gyoza.worker#

gyoza worker runtime.

Provides the heartbeat loop, execution pipeline, and worker context used by the worker process to claim and execute op runs.

class gyoza.worker.ExecutionContext(oprun_id='', image='', constraints=dict(), inputs=dict(), outputs=dict(), attempt=None, worker_id=None, image_digest=None, env_vars=dict(), volumes=dict(), logs='', exit_code=None, extras=dict())[source]#

Bases: object

Mutable context shared across all pipeline steps.

Contains OpRun metadata and container execution state.

Parameters:

oprun_id (str) – Unique identifier for the operation run.
image (str) – Docker image identifier for the container.
constraints (dict[str, Any]) – Hardware constraints (ram_mb, vram_mb, cpu).
inputs (dict[str, Any]) – Input parameters for the operation.
outputs (dict[str, Any]) – Outputs of the operation. Starts as the caller’s partial template (fields pre-filled with upload URLs) and is replaced with the final outputs once the container completes.
env_vars (dict[str, str]) – Environment variables for the container.
volumes (dict[str, dict[str, str]]) – Volume mounts in Docker’s bind format, keyed by host path: {"/host": {"bind": "/container", "mode": "rw"}}.
logs (str) – Container execution logs.
exit_code (int | None) – Exit code from the container process.
extras (dict[str, Any]) – Arbitrary key-value store for step communication.
attempt (int | None) – Attempt number the server handed out; sent with every event so a late event from an abandoned attempt is refused instead of landing on the next one. None for servers that do not report it.
image_digest (str | None) – Digest of the pulled image, reported with STARTED.
worker_id (str | None)

oprun_id: str = ''#

image: str = ''#

constraints: dict[str, Any]#

inputs: dict[str, Any]#

outputs: dict[str, Any]#

attempt: int | None = None#

worker_id: str | None = None#

image_digest: str | None = None#

env_vars: dict[str, str]#

volumes: dict[str, dict[str, str]]#

logs: str = ''#

exit_code: int | None = None#

extras: dict[str, Any]#

class gyoza.worker.WorkerContext(worker_id='', resources=Resources(), tags=list(), running_ops=list())[source]#

Bases: object

Singleton context storing worker state for heartbeats.

Holds running ops (RunningOp) with last known server state, refreshed from the main loop so the executor can check e.g. cancellation.

Parameters:

worker_id (str) – Worker name/identifier.
resources (Resources) – Hardware resources available.
tags (list[str]) – Worker-level tags.
running_ops (list[RunningOp]) – Running OpRuns with last known server state.

worker_id: str = ''#

resources: Resources#

tags: list[str]#

running_ops: list[RunningOp]#

add_running_op(oprun_id, image, inputs, constraints=None, *, attempt=None, reservation=None)[source]#

Add an OpRun to the running operations list.

Parameters:

oprun_id (str) – Unique identifier for the OpRun.
image (str) – Docker image identifier.
inputs (dict[str, Any]) – Input parameters for execution.
constraints (dict[str, Any] | None) – Hardware constraints (for heartbeat payload).
attempt (int | None)
reservation (dict[str, int] | None)

Return type:

None

remove_running_op(oprun_id, attempt=None)[source]#

Remove an OpRun from the running operations list.

Parameters:

oprun_id (str) – Unique identifier for the OpRun to remove.
attempt (int | None)

Return type:

None

set_run_state(run_id, state, attempt=None)[source]#

Update the last known server state for a running op.

Parameters:

run_id (str) – OpRun identifier.
state (str) – Current state from server (e.g. RUNNING, CANCELLED).
attempt (int | None)

Return type:

None

get_run_state(run_id, attempt=None)[source]#

Return the last known server state for a run, or None.

Parameters:

run_id (str) – OpRun identifier.
attempt (int | None)

Returns:

State string or None if not found.

Return type:

str | None

is_run_cancelled(run_id, attempt=None)[source]#

Return True if the run is known to be CANCELLED on the server.

Parameters:

run_id (str) – OpRun identifier.
attempt (int | None)

Returns:

True if state is CANCELLED.

Return type:

bool

update_from_server(run_id, attempt, data)[source]#

Reconcile only the local execution that produced this request.

Return type:

None

Parameters:

run_id (str)
attempt (int | None)
data (dict[str, Any])

