ParallelWork: Feature rich multithreaded fluent task execution library for WPF

Download ParallelWork-src.zip - 3.29 KB

Latest source code is available as part of this open source project. 

Introduction  

ParallelWork is an open source free helper class that lets you run multiple work in parallel threads, get success, failure and progress update on the WPF UI thread, wait for work to complete, abort all work (in case of shutdown), queue work to run after certain time, chain parallel work one after another. It’s more convenient than using .NET’s BackgroundWorker because you don’t have to declare one component per work, nor do you need to declare event handlers to receive notification and carry additional data through private variables. You can safely pass objects produced from different thread to the success callback. Moreover, you can wait for work to complete before you do certain operation and you can abort all parallel work while they are in-flight. If you are building highly responsive WPF UI where you have to carry out multiple job in parallel yet want full control over those parallel jobs completion and cancellation, then the ParallelWork library is the right solution for you.

I am using the ParallelWork library in my PlantUmlEditor project, which is a free open source UML editor built on WPF. You can see some realistic use of the ParallelWork library there. Moreover, the test project comes with 400 lines of Behavior Driven Development flavored tests, that confirms it really does what it says it does.

Source code 

The source code of the library is part of the “Utilities” project in PlantUmlEditor source code hosted at Google Code. 

How to use it 

The library comes in two flavours, one is the ParallelWork static class, which has a collection of static methods that you can call. Another is the Start class, which is a fluent wrapper over the ParallelWork class to make it more readable and aesthetically pleasing code.

ParallelWork allows you to start work immediately on separate thread or you can queue a work to start after some duration. You can start an immediate work in a new thread using the following methods:

• void StartNow(Action doWork, Action onComplete)
• void StartNow(Action doWork, Action onComplete, Action<Exception> failed)

For example,

ParallelWork.StartNow(() =>
{
    workStartedAt = DateTime.Now;
    Thread.Sleep(howLongWorkTakes);
},
() =>
{
    workEndedAt = DateTime.Now; 
});

Or you can use the fluent way Start.Work:

Start.Work(() =>
    {
        workStartedAt = DateTime.Now;
        Thread.Sleep(howLongWorkTakes);
    })
    .OnComplete(() =>
    {
        workCompletedAt = DateTime.Now;
    })
    .Run();

Besides simple execution of work on a parallel thread, you can have the parallel thread produce some object and then pass it to the success callback by using these overloads:

• void StartNow<T>(Func<T> doWork, Action<T> onComplete)
• void StartNow<T>(Func<T> doWork, Action<T> onComplete, Action<Exception> fail)

For example, 

ParallelWork.StartNow<Dictionary<string, string>>(
    () =>
    {                        
        test = new Dictionary<string,string>();
        test.Add("test", "test");

        return test;
    },
    (result) =>
    {
        Assert.True(result.ContainsKey("test"));
    });

Or, the fluent way:

Start<Dictionary<string, string>>.Work(() =>
    {
        test = new Dictionary<string, string>();
        test.Add("test", "test");

        return test;
    })
    .OnComplete((result) =>
    {
        Assert.True(result.ContainsKey("test"));
    })
    .Run(); 

You can send progress update from the background thread while a work is going on to the UI thread in order to update UI. For example, 

Start.Work((onprogress) =>
{
	for (int i = 0; i < 10; i++)
	{
		Thread.Sleep(100);
		onprogress("Working..." + i, i);
	}
})
.OnProgress((msg, progress) =>
{
	UpdateSomeProgressBar(progress);
})
.OnComplete(() =>
{
	UpdateSomeProgressBar(100);
})
.Run(); 

The onprogress callback is passed to the Work's delegate which you can call directly from the background thread. It then fires the delegate passed in the .OnProgress(...) function. 

You can also start a work to happen after some time using these methods:

• DispatcherTimer StartAfter(Action onComplete, TimeSpan duration)
• DispatcherTimer StartAfter(Action doWork,Action onComplete,TimeSpan duration)

You can use this to perform some timed operation on the UI thread, as well as perform some operation in separate thread after some time.

ParallelWork.StartAfter(
    () =>
    {
        workStartedAt = DateTime.Now;
        Thread.Sleep(howLongWorkTakes);
    },
    () =>
    {
        workCompletedAt = DateTime.Now;
    },
    waitDuration);

