3.3. pyopus.parallel.cooperative — Cooperative multitasking OS with task outsourcing

Cooperative multitasking OS with task outsourcing (PyOPUS subsystem name: COS)

This module is based on the greenlet module. Concurrent tasks can be created in a UNIX-like fashion (with the Spawn() method). The return value of a task or multiple tasks is collected with the Join() method. Joins can be blocking or nonblocking.

The cooperative multitasking OS takes care of outsourcing the tasks to computing nodes if it is permitted to do this and there are computing nodes available. Outsourcing uses the virtual machine specified by calling the setVM() method. If no virtual machine is specified outsourcing is not possible.

COS makes it possible to implement multilevel parallelism and asynchronous algorithms in a simple manner. Parallel algorithms can be run on a single CPU by taking advantage of the greenlet module for providing the microthread functionality. Every local task is a microthread that runs concurrently with other microthreads. The microthreads are cooperatively scheduled by COS. Of course such a run is slower than a real parallel run involving multiple tasks across multiple processors.

class pyopus.parallel.cooperative.Task(parent, greenlet=None, remoteTaskID=None, name=None, func=None, args=[], kwargs={})

Task wrapper object. Wraps one microthread or one remote task.

Arguments:

• parent - parent task. Should be None for teh main task.
• greenlet - greenlet of a local task
• remotetaskID - ID of a remote task
• name - task name. If None a name is generated. from greenlet or remoteTaskID
• args - positional arguments to the greenlet passed at startup
• kwargs - keyword arguments to the greenlet passed at startup

If args and kwargs are None the greenlet is assumed to be already running.

Members:

• tid - task id
• ptid - parent task id
• nchildren - number of children
• finishedChildren - dictionary of finished children waiting to be joined. The key is the tid of a child.
• sendval - return value passed to the task at switch
• retval - value returned by the finished task
• waitingOn - list of child task ids the task is waiting on
• status - 0 (created), 1 (running), 2 (stopped), 3 (finished)

localThrow(exception, value)

Throws an exception with value value in the local task represented by this object.

switchToLocal()

Switch control to the local task (microthread) represented by this object.

class pyopus.parallel.cooperative.SysCall

Base class for COS system calls.

class pyopus.parallel.cooperative.Spawn(func, args=[], kwargs={}, remote=False, block=True)

System call for spawning a child task.

• func - function defining the task
• args - positional arguments to the task’s function
• kwargs - keyword arguments to the task’s function
• remote - True if a remote spawn is requested. Spawns a local task if the scheduler has no VM object or remote is False.
• block - True if a remote spawn should block until a slot is free. Has no effect for local spawns. Nonblocking remote spawn returns -1 on failure.

Returns the tid of a spawned task.

class pyopus.parallel.cooperative.Yield

A system call for yielding the control to the scheduler.

The scheduler switches the context to the next scheduled microthread.

class pyopus.parallel.cooperative.Join(tidList=[], block=True)

A system call for joining a child task. Return the task’s tid and its return value.

• tidlist - list of task IDs that we are waiting on. Empty list waits on all child tasks.
• block - True if the call should block until a task can be joined

Returns a dictionary with tid for key holding the return value of the task that was joined. The dictionary has a single entry.

Returns an empty dictionary if there are no children.

Failed nonblocking join returns an empty dictionary.

class pyopus.parallel.cooperative.Scheduler(vm=None)

Cooperative multitasking scheduler based on greenlets.

vm - virtual machine abstraction for spawning remote tasks.

The main loop of the scheduler is entered by calling the scheduler object.

countLocalTasks()

Returns the number of running local tasks including the scheduler and the main task.

countRemoteTasks()

Returns the number of running remote tasks.

countTasks()

Returns the number of running tasks including the scheduler and the main task.

enqueueSpawn(spawnSyscall, spawnerTask)

Equeues a spawn system call waiting on a free slot.

new(parent, func, args=[], kwargs={}, remote=False)

Create a new task.

parent - parent task object func - function defining the task args - positional arguments to the task’s function kwargs - keyword arguments to the task’s function remote - True for a remote task

schedule(task)

Schedules a local task for execution.

setVM(vm)

Sets the VM abstraction object used for spawning remote tasks.

Allowed only when no remote tasks are running. Setting a VM object on remote task has no effect.

class pyopus.parallel.cooperative.OpusOS(vm=None)

Cooperative multitasking OS class.

The user should import the only instance of this class represented by the cOS variable.

vm - virtual machine abstraction for spawning remote tasks.

If vm is None remote spawning is disabled.

static Join(*args, **kwargs)

Invokes the Join system call.

See Join.

static Spawn(*args, **kwargs)

Invokes the Spawn system call.

See Spawn.

static Yield()

Invokes the Yield system call.

See Yield.

static dispatch(jobList, collector=None, remote=True, buildResultList=None)

Dispatches multiple jobs and collects the results. If remote is True the jobs are dispatched asynchronously across the available computing nodes.

A job is a tuple of the form (callable, args, kwargs, extra). A job is evaluated by invoking the callable with given args and kwargs. If kwargs is omitted only positional arguments are passed. If args is also omitted the callable is invoked without arguments. The return value of the callable is the job result. Extra data (extra) can be stored in the optional entries after kwargs. This data is not passed to the callable.

jobList is an iterator object (i,e, list) that holds the jobs. It can also be a generator that produces jobs. A job may not be None.

collector is an optional unprimed coroutine. When a job is finished its result is sent to the collector in the form of a tuple (index, job, result) where index is the index of the job. Collector’s task is to collect the results in a user-defined data structure.

By catching the GeneratorExit exception in the collector a postprocessing step can be performed on the collected results.

Returns the list of results (in the order of generated jobs). When a collector is specified the results are not collected in a list unless buildResultList is set to True.

static dispatchSingle(function, args=[], kwargs={}, remote=True)

Dispatches a single task defined by function, args, and kwargs. If remote is True the task is dispatched to a remote computing node.

This function is used for moving a task to a remote processor and waits for the results. It is not very useful in the sense that it does not introduce any parallelism.

Returns the return value of the function.

static finalize()

Performs cleanup. Calls the finalize() method of the vm.

static freeSlots()

Returns the number of free slots in a vm. If there is no vm, returns -1.

This is not a system call and does not yield execution the the scheduler.

static getTid()

Returns the task id of the running microthread.

This is not a system call and does not yield execution the the scheduler.

setVM(vm=None)

Sets the virtual machine object.

Allowed only when there are no remote tasks running.

This is not a system call and does not yield execution the the scheduler.

static slots()

Returns the number of slots in a vm. If there is no vm, returns -1.

This is not a system call and does not yield execution the the scheduler.

pyopus.parallel.cooperative.cOS

Cooperative multitasking OS instance for accessing its functionality.