Multi-threaded batch job template

Batch jobs in AX are important concepts, required for processing different business logic, filling up tables for reports etc

Multi-threaded batch jobs are an advanced usage of the batch jobs functionality, which can handle even more heavy duty tasks due to parallel processing capability.

Since these jobs have been becoming common place, I thought to create a template to make my life easier.

It’s a project with few objects.

Few pointers:

1. Objects have prefix Template, which I replace with the feature I want the batch job for.
2. "TemplateService" class, "createTasks" method is where the magic happens. This methods spins off a fixed number of threads depending on server thread setting, usually 8. It divides the workload in these 8 threads and makes a list of SysOperationServiceController objects. These objects are basically call to the “run” method of the class, which has the main business logic. For example, you want to process records in CustTable. If you have 32 customers, you will get 8 threads working on a set of (32/8) 4 customers each. This is not the best usage scenario for multi-threading but think 32000 customers, meaning each thread handles 4000 customers in parallel. Your job would finish in roughly 1/8th the time required to finish a single threaded batch job. A less common way to do multi-threading, would be to spin as many threads as required. For example if we want to iterate through all the customers and do some logic for each, we could spin 32 threads. Will try to modify this template to suit that requirement.
3. Another nice logic is in method “AddrunTimeTaskList” in class “ANSysOperationServiceBase”. This class extends SysOperationServiceBase, to enable adding dependent task to a list of parallel tasks. Please see my previous blog here for more info on this. My template project shows how to use this class and add a dependent method to a list of tasks.
4. There are “runInitialTask” and “runFinalTask” methods which can be used to call tasks before and after your parallel list of tasks. 
5. I use paging in query to divide the workload. This requires your AOT query to have its order by node set. 

Please email me for xpo. Cheers.

Creating batch jobs with X++

Two jobs to create batch jobs with X++, one for runBase and other for SysOperation.
The two jobs are essentially the same, except that the SysOperation job uses both controller and contract classes.
You could also choose to add dependency between two tasks.
batchHeader.addTask(batchTaskToRun);
batchHeader.addTask(dependsOnBatchTask);
batchHeader.addDependency(batchTaskToRun, dependsOnBatchTask);

