6
\$\begingroup\$

I've created a simple asynchronous execution engine and I would like some help identifying places it can be improved. The basic function of the system is to monitor MSMQ and Service Broker queues and execute any tasks that come through on them.

Here is the code:

using System;
using System.Collections.Generic;
using System.Linq;
using Castle.Windsor;
using ThreadingTasks = System.Threading.Tasks;

public class Engine
{
    /// <summary>
    /// Start's the Engine and spins up appropriate message listeners
    /// </summary>
    public void Start()
    {
        _logger.Info("Starting the Engine, calling GetAsync for all queues");
        _cancelToken = false;
        GetAsync(AsyncLaunch.All);
    }

    /// <summary>
    /// Sends the stop message to the engine which will let currently running tasks finish
    /// and stop future tasks from being picked up. 
    /// </summary>
    public void Stop()
    {
        _logger.Info("Stopping the Engine");
        _cancelToken = true;
    }

    protected enum AsyncLaunch { All, MSMQ, ServiceBroker };

    /// <summary>
    /// Starts all appropriate asynchronous Message listeners
    /// </summary>
    /// <param name="async"></param>
    protected void GetAsync(AsyncLaunch async)
    {
        if (_cancelToken)
            return;
        _logger.Debug("Attempting to get messages to process Asynchronously");
        if (SingleThreaded)
        {
            _logger.Debug("Engine is running in single-threaded mode and must wait till all tasks finish");
            ThreadingTasks.Task.WaitAll(_threadTasks.ToArray());
            _threadTasks.Clear();
        }

        if (async == AsyncLaunch.ServiceBroker || async == AsyncLaunch.All)
            _serviceBroker.GetAsync(x => RunServiceBrokerJob(x));

        if (async == AsyncLaunch.MSMQ || async == AsyncLaunch.All)
            _msmq.GetManyAsync(x => RunMSMQJob(x));
    }

    /// <summary>
    /// Runs the job for the given message and after the job has launched, calls 
    /// GetAsync to get the next ServiceBroker message
    /// </summary>
    protected void RunServiceBrokerJob(TransportMessage msg)
    {
        RunJob(msg);
        GetAsync(AsyncLaunch.ServiceBroker);
    }

    /// <summary>
    /// Runs the job for the given collection of messages and after the job has 
    /// launched, calls GetAsync to get the next MSMQ message
    /// </summary>
    protected void RunMSMQJob(IEnumerable<TransportMessage> msgs)
    {
        RunJob(msgs);
        GetAsync(AsyncLaunch.MSMQ);
    }


    /// <summary>
    /// Initializes and executes the appropriate job for the given collection of messages
    /// </summary>
    /// <param name="msgs">A collection of messages passed into a Job</param>
    protected ThreadingTasks.Task RunJob(IEnumerable<TransportMessage> msgs)
    {
        try
        {
            var task = ThreadingTasks.Task.Factory.StartNew(() =>
            {
                try
                {
                    _eventSink.Broadcast(msgs);
                }
                catch (Exception ex)
                {
                    _logger.Error(ex, "Error when attempting to run jobs {0} using message types {1} with values {2}", string.Join(",", msgs.Select(x => x.Id).ToArray()), string.Join(",", msgs.Select(x => x.Name).Distinct().ToArray()), string.Join(",",msgs.Select(x => x.ToString())));
                }

            });

            if (SingleThreaded)
                _threadTasks.Add(task);

            return task; 
        }
        catch (Exception ex)
        {
            _logger.Error(ex, "Error when attempting to invoke Task for a collection of {0} messages with ids {1} of type(s) {2}", msgs.Count(), string.Join(",", msgs.Select(x => x.Id).ToArray()), string.Join(",", msgs.Select(x => x.Name).Distinct().ToArray()));
        }

        return null; 
    }

