5
\$\begingroup\$

I'm interested in some feedback for a custom TaskScheduler implementation I wrote today for use on a game server - I wanted to be able to have a single-threaded scheduler for tasks since the game is relatively lightweight and its logic doesn't need to be multi-threaded. However, I wanted to handle networking/IO concurrently to avoid blocking the main thread running game logic, and still wanted the ability to easily schedule tasks onto it, jumping back and forth between executors (e.g. game server receives a request, does some logic, then requires something from a database, makes a call on some other thread pool, then executes a callback on the original thread again). Anyway, that's probably too much about why I wanted to do this; Here's how I attempted it:

public class CustomTaskScheduler : TaskScheduler
{
    private readonly int maxThreads;
    private readonly ConcurrentQueue<Task> tasks;
    private readonly bool unbounded;
    private int threadsRunning;

    private CustomTaskScheduler(int concurrencyLevel, bool unboundedThreads)
    {
        maxThreads = concurrencyLevel;
        tasks = new ConcurrentQueue<Task>();
        threadsRunning = 0;
        unbounded = unboundedThreads;
    }

    protected override IEnumerable<Task> GetScheduledTasks()
    {
        return tasks;
    }

    protected override void QueueTask(Task task)
    {
        tasks.Enqueue(task);

        if (unbounded || Interlocked.Increment(ref threadsRunning) <= maxThreads)
        {
            ThreadPool.UnsafeQueueUserWorkItem(_ =>
            {
                try
                {
                    while (tasks.TryDequeue(out Task toRun))
                    {
                        TryExecuteTask(toRun);
                    }
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex);
                }
                finally
                {
                    Interlocked.Decrement(ref threadsRunning);
                }
            }, null);
        }
        else
        {
            Interlocked.Decrement(ref threadsRunning);
        }
    }

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

    public static CustomTaskScheduler MakeSingleThreaded() => new CustomTaskScheduler(1, false);
    public static CustomTaskScheduler MakeMultiThreaded(int threads) => new CustomTaskScheduler(threads, false);
    public static CustomTaskScheduler MakeUnbounded() => new CustomTaskScheduler(0, true);
}

It isn't that complicated, but there are a few things I was wondering about - It is based off of the sample code here, but elects to use a ConcurrentQueue instead of locking on a LinkedList. From what I can see, this causes no issues. Then, I did an interlocked increment on the number of threads running to make sure it was not exceeding the limit, decrementing it immediately if another thread could not be run. This, too, I think is safe and correct, but it still makes me uncomfortable, and I was wondering if there was a more elegant solution to this. I found the Interlocked.CompareExchange function, but what i'm more looking for is being able to compare if something is less than something else.

Finally, I have tried doing some research on TryExecuteTaskInline, and it seems that this allows you to run a task directly from the thread creating the task, which seems like what I don't want. For now I've set it to always return false, but i'm not sure if this is what i'm supposed to do, or if i'm misunderstanding what it is actually doing.

\$\endgroup\$

closed as off-topic by 200_success, VisualMelon, Mast, dfhwze, pacmaninbw Aug 3 at 16:37

This question appears to be off-topic. The users who voted to close gave this specific reason:

  • "Code not implemented or not working as intended: Code Review is a community where programmers peer-review your working code to address issues such as security, maintainability, performance, and scalability. We require that the code be working correctly, to the best of the author's knowledge, before proceeding with a review." – 200_success, VisualMelon, Mast, dfhwze, pacmaninbw
If this question can be reworded to fit the rules in the help center, please edit the question.

  • 1
    \$\begingroup\$ Nope, this is incorrect. It is possible for the running thread to complete all tasks, then have a new one be queued, fail to start a thread, then have the running thread complete. This would leave tasks in the queue without a thread consuming them. It looks like locking is the most obvious solution to this. \$\endgroup\$ – M Anderson Sep 8 '18 at 14:58
  • \$\begingroup\$ Feel free to edit the question (as long as it has no answers) with new version. The problem indeed is that the increment-check-decrement is not atomic as a whole (and the reason it is done under lock in the example - QueueTask). \$\endgroup\$ – firda Sep 28 '18 at 18:42
  • 1
    \$\begingroup\$ P.S.: If you really want to have single-threaded scheduler then you can use Interlocked.CompareExchange to achieve that (which, together with supposedly lock-free queue, could actually be very good solution, but be careful, still won't be trivial). \$\endgroup\$ – firda Sep 28 '18 at 18:54
  • \$\begingroup\$ Take a look at ConcurrentExclusiveSchedulerPair or more specifically ConcurrentExclusiveSchedulerPair.ConcurrentScheduler which is the built-in way of doing it. \$\endgroup\$ – Johnbot Aug 1 at 9:22