Or, the fluent way:

Start.Work(() =>
    {
        workStartedAt = DateTime.Now;
        Thread.Sleep(howLongWorkTakes);
    })
    .OnComplete(() =>
    {
        workCompletedAt = DateTime.Now;
    })
    .RunAfter(waitDuration);

There are several overloads of these functions to have a exception callback for handling exceptions or get progress update from background thread while work is in progress. For example, I use it in my PlantUmlEditor to perform background update of the application.

// Check if there's a newer version of the app
Start<bool>.Work(() => 
{
    return UpdateChecker.HasUpdate(Settings.Default.DownloadUrl);
})
.OnComplete((hasUpdate) =>
{
    if (hasUpdate)
    {
        if (MessageBox.Show(Window.GetWindow(me),
            "There's a newer version available. 
Do you want to download and install?",
            "New version available",
            MessageBoxButton.YesNo,
            MessageBoxImage.Information) == MessageBoxResult.Yes)
        {
            ParallelWork.StartNow(() => {
                var tempPath = System.IO.Path.Combine(
                    Environment.GetFolderPath(Environment.SpecialFolder.ApplicationData),
                    Settings.Default.SetupExeName);

                UpdateChecker.DownloadLatestUpdate(Settings.Default.DownloadUrl, tempPath);
            }, () => { },
                (x) =>
                {
                    MessageBox.Show(Window.GetWindow(me),
                        "Download failed. When you run next time, 
it will try downloading again.",
                        "Download failed",
                        MessageBoxButton.OK,
                        MessageBoxImage.Warning);
                });
        }
    }
})
.OnException((x) => 
{
    MessageBox.Show(Window.GetWindow(me),
        x.Message,
        "Download failed",
        MessageBoxButton.OK,
        MessageBoxImage.Exclamation);
});

The above code shows you how to get exception callbacks on the UI thread so that you can take necessary actions on the UI. Moreover, it shows how you can chain two parallel works to happen one after another.

Sometimes you want to do some parallel work when user does some activity on the UI. For example, you might want to save file in an editor while user is typing every 10 second. In such case, you need to make sure you don’t start another parallel work every 10 seconds while a work is already queued. You need to make sure you start a new work only when there’s no other background work going on. Here’s how you can do it:

private void ContentEditor_TextChanged(object sender, EventArgs e)
{
  if (!ParallelWork.IsAnyWorkRunning())
  {
     ParallelWork.StartAfter(SaveAndRefreshDiagram, 
                                 TimeSpan.FromSeconds(10));
  }
}

If you want to shutdown your application and want to make sure no parallel work is going on, then you can call the StopAll() method.

ParallelWork.StopAll();

If you want to wait for parallel works to complete without a timeout, then you can call the WaitForAllWork(TimeSpan timeout). It will block the current thread until the all parallel work completes or the timeout period elapses.

result = ParallelWork.WaitForAllWork(TimeSpan.FromSeconds(1));

The result is true, if all parallel work completed. If it’s false, then the timeout period elapsed and all parallel work did not complete.

How does it work? 

Let’s see how the StartNow method works, which is responsible for queuing work in separate thread and firing success, failure and progress update callbacks on the UI thread using Dispatcher.

public static void StartNow(Action doWork, Action onComplete)
{
    StartNow(doWork, onComplete, (x) => { throw x; });
}

public static void StartNow(Action doWork, Action onComplete, 
Action<Exception> failed)
{
    StartNow<object>(() => { doWork(); return true; }, (o) => onComplete(), failed);
}

public static void StartNow<T>(Func<T> doWork, Action<T> onComplete)
{
    StartNow<T>(doWork, onComplete, (x) => { throw x; });
}

public static void StartNow<T>(Func<T> doWork, Action<T> onComplete, 
Action<Exception> fail)
{
    StartNow<object, T>(new object(), 
        (o, progressCallback) => { return doWork(); },
        (o, msg, done) => { },
        (o, result) => onComplete(result),
        (o, x) => { fail(x); }
        );
}

