%SYSTEM.WorkMgr

class %SYSTEM.WorkMgr extends %SYSTEM.AbstractWorkMgr

A typical calling sequence is:

  Set queue=$system.WorkMgr.%New() If queue="" ; Report Error, can check %objlasterror for %Status code
  For i=1:1:100 {
  	Set sc=queue.Queue("##class(MyClass).ClassMethod",i) If $$$ISERR(sc) ; Report Error
  }
  Set sc=queue.Sync() If $$$ISERR(sc) ; Report Error
  

Call Queue() to queue a unit of work to be run, this takes either a class method call, or a '$$func^rtn' reference and then any arguments you need to pass to this function. We support passing arrays here using the standard '.array' syntax although changes in the array by the work unit are not returned to the caller. As soon as the first Queue() is called a worker will start processing this item of work. It is important to make sure that all the units of work are totally independent and they do not rely on other work units. You must not rely on the order in which the units of work are processed. If the units may be changing a common global you will need to add locking to ensure one worker can not change a global while another worker is in the middle of reading this global. When a unit of work is queued the current security context is stored so the work unit will run inside the current security context. Note that the worker jobs are started by the Super Server and so will run as the operating system user that the Super Server process is setup to use, this may be different to your current logged in operating system user.

Finally call Sync() to wait for all the units of work to be complete, display any output each unit produced and report any errors reported from the work units. Instead of waiting for all work units to complete with the Sync() you can obtain notification of each completion using the WaitOne() interface which is an iterator over the work completion events. Another mechanism to run code when each work unit is completed is the QueueCallback() method.

Work units may write to the public variable %result which will be relayed to the parent process in either the WaitOne() call or in the callback function from the QueueCallback(). Worker jobs are owned by the parent process while they are performing work in this group, so when the parent exits the worker jobs will be released immediately. When the object returned by %New is destroyed this will remove all work associated with this group automatically, and release any workers.

The work queued should not perform exclusive kills or argumentless unlocks as this will interfere with the framework. Use of process private globals are only helpful within a chunk of work and not across chunks as different chunks of work are processed by different workers in different jobs. The size of each chunk should be on the order of thousands of lines of ObjectScript code to ensure the overhead of the framework is not a significant factor, also rather than a few very large chunks (e.g. 4 big chunks) if possible it is better to have a fairly large number (100 say) of chunks as this allows us to scale with CPU cores. Worker jobs once started will remain until they time out given a long enough period of inactivity as these jobs are shared among all work entered into the work queue manager in the same 'category'.

The third argument to %New() is the category which specifies which work queue pool to allocate worker jobs from. These categories are independent of each other so for example 'SQL' work is allocated from a different pool to 'Default' work and the load placed on the system from each of the categories can be tuned.

Method Inventory

• ActivateDeferred()
• Attach()
• DefaultNumWorkers()
• Detach()
• Flush()
• IsWorkerJob()
• NumberWorkers()
• Pause()
• Queue()
• QueueCallback()
• QueueDeferred()
• Resume()

Methods

classmethod ActivateDeferred(dtoken As %String) as %Status

This method activates the deferred work unit, represented by its dtoken as output by the QueueDeferred(), making this work unit available for a worker to process.
PARAMETERS:

dtoken - This parameter is the dtoken to represent a deferred work unit.

RETURN VALUE: %Status
This is similar to the Queue() method in the way that it makes the deferred work runnable, but does not guarantee completion/priority. This ActivateDeferred() can be called with a valid dtoken from the process that created the dtoken, or from a different process that has a valid dtoken.

classmethod Attach(token, ByRef sc As %Status) as WorkMgr

If you have called Detach() on a work queue and have the associated token you can pass this into this class method and assuming the work queue still exists it will create an instance of the work queue manager associated this this queue. If it fails then it will return $$$NULLOREF and set sc with the error %Status value.

classmethod DefaultNumWorkers(category="Default") as %Integer

Return the default number of worker jobs we will use if no specific number is specified

method Detach(ByRef token As %String, timeout As %Integer = 86400) as %Status

Detach this oref from the work queue and set token which you can use in another process (or in this process if wanted) to call Attach(). The timeout is how long in seconds we will keep information about this work queue in the system, so if you do not Attach to this within this period of time we will remove all information about this queue and any subsequent call to Attach() will fail.

classmethod Flush() as %Status

Called from a worker job to flush any output produced so far to the parent process. Without this all output from a worker job is buffered until this unit of work is complete and only then is it displayed in the parent process.

classmethod IsWorkerJob() as %Boolean

Returns true/false based on if the current process is a worker job or not.

method NumberWorkers() as %Integer

Return the number of worker jobs that we will request for this group. We request workers as units of work are queued, the first worker is requested when the first unit of work is queued. Once we have requested this number of workers we will not request any more. Note that after a worker is requested it may not be cause any worker to attach to this queue if they are all busy doing other work. If the group is detached it will return -1.

method Pause(timeout As %Integer, ByRef completed As %Boolean = 0) as %Status