clear_running_ops()[source]#

Clear the running operations list.

Return type:: None

to_heartbeat_payload()[source]#

Convert to heartbeat request payload.

Returns:: Payload ready for gyoza_client.heartbeat().
Return type:: dict[str, Any]

gyoza.worker.start()[source]#

Start the worker loop.

Blocks until interrupted with Ctrl-C or a signal. Each cycle:

Sends a heartbeat to the server (registers/updates the worker).
Claims op runs, waiting on the server until some are allocated.
Dispatches each claimed run to a background thread.
Refreshes the server state of the running ops and sweeps the cache.

Return type:: None

Worker main loop.

Entry point called by gyoza worker start. One blocking cycle: report to the server, ask it for work and let it hold the request until there is some, dispatch each op run to a thread pool.

gyoza.worker.runner.start()[source]#

Start the worker loop.

Blocks until interrupted with Ctrl-C or a signal. Each cycle:

Sends a heartbeat to the server (registers/updates the worker).
Claims op runs, waiting on the server until some are allocated.
Dispatches each claimed run to a background thread.
Refreshes the server state of the running ops and sweeps the cache.

Return type:: None

Execution Pipeline#

The Execution pipeline provides a pre-configured execution pipeline that can be used to run the full execution of an OpRun.

Example

>>> from gyoza.worker.execution_pipeline import execution_pipeline, ExecutionContext
>>>
>>> ctx = ExecutionContext(
...     oprun_id="run_abc123",
...     image="geoiahub/product:latest",
...     inputs={"param": "value"},
... )
>>> result = execution_pipeline.run(ctx)
>>> print(result.exit_code)

class gyoza.worker.execution_pipeline.ExecutionContext(oprun_id='', image='', constraints=dict(), inputs=dict(), outputs=dict(), attempt=None, worker_id=None, image_digest=None, env_vars=dict(), volumes=dict(), logs='', exit_code=None, extras=dict())[source]

Bases: object

Mutable context shared across all pipeline steps.

Contains OpRun metadata and container execution state.

Parameters:

oprun_id (str) – Unique identifier for the operation run.
image (str) – Docker image identifier for the container.
constraints (dict[str, Any]) – Hardware constraints (ram_mb, vram_mb, cpu).
inputs (dict[str, Any]) – Input parameters for the operation.
outputs (dict[str, Any]) – Outputs of the operation. Starts as the caller’s partial template (fields pre-filled with upload URLs) and is replaced with the final outputs once the container completes.
env_vars (dict[str, str]) – Environment variables for the container.
volumes (dict[str, dict[str, str]]) – Volume mounts in Docker’s bind format, keyed by host path: {"/host": {"bind": "/container", "mode": "rw"}}.
logs (str) – Container execution logs.
exit_code (int | None) – Exit code from the container process.
extras (dict[str, Any]) – Arbitrary key-value store for step communication.
attempt (int | None) – Attempt number the server handed out; sent with every event so a late event from an abandoned attempt is refused instead of landing on the next one. None for servers that do not report it.
image_digest (str | None) – Digest of the pulled image, reported with STARTED.
worker_id (str | None)

oprun_id: str = ''

image: str = ''

constraints: dict[str, Any]

inputs: dict[str, Any]

outputs: dict[str, Any]

attempt: int | None = None

worker_id: str | None = None

image_digest: str | None = None

env_vars: dict[str, str]

volumes: dict[str, dict[str, str]]

logs: str = ''

exit_code: int | None = None

extras: dict[str, Any]

class gyoza.worker.execution_pipeline.Step(*args, **kwargs)[source]

Bases: Protocol

A single step in the pipeline.

Steps receive the context, process it, and return the (possibly mutated) context.

Parameters:: ctx (ExecutionContext) – Mutable execution context shared across all steps.
Returns:: The context (same or mutated) for the next step.
Return type:: ExecutionContext

class gyoza.worker.execution_pipeline.Pipeline(steps=None)[source]

Bases: object

A sequence of steps executed in order.

Parameters:: steps (list[Step] | None) – Ordered list of steps to execute.

run(ctx=None)[source]

Execute all steps in sequence.

If any step raises an exception the run is marked as FAILED via the server API and the exception is re-raised so the caller can log it.

