Multiple threads, progress in order

[JohnH]

Let's say there is a nested loop where each item has some processing time, each item should be reported (to UI) when finished and the items must be reported in work order. Here is pseudo code of the problem:

for A
   for B
      process AB
      report progress AB

Progress report must be in order A1:B1-B2-B3, A2:B1-B2, etc. Cancellation is also desirable.

It is tempting to do this with a BackgroundWorker, it has easy to use progress reporting to UI thread and can also be set up with cancellation, but it will only use a single thread. This example show how the items can be processed with multiple threads and still report progress in correct order using tasks (TPL).

First set up a test job class, this uses a Task to wait for before reporting progress, which is of course the Task that was started before it in loop. Processing is just a simulation with sleep time.

Public Class TestJob
    Public Property ID As String
    Public Property WaitTask As Task

    Public Sub ProcessSimple(time As Integer, reporter As IProgress(Of String))
        Thread.Sleep(time) 'processing
        WaitTask?.Wait()
        reporter.Report(ID)
    End Sub
End Class

The test code uses regular loops and sets up jobs with ID to see that loop order is followed.

Private reporter As New Progress(Of String)(Sub(status) Debug.WriteLine(status))

Private Sub TestSequenceSimple()
    Dim rnd As New Random
    Dim currentTask As Task = Nothing
    For A = 1 To 3
        For B = 1 To 3
            Dim job As New TestJob With {.ID = $"(A{A},B{B})", .WaitTask = currentTask}
            Dim processingTime = rnd.Next(3000, 5000)
            currentTask = Task.Run(Sub() job.ProcessSimple(processingTime, reporter))
        Next
    Next
End Sub

Adding cancellation to this is not hard, a CancellationTokenSource is the key ingredient in TPL for this. A list is used to hold all tasks and after loop these are waited for in Try-Catch to see the TaskCanceledException when Cancel method of CancellationTokenSource is called.

Private cancelSource As New CancellationTokenSource

Private Async Sub TestSequenceCancel()
    Dim rnd As New Random
    Dim currentTask As Task = Nothing
    Dim allTasks As New List(Of Task)
    For A = 1 To 3
        For B = 1 To 3
            Dim job As New TestJob With {.ID = $"(A{A},B{B})", .WaitTask = currentTask}
            Dim processingTime = rnd.Next(3000, 5000)
            currentTask = Task.Run(Sub() job.ProcessSimple(processingTime, reporter), cancelSource.Token)
            allTasks.Add(currentTask)
        Next
    Next
    Try
        Await Task.WhenAll(allTasks)
    Catch ex As TaskCanceledException
        Debug.WriteLine("operation cancelled")
    End Try
End Sub

Testing cancellation you may do this:

TestSequenceCancel()
Thread.Sleep(1000)
cancelSource.Cancel()

Adding some debug time measurement to code and the output could look like this:

(A1,B1): processing 3594, wait 0
(A1,B2): processing 4849, wait 0
(A1,B3): processing 3654, wait 1194
(A2,B1): processing 4580, wait 269
(A2,B2): processing 3009, wait 836
(A2,B3): processing 3987, wait 0
(A3,B1): processing 4623, wait 0
(A3,B2): processing 3585, wait 438
(A3,B3): processing 3097, wait 25
total processing time 34978 finished in 7721

This means what would have taken 35 seconds with a BackgroundWorker now takes only 8 seconds.

Feedback is welcome, also, would there be a different way of solving the problem at hand using other functionality of TPL?

 

[JohnH]

Here is an alternative using TPL with continuations, in this example I will also use a Function instead of a Sub. In TestJob class add this method:

Public Function ProcessSimpleResult(time) As Task(Of String)
    Thread.Sleep(time) 'processing
    WaitTask?.Wait()
    Return Task.FromResult(ID)
End Function

As you can see the principle is the same (for this problem), it waits for previous task to complete before returning result.

Test code is very similar to previous, but the main task now has a result, notice the type of currentTask. Since the main task is needed as input to the next item in loop the continuation is added next line, this get the main task as input and can access the Result. Instead of IProgress the TaskScheduler is used to continue on UI thread.

Private Sub TestSequenceSimpleResult()
    Dim rnd As New Random
    Dim UI = TaskScheduler.FromCurrentSynchronizationContext
    Dim currentTask As Task(Of String) = Nothing
    For A = 1 To 3
        For B = 1 To 3
            Dim job As New TestJob With {.ID = $"(A{A},B{B})", .WaitTask = currentTask}
            Dim processingTime = rnd.Next(1000, 2000)
            currentTask = Task.Run(Function() job.ProcessSimpleResult(processingTime))
            currentTask.ContinueWith(Sub(ante) Debug.WriteLine(ante.Result), UI)
        Next
    Next
End Sub

Adding cancellation is also similar as before. Here is it natural to wait for the continuation tasks to complete.

Private Async Sub TestSequenceSimpleResultCancel()
    Dim rnd As New Random
    Dim UI = TaskScheduler.FromCurrentSynchronizationContext
    Dim opt = TaskContinuationOptions.OnlyOnRanToCompletion
    Dim currentTask As Task(Of String) = Nothing
    Dim allTasks As New List(Of Task)
    For A = 1 To 3
        For B = 1 To 3
            Dim job As New TestJob With {.ID = $"(A{A},B{B})", .WaitTask = currentTask}
            Dim processingTime = rnd.Next(1000, 2000)
            currentTask = Task.Run(Function() job.ProcessSimpleResult(processingTime), cancelSource.Token)
            Dim continueTask = currentTask.ContinueWith(Sub(ante) Debug.WriteLine(ante.Result), cancelSource.Token, opt, UI)
            allTasks.Add(continueTask)
        Next
    Next
    Try
        Await Task.WhenAll(allTasks)
        Debug.WriteLine("operation complete")
    Catch ex As TaskCanceledException
        Debug.WriteLine("operation cancelled")
    End Try
End Sub

You can read more about continuations here, there are many options and variation that can be used: Chaining Tasks by Using Continuation Tasks | Microsoft Docs