{
"cells": [
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"# Application Level Scheduling\n",
"\n",
"RADICAL-Pilot (RP) by default uses its internal scheduler to efficiently place tasks on the available cluster resources. Some use cases however require more fine-grained and/or explicit control over task placement. RP supports that functionality with *__application level scheduling__*. In that case the pilot will report about the available nodes and resources, but will leave it to the application to assign resources to the tasks. A number of API functions are provided to simplify the development of such application level schedulers, and this tutorial will demonstrate their use.\n",
"\n",
"\n",
"\n",
"__Note:__ At the moment it is not possible to mix application level scheduling and RP's internal scheduling in the same session.\n",
"__Note:__ Application level scheduling is only supported for executable tasks, RAPTOR tasks will ignore any related settings.\n",
"__Note:__ The outputs of the various cells above may differ, depending on your localhost's hardware configuration.\n",
"\n",
"
\n",
"\n",
"\n",
"## Notation\n",
"\n",
"The following terms will be used throughout this tutorial:\n",
"\n",
" - *__task:__* an executable piece of work comprised of one or more processes, all running the same executable on a dedicated set of resources.\n",
" - *__rank:__* one of the processes which comprise a running task. The term _rank_ is frequently used for MPI applications, but we use it generically for any task which uses multiprocessing.\n",
" - *__slot:__* the set of resources which are assigned to a single _rank_, i.e., to a single task process. Note that each _rank_ can utilize multiple cores and/or GPUs, usually by support of libraries and frameworks such as OpenMP, CUDA, OpenCL etc.\n",
" - *__occupation:__* the portion of a resource assigned to a _rank_. For example, two ranks could share a GPU, and then each of the ranks would get `occupation=0.5` assigned for that GPU.\n"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"## Setup\n",
"\n",
"We will first start a normal RP session and submit a pilot, and wait until that pilot becomes active (~15 seconds)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import time\n",
"import copy\n",
"import pprint\n",
"\n",
"import radical.pilot as rp\n",
"import radical.utils as ru"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"vscode": {
"languageId": "plaintext"
}
},
"outputs": [],
"source": [
"session = rp.Session()\n",
"\n",
"pmgr = rp.PilotManager(session)\n",
"tmgr = rp.TaskManager(session)\n",
"\n",
"pilot = pmgr.submit_pilots(rp.PilotDescription(\n",
" {'resource': 'local.localhost',\n",
" 'runtime' : 60,\n",
" 'nodes' : 4}))\n",
"\n",
"tmgr.add_pilots(pilot)\n",
"pilot.wait([rp.PMGR_ACTIVE, rp.FAILED])\n",
"\n",
"assert pilot.state == rp.PMGR_ACTIVE\n",
"\n",
"print('pilot is active (%s)' % pilot.state)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Pilot Resources\n",
"\n",
"We then inspect the pilot's resources by retrieving its nodelist. The nodelist (type: `rp.NodeList`) has the following attributes:\n",
"\n",
" - `uniform`: boolean, indicates if the nodes have identical resources\n",
" - `cores_per_node`, `gpus_per_node`, `mem_per_node`, `lfs_per_node`: amount of resources per node. Those attributes will be `None` for non-uniform nodelists.\n",
" - `nodes`: the actual list of nodes.\n",
"\n",
"Let's inspect one of the nodes (`nodeslist.nodes[0]`). A node in the nodelist has the type `rp.NodeResource` with the following attributes:\n",
"\n",
" - `index`: unique node identifier used within RP\n",
" - `name`: hostname (does not need to be unique!)\n",
" - `mem`: available memory (in MB)\n",
" - `lfs`: available disk storage (in MB)\n",
" - `cores`: available CPU cores and their occupation\n",
" - `gpus`: available GPUs and their occupation.\n",
"\n",
"The core and gpu information are constructed of an integer (the resource index) and a float (the resource occupation where `0.0` is *not used* and `1.0` is *fully used*)."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# inspect the nodelist\n",
"nodelist = pilot.nodelist\n",
"print('#nodes : ', len(nodelist.nodes))\n",
"print('uniform: ', nodelist.uniform)\n",
"print('cpn : ', nodelist.cores_per_node)\n",
"print('gpn : ', nodelist.gpus_per_node)\n",
"\n",
"# inspect one node\n",
"node = nodelist.nodes[0]\n",
"pprint.pprint(node.as_dict())"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Defining Slots for Task Ranks\n",
"\n",
"Based on that information we can now define _slots_, i.e., sets of resources for task ranks to run on. The slot for the simplest possible task (in RP) would just allocate one core. Let's create two slots (core ID 3 and 4) for a two-ranked task which runs `radical-pilot-hello.sh` - that is a script which will report the resources used by the task:\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"slot_1 = rp.Slot(node_index=0, node_name='localhost', cores=[3])\n",
"slot_2 = rp.Slot(node_index=0, node_name='localhost', cores=[4])\n",
"\n",
"td_1 = rp.TaskDescription({'executable': 'radical-pilot-hello.sh', \n",
" 'arguments' : [1],\n",
" 'ranks' : 2,\n",
" 'slots' : [slot_1, slot_2]}) # <--- this is new\n",
"\n",
"task_1 = tmgr.submit_tasks(td_1)\n",
"tmgr.wait_tasks(uids=task_1.uid)\n",
"\n",
"pprint.pprint(task_1.slots)\n",
"print(task_1.stdout)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"We can check the task's rankfile to see if the slot settings were respected:\n",
"\n",
" \n",
"__Note:__ a rankfile will only be available when `mpiexec` is configured as launch method and `-rf` is a supported option on your system.\n",
"
"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"sbox = ru.Url(task_1.sandbox).path\n",
"tid = task_1.uid\n",
"!cat {sbox}/{tid}.rf"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Scheduling Helpers\n",
"\n",
"A simpler way to obtain task slots is to let the node's `NodeResource.find_slot(rp.RankRequirements)` method find it for you. That will return a viable slot, or `None` if the requested resources are not available at this moment."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"rr = rp.RankRequirements(n_cores=1)\n",
"slot_3 = node.find_slot(rr)\n",
"\n",
"assert slot_3\n",
"pprint.pprint(slot_3.as_dict())"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"If we now check the node, we will see that the resource occupation of the first core changed."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"pprint.pprint(node.as_dict())"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"We can also _allocate_ the slots we manually created before so that later calls to `find_slot` will take that information into account (after use, one should _deallocate_ the slots again!)."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"node.allocate_slot(slot_1)\n",
"node.allocate_slot(slot_2)\n",
"\n",
"pprint.pprint(node.as_dict())"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# deallocate all used slots\n",
"node.deallocate_slot(slot_1)\n",
"node.deallocate_slot(slot_2)\n",
"node.deallocate_slot(slot_3)\n",
"\n",
"pprint.pprint(node.as_dict())"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"To allocate a larger set of slots, for example for multi-rank tasks, RP can search the node list itself for available resources. That search might return slots which are distributed across all nodes. For example, the call below will allocate the resources for 4 ranks where each rank uses 4 cores and half a GPU (2 ranks can share one GPU). As the GPU is the limiting resource in this scenario, we will be able to place at most 2 ranks per node:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"rr = rp.RankRequirements(n_cores=4, n_gpus=1, gpu_occupation=0.5)\n",
"slots = nodelist.find_slots(rr, n_slots=4)\n",
"\n",
"for slot in slots:\n",
" print('index:', slot.node_index, 'cores:', slot.cores, 'gpus:', slot.gpus)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"nodelist.release_slots(slots)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Application Level Scheduling\n",
"\n",
"With the above tools, a simple implementation of an application level scheduler would be:\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# ------------------------------------------------------------------------------\n",
"#\n",
"class AppScheduler(object):\n",
"\n",
" def __init__(self, tmgr, nodelist):\n",
" '''\n",
" tmgr : task manager which handles execution of tasks\n",
" nodelist: resources to be used for task execution\n",
" '''\n",
" self._tmgr = tmgr\n",
" self._nodelist = nodelist\n",
" self._running_tasks = dict()\n",
" self._completed_tasks = list()\n",
" self._slots = dict()\n",
"\n",
" self._tmgr.register_callback(self._state_cb)\n",
"\n",
" def _state_cb(self, task, state):\n",
" '''handle task state transitions, specifically for final states'''\n",
" if state in rp.FINAL and task.uid in self._running_tasks:\n",
" print('---> %s: %s' % (task.uid, state))\n",
" assert state == rp.DONE\n",
" self._nodelist.release_slots(self._slots[task.uid])\n",
" del self._running_tasks[task.uid]\n",
" self._completed_tasks.append(task)\n",
"\n",
" def wait_tasks(self, uids=None, timeout=None):\n",
" '''wait for a set of (or all) tasks for a certain time (or forever)'''\n",
" if not uids:\n",
" uids = list(self._running_tasks.keys())\n",
" print('waiting : %s' % uids)\n",
" states = self._tmgr.wait_tasks(uids=uids, timeout=timeout)\n",
" return states\n",
"\n",
" def submit(self, tds):\n",
" '''tds: list of rp.TaskDescriptions - list of tasks to run'''\n",
" while tds:\n",
"\n",
" print('===========================================================')\n",
"\n",
" # find slots for all task descriptions\n",
" allocated = list()\n",
" not_allocated = list()\n",
"\n",
" for td in tds:\n",
" rr = rp.RankRequirements(n_cores=td.cores_per_rank,\n",
" n_gpus=td.gpus_per_rank,\n",
" mem=td.mem_per_rank,\n",
" lfs=td.lfs_per_rank)\n",
" slots = nodelist.find_slots(rr, n_slots=td.ranks)\n",
" if slots:\n",
" # this task can be submitted\n",
" td.slots = slots\n",
" allocated.append(td)\n",
" self._slots[td.uid] = slots\n",
" else:\n",
" # this task has to be retries later on\n",
" not_allocated.append(td)\n",
"\n",
" # submit all tasks for which resources were found\n",
" submitted = tmgr.submit_tasks(allocated)\n",
" for task in submitted:\n",
" self._running_tasks[task.uid] = task\n",
" self._running_tasks[task.uid] = task\n",
" \n",
" print('submitted: %s' % [task.uid for task in allocated])\n",
" print('pending : %s' % [td.uid for td in not_allocated]) \n",
"\n",
" if not_allocated:\n",
"\n",
" if not self._running_tasks:\n",
" # no tasks are running - the remaining tasks will never be\n",
" # able to run, so we have to give up\n",
" td = not_allocated[0]\n",
" pprint.pprint(td.as_dict())\n",
" for node in self._nodelist.nodes:\n",
" pprint.pprint(node.as_dict())\n",
" raise ValueError('can never allocate %s' % td.uid)\n",
"\n",
" # could not submit all tasks - wait for resources to become available\n",
" # on and attempt to schedule the remaining tasks in the next iteration\n",
" while True:\n",
" states = self.wait_tasks(timeout=5.0)\n",
"\n",
" # if we got any free resources, try to schedule more tasks\n",
" if rp.DONE in states or rp.FAILED in states:\n",
" break\n",
"\n",
" time.sleep(3)\n",
" tds = not_allocated\n",
"\n",
"\n",
"# ------------------------------------------------------------------------------\n",
"#\n",
"tds = list()\n",
"for i in range(16):\n",
" td = {'executable' : 'sleep',\n",
" 'arguments' : [2],\n",
" 'ranks' : i + 1, # submit larger and larger tasks\n",
" 'cores_per_rank' : 1, \n",
" 'uid' : 't.%02d' % i}\n",
" tds.append(rp.TaskDescription(td))\n",
"\n",
"scheduler = AppScheduler(tmgr, nodelist)\n",
"scheduler.submit(tds)\n",
"scheduler.wait_tasks()\n",
"\n",
"print('=== all tasks completed ===')\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Use Case 1: Pilot Partitions\n",
"\n",
"Use the application level scheduler to localize certain tasks on one set of nodes (pilot partition), and other tasks on a different set of nodes:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"nodelist_1 = copy.deepcopy(nodelist)\n",
"nodelist_2 = copy.deepcopy(nodelist)\n",
"\n",
"nodelist_1.nodes = nodelist_1.nodes[:2] # first two nodes\n",
"nodelist_2.nodes = nodelist_1.nodes[2:] # remaining nodes\n",
"\n",
"scheduler_1 = AppScheduler(tmgr, nodelist_1)\n",
"scheduler_2 = AppScheduler(tmgr, nodelist_2)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Use Case 2: Virtual Nodes\n",
"\n",
"Use the API to splice nodes into virtual nodes, creating virtual, heterogeneous node list (in the future, we might add APIs to simplify node splicing). Consider the node type we have here:\n",
"```\n",
"{'cores': [{'index': 0, 'occupation': 0.0},\n",
" {'index': 1, 'occupation': 0.0},\n",
" {'index': 2, 'occupation': 0.0},\n",
" {'index': 3, 'occupation': 0.0},\n",
" {'index': 4, 'occupation': 0.0},\n",
" {'index': 5, 'occupation': 0.0},\n",
" {'index': 6, 'occupation': 0.0},\n",
" {'index': 7, 'occupation': 0.0},\n",
" {'index': 8, 'occupation': 0.0},\n",
" {'index': 9, 'occupation': 0.0},\n",
" {'index': 10, 'occupation': 0.0},\n",
" {'index': 11, 'occupation': 0.0}],\n",
" 'gpus': [{'index': 0, 'occupation': 0.0}],\n",
" 'index': 0,\n",
" 'lfs': 1000000,\n",
" 'mem': 65536,\n",
" 'name': 'localhost'}\n",
"```\n",
"\n",
"We could splice that into a GPU node with 2 cores and 1 GPU, and a CPU-only node with the remaining 10 cores:\n",
"```\n",
"[{'cores': [{'index': 0, 'occupation': 0.0},\n",
" {'index': 1, 'occupation': 0.0}],\n",
" 'gpus': [{'index': 0, 'occupation': 0.0}],\n",
" 'index': 0,\n",
" 'lfs': 1000000,\n",
" 'mem': 65536,\n",
" 'name': 'localhost'},\n",
"\n",
" {'cores': [{'index': 2, 'occupation': 0.0},\n",
" {'index': 3, 'occupation': 0.0},\n",
" {'index': 4, 'occupation': 0.0},\n",
" {'index': 5, 'occupation': 0.0},\n",
" {'index': 6, 'occupation': 0.0},\n",
" {'index': 7, 'occupation': 0.0},\n",
" {'index': 8, 'occupation': 0.0},\n",
" {'index': 9, 'occupation': 0.0},\n",
" {'index': 10, 'occupation': 0.0},\n",
" {'index': 11, 'occupation': 0.0}],\n",
" 'index': 1,\n",
" 'lfs': 1000000,\n",
" 'mem': 65536,\n",
" 'name': 'localhost'}]\n",
"```\n",
"\n",
"When splicing all nodes of a cluster allocation, we end up with two virtual allocations, one GPU focused and one CPU focused, and certain task scheduling policies with GPU awareness become almost trivial. One could also, for example, splice nodes at their NUMA domains to trivially get a NUMA-aware scheduler (some care needs to be applied to `lfs` and `mem` resource slicing)."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Use Case 3: NUMA Domains\n",
"\n",
"Similarly, we can splice nodes into individual NUMA-Domains, thus implementing a simple form of NUMA-aware scheduling. As that is a common use case, RP provides a `NumaNode` class to simplify that use case:\n",
"\n",
"\n",
"Assume we have the following node layout with 8 cores and 2 GPUs:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"node = rp.Node({'name' : 'localhost',\n",
" 'cores' : [rp.RO(index=x, occupation=rp.FREE) for x in range(8)],\n",
" 'gpus' : [rp.RO(index=x, occupation=rp.FREE) for x in range(2)]})\n",
"\n",
"pprint.pprint(node.as_dict())"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": "Assume further, that the node is divided into two NUMA domains. That could be expressed as follows:"
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"ndm = rp.NumaDomainMap({0: rp.NumaDomain(cores=range( 0, 4), gpus=[0]),\n",
" 1: rp.NumaDomain(cores=range( 4, 8), gpus=[1])})\n",
"\n",
"numa_node = rp.NumaNode(node, ndm)\n",
"pprint.pprint(numa_node.as_dict())"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"We can now schedule NUMA-agnostic *and* NUMA-aware ranks on that node. Note that the second slot will use the GPU on the *second* NUMA domain as the first NUMA domain does not have sufficient cores left."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"slot_1 = node.find_slot(rp.RankRequirements(n_cores=3))\n",
"slot_2 = numa_node.find_slot(rp.RankRequirements(n_cores=2, n_gpus=1, numa=True))\n",
"print(slot_1)\n",
"print(slot_2)\n",
"\n",
"pprint.pprint(numa_node.as_dict())"
]
}
],
"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.12.7"
}
},
"nbformat": 4,
"nbformat_minor": 4
}