Resident Node Classes (Python)

The processes of a resident program (see the Resident DeBasher Programs Section) are Python classes that derive from the classes below, which the module imports from debasher_runtime_lib:

from debasher_runtime_lib import FBPProcess

Only what a module uses is listed: the hooks that a node redefines, the methods that it calls, and the class attributes that it may set. A subclass sets a class attribute for every process that uses it; some of them can also be set for one process of a program, in the computational specifications that add_debasher_process gives it, as the description of each says.

FBPProcess

class debasher_runtime_lib.FBPProcess

The base class of every node of a resident program. A module defines a node with a subclass, named after its process in CamelCase, that redefines the four hooks process_data(), capture_node_state(), restore_node_state() and initialize_runtime(), and whose Python code creates it and calls run(): Counter().run().

The ports of the node are the options of its process connected to a FIFO, named without their leading dash (inf for -inf). The engine gives them to the node from the options of its module, so the class declares none. Every option of the process is in self.opts, a dict from the name of the option, without its dash, to its value, a string; a flag, an option given with no value, is True when it is given, and its name has to be listed in the class attribute FLAGS (for example FLAGS = ("verbose",)). self.log is a logger from the standard logging module, which writes to the standard error of the process.

A node acts only in reaction to what it receives: the framework calls process_data() once for each message, and a node sends only from there. After a crash, the node is restored from its latest checkpoint and process_data() is called again for every message received since, in the same order, so that the node returns to the state it had.

Hooks

FBPProcess.process_data(port_name, packet)

Processes one message, packet, received on the input port port_name (or under OBSERVE_PORT, from observe()). packet is the value that the sender gave to send_data(), already decoded from JSON. It is the only place from which the node sends, with send_data(). It must be deterministic: given the same node state and the same messages, it makes the same new state and sends the same messages, since a replay after a crash calls it again for every message received since the latest checkpoint. It must not depend on the time, on randomness or on reads from outside the program, and whatever it does outside its FIFOs (writing a file, calling a service) may be done more than once.

FBPProcess.capture_node_state()

Returns the node state: everything that process_data() keeps from one message to the next, as a value that JSON can serialize. It is called when a snapshot is taken, and what it returns is saved in a checkpoint of the node.

FBPProcess.restore_node_state(node_state)

Sets the node state from node_state, what capture_node_state() returned, when the node starts from a checkpoint. It must rebuild the state exactly. It is not called when the node starts with no checkpoint, so the constructor sets the initial state. It cannot send.

FBPProcess.initialize_runtime()

Opens what the node uses and is not part of its state, such as a connection or an open file, before it processes any message. It is called at every start of the node, after restore_node_state(). It cannot send.

FBPProcess.observe()

Optional. Looks at the outside world (a directory, a queue, the batch runs of a launcher node) and brings what it sees into the node with inject(). It runs on a thread of its own, every OBSERVE_INTERVAL_SECS or when observe_now() is called, only while the node is running, never while it replays its input log. What it sees is different every time, so it cannot happen in process_data(): it enters as an input, which a replay receives again. After a crash, observe() may bring in again what it had already brought in, so process_data() keeps in the node state what it has acted on, and drops a repetition. If observe() raises, the node stops sending heartbeats, as a node that crashed does.

Methods

FBPProcess.run()

Runs the node until it is told to stop. The Python code of the process calls it once, after creating the node. It restores the node from its latest checkpoint with restore_node_state(), calls initialize_runtime(), calls process_data() again for every message received since that checkpoint, and then processes what arrives.

FBPProcess.send_data(tag, payload)

Sends payload, any value that JSON can serialize, on the output port tag. A node acts only in reaction to what it receives, so it may be called only from process_data(), and raises anywhere else. It also raises when what the node has not yet written into its FIFOs would go over OUT_BACKLOG_FAIL_BYTES. When it raises, nothing is sent.

FBPProcess.sleep(seconds)

Waits seconds, from process_data(), to set the pace of a node that emits on its own through a self-loop. It does not wait while the node replays its input log after a crash, where the pace would only delay the recovery, and it returns early once the node is told to stop. Use it instead of time.sleep(): it changes only when the next step comes, never the node state or what is sent.

FBPProcess.inject(payload)

Brings payload, something that observe() saw, into the node as a message: process_data() receives it with OBSERVE_PORT as its port_name. The message is recorded like any other, so that a replay after a crash receives it again without looking at the world again. It returns once the message is recorded. It raises when called from process_data(), since an input made by processing would be made again by every replay, or when the class has no OBSERVE_PORT.

FBPProcess.observe_now()

Makes observe() run now, and not at the end of its interval.

FBPProcess.set_notice(text, level='info')

Leaves a notice for whoever watches the program, such as that the configuration file of the node is missing. A node has at most one notice, and a new one replaces the one it had. level is “info” or “warning”; a text longer than NOTICE_MAX_CHARS is cut, with a warning in the log. It may be called from any hook and any thread. The notice is not part of the node state, and each incarnation of the node starts without one, so a notice that follows from the node state is set again in restore_node_state(). debasher_inspect_resident shows it, and the web interface marks the node that has one.