The one exception is a lost connection to the server (httpx.TransportError), which is not reported: it says nothing about the op or its data, and a FAILED written for it ended the run on its first attempt. The attempt is left as it is instead; once the worker stops listing it, the reaper hands the run to a fresh attempt, as it does for any run its worker dropped. HTTPStatusError is a real answer from the server and still fails the run.

Parameters:: ctx (ExecutionContext | None) – Shared context. Created if not provided.
Returns:: The context after all steps have been applied.
Return type:: ExecutionContext
Raises:: Exception – Whatever exception the failing step raised.

gyoza.worker.execution_pipeline.pull_latest_image(ctx)[source]

Pull the latest version of the Docker image before execution.

Parameters:: ctx (ExecutionContext) – Execution context with image name configured.
Returns:: Context with image_digest set to what was actually pulled.
Return type:: ExecutionContext

gyoza.worker.execution_pipeline.start_oprun(ctx)[source]

Send STARTED event to the master API.

Parameters:: ctx (ExecutionContext) – Execution context with oprun_id.
Returns:: Unchanged context.
Return type:: ExecutionContext

Pipeline implementation.

class gyoza.worker.execution_pipeline.pipeline.Pipeline(steps=None)[source]#

Bases: object

A sequence of steps executed in order.

Parameters:: steps (list[Step] | None) – Ordered list of steps to execute.

run(ctx=None)[source]#

Execute all steps in sequence.

If any step raises an exception the run is marked as FAILED via the server API and the exception is re-raised so the caller can log it.

The one exception is a lost connection to the server (httpx.TransportError), which is not reported: it says nothing about the op or its data, and a FAILED written for it ended the run on its first attempt. The attempt is left as it is instead; once the worker stops listing it, the reaper hands the run to a fresh attempt, as it does for any run its worker dropped. HTTPStatusError is a real answer from the server and still fails the run.

Parameters:: ctx (ExecutionContext | None) – Shared context. Created if not provided.
Returns:: The context after all steps have been applied.
Return type:: ExecutionContext
Raises:: Exception – Whatever exception the failing step raised.

Execution context for the pipeline.

class gyoza.worker.execution_pipeline.types.context.ExecutionContext(oprun_id='', image='', constraints=dict(), inputs=dict(), outputs=dict(), attempt=None, worker_id=None, image_digest=None, env_vars=dict(), volumes=dict(), logs='', exit_code=None, extras=dict())[source]#

Bases: object

Mutable context shared across all pipeline steps.

Contains OpRun metadata and container execution state.

Parameters:

oprun_id (str) – Unique identifier for the operation run.
image (str) – Docker image identifier for the container.
constraints (dict[str, Any]) – Hardware constraints (ram_mb, vram_mb, cpu).
inputs (dict[str, Any]) – Input parameters for the operation.
outputs (dict[str, Any]) – Outputs of the operation. Starts as the caller’s partial template (fields pre-filled with upload URLs) and is replaced with the final outputs once the container completes.
env_vars (dict[str, str]) – Environment variables for the container.
volumes (dict[str, dict[str, str]]) – Volume mounts in Docker’s bind format, keyed by host path: {"/host": {"bind": "/container", "mode": "rw"}}.
logs (str) – Container execution logs.
exit_code (int | None) – Exit code from the container process.
extras (dict[str, Any]) – Arbitrary key-value store for step communication.
attempt (int | None) – Attempt number the server handed out; sent with every event so a late event from an abandoned attempt is refused instead of landing on the next one. None for servers that do not report it.
image_digest (str | None) – Digest of the pulled image, reported with STARTED.
worker_id (str | None)

oprun_id: str = ''#

image: str = ''#

constraints: dict[str, Any]#

inputs: dict[str, Any]#

outputs: dict[str, Any]#

attempt: int | None = None#

worker_id: str | None = None#

image_digest: str | None = None#

env_vars: dict[str, str]#

volumes: dict[str, dict[str, str]]#

logs: str = ''#

exit_code: int | None = None#

extras: dict[str, Any]#

Step protocol for the execution pipeline.

class gyoza.worker.execution_pipeline.types.step.Step(*args, **kwargs)[source]#

Bases: Protocol

A single step in the pipeline.

Steps receive the context, process it, and return the (possibly mutated) context.

