6
\$\begingroup\$

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.

enter image description here

  • 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:

enter image description here

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)

\$\endgroup\$
  • \$\begingroup\$ I think you don't have ReSharper installed and screaming at your private field naming convention! Prefixing them with an underscore really helps readability I find - I was WTF-looking for a local variable when I came across jobInstance. Declaring them at the top of the class also helps, especially if you're not prefixing them and you're using them before they're declared. Call that a "readability review" :) Is JobConsumerLog a static class? I'd just replace that with a constructor-injected ILogger, otherwise you've got ambient context here, not obvious the thing is logging anything. \$\endgroup\$ – Mathieu Guindon Oct 1 '13 at 23:45
  • 1
    \$\begingroup\$ Have you considered Quartz.NET, it contains a lot of job running logic out of the box and also retry policies etc. It is more of an scheduling framework but in your case you could it set it to run jobs as fast as possible (fire now). \$\endgroup\$ – Marko Lahma Oct 3 '13 at 5:51
  • \$\begingroup\$ You might want to look into using something like Microsoft HPC if you aren't set on rolling your own solution, as that seems to be what you need. \$\endgroup\$ – krolley Nov 7 '14 at 0:00
  • \$\begingroup\$ Have you consider using Windows Workflow Foundation (WFF) for this? \$\endgroup\$ – Tien Dinh Dec 27 '14 at 17:55
3
\$\begingroup\$

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.
\$\endgroup\$
  • \$\begingroup\$ Adding to your second point: This should be a direct cast rather than "as". \$\endgroup\$ – Roman Reiner Nov 5 '15 at 19:12

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.