admin/mirror-processhacker
1
0
Fork 0
mirror of https://github.com/mirror/processhacker synced 2026-06-08 16:03:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

added new wait manager with support for more than 63 objects

Browse Source 
git-svn-id: svn://svn.code.sf.net/p/processhacker/code@1378 21ef857c-d57f-4fe0-8362-d861dc6d29cd
This commit is contained in:
wj32
2009-06-06 03:37:24 +00:00
parent 67e56d64bb
commit e18334a4bb
4 changed files with 354 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
trunk/ProcessHacker.Common/WorkQueue.cs
+1 -1
View File
@@ -565,7 +565,7 @@ namespace ProcessHacker.Common
}
}
// Check for work
// Check for work.
if (_workQueue.Count > 0)
{
WorkItem workItem = null;
trunk/ProcessHacker.Native/ProcessHacker.Native.csproj
+1 -1
View File
@@ -144,7 +144,7 @@
<Compile Include="Threading\Semaphore.cs" />
<Compile Include="Threading\SpinLock.cs" />
<Compile Include="Threading\Timer.cs" />
<Compile Include="Threading\WaiterThread.cs" />
<Compile Include="Threading\Waiter.cs" />
<Compile Include="Ui\HandlePropertiesWindow.cs">
<SubType>Form</SubType>
</Compile>
trunk/ProcessHacker.Native/Threading/Waiter.cs
+352
View File
@@ -0,0 +1,352 @@
/*
* Process Hacker -
* wait manager
*
* Copyright (C) 2009 wj32
*
* This file is part of Process Hacker.
*
* Process Hacker is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* Process Hacker is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with Process Hacker. If not, see <http://www.gnu.org/licenses/>.
*/
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using ProcessHacker.Native.Api;
using ProcessHacker.Native.Objects;
using ProcessHacker.Native.Security;
namespace ProcessHacker.Native.Threading
{
public delegate void ObjectSignaledDelegate(ISynchronizable obj);
public class Waiter : IDisposable
{
private class WaiterThread : IDisposable
{
public event ObjectSignaledDelegate ObjectSignaled;
private bool _disposed = false;
private object _disposeLock = new object();
private bool _terminating = false;
private Thread _thread;
private List<ISynchronizable> _waitObjects = new List<ISynchronizable>();
private Event _waitChangedEvent = new Event(true, false);
public WaiterThread()
{
_thread = new Thread(this.WaiterThreadStart);
_thread.IsBackground = true;
_thread.SetApartmentState(ApartmentState.STA);
_thread.Start();
}
~WaiterThread()
{
this.Dispose(false);
}
public void Dispose()
{
this.Dispose(true);
GC.SuppressFinalize(this);
}
private void Dispose(bool disposing)
{
if (disposing)
Monitor.Enter(_disposeLock);
try
{
if (!_disposed)
{
this.Terminate();
_waitChangedEvent.Dispose();
_disposed = true;
}
}
finally
{
Monitor.Exit(_disposeLock);
}
}
public int Count
{
get
{
lock (_waitObjects)
return _waitObjects.Count;
}
}
public ISynchronizable[] Objects
{
get
{
lock (_waitObjects)
return _waitObjects.ToArray();
}
}
public bool Add(ISynchronizable obj)
{
lock (_waitObjects)
{
// Check if we already have the maximum number of wait objects.
if (_waitObjects.Count >= Win32.MaximumWaitObjects - 1)
return false;
_waitObjects.Add(obj);
this.NotifyChange();
return true;
}
}
public void NotifyChange()
{
_waitChangedEvent.Set();
}
private void OnObjectSignaled(ISynchronizable obj)
{
if (this.ObjectSignaled != null)
this.ObjectSignaled(obj);
}
public bool Remove(ISynchronizable obj)
{
lock (_waitObjects)
{
if (!_waitObjects.Contains(obj))
return false;
_waitObjects.Remove(obj);
this.NotifyChange();
return true;
}
}
public void Terminate()
{
_terminating = true;
this.NotifyChange();
}
private void WaiterThreadStart()
{
ISynchronizable[] waitObjects;
while (!_terminating)
{
lock (_waitObjects)
{
waitObjects = new ISynchronizable[_waitObjects.Count + 1];
waitObjects[0] = _waitChangedEvent.Handle;
Array.Copy(_waitObjects.ToArray(), 0, waitObjects, 1, _waitObjects.Count);
}
NtStatus waitStatus = NativeHandle.WaitAny(waitObjects);
if (waitStatus == NtStatus.Wait0)
{
// The wait was changed. Go back to refresh the wait objects array.
// The event is also signaled to notify that the thread should terminate.
continue;
}
else if (waitStatus > NtStatus.Wait0 && waitStatus <= NtStatus.Wait63)
{
// One of the objects was signaled.
ISynchronizable signaledObject = waitObjects[(int)(waitStatus - NtStatus.Wait0)];
// Remove the object now that it is signaled.
_waitObjects.Remove(signaledObject);
// Call the object-signaled event.
OnObjectSignaled(signaledObject);
}
}
}
}
/// <summary>
/// Raised when an object is signaled.
/// </summary>
public event ObjectSignaledDelegate ObjectSignaled;
private bool _disposed = false;
private object _disposeLock = new object();
private List<WaiterThread> _waiterThreads = new List<WaiterThread>();
private List<ISynchronizable> _waitObjects = new List<ISynchronizable>();
/// <summary>
/// Creates a waiter.
/// </summary>
public Waiter()
{
}
~Waiter()
{
this.Dispose(false);
}
/// <summary>
/// Releases resources used by the waiter.
/// </summary>
public void Dispose()
{
this.Dispose(true);
GC.SuppressFinalize(this);
}
private void Dispose(bool disposing)
{
if (disposing)
Monitor.Enter(_disposeLock);
try
{
if (!_disposed)
{
// Tell the waiter threads to terminate.
foreach (var waiterThread in _waiterThreads)
waiterThread.Terminate();
_waiterThreads.Clear();
_disposed = true;
}
}
finally
{
Monitor.Exit(_disposeLock);
}
}
public int Count
{
get
{
lock (_waitObjects)
return _waitObjects.Count;
}
}
public ISynchronizable[] Objects
{
get
{
lock (_waitObjects)
return _waitObjects.ToArray();
}
}
/// <summary>
/// Adds an object for the waiter to wait on.
/// </summary>
/// <param name="obj">The object to wait for.</param>
public void Add(ISynchronizable obj)
{
lock (_waitObjects)
_waitObjects.Add(obj);
foreach (var waiterThread in this.GetWaiterThreads())
{
if (waiterThread.Add(obj))
return;
}
// We couldn't add the object to any existing waiter thread.
// Create a new waiter thread and add the object to that.
this.CreateWaiterThread(obj);
}
private void BalanceWaiterThreads()
{
lock (_waitObjects)
{
// Eliminate waiter threads with no objects.
foreach (var waiterThread in this.GetWaiterThreads())
{
if (waiterThread.Count == 0)
this.DeleteWaiterThread(waiterThread);
}
}
}
private WaiterThread CreateWaiterThread()
{
return this.CreateWaiterThread(null);
}
private WaiterThread CreateWaiterThread(ISynchronizable obj)
{
WaiterThread waiterThread = new WaiterThread();
waiterThread.ObjectSignaled += this.OnObjectSignaled;
if (obj != null)
waiterThread.Add(obj);
lock (_waiterThreads)
_waiterThreads.Add(waiterThread);
return waiterThread;
}
private void DeleteWaiterThread(WaiterThread waiterThread)
{
lock (_waiterThreads)
{
_waiterThreads.Remove(waiterThread);
waiterThread.Terminate();
}
}
private WaiterThread[] GetWaiterThreads()
{
lock (_waiterThreads)
return _waiterThreads.ToArray();
}
private void OnObjectSignaled(ISynchronizable obj)
{
if (ObjectSignaled != null)
ObjectSignaled(obj);
}
/// <summary>
/// Removes an object the waiter is waiting on.
/// </summary>
/// <param name="obj">An object which is currently being waited on.</param>
public void Remove(ISynchronizable obj)
{
foreach (var waiterThread in this.GetWaiterThreads())
{
if (waiterThread.Remove(obj))
{
lock (_waitObjects)
_waitObjects.Remove(obj);
this.BalanceWaiterThreads();
return;
}
}
// We couldn't remove the object.
throw new ArgumentException("The object is not being waited on.");
}
}
}
trunk/ProcessHacker.Native/Threading/WaiterThread.cs
-226
View File
@@ -1,226 +0,0 @@
/*
* Process Hacker -
* waiter thread
*
* Copyright (C) 2009 wj32
*
* This file is part of Process Hacker.
*
* Process Hacker is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* Process Hacker is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with Process Hacker. If not, see <http://www.gnu.org/licenses/>.
*/
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using ProcessHacker.Native.Api;
using ProcessHacker.Native.Objects;
using ProcessHacker.Native.Security;
namespace ProcessHacker.Native.Threading
{
public delegate void ObjectSignaledDelegate(ISynchronizable obj);
public class WaiterThread : IDisposable
{
private enum WaiterThreadMessageType
{
AddObject,
RemoveObject
}
private class WaiterThreadMessage
{
private WaiterThreadMessageType _type;
private object _param;
public WaiterThreadMessage(WaiterThreadMessageType type)
: this(type, null)
{ }
public WaiterThreadMessage(WaiterThreadMessageType type, object param)
{
_type = type;
_param = param;
}
public WaiterThreadMessageType Type { get { return _type; } }
public object Param { get { return _param; } }
}
/// <summary>
/// Raised when an object is signaled.
/// </summary>
public event ObjectSignaledDelegate ObjectSignaled;
private bool _disposed = false;
private bool _terminating = false;
private object _disposeLock = new object();
private Thread _waiterThread;
private List<ISynchronizable> _waitObjects = new List<ISynchronizable>();
private Queue<WaiterThreadMessage> _waitMessageQueue = new Queue<WaiterThreadMessage>();
private Event _waitMessageEvent = new Event(true, false);
/// <summary>
/// Creates a waiter thread.
/// </summary>
public WaiterThread()
{
_waiterThread = new Thread(this.WaiterThreadStart);
_waiterThread.IsBackground = true;
_waiterThread.SetApartmentState(ApartmentState.STA);
_waiterThread.Start();
}
~WaiterThread()
{
this.Dispose(false);
}
/// <summary>
/// Releases resources used by the waiter thread.
/// </summary>
public void Dispose()
{
this.Dispose(true);
GC.SuppressFinalize(this);
}
private void Dispose(bool disposing)
{
if (disposing)
Monitor.Enter(_disposeLock);
try
{
if (!_disposed)
{
// Tell the waiter thread to terminate.
_terminating = true;
_waitMessageEvent.Set();
_disposed = true;
}
}
finally
{
Monitor.Exit(_disposeLock);
}
}
/// <summary>
/// Adds an object for the waiter thread to wait on.
/// </summary>
/// <param name="obj">The object to wait for.</param>
public void Add(ISynchronizable obj)
{
lock (_waitMessageQueue)
{
if (_waitObjects.Count >= Win32.MaximumWaitObjects - 1)
throw new TooManyWaitObjectsException();
this.SendWaiterThreadMessage(new WaiterThreadMessage(WaiterThreadMessageType.AddObject, obj));
}
}
private void OnObjectSignaled(ISynchronizable obj)
{
if (ObjectSignaled != null)
ObjectSignaled(obj);
}
/// <summary>
/// Removes an object the waiter thread is waiting on.
/// </summary>
/// <param name="obj">An object which is currently being waited on.</param>
public void Remove(ISynchronizable obj)
{
this.SendWaiterThreadMessage(new WaiterThreadMessage(WaiterThreadMessageType.RemoveObject, obj));
}
private void SendWaiterThreadMessage(WaiterThreadMessage message)
{
lock (_waitMessageQueue)
{
_waitMessageQueue.Enqueue(message);
_waitMessageEvent.Set();
}
}
private void WaiterThreadStart()
{
ISynchronizable[] waitObjects;
while (!_terminating)
{
waitObjects = new ISynchronizable[_waitObjects.Count + 1];
waitObjects[0] = _waitMessageEvent.Handle;
Array.Copy(_waitObjects.ToArray(), 0, waitObjects, 1, _waitObjects.Count);
NtStatus waitStatus = NativeHandle.WaitAny(waitObjects);
if (waitStatus == NtStatus.Wait0)
{
// We have a message in the message queue (or we are terminating).
WaiterThreadMessage message;
// Lock and retrieve the message.
lock (_waitMessageQueue)
{
if (_waitMessageQueue.Count > 0)
message = _waitMessageQueue.Dequeue();
else
continue;
}
switch (message.Type)
{
case WaiterThreadMessageType.AddObject:
{
_waitObjects.Add((ISynchronizable)message.Param);
}
break;
case WaiterThreadMessageType.RemoveObject:
{
_waitObjects.Remove((ISynchronizable)message.Param);
}
break;
}
}
else if (waitStatus > NtStatus.Wait0 && waitStatus <= NtStatus.Wait63)
{
// One of the objects was signaled.
ISynchronizable signaledObject = waitObjects[(int)(waitStatus - NtStatus.Wait0)];
// Remove the object now that it is signaled.
_waitObjects.Remove(signaledObject);
// Call the object-signaled event.
OnObjectSignaled(signaledObject);
}
}
}
}
public class TooManyWaitObjectsException : Exception
{
public override string Message
{
get
{
return "An attempt was made to add too many objects to the wait list.";
}
}
}
}