worker Module

exception pipelet.worker.AbortError(signal=0)

Bases: exceptions.Exception

Extension of the Exception class.

class pipelet.worker.InteractiveWorker(scheduler, logger=None)

Bases: pipelet.worker.Worker

Run segments in interactive mode.

Exceptions are not caught to ease debugging. After a debugging session you may want to call self.terminate() to properly stop the scheduler and update the database.

exec_code(code, glo, env)

Reimplement exec_code from Worker.

Main difference is that exception occuring in the segment code are not caught.

matplotlib_hook()

terminate()

Try to stop the processing of the ongoing task after a bug.

class pipelet.worker.NullHandler(level=0)

Bases: logging.Handler

Extension of the logging handler class.

emit(record)

Avoid warnings.

class pipelet.worker.ProcessWorker(address=('', 50000), authkey='secret', nice=0, **keys)

Bases: pipelet.worker.Worker, multiprocessing.process.Process

The worker is run as an independant process

It access the scheduler through the use of managers

run()

class pipelet.worker.SandBox(pipeline)

Bases: pipelet.worker.Worker

Provide a mean to come back in the execution state of a segment for latter debugging.

setup(seg, global_env)

Setup the working environment for segment seg.

class pipelet.worker.SchedulerManager(address=None, authkey=None, serializer='pickle')

Bases: multiprocessing.managers.BaseManager

Extension of the BaseManager class.

get_scheduler(*args, **kwds)

class pipelet.worker.ThreadWorker(*args, **keys)

Bases: pipelet.worker.Worker, threading.Thread

Run segments in a separate thread.

Jobs are obtained from a QueueScheduler. Exceptions are caught and dumped.

class pipelet.worker.Worker(scheduler, logger=None)

Bases: object

Empty tasks of an execution queue.

A worker is linked to a scheduler. Its job is to empty the scheduler task queue.

This is the base class for all workers type.

exec_code(code, glo, env)

Execute code string with namespace.

Task status is updated after execution. Exceptions are read from the environment log file.

execute_task(task)

Execute a task.

Make task directory and execute segment code string with the environment feeded with the task input. Task output is read from the environment after execution and updated to the task object.

make_dir(task)

Create a directory for a given task.

matplotlib_hook()

Turn the matplotlib backend to Agg.

prepare_env(task)

Build the segment global execution environment for a given task.

Include parameters and a bunch of utilitary routines (see the following members).

run()

Start the worker.

The worker executes the tasks given by its scheduler until the queue is empty. The task status are updated after completion or failure.

terminate()

Disconnect from scheduler.

pipelet.worker.catch_sigterm(signal, frame)