Queue with task parallel library

Question

I am new to programming using the .Net TPL Framework and multi-threading in general. I have googled up some tutorials and articles and put together the solution below.

It is a simple multithreaded queue and a scheduler that polls the queue and processes each item as a seperate task.

It would be great if you'll can comment on the solution and point out areas where I am doing things wrong or how it can be done more efficiently.

//Console app that starts the Queue and Scheduler
class Program
    {        
        static void Main(string[] args)
        {
            Console.WriteLine("Job Scheduler Started. Press Ctrl-C to end");

            Parallel.Invoke(
                () => clJobQueue.Instance.Populate(),
                () => clJobRunner.Instance.Start()
                );

            var autoResetEvent = new AutoResetEvent(false);
            Console.CancelKeyPress += (sender, eventArgs) =>
            {
                eventArgs.Cancel = true;
                autoResetEvent.Set();
            };

            autoResetEvent.WaitOne();

            Stop();
            Console.WriteLine("Job Scheduler Shutting Down");
            Console.ReadLine();
        }

        private static void Stop()
        {            
            Parallel.Invoke(
             () => clJobQueue.Instance.CloseQueue(),
             () => clJobRunner.Instance.Stop()
             ); 
        }
    }

//Singelton pattern Queue using a BlockingCollection.
//Contains a Populate method that polls a db and retreives new items
public sealed class clJobQueue
{
    private static volatile clJobQueue instance;
    private static volatile BlockingCollection<clJob> queue;
    private static volatile CancellationTokenSource cts;

    private static object syncRoot = new Object();

    static clJobQueue() { }
    private clJobQueue() { }

    public static clJobQueue Instance
    {
        get
        {
            if (instance == null)
            {
                lock (syncRoot)
                {
                    if (instance == null)
                    {
                        instance = new clJobQueue();
                        queue = new BlockingCollection<clJob>(10);
                        cts = new CancellationTokenSource();
                    }
                }
            }

            return instance;
        }
    }

    public bool Add(clJob jobItem)
    {
        queue.Add(jobItem, cts.Token);
        jobItem.setJobStatus(JobStatus.Queued);
        jobItem.JobQueue = this;
        return true;
    }

    public clJob GetNextJob()
    {
        try
        {
            clJob nextJob = queue.Take(cts.Token);
            nextJob.setJobStatus(JobStatus.Scheduled);
            return nextJob;
        }
        catch (Exception ex)
        {
            Console.WriteLine("Cancelling Queue");
        }
        return null;
    }

    public void CloseQueue()
    {
        cts.Cancel();
        Console.WriteLine("Closing Queue");
        while (!IsEmpty)
        {
            clJob job = queue.Take();
            job.setJobStatus(JobStatus.Scheduled);
        }
    }

    public int Count
    {
        get { return queue.Count; }
    }

    public bool IsEmpty
    {
        get { return Count <= 0; }
    }

    public void Populate()
    {
        Parallel.Invoke(
                () => {
                    while (!cts.IsCancellationRequested)
                    {
                        Console.WriteLine("Looking for X Job");
                        clJob job = clXJob.getJob();
                        if (job.JobID == 0)
                        {
                            Console.WriteLine("No X Jobs in DB, Sleeping for 60s");
                            Thread.Sleep(TimeSpan.FromSeconds(60));
                        }
                        else
                        {
                            Console.WriteLine("Adding X Job to Queue - {0} {1}", job.JobID, job.JobName);
                            Add(job);
                        }
                    }
                },
            () => {
                while (!cts.IsCancellationRequested)
                {
                    Console.WriteLine("Looking for Y Job");
                    clJob job = clYJob.getJob();
                    if (job == null || job.JobID == 0)
                    {
                        Console.WriteLine("No Y Jobs in DB, Sleeping for 60s");
                        Thread.Sleep(TimeSpan.FromSeconds(60));
                    }
                    else
                    {
                        Console.WriteLine("Adding Y Job to Queue - {0} {1}", job.JobID, job.JobName);
                        Add(job);
                    }
                }
            });
    }
}

//Polls the queue for items and initiates the items Run() method as a new Task
public sealed class clJobRunner
    {
        private static volatile clJobRunner instance;
        private static volatile CancellationTokenSource cts;

        static clJobRunner() { }
        private clJobRunner() { }

        public static clJobRunner Instance
        {
            get
            {
                if (instance == null)
                {
                    instance = new clJobRunner();
                    cts = new CancellationTokenSource();
                }

                return instance;
            }
        }

        public void Start()
        {
            Task.Run(() =>
            {
                while (!cts.IsCancellationRequested)
                {
                    clJob job;
                    if ((job = clJobQueue.Instance.GetNextJob()) == null)
                    {
                        Console.WriteLine("Queue Empty, Sleeping for 30s");
                        Thread.Sleep(TimeSpan.FromSeconds(30));
                        continue;
                    }

                    Console.WriteLine("Running Job - {0} {1}", job.JobID, job.JobName);
                    Task.Run(() => job.Run());
                }
                Console.WriteLine("CancellationRequested!");
            }, cts.Token);
        }

        public void Stop()
        {
            cts.Cancel();
            Console.WriteLine("Stopping Job Runner");
        }
}

Answer 1

Are you really sure you want to start n number of tasks? What's the point of a queue if you're just going to try invoking all the tasks in the queue at once? In other words, you're taking a job from a queue, then queuing a Task, i.e. you're not really implementing your own queue.

I would also suggest using a better singleton pattern, double-checked locking can be easy to mess up: http://csharpindepth.com/Articles/General/Singleton.aspx in which case you wouldn't need to make your queue, instance, and cts members to be volatile.

Also, since you're not making the queue instance public, I would suggest not using a blocking queue just making your access to a non-blocking queue thread-safe (See ReaderWriterLockSlim) you will be able to optimize that much better. After all, you're making a multi-threaded queue, you're already trying to optimize the use of the CPU.

If your jobs are going to be very long (greater than 1 second) Task.StartNew might be a better option than Task.Run for the jobs, whereby you can include TaskCreationOption.LongRunning. For example:

Task.Factory.StartNew(() =>
{
    while (!cts.IsCancellationRequested)
    {
        clJob job;
        if ((job = clJobQueue.Instance.GetNextJob()) == null)
        {
            Console.WriteLine("Queue Empty, Sleeping for 30s");
            Thread.Sleep(TimeSpan.FromSeconds(30));
            continue;
        }

        Console.WriteLine("Running Job - {0} {1}", job.JobID, job.JobName);
        Task.Factory.StartNew(() => job.Run(), TaskCreationOptions.LongRunning);
    }
    Console.WriteLine("CancellationRequested!");
}, cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);

This doesn't mean your jobs have to be long-running; but if they end up being long running, it will reduce the stress on the thread pool.

Answer 2

clJobQueue

What does cl stand for? Class? It looks like some weird form of Hungarian notation, you shouldn't use that.

---

clJobQueue.Instance

Why is it a singleton? If you want to be able to easily access the queue from multiple places in your code, then a static class would work just as well. But I think a normal (non-static, non-singleton) class would be an even better option, since that means you can have multiple queues in your application.

---

clJobQueue.Instance.Populate()

Why does the queue know how to populate itself? I think that a better approach would be decouple the queue from getting the jobs, and have some external code that calls Add() on the queue.

---

Because your Parallel.Invoke() calls methods that don't return (at least not until they're explicitly canceled), it will never return itself. This means that your Ctrl+C-handling code will never be executed and that pressing Ctrl+C just terminates the application as it does normally, it won't cause Stop() to be called.