Invoking a callback on a pool of threads

Question

This class acts like a synchronization context except the work could be done on any one of the available threads.

I've done quite a bit testing on my own, but I'm generally concerned about potential deadlocks. (I'm also curious if there are better ways to accomplish my goal.)

I created it because I have a 3rd party library that can only be executed on threads that have been setup to execute, this setup and tear down is expensive (and if not done, causes a resource leak). The 3rd party library accessed via a remoted object, so I cannot control the thread that requests appear on. A recent change made to the software made the resource leak apparent, and now it needs to be fixed... (previously it was leaking to slow to notice)

public delegate void WorkerThreadTask();

public sealed class MultiThreadWorker
{
    private const int WORKER_THREAD_BUSY = 1;
    private const int WORKER_THREAD_FREE = 0;

    public const int MaxThreads = 8;

    private string m_Name;
    private int m_ThreadCount;
    private SemaphoreSlim m_Semaphore;
    private int[] m_ThreadStates;
    private WorkerThreadData[] m_Threads = null;
    private bool m_Started;
    private object m_StartStopSync = new object();

    public MultiThreadWorker(int threadCount, string name)
    {
        if (threadCount > MaxThreads)
        {
            throw new ArgumentOutOfRangeException("threadCount", threadCount, "threadCount cannot be greater than MaxThreads.");
        }

        m_ThreadCount = threadCount;
        m_Name = name;

        m_Threads = new WorkerThreadData[threadCount];
        m_ThreadStates = new int[threadCount];

        for (int threadIdx = 0; threadIdx < threadCount; threadIdx++)
        {
            string threadName = String.Empty;
            if (name != String.Empty)
            {
                threadName = String.Format("{0}_{1}", name, threadIdx);
            }

            m_Threads[threadIdx] = new WorkerThreadData(threadName);

            m_Threads[threadIdx].InitializeThread += OnInitializeThread;
            m_Threads[threadIdx].ThreadTerminating += OnThreadTerminating;

            m_ThreadStates[threadIdx] = 0;
        }

        m_Semaphore = new SemaphoreSlim(0, threadCount);
        m_Started = false;
    }

    public bool Started
    {
        get
        {
            return m_Started;
        }
    }

    public void Start()
    {
        lock (m_StartStopSync)
        {
            if (!m_Started)
            {
                m_Started = true;

                for (int threadIdx = 0; threadIdx < m_ThreadCount; threadIdx++)
                {
                    m_Threads[threadIdx].Start();
                }

                m_Semaphore.Release(m_ThreadCount);
            }
        }
    }

    public void Stop()
    {
        lock (m_StartStopSync)
        {
            if (m_Started)
            {
                // Wait until all pending work is complete.
                for (int i = 0; i < m_ThreadCount; i++)
                {
                    m_Semaphore.Wait();
                }

                m_Started = false;

                // Stop the child threads.
                for (int i = 0; i < m_ThreadCount; i++)
                {
                    m_Threads[i].Stop();
                }
            }
        }
    }

    public void Invoke(WorkerThreadTask work)
    {
        m_Semaphore.Wait();
        try
        {
            if (!Started)
            {
                throw new InvalidOperationException("Cannot do work on an unstarted worker.");
            }

            // Pick a thread
            int threadIndex = -1;
            for (int tIndex = 0; tIndex < m_ThreadCount; tIndex++)
            {
                // Atomicly set thread state to 1 if it is currently 0.
                int threadState = Interlocked.CompareExchange(ref m_ThreadStates[tIndex], WORKER_THREAD_BUSY, WORKER_THREAD_FREE);
                if (threadState == WORKER_THREAD_FREE)
                {
                    threadIndex = tIndex;
                    break;
                }
            }

            Debug.Assert(threadIndex >= 0);

            m_Threads[threadIndex].Invoke(work);

            Interlocked.Exchange(ref m_ThreadStates[threadIndex], WORKER_THREAD_FREE);
        }
        finally
        {
            m_Semaphore.Release();
        }
    }

    private event EventHandler<EventArgs> m_InitializeThread;
    public event EventHandler<EventArgs> InitializeThread
    {
        add
        {
            m_InitializeThread += value;
        }
        remove
        {
            m_InitializeThread -= value;
        }
    }

    private void OnInitializeThread(object sender, EventArgs e)
    {
        var handlers = m_InitializeThread;
        if (handlers != null)
        {
            handlers(this, e);
        }
    }

    private event EventHandler<EventArgs> m_ThreadTerminating;
    public event EventHandler<EventArgs> ThreadTerminating
    {
        add
        {
            m_ThreadTerminating += value;
        }
        remove
        {
            m_ThreadTerminating -= value;
        }
    }

    private void OnThreadTerminating(object sender, EventArgs e)
    {
        var handlers = m_ThreadTerminating;
        if (handlers != null)
        {
            handlers(this, e);
        }
    }

    public int[] Counts
    {
        get
        {
            return m_Threads.Select(t => t.ExecutionCount).ToArray();
        }
    }

    public override string ToString()
    {
        return "MultiThreadWorker";
    }
}

internal class WorkerThreadData
{
    bool m_Started = false;
    string m_Name = String.Empty;
    Thread m_Thread = null;
    Object m_ThreadSync = new object();
    int m_ExecutionCount = 0;
    WorkerThreadTask m_Work = null;

    public WorkerThreadData(string name)
    {
        m_Name = name;
    }

    public void Start()
    {
        if (!m_Started)
        {
            m_Started = true;
            Debug.Assert(m_Thread == null);
            m_Thread = new Thread(ThreadLoop);
            m_Thread.IsBackground = true;
            if (m_Name != String.Empty)
            {
                m_Thread.Name = m_Name;
            }
            m_Thread.Start();
        }
    }

    public void Stop()
    {
        if (m_Started)
        {
            m_Started = false;
            Invoke(() => { });
            m_Thread.Join();
            m_Thread = null;
        }
    }

    public void Invoke(WorkerThreadTask work)
    {
        lock (m_ThreadSync)
        {
            Debug.Assert(m_Work == null);
            System.Threading.Monitor.Pulse(m_ThreadSync);
            m_Work = work;
            System.Threading.Monitor.Wait(m_ThreadSync);
            Debug.Assert(m_Work == null);
        }
    }

    private void ThreadLoop()
    {
        OnInitializeThread();

        lock (m_ThreadSync)
        {
            while (m_Started)
            {
                System.Threading.Monitor.Wait(m_ThreadSync);
                if (m_Work != null)
                {
                    m_Work();
                    m_Work = null;
                }
                m_ExecutionCount++;
                System.Threading.Monitor.Pulse(m_ThreadSync);
            }
        }

        OnThreadTerminating();
    }

    private event EventHandler<EventArgs> m_InitializeThread;
    public event EventHandler<EventArgs> InitializeThread
    {
        add
        {
            m_InitializeThread += value;
        }
        remove
        {
            m_InitializeThread -= value;
        }
    }

    private void OnInitializeThread()
    {
        var handlers = m_InitializeThread;
        if (handlers != null)
        {
            handlers(this, e);
        }
    }

    private event EventHandler<EventArgs> m_ThreadTerminating;
    public event EventHandler<EventArgs> ThreadTerminating
    {
        add
        {
            m_ThreadTerminating += value;
        }
        remove
        {
            m_ThreadTerminating -= value;
        }
    }

    private void OnThreadTerminating()
    {
        var handlers = m_ThreadTerminating;
        if (handlers != null)
        {
            handlers(this, e);
        }
    }

    public int ExecutionCount
    {
        get
        {
            return m_ExecutionCount;
        }
    }
}

I intend to use the class like this (the actual class has 90 methods):

public class RemotedWrapperObject
{
    // Initialization not shown.
    private MultiThreadWorker m_Worker; 
    private Some3rdPartyLibrary m_Instance;

    public void Initialize()
    {
        m_Worker = new MultiThreadWorker(4, "LibraryThread");
        m_Worker.Start();
        m_Worker.InitializeThread += m_Worker3_InitializeThread;
        m_Worker.ThreadTerminating += m_Worker3_ThreadTerminating;
    }

    void m_Worker_ThreadTerminating(object sender, EventArgs e)
    {
        m_Instance.ThreadCleanup();
    }

    void m_Worker_InitializeThread(object sender, EventArgs e)
    {
        m_Instance.ThreadSetup();
    }

    public bool SomeMethod(int param1, int param2)
    {
        bool retValue = false;

        m_Worker.Invoke(() =>
        {
            retValue = m_Instance.SomeMethod(param1, param2);
        });

        return retValue;
    }

    public void SomeResult(int param1, out int param2)
    {
        int param2Out = 0;

        m_Worker.Invoke(() =>
        {
            m_Instance.SomeResult(param1, out param2Out);
        });

        param2 = param2Out;
    }
}

A couple of notes: 1. Most of the code I am modifying is very old this MultiThreadWorker is new. (i.e. the remoting portions cannot be changed at this stage) 2. I've removed all the xmldoc comments above the methods to save space.

Answer 1

Some initial thoughts about your code (I would recommend ReSharper to detect most of them):

• don't use Hungarian notation (m_VariableName)
• in most cases you don't need to define events explicitly (with add/remove)
• do use readonly where possible
• no need to initialise fields with their default values (bool _b = false, string[] _arr = null)

But the main issue is that you're trying to do all the synchronisation yourselves instead of using support from the .NET framework. Your task is a typical producer/consumer problem where producers generate workload (request some work to be done on one of the threads), and consumers are the threads that execute that work. Producer and consumer communicate with each other via queue, and for multithreaded environment .NET offers ConcurrentQueue<T>. Since we need to wait for new items we'll use BlockingCollection<T> (with ConcurrentQueue<T> under the hood).

public sealed class MultiThreadWorker
{
    public const int MaxThreads = 8;

    private readonly object _startStopSync = new object();
    private readonly int _threadCount;
    private Task[] _tasks;
    private bool _started;
    private CancellationTokenSource _cancellationTokenSource;

    private readonly BlockingCollection<AsyncJob> _queue = new BlockingCollection<AsyncJob>();

    public event EventHandler InitializeThread;
    public event EventHandler ThreadTerminating;

    public MultiThreadWorker(int threadCount)
    {
        if (threadCount > MaxThreads)
            throw new ArgumentOutOfRangeException("threadCount", threadCount, "threadCount cannot be greater than MaxThreads.");

        _threadCount = threadCount;
    }

    public bool Started { get { return _started; } }

    private static void RunAsyncJob(AsyncJob asyncJob)
    {
        try
        {
            asyncJob.Job();
            asyncJob.TaskCompletionSource.SetResult(null); //notifying that task has completed
        }
        catch (Exception ex)
        {
            asyncJob.TaskCompletionSource.SetException(ex); //notifying that task has failed with exception
        }
    }

    private bool TryDequeue(CancellationToken cancellationToken, out AsyncJob asyncJob)
    {
        try
        {
            asyncJob = _queue.Take(cancellationToken);
            return true;
        }
        catch (OperationCanceledException)
        {
            asyncJob = null;
            return false;
        }
    }

    private void MainTaskCycle(CancellationToken cancellationToken)
    {
        OnInitializeThread();

        AsyncJob asyncJob;
        while (!cancellationToken.IsCancellationRequested && TryDequeue(cancellationToken, out asyncJob))
        {
            RunAsyncJob(asyncJob);
        }

        OnThreadTerminating();
    }

    private void OnInitializeThread()
    {
        var handlers = InitializeThread;
        if (handlers != null)
            handlers(this, null);
    }

    private void OnThreadTerminating()
    {
        var handlers = ThreadTerminating;
        if (handlers != null)
            handlers(this, null);
    }

    public void Start()
    {
        if (_started)
            return;

        lock (_startStopSync)
        {
            if (_started)
                return;

            _started = true;

            _cancellationTokenSource = new CancellationTokenSource();

            _tasks = Enumerable.Range(0, _threadCount)
                .Select(i => Task.Factory.StartNew(() => MainTaskCycle(_cancellationTokenSource.Token), _cancellationTokenSource.Token)).ToArray();
        }
    }

    public void Stop()
    {
        if (!_started)
            return;

        lock (_startStopSync)
        {
            if (!_started)
                return;

            _cancellationTokenSource.Cancel();
            Task.WaitAll(_tasks);
            _started = false;
        }
    }

    public Task InvokeAsync(Action action)
    {
        var completionSource = new TaskCompletionSource<object>();
        _queue.Add(new AsyncJob { Job = action, TaskCompletionSource = completionSource });
        return completionSource.Task;
    }

    private class AsyncJob
    {
        public TaskCompletionSource<object> TaskCompletionSource { get; set; }
        public Action Job { get; set; }
    }
}

And the usage will look like this (note that now all exceptions from third-party library will be passed through as if the code ran on the same thread)

public class RemotedWrapperObject
{
    // Initialization not shown.
    private MultiThreadWorker _worker;
    private Some3rdPartyLibrary _instance;

    public void Initialize()
    {
        _worker = new MultiThreadWorker(4);
        _worker.InitializeThread += ThreadSetup;
        _worker.ThreadTerminating += ThreadCleanup;
        _worker.Start();
    }

    void ThreadSetup(object sender, EventArgs e)
    {
        _instance.ThreadSetup();
    }

    void ThreadCleanup(object sender, EventArgs e)
    {
        _instance.ThreadCleanup();
    }

    public bool SomeMethod(int param1, int param2)
    {
        bool retValue = false;

        _worker.InvokeAsync(() =>
        {
            retValue = _instance.SomeMethod(param1, param2);
        }).Wait();

        return retValue;
    }

    public void SomeResult(int param1, out int param2)
    {
        int param2Out = 0;

        _worker.InvokeAsync(() =>
        {
            _instance.SomeResult(param1, out param2Out);
        }).Wait();

        param2 = param2Out;
    }
}

Answer 2

It looks to me like you've reinvented the thread pool. What exactly is this giving you that the thread pool doesn't have? I see you've got events for termination and whatnot, but if you use TPL you can easily emulate everything you've written.

One thing I would like to point out is that the way you fire events isn't thread-safe. Don't do this:

if(m_foo != null) m_foo(this, e);

You should instead do this:

var foo = m_foo;
if(foo != null) foo(this, e);

Why? Because m_foo could potentially become null between your conditional and your invocation.

Also you said you have 90 methods on a class? That's a code smell right there. It sounds like it's doing too much.