ex_01_remo_func.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A Guided Tour of Ray Core: Remote Functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[*Remote Functions*](https://docs.ray.io/en/latest/walkthrough.html#remote-functions-tasks)\n",
    "involve using a `@ray.remote` decorator on a function. \n",
    "\n",
    "This implements a [*task parallelism*](https://patterns.eecs.berkeley.edu/?page_id=208) pattern, with properties: *data independence*, *stateless*\n",
    "\n",
    "But first the concept of tasks and ownership.\n",
    "\n",
    "Most of the system metadata is managed according to a decentralized concept called ownership: Each worker process manages and owns the tasks that it submits and the `ObjectRef`s returned by those tasks. The owner is responsible for ensuring execution of the task and facilitating the resolution of an `ObjectRef` to its underlying value. Similarly, a worker owns any objects that it created through a `ray.put` call.\n",
    "\n",
    "<img src=\"images/task_ownership.png\" height=350, width=650>\n",
    "\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true,
    "tags": []
   },
   "source": [
    "### Lifetime of a Ray Task\n",
    "\n",
    "The owner is responsible for ensuring execution of a submitted task and facilitating the resolution of the returned `ObjectRef` to its underlying value.\n",
    "The process that submits a task is considered to be the owner of the result and is responsible for acquiring resources from the raylet to execute the task. Here, the driver owns the result of `A`, and `Worker 1` owns the result of `B`.\n",
    "\n",
    "<img src=\"images/task_life.png\" height=350, width=650>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Launch a Ray locally, a single head node"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, let's start Ray… \n",
    "\n",
    "This will start Ray on the local host, with headnode and workers for each core or CPU available.\n",
    "You can check the resources being used."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'CPU': 12.0,\n",
       " 'memory': 13590429696.0,\n",
       " 'node:127.0.0.1': 1.0,\n",
       " 'object_store_memory': 6795214848.0}"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import logging\n",
    "import ray\n",
    "\n",
    "ray.init(\n",
    "    ignore_reinit_error=True,              # Don't print error messages if a Ray instance is already running. Attach to it\n",
    "    logging_level=logging.ERROR,           \n",
    ")\n",
    "ray.cluster_resources()                    # get the cluster resources"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ray Dashboard accessible at URI: [http://127.0.0.1:8265](http://127.0.0.1:8265)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Remote Functions example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following is just a regular Python function..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "def my_function ():\n",
    "    return 42"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When called, it simply returns an integer:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "42"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "my_function()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you were to iterate through a sequence of calls to a function such as that, these calls would be performed *sequentially*."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "However, by adding the `@ray.remote` decorator, a regular Python function becomes a Ray remote function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def my_function ():\n",
    "    return 42"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To invoke this remote function, use the `remote` method. This will immediately return an object ref (a *future* in Python) and then create a task that will be executed on a worker process."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.42 ms, sys: 1.11 ms, total: 2.53 ms\n",
      "Wall time: 1.79 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "ObjectRef(bbde8638d39a1245ffffffffffffffffffffffff0100000001000000)"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "obj_ref = my_function.remote()\n",
    "obj_ref"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The result can be retrieved with `ray.get`, which is a blocking call if the task is still not finished"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 205 µs, sys: 99 µs, total: 304 µs\n",
      "Wall time: 286 µs\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "42"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "ray.get(obj_ref)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Invocations of Ray *remote functions* happen in parallel, and all computation gets performed in the background, driven by Ray's internal event loop. Remote calls return immediately, with a Python Future Object reference.\n",
    "\n",
    "To illustrate this, first let's define a relatively \"slow\" function, by adding a 10 second delay..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "import random\n",
    "\n",
    "@ray.remote\n",
    "def slow_function ():\n",
    "  time.sleep(10)\n",
    "  return random.randint(0, 9)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we'll iterate through multiple calls, showing that this does not block:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0\n",
      "1\n",
      "2\n",
      "3\n",
      "[ObjectRef(44ed5e1383be6308ffffffffffffffffffffffff0100000001000000), ObjectRef(d56d800cbde6ca14ffffffffffffffffffffffff0100000001000000), ObjectRef(38db0ab51c6b6cfbffffffffffffffffffffffff0100000001000000), ObjectRef(4e2ab276f14c37c2ffffffffffffffffffffffff0100000001000000)]\n",
      "CPU times: user 2.31 ms, sys: 1.84 ms, total: 4.15 ms\n",
      "Wall time: 2.31 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "futures_list = []\n",
    "\n",
    "for i in range(4):\n",
    "    future = slow_function.remote()\n",
    "    futures_list.append(future)\n",
    "    print(i)\n",
    "print(futures_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5\n",
      "9\n",
      "9\n",
      "5\n",
      "CPU times: user 288 ms, sys: 184 ms, total: 472 ms\n",
      "Wall time: 8.77 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "for future in futures_list:\n",
    "    print(ray.get(future))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "futures_list = []\n",
    "\n",
    "for i in range(4):\n",
    "    future = slow_function.remote()\n",
    "    futures_list.append(future)\n",
    "print(futures_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "for future in futures_list:\n",
    "    print(ray.get(future))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note the difference between CPU times and wall clock?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Another way to do this is using list comprehension"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.14 ms, sys: 524 µs, total: 1.66 ms\n",
      "Wall time: 838 µs\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[ObjectRef(d5a75db31f99bd73ffffffffffffffffffffffff0100000001000000),\n",
       " ObjectRef(d68fec326c8433c9ffffffffffffffffffffffff0100000001000000),\n",
       " ObjectRef(8e088f779f48acd6ffffffffffffffffffffffff0100000001000000),\n",
       " ObjectRef(65a1a3aaa614cba3ffffffffffffffffffffffff0100000001000000)]"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "futures_list = [slow_function.remote() for _ in range(4)]\n",
    "futures_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[7, 6, 2, 1]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "values = [ray.get(future) for future in futures_list]\n",
    "values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "futures_list = [slow_function.remote() for _ in range(4)]\n",
    "futures_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "values_list = [ray.get(future) for future in futures_list]\n",
    "values_list"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note the difference between CPU times and wall clock in comprehensions? \n",
    "Comprehensions seems faster."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, shutdown Ray"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.shutdown()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## References"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[*Patterns for Parallel Programming*](https://www.goodreads.com/book/show/85053.Patterns_for_Parallel_Programming)  \n",
    "Timothy G. Mattson, Beverly A. Sanders, Berna L. Massingill  \n",
    "Addison-Wesley (2004)\n",
    "\n",
    "[Ray Core Walkthrough](https://docs.ray.io/en/latest/walkthrough.html)\n",
    "Ray Documentation and Gettting started materials\n",
    "\n",
    "[Ray Architecture Reference](https://docs.google.com/document/d/1lAy0Owi-vPz2jEqBSaHNQcy2IBSDEHyXNOQZlGuj93c/preview#)\n",
    "Ray 1.x Architecture Technical Paper\n",
    "\n",
    "[Ray Internals: A peek at ray,get](https://www.youtube.com/watch?v=a1kNnQu6vGw)\n",
    "\n",
    "[Ray Internals: Object management with Ownership Model](https://www.anyscale.com/events/2021/06/22/ray-internals-object-management-with-the-ownership-model)"
   ]
  }
 ],
 "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.8.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}