{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Using Multiple Pilots\n",
    "\n",
    "RADICAL-Pilot supports managing multiple pilots during a single run, while workload (bag of computing tasks) will be distributed among all available pilots. [rp.TaskManager](../apidoc.rst#radical.pilot.TaskManager) dispatches tasks to pilots according to a scheduling algorithm ([rp.SCHEDULER_ROUND_ROBIN](../apidoc.rst#radical.pilot.TaskManager.scheduler), [rp.SCHEDULER_BACKFILLING](../apidoc.rst#radical.pilot.TaskManager.scheduler)), which is provided during its initialization.\n",
    "\n",
    "<div class=\"alert alert-info\">\n",
    "\n",
    "__Note:__ RADICAL-Pilot enacts a second scheduling step within each launched pilot: once the Agent takes ownership of tasks, which the [rp.TaskManager](../apidoc.rst#radical.pilot.TaskManager) scheduler assigned to it, the Agent Scheduler will place the tasks on the set of resources (CPU cores, GPUs) that the Agent is managing. The Agent Scheduler can be configured via resource configuration files (see the tutorial [RADICAL-Pilot Configuration System](configuration.ipynb))\n",
    "\n",
    "</div>\n",
    "\n",
    "* Round-Robin Scheduler, RR (`rp.SCHEDULER_ROUND_ROBIN`, __default__) fairly distributes arriving tasks over the set of known pilots, independent of task state, expected workload, pilot state or pilot lifetime. As such, RR is a fairly simplistic, but also a very fast scheduler, which does not impose any additional communication round trips between [rp.TaskManager](../apidoc.rst#radical.pilot.TaskManager) and pilot agents.\n",
    "\n",
    "* Backfilling Scheduler, BF (`rp.SCHEDULER_BACKFILLING`) provides a better load balancing, but at the cost of additional communication round trips. It depends on the actual application workload if that load balancing is beneficial or not. It is most beneficial for the large number of pilots and for relatively long-running tasks (e.g., the task runtime is significantly longer than the communication round trip time between [rp.TaskManager](../apidoc.rst#radical.pilot.TaskManager) and pilot agents).\n",
    "\n",
    "   * It is __not recommended__ to use BF for: (i) a single pilot, and/or (ii) many short-running tasks.\n",
    "   * BF will only dispatch tasks to pilot agents once the pilot agent is in the `rp.PMGR_ACTIVE` state. The tasks will thus get executed even if one of the pilots never reaches that state - the load will be distributed between pilots which become active (`rp.PMGR_ACTIVE`).\n",
    "   * BF will only dispatch as many tasks to the Agent as they can be executed concurrently. No tasks will be waiting in the Agent Scheduler queue. BF will react on task termination events, and will then backfill (!) the Agent with any remaining tasks. The Agent will remain under-utilized during that communication.\n",
    "\n",
    "## Examples\n",
    "\n",
    "<div class=\"alert alert-info\">\n",
    "\n",
    "__Note:__ In our examples, we will not show a progression bar while waiting for some operation to complete, e.g., while waiting for a pilot to stop. That is because the progression bar offered by RP's reporter does not work within a notebook. You could use it when executing an RP application as a standalone Python script.\n",
    "\n",
    "</div>\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%env RADICAL_REPORT=TRUE\n",
    "%env RADICAL_REPORT_ANIME=FALSE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import radical.pilot as rp\n",
    "import radical.utils as ru"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "session = rp.Session()\n",
    "pmgr    = rp.PilotManager(session=session)\n",
    "tmgr    = rp.TaskManager(session=session)  # -> scheduler=rp.SCHEDULER_ROUND_ROBIN"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We run multiple pilots using a description for a \"localhost\" platform (`local.localhost`), which is provided by the built-in configuration file `resource_local.json` and mimics a configuration for CPU cores and GPUs.\n",
    "\n",
    "<div class=\"alert alert-info\">\n",
    "\n",
    "__Note:__ The amount of resources for \"localhost\" is arbitrary, and it does not reflect the number of physical resources available.\n",
    "\n",
    "</div>\n",
    "\n",
    "<div class=\"alert alert-info\">\n",
    "\n",
    "__Note:__ Before running multiple pilots on a single HPC platform, ensure that its scheduling policy conforms the use case, since each pilot represents a batch job and HPC's scheduling policy might limit the number of jobs running simultaneously per user. For the remote submission see the tutorial [Using RADICAL-Pilot on HPC Platforms](submission.ipynb).\n",
    "\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pilots = []\n",
    "\n",
    "# submission of a single pilot\n",
    "pd0 = rp.PilotDescription({\n",
    "    'resource': 'local.localhost',\n",
    "    'cores'   : 4,\n",
    "    'runtime' : 10\n",
    "})\n",
    "pilots.append(pmgr.submit_pilots(pd0))\n",
    "\n",
    "# submission of multiple pilots at the same time\n",
    "pd1 = rp.PilotDescription({\n",
    "    'resource': 'local.localhost',\n",
    "    'cores'   : 2,\n",
    "    'runtime' : 7\n",
    "})\n",
    "pd2 = rp.PilotDescription({\n",
    "    'resource': 'local.localhost',\n",
    "    'cores'   : 3,\n",
    "    'runtime' : 7\n",
    "})\n",
    "pilots.extend(pmgr.submit_pilots([pd1, pd2]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After pilots are submitted, they should be added to [rp.TaskManager](../apidoc.rst#radical.pilot.TaskManager) for the task dispatching."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tmgr.add_pilots(pilots)\n",
    "tmgr.list_pilots()  # lists all pilots available for TaskManager"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "N_TASKS = 4\n",
    "tds     = []  # list of task descriptions\n",
    "\n",
    "for _ in range(N_TASKS):\n",
    "\n",
    "    td = rp.TaskDescription({\n",
    "        'executable': '/bin/echo',\n",
    "        'arguments' : ['pilot_id=$RP_PILOT_ID']\n",
    "    })\n",
    "\n",
    "    tds.append(td)\n",
    "\n",
    "tasks = tmgr.submit_tasks(tds)\n",
    "tmgr.wait_tasks()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Each task was assigned to a corresponding pilot according to a default scheduling algorithm `rp.SCHEDULER_ROUND_ROBIN`, but it is possible to assign a task to a particular pilot explicitly with the [rp.TaskDescription.pilot](../apidoc.rst#radical.pilot.TaskDescription.pilot) attribute. Thus, in the example below, we submit another pilot and assign a new task to it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pd3 = rp.PilotDescription({\n",
    "    'resource': 'local.localhost',\n",
    "    'cores'   : 2,\n",
    "    'runtime' : 5\n",
    "})\n",
    "\n",
    "pilot = pmgr.submit_pilots(pd3)\n",
    "tmgr.add_pilots(pilot)\n",
    "\n",
    "td = rp.TaskDescription()\n",
    "td.executable = '/bin/echo'\n",
    "td.arguments  = ['task is assigned to $RP_PILOT_ID']\n",
    "td.pilot      = pilot.uid\n",
    "\n",
    "task = tmgr.submit_tasks(td)\n",
    "tmgr.wait_tasks(uids=task.uid)\n",
    "\n",
    "tasks.append(task)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If the task is assigned to a pilot, which wasn't added to [rp.TaskManager](../apidoc.rst#radical.pilot.TaskManager), then that pilot is unknown for the task dispatching, and the task is pending for a corresponding pilot to arrive. In the example below, we set a timeout for the [rp.TaskManager.wait_tasks()](../apidoc.rst#radical.pilot.TaskManager.wait_tasks) method, since by default, that method waits for tasks reaching their final state, but this task will have the state `rp.TMGR_SCHEDULING` until the session will be closed (since we will not submit an associated pilot)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "td = rp.TaskDescription()\n",
    "\n",
    "td.executable = '/bin/echo'\n",
    "td.arguments  = ['pilot_id=$RP_PILOT_ID']\n",
    "td.pilot      = 'unknown_pilot_id'\n",
    "\n",
    "task = tmgr.submit_tasks(td)\n",
    "tmgr.wait_tasks(uids=task.uid, timeout=15)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As a final step, we go through all tasks outputs to see which pilot was assigned to each task."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for task in tasks:\n",
    "    stdout = task.stdout.strip()[:35]\n",
    "    print('%s - output: %-30s (in task description: pilot=\"%s\")' %\n",
    "          (task.uid, stdout, task.description['pilot']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "session.close()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.9.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}