public static void StartNow<T, R>(
    T arg,
    Func<T, Action<T, string, int>, R> doWork,
    Action<T, string, int> progress,
    Action<T, R> onComplete,
    Action<T, Exception> fail)
{
    Weak<Dispatcher> currentDispatcher = Dispatcher.CurrentDispatcher;
    Thread newThread = new Thread(new ParameterizedThreadStart( (thread)=>
        {
            var currentThread = thread as Thread;

            Dispatcher dispatcher = currentDispatcher;
            if (null == dispatcher)
                fail(arg, new ApplicationException("Dispatcher is unavailable"));
                
            try
            {
                Debug.WriteLine(currentThread.ManagedThreadId 
                     + " Work execution stated: " + DateTime.Now.ToString());
                
                R result = doWork(arg,
                    (data, message, percent) => dispatcher
                         .BeginInvoke(progress, arg, message, percent));

                if (null == result)
                {
                    try
                    {
                        dispatcher.BeginInvoke(fail, arg, null);
                    }
                    catch
                    {
                        // Incase the error handler produces exception, 
                           we have to gracefully
                        // handle it since this is a background thread
                    }
                    finally
                    {
                        // Nothing to do, error handler is not supposed to 
                           produce more error
                    }
                }
                else
                {
                    try
                    {
                        dispatcher.BeginInvoke(onComplete, arg, result);
                    }
                    catch (Exception x)
                    {
                        dispatcher.BeginInvoke(fail, arg, x);
                    }
                }
            }
            catch (ThreadAbortException ex)
            {
                Debug.WriteLine(ex);
            }
            catch (Exception x)
            {
                dispatcher.BeginInvoke(fail, arg, x);
            }
            finally
            {
                currentDispatcher.Dispose();
                Debug.WriteLine(currentThread.ManagedThreadId 
                   + " Work execution completed: " + DateTime.Now.ToString());

                lock (_threadPool)
                {
                    _threadPool.Remove(thread as Thread);
                    if (_threadPool.Count == 0)
                    {
                        _AllBackgroundThreadCompletedEvent.Set();
                        Debug.WriteLine("All Work completed: " 
                           + DateTime.Now.ToString());
                    }
                }
            }
        }));
    
    // Store the thread in a pool so that it is not garbage collected
    lock(_threadPool) 
        _threadPool.Add(newThread);

    _AllBackgroundThreadCompletedEvent.Reset();
    Debug.WriteLine(newThread.ManagedThreadId + " Work queued at: " 
              + DateTime.Now.ToString());            

    newThread.SetApartmentState(ApartmentState.STA);
    newThread.Start(newThread);            
}

First it takes the current dispatcher from the caller thread. Since the caller thread is the UI thread, as you are calling ParallelWork.StartNow(), from UI thread, the Dispatcher reference resolves the UI thread. The reference is stored in a Weak reference so that the reference to the Dispatcher is not kept alive until all threads have a chance to get called and release the reference. This is the classic “closure” problem you mostly hear in the javascript world.

Then it creates a new thread and resets a ManualWaitEvent which is signaled when all threads are completed. It’s used in WaitForAllWork function to wait until all threads have completed their work.

In the thread callback, it receives the Dispatcher reference and then fires the success, failure and progress update callbacks.

The other sophisticated function is the StartNowAfter, which queues a work to happen in separate thread after some duration.

public static DispatcherTimer StartNowAfter<T, R>(
    T arg,
    Func<T, Action<T, string, int>, R> doWork, 
    Action<T, string, int> onProgress,
    Action<T, R> onComplete, 
    Action<T, Exception> onError, 
    TimeSpan duration)
{
    var timer = new DispatcherTimer(duration, DispatcherPriority.Normal, 
    new EventHandler((sender, e) =>
    {
        var currentTimer = (sender as DispatcherTimer);
        currentTimer.Stop();
        lock (_timerPool)
        {
            _timerPool.Remove(currentTimer);
            if (_timerPool.Count == 0)
                _AllTimerFiredEvent.Set();
        }

        ParallelWork.StartNow<T, R>(arg, doWork, onProgress, onComplete, onError);
    }),
    Dispatcher.CurrentDispatcher);

    _AllTimerFiredEvent.Reset(); 


    lock(_timerPool)
        _timerPool.Add(timer);
    timer.Start();

    return timer;
}

It basically reuses the StartNow function.

The other functions like StopAll, WaitForWork are straightforward. Nothing worth mentioning here.

Unit testing the library 

