Parallel Job Consumer

Question

Edit: A second iteration of this problem here.

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. They are generally long-running and are not suitable to run on the web server.

Important factors

The solution needs to be a very extensible and maintainable solution, as new job implementations will be created by developers all the time.
The execution of these jobs is to be abstracted away from the client code. All developers need to do is to create the new job and add it to the job data source (database, MSMQ, Azure Queue, etc).

Solution

Deriving from AbstractJob and overriding the abstract Execute() method to provide implementation for the new job:

public abstract class AbstractJob { public JobState State { get; protected set; } public AbstractJob() { State = JobState.New; } protected abstract bool Execute(); public sealed void ExecuteJob() { State = JobState.Executing; try { if (Execute()) { State = JobState.Successful; } else { State = JobState.Failed; } } catch (Exception) { State = JobState.FailedOnException; } } } public class TestJob : AbstractJob { protected override bool Execute() { Debug.WriteLine("Test job!"); return true; } }

The following JobConsumer would reside on a Windows Service or Azure Worker Role. IJobSource provides the consumer with jobs. These could be from the database, a MSMQ, an Azure Queue, etc (I have left this implementation out).

Every 5 seconds (this can obviously be configurable), the JobConsumer executes new jobs when there are available cores.

public class JobConsumer { public List<AbstractJob> Jobs { get; set; } public IJobSource JobSource { get; private set; } public int ParallelExecutionCount { get; set; } protected List<Task> Tasks { get; set; } public JobConsumer(IJobSource jobSource, int parallelExecutionCount) { JobSource = jobSource; ParallelExecutionCount = parallelExecutionCount; } public void StartConsumer() { Task.Factory.StartNew(() => { TaskScheduler currentScheduler = TaskScheduler.Current; while (true) { int availableCores = ParallelExecutionCount - Tasks.Count(); IEnumerable<AbstractJob> newJobs = JobSource.ReadNewJobs().Take(availableCores); foreach (var job in newJobs) { System.Threading.Tasks.Task task = null; task = Task.Factory.StartNew(() => { Tasks.Add(task); }, CancellationToken.None, TaskCreationOptions.None, currentScheduler) .ContinueWith(t => { // Execute job on a new thread job.ExecuteJob(); }) .ContinueWith(t => { Tasks.Remove(task); }, currentScheduler); } Thread.Sleep(5000); } }); } } public interface IJobSource { public IEnumerable<AbstractJob> ReadNewJobs(); }

Input, critique and suggestions very welcome!

EDIT: I am using .NET 4.5

I wrote similar code polling a serial port and used the same looping methods you've used. Except I added a timeout error count, which would reset every time it connected successfully. I don't know if you want to count your job errors, to create some system flag?? — user90823
– user90823, Commented Sep 27, 2013 at 15:16
Wouldn't it actually be easier to convert the Execute to ExecuteAsync returning a Task and the AbstractJob to IJob? Task carries everything that you put in the JobState flag and more (e.g. if it failed, why). Then you could also drop the task creation on consumer side and use either ConcurrentQueue<T> or BlockingCollection<T> to process the jobs, but that depends on .NET version. — Patryk Ćwiek
– Patryk Ćwiek, Commented Sep 27, 2013 at 15:42
@PatrykĆwiek I think that unless explicitly mentioned, you can safely assume at least .Net 4.0. — svick
– svick, Commented Sep 27, 2013 at 16:49

svick · Accepted Answer · 2013-09-27 19:33:04Z

The way you're using TaskScheduler.Current to try to synchronize access to Tasks won't work (unless this code runs under a very specific custom TaskScheduler).

With the default scheduler, a Task can be executed on any ThreadPool thread, which means that multiple threads could access Tasks at the same time. The simplest fix for that is to use lock instead of trying to use TaskScheduler (which will also simplify your code a bit).

Since you seem to be using Tasks just for counting, another option would be to replace it with int field and then use Interlocked.Increment()/Decrement() to modify it.
I think you should save the exception that's caught in ExecuteJob() somewhere, so that when you encounter an issue, you can actually diagnose it.

Also, it looks like you're not saving the results of jobs anywhere. I think that's a bad idea, you should have some sort of logging, to see where the problem was when something doesn't work right. (Was the job even scheduled? Did it complete successfully? Etc.)

And I'm not sure you want to differentiate between "failed" and "failed on exception". When something fails, you usually want to know why it failed, which is what exceptions are for.
Why are most of the properties of JobConsumer public? If you want to allow adding a new job to Jobs, but nothing else, then create a method for that.

And why is Tasks protected, when you don't have any virtual methods? The default accessibility should be private (which means you don't need a property anymore and you can use just a field) and you should change that only when you have good reason.
The way you use ReadNewJobs() seems to indicate that items are actually consumed only when the enumerable is iterated. That's not a bad idea (BlockingCollection.GetConsumingEnumerable() does the same thing), but it should be documented very clearly that that's what is expected from the method.
Since you're on .Net 4.5, you can use await Task.Delay() instead of Thread.Sleep(). This means you would need to make the lambda async and you should probably also use Task.Run() instead of Task.Factory.StartNew() (which is a good idea anyway, because it's shorter).
There is no need to spell the name System.Threading.Tasks.Task in full, use just Task.
I don't think that the name AbstractJob needs to emphasize that the class is abstract, calling it just Job would work just as well. Another possible name is JobBase.

Thanks for the great reply. You have made some really helpful point. I have a question. Would you use an abstract Job class or an IJob interface for implementation of new jobs? I feel that an IJob is the more correct choice, but I really want the job to have a JobState. — Dave New
– Dave New, Commented Sep 30, 2013 at 6:54
@davenewza You could always do both. But I think that abstract class makes more sense here, because of ExecuteJob() (if you used an interface, each implementation would have to repeat the code for that method). — svick
– svick, Commented Sep 30, 2013 at 8:45
I have created a second iteration of the above code (and have implemented a lot of your comments). Your feedback/review would be awesome: post here. — Dave New
– Dave New, Commented Oct 1, 2013 at 14:49

Stack Exchange Network

Parallel Job Consumer

1 Answer 1

You must log in to answer this question.

Linked

Hot Network Questions

Parallel Job Consumer

1 Answer 1

You must log in to answer this question.

Linked

Related

Hot Network Questions