Running a Workflow Asynchronously To Improve The Performance

• Download source - 70 KB

Introduction 

In this article a piece of code is implemented in three different Workflows. The first one implements it in a Foreach activity. The second one in a ParellelForeach and the third one runs that piece of code on different workflow instances on seperate threads. The goal of this article is to demonestrate how WorkflowApplication object runs a number of workflows asynchronously. In the end, I will compare the execution time of these methods.

Note: In app.config, remember to change the Trace tag in order to log execution information properly:

<system.diagnostics>
  <trace>
    <listeners>
      <clear/>
      <add name="MyFileListener" type="System.Diagnostics.TextWriterTraceListener"
           initializeData="TraceLog.TXT" />
    </listeners>
  </trace>
</system.diagnostics>

Requirements 

To follow along with this article, you will need the following installed:

• Visual Studio 2010 or higher and .NET 4.0 or higher.
• The code for this article.

Background

In WF 4.0, WorkflowApplication class is used to run a workflow(WorkflowInvoker is used either). Using WorkflowApplication.Run We have a better control over workflow execution. In WorkflowApplication, we can set actions for Completed, Aborted, Idle, OnUnhandledException in order to handle various events on the workflow. We will see how asynchronously running a workflow using WorkflowApplication will increase the performance of our code.

Using the Code

Our sample is a C# Workflow Console application. First of all, we need to create three seperate workflows like the following Workflows.

ForeachWorkflow.xoml: Here a Sequence activity has been used to create a simple math calculation and then the Sequence has been dropped in the Foreach Activity. Foreach Activity will enumerates the elements of our collection (itemList as an argument) and our Sequence will be executed once for each value in the itemList.  

ParallelForeachWorkflow.xoml: Here, as previous Workflow, a Sequence activity has been used and has been dropped in the ParallelForeach Activity. What differentiates ParallelForeach from Foreach Activity is that if any of the embedded activities in the ParallelForeach are asynchronous or go idle, like a Delay activity or an activity which derives from AsyncCodeActivity, then, The embedded statements are scheduled together and run asynchronously. However, if none of the embedded activities are asynchronous or go idle, Foreach and ParallelForeach will execute in the same way. In our Workflow, ParallelForeachWorkflow.xoml, every Sequence is executed on different thread as there is a Delay activity among our activities in the ParallelForeach.

MultiThreadWorkflow.xaml: In this workflow, there is no Foreach or ParallelForeach activity. The iteration will be implemented in the code which will be discussed in the following.

 

Here is our custom activity which simulates a long lasting calculation. Here a Delay activity  has been used which just works like a time consuming calculation:

public sealed class LongRunningCalcActivity : CodeActivity
{
    public InArgument<TimeSpan> Duration { get; set; }

    protected override void Execute(CodeActivityContext context)
    {
        Thread.Sleep(context.GetValue<TimeSpan>(Duration));
    }
}

In our above workflows, in the LongRunningCalcActivity, I have set the Duration argument  to 00:00:01 which means a delay of 1 second.

Executing the Workflow

WorkflowApplication is an asynchronouse class unlike ActivityInvoker which runs the workflow in the current thread so WorkflowApplication.Run() will execute workflow asynchronously. In addition using WorkflowApplication we have greater control over execution. Here is my code to create a WorkflowApplication and run the Workflow:

WorkflowApplication workflowApp = new WorkflowApplication(new MultiThreadWorkflow());
workflowApp.Completed = WorkflowAppCompleted;
workflowApp.Run(); 

WorkflowAppCompleted is a method which will be called once the workflow execution finishes. In the below code, numberOfThreadsRunning is a variable which is used to be decremented once a workflow execution finishes. numberOfThreadsRunning is decremented until it reaches to 0. If so, ManualResetEvent is set and the main thread's execution resumes. Here is WorkflowAppCompleted method:

static int numberOfCalcs = 20; // number of workflows
static int numberOfThreadsRunning = numberOfCalcs; 
static ManualResetEvent eventDone = new ManualResetEvent(false);
public static void WorkflowAppCompleted(WorkflowApplicationCompletedEventArgs e)
{
   if (e.CompletionState == ActivityInstanceState.Faulted)
   {
       Trace.WriteLine(string.Format("Workflow {0} Terminated.", e.InstanceId.ToString()));
       Trace.WriteLine(string.Format("Exception: {0}\n{1}",
			e.TerminationException.GetType().FullName,
			e.TerminationException.Message));
   }
   else if (e.CompletionState == ActivityInstanceState.Canceled)
   {
       Trace.WriteLine("Canceled!");
   }
   else
   {
       if (Interlocked.Decrement(ref numberOfThreadsRunning) == 0)
          eventDone.Set();
   } 
}

