Is this Agent/Actor implementation issue free?

Question

I implemented this Agent class for a project recently and was wondering if I could get some other eyes to look at it -- I'm currently the only developer where I work so I can't exactly ask someone here to do it.

I'm pretty sure it's correct, but then I'm too close to it.

public abstract class Agent<M> : IDisposable
{
    private Thread thread;
    private Queue<M> messageQueue;
    private bool quit;

    public Agent()
    {
        this.thread = new Thread(new ThreadStart(AgentThreadBody));
        this.messageQueue = new Queue<M>();
        this.quit = false;

        this.thread.Start();
    }

    /// <summary>
    /// Stops the message-handling thread.
    /// 
    /// Do not call this method from within the message-
    /// handling method, or it will result in a deadlock
    /// (because this method waits for the message-handling
    /// thread to stop).
    /// </summary>
    public virtual void Dispose()
    {
        this.quit = true;
        this.thread.Interrupt();
        this.thread.Join();
        this.thread = null;
        // clear messageQueue before nulling?
        // (would do this to dispose queued items)
        this.messageQueue = null;
    }

    public void QueueMessage(M message)
    {
        lock (this.messageQueue)
        {
            this.messageQueue.Enqueue(message);
            this.thread.Interrupt();
        }
    }

    private void AgentThreadBody()
    {
        while (!this.quit)
        {
            M message = default(M);
            bool messageAvailable = false;

            lock (messageQueue)
            {
                if (messageQueue.Count > 0)
                {
                    message = messageQueue.Dequeue();
                    messageAvailable = true;
                }
            }

            try
            {
                if (!messageAvailable)
                {
                    // if the Interrupt() method was
                    // called before we sleep or is
                    // called while we're sleeping,
                    // this will throw:
                    // ThreadInterruptedException
                    Thread.Sleep(Timeout.Infinite);
                }
            }
            catch (ThreadInterruptedException)
            {
                // we have a new message to handle,
                // so get it, or we've been told
                // to quit.
                continue;
            }

            ProcessMessage(message);
        }
    }

    protected abstract void ProcessMessage(M message);
}

Also, are there any special considerations you can think of that should be made by a class inheriting from this base class? (I can't think of any.)

I could add start/stop methods, but at the moment they're not needed.

Btw, I have to use .NET 3.5.

Answer 1

Instead of sleeping for short periods or using Thread.Interrupt() it's better to use waithandles, ManualResetEvent in your case. I don't have Visual Studio at hand, but the following example should give you the basic idea (most of the complexity will be gone if you start using .NET 4 or 4.5, in particular BlockingCollection<T>):

public abstract class Agent<TMessage> : IDisposable
{
    private readonly Thread _thread = new Thread(AgentThreadBody);
    private readonly Queue<TMessage> _messageQueue = new Queue<TMessage>();
    private readonly ManualResetEvent _waitHandle = new ManualResetEvent(false);
    private volatile bool _quit;
    private bool _disposed;

    protected Agent()
    {
        _thread.Start();
    }

    public void QueueMessage(M message)
    {
        lock (_messageQueue)
        {
            _messageQueue.Enqueue(message);
            _waitHandle.Set();
        }
    }

    private void AgentThreadBody()
    {
        while (!_quit)
        {
            _waitHandle.WaitOne();
            TMessage message;

            lock (messageQueue)
            {
                if (messageQueue.Count == 0)
                {
                    _waitHandle.Reset();
                    continue;
                }

                message = messageQueue.Dequeue();
            }

            ProcessMessage(message);
        }
    }

    protected abstract void ProcessMessage(M message);

    /// <summary>
    /// Stops the message-handling thread.
    /// 
    /// Do not call this method from within the message-
    /// handling method, or it will result in a deadlock
    /// (because this method waits for the message-handling
    /// thread to stop).
    /// </summary>
    public void Dispose()
    {
        Dispose(true);
        GC.SupressFinalize(this);
    }

    protected virtual void Dispose(bool disposing)
    {
        if (_disposed || !disposing)
            return;

        _quit = true;
        _waitHandle.Set();
        _thread.Join();
        _waitHandle.Dispose();
        _disposed = true;
    }
}

Answer 2

Few minor points:

• Generic type parameter should be called <T> as per convention.
• public constructors in abstract classes make no sense - make it protected.
• Make thread and messageQueue readonly and don't set them to null in the Dispose() method.
• Better yet, implement the Disposable Pattern correctly. Though note, I do disagree with their setting of the IDisposable members to null. It's really not necessary at all; but calling Dispose() is. Their _disposed class member conveys enough information necessary.
• You should also lock on a dedicated locking class member when accessing your quit class member since it is accessed via multiple threads.
• Possibly wrap your call to ProcessMessage(message) in a try..catch block with appropriate error handling (or not, as the case could be). Unless, of course, you trust any subclasses from doing nasty things in their override of it.

Just my initial thoughts. Hope they help.

Answer 3

No, this implementation is not issue free!

The Thread.Interrupt() method apparently causes a ThreadInterruptedException to be thrown whenever the thread is blocked. The documentation led me to believe that it was only thrown under certain circumstances:

If this thread is not currently blocked in a wait, sleep, or join state, [emph mine] it will be interrupted when it next begins to block.

However, a lot more can block a thread than you'd think! I first started seeing what I thought were rogue instances of this exception in a derived agent instance when accessing DateTime.Now! (This is apparently due to the Now property requiring blocking IO to obtain a value.)

Locking can also cause a thread to block, as lock gets translated into Monitors (which block).

Looks like I'll have to go with a solution that just Sleep()s for short periods of time instead of waiting indefinitely until new input became available :-/