C# equivalent for Java ExecutorService.newSingleThreadExecutor(), or: how to serialize mulithreaded access to a resource

Question

I have a couple of situations in my code where various threads can create work items that, for various reasons, shouldn't be done in parallel. I'd like to make sure the work gets done in a FIFO manner, regardless of what thread it comes in from. In Java, I'd put the work items on a single-threaded ExecutorService; is there an equivalent in C#? I've cobbled something together with a Queue and a bunch of lock(){} blocks, but it'd be nice to be able to use something off-the-shelf and tested.

Update: Does anybody have experience with System.Threading.Tasks? Does it have a solution for this sort of thing? I'm writing a Monotouch app so who knows if I could even find a backported version of it that I could get to work, but it'd at least be something to think about for the future.

Update #2 For C# developers unfamiliar with the Java libraries I'm talking about, basically I want something that lets various threads hand off work items such that all those work items will be run on a single thread (which isn't any of the calling threads).

---

Update, 6/2018: If I was architecting a similar system now, I'd probably use Reactive Extensions as per Matt Craig's answer. I'm leaving Zachary Yates' answer the accepted one, though, because if you're thinking in Rx you probably wouldn't even ask this question, and I think ConcurrentQueue is easier to bodge into a pre-Rx program.

Answer 1

Update: To address the comments on wasting resources (and if you're not using Rx), you can use a BlockingCollection (if you use the default constructor, it wraps a ConcurrentQueue) and just call .GetConsumingEnumerable(). There's an overload that takes a CancellationToken if the work is long-running. See the example below.

---

You can use ConcurrentQueue, (if monotouch supports .net 4?) it's thread safe and I think the implementation is actually lockless. This works pretty well if you have a long-running task (like in a windows service).

Generally, your problem sounds like you have multiple producers with a single consumer.

var work = new BlockingCollection<Item>();
var producer1 = Task.Factory.StartNew(() => {
    work.TryAdd(item); // or whatever your threads are doing
});
var producer2 = Task.Factory.StartNew(() => {
    work.TryAdd(item); // etc
});
var consumer = Task.Factory.StartNew(() => {
    foreach (var item in work.GetConsumingEnumerable()) {
        // do the work
    }
});
Task.WaitAll(producer1, producer2, consumer);

You should use BlockingCollection if you have a finite pool of work items. Here's an MSDN page showing all of the new concurrent collection types.

Answer 2

I believe this can be done using a SynchronizationContext. However, I have only done this to post back to the UI thread, which already has a synchronization context (if told to be installed) provided by .NET -- I don't know how to prepare it for use from a "vanilla thread" though.

Some links I found for "custom synchronizationcontext provider" (I have not had time to review these, do not fully understand the working/context, nor do I have any additional information):

1. Looking for an example of a custom SynchronizationContext (Required for unit testing)

2. http://codeidol.com/csharp/wcf/Concurrency-Management/Custom-Service-Synchronization-Context/

Happy coding.

Answer 3

There is a more contemporary solution now available - the EventLoopScheduler class.

Answer 4

Not native AFAIK, but look at this: Serial Task Executor; is this thread safe?

Answer 5

I made an example here https://github.com/embeddedmz/message_passing_on_csharp which makes use of BlockingCollection.

So you will have a class that manages a resource and you can use the class below which creates a thread that will be the only one to manage it :

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;

    public class ResourceManagerThread<Resource>
        {
            private readonly Resource _managedResource;
            private readonly BlockingCollection<Action<Resource>> _tasksQueue;
            private Task _task;
            private readonly object _taskLock = new object();
    
            public ResourceManagerThread(Resource resource)
            {
                _managedResource = (resource != null) ? resource : throw new ArgumentNullException(nameof(resource));
                _tasksQueue = new BlockingCollection<Action<Resource>>();
            }
    
            public Task<T> Enqueue<T>(Func<Resource, T> method)
            {
                var tcs = new TaskCompletionSource<T>();
                _tasksQueue.Add(r => tcs.SetResult(method(r)));
                return tcs.Task;
            }
    
            public void Start()
            {
                lock (_taskLock)
                {
                    if (_task == null)
                    {
                        _task = Task.Run(ThreadMain);
                    }
                }
            }
    
            public void Stop()
            {
                lock (_taskLock)
                {
                    if (_task != null)
                    {
                        _tasksQueue.CompleteAdding();
                        _task.Wait();
                        _task = null;
                    }
                }
            }
    
            public bool HasStarted
            {
                get
                {
                    lock (_taskLock)
                    {
                        if (_task != null)
                        {
                            return _task.IsCompleted == false || 
                                _task.Status == TaskStatus.Running ||
                                _task.Status == TaskStatus.WaitingToRun ||
                                _task.Status == TaskStatus.WaitingForActivation;
                        }
                        else
                        {
                            return false;
                        }
                    }
                }
            }
    
            private void ThreadMain()
            {
                try
                {
                    foreach (var action in _tasksQueue.GetConsumingEnumerable())
                    {
                        try
                        {
                            action(_managedResource);
                        }
                        catch
                        {
                            //...
                        }
                    }
                }
                catch
                {
                }
            }
        }

Example :

private readonly DevicesManager _devicesManager;
private readonly ResourceManagerThread<DevicesManager> _devicesManagerThread;
        
//...
        
_devicesManagerThread = new ResourceManagerThread<DevicesManager>(_devicesManager);
_devicesManagerThread.Start();
_devicesManagerThread.Enqueue((DevicesManager dm) =>
{
    return dm.Initialize();
});

    // Enqueue will return a Task. Use the 'Result' property to get the result of the 'message' or 'request' sent to the the thread managing the resource

Answer 6

As I wrote in comments, you discovered by yourself that the lock statement can do the work.

If you are interested in getting a "container" that can make simpler the job of managing a queue of work items, look at the ThreadPool class.

I think that, in a well designed architecture, with these two elemnts (ThreadPool class and lock statement) you can easily and succesfully serialize access to resources.