pynenc.orchestrator.redis_orchestrator

Module Contents

Classes

RedisCycleControl	A Redis-based implementation of cycle control using a directed acyclic graph (DAG).
RedisBlockingControl	A Redis-based implementation of blocking control for task invocations.
TaskRedisCache	A Redis-based cache for managing task invocation statuses and retries.
RedisOrchestrator	An orchestrator implementation using Redis for managing distributed task invocations.

API

class pynenc.orchestrator.redis_orchestrator.RedisCycleControl(app: pynenc.app.Pynenc, client: redis.Redis)

Bases: pynenc.orchestrator.base_orchestrator.BaseCycleControl

A Redis-based implementation of cycle control using a directed acyclic graph (DAG).

This class manages the dependencies between task invocations in Redis to prevent cycles in task calling patterns, which could lead to deadlocks or infinite loops.

Parameters:

• app (Pynenc) – The Pynenc application instance.

• client (redis.Redis) – The Redis client instance.

Initialization

purge() → None

Purges all data related to cycle control from Redis. This includes all stored invocations, calls, and their relationships.

add_call_and_check_cycles(caller: pynenc.invocation.dist_invocation.DistributedInvocation, callee: pynenc.invocation.dist_invocation.DistributedInvocation) → None

Adds a new call dependency between caller and callee invocations and checks for potential cycles.

Parameters:

• caller (DistributedInvocation) – The invocation that is making the call.

• callee (DistributedInvocation) – The invocation that is being called.

Raises:

CycleDetectedError – If adding the call creates a cycle.

remove_edges(call_id: str) → None

Recursively removes all edges from a given call in the graph.

Parameters:

call_id – The ID of the call from which to remove edges.

clean_up_invocation_cycles(invocation: pynenc.invocation.dist_invocation.DistributedInvocation) → None

Cleans up the graph by removing a given invocation and its associated edges.

Parameters:

invocation (DistributedInvocation) – The DistributedInvocation instance to be removed.

find_cycle_caused_by_new_invocation(caller: pynenc.invocation.dist_invocation.DistributedInvocation, callee: pynenc.invocation.dist_invocation.DistributedInvocation) → list[pynenc.call.Call]

Checks if adding a new call from caller to callee would create a cycle.

Parameters:

• caller (DistributedInvocation) – The invocation making the call.

• callee (DistributedInvocation) – The invocation being called.

Returns:

List of Call objects forming the cycle, if a cycle is detected; otherwise, an empty list.

_is_cyclic_util(current_call_id: str, visited: set[str], path: list[str]) → list[pynenc.call.Call]

A utility function for cycle detection.

Parameters:

• current_call_id (str) – The current call ID being examined.

• visited (set[str]) – A set of visited call IDs for cycle detection.

• path (list[str]) – A list representing the current path of call IDs.

Returns:

List of Call objects forming a cycle, if a cycle is detected; otherwise, an empty list.

class pynenc.orchestrator.redis_orchestrator.RedisBlockingControl(app: pynenc.app.Pynenc, client: redis.Redis)

Bases: pynenc.orchestrator.base_orchestrator.BaseBlockingControl

A Redis-based implementation of blocking control for task invocations.

Manages invocation dependencies and blocking states in a Redis-backed environment, ensuring that invocations waiting for others are properly tracked and released.

Parameters:

• app (Pynenc) – The Pynenc application instance.

• client (redis.Redis) – The Redis client instance.

Initialization

purge() → None

Purges all data related to blocking control from Redis. This includes all stored invocations and their waiting relationships.

waiting_for_results(waiter: pynenc.invocation.dist_invocation.DistributedInvocation, waiteds: list[pynenc.invocation.dist_invocation.DistributedInvocation]) → None

Registers that an invocation (waiter) is waiting for the results of other invocations (waiteds).

Parameters:

• waiter (DistributedInvocation) – The invocation that is waiting.

• waiteds (DistributedInvocation) – A list of invocations that the waiter is waiting for.

release_waiters(invocation: pynenc.invocation.dist_invocation.DistributedInvocation) → None

Removes an invocation from the tracking system. Also removes any dependencies related to the invocation.

Parameters:

invocation (DistributedInvocation) – The DistributedInvocation instance to be removed.

get_blocking_invocations(max_num_invocations: int) → Iterator[DistributedInvocation[Params, Result]]

Returns the invocations that are blocking others but are not waiting for anything themselves.

Parameters:

max_num_invocations (int) – The maximum number of blocking invocations to retrieve.

Returns:

An iterator of blocking DistributedInvocation instances.

exception pynenc.orchestrator.redis_orchestrator.StatusNotFound

Bases: Exception

Raised when a status is not found in Redis

Initialization

Initialize self. See help(type(self)) for accurate signature.

class pynenc.orchestrator.redis_orchestrator.TaskRedisCache(app: pynenc.app.Pynenc, client: redis.Redis)

A Redis-based cache for managing task invocation statuses and retries.

