{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "07789668",
   "metadata": {},
   "source": [
    "# How to follow, stop and time-limit long-running tasks\n",
    "\n",
    "`nr.run()` returns when every host has finished. For long operations, such as a software upgrade, an image copy or a commit-confirm, that is often not enough: you want to see each host's progress while it runs, stop the hosts that are still running when one host fails, and give each host a deadline.\n",
    "\n",
    "This guide uses `ThreadedRunner`. The same calls work on `AsyncioRunner` with `asubmit` and `astream`, and `stream` works on `SerialRunner`. The tasks below sleep instead of talking to devices, so the guide runs anywhere."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7276a4f0",
   "metadata": {},
   "source": [
    "## Set up\n",
    "\n",
    "The configuration selects `ThreadedRunner` with one worker, so that the output below always comes in the same order. Paths are relative to the notebook working directory, `docs/howto/`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "99860eed",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-11-14T11:19:06.258821Z",
     "iopub.status.busy": "2026-11-14T11:19:06.258751Z",
     "iopub.status.idle": "2026-11-14T11:19:06.489383Z",
     "shell.execute_reply": "2026-11-14T11:19:06.489051Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Runner: ThreadedRunner, hosts: edge-1, edge-2, edge-3, edge-4\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "\n",
    "from nornir import InitNornir\n",
    "from nornir.core.exceptions import TaskCancelledError, TaskTimeoutError\n",
    "from nornir.core.run import RunOptions\n",
    "from nornir.core.task import PartialResult, Result, Task\n",
    "from nornir.plugins.runners import ThreadedRunner\n",
    "\n",
    "nr = InitNornir(config_file=\"long_running_tasks/config.yaml\")\n",
    "print(f\"Runner: {type(nr.runner).__name__}, hosts: {', '.join(nr.inventory.hosts)}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "099dee46",
   "metadata": {},
   "source": [
    "## Report progress from a task\n",
    "\n",
    "A task reports progress by yielding `PartialResult` objects. Its `return` value is its final result, as for any other task. A partial result never marks a host as failed and is not stored in the host's results.\n",
    "\n",
    "`nr.stream()` runs the task and yields each event as it happens: the partial results, then the final `Result` of each host."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "3985e0f7",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-11-14T11:19:06.490371Z",
     "iopub.status.busy": "2026-11-14T11:19:06.490237Z",
     "iopub.status.idle": "2026-11-14T11:19:06.573811Z",
     "shell.execute_reply": "2026-11-14T11:19:06.573580Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "edge-1: copying eos-4.32.bin (30%)\n",
      "edge-1: rebooting (80%)\n",
      "edge-1: done, running eos-4.32.bin\n",
      "edge-2: copying eos-4.32.bin (30%)\n",
      "edge-2: rebooting (80%)\n",
      "edge-2: done, running eos-4.32.bin\n",
      "edge-3: copying eos-4.32.bin (30%)\n",
      "edge-3: rebooting (80%)\n",
      "edge-3: done, running eos-4.32.bin\n",
      "edge-4: copying eos-4.32.bin (30%)\n",
      "edge-4: rebooting (80%)\n",
      "edge-4: done, running eos-4.32.bin\n"
     ]
    }
   ],
   "source": [
    "def upgrade(task: Task, image: str):\n",
    "    yield PartialResult(host=task.host, result=f\"copying {image}\", progress=0.3)\n",
    "    time.sleep(0.01)  # stands in for the image copy\n",
    "    yield PartialResult(host=task.host, result=\"rebooting\", progress=0.8)\n",
    "    time.sleep(0.01)  # stands in for the reboot\n",
    "    return Result(host=task.host, result=f\"running {image}\", changed=True)\n",
    "\n",
    "\n",
    "for event in nr.stream(task=upgrade, image=\"eos-4.32.bin\"):\n",
    "    if isinstance(event, PartialResult):\n",
    "        print(f\"{event.host.name}: {event.result} ({event.progress:.0%})\")\n",
    "    else:\n",
    "        print(f\"{event.host.name}: done, {event.result}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e0ed446e",
   "metadata": {},
   "source": [
    "`nr.run()` runs the same task and returns the usual `AggregatedResult`. The partial results still reach processors that define `task_instance_progress`, so a progress display can live in a processor:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "60451722",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-11-14T11:19:06.574728Z",
     "iopub.status.busy": "2026-11-14T11:19:06.574656Z",
     "iopub.status.idle": "2026-11-14T11:19:06.658342Z",
     "shell.execute_reply": "2026-11-14T11:19:06.658038Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[progress] edge-1: copying eos-4.32.bin\n",
      "[progress] edge-1: rebooting\n",
      "[progress] edge-2: copying eos-4.32.bin\n",
      "[progress] edge-2: rebooting\n",
      "[progress] edge-3: copying eos-4.32.bin\n",
      "[progress] edge-3: rebooting\n",
      "[progress] edge-4: copying eos-4.32.bin\n",
      "[progress] edge-4: rebooting\n",
      "Results stored for edge-1: ['running eos-4.32.bin']\n"
     ]
    }
   ],
   "source": [
    "class PrintProgress:\n",
    "    def task_instance_progress(self, task, host, partial):\n",
    "        print(f\"[progress] {host.name}: {partial.result}\")\n",
    "\n",
    "    def task_started(self, task): ...\n",
    "    def task_completed(self, task, result): ...\n",
    "    def task_instance_started(self, task, host): ...\n",
    "    def task_instance_completed(self, task, host, result): ...\n",
    "    def subtask_instance_started(self, task, host): ...\n",
    "    def subtask_instance_completed(self, task, host, result): ...\n",
    "\n",
    "\n",
    "result = nr.with_processors([PrintProgress()]).run(task=upgrade, image=\"eos-4.32.bin\")\n",
    "print(f\"Results stored for edge-1: {[r.result for r in result['edge-1']]}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "67d2bc1e",
   "metadata": {},
   "source": [
    "## Stop the remaining hosts after a failure\n",
    "\n",
    "`nr.submit()` starts the run in the background and returns a handle at once. `handle.results()` yields the final result of each host in the order the hosts finish, and `handle.cancel()` stops the run.\n",
    "\n",
    "Cancellation is delivered at the task's next *checkpoint*: a call to `task.checkpoint()`, a `yield`, or the start of a subtask. Hosts that have not started yet never start."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "2bff9a86",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-11-14T11:19:06.659285Z",
     "iopub.status.busy": "2026-11-14T11:19:06.659208Z",
     "iopub.status.idle": "2026-11-14T11:19:07.075965Z",
     "shell.execute_reply": "2026-11-14T11:19:07.075611Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "edge-2 failed, cancelling the run\n",
      "edge-1: change applied\n",
      "edge-2: ValueError\n",
      "edge-3: TaskCancelledError\n",
      "edge-4: TaskCancelledError\n"
     ]
    }
   ],
   "source": [
    "def change(task: Task):\n",
    "    for _ in range(20):\n",
    "        task.checkpoint()\n",
    "        time.sleep(0.01)  # one step of the change\n",
    "    if task.host.name == \"edge-2\":\n",
    "        raise ValueError(\"the new configuration was rejected\")\n",
    "    return Result(host=task.host, result=\"change applied\", changed=True)\n",
    "\n",
    "\n",
    "handle = nr.submit(task=change)\n",
    "for res in handle.results():\n",
    "    if res.failed:\n",
    "        print(f\"{res.host.name} failed, cancelling the run\")\n",
    "        handle.cancel()\n",
    "        break\n",
    "\n",
    "result = handle.result()\n",
    "for name, host_results in result.items():\n",
    "    outcome = host_results[0]\n",
    "    print(f\"{name}: {type(outcome.exception).__name__ if outcome.failed else outcome.result}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "71bb03c1",
   "metadata": {},
   "source": [
    "Every selected host has exactly one final result. A cancelled host carries `TaskCancelledError`, and its `started` field says whether the task had started. Cancelled hosts are added to `nr.data.failed_hosts`, as any failed host is."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "9865cb6e",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-11-14T11:19:07.076847Z",
     "iopub.status.busy": "2026-11-14T11:19:07.076767Z",
     "iopub.status.idle": "2026-11-14T11:19:07.078470Z",
     "shell.execute_reply": "2026-11-14T11:19:07.078250Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['edge-2', 'edge-3', 'edge-4']\n",
      "Task 'change' on host 'edge-4' was cancelled before it started.\n"
     ]
    }
   ],
   "source": [
    "print(sorted(nr.data.failed_hosts))\n",
    "print(result[\"edge-4\"][0].exception)\n",
    "nr.data.reset_failed_hosts()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b97d85ac",
   "metadata": {},
   "source": [
    "`TaskCancelledError` subclasses `BaseException`, like `asyncio.CancelledError`, so an `except Exception` block in a task does not stop a cancellation.\n",
    "\n",
    "## Give each host a deadline\n",
    "\n",
    "A runner takes a per-host deadline (`per_host_timeout`) and a deadline for the whole run (`timeout`), in seconds. They also work from the configuration file, under `runner.options`. A host that passes its deadline is recorded with `TaskTimeoutError`; the other hosts are not affected."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "c26ae1fb",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-11-14T11:19:07.079132Z",
     "iopub.status.busy": "2026-11-14T11:19:07.079067Z",
     "iopub.status.idle": "2026-11-14T11:19:07.645696Z",
     "shell.execute_reply": "2026-11-14T11:19:07.645307Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "edge-1: finished\n",
      "edge-2: finished\n",
      "edge-3: Task 'slow_on_edge_3' on host 'edge-3' exceeded its per-host timeout of 0.5 seconds.\n",
      "edge-4: finished\n"
     ]
    }
   ],
   "source": [
    "def slow_on_edge_3(task: Task):\n",
    "    steps = 100 if task.host.name == \"edge-3\" else 2\n",
    "    for _ in range(steps):\n",
    "        task.checkpoint()\n",
    "        time.sleep(0.01)\n",
    "    return Result(host=task.host, result=\"finished\")\n",
    "\n",
    "\n",
    "deadline_nr = nr.with_runner(ThreadedRunner(num_workers=1, per_host_timeout=0.5))\n",
    "result = deadline_nr.run(task=slow_on_edge_3)\n",
    "for name, host_results in result.items():\n",
    "    print(f\"{name}: {host_results[0].result}\")\n",
    "deadline_nr.data.reset_failed_hosts()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a68e8865",
   "metadata": {},
   "source": [
    "`submit` and `stream` also take `run_options`, which overrides the runner's deadlines for one run. It is a separate argument, so it never takes a name from the task's own arguments:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "883f8c12",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-11-14T11:19:07.646646Z",
     "iopub.status.busy": "2026-11-14T11:19:07.646520Z",
     "iopub.status.idle": "2026-11-14T11:19:08.011624Z",
     "shell.execute_reply": "2026-11-14T11:19:08.011352Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "True host 0.3\n"
     ]
    }
   ],
   "source": [
    "handle = nr.submit(task=slow_on_edge_3, run_options=RunOptions(per_host_timeout=0.3))\n",
    "error = handle.result()[\"edge-3\"][0].exception\n",
    "print(isinstance(error, TaskTimeoutError), error.scope, error.timeout)\n",
    "nr.data.reset_failed_hosts()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cc40c4b8",
   "metadata": {},
   "source": [
    "On `ThreadedRunner`, a deadline is enforced even when a task is blocked in a call that never returns: the run records the host as timed out and returns on time. The blocked thread keeps its worker slot until the call returns, so tasks built on blocking libraries should call `task.checkpoint()` between commands. `AsyncioRunner` interrupts a host at the call it is awaiting, without checkpoints.\n",
    "\n",
    "See [the execution model](../plugins/execution_model.rst) for what each runner supports."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.14.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}