Tasks provide a way of encapsulating and managing asynchronous actions in a composable manner. They are primarily used to perform some sort of I/O operation and/or manage a series of sub-tasks. Beam provides several classes to define your own tasks as well as classes that connect tasks to one another.

The task API in Beam is defined by two base classes, the class Beam::Tasks::Task which represents a task, and the class Beam::Tasks::TaskFactory which constructs tasks. In this article we will explore how we can use these two classes in ways that allow us to start, stop, modify and resume asynchronous actions.

Interface

Task

The Task class defines two methods, an Execute() method to start the task and a Cancel() method to request that the task stop. Both operations are asynchronous, meaning that they return immediately and the actual operation of the task takes place in a separate thread of execution. In order to keep track of the progress of a task one must monitor its state which can be any of the following:

State	Description
NONE	The task has not yet been executed.
INITIALIZING	The task is performing some initialization, during this state no sub-tasks may be executed.
ACTIVE	The task is running, during this state sub-tasks may be executed.
PENDING_CANCEL	A request to cancel the task has been made, no new sub-tasks may be executed.
CANCELED	The task was canceled.
FAILED	The task could not complete due to an error.
EXPIRED	The task could not complete due to a time constraint.
COMPLETE	The task completed successfully.

The states CANCELED, FAILED, EXPIRED and COMPLETE all represent terminal states. Before a task can transition to a terminal state, all of its sub tasks must be in a terminal state. Furthermore once a task is in a terminal state the task is not permitted to perform any additional action or transition to a different state.

The current state of the task along with all of its transitions is accessed through its Publisher by calling the GetPublisher() method. The publisher will emit objects of type Beam::Tasks::Task::StateEntry which contains two fields as follows:

Name	Type	Description
m_state	Beam::Tasks::Task::State	The state of the task.
m_message	std::string	A message describing the reason for the transition.

In effect a task begins in the NONE state, is then started via the Execute() method, transitions into the INITIALIZATION state where it performs any setup needed, then proceeds to the ACTIVE state where it performs its operation including executing any required sub-tasks, and then finally terminates all of its sub-tasks and performs any final clean-up before ending in a terminal state. During any point after the INITIALIZATION state it may encounter a cancel request (via the Cancel() method) or an error, either of which should result in cleaning-up and terminating its sub-tasks before transitioning to the FAILED or CANCELED state.

Task Factory

The task factory is responsible for creating new tasks by invoking its Create() method. In addition to this a task factory also keeps track of the parameters and state needed to create tasks. Setting a task's parameter is done by calling the factory's Set(name, value) method and retrieving a parameter is done via the Get(name) method.

Continuations

One additional operation that task factories perform is constructing what is called a continuation task. Continuations are a mechanism that allow us to resume a task that previously terminated, but more than that they also allow us to modify the parameters of that task. In a sense, a continuation lets us take a terminated task, modify it, and then continue running it with those modifications.

To do this, task factories have a method PrepareContinuation(task). To use it you pass into it a task that has terminated, then you modify its properties using the Set(name, value) method, and then finally you invoke the Create() method to get a newly constructed task that represents the continuation of the old task.

Not all tasks support continuations, those that do not will throw a Beam::NotSupportedException.

BasicTask

As a great deal of care must be taken to ensure that tasks properly transition from State to State, including managing sub-tasks, handling cancel requests etc... Beam provides the Beam::Tasks::BasicTask as a base class which can be inherited from to take care of much of the work needed to write a proper task. For example it ensures that upon a cancel request that the task transitions into the PENDING_CANCEL state and terminates all managed sub-tasks.

To make use of a the BasicTask you need only implement the OnExecute() method to start your task, and the OnCancel() method to handle cancelling your class.

Example

To showcase a simple example, let's build a task that prints "hello" every second a given number of times. We will write a HelloTask class which inherits from BasicTask, and a HelloTaskFactory.

First Attempt

import beam import datetime import time

