Encoding Task to run on a specific thread

Question

I've updated an old project that was using SuperSocket to connect to old c++ servers. With the latest version (from 0.7.0 => 0.8.0.8) I got en exception when trying to reconnect (It says that the socket was opened on a different thread) I would like to have a class that enqueues tasks (First connection and reconnection) and runs them on a specific Thread.

I've seen this approach but when I try to run the task created as got an exception

ExecuteTask may not be called for a task which was previously queued to a different TaskScheduler.

Here's the class I've taken from the link above

public class SameThreadTaskScheduler : TaskScheduler, IDisposable
{
    #region publics
    public SameThreadTaskScheduler(string name)
    {
        scheduledTasks = new Queue<Task>();
        threadName = name;
    }
    public override int MaximumConcurrencyLevel => 1;

    public void Dispose()
    {
        lock (scheduledTasks)
        {
            quit = true;
            Monitor.PulseAll(scheduledTasks);
        }
    }
    #endregion

    #region protected overrides
    protected override IEnumerable<System.Threading.Tasks.Task> 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);
        }
    }

    public void Queue(Task task)
    {
        QueueTask(task);
    }

    protected override bool TryExecuteTaskInline(Task task, bool task_was_previously_queued)
    {
        return false;
    }

    #endregion

    private readonly Queue<System.Threading.Tasks.Task> 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();
    }
}

Here's how I call the task

public void RegisterServers()
{
    sameThreadTaskScheduler.Queue(new Task(() =>
    {
        ...something
    }));

What am I doing wrong?

Try using `Task.Start(TaskScheduler)`, or `Task.Factory.StartNew` with your task scheduler, instead of this public `Queue` method. — acelent, Jan 23 '17 at 19:39

score 2 · Answer 1 · answered Jan 25 '17 at 17:19

You must start a task to bind it to a TaskScheduler. In your code, you're manually queuing the task, doing it the other way around, so the task doesn't get bound to your task scheduler (or any at all), and TryExecuteTask will fail with that error. The description is rather cryptic, as any actual TaskScheduler is different from null.

For a task that you created without running, there are the overloads of Task.RunSynchronously and Task.Start that take a TaskScheduler.

For a task that starts running automatically, there are the overloads of TaskFactory.StartNew and TaskFactory<TResult>.StartNew that take a TaskScheduler.

For continuation tasks, there are the overloads of Task.ContinueWith, Task<TResult>.ContinueWith, TaskFactory.ContinueWhenAll and TaskFactory.ContinueWhenAny that take a TaskScheduler.

The overloads that don't take a task scheduler are equivalent to specifying TaskScheduler.Current. In the case of TaskFactory, this is true for the default Task.Factory or one created without a task scheduler at the time that the factory's method is called, otherwise the factory's task scheduler is used.

Contrast with the overloads of the newer Task.Run, which always use TaskScheduler.Default. According to most experienced folks, the thread-pool scheduler is desired more often as the default than TaskScheduler.Current which may be thread-bound, but it was too late to change the contract for the existing APIs.

Encoding Task to run on a specific thread

1 Answers1