There’s a xUnit test suite which tests the ParallelWork class using BDD approach. I find BDD approach better than traditional TDD approach where a single method tests a single fact only. I prefer testing a method against its overall behavior, not using a fraction of its behavior, which is a ‘fact’.

Unit testing multithreaded libraries is complex. It’s even more complicated when WPF is involved and we are using Dispatcher which is not a simple thing to test. Hope the following tests will give you some ideas on how to write tests that confirms multithreaded behavior of your components.

Let’s test the most simple one.

[Specification][STAThread]
public void WaitForAllWork_should_return_immediately_if_no_work_queued()
{
    var stopWatch = new Stopwatch();
    "Given no work going on".Context(() =>
    {
        Assert.False(ParallelWork.IsWorkOrTimerQueued());
    });

    var result = default(bool);
    "When WaitForAllWork is called".Do(() =>
    {
        stopWatch.Start();
        result = ParallelWork.WaitForAllWork(TimeSpan.FromSeconds(1));
    });

    "It should return immediately without going into any wait period".Assert(() =>
    {
        Assert.True(stopWatch.Elapsed < TimeSpan.FromSeconds(1));
    });

    "It should return true".Assert(() =>
    {
        Assert.True(result);
    });
}

Next is to test the StartNow function to ensure the objects produced in other thread can be accessible from the UI thread where the success callback is fired.

public void StartNow_should_return_objects_from_parallel_thread_to_callbacks()
{
    TimeSpan howLongWorkTakes = TimeSpan.FromSeconds(1);
    TimeSpan timeout = howLongWorkTakes.Add(TimeSpan.FromMilliseconds(500));

    DispatcherFrame frame = default(DispatcherFrame);

    "Given no background work running".Context(() =>
    {
        Assert.False(ParallelWork.IsWorkOrTimerQueued());
        frame = new DispatcherFrame();
    });

    var test = default(Dictionary<string,string>);
    
    var output = default(Dictionary<string, string>);
    "When StartNow<> is called".Do(() =>
    {
        ParallelWork.StartNow<Dictionary<string, string>>(
            () =>
            {                        
                test = new Dictionary<string,string>();
                test.Add("test", "test");

                return test;
            },
            (result) =>
            {
                output = result;
                frame.Continue = false;
            });
    });

    @"It should return the object produced in separate thread 
    and the object should be modifiable"
        .Assert(() =>
        {
            Assert.True(WaitForWorkDoneAndFireCallback(timeout, frame));

            Assert.True(output.ContainsKey("test"));
            Assert.DoesNotThrow(() => output.Add("another key", "another value"));
        });
}

Here it shows that inside a parallel thread, you can create a dictionary and that’s returned to the callback and you can access dictionary and make changes to it from the callback thread.

This test shows you some of the pains of unit testing WPF Dispatcher. In order to make Dispatcher.BeginInvoke to work in a unit test, without having a WPF UI running, you have to first create a DispatcherFrame, and then call Dispatcher.PushFrame(frame) to execute the BeginInvoke calls. The problem and the solution is explained in great details in this blog post. The function WaitForWorkDoneAndFireCallback does the job of running Dispatcher and making it process the BeginInvoke calls.

Now let’s test a slightly more complicated behavior. We want to ensure the StartNowAfter does the job after the specified time, not before that and the job is carried in a separate thread, without blocking caller. The following test ensures the behavior:

