I am using Paralle.Invoke() with a Windows Workflow activity to launch a set of long running tasks. I need some help on how to handle the AggregationException and Cancelling the tasks.
Below is the code I am using. The two issues that I am facing are:
1) I randomly see AggregateException thrown my code - I can't explain why. Would like to know if simply logging the exception would suffice.
2) Some tasks could take longer than the Task.Wait timeout. So I want to check that if any tasks are still pending, I want to cancel them out and essentially return control from the method. But, I am seeing that although I call "cancel",
control doesn't return from the method until all the tasks are done. So lets say if my Wait timeout is 1 minute and one of the tasks takes 5 mins, then the method will not return until 5 minutes despite doing a cancel. I want it to return as soon as the 1
min wait timeout is met. Please note that DoTask() is the method doing the long running work.
Code below:
protected override IAsyncResult BeginExecute(AsyncCodeActivityContext context, AsyncCallback callback, object state)
{
ParallelOptions options = new ParallelOptions();
CancellationTokenSource applicationTokenSource = new CancellationTokenSource();
CancellationToken applicationToken = applicationTokenSource.Token;
context.UserState = applicationTokenSource;
options.MaxDegreeOfParallelism = this.MaxDegreeOfParallelism.Get(context);
options.CancellationToken = applicationToken;
//setup up the Delegates
List<Action> workTasks = new List<Action>();
BatchDetailInfo[] workObjects = BatchDetails.Get(context);
int i = 0;
bool?[] results = new bool?[workObjects.Length];
foreach (BatchDetailInfo workObject in workObjects)
{
int k = i;// This is needed to work around the "closure" issue
BatchDetailInfo workObjectCopy = workObject; // This is needed to work around the "closure" issue
workTasks.Add(() => { results[k] = this.DoTask(workObjectCopy, 1, "", context); });
i++;
}
//run ParallelInvoke async
System.Threading.Tasks.Task parallel = Task.Factory.StartNew((taskState) =>
{
try
{
Parallel.Invoke(options, workTasks.ToArray());
}
catch (OperationCanceledException)
{
//expected if the token was cancelled
}
}, state);
try
{
parallel.Wait(ParallelInvokationTimeout.Get(context));
}
catch (AggregateException ex)
{
}
catch (Exception ex)
{
}
int batchDetailsFailedCount = results.Where(n => n == false).Count();
int runningCountOfBatchDetailsFailed = NumBatchDetailsFailed.Get(context);
runningCountOfBatchDetailsFailed += batchDetailsFailedCount;
NumBatchDetailsFailed.Set(context, runningCountOfBatchDetailsFailed);
int batchDetailsSuccessCount = results.Where(n => n == true).Count();
int runningCountOfBatchDetailsSucceeded = NumBatchDetailsProcessed.Get(context);
runningCountOfBatchDetailsSucceeded += batchDetailsSuccessCount;
NumBatchDetailsProcessed.Set(context, runningCountOfBatchDetailsSucceeded);
int batchDetailsPending = results.Where(n => !n.HasValue).Count();
if (batchDetailsPending > 0)
{
try
{
applicationTokenSource.Cancel(true);
}
catch (AggregateException ex)
{
}
}
//force the EndExecute
parallel.ContinueWith((task) => { callback.Invoke(task); });
return parallel;
}