You will get the same result as here, my previous blog post on adding dependency between jobs manually.

 static void batchRunBase(Args _args)  
 {  
   #define.recurrenceStartTime('23:00')  
   #define.recurrenceInterval(1)  
   #define.noOfRetriesOnFailure(4)  
   BatchHeader                  batchHeader;  
   BatchInfo                    batchInfo;  
   SysRecurrenceData            sysRecurrenceData;  
   utcDateTime                  batchStartTime;  
   Batch                        batch;  
   Tutorial_RunBaseBatch        yourRunBaseBatchClass;  
   ;  
   select firstOnly RecId from batch  
     where batch.ClassNumber == classNum(Tutorial_RunBaseBatch) &&  
       batch.Company == curext() &&  
       batch.DataPartition == getcurrentpartition();  
   if (!batch.RecId)  
   {  
     yourRunBaseBatchClass = Tutorial_RunBaseBatch::construct();  
     batchInfo = yourRunBaseBatchClass.batchinfo();  
     batchInfo.parmRetriesOnFailure(#noOfRetriesOnFailure);  
     // Create the batch job  
     batchHeader = BatchHeader::construct();  
     batchHeader.addTask(yourRunBaseBatchClass); // or controller class  
     //batchHeader.parmCaption("Tutorial");  
     batchStartTime = DateTimeUtil::newDateTime(systemDateGet(), str2time(#recurrenceStartTime), DateTimeUtil::getUserPreferredTimeZone());  
     // Set the recurrence data  
     sysRecurrenceData = SysRecurrence::defaultRecurrence();  
     sysRecurrenceData = SysRecurrence::setRecurrenceStartDateTime(sysRecurrenceData, batchStartTime);  
     sysRecurrenceData = SysRecurrence::setRecurrenceNoEnd(sysRecurrenceData);  
     sysRecurrenceData = SysRecurrence::setRecurrenceUnit(sysRecurrenceData, SysRecurrenceUnit::Day, #recurrenceInterval);  
     batchHeader.parmRecurrenceData(sysRecurrenceData);  
     // Set the batch alert configurations  
     batchHeader.parmAlerts(NoYes::No, NoYes::Yes, NoYes::No, NoYes::Yes, NoYes::Yes);  
     batchHeader.parmStartDateTime(batchStartTime);  
     // Save the batch job  
     batchHeader.save();  
   }  
 }  

 static void batchSys(Args _args)  
 {  
   #define.recurrenceStartTime('23:00')  
   #define.recurrenceInterval(1)  
   #define.noOfRetriesOnFailure(4)  
   BatchHeader                                 batchHeader;  
   BatchInfo                                   batchInfo;  
   SysRecurrenceData                           sysRecurrenceData;  
   utcDateTime                                 batchStartTime;  
   Batch                                       batch;  
   AbcDemandForecastServiceController          AbcDemandForecastServiceController;  
   AbcDemandForecastServiceContract            AbcDemandForecastServiceContract;  
   ;  
   select firstOnly RecId from batch  
     where batch.ClassNumber == classNum(AbcDemandForecastServiceController) &&  
       batch.Company == curext() &&  
       batch.DataPartition == getcurrentpartition();  
   if (!batch.RecId)  
   {  
     AbcDemandForecastServiceController = new AbcDemandForecastServiceController(classStr(AbcDemandForecastServiceOperation), methodStr(AbcDemandForecastServiceOperation, calcDemandForecast), SysOperationExecutionMode::Synchronous);  
     AbcDemandForecastServiceContract = AbcDemandForecastServiceController.getDataContractObject(classStr(AbcDemandForecastServiceContract));  
     batchInfo = AbcDemandForecastServiceController.batchInfo();  
     batchInfo.parmRetriesOnFailure(#noOfRetriesOnFailure);  
     // Create the batch job  
     batchHeader = BatchHeader::construct();  
     batchHeader.addTask(AbcDemandForecastServiceController);   
     //batchHeader.parmCaption("Tutorial SysOperation");  
     batchStartTime = DateTimeUtil::newDateTime(systemDateGet(), str2time(#recurrenceStartTime), DateTimeUtil::getUserPreferredTimeZone());  
     // Set the recurrence data  
     sysRecurrenceData = SysRecurrence::defaultRecurrence();  
     sysRecurrenceData = SysRecurrence::setRecurrenceStartDateTime(sysRecurrenceData, batchStartTime);  
     sysRecurrenceData = SysRecurrence::setRecurrenceNoEnd(sysRecurrenceData);  
     sysRecurrenceData = SysRecurrence::setRecurrenceUnit(sysRecurrenceData, SysRecurrenceUnit::Day, #recurrenceInterval);  
     batchHeader.parmRecurrenceData(sysRecurrenceData);  
     // Set the batch alert configurations  
     batchHeader.parmAlerts(NoYes::No, NoYes::Yes, NoYes::No, NoYes::Yes, NoYes::Yes);  
     batchHeader.parmStartDateTime(batchStartTime);  
     // Save the batch job  
     batchHeader.save();  
   }  
 }  

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.

Capturing infolog messages

I had a requirement to retrieve the content of infolog and store it in log tables, made to track the progress of a batch job and take action on any errors. I used the below method in a catch statement handling all unexpected errors and at the same time updating error log tables. The while loop retrieves all the infolog errors using SysInfologEnumerator and SysInfologMessageStruct infolog classes and concatenating a string with the error messages.

 private str getErrorStr()  
 {  
   SysInfologEnumerator enumerator;  
   SysInfologMessageStruct msgStruct;  
   Exception exception;  
   str error;  
   enumerator = SysInfologEnumerator::newData(infolog.cut());  
   while (enumerator.moveNext())  
   {  
     msgStruct = new SysInfologMessageStruct(enumerator.currentMessage());  
     exception = enumerator.currentException();  
     error = strfmt("%1 %2", error, msgStruct.message());  
   }  
   return error;  
 }  

And the catch statement calling the above method.

   catch (Exception::Error)  
   {  
     ttsbegin;  
     mmsStagingPurchImport.selectForUpdate(true);  
     mmsStagingPurchImport.Error = true;  
     mmsStagingPurchImport.ErrorLog = this.getErrorStr();  
     mmsStagingPurchImport.update();  
     ttscommit;  
     retry;  
   }  

Playing with controller class

SysOperation framework relies on SysOperationServiceController class for passing the args argument from the menu item to the framework. Overriding some of the methods on this class gives you more control over its functioning. Same way when you want to modify the user interface of a SysOperation service, UI Builder classes are the way to go. Two methods of particular interest to me are:

showQueryValues When a query is used and this method returns true, then the fields with ranges and a Select button will be shown on the dialog. Return false in this method to hide them.

showQuerySelectButton This method does the same as the showQueryValues method, but only affects the Select button.

Some cases we want to create a batchable job, which a user can use but don't want to give him the option to 'select' any particular records or its meaningless to show the query fields on dialog. Overriding the above two methods to return false can solve the purpose.

You will also need to override main and construct methods in this case, like below.

 public static void main(args _args)  
 {  
   MMSPriceUpdatePublishDataController  controller = MMSPriceUpdatePublishDataController::construct();  
   controller.startOperation();  
 }  

 private static MMSPriceUpdatePublishDataController construct(SysOperationExecutionMode _mode = SysOperationExecutionMode::ReliableAsynchronous)  
 {  
   MMSPriceUpdatePublishDataController  controller = new MMSPriceUpdatePublishDataController(classStr(MMSPriceUpdatePublishData),methodStr(MMSPriceUpdatePublishData, run), _mode);  
   return controller;  
 }  

 public boolean showQuerySelectButton(str parameterName)  
 {  
   return false;  
 }  

Also set the Object property on the menu item as the name of your controller class.

X++ Vs CIL

In one of the assignments, I had to make a batch job using Sys operation framework.

Requirement was to create and post trade agreements, and for that I used PricePriceDiscJourService.

So I created a service class, a data contract class, a service, a query and a menu item. All standard AX objects required for a batch job to run.

The issue was that if i put all my methods/logic  in a standalone class, everything worked fine. That means records get created in all the right tables like PriceDiscAdmTrans, PriceDiscAdmtable, and PriceDiscTable. But if try to run the batch job (after CIL compile of course), it doesn’t create all the records. Some records were getting skipped, which didn’t make sense since my standalone class and the batch job class were logically speaking the same. 

I used a while(query.next()) to loop through two header/line tables and fetch records.

Now the fun part. Debugging in Visual Studio (since it’s a service you can’t debug in X++) i found that the while loop works little differently in Visual Studio than in X++. In Visual Studio, the line table's RecId gets reset even when the while loop moves to the next parent record when in fact you would expect it to be there till the cursor moves to the line record. AX, as expected, keeps the RecId value until the line table gets reassigned the next table record. My logic depended on this and it failed. Finally I changed the logic to store the table buffer in a variable and managed to get the code working.

Does anyone have a similar experience or any suggestions?

Tackling dependent tasks in a Batch job

I am sure most of you must be familiar with Batch jobs. If not created one, you must have at least seen some in action. A simple batch job consists of a single task with no dependencies. On the other hand, a complex batch job can have multiple tasks having dependencies with each other.

Basic info on how to create and run batch jobs can be found here.
http://msdn.microsoft.com/en-us/library/cc636647.aspx

In my post, I will demonstrate how to define dependencies between the tasks. This has application when you want to have complex dependency hierarchies, allowing you to schedule tasks in parallel, and choose multiple execution paths etc.

In short, I will create 2 tasks where second task will be dependent on the first task. Let's go step by step.

1. I am using the Tutorial_RunBaseBatch class. Just open/run the class, check the Batch processing checkbox and click OK. This will create a new batch job named Tutorial with one task.

2. Open the Batch job form. Select the newly created batch job named Tutorial. Click on View tasks button. You can see the only task available named Tutorial. Rename it to TutorialTask1 so that it makes more sense. Also, I kept a copy of Tutorial_RunBaseBatch class named CopyOfTutorial_RunBaseBatch. (I won't rename it as I am a little lazy. :P) Since each task represents a class, I will create a second task named TutorialTask2 running on CopyOfTutorial_RunBaseBatch class. Press Ctrl+N to create TutorialTask2.

3. Now select TutorialTask2, click the Has Conditions grid in the lower section of the Batch Tasks form and press Ctrl+N to create a new condition.

4. Select the task ID of the parent task, in our case TutorialTask1.
5. Select the status the parent task must have before the child/dependent task can run. In our case, TutorialTask2 starts when TutorialTask1 becomes 'Ended'. Save the condition.

It works more or less like a workflow. :)