from abc import ABC, ABCMeta
from dataclasses import dataclass
from datetime import datetime
from http import HTTPMethod
from importlib import invalidate_caches, import_module, reload
from inspect import signature, Signature, ismethod, isfunction
from logging import getLogger, Logger
from multiprocessing import Process, Pipe, Queue
from multiprocessing.connection import Connection, wait
from pathlib import Path
from os import getpid
from time import sleep, monotonic
from threading import active_count
from traceback import format_exc
from types import UnionType
from typing import Optional, Type, Tuple, Callable, Union, TypeVar, get_args, Iterable, get_origin, Any, cast
from uuid import UUID
from edri.abstract.manager.worker import Worker, WorkerFactoryProtocol, WorkerState
from edri.abstract.manager.worker_controller import WorkerController
from edri.api.dataclass.api_event import api_events
from edri.config.setting import API_CACHE_CONTROL, API_CACHE_HEADERS, HEALTH_CHECK_LOG
from edri.dataclass.directive.http import NotModifiedResponseDirective, HeaderResponseDirective
from edri.dataclass.event import Event
from edri.dataclass.health_checker import Status, Vitals, HealthReport
from edri.dataclass.response import Response, ResponseStatus
from edri.events.edri.group import Manager
from edri.events.edri.manager import WorkerQuit, WorkerStarted, Restart
from edri.events.edri.router import Subscribe, HealthCheck, UnsubscribeAll, Unsubscribe
from edri.events.edri.store import Get
from edri.utility.function import (install_graceful_sigterm,
current_rss_bytes, peak_rss_bytes, cpu_seconds_total, open_fd_count)
from edri.utility.cache import Cache
from warnings import warn
__all__ = ["ManagerBase"]
T = TypeVar("T", bound=Event)
class ManagerBaseMeta(ABCMeta):
def __new__(mcls: type, name: str, bases: tuple[type, ...], namespace: dict[str, Any], /, **kwargs: Any) -> Any:
namespace["_cache_keys"] = dict()
namespace["_cache_methods"] = dict()
for attr_name, attr_value in list(namespace.items()):
if not callable(attr_value):
continue
if purpose := getattr(attr_value, "__purpose__", None): # Check for the decorator's marker
if purpose == "request":
new_name = f"solve_req_{attr_name}"
if cache := getattr(attr_value, "__cache__", None):
namespace["_cache_keys"][attr_name] = cache
event = signature(attr_value).parameters["event"].annotation
for api_event in api_events:
if api_event.event == event:
break
namespace["_cache_methods"][event] = api_event.method
elif purpose == "response":
new_name = f"solve_res_{attr_name}"
else:
raise ValueError(f"Invalid purpose: {purpose}")
if new_name in namespace:
raise KeyError(f"This method is in conflict: {attr_name}")
namespace[new_name] = namespace.pop(attr_name)
# At runtime mcls is the class being created, whose metaclass is
# ManagerBaseMeta, so it is always an instance of ManagerBaseMeta;
# mypy cannot verify that from the bare `type` annotation.
return super().__new__(mcls, name, bases, namespace, **kwargs) # type: ignore[misc]
@dataclass(frozen=True, slots=True)
class ResolverCacheState:
key: Any | None = None
etag: str | None = None
method: HTTPMethod | None = None
@property
def enabled(self) -> bool:
return bool(self.key)
[docs]
class ManagerBase(ABC, Process, metaclass=ManagerBaseMeta):
"""
Base class for a manager that handles events and manages worker processes or threads.
The manager subscribes to request and response event types, resolves incoming
events, and may delegate work to worker threads or processes. A configurable
worker limit can be applied through `max_workers`. When the limit is reached,
additional worker start requests are queued and launched later as running
workers finish.
Attributes:
router_queue (Optional["Queue[Event]"]):
Queue for communicating with the router.
logger (Optional[Logger]):
Logger instance for logging events.
_from_time (Optional[datetime]):
Start time for filtering events.
max_workers (int):
Maximum number of workers allowed to run concurrently. `-1` means
unlimited and is the default.
Methods:
__init__: Constructor for initializing the manager.
_find_worker: Finds a worker based on its communication pipe.
_prepare: Prepares the manager before starting the event resolver loop.
_prepare_resolvers: Prepares methods for resolving request and response events.
_remove_worker: Removes a worker from the manager's tracking.
_resolve_undelivered_events: Resolves any undelivered events.
_subscribe: Subscribes to events based on requests and responses.
_subscribe_static_resolvers: Subscribes to static event resolvers.
additional_pipes: Property for specifying additional pipes.
name: Property for specifying the manager's name.
workers: Property for defining workers and their associated events.
after_start: Hook for actions after the manager starts.
get_pipes: Gets a set of all pipes including the router and worker pipes.
health_check: Handles health check events from the router.
resolve: Resolves an event through independently overridable stages.
prepare_outgoing_event: Hook run immediately before router queue output.
send_to_router: Common output path for every event sent to the router.
resolve_callback_pipe: Handles events received from unknown pipes.
resolve_callback_worker: Default method for resolving events received from workers.
resolve_unknown: Handles unknown events received from the router pipe.
run: Main entry point for the manager process.
run_resolver: Main loop for resolving events from all pipes.
run_resolver_cycle: Executes one independently testable resolver iteration.
receive_pipe_event: Applies the shared receive and pipe-error policy.
start: Starts the manager process with an optional router queue.
start_worker: Starts a new worker for handling an event.
store_get: Handles store get requests.
"""
# Populated by ManagerBaseMeta.__new__ on the class namespace; declared
# here so the instance-level lookups in the resolver cache state type-check.
_cache_keys: dict[str, str]
_cache_methods: dict[Type[Event], HTTPMethod | None]
# ``router_queue`` is annotated Any rather than Queue[Event] because the
# multiprocessing.Queue factory function is not subscriptable at runtime and
# Python 3.14's lazy annotation evaluation (PEP 649) evaluates __init__
# signature annotations inside inspect.signature().
def __init__(self, router_queue: Optional[Any] = None, logger: Optional[Logger] = None,
from_time: Optional[datetime] = None, max_workers: int = -1) -> None:
"""
Initializes the manager with an optional router queue, logger, and start time for event filtering.
Parameters:
router_queue (Optional["Queue[Event]"]): Queue for communication with the router. Default is None.
logger (Optional[Logger]): Logger instance for logging. If None, a default logger is created.
from_time (Optional[datetime]): Start time for filtering events. Default is None.
"""
super().__init__()
self.router_queue: Queue[Event]
if router_queue:
self.router_queue = router_queue
self.router_pipe: Connection
self._unresolved: list[Event] = []
if logger is None:
self.logger = getLogger(self.name)
else:
self.logger = logger
self._requests: dict[Type[Event], Callable[..., Any]] = {}
self._responses: dict[Type[Event], Callable[..., Any]] = {}
self._max_workers = -1
self.workers = WorkerController(self)
# event._id -> (event type, Worker) for events a worker sent out and is
# awaiting a reply for; _pending_response_counts refcounts each type
# across all workers so the shared _responses/_auto_responses entry for
# it (see _track_worker_response/_release_pending_response_type) is only
# torn down once nobody is waiting on it anymore. Cleaned up when the
# worker exits (central WorkerController finalization) — a reply that's permanently lost (not
# just slow) for a long-lived worker stays orphaned here until that
# worker eventually exits, since nothing but worker exit sweeps this.
# Acceptable: it takes a genuine double fault (persistent worker +
# a reply that never comes back, not merely a slow one — the 30s
# dispatch-side timeout in pydantic_ai.BaseAgentWorker doesn't touch
# this state at all), and the leak is one entry per such fault, not
# unbounded growth from normal operation.
self._pending_responses: dict[UUID, tuple[Type[Event], Worker]] = {}
self._pending_response_counts: dict[Type[Event], int] = {}
self._auto_responses: set[Type[Event]] = set()
self._from_time = from_time
self._store_get: Optional[Callable[..., Any]] = None
self._exceptions: list[tuple[str, dict[str, Any], str, str, datetime]] = []
# Counters for health-check vitals. ``_processed_total``/``_failed_total``
# and the cumulative CPU clock are monotonic; ``health_check`` reports the
# delta since the previous check and advances each watermark, so every
# check describes only its own interval. ``_started_at`` is re-stamped in
# ``run()`` once the process is actually live (see uptime vital).
self._processed_total = 0
self._processed_reported = 0
self._failed_total = 0
self._failed_reported = 0
self._cpu_reported = 0.0
self._started_at = monotonic()
self.cache: Cache
self._cache_vary = "Accept,Accept-Encoding"
if API_CACHE_HEADERS:
self._cache_vary += f",{API_CACHE_HEADERS}"
self._disabled_additional_pipes: set[Connection] = set()
self._resolver_running = False
self.max_workers = max_workers
@property
def max_workers(self) -> int:
return self._max_workers
@max_workers.setter
def max_workers(self, value: int) -> None:
if value == 0 or value < -1:
raise ValueError("max_workers must be -1 or a positive integer")
self._max_workers = value
self.workers._start_pending()
@max_workers.deleter
def max_workers(self) -> None:
self._max_workers = -1
self.workers._start_pending()
@property
def _cache(self) -> Cache:
"""Deprecated alias for :attr:`cache`. Removed in a future release."""
warn(
"ManagerBase._cache is deprecated; use ManagerBase.cache instead.",
DeprecationWarning,
stacklevel=2,
)
return self.cache
@_cache.setter
def _cache(self, value: Cache) -> None:
warn(
"ManagerBase._cache is deprecated; use ManagerBase.cache instead.",
DeprecationWarning,
stacklevel=2,
)
self.cache = value
def _subscribe(self) -> None:
"""
Subscribes the manager to events by sending subscription requests to the router queue.
It handles both requests and responses, registering them for the manager to resolve.
"""
count = len(self._requests) + len(self._responses)
if count == 0:
self.logger.info("No events to subscribe!")
return
self.router_pipe, pipe = Pipe(duplex=False)
self.logger.debug("Count of events to register: %s", count)
def register_events(events: dict[Type[Event], Callable[..., Any]], is_request: bool) -> None:
for event_type in events.keys():
self.logger.debug("Registering of event: %s", event_type)
self.send_to_router(
# The @event decorator rewrites the dataclass 'response' field
# to init=False with a default_factory, which mypy's dataclass
# transform cannot see; the proper fix belongs in
# edri/events/edri/router/subscribe.py (outside this scope).
Subscribe(pipe=pipe, name=self.name, event_type=event_type, request=is_request,
from_time=self._from_time) # type: ignore[call-arg]
)
while True:
event = self.router_pipe.recv()
if isinstance(event, Subscribe):
self.logger.debug("Event was registered: %s", event.event_type)
break
else:
self.logger.debug("Unresolved event received during registration: %s", event)
self._unresolved.append(event)
register_events(self._requests, is_request=True)
register_events(self._responses, is_request=False)
# Kept open (not closed) so _track_worker_response can mint further
# Subscribe events for types discovered dynamically after startup.
self._router_send_pipe = pipe
# Registered here, after the loop above, not in _subscribe_static_resolvers:
# that runs before _subscribe() and everything in _requests/_responses at
# that point gets its own Router-level Subscribe — these two are acks that
# arrive already addressed to us and must NOT be requested from the Router.
self._responses.setdefault(Subscribe, self._ignore_router_ack)
self._responses.setdefault(Unsubscribe, self._ignore_router_ack)
unresolved_count = len(self._unresolved)
if unresolved_count > 0:
self.logger.debug("Received %s undelivered events!", unresolved_count)
self.logger.debug("All events were successfully registered!")
def _unsubscribe_all(self) -> None:
"""
...
"""
count = len(self._requests) + len(self._responses)
if count == 0:
self.logger.info("No events to unsubscribe!")
return
self.logger.debug("Count of events to unregister: %s", count)
self._unresolved = list()
self.send_to_router(
# The @event decorator rewrites the dataclass 'response' field to
# init=False with a default_factory, which mypy's dataclass transform
# cannot see; the proper fix belongs in
# edri/events/edri/router/unsubscribe_all.py (outside this scope).
UnsubscribeAll(name=self.name) # type: ignore[call-arg]
)
while True:
event = self.router_pipe.recv()
if isinstance(event, UnsubscribeAll):
self.logger.debug("All event was unregistered", )
break
else:
self.logger.debug("Unresolved event received during registration: %s", event)
self._unresolved.append(event)
unresolved_count = len(self._unresolved)
if unresolved_count > 0:
self.logger.debug("Received %s undelivered events!", unresolved_count)
self.logger.debug("All events were successfully unregistered!")
def _resolve_undelivered_events(self) -> None:
"""
Processes any events that were received but not resolved immediately upon subscription.
This ensures no events are lost during the setup phase of the manager.
"""
if self._unresolved:
while True:
try:
event = self._unresolved.pop(0)
self.logger.debug("Solving undelivered event: %s", event)
except IndexError:
self.logger.debug("Unresolved events were resolved!")
break
event._timing.stamp(self.__class__.__name__, "Received as undelivered")
self.resolve(event)
event._timing.stamp(self.__class__.__name__, "Resolved")
else:
self.logger.debug("No unresolved events!")
del self._unresolved
def _prepare_resolvers(self) -> None:
"""
Prepares the resolvers by dynamically associating methods in the class with
specific event types for handling requests and responses.
This method iterates over all methods in the class that have names starting
with 'solve_req_' or 'solve_res_'. It checks the signature of each method to
ensure that its parameter is named 'event' and is a subclass of the `Event`
dataclass. Based on the method's prefix, it associates the method with either
a request or response handler in the `_requests` or `_responses` dictionaries
respectively.
Raises:
AttributeError: If the parameter of the method is not named 'event'
or if the parameter is not a subclass of the `Event` dataclass.
Methods:
- `get_event(sig: Signature) -> Type[Event]`: Extracts and validates the event
parameter from the method's signature. Ensures that the event parameter is
named 'event' and is a subclass of the `Event` dataclass.
- `methods_name`: A generator expression that yields method names starting with
'solve_req_' or 'solve_res_'.
Workflow:
1. The method iterates over all the method names in the class that start with
'solve_req_' or 'solve_res_'.
2. For each method, it checks if it is a valid instance method. If not, it logs
a warning message and continues to the next method.
3. For valid methods, it retrieves the `event` type by calling `get_event()`
with the method's signature.
4. If the method name starts with 'solve_req_', it associates the method with
the event type in the `_requests` dictionary.
5. If the method name starts with 'solve_res_', it associates the method with
the event type in the `_responses` dictionary.
Attributes:
- `_requests`: A dictionary that stores request handler methods, keyed by event type.
- `_responses`: A dictionary that stores response handler methods, keyed by event type.
- `logger`: A logging object used to log warnings for methods that do not meet
the required criteria.
"""
def get_events(sig: Signature) -> Iterable[Type[Event]]:
# Extract the event parameter from the signature
parameters = list(sig.parameters.values())
if len(parameters) < 1:
raise AttributeError("The solve method must have at least two parameters, "
"with the second named 'event'.")
parameter_event = parameters[0]
if parameter_event.name != "event":
raise AttributeError(
f"Second parameter of solve method must be named 'event', not '{parameter_event.name}'."
)
annotation = parameter_event.annotation
# Check if it's a union of event types
if get_origin(annotation) is UnionType:
args = get_args(annotation)
# Verify each type in the union is a valid subclass of Event
for arg in args:
if not (isinstance(arg, type) and issubclass(arg, Event)):
raise AttributeError(
f"Each event type in the union must be a subclass of Event, not: {arg}"
)
return args
# If not a union, it must be a single event class
# Check if it's actually a type and a subclass of Event
if not (isinstance(annotation, type) and issubclass(annotation, Event)):
raise AttributeError(
f"Type of event must be a subclass of Event dataclass, not: {annotation}"
)
return (annotation,)
methods_name = (method_name for method_name in dir(self) if
method_name.startswith(("solve_req_", "solve_res_")))
for method_name in methods_name:
method = getattr(self, method_name)
if not (ismethod(method) or isfunction(method)):
self.logger.warning("%s suppose to be a method", method_name)
continue
for event in get_events(signature(method)):
if method_name.startswith("solve_req_"):
if event in self._requests:
raise TypeError(f"Event is already registered: {event}")
self._requests[event] = method
else:
if event in self._responses:
raise TypeError(f"Event is already registered: {event}", )
self._responses[event] = method
@property
def additional_pipes(self) -> set[Connection]:
"""
Specifies additional pipes for the manager to listen on. Override to add custom pipes.
Returns:
set[Connection]: A set of additional pipes.
"""
return set() # pragma: no cover
# Process declares `name` as a writeable instance attribute; this read-only
# property intentionally overrides it (returns the class name). Adding a
# setter would change runtime behavior, so the override is silenced.
@property
def name(self) -> str: # type: ignore[override]
"""
Specifies the manager's name. Override to provide a custom name.
Returns:
str: The manager's name.
"""
return self.__class__.__name__
[docs]
def resolve_callback_worker(self, event: Event, worker: Worker) -> None:
"""
Default method for handling events received from workers. It forwards received events to the router queue.
Parameters:
event (Event): The received event.
worker (Worker): Information about the worker that sent the event.
"""
if worker.runtime is None:
raise RuntimeError("worker runtime is not available")
event._worker = worker.runtime.name
event._timing.stamp(self.__class__.__name__, "Returned from worker")
if event.response is not None and not event.has_response():
self._track_worker_response(event, worker)
self.send_to_router(event)
def _track_worker_response(self, event: Event, worker: Worker) -> None:
"""
Registers that ``worker`` is awaiting the reply to ``event`` so it can
be routed back once it arrives (see :meth:`_resolve_worker_response`),
auto-subscribing to the reply type the first time it's seen. Does
nothing if the application already owns a handler for this type.
"""
event_type = event.__class__
if event_type in self._responses and event_type not in self._auto_responses:
self.logger.warning(
"Worker %s dispatched %s, but this Manager already owns a "
"response handler for it — the reply will go there, not back "
"to the worker, which will time out waiting.",
worker.runtime.name if worker.runtime is not None else "pending worker",
event_type.__name__,
)
return
self._pending_responses[event._id] = (event_type, worker)
worker.pending_responses.add(event._id)
self._pending_response_counts[event_type] = self._pending_response_counts.get(event_type, 0) + 1
if event_type not in self._auto_responses:
self._responses[event_type] = self._resolve_worker_response
self._auto_responses.add(event_type)
# from_time=False explicitly (not the None default, which means "no
# preference, replay everything cached" — see Subscribe.from_time):
# this is a dynamic, well-after-startup subscribe, so replaying old,
# already-resolved requests of this type would land in
# resolve_unknown() as spurious "unknown event" warnings.
# self._from_time (the manager's own startup timestamp) is stale by
# the time a type is first dispatched at runtime and would not help.
self.send_to_router(
# The @event decorator rewrites the dataclass 'response' field to
# init=False with a default_factory, which mypy's dataclass
# transform cannot see; the proper fix belongs in
# edri/events/edri/router/subscribe.py (outside this scope).
Subscribe(pipe=self._router_send_pipe, name=self.name,
event_type=event_type, request=False, from_time=False) # type: ignore[call-arg]
)
def _resolve_worker_response(self, event: Event) -> None:
"""Delivers an auto-subscribed reply back to the worker that dispatched it."""
entry = self._pending_responses.pop(event._id, None)
if entry is None:
self.logger.debug("Unmatched worker response (no pending worker): %s", event)
return
event_type, worker = entry
worker.pending_responses.discard(event._id)
self._release_pending_response_type(event_type)
if worker.pipe is None:
self.logger.debug("Worker response arrived after worker pipe closed: %s", event)
return
worker.pipe.send(event)
def _release_pending_response_type(self, event_type: Type[Event]) -> None:
"""Drops the shared auto-registration for ``event_type`` once nothing is pending on it anymore."""
remaining = self._pending_response_counts.get(event_type, 0) - 1
if remaining > 0:
self._pending_response_counts[event_type] = remaining
return
self._pending_response_counts.pop(event_type, None)
self._auto_responses.discard(event_type)
self._responses.pop(event_type, None)
self.send_to_router(
# The @event decorator rewrites the dataclass 'response' field to
# init=False with a default_factory, which mypy's dataclass transform
# cannot see; the proper fix belongs in
# edri/events/edri/router/unsubscribe.py (outside this scope).
Unsubscribe(name=self.name, event_type=event_type, request=False) # type: ignore[call-arg]
)
def _ignore_router_ack(self, event: Event) -> None:
"""No-op handler for the Subscribe/Unsubscribe acks that dynamic (un)registration
(see :meth:`_track_worker_response`/:meth:`_release_pending_response_type`) causes
to arrive on the router pipe during normal operation, not just at startup."""
pass
[docs]
def resolve_callback_pipe(self, event: Event, pipe: Connection) -> None:
"""
Handles events received from unknown pipes, which are not associated with known workers.
Parameters:
event (Event): The received event.
pipe (Connection): The pipe from which the event was received.
"""
self.logger.warning("Unknown event from unknown pipe: %s %s", event, pipe) # pragma: no cover
[docs]
def resolve_unknown(self, event: Event) -> None:
"""
Handles unknown events received from the router pipe that do not match any registered request or response.
Parameters:
event (Event): The unknown event received.
"""
self.logger.warning("Unknown event: %s", event) # pragma: no cover
def _prepare(self) -> None:
"""
Prepares the manager before starting the event resolver loop, initializing necessary attributes and structures.
"""
pass
def _subscribe_static_resolvers(self) -> None:
"""
Registers static resolver methods for specific event types, such as health checks and store get requests.
"""
self.logger.debug("Registering static resolvers")
self._requests[HealthCheck] = self.health_check
self._requests[Restart] = self._restart
self._store_get = self._responses.get(Get, None)
self._responses[Get] = self.store_get
def _restart(self, event: Restart) -> None:
if event.manager == type(self):
self.logger.warning("Restarting manager: %s", self.name)
self.logger.debug("Reloading manager source code", )
try:
invalidate_caches()
mod = import_module(self.__class__.__module__)
mod = reload(mod)
except Exception as e:
self.logger.error("Unknown error during reloading of manager", exc_info=e)
return
try:
klass_import = getattr(mod, self.__class__.__name__)
except AttributeError as e:
self.logger.error("Class %s not found", self.__class__.__name__, exc_info=e)
return
except Exception as e:
self.logger.error("Unknown error during importing of manager", exc_info=e)
return
self.logger.info("Removing old subscription: %s", self.name)
now = datetime.now()
self._unsubscribe_all()
self._resolve_undelivered_events()
self.logger.info("Starting new manager: %s", self.name)
klass_import(self.router_queue, from_time=now).start()
self.router_pipe.close()
self.router_queue.close()
self.logger.warning("Stopping manager: %s", self.name)
exit(0)
def _find_worker(self, pipe: Connection) -> Tuple[str, Worker]:
"""
Finds and returns the worker associated with a given pipe. This method iterates through all registered
workers, comparing the given pipe with each worker's associated pipe. When it finds a match, it returns
both the identifier of the worker and the worker instance itself.
Parameters:
pipe (Connection): The communication pipe to match with a worker. This is used to identify the specific
worker associated with the given communication channel.
Returns:
Tuple[str, Worker]: A tuple containing the worker's identifier and the worker instance itself. The
identifier is a unique string that represents the worker, and the Worker is the
actual instance handling tasks for a specific pipe.
Raises:
ValueError: Raised if no worker is found that matches the given pipe. This exception indicates that
the pipe provided does not correspond to any currently registered worker. It's crucial for
callers to handle this exception, especially in cases where dynamic registration and
deregistration of workers may lead to situations where a pipe is no longer valid.
Example:
To find a worker associated with a specific pipe, you would call this method with the pipe as the argument.
If a worker is found that uses this pipe for communication, the method will return the worker's identifier
and the worker instance. If no such worker exists, a ValueError will be raised.
"""
return self.workers.find_by_pipe(pipe)
def _worker_runtime_started(self, worker_id: str, worker: Worker) -> None:
"""Hook for Manager variants that must wake after a runtime pipe appears."""
pass
def _worker_deadline_changed(self) -> None:
"""Hook for Manager variants whose current wait may need shortening."""
pass
def _worker_runtime_finishing(self, worker_id: str, worker: Worker) -> None:
"""Hook invoked before the controller closes a runtime's manager pipe."""
pass
[docs]
def resolve(self, event: Event) -> None:
"""Resolves one event through independently overridable stages."""
resolver = self.get_event_resolver(event)
if resolver is None:
self.resolve_unknown(event)
return
had_response = event.has_response()
cache = self.get_resolver_cache_state(event, resolver)
if self.resolve_not_modified(event, cache):
return
self.execute_event_resolver(event, resolver)
self.record_resolved_event(event)
if not self.should_send_resolved_event(event, had_response):
return
self.prepare_resolved_event(event, cache)
self.send_to_router(event)
[docs]
def get_event_resolver(self, event: Event) -> Callable[..., Any] | None:
"""Returns the registered request or response resolver for ``event``."""
resolvers = self._responses if event.has_response() else self._requests
return resolvers.get(event.__class__)
[docs]
def get_resolver_cache_state(
self,
event: Event,
resolver: Callable[..., Any],
) -> ResolverCacheState:
"""Builds the cache state used while resolving one event."""
cache_key = self._cache_keys.get(resolver.__name__)
if not cache_key:
return ResolverCacheState()
return ResolverCacheState(
key=cache_key,
etag=self.cache.tag(cache_key),
method=self._cache_methods[event.__class__],
)
[docs]
def resolve_not_modified(
self,
event: Event,
cache: ResolverCacheState,
) -> bool:
"""Sends a cached ``304 Not Modified`` response when applicable."""
if not self.is_not_modified_request(event, cache):
return False
cast(Response, event.response).add_directive(NotModifiedResponseDirective())
self.send_to_router(event)
return True
[docs]
def is_not_modified_request(
self,
event: Event,
cache: ResolverCacheState,
) -> bool:
"""Returns whether ``event`` can be answered with ``304 Not Modified``."""
return bool(
cache.enabled
and cache.etag
and cache.method == HTTPMethod.GET
and hasattr(event, "etag")
and event.etag
and f'"{cache.etag}"' in event.etag
)
[docs]
def execute_event_resolver(
self,
event: Event,
resolver: Callable[..., Any],
) -> None:
"""Executes the selected resolver."""
resolver(event)
[docs]
def record_resolved_event(self, event: Event) -> None:
"""Updates resolver health counters after successful resolver execution."""
self._processed_total += 1
if event.has_response() and cast(Response, event.response).get_status() == ResponseStatus.FAILED:
self._failed_total += 1
[docs]
def should_send_resolved_event(
self,
event: Event,
had_response: bool,
) -> bool:
"""Returns whether resolver execution created a changed response to send."""
return cast(
bool,
not had_response
and event.has_response()
and cast(Response, event.response)._changed,
)
[docs]
def prepare_resolved_event(
self,
event: Event,
cache: ResolverCacheState,
) -> None:
"""Applies framework response metadata before the common outgoing hook."""
if event._switch:
event._switch.received = False
self.prepare_cache_response(event, cache)
[docs]
def prepare_cache_response(
self,
event: Event,
cache: ResolverCacheState,
) -> None:
"""Renews cache state or attaches HTTP cache response directives."""
if not cache.enabled or not cache.etag:
return
if (
cast(Response, event.response).get_status() == ResponseStatus.OK
and cache.method in (
HTTPMethod.POST,
HTTPMethod.PUT,
HTTPMethod.PATCH,
HTTPMethod.DELETE,
)
):
self.cache.renew(cast(str, cache.key))
return
cast(Response, event.response).add_directive(
HeaderResponseDirective(name="ETag", value=f'"{cache.etag}"')
)
cast(Response, event.response).add_directive(
HeaderResponseDirective(name="Cache-Control", value=API_CACHE_CONTROL)
)
cast(Response, event.response).add_directive(
HeaderResponseDirective(name="Vary", value=self._cache_vary)
)
[docs]
def prepare_outgoing_event(self, event: Event) -> None:
"""Hook called immediately before an event enters the router queue."""
pass
[docs]
def send_to_router(self, event: Event) -> None:
"""Runs the common outgoing hook and sends ``event`` to the router."""
self.prepare_outgoing_event(event)
self.router_queue.put(event)
[docs]
def get_pipes(self) -> set[Connection]:
"""
Collects all communication pipes used by the manager, including the router pipe and any worker pipes.
Returns:
set[Connection]: A set of all pipes for the manager to listen on.
"""
pipes = set()
if self.router_pipe:
pipes.add(self.router_pipe)
if self.workers:
worker_pipes = self.workers.pipes()
pipes.update(worker_pipes)
self.logger.debug(
"Count of workers to listen to: running=%s pending=%s limit=%s",
len(worker_pipes),
len(self.workers.pending),
self.max_workers,
)
pipes.update(self.additional_pipes)
pipes.difference_update(self._disabled_additional_pipes)
return pipes
resolver_wait_timeout: float = 1.0
resolver_idle_delay: float = 1.0
[docs]
def run_resolver(self) -> None:
"""Runs resolver cycles until interrupted or explicitly stopped."""
self._resolver_running = True
try:
while self._resolver_running:
self.run_resolver_cycle()
except KeyboardInterrupt:
pass
finally:
self._resolver_running = False
[docs]
def stop_resolver(self) -> None:
"""Requests a clean exit from :meth:`run_resolver`."""
self._resolver_running = False
[docs]
def run_resolver_cycle(self) -> None:
"""Executes one iteration of the resolver loop."""
pipes = self.get_pipes()
if not pipes:
self.workers._process_timeouts()
self.on_resolver_idle()
return
active_pipes = self.wait_for_active_pipes(pipes)
# A lifecycle frame that became readable by its deadline is evidence
# that the transition completed in time. Consume all ready worker
# frames before expiring deadlines so manager scheduling latency cannot
# turn WorkerStarted/WorkerQuit into a false timeout.
worker_pipes = self.workers.pipes()
ready_workers = [pipe for pipe in active_pipes if pipe in worker_pipes]
ready_others = [pipe for pipe in active_pipes if pipe not in worker_pipes]
for active_pipe in ready_workers:
self.resolve_active_pipe(active_pipe)
self.workers._process_timeouts()
for active_pipe in ready_others:
self.resolve_active_pipe(active_pipe)
self.workers._process_timeouts()
if not self._resolver_running:
break
[docs]
def on_resolver_idle(self) -> None:
"""Handles a resolver cycle in which there are no pipes to monitor."""
sleep(self.workers._next_timeout(self.resolver_idle_delay) or 0.0)
[docs]
def wait_for_active_pipes(self, pipes: set[Connection]) -> list[Connection]:
"""Waits until one or more monitored pipes become readable."""
# multiprocessing.connection.wait() is typed to return a union list that
# includes socket/int; only Connection pipes are waited on and resolved,
# so the result is narrowed with an erased cast (no runtime change).
return cast(list[Connection], wait(pipes, timeout=self.workers._next_timeout(self.resolver_wait_timeout)))
[docs]
def resolve_active_pipe(self, pipe: Connection) -> None:
"""Dispatches a readable pipe to its source-specific resolver."""
if self.is_router_pipe(pipe):
self.resolve_router_pipe(pipe)
return
self.resolve_non_router_pipe(pipe)
[docs]
def is_router_pipe(self, pipe: Connection) -> bool:
"""Returns whether ``pipe`` is the manager's router connection."""
return pipe == self.router_pipe
[docs]
def resolve_router_pipe(self, pipe: Connection) -> None:
"""Receives and resolves one event from the router connection."""
event = self.receive_pipe_event(pipe)
if event is None:
return
if HEALTH_CHECK_LOG or not isinstance(event, HealthCheck):
self.logger.debug("Received event from router: %s", event)
self.resolve_router_event(event)
[docs]
def resolve_router_event(self, event: Event) -> None:
"""Resolves one router event and records resolver failures."""
event._timing.stamp(self.__class__.__name__, "Received")
try:
self.resolve(event)
except Exception as error:
self.handle_resolver_exception(event, error)
else:
event._timing.stamp(self.__class__.__name__, "Resolved")
[docs]
def handle_resolver_exception(self, event: Event, error: Exception) -> None:
"""Logs and stores an exception raised while resolving a router event."""
self.logger.error("Error while resolving event", exc_info=error)
self._exceptions.append(
(
event.__class__.__name__,
event.as_dict(transform=False, keep_concealed=False),
str(error),
format_exc(),
datetime.now(),
)
)
[docs]
def resolve_non_router_pipe(self, pipe: Connection) -> None:
"""Receives and dispatches one event from a worker or additional pipe."""
event = self.receive_pipe_event(pipe)
if event is None:
return
self.logger.debug("Received event: %s", event)
self.resolve_non_router_event(event, pipe)
[docs]
def receive_pipe_event(self, pipe: Connection) -> Event | None:
"""Receives one event using the shared pipe error policy."""
try:
event: Event = pipe.recv()
except (EOFError, OSError) as error:
self.handle_pipe_receive_error(pipe, error)
return None
return event
[docs]
def handle_pipe_receive_error(
self,
pipe: Connection,
error: EOFError | OSError,
) -> None:
"""Dispatches a pipe read failure according to the pipe's owner."""
if self.is_router_pipe(pipe):
self.handle_router_pipe_receive_error(pipe, error)
return
try:
worker_key, worker = self._find_worker(pipe)
except ValueError:
self.handle_additional_pipe_receive_error(pipe, error)
return
self.handle_worker_pipe_receive_error(
pipe,
worker_key,
worker,
error,
)
[docs]
def handle_router_pipe_receive_error(
self,
pipe: Connection,
error: EOFError | OSError,
) -> None:
"""Stops the resolver when its router connection is no longer usable."""
self.logger.error("Router communication failed", exc_info=error)
self.close_failed_pipe(pipe)
self.stop_resolver()
[docs]
def handle_worker_pipe_receive_error(
self,
pipe: Connection,
worker_key: str,
worker: Worker,
error: EOFError | OSError,
) -> None:
"""Removes a worker whose communication pipe has closed or failed."""
self.logger.error(
"Worker %s communication failed",
worker.runtime.name if worker.runtime is not None else worker_key,
exc_info=error,
)
self.workers.finish_lost(worker_key)
[docs]
def handle_additional_pipe_receive_error(
self,
pipe: Connection,
error: EOFError | OSError,
) -> None:
"""Disables an application-defined pipe after a failed read."""
self.logger.error(
"Additional pipe communication failed: %s",
pipe,
exc_info=error,
)
self.disable_additional_pipe(pipe)
[docs]
def disable_additional_pipe(self, pipe: Connection) -> None:
"""Excludes a failed pipe from future wait sets and closes it."""
self._disabled_additional_pipes.add(pipe)
self.close_failed_pipe(pipe)
[docs]
def close_failed_pipe(self, pipe: Connection) -> None:
"""Closes a failed pipe while suppressing secondary close errors."""
try:
pipe.close()
except OSError as error:
self.logger.debug("Failed pipe could not be closed", exc_info=error)
[docs]
def resolve_non_router_event(self, event: Event, pipe: Connection) -> None:
"""Routes an event from a worker pipe or an application-defined pipe."""
try:
worker_key, worker = self._find_worker(pipe)
except ValueError:
self.resolve_callback_pipe(event, pipe)
return
self.resolve_worker_event(event, worker_key, worker)
[docs]
def resolve_worker_event(
self,
event: Event,
worker_key: str,
worker: Worker,
) -> None:
"""Routes a worker event either as control traffic or application traffic."""
if isinstance(event, Manager):
self.resolve_worker_control_event(event, worker_key, worker)
return
self.resolve_callback_worker(event, worker)
[docs]
def resolve_worker_control_event(
self,
event: Manager,
worker_key: str,
worker: Worker,
) -> None:
"""Handles lifecycle and other control events received from a worker."""
if isinstance(event, WorkerStarted):
self.workers.started(worker_key, event.started)
elif isinstance(event, WorkerQuit):
self.workers.finish_from_quit(worker_key, event)
[docs]
def start_worker(
self,
event: Optional[Event],
resolver: WorkerFactoryProtocol,
/,
*args: Any,
environment: Path | None = None,
**kwargs: Any,
) -> None:
warn(
"ManagerBase.start_worker() is deprecated; use self.workers.start()",
DeprecationWarning,
stacklevel=2,
)
self.workers.start(
event,
resolver,
*args,
environment=environment,
**kwargs,
)
[docs]
def health_check(self, event: HealthCheck) -> None:
"""
Responds to a health check request from the router, indicating the manager's status.
Parameters:
event (HealthCheck): The health check event received from the router.
"""
report = self.health()
event.response.name = self.name
event.response.status = report.status
event.response.ready = report.ready
event.response.reason = report.reason
event.response.exceptions = self._exceptions
event.response.vitals = self._collect_vitals()
self._exceptions = []
[docs]
def health(self) -> HealthReport:
"""
Assess this component's health for a health-check response — override hook.
The default derives status purely from exceptions caught since the last
check (any → WARNING) and always reports the component as ready. Override
to add domain knowledge — typically checking the dependencies this
component needs and downgrading accordingly::
def health(self) -> HealthReport:
report = super().health()
if not self._database.is_connected():
return HealthReport(Status.DEGRADED, ready=False,
reason="database unreachable")
return report
The caught exceptions are still reported separately (for the detail view)
regardless of what this returns, so an override should escalate status,
not suppress exceptions. Keep it cheap and non-blocking: it runs on every
health-check round-trip, inline on the component's main loop.
Returns:
HealthReport: The component's self-assessment (status, readiness, reason).
"""
return HealthReport(status=Status.WARNING if self._exceptions else Status.OK)
[docs]
def collect_custom_vitals(self) -> dict[str, Any]:
"""
Add application-defined metrics to the health-check vitals — override hook.
The base implementation reports nothing. Override to attach your own
gauges and counters, which ride along on the periodic health round-trip
and appear under ``vitals.custom`` on the status endpoint::
def collect_custom_vitals(self) -> dict[str, Any]:
return {
"queue_depth": self._inbox.qsize(),
"cache_entries": len(self.cache),
"last_sync_age_seconds": time() - self._last_sync,
}
Keep it cheap and non-blocking: it runs inline on the component's main
loop on every check. Return JSON-serializable primitives so the values
survive both the pipe round-trip and the endpoint's JSON encoding. Any
exception raised here is caught and logged — it degrades to empty custom
vitals rather than wiping out the standard ones.
Returns:
dict[str, Any]: Custom metric name → value. Empty by default.
"""
return {}
def _collect_vitals(self) -> Vitals:
"""
Snapshots the component's operational metrics for a health-check response.
Point-in-time metrics (worker occupancy and backlog, oldest worker age,
memory, threads, fds) are read at call time. Interval deltas (events
processed, of which failed, and CPU time consumed) are computed against
watermarks that this method advances, so every check describes only its
own interval.
Returns:
Vitals: The metrics to ride along on the health-check round-trip.
"""
processed = self._processed_total - self._processed_reported
self._processed_reported = self._processed_total
failed = self._failed_total - self._failed_reported
self._failed_reported = self._failed_total
cpu_total = cpu_seconds_total()
cpu_seconds = cpu_total - self._cpu_reported
self._cpu_reported = cpu_total
active_workers = [
worker for worker in self.workers.values()
if worker.state is not WorkerState.PENDING
]
oldest_age: float | None = None
if active_workers:
now = monotonic()
anchors = (
worker.started if worker.started is not None else worker.spawned
for worker in active_workers
)
ages = [now - anchor for anchor in anchors if anchor is not None]
if ages:
oldest_age = max(ages)
try:
custom = self.collect_custom_vitals()
except Exception as e:
# A buggy custom probe must not wipe out the standard vitals.
self.logger.error("collect_custom_vitals failed", exc_info=e)
custom = {}
return Vitals(
# STARTING/STOPPING/TERMINATING still occupy max_workers slots.
workers_running=len(active_workers),
workers_pending=len(self.workers.pending),
workers_limit=self.max_workers,
workers_oldest_age_seconds=oldest_age,
processed=processed,
failed=failed,
cpu_seconds=cpu_seconds,
rss_bytes=current_rss_bytes(),
rss_peak_bytes=peak_rss_bytes(),
threads=active_count(),
open_fds=open_fd_count(),
pid=getpid(),
uptime_seconds=monotonic() - self._started_at,
custom=custom,
)
[docs]
def store_get(self, event: Get) -> None:
"""
Handles the store get response for retrieving data from the Store. This method determines whether
the response should be directed to a specific worker or handled by the manager itself.
The method first checks if the event is associated with a worker. If the event's `_worker` attribute
is set and matches a worker managed by this instance, the event is sent to that worker via its communication
pipe. If no matching worker is found, or if the `_worker` attribute is not set, the method attempts to handle
the event using the `_store_get` method, if it is defined.
Parameters:
event (Get): The get response event, containing details of the data to retrieve and the retrieved data.
Behavior:
- If the event specifies a worker (via the `_worker` attribute), the method directs the response to that worker.
- If no worker is specified or found, the event is passed to the manager's own `_store_get` method for standard handling,
provided that `_store_get` has been defined. If `_store_get` is not defined, the event is effectively ignored.
Note:
- The `_store_get` method is an optional user-defined method. If it is not defined, this method will not attempt to
handle the event beyond checking for a worker.
"""
if self.workers and event._worker is not None:
for worker in self.workers.values():
if worker.runtime is not None and worker.runtime.name == event._worker:
assert worker.pipe is not None
worker.pipe.send(event)
return
if self._store_get:
self._store_get(event)
[docs]
def run(self) -> None:
"""
Main entry point for the manager process. Sets up event subscriptions, prepares the manager,
and enters the main event resolution loop.
"""
self.logger.debug(f"{self.name} is running!")
# Re-stamp the start clock in the child process: __init__ runs in the
# parent (and a spawned child may not re-run it), so uptime must be
# anchored here, where the process's own run loop actually begins.
self._started_at = monotonic()
# Shut down cleanly on SIGTERM (systemd/docker/k8s default) the same way
# the resolver loop already does on SIGINT/KeyboardInterrupt.
install_graceful_sigterm()
try:
self._start_and_resolve()
finally:
self.workers._close()
def _start_and_resolve(self) -> None:
"""Runs start-up and then the resolver loop; async managers run both inside the event loop."""
self._start_up()
self.run_resolver()
def _start_up(self) -> None:
"""Subscribes, prepares the manager, and resolves the events received while subscribing."""
self._prepare_resolvers()
self._subscribe_static_resolvers()
self._subscribe()
self._prepare()
self.after_start()
self._resolve_undelivered_events()
# Same PEP 649 lazy-evaluation concern as __init__ (see above): keep the
# parameter annotation Any so the signature stays evaluable at runtime.
[docs]
def start(self, router_queue: Optional[Any] = None) -> None:
"""
Starts the manager process with an optional router queue specified. This method initiates the manager's
operation, allowing it to begin processing events. The router queue is a crucial component for communication
between the manager and the router, facilitating the dispatch and handling of events.
Parameters:
router_queue (Optional[Queue[Event]]): The event queue used for communication with the router. If provided,
this queue will be set as the current router queue for the manager.
This queue is essential for the manager to receive and process events.
Raises:
RuntimeError: Raised if no router queue is provided and no router queue has been set previously. This
exception ensures that the manager cannot start without a means to communicate with the
router, as the lack of a router queue would prevent the manager from functioning correctly.
Note:
If a router queue is not provided as an argument to this method, the manager will attempt to use a
previously set router queue. It's essential to either provide a router queue when calling this method
or set one before its execution. Failure to do so will result in a RuntimeError, indicating the
manager's inability to start due to the absence of a communication channel with the router.
Example:
To start the manager, you may either provide a router queue directly when calling the start method
or ensure that the manager has a router queue set beforehand. This flexibility allows for different
use cases, such as reusing an existing queue or setting a new one as needed.
"""
if router_queue is not None:
self.router_queue = router_queue
if self.router_queue is None:
raise RuntimeError("Queue must be specified")
self.router_queue = self.router_queue
super().start()
[docs]
def after_start(self) -> None:
"""
Hook method called after the manager process starts. Can be overridden to perform initialization tasks.
"""
self.cache = Cache()
[docs]
def quit(self) -> None:
self.stop_resolver()
self.router_queue.close()
count = len(self._requests) + len(self._responses)
if count > 0:
self.router_pipe.close()