Distribution

Before proceeding with this section, the reader should ensure they are familiar with the recipe author’s perspective on how tasks can be distributed within the pipeline framework. This section will describe in detail what goes on behind the scenes and how it can be extended.

Todo

Details!

Node scripts

class lofarpipe.support.lofarnode.LOFARnode(loghost=None, logport=9020)

Base class for node jobs called through IPython or directly via SSH.

Sets up TCP based logging.

run_with_logging(*args)

Calls the run() method, ensuring that the logging handler is added and removed properly.

class lofarpipe.support.lofarnode.LOFARnodeTCP(job_id, host, port)

This node script extends LOFARnode to receive instructions via TCP from a JobSocketReceiver.

run_with_stored_arguments()

After fetching arguments remotely, use them to run the standard run_with_logging() method.

Compute jobs

class lofarpipe.support.remotecommand.ComputeJob(host, command, arguments=[])

Container for information about a job to be dispatched to a compute node.

Parameters:

• host – Target host for job
• command – Full path to command to be run on target host
• arguments – List of arguments which will be passed to command

dispatch(logger, config, limiter, id, jobhost, jobport, error, killswitch)

Dispatch this job to the relevant compute node.

Note that error is an instance of threading.Event, which will be set if the remote job fails for some reason.

Scheduling and rate limiting

See RemoteCommandRecipeMixIn.

class lofarpipe.support.remotecommand.ProcessLimiter(nproc=None)

Provide a dictionary-like structure of bounded semaphores with arbitrary keys.

This gives a convenient way to keep tabs on the number of simultaneous jobs running on a given host.

Parameters:nproc (integer or none) – Bound value for semaphore (ie, maximum number of jobs)

Dispatch mechanisms

Todo

Extenting this system.

Dispatch mechanism can be specified in the remote section of the pipeline configuration

lofarpipe.support.remotecommand.run_remote_command(config, logger, host, command, env, arguments=None)

Run command on host, passing it arguments from the arguments list and exporting key/value pairs from env(a dictionary).

Returns an object with poll() and communicate() methods, similar to subprocess.Popen.

This is a generic interface to potentially multiple ways of running commands (SSH, mpirun, etc). The appropriate method is chosen from the config block supplied (with SSH as a fallback).

lofarpipe.support.remotecommand.run_via_mpirun(logger, host, command, environment, arguments)

Dispatch a remote command via mpirun.

Return a Popen object pointing at the MPI command, to which we add a kill method for shutting down the connection if required.

lofarpipe.support.remotecommand.run_via_paramiko(logger, host, command, environment, arguments, key_filename)

Dispatch a remote command via paramiko.

We return an instance of ParamikoWrapper.

class lofarpipe.support.remotecommand.ParamikoWrapper(paramiko_client, command)

Sends an SSH command to a host using paramiko, then emulates a Popen-like interface so that we can pass it back to pipeline recipes.

Exchanging data between master and nodes

lofarpipe.support.jobserver.job_server(*args, **kwds)

Provides a context in which job dispatch is available.

Yields a host name & port which clients can connect to for job details.

class lofarpipe.support.jobserver.JobSocketReceiver(logger, jobpool, error, host=None, port=9020)

Bases: SocketServer.ThreadingTCPServer

Simple TCP socket-based job dispatch and results collection as well as network logging.

class lofarpipe.support.jobserver.JobStreamHandler(request, client_address, server)

Bases: SocketServer.StreamRequestHandler

Networked job server.

This will listen for:

• GET <jobid> – reply with a list of all job arguments
• PUT <jobid> <results> – receive and unpickle job results
• Pickled log records

Log records are logs using whatever local logger is supploed to the SocketReceiver.

Clean shut-down and recovery

lofarpipe.support.remotecommand.threadwatcher(threadpool, logger, killswitch)

Start and watch a pool of threads. If an exception is thrown during processing, set the killswitch so that all threads can shut down cleanly, then join all the threads to wait for them to finish.

Parameters:

• threadpool – Pool of threads to handle
• logger (logging.Logger or descendant) – Logger
• killswitch (threading.Event) – Indication for threads to abort