Pause any work in this work queue, this stops any workers from picking up additional items from this queue, but leaves the work itself so you can call Resume() at a later point. When no timeout is passed this will return immediately so there could still be work in progress from one of the work units that was being process at the time this function was called. If you pass in a non-null timeout it will wait for up to this timeout value in seconds for work currently in progress to finish. If after the timeout the work in progress has exited it will set completed=1 else this will be 0.

method Queue(work As %String, args... As %String) as %Status

Queues a specific unit of work, you pass the entry point to call in 'work' argument. This can be either '##class(Classname).ClassMethod' or '$$entry^rtn' and it is expected to return a %Status code on completion. To call a function that does not return any value on completion then prepend the class syntax with '=' e.g. '=##class(Classname).ClassMethod' or to call a function do not include the '$$' e.g. 'entry^rtn'. The item being called may also throw exceptions in order to indicate an error happened which is trapped and converted to a %Status value to be returned in the parent process. You can also pass additional arguments including arrays by reference. Note that the size of the data passed in these arguments should be kept relatively small, if there is a large amount of information that needs to be passed then put this in a global. The security context of the caller is recorded when this function is called and is used when the work is executed. Output the work unit makes to the current device will be trapped and relayed to the parent process after this work unit has completed and when the parent process is waiting for work to be completed. If a work unit wishes to flush output before the work unit has completed then then call 'Do $system.WorkMgr.Flush()'

method QueueCallback(work As %String, callback As %String, args... As %String) as %Status

Similar to Queue() except you can also pass in a 'callback' which is a function or class method that is called in the parent process when this unit of work is complete.
This function is called with the same arguments the original 'work' is called with so it can tell which unit of work is complete.
Also the callback function can access the '%job' public variable which is the $job of the process which really did the work; the '%status' public variable which is the %Status return code from the work unit this is the callback for; the '%workqueue' public variable which is the oref of the work queue instance; and the public variable %result array if it is set in the work unit.
Any error returned by the work unit will by default be added to the Sync() return %Status, but the callback may alter the the work units %Status by modifying the public variable '%status'. For example if the callback detects a specific error %Status from the work unit and does a 'Set %status=$$$OK' it will mark this error as handled and no error %Status will be added to the return from Sync().
If using the Wait() to wait for the work to be completed the callback can signal that it should return to the caller rather than waiting for more events by setting the public variable '%exit' to 1.

method QueueDeferred(Output dtoken As %String, work As %String, args... As %String) as %Status

This method allows a user to queue a deferred unit of work, and then return the deferred token dtoken to use for ActivateDeferred() later. The deferred work unit will not be available to be processed until it's activated using ActivateDeferred() with its corresponding dtoken. This deferrable queueing mechanism allows users to schedule work up front that cannot be processed immediately, for example because the work unit requires data that is not yet available. The deferred token can then be used to activate this work unit when the required data eventually becomes available. This can often be from another process than the process that originally queued said deferred work unit, for example by sharing the corresponding deferred token with separate logic that is triggered when the required data becomes available. This other process can pass the deferred token to ActivateDeferred() to make the deferred work unit available for processing.
PARAMETERS:

dtoken - This is an output variable to output the dtoken representing the deferred work unit. This dtoken can be passed to ActivateDeferred() later to activate the deferred unit.

work - This parameter is the same purpose and format as the work parameter of the Queue() method.

args... - This parameter is the same purpose and format as the args... parameter of the Queue() method.

RETURN VALUE: %Status
Unlike work requested by Queue(), the work requested by this QueueDeferred() call is not made available for a worker job to process, as the unit is in a 'deferred' state until explicitly activated. A work queue can simultaneously contain both normal work units (i.e. units queued by Queue()) and deferred work units (i.e. units queued by QueueDeferred()).
EXAMPLE USAGE:

  Set queue=$system.WorkMgr.%New() If queue="" ; Report Error, can check %objlasterror for %Status code
  For i=1:1:10 {       ; queue some normal work units 1-10.
  	Set sc=queue.Queue("##class(MyClass).ClassMethod",i) If $$$ISERR(sc) ; Report Error
  }
  For i=11:1:20 {      ; queue some deferred work units 11-20.
  	Set sc=queue.QueueDeferred(.dtoken,"##class(MyClass).ClassMethod",i) If $$$ISERR(sc) ; Report Error
  	Set dtokens($i(dtokens))=dtoken ; store the 'dtoken's for these deferred work units
  }
  For i=1:1:dtokens {  ; activate deferred work units 11-20. This can also be done from a different process with the correct 'dtoken's information.
  	Set sc = ##class(%SYSTEM.WorkMgr).ActivateDeferred(dtokens(i))
      ; Or: "Set sc = queue.ActivateDeferred(dtokens(i))"
  	If $$$ISERR(sc)  ; Report Error
  }
  ; Note that here the Sync() will wait for both the 1-10 normal work units and the
  ; 11-20 deferred work units to finish. Without the final loop above to call 
  ; ActivateDeferred() for each 'dtoken', this Sync() will finish the 1-10 normal
  ; work units and then hang, waiting for all of the deferred work units to be activated!
  Set sc=queue.Sync() If $$$ISERR(sc) ; Report Error
  

method Resume() as %Status

Resume any work in this work queue which was paused with a call to Pause().