[Specification]
[STAThread]
public void StartNowAfter_should_execute_the_work_after_given_duration()
{
    DateTime workStartedAt = default(DateTime);
    DateTime workCompletedAt = default(DateTime);
    DateTime countDownStartedAt = default(DateTime);
    TimeSpan waitDuration = TimeSpan.FromSeconds(2);
    TimeSpan howLongWorkTakes = TimeSpan.FromSeconds(1);
    TimeSpan timeout = waitDuration.Add(howLongWorkTakes.Add(
          TimeSpan.FromMilliseconds(500)));

    DispatcherFrame frame = default(DispatcherFrame);

    "Given no background work running".Context(() =>
    {
        Assert.False(ParallelWork.IsAnyWorkRunning());
        frame = new DispatcherFrame();

        workStartedAt = default(DateTime);
        workCompletedAt = default(DateTime);
    });

    "When StartNowAfter is called".Do(() =>
    {
        ParallelWork.StartNowAfter(
            () =>
            {
                workStartedAt = DateTime.Now;
                Thread.Sleep(howLongWorkTakes);
            },
            () =>
            {
                workCompletedAt = DateTime.Now;
                frame.Continue = false;
            },
            waitDuration);

        countDownStartedAt = DateTime.Now;
    });

    "It should not start the work until the duration has elapsed".Assert(() =>
    {
        Assert.Equal(default(DateTime), workStartedAt);
        Assert.Equal(default(DateTime), workCompletedAt);

        Assert.True(WaitForWorkDoneAndFireCallback(timeout, frame));
    });

    "It should start the work after duration has elapsed".Assert(() =>
    {
        Assert.True(WaitForWorkDoneAndFireCallback(timeout, frame));

        // Work should start within 500ms of start time
        DateTime expectedStartTime = countDownStartedAt.Add(waitDuration);
        Assert.InRange(workStartedAt,
            expectedStartTime,
            expectedStartTime.AddMilliseconds(500));

        // Work should end within 500ms of expected end time
        DateTime expectedEndTime = countDownStartedAt
                                        .Add(waitDuration)
                                        .Add(howLongWorkTakes);
        Assert.InRange(workCompletedAt,
            expectedEndTime,
            expectedEndTime.AddMilliseconds(500));
    });
}

The above test confirms that the StartNowAfter does not start the work before the expected time and the work completes within the expected time.

Next is to ensure the StopAllWork really stops all parallel running threads. The following test starts two parallel work and while the work is in-progress, it calls StopAllWork and ensures the threads abort and do not complete their job.

[Specification]
[STAThread]
public void StopAllWork_should_stop_all_parallel_work()
{
    DateTime work1StartedAt = default(DateTime);
    DateTime work2StartedAt = default(DateTime);

    DateTime work1EndedAt = default(DateTime);
    DateTime work2EndedAt = default(DateTime);

    TimeSpan waitDuration = TimeSpan.FromSeconds(2);
    TimeSpan howLongWorkTakes = TimeSpan.FromSeconds(1);
    TimeSpan timeout = waitDuration.Add(howLongWorkTakes
            .Add(TimeSpan.FromMilliseconds(500)));

    DispatcherFrame frame = default(DispatcherFrame);
    
    "Given some parallel work running".Context(() =>
        {
            frame = new DispatcherFrame();
            ParallelWork.StartNow(() =>
            {
                work1StartedAt = DateTime.Now;
                Thread.Sleep(howLongWorkTakes);
                work1EndedAt = DateTime.Now;
            },
            () =>
            {
                frame.Continue = false;
            });

            ParallelWork.StartNow(() =>
            {
                work2StartedAt = DateTime.Now;
                Thread.Sleep(howLongWorkTakes);
                work2EndedAt = DateTime.Now;
            },
            () =>
            {
                frame.Continue = false;
            });
        });

    "When StopAllWork is called".Do(() =>
        {
            // Let the work be half way through
            Thread.Sleep((int)howLongWorkTakes.TotalMilliseconds / 2);
            ParallelWork.StopAll();
        });

    "It should stop all work from completing".Assert(() =>
        {   
            // Confirm the work started
            Assert.NotEqual(default(DateTime), work1StartedAt);
            Assert.NotEqual(default(DateTime), work2StartedAt);
            
            // Confirm the work did not end
            Assert.Equal(default(DateTime), work1EndedAt);
            Assert.Equal(default(DateTime), work2EndedAt);

            Thread.Sleep(timeout);
        });
}

If you look at the Debug output, you will notice ThreadAbortException happening, confirming the threads really abort. Calling StopAllWork in real life scenario is dangerous. Since it might leave resources open and cause severe memory leak and resource contention. Use it with care.

Now let’s look at the most complicated test which ensure StartNow does what it’s supposed to do. I have been saving this for the last since the test is really long and complicated. There’s a lot of expectation from this function. It should execute the work in parallel thread without blocking the caller thread. It should fire the success, failure and progress update callbacks on the UI thread and so on. The following test ensures the StartNow works as expected and ensure all the major side-effects are also tested.

