Parallel Job Consumer using TPL

Question

I need to provide a service (either a Windows Service or an Azure Worker Role) which will handle the parallel execution of jobs. These jobs could be anything from importing data to compiling of reports, or to sending out mass notifications, etc.

Database Design:

The whole process of running these jobs is persistent.

• JobDefinition stores varchar reference to the IJob concrete type class.
• JobInstance is an instance of JobDefinition which needs to be executed as a job.
• JobInstanceEvent stores the process of executing an instance (the change of states).

Class Structure:

In order to create new jobs the simple IJob interface needs to be implemented:

public interface IJob
{
    bool Execute();
}

JobHandler is responsible for deserializing the serialized varchar in the database and also saving state changes:

public class JobHandler
{
    public string State 
    {
        get
        {
            return jobInstance.State;
        }
        set
        {
            using (MyEntities context = new MyEntities())
            {
                context.JobInstances.Attach(jobInstance);
                jobInstance.State = value;

                context.SaveChanges();
            }

            SaveJobEvent(String.Empty);
        }
    }

    private JobInstance jobInstance;

    public JobHandler(JobInstance job)
    {
        jobInstance = job;
    }

    public IJob CreateJobObject()
    {
        Type type = Type.GetType(jobInstance.JobDefinition.FullAssemblyType);

        XmlSerializer serializer = new XmlSerializer(type);
        using (StringReader reader = new StringReader(jobInstance.SerialisedObject))
        {
            return serializer.Deserialize(reader) as IJob;
        }
    }

    public void SaveJobEvent(string jobEventMessage)
    {
        using (MyEntities context = new MyEntities())
        {
            context.JobInstances.Attach(jobInstance);

            jobInstance.JobInstanceEvents.Add(
                new JobInstanceEvent() 
                { 
                    State = jobInstance.State,
                    Date = DateTime.UtcNow,
                    Message = jobEventMessage
                }
            );

            context.SaveChanges();
        }
    }
}

The JobConsumer is what will run in the service. It performs two primary parallel tasks:

• Adding items to the blocking queue (through polling)
• Executing items from the blocking queue (through thread blocking)

It is able to run multiple jobs concurrently (ParallelExecutionSlots).

public class JobConsumer
{
    public IJobSource JobSource { get; private set; }

    public BlockingCollection<JobHandler> JobQueue { get; private set; }

    public int ParallelExecutionSlots { get; private set; }

    private bool isPopulating;

    private CancellationTokenSource consumerCancellationTokenSource;

    public JobConsumer(IJobSource jobSource, int parallelExecutionSlots)
    {
        JobQueue = new BlockingCollection<JobHandler>();

        JobSource = jobSource;

        ParallelExecutionSlots = parallelExecutionSlots;
        consumerCancellationTokenSource = new CancellationTokenSource();

        StartPopulatingQueue();
    }

    public void StartPopulatingQueue()
    {
        StartPopulatingQueue(500);
    }

    public void StartPopulatingQueue(int pollingDelay)
    {
        if (!isPopulating)
        {
            isPopulating = true;

            var task = Task.Factory.StartNew(() =>
            {
                while (isPopulating)
                {
                    foreach (var job in JobSource.ReadNewJobs())
                    {
                        job.State = "Queued";

                        JobQueue.Add(job);
                    }

                    Tasks.Task.Delay(pollingDelay).Wait();
                }
            })
            .ContinueWith(t =>
            {
                isPopulating = false;
                JobConsumerLog.LogException("JobConsumer populating process failed", t.Exception);
            },
                CancellationToken.None,
                TaskContinuationOptions.OnlyOnFaulted,
                TaskScheduler.Current
            );
        }
    }

    public void StopPopulatingQueue()
    {
        isPopulating = false;
    }

    public void StartConsumer()
    {
        Task.Factory.StartNew(() =>
        {
            SemaphoreSlim slotsSemaphore = new SemaphoreSlim(ParallelExecutionSlots);

            foreach (var job in JobQueue.GetConsumingEnumerable(consumerCancellationTokenSource.Token))
            {
                slotsSemaphore.Wait();

                JobHandler jobHandler = job;

                Task.Factory.StartNew(() =>
                {
                    try
                    {
                        ExecuteJob(jobHandler);
                    }
                    finally
                    {
                        slotsSemaphore.Release();
                    }
                },
                    TaskCreationOptions.LongRunning
                );
            }

            consumerCancellationTokenSource.Token.ThrowIfCancellationRequested();

        }, consumerCancellationTokenSource.Token)
        .ContinueWith(t =>
        {
            JobConsumerLog.LogException("JobConsumer execution process failed", t.Exception);
        },
            CancellationToken.None,
            TaskContinuationOptions.OnlyOnFaulted,
            TaskScheduler.Current
        )
        .ContinueWith(t =>
        {
            JobConsumerLog.LogException("JobConsumer execution process stopped", t.Exception);
        },
            CancellationToken.None,
            TaskContinuationOptions.OnlyOnCanceled,
            TaskScheduler.Current
        );
    }

    public void StopConsumer()
    {
        consumerCancellationTokenSource.Cancel();
    }

    private void ExecuteJob(JobHandler job)
    {
        job.State = "Executing";

        try
        {
            if (job.CreateJobObject().Execute())
            {
                job.State = "Successful";
            }
            else
            {
                job.State = "Failed";
            }
        }
        catch (Exception ex)
        {
            job.State = "FailedOnException";
        }
    }
}

Feedback on the overall design would be much appreciated. I'm looking for a maintainable and extensible solution.

I am using .NET 4.5

(Note: this is a second iteration of the following post)

Answer 1

Only some few minor things:

1. In StartPopulatingQueue to me as the reader it's unclear in what unit pollingDelay is in. Two classic solutions:
   • Change it to be a TimeSpan - I usually prefer this since it provides the most flexibility.
   • Add the unit suffix to the parameter name. like Ms for milliseconds or Sec for seconds etc.
2. CreateJobObject can potentially return null if the deserialized object cannot be cast to IJob in which case ExecuteJob will throw a NullReferenceException which is typically not very meaningful. You should throw a more meaningful exception on deserialization (as mentioned in the comment by Roman you could use a direct cast instead of as which would throw an InvalidCastException)
3. It feels wrong to me that the consumer defines the job execution states. The consumer seems largely responsible for managing the job queue and it should stick to that single responsibility. Hence I think Execute should be moved into JobHandler.
4. I'd also consider moving CreateJobObject to JobInstance so JobInstance is responsible to provide the actual object instance. Then JobHandler is just concerned with the state changes of the job.