class HelloTask(beam.tasks.BasicTask):

 Prints hello a specified number of times.

 def __init__(self, timer, iterations):
    Constructs this task.
   :param timer: The Timer used to wait between successive prints.
   :param iterations: The number of times to print.

   # We must make sure to initialize the base class.
   beam.tasks.BasicTask.__init__(self)
   self.timer = timer
   self.iterations = iterations

   # The number of times we've printed so far.
   self.counter = 0

   # Used to handle timer callbacks.
   self.tasks = beam.RoutineTaskQueue()

 # This overrides BasicTask.on_execute.
 # When called we can assume that our Task is in the INITIALIZATION state.
 def on_execute(self):

   # To initialize our task we will monitor our timer and then start the timer.
   self.timer.get_publisher().monitor(self.tasks.get_slot(self.on_timer))
   self.timer.start()

   # Once the initialization is complete we transition to the ACTIVE state.
   self.set_active()

 # This overrides BasicTask.on_cancel.
 # When called we can assume that our Task is in the PENDING_CANCEL state.
 # No new sub-tasks may be executed.
 def on_cancel(self):

   # In order to synchronize handling the cancel operation with the timer,
   # we will push a helper function onto our RoutineTaskQueue.
   # This ensures no race-conditions take place between the timer and
   # the cancel request.
   self.tasks.push(self._on_cancel)

 # This handles the timer expiry.
 def on_timer(self, result):
   if result == beam.threading.Timer.Result.EXPIRED:

     # This branch implies that our timer expired normally.
     print 'hello'
     self.counter += 1
     if self.counter >= self.iterations:

       # There is nothing more to print, so transition to
       # a terminal state, which by default is the COMPLETE state.
       self.set_terminal()
     else:

       # There are still further iterations, restart the timer.
       self.timer.start()
   else:

       # This branch implies that we canceled the timer in response
       # to a cancel request.
       # In this case we set our state to CANCELED.
       self.set_terminal(beam.tasks.Task.State.CANCELED)

 def _on_cancel(self):
   if self.counter < self.iterations:

     # Only cancel the timer if there are iterations remaining.
     self.timer.cancel()

class HelloTaskFactory(beam.tasks.TaskFactory):

 Builds HelloTasks.

 # Typically the parameters that get passed into the Task are
 # defined as static constant strings in the TaskFactory.
 # This makes it easier to identify the properties of a task
 # as well as reference them (avoiding potential typos).
 ITERATIONS = 'iterations'

 def __init__(self, timer_factory):
   Constructs a HelloTaskFactory.
   :param timer_factory: Used to construct Timers.

   # Factories will typically inherit from TaskFactory as a base class.
   beam.tasks.TaskFactory.__init__(self)
   self.timer_factory = timer_factory

   # The constructor should define all the properties and default
   # values for those properties.
   self.define_property(HelloTaskFactory.ITERATIONS, 10)

   # This variable keeps track of continuations.
   self.continuation_task = None

 # This overrides the TaskFactory create method.
 def create(self):

   if self.continuation_task is None:

     # We are not creating a continuation task, so all we need to do is
     # construct a HelloTask using the ITERATIONS property defined above.
     return HelloTask(self.timer_factory(datetime.timedelta(seconds = 1)),
       self.get(HelloTaskFactory.ITERATIONS))
   else:

     # We are creating a continuation task.  The continuation of a HelloTask of
     # N iterations that has already printed C times is basically a HelloTask
     # that prints N - C times.
     continuation = HelloTask(
       self.timer_factory(datetime.timedelta(seconds = 1)),
       self.get(HelloTaskFactory.ITERATIONS) - self.continuation_task.counter)
     self.continuation_task = None
     return continuation

 # This overrides the TaskFactory prepare_continuation method.
 def prepare_continuation(self, task):

   # We store the task to continue.
   self.continuation_task = task