Parameters:: ctx (ExecutionContext) – Mutable execution context shared across all steps.
Returns:: The context (same or mutated) for the next step.
Return type:: ExecutionContext

Steps#

class gyoza.worker.execution_pipeline.steps.prepare_inputs.PrepareInputs[source]#

Bases: object

Prepare the inputs for the container execution.

Creates a unique gyoza directory for the run, saves inputs as a JSON file, and sets GYOZA_INPUT_PATH and GYOZA_OUTPUT_PATH environment variables.

When the run carries an outputs template (fields pre-filled with upload URLs by the caller), it is pre-written to the output file so the op runtime inside the container can deliver those outputs.

Pull latest image step - ensures the latest version of the image is available.

gyoza.worker.execution_pipeline.steps.pull_latest_image.pull_latest_image(ctx)[source]#

Pull the latest version of the Docker image before execution.

Parameters:: ctx (ExecutionContext) – Execution context with image name configured.
Returns:: Context with image_digest set to what was actually pulled.
Return type:: ExecutionContext

Run container step - executes container and streams logs.

class gyoza.worker.execution_pipeline.steps.run_container.RunContainer[source]#

Bases: object

Run the container and stream logs.

Runs the container attached, streaming logs in real-time. Logs compatible with the gyoza events are parsed and sent to master.

Start OpRun step - sends STARTED event to master.

gyoza.worker.execution_pipeline.steps.start_oprun.start_oprun(ctx)[source]#

Send STARTED event to the master API.

Parameters:: ctx (ExecutionContext) – Execution context with oprun_id.
Returns:: Unchanged context.
Return type:: ExecutionContext

gyoza.worker.execution_pipeline.steps.complete_oprun.complete_oprun(ctx)[source]#

Send COMPLETED or FAILED event to the master API based on exit code.

Loads outputs from the pickle file and sends them as payload on success. If the server returns 400 (run already terminal) or 409 (the run moved on to another attempt), the update is skipped and context is returned without raising, so the worker does not treat a reclaimed run as a local failure.

Parameters:: ctx (ExecutionContext) – Execution context with exit_code and env_vars set.
Returns:: Context with outputs populated on success.
Return type:: ExecutionContext

gyoza.worker.execution_pipeline.steps.cleanup_op_dir.cleanup_op_dir(ctx)[source]#

Delete the op directory created by prepare_inputs.

Removes the gyoza_dir generated by the pipeline, which contains inputs, outputs, and any other files produced during execution.

Parameters:: ctx (ExecutionContext) – Execution context with gyoza_dir in extras.
Returns:: Context unchanged (directory removed from disk).
Return type:: ExecutionContext

Worker Context#

Worker context.

Provides a easy to use worker context that can be used to: track worker identity, resources, and running operations.

Example

>>> from gyoza.worker.worker_context import worker_context
>>>
>>> # Resources and worker_id are auto-detected at import time
>>> print(worker_context.worker_id)  # from GYOZA_WORKER_ID env var
>>> print(worker_context.resources.cpu_cores)
>>> print(worker_context.resources.ram_mb)
>>>
>>> # Use for heartbeat
>>> from gyoza.client import gyoza_client
>>> payload = worker_context.to_heartbeat_payload()
>>> gyoza_client.heartbeat(**payload)

class gyoza.worker.worker_context.GPU(id, vram_mb, tags=list())[source]

Bases: object

Representation of a single GPU.

Parameters:

id (int) – Numeric GPU identifier (0, 1, 2…).
vram_mb (int) – VRAM in megabytes.
tags (list[str]) – GPU tags (e.g., “cuda”, “nvidia-a100”).

id: int

vram_mb: int

tags: list[str]

to_dict()[source]

Convert to dictionary representation.

Return type:: dict[str, Any]

class gyoza.worker.worker_context.Resources(cpu_cores=0, ram_mb=0, gpus=list())[source]

Bases: object

Hardware resources available on the worker.

Parameters:

cpu_cores (int) – Number of CPU cores.
ram_mb (int) – RAM in megabytes.
gpus (list[GPU]) – List of available GPUs.

cpu_cores: int = 0

ram_mb: int = 0

gpus: list[GPU]

to_dict()[source]

Convert to dictionary representation.

Return type:: dict[str, Any]

