Adding Tasks

In this section, we will take a look at how we can add more tasks to our base script from the Getting Started section.

Note

The reader is assumed to be familiar with the PST Model and to have read through the Introduction of Ensemble Toolkit.

Note

This chapter assumes that you have successfully installed Ensemble Toolkit, if not see Installation.

You can download the complete code discussed in this section here or find it in your virtualenv under share/radical.entk/user_guide/scripts.

Below, you can see the code snippet that shows how you can create more Task objects and add them to the Stage using the add_task() method.

    t.name = 'my.task'        # Assign a name to the task (optional, do not use ',' or '_')
    t.executable = '/bin/echo'   # Assign executable to the task
    t.arguments = ['I am task %s' % cnt]  # Assign arguments for the task executable

    # Add the Task to the Stage
    s.add_tasks(t)

# Add Stage to the Pipeline
p.add_stages(s)

# Create Application Manager
appman = AppManager()

python add_tasks.py

Note

Individual Task must be created each time in order to treat all the items uniquely inside a Stage. For example, for loop can be used for creating multiple tasks and creating Task object needs to be included inside iteration.

# Five tasks to be added
s = Stage()
task_dict = {
    't.cpu_reqs': {
    'processes'          : 10,
    'threads_per_process': 1,
    'process_type'       : "MPI",
    'thread_type'        : "OpenMP"
    }}

for i in range(5):
    task_dict['name']      = "task-{}".format(i)
    task_dict['arguments'] = ["file-{}".format(i)]
    # Creating new Task object and adding to Stage at every iteration
    s.add_tasks(Task(from_dict=task_dict))

print("Adding shared tasks. Total: {}".format(len(s.tasks)))

Let’s take a look at the complete code in the example. You can generate a more verbose output by setting the environment variable RADICAL_ENTK_VERBOSE=DEBUG.

A look at the complete code in this section:

#!/usr/bin/env python

from radical.entk import Pipeline, Stage, Task, AppManager
import os

# ------------------------------------------------------------------------------
# Set default verbosity
if os.environ.get('RADICAL_ENTK_VERBOSE') is None:
    os.environ['RADICAL_ENTK_REPORT'] = 'True'


if __name__ == '__main__':

    # Create a Pipeline object
    p = Pipeline()

    # Create a Stage object
    s = Stage()

    for cnt in range(10):

        # Create a Task object
        t = Task()
        t.name = 'my.task'        # Assign a name to the task (optional, do not use ',' or '_')
        t.executable = '/bin/echo'   # Assign executable to the task
        t.arguments = ['I am task %s' % cnt]  # Assign arguments for the task executable

        # Add the Task to the Stage
        s.add_tasks(t)

    # Add Stage to the Pipeline
    p.add_stages(s)

    # Create Application Manager
    appman = AppManager()

    # Create a dictionary describe four mandatory keys:
    # resource, walltime, and cpus
    # resource is 'local.localhost' to execute locally
    res_dict = {

        'resource': 'local.localhost',
        'walltime': 10,
        'cpus': 1
    }

    # Assign resource request description to the Application Manager
    appman.resource_desc = res_dict

    # Assign the workflow as a set or list of Pipelines to the Application Manager
    # Note: The list order is not guaranteed to be preserved
    appman.workflow = set([p])

    # Run the Application Manager
    appman.run()