[Specification][STAThread]
public void StartNow_should_queue_a_new_thread_to_do_the_work()
{
    TimeSpan howLongWorkTakes = TimeSpan.FromSeconds(2);
    TimeSpan timeout = howLongWorkTakes.Add(TimeSpan.FromMilliseconds(500));
    
    var doWorkCalled = false;
    var successCallbackFired = false;
    var onExceptionFired = false;

    var doWorkThreadId = default(int);
    var onCompleteThreadId = default(int);
    var onExceptionThreadId = default(int);
    var letsThrowException = false;

    Stopwatch stopWatch = new Stopwatch();
    DispatcherFrame frame = default(DispatcherFrame);
            
    Action callbackFiredOnDispatcher = () => {
        frame.Continue = false; // Dispatcher should stop now
    };

    "Given no background work running".Context(() =>
    {
        Assert.False(ParallelWork.IsWorkOrTimerQueued());                
        frame = new DispatcherFrame();

        doWorkCalled = false;
        successCallbackFired = false;
        onExceptionFired = false;

        doWorkThreadId = default(int);
        onCompleteThreadId = default(int);
        onExceptionThreadId = default(int);
        
        stopWatch.Reset();                
    });

    "When a new work is queued".Do(() =>
    {
        var shouldThrowException = letsThrowException;
        ParallelWork.StartNow(() => // doWork
        {
            doWorkThreadId = Thread.CurrentThread.ManagedThreadId;
            doWorkCalled = true;

            // Simulate some delay in background work
            Thread.Sleep(howLongWorkTakes); 

            if (shouldThrowException)
            {
                throw new ApplicationException("Exception");
            }                                
        }, () => // onComplete
        {
            onCompleteThreadId = Thread.CurrentThread.ManagedThreadId;
            successCallbackFired = true;

            callbackFiredOnDispatcher();
        }, (x) => // onException
        {
            onExceptionThreadId = Thread.CurrentThread.ManagedThreadId;
            onExceptionFired = true;

            callbackFiredOnDispatcher();
        });

        stopWatch.Start();
    });

    "It should return control immediately without blocking the current thread".Assert(() =>
    {
        Assert.True(stopWatch.Elapsed < howLongWorkTakes, 
            string.Format("{0}<{1}", stopWatch.Elapsed, howLongWorkTakes));
        Assert.True(WaitForWorkDoneAndFireCallback(timeout, frame));
    });

    "It should return true if IsWorkQueued is called".Assert(() =>
    {
        Assert.True(ParallelWork.IsWorkOrTimerQueued());
        Assert.True(WaitForWorkDoneAndFireCallback(timeout, frame));
    });

    "It should wait for the work to complete if WaitForAllWork is called".Assert(() =>
    {
        Assert.True(WaitForWorkDoneAndFireCallback(timeout, frame));

        // The work should finish within the duration it takes with max 1 sec buffer
        // for additional stuff xunit does.
        Assert.True(stopWatch.Elapsed < howLongWorkTakes.Add(TimeSpan.FromSeconds(1)));
    });

    "It should execute the work in a separate thread".Assert(() => 
    {
        Assert.True(WaitForWorkDoneAndFireCallback(timeout, frame));

        Assert.True(doWorkCalled);
        Assert.NotEqual(Thread.CurrentThread.ManagedThreadId, doWorkThreadId);
    });

    "It should fire onComplete on the same thread as the UI thread".Assert(() =>
    {
        Assert.True(WaitForWorkDoneAndFireCallback(timeout, frame));

        Assert.True(successCallbackFired);
        Assert.Equal(Thread.CurrentThread.ManagedThreadId, onCompleteThreadId);
    });
    
    "It should not fire onException if there's no exception".Assert(() =>
    {
        Assert.True(WaitForWorkDoneAndFireCallback(timeout, frame));

        Assert.False(onExceptionFired);

        letsThrowException = true; // This is for next assert                
    });

    "It should fire exception on UI thread".Assert(() =>
    {
        Assert.True(WaitForWorkDoneAndFireCallback(timeout, frame));

        Assert.False(successCallbackFired);
        Assert.True(onExceptionFired);
        Assert.Equal(Thread.CurrentThread.ManagedThreadId, onExceptionThreadId);
    });
}

There’s a lot of code here to look at. I will save you the boring C# to English translation. The code should be clear enough to read. If not, I am a bad coder. Feel free to bash me in the forum.

Conclusion  

So there you have it. A feature rich multithreaded task execution library, backed by 400 lines of test code and a complete WPF smart client example to prove it really does what it says it does.