    /// <summary>
    /// Initializes and executes the appropriate job for the given message
    /// </summary>
    /// <param name="msg">Message containing information pertinent to the executed Job</param>
    protected ThreadingTasks.Task RunJob(TransportMessage msg)
    {
        _logger.Info("RunJob was invoked for msg id " + msg.Id);
        try
        {
            var task = ThreadingTasks.Task.Factory.StartNew(() =>
            {
                _logger.Debug("Attempting to Run Job with ID: " + msg.Id);
                try
                {
                    _eventSink.Broadcast(msg);
                }
                catch (Exception ex)
                {
                    _logger.Error(ex, "Error when attempting to run job {0} using message type {1} with values {2}", msg.Id, msg.GetType(), msg.ToString());
                }

            });

            if (SingleThreaded)
                _threadTasks.Add(task);

            return task; 
        }
        catch (Exception ex)
        {
            _logger.Error(ex, "Error when attempting to invoke Task for message with id {0} of type {1}", msg.Id, msg.GetType());
        }

        return null; 
    }

    private void ValidateObjectIsNotNull(object obj, string objectName)
    {
        if (obj == null) throw new ArgumentNullException(objectName + " can not be null");
    }

    /// <summary>
    /// Instantiate a new engine used to access messaging queues and launch Tasks
    /// </summary>
    public Engine(IWindsorContainer container, IEventSink eventSink, IEngineConfiguration config, ILogging logger, IDataAccess<MessageBase> serviceBroker, IDataAccess<TransportMessage> msmq)
    {
        _eventSink = eventSink;
        _config = config;
        _threadTasks = new List<ThreadingTasks.Task>();
        _logger = logger;
        _container = container;
        _serviceBroker = serviceBroker;
        _msmq = msmq;

        //ThreadCount 0 is unlimited threads
        if (SingleThreaded || _config.ThreadCount == 0)
            return;

        // We need at least 3 threads, otherwise on a single CPU box we are stuck with a single thread
        // and only two threads might stop one of the job pollers from running
        int minThreadCount = System.Math.Max(Environment.ProcessorCount, 3);

        // do not set maxThreadCount less than the minThreadCount
        int maxThreadCount = System.Math.Max(minThreadCount, _config.ThreadCount);

        System.Threading.ThreadPool.SetMaxThreads(maxThreadCount, maxThreadCount);
    }

    private readonly IEngineConfiguration _config;
    private readonly IEventSink _eventSink;
    private readonly IList<ThreadingTasks.Task> _threadTasks;
    private readonly ILogging _logger;
    private readonly IWindsorContainer _container;
    private readonly IDataAccess<MessageBase> _serviceBroker;
    private readonly IDataAccess<TransportMessage> _msmq;
    private bool _cancelToken;

    private bool SingleThreaded { get { return _config.ThreadCount == 1; } }
}

A couple of issues I have: - If the TPL Tasks blow up, I don't have a good way of getting that information - There is nothing that limits how "greedy" this engine is, meaning it will pull off Tasks even if it's extremely busy, possibly deriving other instances of this engine of work

Any suggestions are greatly appreciated, but those are the two primary areas of concern for me.

\$\endgroup\$
1
  • \$\begingroup\$ Didn't read the whole thing, but GetAsync() is not a good name for that method IMO. \$\endgroup\$ Oct 7 '13 at 5:19
3
\$\begingroup\$
  1. GetAsync is a bad name for that method. StartListening seems to express better what it does.

  2. AsyncLaunch is a bad name for that enum. It seems to describe which target should be listened to so why not call it ListenTarget.

  3. enums can be used as flags:

    enum ListenTarget
    {
        MSMQ          = 1 << 0,
        ServiceBroker = 1 << 1,
        All           = ~0,
    }
    

    Then your testing code becomes:

    if (async & AsyncLaunch.ServiceBroker)
        _serviceBroker.GetAsync(x => RunServiceBrokerJob(x));
    
    if (async & AsyncLaunch.MSMQ)
        _msmq.GetManyAsync(x => RunMSMQJob(x));
    

    plus you can pass in an arbitrary combination of targets (in case you add more later).

  4. ValidateIsNotNull should be renamed to state more clearly what it does: ThrowIfObjectIsNull.

  5. You have a lot of duplicate code in your two RunJob methods. The single message one could be shortened to one line of code RunJobs(new [] { msg }).

\$\endgroup\$
1
  • 1
    \$\begingroup\$ Also declare fields at the top, followed by constructor; had to scroll all the way down to check if _logger was readonly (kudos for private readonly fields). \$\endgroup\$ Oct 7 '13 at 10:42

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.