FBPProcess.clear_notice()

Takes away the notice of the node, if it has one (see set_notice).

Class attributes

FBPProcess.OBSERVE_PORT = None

The name under which what observe() brings in with inject() reaches process_data(), as its port_name. It is not a FIFO and not one of the ports of the node. None, the default, for a node that observes only to act, which cannot call inject().

FBPProcess.OBSERVE_INTERVAL_SECS = 1.0

How often, in seconds, observe() runs.

FBPProcess.HEARTBEAT_INTERVAL_SECONDS = 5

How often, in seconds, the node tells the Supervisor of the program that it is alive.

FBPProcess.INPUT_LOG_MAX_BYTES = 104857600

The size, in bytes, that the input log of the node may reach (100 MiB by default). The log holds what the node received since its latest checkpoint, so it only reaches this size when no snapshot has been taken for a long time; the node then stops with an error. A program sets it for one of its processes with the computational specification input_log_max_mb.

FBPProcess.OUT_BACKLOG_MAX_BYTES = 8388608

The size, in bytes, of what send_data() has taken and the node has not yet written into its FIFOs, since their readers are slow, over which the node skips a checkpoint and waits for the next round (8 MiB by default). A program sets it for one of its processes with the computational specification out_backlog_max_mb.

FBPProcess.OUT_BACKLOG_FAIL_BYTES = 67108864

The size, in bytes, of what send_data() has taken and the node has not yet written into its FIFOs, over which send_data() raises, since a reader of the node is not reading (64 MiB by default). A program sets it for one of its processes with the computational specification out_backlog_fail_mb.

FBPProcess.GIL_SWITCH_INTERVAL_SECS = 0.0005

The GIL switch interval of the process of the node, in seconds (0.5 ms by default). A program sets it for one of its processes with the computational specification gil_switch_interval_ms.

FBPProcess.NOTICE_MAX_CHARS = 1000

The length at which set_notice() cuts the text of a notice.

DirectoryWatcher

class debasher_runtime_lib.DirectoryWatcher

A node that watches a directory and sends a request for each file that arrives there, typically to a launcher node (see ProgramLauncher). A module declares it with a subclass that sets WATCH_DIR (or gives the process the option -watchdir) and, if needed, PATTERN.

WATCH_DIR is the directory to watch: an absolute path, or one relative to the directory of the module that declares the node; an option of the node named WATCH_DIR_OPTION (-watchdir), if it has one, gives it instead, for a directory that the command line chooses. The files are those whose name matches PATTERN, never one whose name starts with a dot, which is how a file that is still being written is often named.

A file is complete once its size and its modification time have stayed the same for STABLE_OBSERVATIONS observations in a row, OBSERVE_INTERVAL_SECS apart. A module whose files are complete in another way (a companion file, a rename at the end of a copy) redefines is_complete().

For each complete file, the node sends on REQUESTS_PORT what request_for() makes of its path: by default a request for a launcher node (see ProgramLauncher), with the path as the value of FILE_OPTION and the name of the file without its extension as the name of the run.

Methods a module may redefine

DirectoryWatcher.request_for(path)

The request sent for the complete file at path: by default {"opts": {FILE_OPTION: path}, "run": <name of the file without its extension>}. A module redefines it to send something else.

DirectoryWatcher.is_complete(path, stable_observations)

Whether the file at path is complete, after its size and modification time have not changed for stable_observations observations in a row.

Class attributes

DirectoryWatcher.WATCH_DIR = None

The directory to watch, an absolute path or one relative to the directory of the module.

DirectoryWatcher.WATCH_DIR_OPTION = 'watchdir'

The option of the process, without its dash, that gives the directory to watch instead of WATCH_DIR, when the process has it.

DirectoryWatcher.PATTERN = '*'

The pattern, in the syntax of the shell, that the name of a file has to match.

DirectoryWatcher.STABLE_OBSERVATIONS = 2

How many observations in a row the size and modification time of a file have to stay the same for the file to be complete.

DirectoryWatcher.REQUESTS_PORT = 'outrequests'

The output port on which the requests are sent.

DirectoryWatcher.FILE_OPTION = '-infile'

The option of the general program to launch that receives the path of the file, in the default request.

ProgramLauncher

class debasher_runtime_lib.ProgramLauncher

A node that launches a general program once for each request that it receives, each in a run directory of its own, and can tell a node downstream when each of those batch runs ends.

A subclass names the general program to launch, PFILE, as a module that declares the node would load it: a relative path is looked for in the directory of that module, where a program keeps its files, and then in the directories of DEBASHER_MOD_DIR; an absolute one is accepted with a warning, since it ties the program to one machine. With PROCESS, the name of a process of that module, it launches that process alone, with debasher_exec_process, instead of the whole program. It may give RUNS_ROOT, an absolute path under which the run directories go; without it, the run directories go in the output directory of the process.

