The following is an TaskScheduler that always run tasks in a thread it maintains.
When created, a name of the thread was specified. Once you schedule the first task, until it is been Disposeed, a thread will be created and wait for tasks to execute.
The reason of this class is that sometimes there is a need to guarantee that some tasks must be always scheduled in a specific thread (not the UI thread though). For example, some 3 party dll may have resource leak if you keep creating new threads to call its functions.
using Task = System.Threading.Tasks.Task;
using Thread = System.Threading.Thread;
using Barrier = System.Threading.Barrier;
using Monitor = System.Threading.Monitor;
using IDisposable = System.IDisposable;
using TaskEnum = System.Collections.Generic.IEnumerable<System.Threading.Tasks.Task>;
using TaskQueue = System.Collections.Generic.Queue<System.Threading.Tasks.Task>;
using Enumerable = System.Linq.Enumerable;
using ObjectDisposedException = System.ObjectDisposedException;
using _Imported_Extensions_;
namespace _Imported_Extensions_
{
public static class Extensions
{
public static bool Any(this TaskEnum te)
{
return Enumerable.Any(te);
}
public static TaskEnum ToList(this TaskEnum te)
{
return Enumerable.ToList(te);
}
}
}
namespace TaskUtils
{
public class SameThreadTaskScheduler : System.Threading.Tasks.TaskScheduler, IDisposable
{
#region publics
public SameThreadTaskScheduler(string name)
{
scheduledTasks = new TaskQueue();
threadName = name;
}
public override int MaximumConcurrencyLevel { get { return 1; } }
public void Dispose()
{
lock (scheduledTasks)
{
quit = true;
Monitor.PulseAll(scheduledTasks);
}
}
#endregion
#region protected overrides
protected override TaskEnum GetScheduledTasks()
{
lock (scheduledTasks)
{
return scheduledTasks.ToList();
}
}
protected override void QueueTask(Task task)
{
if (myThread == null)
myThread = StartThread(threadName);
if (!myThread.IsAlive)
throw new ObjectDisposedException("My thread is not alive, so this object has been disposed!");
lock (scheduledTasks)
{
scheduledTasks.Enqueue(task);
Monitor.PulseAll(scheduledTasks);
}
}
protected override bool TryExecuteTaskInline(Task task, bool task_was_previously_queued)
{
return false;
}
#endregion
private readonly TaskQueue scheduledTasks;
private Thread myThread;
private readonly string threadName;
private bool quit;
private Thread StartThread(string name)
{
var t = new Thread(MyThread) { Name = name };
using (var start = new Barrier(2))
{
t.Start(start);
ReachBarrier(start);
}
return t;
}
private void MyThread(object o)
{
Task tsk;
lock (scheduledTasks)
{
//When reaches the barrier, we know it holds the lock.
//
//So there is no Pulse call can trigger until
//this thread starts to wait for signals.
//
//It is important not to call StartThread within a lock.
//Otherwise, deadlock!
ReachBarrier(o as Barrier);
tsk = WaitAndDequeueTask();
}
for (; ; )
{
if (tsk == null)
break;
TryExecuteTask(tsk);
lock (scheduledTasks)
{
tsk = WaitAndDequeueTask();
}
}
}
private Task WaitAndDequeueTask()
{
while (!scheduledTasks.Any() && !quit)
Monitor.Wait(scheduledTasks);
return quit ? null : scheduledTasks.Dequeue();
}
private static void ReachBarrier(Barrier b)
{
if (b != null)
b.SignalAndWait();
}
}
}
I used an unusual using block and put all method extensions in use into a single class. The reason is that I want to specify exactly what I wanted from the outside of the code.
It is fine to use traditional using block instead without change any class code, but anyway focus on the class!
What I am concerning is its concurrency correctness. I want to know although this seems to be working, is it actually correct? Are there better way (simpler) to achieve this? Coding style advises are also welcome, thanks.
Specific Questions
Is it safe to use Pulse rather than PulseAll in this case?