Provides methods to set and get the status and retry counts of task invocations, leveraging Redis for scalable and efficient storage.

Parameters:

• app (Pynenc) – The Pynenc application instance.

• client (redis.Redis) – The Redis client instance.

Initialization

purge() → None

Purges all data related to task invocations from Redis.

set_status(invocation: DistributedInvocation[Params, Result], status: pynenc.invocation.status.InvocationStatus) → None

Set the status of an invocation in Redis.

Parameters:

• invocation (DistributedInvocation) – The invocation to be updated.

• status (InvocationStatus) – The new status of the invocation.

set_pending_status(invocation: DistributedInvocation[Params, Result]) → None

Set the status of an invocation to pending in Redis.

Danger

We have to ensure that the invocation is not already pending.
Otherwise, we could end up with a deadlock.

Parameters:

invocation (DistributedInvocation) – The invocation to be updated.

Raises:

PendingInvocationLockError – If was not possible to lock the Invocation.

set_up_invocation_auto_purge(invocation: DistributedInvocation[Params, Result]) → None

Sets up an invocation for automatic purging after a specified time.

Parameters:

invocation (DistributedInvocation) – The invocation to be updated.

auto_purge() → None

Automatically purges invocations that have been in their final state beyond a specified duration.

Note

The duration is specified in the configuration file using the `auto_final_invocation_purge_hours` parameter.

clean_pending_status(invocation: DistributedInvocation[Params, Result]) → None

Cleans up the pending status of a task invocation.

Parameters:

invocation – The task invocation instance.

_get_invocation_status(invocation_id: str) → pynenc.invocation.status.InvocationStatus

Gets the status of a specific task invocation.

Parameters:

invocation_id – The ID of the task invocation.

Returns:

The current status of the task invocation.

get_invocation_status(invocation: pynenc.invocation.dist_invocation.DistributedInvocation) → pynenc.invocation.status.InvocationStatus

Retrieves the status of a task invocation.

Parameters:

invocation – The task invocation instance.

Returns:

The current status of the task invocation.

get_invocation_retries(invocation: pynenc.invocation.dist_invocation.DistributedInvocation) → int

Gets the number of retries for a specific task invocation.

Parameters:

invocation – The task invocation instance.

Returns:

The number of retries of the task invocation.

increment_invocation_retries(invocation: pynenc.invocation.dist_invocation.DistributedInvocation) → None

Increments the retry count for a specific task invocation.

Parameters:

invocation – The task invocation instance.

get_invocations(task_id: str, key_arguments: Optional[dict[str, str]], statuses: Optional[list[pynenc.invocation.status.InvocationStatus]]) → Iterator[pynenc.invocation.dist_invocation.DistributedInvocation]

Retrieves task invocations based on task ID, key arguments, and status.

Parameters:

• task_id (str) – The ID of the task.

• key_arguments (Optional[dict[str, str]]) – Optional key arguments for filtering.

• statuses (Optional[list[“InvocationStatus”]]) – Optional status for filtering.

Returns:

An iterator of task invocations that match the criteria.

class pynenc.orchestrator.redis_orchestrator.RedisOrchestrator(app: pynenc.app.Pynenc)

Bases: pynenc.orchestrator.base_orchestrator.BaseOrchestrator

An orchestrator implementation using Redis for managing distributed task invocations.

This orchestrator leverages Redis to handle invocation status, retries, auto-purge, and cycle and blocking controls in a distributed task environment. It extends BaseOrchestrator and integrates closely with Redis to provide efficient and scalable task orchestration.

Parameters:

app (Pynenc) – The Pynenc application instance.

Initialization

conf() → pynenc.conf.config_orchestrator.ConfigOrchestratorRedis

property cycle_control: pynenc.orchestrator.redis_orchestrator.RedisCycleControl

property blocking_control: pynenc.orchestrator.redis_orchestrator.RedisBlockingControl

get_existing_invocations(task: Task[Params, Result], key_serialized_arguments: Optional[dict[str, str]] = None, statuses: Optional[list[pynenc.invocation.status.InvocationStatus]] = None) → Iterator[pynenc.invocation.dist_invocation.DistributedInvocation]

_set_invocation_status(invocation: DistributedInvocation[Params, Result], status: pynenc.invocation.status.InvocationStatus) → None

_set_invocation_pending_status(invocation: pynenc.invocation.dist_invocation.DistributedInvocation) → None

set_up_invocation_auto_purge(invocation: pynenc.invocation.dist_invocation.DistributedInvocation[pynenc.types.Params, pynenc.types.Result]) → None

auto_purge() → None

get_invocation_status(invocation: DistributedInvocation[Params, Result]) → pynenc.invocation.status.InvocationStatus

get_invocation_retries(invocation: DistributedInvocation[Params, Result]) → int

increment_invocation_retries(invocation: DistributedInvocation[Params, Result]) → None

purge() → None

Remove all invocations from the orchestrator