Multi-threading - Dependent tasks in batch using SysOperation framework

For a recent requirement, we had to make a custom batch job that could process thousands of lines coming from retail stores and post sales orders, purchase orders and counting journals for stock adjustments in AX. This batch job was to run on a regular basis daily. And to raise the bar of complexity even further, there was a sequence to the postings. First goes purchase, then sales and finally stock. Plus there were methods at the beginning and at the end for preparing the data and cleaning up activities at the end. So all in all we have a heavy duty batch job doing lots of heavy lifting and at the same time maintaining a sequence. That's when i started looking at multi-threading in AX.

Adding dependent tasks in AX 2009 has been possible using RunBaseBatch framework. Read this post
http://blogs.msdn.com/b/axsupport/archive/2011/04/13/threading-in-dynamics-ax.aspx

But we needed to tweak this approach as we have the SysOperation framework replacing most of the RunBase code going in future. All new batch jobs, ssrs reports rely on SysOperation classes.

I am going to show few of the tweakings we did and how we used them in the batch job.

First of all we needed a way to have a list of tasks that could be executed in parallel and at the same time define a dependent task. We created a custom class that extended SysOperationServiceBase and custom method that took a list and a task(class) as parameter. We called it addRunTimeTaskList().

Notice how we loop through the list of tasks, calling addRunTimeTask() for each and at the end call addDependency(). This is important you can't call addDependency() without calling addRunTimeTask() first. Note addDependency() method makes task _batchTaskAfter a dependent of each batchTask of the list. Enum BatchDependencyStatus::Finished tells that dependent task will only start executing once the parent tasks have finished successfully.

 /// <summary>  
 /// Used to add a list of runtime tasks to the current batch job  
 /// </summary>  
 /// <param name="_batchTasks">  
 /// List of tasks to be added  
 /// </param>  
 /// <param name="_batchTaskAfter">  
 /// task to be run after all other tasks are complete  
 /// </param>  
 /// <remarks>  
 ///  
 /// </remarks>  
 protected void AddRunTimeTaskList(List _batchTasks, Batchable _batchTaskAfter = null)  
 {  
   BatchHeader bh = this.getCurrentBatchHeader();  
   Batch b = this.getCurrentBatchTask();  
   Batchable batchTask;  
   ListEnumerator listEnum;  
   boolean isRunTimeJob = bh.parmRuntimeJob();  
   listEnum = _batchTasks.getEnumerator();  
   listEnum.reset();  
   if( this.isExecutingInBatch() )  
   {  
     if( _batchTaskAfter )  
     {  
       bh.addRuntimeTask(_batchTaskAfter,b.RecId);  
     }  
     /*Now loop through all the tasks and add them*/  
     while(listEnum.moveNext())  
     {  
       batchTask = listEnum.current();  
       //b.RunTimeTask = true;  
       //bh.addTask(batchTask);  
       bh.addRuntimeTask(batchTask,b.RecId);  
       if(_batchTaskAfter)  
       {  
         bh.addDependency(_batchTaskAfter,batchTask,BatchDependencyStatus::Finished);  
       }  
     }  
     try  
     {  
       ttsBegin;  
       //Need to restore this value as there is an issue in this version of AX  
       bh.parmRuntimeJob(isRunTimeJob);  
       bh.save();  
       ttsCommit;  
     }  
     catch( Exception::UpdateConflict )  
     {  
       error("Failed to add child task");  
     }  
   }  
   else  
   {  
     /*Now loop through all the tasks and add them*/  
     while(listEnum.moveNext())  
     {  
       batchTask = listEnum.current();  
       batchTask.run();  
     }  
     /*See if there is a batch to run after*/  
     if(_batchTaskAfter)  
     {  
       _batchTaskAfter.run();  
     }  
   }  
 }  

This is how we call the method addRunTimeTaskList() passing a list and a call to another method as dependent.

 this.AddRunTimeTaskList(batchTasks_Sales, this.batchTaskAfterSales());  

Lets see how we populate batchTasks_Sales. We while through a query each time calling method addUpdateBatch_Sales(), in which we fill a list with instances of SysOperationServiceController class and setting the data contract alongwith.

   while (queryRun.next())  
   {  
     mmsStagingImportTable     = queryRun.get(tableNum(MMSStagingImportTable));  
     if (queryRun.changed(tableNum(MMSStagingImportTable)))  
     {  
       updateBatch.parmDescription(strFmt("@SYS76785", startingPosition, "@MMS2867"));  
       startingPosition++;  
       updateBatch.parmStartDate(processingDate);  
       updateBatch.parmImportId(mmsStagingImportTable.ImportId);  
       this.addUpdateBatch_Sales(updateBatch, classStr(MMSStagingDataSalesCopy), methodStr(MMSStagingDataSalesCopy, copyData));  
     }  
   }  

 public void addUpdateBatch_Sales(MMSStagingDataSalesCopyBatchDC _updateBatch, ClassName _class, MethodName _method)  
 {  
   MMSStagingDataSalesCopyBatchDC dataContract;  
   SysOperationServiceController sosc = new SysOperationServiceController(_class, _method, SysOperationExecutionMode::Synchronous );  
   dataContract = sosc.getDataContractObject();  
   if (dataContract)  
   {  
     dataContract.parmDescription(_updateBatch.parmDescription());  
     dataContract.parmStartDate(_updateBatch.parmStartDate());  
     dataContract.parmImportId(_updateBatch.parmImportId());  
   }  
   batchTasks_Sales.addEnd(sosc);  
 }  

Final piece of the puzzle, dependent method batchTaskAfterSales() which calls the next method.

 private SysOperationServiceController batchTaskAfterSales()  
 {  
   MMSStagingMasterPostingDC     dataContractStock;  
   SysOperationServiceController sosc = new SysOperationServiceController(classStr(MMSStagingDataMasterPosting), methodStr(MMSStagingDataMasterPosting, batchStores), SysOperationExecutionMode::Synchronous);  
   dataContractStock = sosc.getDataContractObject();  
   return sosc;  
 }  

If you run the batch job and have it executing, going to the "View tasks" and selecting dependent task, you could see in the bottom grid the parent tasks.

EDIT: Its important that you use SysOperationExecutionMode::Synchronous in instantiating SysOperationServiceController otherwise your dependent task would run before.