def main():

 # Construct the HelloTaskFactory using the LiveTimer for 5 iterations.
 factory = HelloTaskFactory(beam.threading.LiveTimer)
 factory.set(HelloTaskFactory.ITERATIONS, 5)
 task = factory.create()
 task.execute()

 # Let's sleep for 2 seconds before canceling the task.
 # It should print hello twice.
 time.sleep(2)
 task.cancel()

 # Wait for the task to enter the CANCELED state.
 beam.tasks.wait(task)

 # Build the continuation task and execute it.  To do this we first
 # call prepare_continuation, then we make our modifications using the set
 # method, then we create the task and execute it.
 factory.prepare_continuation(task)
 factory.set(HelloTaskFactory.ITERATIONS, 7)
 task = factory.create()

 # We expect it to print hello three times before coming to an end.
 task.execute()
 beam.tasks.wait(task)

if __name__ == '__main__':

 main()

Decomposition

The above example works, but upon reflection we should notice that this Task is responsible for two distinct things. One is the responsibility of printing, and the other is the responsibility of repeating. Given that the purpose of tasks is to be able to compose asynchronous operations we should separate these two responsibilities from one another. To do that we will change our HelloTask so that all it does is print hello after a specified time period, followed by a RepetitionTask which repeats a task a specified number of times. The benefit of this is that our RepetitionTask can be reused to repeat any task whatsoever down the road.

We will define it as follows:

class RepetitionTask(beam.tasks.BasicTask):

 Repeats a Task a specified number of times.

 def __init__(self, task_factory, iterations):
   Constructs the Task.
   :param task_factory: Builds the Task to repeat.
   :param iterations: The number of times to repeat the Task.
   
   beam.tasks.BasicTask.__init__(self)

   # We should always make a deep copy of factories in order to
   # avoid modifying a factory belonging to another task or having
   # another task modify our factory.
   self.task_factory = copy.deepcopy(task_factory)
   self.iterations = iterations
   self.counter = 0

   # This stores the task currently being executed.
   self.task = None

   # This is used to handle callbacks from our tasks.
   self.tasks = beam.RoutineTaskQueue()

 def on_execute(self):

   # Defer to a helper function.
   self.execute_task()

 def on_cancel(self):

   # As before, to avoid race conditions between cancels and
   # our task we will push a callback onto a RoutineTaskQueue
   # to handle cancellations.
   self.tasks.push(self._on_cancel)

 def on_state(self, state_entry):

   # This method handles transitions of the task we're repeating.
   if state_entry.state == beam.tasks.Task.State.CANCELED:

     # This branch indicates that we canceled our task which
     # means that we're handling a cancel request.
     self.set_terminal(beam.tasks.Task.State.CANCELED)
   elif beam.tasks.is_terminal(state_entry.state):

     # This branch indicates that our task terminated and
     # hence we should repeat.
     self.execute_task()

 def _on_cancel(self):
   if self.counter < self.iterations:

     # Similar to before, only cancel if we still have
     # repetitions to process.
     self.task.cancel()

 def execute_task(self):
   if self.counter >= self.iterations:

     # This branch indicates there are no more iterations left.
     self.set_terminal()
   else:

     # This branch indicates that we need to repeat the task
     # by constructing a new one and executing it.
     self.counter += 1
     self.task = self.task_factory.create()
     self.task.get_publisher().monitor(self.tasks.get_slot(self.on_state))
     self.task.execute()

This class is very similar to the HelloTaskFactory.

class RepetitionTaskFactory(beam.tasks.TaskFactory):

 ITERATIONS = 'iterations'

 def __init__(self, task_factory):
   beam.tasks.TaskFactory.__init__(self)
   self.task_factory = copy.deepcopy(task_factory)
   self.define_property(RepetitionTaskFactory.ITERATIONS, 10)
   self.continuation_task = None

 def create(self):
   if self.continuation_task is None:
     return RepetitionTask(self.task_factory,
       self.get(RepetitionTaskFactory.ITERATIONS))
   else:
     continuation = RepetitionTask(self.task_factory,
       self.get(RepetitionTaskFactory.ITERATIONS) -
       self.continuation_task.counter)
     self.continuation_task = None
     return continuation

 def prepare_continuation(self, task):
   self.continuation_task = task

Now that we have factored out the job of repeating tasks we can rewrite our HelloTask as follows: