hololinked.core.eventloop.operations.Job dataclass

The operation wrapped around its lifecycle within the event loop.

Source code in repo/hololinked/hololinked/core/eventloop/operations.py

@dataclass
class Job:
    """The operation wrapped around its lifecycle within the event loop."""

    operation: Operation
    """what to do."""
    future: Future
    """Resolved with a `Reply`"""
    # A `concurrent.futures.Future` rather than an `asyncio` one: it is created by whoever submitted
    # and resolved by whichever thread the `Thing` ran on, and only this kind is safe across that gap.
    # An async caller wraps it onto its own loop - see `EventLoop.execute()`.
    started: CrossLoopEvent
    """set when the operation started. The invokation timeout races against this."""
    completed: CrossLoopEvent
    """set when the operation has completed. The execution timeout races against this."""
    invokation_timeout_task: Future | None = None
    """The timeout racing `started`. Awaited once `started` is set, to settle the race."""
    execution_timeout_task: asyncio.Task | None = None
    """The timeout racing `completed`. Awaited once `completed` is set, to settle the race."""

    def answer(self, reply: Reply) -> None:
        """
        Resolve the caller's future, unless a timeout already answered for us.

        Parameters
        ----------
        reply: Reply
            the outcome to hand back to whoever submitted
        """
        try:
            self.future.set_result(reply)
        except InvalidStateError:
            pass  # somebody got there first: a timeout, or a second answer for the same job

    async def answer_if_never_started(self, timeout: float) -> bool:
        """
        Answer the caller with an invokation timeout if this job has not left the queue in time.

        Parameters
        ----------
        timeout: float
            seconds to wait for `started`

        Returns
        -------
        bool
            `True` if it timed out and answered, `False` if the job started first
        """
        try:
            await asyncio.wait_for(self.started.wait(), timeout)
            return False
        except TimeoutError:
            self.answer(TIMED_OUT_REPLY[ReplyKind.INVOKATION_TIMEOUT])
            return True

    async def answer_if_overdue(self, timeout: float) -> bool:
        """
        Answer the caller with an execution timeout if the operation has not finished in time.

        The operation is not cancelled - it cannot be - so its eventual reply is still drained by
        `EventLoop.tunnel_message_to_things()`, then dropped.

        Parameters
        ----------
        timeout: float
            seconds to wait for `completed`

        Returns
        -------
        bool
            `True` if it timed out and answered, `False` if the operation finished first
        """
        try:
            await asyncio.wait_for(self.completed.wait(), timeout)
            return False
        except TimeoutError:
            self.answer(TIMED_OUT_REPLY[ReplyKind.EXECUTION_TIMEOUT])
            return True

Attributes

operation instance-attribute

operation: Operation

what to do.

future instance-attribute

future: Future

Resolved with a Reply

started instance-attribute

started: CrossLoopEvent

set when the operation started. The invokation timeout races against this.

completed instance-attribute

completed: CrossLoopEvent

set when the operation has completed. The execution timeout races against this.

invokation_timeout_task class-attribute instance-attribute

invokation_timeout_task: Future | None = None

The timeout racing started. Awaited once started is set, to settle the race.

execution_timeout_task class-attribute instance-attribute

execution_timeout_task: Task | None = None

The timeout racing completed. Awaited once completed is set, to settle the race.

Functions

__init__

__init__(operation: Operation, future: Future, started: CrossLoopEvent, completed: CrossLoopEvent, invokation_timeout_task: Future | None = None, execution_timeout_task: Task | None = None) -> None

answer

answer(reply: Reply) -> None

Resolve the caller's future, unless a timeout already answered for us.

Parameters:

Name	Type	Description	Default
reply	Reply	the outcome to hand back to whoever submitted	required

Source code in repo/hololinked/hololinked/core/eventloop/operations.py

def answer(self, reply: Reply) -> None:
    """
    Resolve the caller's future, unless a timeout already answered for us.

    Parameters
    ----------
    reply: Reply
        the outcome to hand back to whoever submitted
    """
    try:
        self.future.set_result(reply)
    except InvalidStateError:
        pass  # somebody got there first: a timeout, or a second answer for the same job

answer_if_never_started async

answer_if_never_started(timeout: float) -> bool

Answer the caller with an invokation timeout if this job has not left the queue in time.

Parameters:

Name	Type	Description	Default
timeout	float	seconds to wait for started	required

Returns:

Type	Description
bool	True if it timed out and answered, False if the job started first

Source code in repo/hololinked/hololinked/core/eventloop/operations.py

async def answer_if_never_started(self, timeout: float) -> bool:
    """
    Answer the caller with an invokation timeout if this job has not left the queue in time.

    Parameters
    ----------
    timeout: float
        seconds to wait for `started`

    Returns
    -------
    bool
        `True` if it timed out and answered, `False` if the job started first
    """
    try:
        await asyncio.wait_for(self.started.wait(), timeout)
        return False
    except TimeoutError:
        self.answer(TIMED_OUT_REPLY[ReplyKind.INVOKATION_TIMEOUT])
        return True

answer_if_overdue async

answer_if_overdue(timeout: float) -> bool

Answer the caller with an execution timeout if the operation has not finished in time.

The operation is not cancelled - it cannot be - so its eventual reply is still drained by EventLoop.tunnel_message_to_things(), then dropped.

Parameters:

Name	Type	Description	Default
timeout	float	seconds to wait for completed	required

Returns:

Type	Description
bool	True if it timed out and answered, False if the operation finished first

Source code in repo/hololinked/hololinked/core/eventloop/operations.py

async def answer_if_overdue(self, timeout: float) -> bool:
    """
    Answer the caller with an execution timeout if the operation has not finished in time.

    The operation is not cancelled - it cannot be - so its eventual reply is still drained by
    `EventLoop.tunnel_message_to_things()`, then dropped.

    Parameters
    ----------
    timeout: float
        seconds to wait for `completed`

    Returns
    -------
    bool
        `True` if it timed out and answered, `False` if the operation finished first
    """
    try:
        await asyncio.wait_for(self.completed.wait(), timeout)
        return False
    except TimeoutError:
        self.answer(TIMED_OUT_REPLY[ReplyKind.EXECUTION_TIMEOUT])
        return True