Note: In the above code Interlocked.Decrement is used to make numberOfThreadsRunning modification mutually exclusive.

Note: In many situations we may use something like the following code in order to wait for other threads to finish:

WaitHandle[] waitHandles = new WaitHandle[2] { new AutoResetEvent(false), new AutoResetEvent(false) };
//Call Async methods
WaitHandle.WaitAll(waitHandles);

However, In this case, the number of WaitHandles must be limited (64) otherwise an exception will be thrown!

Here is the completed code:

class Program
{
  static int numberOfCalcs = 100;
  static int numberOfThreadsRunning = numberOfCalcs;
  static string workflowInArg = "itemList";
  static ManualResetEvent eventDone = new ManualResetEvent(false);

  static void Main(string[] args)
  {
     List<Double> items = new List<Double>();

     for (int i = 0; i < numberOfCalcs; i++)
        items.Add(System.Convert.ToDouble(i));

     Dictionary<string, object> p = new Dictionary<string, object>();
     p.Add(workflowInArg, items);

     //Linear Foreach
     RunSimpleForeach(p);

     //Parallel Foreach
     RunParallelForeach(p);

     //MultiThreading
     RunMultiThreading();

     Trace.Flush();

 }

 private static void RunMultiThreading()
 {
    Stopwatch sw = Stopwatch.StartNew();
    for (int workflowIndex = 0; workflowIndex < numberOfCalcs; workflowIndex++)
    {
       WorkflowApplication workflowApp = new WorkflowApplication(new MultiThreadWorkflow());
       workflowApp.Completed = WorkflowAppCompleted;
       workflowApp.Run();
    }

    eventDone.WaitOne();
    Trace.WriteLine("**** MultiThread Workflow, Elapsed Time: " + sw.Elapsed.ToString());
 }

 private static void RunParallelForeach( Dictionary<string, object> p)
 {
    Stopwatch sw = Stopwatch.StartNew();
    WorkflowInvoker.Invoke(new ParallelForeachWorkflow(), p);
    Trace.WriteLine("**** Parallel Foreach, Elapsed Time: " + sw.Elapsed.ToString());
 }

 private static void RunSimpleForeach(Dictionary<string, object> p)
 {
    Stopwatch sw = Stopwatch.StartNew();
    WorkflowInvoker.Invoke(new ForeachWorkflow(), p);
    Trace.WriteLine("**** Simple Foreach, Elapsed Time: " + sw.Elapsed.ToString());
 }

public static void WorkflowAppCompleted(WorkflowApplicationCompletedEventArgs e)
{
   if (e.CompletionState == ActivityInstanceState.Faulted)
   {
       Trace.WriteLine(string.Format("Workflow {0} Terminated.", e.InstanceId.ToString()));
       Trace.WriteLine(string.Format("Exception: {0}\n{1}",
			e.TerminationException.GetType().FullName,
			e.TerminationException.Message));
   }
   else if (e.CompletionState == ActivityInstanceState.Canceled)
   {
       Trace.WriteLine("Canceled!");
   }
   else
   {
       if (Interlocked.Decrement(ref numberOfThreadsRunning) == 0)
          eventDone.Set();
   } 
}
} 

Execution Result

Here is the result of running of our program:

Number of Threads: 100
**** Simple Foreach, Elapsed Time: 00:01:40.3350970
**** Parallel Foreach, Elapsed Time: 00:01:40.0157880
**** MultiThread Workflow, Elapsed Time: 00:00:10.0113136

As can be seen, MultiThread Workflow is considerably faster than the other ones! 

Note: The above results may be different in your machine. However, MultiThread Workflow will be far faster than the other ones in your machines too!   

Discussion  

The question which arise here is, why the result of Foreach and ParallelForeach Activity are almost alike?

The reason for this is that the workflow instance is run on a single thread in both activities. The WF runtime schedules work on that thread using a queue-like structure. When the ParallelForEach activity executes, it simply schedules its child activities with the runtime. Because the activities do blocking, synchronous work, they block the thread the workflow is running on. The solution is to run the workflow asynchronously using WorkflowApplication object. An alternative way is to inherit LongRunningCalcActivity from AsyncCodeActivity rather than CodeActivity in order to run our activity in a different thread. 

Conclusion 

WorkflowApplication object runs a workflow instance asynchronously. In order to use the Memory and the CPU of a machine in long running Workflows effectively, we can use this object to improve the performance. 

History

• March, 2012: Initial post.