farcasclaudiu/SmartThreadPool
1
0
Fork 0
mirror of https://github.com/farcasclaudiu/SmartThreadPool.git synced 2026-06-28 21:01:11 +03:00
Code Issues Packages Projects Releases Wiki Activity

Enabled cancel abort after cancel. (Suggested by alecswan)

Browse Source 
Also added a test for it.
This commit is contained in:
Ami Bar
2012-08-25 00:48:41 +03:00
parent 325369cd00
commit 4af5ce990d
2 changed files with 598 additions and 999 deletions
Download Patch File Download Diff File Expand all files Collapse all files
STPTests/TestCancel.cs
+42 -454
View File
@@ -77,6 +77,47 @@ namespace SmartThreadPoolTests
}
}
/// <summary>
/// 1. Create STP
/// 2. Queue work item that takes some time
/// 3. Wait for it to start
/// 4. Cancel the work item (soft)
/// 5. Work item's GetResult should throw WorkItemCancelException
/// </summary>
[Test]
[ExpectedException(typeof(WorkItemCancelException))]
public void CancelCancelledWorkItemAbort()
{
ManualResetEvent waitToStart = new ManualResetEvent(false);
SmartThreadPool stp = new SmartThreadPool();
IWorkItemResult wir = stp.QueueWorkItem(
state => { waitToStart.Set(); while (true) { Thread.Sleep(1000); } return null; }
);
waitToStart.WaitOne();
wir.Cancel(false);
Assert.IsTrue(wir.IsCanceled);
bool completed = stp.WaitForIdle(1000);
Assert.IsFalse(completed);
wir.Cancel(true);
try
{
wir.GetResult();
}
finally
{
stp.Shutdown();
}
}
/// <summary>
/// 1. Create STP
/// 2. Queue work item that:
@@ -191,6 +232,7 @@ namespace SmartThreadPoolTests
stp.WaitForIdle();
// Throws WorkItemCancelException
wir.GetResult();
stp.Shutdown();
@@ -428,459 +470,5 @@ namespace SmartThreadPoolTests
stp.Shutdown();
}
//////////////////////////////////////////////////////////////////////////////////////////////////
/*
private int _counter;
/// <summary>
/// Example of how to queue a work item and then cancel it while it is in the queue.
/// </summary>
[Test]
[ExpectedException(typeof(WorkItemCancelException))]
public void WorkItemCanceling()
{
// Create a SmartThreadPool with only one thread.
// It just to show how to use the work item canceling feature
SmartThreadPool smartThreadPool = new SmartThreadPool(10*1000, 1);
// Queue a work item that will occupy the thread in the pool
IWorkItemResult wir1 =
smartThreadPool.QueueWorkItem(new WorkItemCallback(this.DoSomeWork), null);
// Queue another work item that will wait for the first to complete
IWorkItemResult wir2 =
smartThreadPool.QueueWorkItem(new WorkItemCallback(this.DoSomeWork), null);
// Wait a while for the thread pool to start executing the first work item
Thread.Sleep(100);
try
{
// The first work item cannot be canceled since it is currently executing
if (!wir1.Cancel())
{
// Cancel the second work item while it still in the queue
if (wir2.Cancel())
{
// Retreiving result of a canceled work item throws WorkItemCancelException exception
wir2.GetResult();
}
}
}
finally
{
smartThreadPool.Shutdown();
}
}
/// <summary>
/// </summary>
[Test]
public void WorkItemCancelingAndInUseWorkerThreads()
{
SmartThreadPool smartThreadPool = new SmartThreadPool(10*1000, 10);
IWorkItemResult [] wirs = new IWorkItemResult[100];
for(int i = 0; i < 100; ++i)
{
wirs[i] = smartThreadPool.QueueWorkItem(new WorkItemCallback(this.DoSomeWork), null);
}
// Wait a while for the thread pool to start executing the first work item
Thread.Sleep(100);
for(int i = 0; i < 100; ++i)
{
wirs[i].Cancel();
}
smartThreadPool.WaitForIdle(2000);
int inUseThreads = smartThreadPool.InUseThreads;
smartThreadPool.Shutdown();
Assert.AreEqual(0, inUseThreads);
}
private object DoSomeWork(object state)
{
Thread.Sleep(1000);
return 1;
}
/// <summary>
/// Check within the work item if it was cancelled
/// </summary>
[Test]
public void SampleIfWorkItemCancelled()
{
_counter = 0;
STPStartInfo stpStartInfo = new STPStartInfo();
stpStartInfo.StartSuspended = true;
// Create a SmartThreadPool with only one thread.
SmartThreadPool smartThreadPool = new SmartThreadPool(stpStartInfo);
IWorkItemResult[] wirs = new IWorkItemResult[100];
for (int i = 0; i < 100; ++i)
{
wirs[i] = smartThreadPool.QueueWorkItem(new WorkItemCallback(this.DoCheckForCancelledWorkItem), null);
}
for (int i = 0; i < 50; ++i)
{
wirs[i].Cancel();
}
smartThreadPool.Start();
smartThreadPool.WaitForIdle();
smartThreadPool.Shutdown();
Assert.AreEqual(50, _counter);
}
/// <summary>
/// 1. Create STP
/// 2. Queue work item into the STP
/// 4. Cancel the work item
/// 5. Work item doesn't check for cancel
/// 6. Work item quits
/// 7. Make sure the work item result is ok, and not an exception
/// </summary>
[Test]
public void TestWorkItemCancelledWorkItemOK()
{
// Create a SmartThreadPool with only one thread.
SmartThreadPool smartThreadPool = new SmartThreadPool();
AutoResetEvent start = new AutoResetEvent(false);
// Queue the work item
IWorkItemResult wir = smartThreadPool.QueueWorkItem(new WorkItemCallback(this.DoWaitForCancel), start);
// Wait for it to start executing
bool success = start.WaitOne(3000, false);
// Make sure it was started
Assert.IsTrue(success);
// Cancel the work item
wir.Cancel();
// Let it complete
start.Set();
// Wait for the work item to complete
smartThreadPool.WaitForIdle();
// Check the work item's result
// The work item started running after it was start its execution.
// Since the work item didn't sample the cancel, it was not aware that it
// was canceled. Therefore the result should be what the work item returned
// and not the cancel exeception
int result = (int)wir.GetResult(0, false);
smartThreadPool.Shutdown();
Assert.AreEqual(1, result);
}
/// <summary>
/// 1. Create STP
/// 2. Create WIG
/// 3. Queue work item into the WIG
/// 4. Cancel the work items in the WIG
/// 5. Work item doesn't check for cancel
/// 6. Work item quits
/// 7. Make sure the work item result is ok, and not an exception
/// </summary>
[Test]
public void TestWIGCancelledWorkItemOK()
{
// Create a SmartThreadPool with only one thread.
SmartThreadPool smartThreadPool = new SmartThreadPool();
IWorkItemsGroup wig = smartThreadPool.CreateWorkItemsGroup(1);
AutoResetEvent start = new AutoResetEvent(false);
// Queue the work item
IWorkItemResult wir = wig.QueueWorkItem(new WorkItemCallback(this.DoWaitForCancel), start);
// Wait for it to start executing
bool success = start.WaitOne(3000, false);
// Make sure it was started
Assert.IsTrue(success);
// Cancel the work item
wig.Cancel();
// Let it complete
start.Set();
// Wait for the work item to complete
smartThreadPool.WaitForIdle();
// Check the work item's result
// The work item started running after it was start its execution.
// Since the work item didn't sample the cancel, it was not aware that it
// was canceled. Therefore the result should be what the work item returned
// and not the cancel exeception
int result = (int)wir.GetResult(0, false);
smartThreadPool.Shutdown();
Assert.AreEqual(1, result);
}
/// <summary>
/// 1. Create STP
/// 2. Queue work item into the STP
/// 3. Cancel the work items in the STP
/// 4. Work item doesn't check for cancel
/// 5. Work item quits
/// 6. Make sure the work item result is ok, and not an exception
/// </summary>
[Test]
public void TestSTPCancelledWorkItemOK()
{
// Create a SmartThreadPool with only one thread.
SmartThreadPool smartThreadPool = new SmartThreadPool();
AutoResetEvent start = new AutoResetEvent(false);
// Queue the work item
IWorkItemResult wir = smartThreadPool.QueueWorkItem(new WorkItemCallback(this.DoWaitForCancel), start);
// Wait for it to start executing
bool success = start.WaitOne(3000, false);
// Make sure it was started
Assert.IsTrue(success);
// Cancel the work item
smartThreadPool.Cancel();
// Let it complete
start.Set();
// Wait for the work item to complete
smartThreadPool.WaitForIdle();
// Check the work item's result
// The work item started running after it was start its execution.
// Since the work item didn't sample the cancel, it was not aware that it
// was canceled. Therefore the result should be what the work item returned
// and not the cancel exeception
int result = (int)wir.GetResult(0, false);
smartThreadPool.Shutdown();
Assert.AreEqual(1, result);
}
/// <summary>
/// 1. Create STP
/// 2. Queue work item
/// 3. Cancel the work item
/// 4. Work item checks the cancel and quits
/// 5. Make sure the work item result throws exception
/// </summary>
[Test]
[ExpectedException(typeof(WorkItemCancelException))]
public void TestWorkItemCanceledWorkItemCancelException()
{
// Create a SmartThreadPool with only one thread.
SmartThreadPool smartThreadPool = new SmartThreadPool();
AutoResetEvent start = new AutoResetEvent(false);
// Queue the work item
IWorkItemResult wir = smartThreadPool.QueueWorkItem(new WorkItemCallback(this.DoCheckForCancel), start);
// Wait for it to start executing
bool success = start.WaitOne(3000, false);
// Make sure it was started
Assert.IsTrue(success);
// Cancel the work item
wir.Cancel();
// Let it complete
start.Set();
// Wait for the work item to complete
smartThreadPool.WaitForIdle();
try
{
// Check the work item's result
// The work item started running after it was start its execution.
// The work item samples the cancel, therefore it is aware that it
// was canceled. Using the GetResult should throw the cancel exeception
wir.GetResult(0, false);
}
finally
{
smartThreadPool.Shutdown();
}
}
/// <summary>
/// 1. Create STP
/// 2. Create WIG
/// 3. Queue work item into the WIG
/// 4. Cancel the work items in the WIG
/// 5. Work item checks the cancel and quits
/// 6. Make sure the work item result throws exception
/// </summary>
[Test]
[ExpectedException(typeof(WorkItemCancelException))]
public void TestWIGCancelledWorkItemCancelException()
{
// Create a SmartThreadPool with only one thread.
SmartThreadPool smartThreadPool = new SmartThreadPool();
IWorkItemsGroup wig = smartThreadPool.CreateWorkItemsGroup(1);
AutoResetEvent start = new AutoResetEvent(false);
// Queue the work item
IWorkItemResult wir = wig.QueueWorkItem(new WorkItemCallback(this.DoCheckForCancel), start);
// Wait for it to start executing
bool success = start.WaitOne(3000, false);
// Make sure it was started
Assert.IsTrue(success);
// Cancel the work item
wig.Cancel();
// Let it complete
start.Set();
// Wait for the work item to complete
smartThreadPool.WaitForIdle();
try
{
// Check the work item's result
// The work item started running after it was start its execution.
// The work item samples the cancel, therefore it is aware that it
// was canceled. Using the GetResult should throw the cancel exeception
wir.GetResult(0, false);
}
finally
{
smartThreadPool.Shutdown();
}
}
/// <summary>
/// 1. Create STP
/// 3. Queue work item into the STP
/// 4. Cancel the work items in the STP
/// 5. Work item checks the cancel and quits
/// 6. Make sure the work item result throws exception
/// </summary>
[Test]
[ExpectedException(typeof(WorkItemCancelException))]
public void TestSTPCancelledWorkItemCancelException()
{
// Create a SmartThreadPool with only one thread.
SmartThreadPool smartThreadPool = new SmartThreadPool();
AutoResetEvent start = new AutoResetEvent(false);
// Queue the work item
IWorkItemResult wir = smartThreadPool.QueueWorkItem(new WorkItemCallback(this.DoCheckForCancel), start);
// Wait for it to start executing
bool success = start.WaitOne(3000, false);
// Make sure it was started
Assert.IsTrue(success);
// Cancel the work item
smartThreadPool.Cancel();
// Let it complete
start.Set();
// Wait for the work item to complete
smartThreadPool.WaitForIdle();
try
{
// Check the work item's result
// The work item started running after it was start its execution.
// The work item samples the cancel, therefore it is aware that it
// was canceled. Using the GetResult should throw the cancel exeception
wir.GetResult(0, false);
}
finally
{
smartThreadPool.Shutdown();
}
}
private object DoCheckForCancelledWorkItem(object state)
{
if (!SmartThreadPool.IsWorkItemCanceled)
{
Interlocked.Increment(ref _counter);
}
return null;
}
private object DoWaitForCancel(object state)
{
AutoResetEvent start = state as AutoResetEvent;
// Signal the test that the work item started
start.Set();
// Let the test run (or else the next line may reset the signaled event)
Thread.Sleep(10);
// Wait for the test to cancel the work item
start.WaitOne();
return 1;
}
private object DoCheckForCancel(object state)
{
AutoResetEvent start = state as AutoResetEvent;
// Signal the test that the work item started
start.Set();
// Let the test run (or else the next line may reset the signaled event)
Thread.Sleep(10);
// Wait for the test to cancel the work item
start.WaitOne();
// Sample if the work item was cancelled
bool cancelled = SmartThreadPool.IsWorkItemCanceled;
return 1;
}
*/
}
}
SmartThreadPool/WorkItem.cs
+29 -18
View File
@@ -272,7 +272,7 @@ namespace Amib.Threading.Internal
_waitingOnQueueStopwatch.Stop();
_processingStopwatch.Start();
lock(this)
lock (this)
{
if (IsCanceled)
{
@@ -305,7 +305,7 @@ namespace Amib.Threading.Internal
CallToPostExecute currentCallToPostExecute = 0;
// Execute the work item if we are in the correct state
switch(GetWorkItemState())
switch (GetWorkItemState())
{
case WorkItemState.InProgress:
currentCallToPostExecute |= CallToPostExecute.WhenWorkItemNotCanceled;
@@ -338,7 +338,7 @@ namespace Amib.Threading.Internal
}
}
catch // Suppress exceptions
{}
{ }
}
internal void FireWorkItemStarted()
@@ -392,7 +392,7 @@ namespace Amib.Threading.Internal
if (null == executionThread)
{
// Oops! we are going to be aborted..., Wait here so we can catch the ThreadAbortException
Thread.Sleep(60*1000);
Thread.Sleep(60 * 1000);
// If after 1 minute this thread was not aborted then let it continue working.
}
@@ -502,14 +502,14 @@ namespace Amib.Threading.Internal
int millisecondsLeft = millisecondsTimeout;
Stopwatch stopwatch = Stopwatch.StartNew();
WaitHandle [] whs;
WaitHandle[] whs;
if (null != cancelWaitHandle)
{
whs = new WaitHandle [] { null, cancelWaitHandle };
whs = new WaitHandle[] { null, cancelWaitHandle };
}
else
{
whs = new WaitHandle [] { null };
whs = new WaitHandle[] { null };
}
bool waitInfinitely = (Timeout.Infinite == millisecondsTimeout);
@@ -534,7 +534,7 @@ namespace Amib.Threading.Internal
break;
}
if(!waitInfinitely)
if (!waitInfinitely)
{
// Update the time left to wait
millisecondsLeft = millisecondsTimeout - (int)stopwatch.ElapsedMilliseconds;
@@ -565,7 +565,7 @@ namespace Amib.Threading.Internal
bool exitContext,
WaitHandle cancelWaitHandle)
{
WaitHandle [] waitHandles;
WaitHandle[] waitHandles;
if (null != cancelWaitHandle)
{
@@ -602,7 +602,7 @@ namespace Amib.Threading.Internal
/// <param name="waitHandles">An array of wait handles to fill</param>
private static void GetWaitHandles(
IWaitableResult[] waitableResults,
WaitHandle [] waitHandles)
WaitHandle[] waitHandles)
{
for (int i = 0; i < waitableResults.Length; ++i)
{
@@ -668,7 +668,7 @@ namespace Amib.Threading.Internal
/// <param name="workItemState">The state to set the work item to</param>
private void SetWorkItemState(WorkItemState workItemState)
{
lock(this)
lock (this)
{
if (IsValidStatesTransition(_workItemState, workItemState))
{
@@ -684,7 +684,7 @@ namespace Amib.Threading.Internal
private void SignalComplete(bool canceled)
{
SetWorkItemState(canceled ? WorkItemState.Canceled : WorkItemState.Completed);
lock(this)
lock (this)
{
// If someone is waiting then signal.
if (null != _workItemCompleted)
@@ -720,10 +720,21 @@ namespace Amib.Threading.Internal
lock (this)
{
switch(GetWorkItemState())
switch (GetWorkItemState())
{
case WorkItemState.Canceled:
//Debug.WriteLine("Work item already canceled");
if (abortExecution)
{
Thread executionThread = Interlocked.CompareExchange(ref _executingThread, null, _executingThread);
if (null != executionThread)
{
executionThread.Abort(); // "Cancel"
// No need to signalComplete, because we already cancelled this work item
// so it already signaled its completion.
//signalComplete = true;
}
}
success = true;
break;
case WorkItemState.Completed:
@@ -831,7 +842,7 @@ namespace Amib.Threading.Internal
int result = STPEventWaitHandle.WaitAny(new WaitHandle[] { wh, cancelWaitHandle });
ReleaseWaitHandle();
switch(result)
switch (result)
{
case 0:
// The work item signaled
@@ -867,7 +878,7 @@ namespace Amib.Threading.Internal
/// </summary>
private WaitHandle GetWaitHandle()
{
lock(this)
lock (this)
{
if (null == _workItemCompleted)
{
@@ -880,7 +891,7 @@ namespace Amib.Threading.Internal
private void ReleaseWaitHandle()
{
lock(this)
lock (this)
{
if (null != _workItemCompleted)
{
@@ -901,7 +912,7 @@ namespace Amib.Threading.Internal
{
get
{
lock(this)
lock (this)
{
WorkItemState workItemState = GetWorkItemState();
return ((workItemState == WorkItemState.Completed) ||
@@ -917,7 +928,7 @@ namespace Amib.Threading.Internal
{
get
{
lock(this)
lock (this)
{
return (GetWorkItemState() == WorkItemState.Canceled);
}