pyantz.infrastructure.runner.local package

pyantz.infrastructure.runner.local package#

Submodules#

pyantz.infrastructure.runner.local.local_runner module#

A local runner executes code on the device.

The jobs are put in a SqliteQueue in the working directory, and worker processes (or this process, with use_same_proc) take and run them until the queue is complete.

DEADLOCK_POLLS = 3#: Consecutive polls with only pending jobs before they are declared deadlocked.

DEATH_CHECK_SECONDS = 0.2#: How often the pool checks whether a worker died.

STARTUP_WAIT_SECONDS = 60.0#: Seconds to wait for a job’s process to start.

STOP_WAIT_SECONDS = 5.0#: Seconds to wait for a job’s process to stop after being asked to.

WORKER_STOP_TIMEOUT = 10.0#: Seconds workers get to mark their job as failed before being killed.

exception WorkerTerminatedError[source]#

Bases: BaseException

Raised inside a worker process when it is asked to terminate (SIGTERM).

Deliberately not an Exception so that job code cannot swallow it.

_POOL_CONTEXT = <multiprocessing.context.ForkServerContext object>#: Multiprocessing context of the worker processes.

_check_for_deadlock(queue: SqliteQueue, report: JobsReport, deadlocked_polls: int)[source]#

Fail jobs which can never run and return the updated deadlocked poll count.

A queue with only pending jobs cannot make progress. To avoid racing with another worker that is between two queue updates, the state must persist for a few consecutive polls before the pending jobs are marked as ERROR.

Parameters:

queue (SqliteQueue) – the queue of the run.
report (JobsReport) – the latest report of the queue.
deadlocked_polls (int) – how many consecutive polls so far had only pending jobs.

Returns:

the new count of consecutive deadlocked polls (0 once the stuck jobs are failed or the queue makes progress).

Return type:

int

Raises:

sqlalchemy.exc.OperationalError – if the queue can’t be used.

_fill_queue(config: InitialConfig, queue: SqliteQueue)[source]#

Fill the queue with the initial jobs to run.

Each job is given the configuration’s variables.

Parameters:

config (InitialConfig) – the validated configuration.
queue (SqliteQueue) – the queue to fill.

Return type:

None

Returns:

None.

Raises:

sqlalchemy.exc.OperationalError – if the queue can’t be used.

_log_failures(queue: SqliteQueue)[source]#

Log a table with each failed job and the first line of its error.

Parameters:: queue (SqliteQueue) – the queue of the finished run.
Return type:: None
Returns:: None.
Raises:: sqlalchemy.exc.OperationalError – if the queue can’t be read.

_log_progress(queue: SqliteQueue)[source]#

Log a one-line progress report of the queue.

Parameters:: queue (SqliteQueue) – the queue of the run.
Return type:: None
Returns:: None.
Raises:: sqlalchemy.exc.OperationalError – if the queue can’t be read.

_make_submit_fn(queue: SqliteQueue)[source]#

Create the function which jobs use to add jobs to the local queue.

It needs to know which parent submitted a job.

Parameters:: queue (SqliteQueue) – the queue to add the jobs to.
Returns:: the submit function, taking a job and the id of its parent.
Return type:: Callable[[JobWithContext, str | UUID | None], None]
Raises:: None. –

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue
from pyantz.infrastructure.runner.local.local_runner import _make_submit_fn

queue = SqliteQueue(Path("queue.db"))
submit = _make_submit_fn(queue)
job = JobWithContext(
    job_id="a", function="pyantz.jobs.testing.nop.do_nothing", parameters={}
)
submit(job, None)
try:
    submit(job, None)
except RuntimeError as exc:
    print(exc)

Could not add job a to the queue

_record_result(queue: SqliteQueue, job_config: JobWithContext, result: JobResult)[source]#

Store the outcome of a job, or put it back in the queue to be retried.

A failed job is retried while it has been run fewer than max_retries extra times.

Parameters:

queue (SqliteQueue) – the queue of the run.
job_config (JobWithContext) – the job which ran.
result (JobResult) – the job’s outcome.

Return type:

None

Returns:

None.

Raises:

sqlalchemy.exc.OperationalError – if the queue can’t be used.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue
from pyantz.infrastructure.runner.job_manager import JobResult
from pyantz.infrastructure.runner.local.local_runner import _record_result

queue = SqliteQueue(Path("queue.db"))
job = JobWithContext(
    job_id="flaky",
    function="pyantz.jobs.testing.nop.do_nothing",
    parameters={},
    max_retries=1,
)
queue.add_job(job, None)
_record_result(queue, job, JobResult(success=False, details="oops"))
print(queue.get_job_status("flaky").status.name)  # to be retried

READY

_remove_stale_queue(queue_file: Path)[source]#

Delete the queue (and sqlite side files) left behind by a previous run.

Every run starts from an empty queue, otherwise old jobs and statuses would mix with the new run and re-adding jobs with the same ids would fail.

Parameters:: queue_file (Path) – the queue’s SQLite file.
Return type:: None
Returns:: None.
Raises:: OSError – if a file exists but can’t be deleted.

Example

import os
from pathlib import Path

from pyantz.infrastructure.runner.local.local_runner import (
    _remove_stale_queue,
)

for name in ("queue.db", "queue.db-wal", "keep.txt"):
    Path(name).touch()
_remove_stale_queue(Path("queue.db"))
print(os.listdir())

['keep.txt']

_run_in_child(job_json: str, queue_file: str, ready: Event)[source]#

Run a job in this (child) process; exit with a failing status if it failed.

Parameters:

job_json (str) – the job, serialized as JSON.
queue_file (str) – path of the queue’s SQLite file, for the jobs it submits.
ready (Event) – event set once this process has started.

Returns:

never returns.

Return type:

None

Raises:

SystemExit – always: with code 0 if the job succeeded, otherwise 1.

_run_with_timeout(job_config: JobWithContext, queue_file: Path, timeout: float)[source]#

Run the job in a separate process, which is stopped if it takes too long.

The timeout starts once the process has started.

Parameters:

job_config (JobWithContext) – the job to run.
queue_file (Path) – the queue’s SQLite file, for the jobs it submits.
timeout (float) – seconds the job may run.

Returns:

whether the job succeeded; a job which timed out failed.

Return type:

JobResult

Raises:

OSError – if the process can’t be started.

_run_within_same_proc(runner_config: LocalRunnerConfig, queue_file: Path)[source]#

Run the local runner pool in this process.

Mostly used for testing.

Parameters:

runner_config (LocalRunnerConfig) – the local runner’s configuration.
queue_file (Path) – the queue’s SQLite file.

Return type:

None

Returns:

None.

Raises:

KeyboardInterrupt – if interrupted (Ctrl+C or SIGTERM); the running job is marked as failed.

_sigterm_as_keyboard_interrupt()[source]#

Treat SIGTERM like Ctrl+C while inside the context (main thread only).

Outside of the main thread, signal handlers can’t be set, so the context does nothing.

Parameters:: None.
Yields:: None – control, with the SIGTERM handler set.
Returns:: the context manager’s generator.
Return type:: Generator[None]
Raises:: None. –

Example

import os
import signal

from pyantz.infrastructure.runner.local.local_runner import (
    _sigterm_as_keyboard_interrupt,
)

try:
    with _sigterm_as_keyboard_interrupt():
        os.kill(os.getpid(), signal.SIGTERM)
except KeyboardInterrupt:
    print("stopped")

stopped

_start_pool(runner_config: LocalRunnerConfig, queue_file: Path)[source]#

Start the local runner pool.

Starts number_processes worker processes and waits for them to finish.

Parameters:

runner_config (LocalRunnerConfig) – the local runner’s configuration.
queue_file (Path) – the queue’s SQLite file.

Return type:

None

Returns:

None.

Raises:

KeyboardInterrupt – if interrupted (Ctrl+C or SIGTERM), once the workers are stopped.

_stop_workers(procs: Sequence[BaseProcess])[source]#

Ask the workers to stop (they mark their job as failed), then force them.

Workers still alive WORKER_STOP_TIMEOUT seconds after being asked are killed.

Parameters:: procs (Sequence[BaseProcess]) – the worker processes.
Return type:: None
Returns:: None.
Raises:: None. –

_wait_for_workers(procs: Sequence[BaseProcess], queue_file: Path, progress_interval: float | None = None)[source]#

Wait for all the workers, failing the jobs of those which die abnormally.

A worker killed while running a job (out of memory, a crash in a C extension) can never report back, so its job would stay RUNNING and the other workers would wait for it forever.

Parameters:

procs (Sequence[BaseProcess]) – the worker processes.
queue_file (Path) – the queue’s SQLite file.
progress_interval (float | None) – if set, log a progress report this often (seconds).

Return type:

None

Returns:

None.

Raises:

sqlalchemy.exc.OperationalError – if the queue can’t be used.

_worker(queue_file: Path, poll_time: float, timeout: float | None)[source]#

Run a job.

Takes jobs from the queue and runs them until the queue is complete, failing jobs which can never run on the way.

Parameters:

queue_file (Path) – the queue’s SQLite file.
poll_time (float) – seconds to wait between checks of the queue.
timeout (float | None) – if set, stop after this many seconds without a job.

Return type:

None

Returns:

None.

Raises:

KeyboardInterrupt – if interrupted; the running job is marked as failed.
WorkerTerminatedError – if the worker process is terminated; the running job is marked as failed.

_worker_process(queue_file: Path, poll_time: float, timeout: float | None)[source]#

Entry point of a worker process in the pool.

SIGTERM stops the worker (its running job is marked as failed).

Parameters:

queue_file (Path) – the queue’s SQLite file.
poll_time (float) – seconds to wait between checks of the queue.
timeout (float | None) – if set, stop after this many seconds without a job.

Return type:

None

Returns:

None.

Raises:

None – interruptions end the worker quietly.

start(config: InitialConfig)[source]#

Start the local runner processes and return the final report of the jobs.

Blocks until every job has finished (or the runner times out). A queue left in the working directory by a previous run is removed first.

Parameters:

config (InitialConfig) – the validated configuration, with a local_proc submitter.

Returns:

the final number of jobs in each status.

Return type:

JobsReport

Raises:

TypeError – if the submitter of the configuration is not a local runner
KeyboardInterrupt – if interrupted (Ctrl+C or SIGTERM), once the workers are stopped.

Example

from pyantz.infrastructure.config import InitialConfig
from pyantz.infrastructure.runner.local import start

config = InitialConfig.model_validate(
    {
        "submitter": {"type_": "local_proc", "working_directory": "."},
        "jobs": [
            {
                "function": "pyantz.jobs.files.simple.mkdir",
                "parameters": {"dir_path": "out/%{name}"},
            },
        ],
        "variables": {"name": "first"},
    }
)
print(start(config).success)

Module contents#

Run jobs in a multiprocessing pool on the local machine.

pyantz.infrastructure.runner.local package

Contents

pyantz.infrastructure.runner.local package#

Submodules#

pyantz.infrastructure.runner.local.local_runner module#

Module contents#