An option of the node gives any of the three instead, for a value that the command line chooses, when the node has it: PFILE_OPTION (-pfile), PROCESS_OPTION (-process) and RUNS_ROOT_OPTION (-runsdir). -pfile is not a file option (define_cmdline_opt, not define_cmdline_infile_opt), so that the engine leaves a relative name as it is and the node looks for it as PFILE; -runsdir has to be an absolute path.

Every input port is a port of requests. A request is a JSON object with opts, the options of the general program (names with their leading dash, values as strings), and optionally run, the name of its run directory, a relative path none of whose parts starts with a dot: {"opts": {"-bam": "/data/s17.bam"}, "run": "s17"}. Without run, the run directory is named after the position of the request. A request that does not follow this, or whose run directory belongs to another request, is logged as an error and dropped.

If the node has an output port DONE_PORT, observe() brings in the end of every batch run under the name RUNS_DONE_PORT (its observe port, which is not a fifo), and the node sends on DONE_PORT a message for each, {"run": ..., "run_dir": ..., "status": ..., "exit_code": ...}, with run_dir the absolute path of the run directory, and status “finished” for an exit code of 0 and “failed” for any other.

A launcher node cannot be a task of an array process.

Class attributes

ProgramLauncher.PFILE = None

The module of the general program to launch (required, unless the process has the option PFILE_OPTION).

ProgramLauncher.PFILE_OPTION = 'pfile'

The option of the process, without its dash, that gives the module of the general program instead of PFILE, when the process has it.

ProgramLauncher.PROCESS = None

The name of a process of PFILE to launch alone, instead of the whole program, or None.

ProgramLauncher.PROCESS_OPTION = 'process'

The option of the process, without its dash, that gives the process to launch alone instead of PROCESS, when the process has it.

ProgramLauncher.RUNS_ROOT = None

The absolute path under which the run directories go, or None for the output directory of the process.

ProgramLauncher.RUNS_ROOT_OPTION = 'runsdir'

The option of the process, without its dash, that gives the runs root instead of RUNS_ROOT, as an absolute path, when the process has it.

ProgramLauncher.DONE_PORT = 'outdone'

The output port on which the end of each batch run is sent, if the node has it.

ProgramLauncher.MAX_CONCURRENT_RUNS = 1

How many batch runs run at a time. A program sets it for a process in its computational specifications, max_concurrent_runs.

ProgramLauncher.BATCH_SCHED = 'BUILTIN'

The scheduler the batch runs use, given to debasher_exec as --sched: the built-in one by default, as for the resident program itself, so that the batch runs do not depend on the scheduler a machine would pick on its own. A program sets it for a process in its computational specifications, batch_sched (BUILTIN or SLURM).

ProgramLauncher.STATUS_CHECK_INTERVAL_SECS = 5.0

How often, at most, in seconds, observe() asks debasher_status how a batch run is going (OBSERVE_INTERVAL_SECS is how often it looks at the run directories).

Supervisor

class debasher_runtime_lib.Supervisor

The optional process of a resident program, at most one, that watches its nodes. Every node sends it heartbeats; when they stop, or the process of a node is gone, the Supervisor relaunches the node, and it gives up on a node that keeps failing, stopping the rest of the program in order. It relays to the initiators the triggers that start a snapshot or a shutdown, written into its manual trigger port from outside the program. It also holds open the FIFO of every channel between two nodes, so that what a FIFO holds outlives the crash of both of them, unless the process is given the flag -no-hold-fifos.

A module declares it with an empty subclass, class Sup(Supervisor): pass, whose ports the engine gives, like those of a node. It keeps no state: a Supervisor relaunched by hand starts afresh.

Methods a module may redefine

Supervisor.on_node_down(node_name)

Called when the node node_name is declared down. By default it relaunches the node exactly as the program first launched it. A subclass may redefine it for a node that needs something else.

Supervisor.on_node_permanently_failed(node_name)

Called when the Supervisor gives up on the node node_name, after MAX_RELAUNCH_ATTEMPTS relaunches. By default it stops the rest of the program in order with debasher_stop_resident, and kills it if that takes longer than FORCE_STOP_TIMEOUT_SECS.

Class attributes

Supervisor.HEARTBEAT_TIMEOUT_SECS = 30

How long, in seconds, a node may go without sending a heartbeat before it is declared down. A program sets it with the computational specification heartbeat_timeout_s of the Supervisor.

Supervisor.HEARTBEAT_CHECK_INTERVAL_SECS = 5

How often, in seconds, the Supervisor checks the heartbeats.

Supervisor.STARTUP_TIMEOUT_SECS = None

How long, in seconds, a node has, from each launch, to send its first heartbeat, which comes once it has restored its checkpoint and replayed its input log; never shorter than HEARTBEAT_TIMEOUT_SECS. A program sets it with the computational specification startup_timeout_s, of the Supervisor for every node, or of a node for that node.

Supervisor.MAX_RELAUNCH_ATTEMPTS = 3

How many times in a row a node is relaunched without sending a heartbeat before the Supervisor gives up on it.

Supervisor.FORCE_STOP_TIMEOUT_SECS = 60

How long, in seconds, the orderly stop that follows giving up on a node may take before the program is killed.