class gyoza.worker.worker_context.RunningOp(id, image, inputs=dict(), constraints=dict(), state=None, attempt=None, reservation=None)[source]

Bases: object

Simplified run representation with last known server state.

Stored in worker context for each op the worker is executing. Server state is refreshed from the main loop via GET /runs/{id}.

Parameters:

id (str) – OpRun identifier.
image (str) – Docker image.
inputs (dict[str, Any]) – Input parameters.
constraints (dict[str, Any]) – Hardware constraints (for heartbeat payload).
state (str | None) – Last known server state (PENDING, ASSIGNED, RUNNING, CANCELLED, COMPLETED, FAILED). None until refreshed from server.
attempt (int | None)
reservation (dict[str, int] | None)

id: str

image: str

inputs: dict[str, Any]

constraints: dict[str, Any]

state: str | None = None

attempt: int | None = None

reservation: dict[str, int] | None = None

to_heartbeat_dict()[source]

Dict for heartbeat running_ops (id, image, inputs, constraints).

Return type:: dict[str, Any]

class gyoza.worker.worker_context.WorkerContext(worker_id='', resources=Resources(), tags=list(), running_ops=list())[source]

Bases: object

Singleton context storing worker state for heartbeats.

Holds running ops (RunningOp) with last known server state, refreshed from the main loop so the executor can check e.g. cancellation.

Parameters:

worker_id (str) – Worker name/identifier.
resources (Resources) – Hardware resources available.
tags (list[str]) – Worker-level tags.
running_ops (list[RunningOp]) – Running OpRuns with last known server state.

worker_id: str = ''

resources: Resources

tags: list[str]

running_ops: list[RunningOp]

add_running_op(oprun_id, image, inputs, constraints=None, *, attempt=None, reservation=None)[source]

Add an OpRun to the running operations list.

Parameters:

oprun_id (str) – Unique identifier for the OpRun.
image (str) – Docker image identifier.
inputs (dict[str, Any]) – Input parameters for execution.
constraints (dict[str, Any] | None) – Hardware constraints (for heartbeat payload).
attempt (int | None)
reservation (dict[str, int] | None)

Return type:

None

remove_running_op(oprun_id, attempt=None)[source]

Remove an OpRun from the running operations list.

Parameters:

oprun_id (str) – Unique identifier for the OpRun to remove.
attempt (int | None)

Return type:

None

set_run_state(run_id, state, attempt=None)[source]

Update the last known server state for a running op.

Parameters:

run_id (str) – OpRun identifier.
state (str) – Current state from server (e.g. RUNNING, CANCELLED).
attempt (int | None)

Return type:

None

get_run_state(run_id, attempt=None)[source]

Return the last known server state for a run, or None.

Parameters:

run_id (str) – OpRun identifier.
attempt (int | None)

Returns:

State string or None if not found.

Return type:

str | None

is_run_cancelled(run_id, attempt=None)[source]

Return True if the run is known to be CANCELLED on the server.

Parameters:

run_id (str) – OpRun identifier.
attempt (int | None)

Returns:

True if state is CANCELLED.

Return type:

bool

update_from_server(run_id, attempt, data)[source]

Reconcile only the local execution that produced this request.

Return type:

None

Parameters:

run_id (str)
attempt (int | None)
data (dict[str, Any])

clear_running_ops()[source]

Clear the running operations list.

Return type:: None

to_heartbeat_payload()[source]

Convert to heartbeat request payload.

Returns:: Payload ready for gyoza_client.heartbeat().
Return type:: dict[str, Any]

gyoza.worker.worker_context.detect_resources()[source]

Auto-detect hardware resources available on this machine.

Returns:: Detected CPU cores, RAM, and GPUs.
Return type:: Resources

Types for the worker context module.

class gyoza.worker.worker_context.types.RunningOp(id, image, inputs=dict(), constraints=dict(), state=None, attempt=None, reservation=None)[source]#

Bases: object

Simplified run representation with last known server state.

Stored in worker context for each op the worker is executing. Server state is refreshed from the main loop via GET /runs/{id}.

Parameters:

id (str) – OpRun identifier.
image (str) – Docker image.
inputs (dict[str, Any]) – Input parameters.
constraints (dict[str, Any]) – Hardware constraints (for heartbeat payload).
state (str | None) – Last known server state (PENDING, ASSIGNED, RUNNING, CANCELLED, COMPLETED, FAILED). None until refreshed from server.
attempt (int | None)
reservation (dict[str, int] | None)

