2
\$\begingroup\$

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");
        }
}
\$\endgroup\$
  • \$\begingroup\$ Why don't you use ActionBlock from TPL Dataflow? Are you limited to .Net 4.0? \$\endgroup\$ – svick Oct 7 '14 at 9:29
2
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ I think you got the last part backwards: both Task.Run and Task.Factory.StartNew work equally well for short jobs, but Task.Factory.StartNew with TaskCreationOptions.LongRunning is better for long jobs. \$\endgroup\$ – svick Oct 7 '14 at 9:34
  • \$\begingroup\$ @svick You're right. I thought I read a BCL Team blog a couple of days ago that suggested Task.Run for long-running tasks. But, I can't find it and looking at referencesource shows it's not using TaskCreateOption.LongRunning; so if it did exist, it is wrong. I've corrected the answer and added some detail. \$\endgroup\$ – Peter Ritchie Oct 7 '14 at 14:30
0
\$\begingroup\$
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.

\$\endgroup\$

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.