radical.asyncflow.backends¶

def state(self):
    """Get the current state of the no-op execution backend.

    Returns:
        str: Always returns 'IDLE' as this backend performs no actual work.
    """
    return "IDLE"

def task_state_cb(self, task: dict, state: str) -> None:
    """Callback function invoked when a task's state changes.

    Args:
        task: Dictionary containing task information and metadata.
        state: The new state of the task.

    Note:
        This is a no-op implementation that performs no actions.
    """
    pass

def get_task_states_map(self):
    """Retrieve a mapping of task IDs to their current states.

    Returns:
        StateMapper: Object containing the mapping of task states for this backend.
    """
    return StateMapper()

def register_callback(self, func: Callable):
    """Register a callback for task state changes.

    Args:
        func: Function to be called when task states change. Should accept
            task and state parameters.
    """
    self._callback_func = func

def build_task(self, uid, task_desc, task_specific_kwargs):
    """Build or prepare a task for execution.

    Args:
        uid: Unique identifier for the task.
        task_desc: Dictionary containing task description and metadata.
        task_specific_kwargs: Backend-specific keyword arguments for the task.

    Note:
        This is a no-op implementation that performs no actual task building.
    """
    pass

async def submit_tasks(self, tasks):
    """Submit tasks for mock execution.

    Immediately marks all tasks as completed with dummy output without
    performing any actual computation.

    Args:
        tasks: List of task dictionaries to be processed. Each task will
            receive dummy stdout and return_value before being marked as DONE.
    """
    for task in tasks:
        if task.get("prompt"):
            task["stdout"] = "Dummy Prompt Output"
            task["return_value"] = "Dummy Prompt Output"
        else:
            task["stdout"] = "Dummy Output"
            task["return_value"] = "Dummy Output"
        self._callback_func(task, "DONE")

def link_explicit_data_deps(
    self, src_task=None, dst_task=None, file_name=None, file_path=None
):
    """Handle explicit data dependencies between tasks.

    Args:
        src_task: The source task that produces the dependency.
        dst_task: The destination task that depends on the source.
        file_name: Name of the file that represents the dependency.
        file_path: Full path to the file that represents the dependency.

    Note:
        This is a no-op implementation as this backend doesn't handle dependencies.
    """
    pass

def link_implicit_data_deps(self, src_task, dst_task):
    """Handle implicit data dependencies for a task.

    Args:
        src_task: The source task that produces data.
        dst_task: The destination task that depends on the source task's output.

    Note:
        This is a no-op implementation as this backend doesn't handle dependencies.
    """
    pass

async def shutdown(self) -> None:
    """Shutdown the no-op execution backend.

    Performs cleanup operations. Since this is a no-op backend, no actual resources
    need to be cleaned up.
    """
    pass

def __await__(self):
    """Make backend awaitable."""
    return self._async_init().__await__()

def register_callback(self, func: Callable):
    """Register a callback for task state changes.

    Args:
        func: Function to be called when task states change. Should accept
            task and state parameters.
    """
    self._callback_func = func

async def submit_tasks(self, tasks: list[dict[str, Any]]) -> list[asyncio.Task]:
    """Submit tasks for execution."""

    submitted_tasks = []

    for task in tasks:
        future = asyncio.create_task(self._handle_task(task))
        submitted_tasks.append(future)

        self.tasks[task["uid"]] = task
        self.tasks[task["uid"]]["future"] = future

    return submitted_tasks

async def cancel_task(self, uid: str) -> bool:
    """Cancel a task by its UID.

    Args:
        uid (str): The UID of the task to cancel.

    Returns:
        bool: True if the task was found and cancellation was attempted,
              False otherwise.
    """
    if uid in self.tasks:
        task = self.tasks[uid]
        future = task["future"]
        if future and future.cancel():
            # Set state on the task object itself before callback
            self._callback_func(task, "CANCELED")
            return True
    return False

async def cancel_all_tasks(self) -> int:
    """Cancel all running tasks."""
    cancelled_count = 0
    for task in self.tasks.values():
        future = task["future"]
        future.cancel()
        cancelled_count += 1
    self.tasks.clear()
    return cancelled_count

async def shutdown(self) -> None:
    """Shutdown the executor."""
    # Set backend state to SHUTDOWN
    await self.cancel_all_tasks()
    self.executor.shutdown(wait=True)
    logger.info("Concurrent execution backend shutdown complete")

def state(self) -> str:
    """Get backend state.

    Returns:
        str: Current backend state (INITIALIZED, RUNNING, SHUTDOWN)
    """
    return "INITIALIZED" if self._initialized else "IDLE"

async def __aenter__(self):
    """Async context manager entry."""
    if not self._initialized:
        await self._async_init()
    return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
    """Async context manager exit."""
    await self.shutdown()

@classmethod
async def create(cls, executor: Executor) -> "LocalExecutionBackend":
    """Alternative factory method for creating initialized backend.

    Args:
        executor: A concurrent.Executor instance (ThreadPoolExecutor
                  or ProcessPoolExecutor).

    Returns:
        Fully initialized LocalExecutionBackend instance.
    """
    backend = cls(executor)
    return await backend