id: str#

image: str#

inputs: dict[str, Any]#

constraints: dict[str, Any]#

state: str | None = None#

attempt: int | None = None#

reservation: dict[str, int] | None = None#

to_heartbeat_dict()[source]#

Dict for heartbeat running_ops (id, image, inputs, constraints).

Return type:: dict[str, Any]

class gyoza.worker.worker_context.types.GPU(id, vram_mb, tags=list())[source]#

Bases: object

Representation of a single GPU.

Parameters:

id (int) – Numeric GPU identifier (0, 1, 2…).
vram_mb (int) – VRAM in megabytes.
tags (list[str]) – GPU tags (e.g., “cuda”, “nvidia-a100”).

id: int#

vram_mb: int#

tags: list[str]#

to_dict()[source]#

Convert to dictionary representation.

Return type:: dict[str, Any]

class gyoza.worker.worker_context.types.Resources(cpu_cores=0, ram_mb=0, gpus=list())[source]#

Bases: object

Hardware resources available on the worker.

Parameters:

cpu_cores (int) – Number of CPU cores.
ram_mb (int) – RAM in megabytes.
gpus (list[GPU]) – List of available GPUs.

cpu_cores: int = 0#

ram_mb: int = 0#

gpus: list[GPU]#

to_dict()[source]#

Convert to dictionary representation.

Return type:: dict[str, Any]

class gyoza.worker.worker_context.types.WorkerContext(worker_id='', resources=Resources(), tags=list(), running_ops=list())[source]#

Bases: object

Singleton context storing worker state for heartbeats.

Holds running ops (RunningOp) with last known server state, refreshed from the main loop so the executor can check e.g. cancellation.

Parameters:

worker_id (str) – Worker name/identifier.
resources (Resources) – Hardware resources available.
tags (list[str]) – Worker-level tags.
running_ops (list[RunningOp]) – Running OpRuns with last known server state.

worker_id: str = ''#

resources: Resources#

tags: list[str]#

running_ops: list[RunningOp]#

add_running_op(oprun_id, image, inputs, constraints=None, *, attempt=None, reservation=None)[source]#

Add an OpRun to the running operations list.

Parameters:

oprun_id (str) – Unique identifier for the OpRun.
image (str) – Docker image identifier.
inputs (dict[str, Any]) – Input parameters for execution.
constraints (dict[str, Any] | None) – Hardware constraints (for heartbeat payload).
attempt (int | None)
reservation (dict[str, int] | None)

Return type:

None

remove_running_op(oprun_id, attempt=None)[source]#

Remove an OpRun from the running operations list.

Parameters:

oprun_id (str) – Unique identifier for the OpRun to remove.
attempt (int | None)

Return type:

None

set_run_state(run_id, state, attempt=None)[source]#

Update the last known server state for a running op.

Parameters:

run_id (str) – OpRun identifier.
state (str) – Current state from server (e.g. RUNNING, CANCELLED).
attempt (int | None)

Return type:

None

get_run_state(run_id, attempt=None)[source]#

Return the last known server state for a run, or None.

Parameters:

run_id (str) – OpRun identifier.
attempt (int | None)

Returns:

State string or None if not found.

Return type:

str | None

is_run_cancelled(run_id, attempt=None)[source]#

Return True if the run is known to be CANCELLED on the server.

Parameters:

run_id (str) – OpRun identifier.
attempt (int | None)

Returns:

True if state is CANCELLED.

Return type:

bool

update_from_server(run_id, attempt, data)[source]#

Reconcile only the local execution that produced this request.

Return type:

None

Parameters:

run_id (str)
attempt (int | None)
data (dict[str, Any])

clear_running_ops()[source]#

Clear the running operations list.

Return type:: None

to_heartbeat_payload()[source]#

Convert to heartbeat request payload.

Returns:: Payload ready for gyoza_client.heartbeat().
Return type:: dict[str, Any]

Utilities for auto-detecting worker resources.

gyoza.worker.worker_context.utils.detect_resources()[source]#

Auto-detect hardware resources available on this machine.

Returns:: Detected CPU cores, RAM, and GPUs.
Return type:: Resources