ex_06_ray_api_calls.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Exploring Ray API Calls\n",
    "\n",
    "© 2019-2022, Anyscale. All Rights Reserved\n",
    "\n",
    "📖 [Back to Table of Contents](./ex_00_tutorial_overview.ipynb)<br>\n",
    "➡ [Next notebook](./ex_07_ray_data.ipynb) <br>\n",
    "⬅️ [Previous notebook](./ex_05_multiprocess_pool.ipynb) <br>\n",
    "\n",
    "### Overview\n",
    "Ray offers a rich and wide set of APIs across all its components. Discussing or covering all of them is out of the scope of this notebook (and tutorial), albeit we will touch upon some common APIs, its use, and arguments. A more exhaustive references below in the documentation provides a full set.\n",
    "\n",
    " * [Ray Core](https://docs.ray.io/en/latest/ray-core/package-ref.html)\n",
    " * [Ray AIR](https://docs.ray.io/en/latest/ray-air/package-ref.html)\n",
    " * [Ray Data](https://docs.ray.io/en/latest/data/dataset.html)\n",
    " * [Ray Train](https://docs.ray.io/en/latest/train/train.html)\n",
    " * [Ray Tune](https://docs.ray.io/en/latest/tune/index.html)\n",
    " * [Ray Serve](https://docs.ray.io/en/latest/serve/index.html)\n",
    " * [RLlib](https://docs.ray.io/en/latest/rllib/index.html)\n",
    "\n",
    "\n",
    "### Learning objectives\n",
    "In this quick tour of the API, you will learn about:\n",
    "\n",
    " * Common Ray Core APIs\n",
    " * Some useful arguments to these APIs \n",
    " * Tips and Tricks for first-time users\n",
    "\n",
    "\n",
    "This lesson explores a few of the other API calls you might find useful, as well as options that can be used with the API calls we've already used in the previous lessons. Additionally, we will walk through some tips and tricks for first time users.\n",
    "\n",
    "> **Tip:** The [Ray Package Reference](https://docs.ray.io/en/latest/package-ref.html) in the [Ray Docs](https://docs.ray.io/en/latest/) is useful for exploring the API features we'll learn."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time \n",
    "import sys\n",
    "import logging \n",
    "import random\n",
    "import math\n",
    "import json\n",
    "import requests\n",
    "import numpy as np\n",
    "from typing import List, Tuple\n",
    "\n",
    "import ray"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ray.init()\n",
    "\n",
    "When we used [`ray.init()`](https://ray.readthedocs.io/en/latest/package-ref.html#ray.init), we used it to start Ray on our local machine. When the optional `address=...` argument is specified, the driver connects to the corresponding Ray cluster.\n",
    "\n",
    "There are a lot of optional keyword arguments you can pass to `ray.init()`. Here are some of them. All options are described in the [documentation](https://ray.readthedocs.io/en/latest/package-ref.html#ray.init). \n",
    "\n",
    "| Name | Type | Example | Description |\n",
    "| :--- | :--- | :------ | :---------- |\n",
    "| `address` | `str` | `address='auto'` | The address of the Ray cluster to connect to. If this address is not provided, then this command will start a raylet, a plasma store, a plasma manager, and some workers. It will also kill these processes when Python exits. If the driver is running on a node in a Ray cluster, using `auto` as the value tells the driver to detect the the cluster, removing the need to specify a specific node address. |\n",
    "| `num_cpus` | `int` | `num_cpus=4` | Number of CPUs the user wishes to assign to each _raylet_. |\n",
    "| `num_gpus` | `int` | `num_gpus=1` | Number of GPUs the user wishes to assign to each _raylet_. |\n",
    "| `resources` | `dictionary` | `resources={'resource1': 4, 'resource2': 16}` | Maps the names of custom resources to the quantities of those resources available. |\n",
    "| `memory` | `int` | `memory=1000000000` | The amount of memory (in bytes) that is available for use by workers requesting memory resources. By default, this is automatically set based on the available system memory. |\n",
    "| `object_store_memory` | `int` | `object_store_memory=1000000000` | The amount of memory (in bytes) for the object store. By default, this is automatically set based on available system memory, subject to a 20GB cap. |\n",
    "| `log_to_driver` | `bool` | `log_to_driver=True` | If true, then the output from all of the worker processes on all nodes will be directed to the driver program. |\n",
    "| `local_mode` | `bool` | `local_mode=True` | If true, the code will be executed serially. This is useful for debugging. |\n",
    "| `ignore_reinit_error` | `bool` | `ignore_reinit_error=True` | If true, Ray suppresses errors from calling `ray.init()` a second time (as we've done in these notebooks). Ray won't be restarted. |\n",
    "| `configure_logging` | `bool` | `configure_logging=True` | If true (default), configuration of logging is allowed here. Otherwise, the user may want to configure it separately. |\n",
    "| `logging_level` | _Flag_ | `logging_level=logging.INFO` | The logging level, defaults to `logging.INFO`. Ignored unless \"configure_logging\" is true. |\n",
    "| `logging_format` | `str` | `logging_format='...'` | The logging format to use, defaults to a string containing a timestamp, filename, line number, and message. See the Ray source code `ray_constants.py` for details. Ignored unless \"configure_logging\" is true. |\n",
    "| `runtime_env` | `map` | `{\"working_dir\": \"/path/to/files\"}` | Your Ray application might depend on source files or data files. For a development workflow, these might live on your local machine, but when it comes time to run things at scale, you will need to get them to your remote cluster. A way to send these files across all nodes in the cluster so that your distributed tasks or actors can access them, use this option, [for example.](https://docs.ray.io/en/latest/ray-core/handling-dependencies.html#runtime-environments) |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's define a global runtime enviornment accessible or available across all nodes by all Ray tasks and actors.\n",
    " * we want our working directory with some data files available across these tasks and actors. There is a limit of 2GB upload\n",
    " * we additional packages available too; they will be installed automatically upon start or launch of the cluster\n",
    " \n",
    "A runtime environment describes the dependencies your Ray application needs to run, including files, packages, environment variables, and more. It is installed dynamically on the cluster at runtime and cached for future use. A short [blog](https://medium.com/@2twitme/day-10-how-to-use-ray-runtime-environment-dependencies-5d6266c48c9) describing how to use Runtime environments. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "ENV_VARS = {\"BUCKET_LOC\": \"s3://models\",\n",
    "            \"MODEL_NAME\": \"lr_model.pkl\"\n",
    "           }\n",
    "MY_RUNTIME_ENV = {\"working_dir\": \"data\", \n",
    "                   \"env_vars\": ENV_VARS, \n",
    "                   \"pip\": [\"requests\", \"pendulum==2.1.2\"]\n",
    "                 }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "    <div style=\"margin-left: 50px;display: flex;flex-direction: row;align-items: center\">\n",
       "        <h3 style=\"color: var(--jp-ui-font-color0)\">Ray</h3>\n",
       "        <svg version=\"1.1\" id=\"ray\" width=\"3em\" viewBox=\"0 0 144.5 144.6\" style=\"margin-left: 3em;margin-right: 3em\">\n",
       "            <g id=\"layer-1\">\n",
       "                <path fill=\"#00a2e9\" class=\"st0\" d=\"M97.3,77.2c-3.8-1.1-6.2,0.9-8.3,5.1c-3.5,6.8-9.9,9.9-17.4,9.6S58,88.1,54.8,81.2c-1.4-3-3-4-6.3-4.1\n",
       "                    c-5.6-0.1-9.9,0.1-13.1,6.4c-3.8,7.6-13.6,10.2-21.8,7.6C5.2,88.4-0.4,80.5,0,71.7c0.1-8.4,5.7-15.8,13.8-18.2\n",
       "                    c8.4-2.6,17.5,0.7,22.3,8c1.3,1.9,1.3,5.2,3.6,5.6c3.9,0.6,8,0.2,12,0.2c1.8,0,1.9-1.6,2.4-2.8c3.5-7.8,9.7-11.8,18-11.9\n",
       "                    c8.2-0.1,14.4,3.9,17.8,11.4c1.3,2.8,2.9,3.6,5.7,3.3c1-0.1,2,0.1,3,0c2.8-0.5,6.4,1.7,8.1-2.7s-2.3-5.5-4.1-7.5\n",
       "                    c-5.1-5.7-10.9-10.8-16.1-16.3C84,38,81.9,37.1,78,38.3C66.7,42,56.2,35.7,53,24.1C50.3,14,57.3,2.8,67.7,0.5\n",
       "                    C78.4-2,89,4.7,91.5,15.3c0.1,0.3,0.1,0.5,0.2,0.8c0.7,3.4,0.7,6.9-0.8,9.8c-1.7,3.2-0.8,5,1.5,7.2c6.7,6.5,13.3,13,19.8,19.7\n",
       "                    c1.8,1.8,3,2.1,5.5,1.2c9.1-3.4,17.9-0.6,23.4,7c4.8,6.9,4.6,16.1-0.4,22.9c-5.4,7.2-14.2,9.9-23.1,6.5c-2.3-0.9-3.5-0.6-5.1,1.1\n",
       "                    c-6.7,6.9-13.6,13.7-20.5,20.4c-1.8,1.8-2.5,3.2-1.4,5.9c3.5,8.7,0.3,18.6-7.7,23.6c-7.9,5-18.2,3.8-24.8-2.9\n",
       "                    c-6.4-6.4-7.4-16.2-2.5-24.3c4.9-7.8,14.5-11,23.1-7.8c3,1.1,4.7,0.5,6.9-1.7C91.7,98.4,98,92.3,104.2,86c1.6-1.6,4.1-2.7,2.6-6.2\n",
       "                    c-1.4-3.3-3.8-2.5-6.2-2.6C99.8,77.2,98.9,77.2,97.3,77.2z M72.1,29.7c5.5,0.1,9.9-4.3,10-9.8c0-0.1,0-0.2,0-0.3\n",
       "                    C81.8,14,77,9.8,71.5,10.2c-5,0.3-9,4.2-9.3,9.2c-0.2,5.5,4,10.1,9.5,10.3C71.8,29.7,72,29.7,72.1,29.7z M72.3,62.3\n",
       "                    c-5.4-0.1-9.9,4.2-10.1,9.7c0,0.2,0,0.3,0,0.5c0.2,5.4,4.5,9.7,9.9,10c5.1,0.1,9.9-4.7,10.1-9.8c0.2-5.5-4-10-9.5-10.3\n",
       "                    C72.6,62.3,72.4,62.3,72.3,62.3z M115,72.5c0.1,5.4,4.5,9.7,9.8,9.9c5.6-0.2,10-4.8,10-10.4c-0.2-5.4-4.6-9.7-10-9.7\n",
       "                    c-5.3-0.1-9.8,4.2-9.9,9.5C115,72.1,115,72.3,115,72.5z M19.5,62.3c-5.4,0.1-9.8,4.4-10,9.8c-0.1,5.1,5.2,10.4,10.2,10.3\n",
       "                    c5.6-0.2,10-4.9,9.8-10.5c-0.1-5.4-4.5-9.7-9.9-9.6C19.6,62.3,19.5,62.3,19.5,62.3z M71.8,134.6c5.9,0.2,10.3-3.9,10.4-9.6\n",
       "                    c0.5-5.5-3.6-10.4-9.1-10.8c-5.5-0.5-10.4,3.6-10.8,9.1c0,0.5,0,0.9,0,1.4c-0.2,5.3,4,9.8,9.3,10\n",
       "                    C71.6,134.6,71.7,134.6,71.8,134.6z\"/>\n",
       "            </g>\n",
       "        </svg>\n",
       "        <table>\n",
       "            <tr>\n",
       "                <td style=\"text-align: left\"><b>Python version:</b></td>\n",
       "                <td style=\"text-align: left\"><b>3.8.13</b></td>\n",
       "            </tr>\n",
       "            <tr>\n",
       "                <td style=\"text-align: left\"><b>Ray version:</b></td>\n",
       "                <td style=\"text-align: left\"><b> 2.2.0</b></td>\n",
       "            </tr>\n",
       "            <tr>\n",
       "    <td style=\"text-align: left\"><b>Dashboard:</b></td>\n",
       "    <td style=\"text-align: left\"><b><a href=\"http://127.0.0.1:8265\" target=\"_blank\">http://127.0.0.1:8265</a></b></td>\n",
       "</tr>\n",
       "\n",
       "        </table>\n",
       "    </div>\n",
       "</div>\n"
      ],
      "text/plain": [
       "RayContext(dashboard_url='127.0.0.1:8265', python_version='3.8.13', ray_version='2.2.0', ray_commit='b6af0887ee5f2e460202133791ad941a41f15beb', address_info={'node_ip_address': '127.0.0.1', 'raylet_ip_address': '127.0.0.1', 'redis_address': None, 'object_store_address': '/tmp/ray/session_2023-01-01_08-43-09_914236_61566/sockets/plasma_store', 'raylet_socket_name': '/tmp/ray/session_2023-01-01_08-43-09_914236_61566/sockets/raylet', 'webui_url': '127.0.0.1:8265', 'session_dir': '/tmp/ray/session_2023-01-01_08-43-09_914236_61566', 'metrics_export_port': 61622, 'gcs_address': '127.0.0.1:54042', 'address': '127.0.0.1:54042', 'dashboard_agent_listen_port': 52365, 'node_id': 'b5a66792d6abaa014788485a35caaa837573e16d451ea93aef504f8d'})"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "if ray.is_initialized:\n",
    "    ray.shutdown()\n",
    "ray.init(num_cpus=4, runtime_env=MY_RUNTIME_ENV,logging_level=logging.ERROR)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "See also the documentation for [ray.shutdown()](https://ray.readthedocs.io/en/latest/package-ref.html#ray.shutdown), which is needed in some contexts."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fetching Cluster Information\n",
    "\n",
    "Many methods return information:\n",
    "\n",
    "| Method | Brief Description |\n",
    "| :----- | :---------------- |\n",
    "| [`ray.get_gpu_ids()`](https://ray.readthedocs.io/en/latest/package-ref.html#ray.get_gpu_ids) | GPUs |\n",
    "| [`ray.nodes()`](https://ray.readthedocs.io/en/latest/package-ref.html#ray.nodes) | Cluster nodes |\n",
    "| [`ray.cluster_resources()`](https://ray.readthedocs.io/en/latest/package-ref.html#ray.cluster_resources) | All the available resources, used or not |\n",
    "| [`ray.available_resources()`](https://ray.readthedocs.io/en/latest/package-ref.html#ray.available_resources) | Resources not in use |\n",
    "\n",
    "You can see the full list of methods in the [ray.init()](https://docs.ray.io/en/latest/ray-core/package-ref.html#python-api) API documention."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "ray.get_gpu_ids():          []\n",
      "ray.nodes():                [{'NodeID': 'b5a66792d6abaa014788485a35caaa837573e16d451ea93aef504f8d', 'Alive': True, 'NodeManagerAddress': '127.0.0.1', 'NodeManagerHostname': 'Juless-MacBook-Pro-16', 'NodeManagerPort': 50600, 'ObjectManagerPort': 50599, 'ObjectStoreSocketName': '/tmp/ray/session_2023-01-01_08-43-09_914236_61566/sockets/plasma_store', 'RayletSocketName': '/tmp/ray/session_2023-01-01_08-43-09_914236_61566/sockets/raylet', 'MetricsExportPort': 61622, 'NodeName': '127.0.0.1', 'alive': True, 'Resources': {'node:127.0.0.1': 1.0, 'CPU': 4.0, 'object_store_memory': 2147483648.0, 'memory': 49037236634.0}}]\n",
      "ray.cluster_resources():    {'memory': 49037236634.0, 'object_store_memory': 2147483648.0, 'node:127.0.0.1': 1.0, 'CPU': 4.0}\n",
      "ray.available_resources():  {'memory': 49037236634.0, 'CPU': 4.0, 'object_store_memory': 2147483648.0, 'node:127.0.0.1': 1.0}\n",
      "\n"
     ]
    }
   ],
   "source": [
    "print(f\"\"\"\n",
    "ray.get_gpu_ids():          {ray.get_gpu_ids()}\n",
    "ray.nodes():                {ray.nodes()}\n",
    "ray.cluster_resources():    {ray.cluster_resources()}\n",
    "ray.available_resources():  {ray.available_resources()}\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Recall that we used `ray.init(num_cpus=4...)` above to initialize so `ray.nodes()[0]['Resources']['CPU']` returns number of CPU cores on our machines:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4.0"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ray.nodes()[0]['Resources']['CPU']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## @ray.remote()\n",
    "\n",
    "We've used [@ray.remote](https://ray.readthedocs.io/en/latest/package-ref.html#ray.remote) a lot. You can pass arguments when using it. Here are some of them.\n",
    "\n",
    "| Name | Type | Example | Description |\n",
    "| :--- | :--- | :------ | :---------- |\n",
    "| `num_cpus` | `int` | `num_cpus=4` | The number of CPU cores to reserve for this task or for the lifetime of the actor. |\n",
    "| `num_gpus` | `int` | `num_gpus=1` | The number of GPU cores to reserve for this task or for the lifetime of the actor. |\n",
    "| `num_returns` | `int` | `num_returns=2` | (Only for tasks, not actors.) The number of object refs returned by the remote function invocation. |\n",
    "| `runtime_env` | `map` | `runtime_env = {\"working_dir\": \".\", \"pip\": [\"requests\"]}}` | The runtime environment to use for this job (see [Runtime environments](https://docs.ray.io/en/latest/ray-core/handling-dependencies.html#runtime-environments) for details. |\n",
    "| `max_calls` | `int` | `max_calls=5` | Only for *remote tasks*. This specifies the maximum of times that a given worker can execute the given remote function before it must exit (this can be used to address memory leaks in third-party libraries or to reclaim resources that cannot easily be released, e.g., GPU memory that was acquired by TensorFlow). By default this is infinite. |\n",
    "| `max_restarts` | `int` | `max_restarts=-1` | Only for *actors*. This specifies the maximum number of times that the actor should be restarted when it dies unexpectedly. The minimum valid value is 0 (default), which indicates that the actor doesn't need to be restarted. A value of -1 indicates that an actor should be restarted indefinitely. |\n",
    "| `max_task_retries` | `int` | `max_task_retries=-1` | Only for *actors*. How many times to retry an actor task if the task fails due to a system error, e.g., the actor has died. If set to -1, the system will retry the failed task until the task succeeds, or the actor has reached its max_restarts limit. If set to n > 0, the system will retry the failed task up to n times, after which the task will throw a `RayActorError` exception upon `ray.get`. Note that Python exceptions are not considered system errors and will not trigger retries. |\n",
    "| `max_retries` | `int` | `max_retries=-1` | Only for *remote functions*. This specifies the maximum number of times that the remote function should be rerun when the worker process executing it crashes unexpectedly. The minimum valid value is 0, the default is 4 (default), and a value of -1 indicates infinite retries. |\n",
    "\n",
    "Here's an example with and without `num_return_vals`:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## @ray.method()\n",
    "\n",
    "Related to `@ray.remote()`, [@ray.method()](https://ray.readthedocs.io/en/latest/package-ref.html#ray.method) allows you to specify the number of return values for a method in a task or an actor, by passing the `num_returns` keyword argument. None of the other `@ray.remote()` keyword arguments are allowed. Here is an example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(LIONEL MESSIE, 8, 12.100000000000001)\n"
     ]
    }
   ],
   "source": [
    "@ray.remote(num_cpus=1, num_returns=3)\n",
    "def tuple3(id: str, two:int, lst: List[float]) -> Tuple[str, int, float]:\n",
    "    one = id.upper()\n",
    "    two = random.randint(5, 10)\n",
    "    three = sum(lst)\n",
    "    return (one, two, three)\n",
    "\n",
    "# Return three object references with three distinct values in each \n",
    "x_ref, y_ref, z_ref = tuple3.remote(\"lionel messie\", 1, [2.2, 4.4, 5.5])\n",
    "\n",
    "# Fetch the entire list\n",
    "x, y, z = ray.get([x_ref, y_ref, z_ref])\n",
    "print(f'({x}, {y}, {z})')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A slight variation of the above example is pack all values in a single return, and then unpack them."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(LIONEL MESSIE, 5, 12.100000000000001)\n"
     ]
    }
   ],
   "source": [
    "@ray.remote(num_returns=1)\n",
    "def tuple3_packed(id: str, two:int, lst: List[float]) -> Tuple[str, int, float]:\n",
    "    one = id.upper()\n",
    "    two = random.randint(5, 10)\n",
    "    three = sum(lst)\n",
    "    return (one, two, three)\n",
    "\n",
    "# Returns one object references with three values in it\n",
    "xyz_ref = tuple3_packed.remote(\"lionel messie\", 1, [2.2, 4.4, 5.5])\n",
    "\n",
    "# Fetch from a single object ref and unpack into three values\n",
    "x, y, z = ray.get(xyz_ref)\n",
    "print(f'({x}, {y}, {z})')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's do the same for an Ray actor method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(LIONEL MESSIE, 7, 12.100000000000001)\n"
     ]
    }
   ],
   "source": [
    "@ray.remote\n",
    "class TupleActor:\n",
    "    @ray.method(num_returns=3)\n",
    "    def tuple3(self, id: str, two:int, lst: List[float]) -> Tuple[str, int, float]:\n",
    "        one = id.upper()\n",
    "        two = random.randint(5, 10)\n",
    "        three = sum(lst)\n",
    "        return (one, two, three)\n",
    "    \n",
    "# Create an instance of an actor\n",
    "actor = TupleActor.remote()\n",
    "x_ref, y_ref, z_ref = actor.tuple3.remote(\"lionel messie\", 1, [2.2, 4.4, 5.5])\n",
    "x, y, z = ray.get([x_ref, y_ref, z_ref])\n",
    "print(f'({x}, {y}, {z})')   "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Tips and Tricks for first-time users\n",
    "Because Ray's core APIs are simple and flexible, first time users can trip upon certain API calls in Ray's usage patterns. This short tips & tricks will insure you against unexpected results. Below we briefly explore a handful of API calls and their best practices."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Tip 1: Delay ray.get()\n",
    "\n",
    "With Ray, all invocations of `.remote()` calls are asynchronous, meaning the operation returns immediately with a promise/future object Reference ID. This is key to achieving massive parallelism, for it allows a devloper to launch many remote tasks, each returning a remote future object ID. Whenever needed, this object ID is fetched with `ray.get`. Because `ray.get` is a blocking call, where and how often you use can affect the performance of your Ray application. \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def do_some_work(x):\n",
    "    # Assume doing some computation\n",
    "    time.sleep(0.5)\n",
    "    return math.exp(x)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Bad usage\n",
    "We use `ray.get` inside a list comprehension loop, hence it blocks on each call of `.remote()`, delaying until the task is finished and the value\n",
    "is materialized and fetched from the Ray object store."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 45 ms, sys: 21.5 ms, total: 66.5 ms\n",
      "Wall time: 5.09 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[1.0,\n",
       " 2.718281828459045,\n",
       " 7.38905609893065,\n",
       " 20.085536923187668,\n",
       " 54.598150033144236,\n",
       " 148.4131591025766,\n",
       " 403.4287934927351,\n",
       " 1096.6331584284585,\n",
       " 2980.9579870417283,\n",
       " 8103.083927575384]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "results = [ray.get(do_some_work.remote(x)) for x in range(10)]\n",
    "results"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Good usage\n",
    "We delay `ray.get` after all the tasks have been invoked and their references have been returned. That is,\n",
    "we don't block on each call but instead do outside the comprehension loop.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 15 ms, sys: 9.86 ms, total: 24.8 ms\n",
      "Wall time: 2.26 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[1.0,\n",
       " 2.718281828459045,\n",
       " 7.38905609893065,\n",
       " 20.085536923187668,\n",
       " 54.598150033144236,\n",
       " 148.4131591025766,\n",
       " 403.4287934927351,\n",
       " 1096.6331584284585,\n",
       " 2980.9579870417283,\n",
       " 8103.083927575384]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "results = ray.get([do_some_work.remote(x) for x in range(10)])\n",
    "results"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Takeway tip 1: \n",
    "Since `ray.get` is a blocking call, postpone its use only when you need object ID's value. If called eagerly, it can\n",
    "affect the performance of your desired parallelism."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Tip 2: Avoid tiny remote tasks\n",
    "Ray APIs are general and simple to use. As a result, new comers' natural instinct is to parallelize all tasks, including tiny ones, which can incur an overhead overtime. \n",
    "In short, if the Ray remote tasks are tiny or miniscule in compute, they may take longer to execute than their serial Python equivalents."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "def tiny_task(x):\n",
    "    time.sleep(0.0001)\n",
    "    return 2 * x"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 136 ms, sys: 175 ms, total: 311 ms\n",
      "Wall time: 12.9 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "results = [tiny_task(x) for x in range(100000)]\n",
    "results[:10]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now convert this into Ray remote task"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def remote_tiny_task(x):\n",
    "    time.sleep(0.0001)\n",
    "    return x"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 7.22 s, sys: 3.33 s, total: 10.5 s\n",
      "Wall time: 11.9 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "result_ids = [remote_tiny_task.remote(x) for x in range(100000)]\n",
    "results = ray.get(result_ids)\n",
    "results[:10]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Surprisingly, Ray didn’t improve the execution time. Ray program is actually slower or closer in execution time than the sequential program! \n",
    "\n",
    "What's going on? What can we do to remedy it?\n",
    "\n",
    "Well, the issue here is that every task invocation has a non-trivial overhead (e.g., scheduling, inter-process communication, updating the system state), and this overhead dominates the actual time it takes to execute the task.\n",
    "\n",
    "One way to mitigate is to make the remote tasks \"larger\" in order to amortize invocation overhead. This is achieved by aggregating tasks into bigger chunks of 1000.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Bad Usage\n",
    "Avoid small many tiny tasks as the overhead to scheduler may be slower than serial execution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def mega_work(start, end):\n",
    "    return [tiny_task(x) for x in range(start, end)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 223 ms, sys: 33.3 ms, total: 257 ms\n",
      "Wall time: 3.95 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "result_ids = []\n",
    "[result_ids.append(mega_work.remote(x*1000, (x+1)*1000)) for x in range(100)]\n",
    "results = ray.get(result_ids)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A huge difference in execution time, almost **4X** faster!\n",
    "\n",
    "#### Good Usage\n",
    "Break or restructure many small tasks into batches or chunks of large Ray remote tasks, as demonstrated above\n",
    "\n",
    "#### Takeway tip 2:\n",
    "Where possible strive to batch tiny smaller Ray tasks into chuncks to reap the benefits of distributing them."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Tip 3: Using ray.wait() with ray.get()\n",
    "\n",
    "| Name | Argument Type |  Description |\n",
    "| :--- | :---     |  :---------- |\n",
    "| `ray.get()`     | `ObjectRef` or `List[ObjectRefs]`   | Return a value in the object ref or list of values from the object IDs. This is a synchronous (i.e., blocking) operation. |\n",
    "| `ray.wait()`    | `List[ObjectRefs]`  | From a list of object IDs, returns (1) the list of IDs of the objects that are ready, and (2) the list of IDs of the objects that are not ready yet. By default, it returns one ready object ID at a time. However, by specifying `num_returns=<value>` it will return all object IDs whose tasks are finished and there respective values materialized and available in the object\n",
    "store.\n",
    "\n",
    "\n",
    "As we noted above, an idiomatic way of using `ray.get()` is to delay fetching the object until you need them. Another way is to use it is with `ray.wait()`. Only fetch values that are already available or materialized in the object store. This is a way to [pipeline the execution](https://docs.ray.io/en/latest/ray-core/tips-for-first-time.html#tip-4-pipeline-data-processing), especially when you want to process the results of completed Ray tasks.\n",
    "\n",
    "|<img src=\"https://docs.ray.io/en/latest/_images/pipeline.png\" height=\"20%\" width=\"40%\">|\n",
    "|:--|\n",
    "|Execution timeline in both cases: when using `ray.get()` to wait for all results to become available before processing them, and using `ray.wait()` to start processing the results as soon as they become available.|\n",
    "\n",
    "\n",
    "If we use `ray.get()` on the results of multiple tasks we will have to wait until the last one of these tasks finishes. This can be an issue if tasks take widely different amounts of time.\n",
    "\n",
    "To illustrate this issue, consider the following example where we run four `transform_images()` tasks in parallel, with each task taking a time uniformly distributed between 0 and 4 seconds. Next, assume the results of these tasks are processed by `classify_images()`, which takes 1 sec per result. The expected running time is then (1) the time it takes to execute the slowest of the `transform_images()` tasks, plus (2) 4 seconds which is the time it takes to execute `classify_images()`.\n",
    "\n",
    "Let's look at a simple example."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Not using ray.wait and no pipelining"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "from PIL import Image, ImageFilter\n",
    "from torchvision import transforms as T"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "random.seed(42)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "import random\n",
    "import ray\n",
    "\n",
    "@ray.remote\n",
    "def transform_images(x):\n",
    "    imarray = np.random.rand(x, x , 3) * 255\n",
    "    img = Image.fromarray(imarray.astype('uint8')).convert('RGBA')\n",
    "    \n",
    "    # Make the image blur with specified intensify\n",
    "    img = img.filter(ImageFilter.GaussianBlur(radius=20))\n",
    "    \n",
    "    time.sleep(random.uniform(0, 4)) # Replace this with extra work you need to do.\n",
    "    return img\n",
    "\n",
    "def predict(image):\n",
    "    size = image.size[0]\n",
    "    if size == 16 or size == 32:\n",
    "        return 0\n",
    "    elif size == 64 or size == 128:\n",
    "        return 1\n",
    "    elif size == 256:\n",
    "        return 2\n",
    "    else:\n",
    "        return 3\n",
    "\n",
    "def classify_images(images):\n",
    "    preds = []\n",
    "    for image in images:\n",
    "        pred = predict(image)\n",
    "        time.sleep(1)\n",
    "        preds.append(pred)\n",
    "    return preds\n",
    "\n",
    "def classify_images_inc(images):\n",
    "    preds = [predict(img) for img in images]\n",
    "    time.sleep(1)\n",
    "    return preds\n",
    "\n",
    "SIZES = [16, 32, 64, 128, 256, 512]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Duration: 8.96 seconds and predictions: [0, 0, 1, 1, 2, 3]\n",
      "CPU times: user 61.2 ms, sys: 33.2 ms, total: 94.4 ms\n",
      "Wall time: 8.96 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "start = time.time()\n",
    "images = ray.get([transform_images.remote(image) for image in SIZES])\n",
    "predictions = classify_images(images)\n",
    "print(f\"Duration: {round(time.time() - start, 2)} seconds and predictions: {predictions}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Bad Usage\n",
    "Waiting for large number of tasks to finish using `ray.get()` on all of them before processing\n",
    "the results returned."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Using ray.wait and pipelining"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Duration: 6.88 seconds and predictions: [0, 1, 0, 1, 2, 3]\n",
      "CPU times: user 50.5 ms, sys: 28.3 ms, total: 78.9 ms\n",
      "Wall time: 6.88 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "start = time.time()\n",
    "result_images_refs = [transform_images.remote(image) for image in SIZES] \n",
    "predictions =[]\n",
    "\n",
    "# Loop until all tasks are finished\n",
    "while len(result_images_refs):\n",
    "    done_image_refs, result_images_refs = ray.wait(result_images_refs, num_returns=1)\n",
    "    preds = classify_images_inc(ray.get(done_image_refs))\n",
    "    predictions.extend(preds)\n",
    "print(f\"Duration: {round(time.time() - start, 2)} seconds and predictions: {predictions}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Notice**: You get some incremental difference. However, for compute intensive and many tasks, and overtime, this difference will be in order of magnitude."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Good Usage:\n",
    "For large number of tasks in flight, use `ray.get()` and `ray.wait()` to implement pipeline execution of processing\n",
    "those tasks already finished. \n",
    "\n",
    "#### Takeaway Tip 3: \n",
    "Use pipeline execution to process results returned from the finished Ray tasks using `ray.get()` and `ray.wait()`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Exercise for **Tip 3**:\n",
    " * Extend or add more images of sizes: 1024, 2048, ...\n",
    " * Increase the number of returns to 2 from the `ray.wait`()`\n",
    " * Process the images\n",
    " \n",
    " \n",
    " Is there a difference in processing time between serial and pipelining?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Tip 4: Avoid passing same object repeatedly to remote tasks¶\n",
    "\n",
    "When you pass a large object as an argument to a remote function, Ray calls `ray.put()` under the hood to store that object in the local object store. Done once with the same object, say outside a loop, can significantly improve the performance of a remote task invocation when the remote task is executed locally, or if running on the same cluster node, as all tasks on the same node share the object store in shared memory. But if done by passing the same object repeatedly inside a loop can degrade the performance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def do_work(a):\n",
    "    return np.sum(a)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " results = 124995931.27 and duration = 0.729 sec\n"
     ]
    }
   ],
   "source": [
    "start = time.time()\n",
    "a = np.random.rand(5000, 5000)\n",
    "\n",
    "# Sending the big array to each remote task\n",
    "result_ids = [do_work.remote(a) for x in range(10)]\n",
    "results = math.fsum(ray.get(result_ids))\n",
    "print(f\" results = {results:.2f} and duration = {time.time() - start:.3f} sec\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Bad Usage\n",
    "Avoid sending the same large objects to Ray remote tasks. This creates multiple copies of the same\n",
    "object in the Ray distributed object store. Storing and fetching and copying identical object can degrade performance overtime."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we going to store the large array into the object store, so there is only a\n",
    "single copy and added only once. Plus we sending not the array but a reference to it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " results = 124998578.35 and duration = 0.418 sec\n"
     ]
    }
   ],
   "source": [
    "start = time.time()\n",
    "\n",
    "# Adding the big array into the object store\n",
    "a_id = ray.put(np.random.rand(5000, 5000))\n",
    "result_ids = [do_work.remote(a_id) for x in range(10)]\n",
    "results = math.fsum(ray.get(result_ids))\n",
    "print(f\" results = {results:.2f} and duration = {time.time() - start:.3f} sec\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Good Usage\n",
    "Place or insert the large object store into Ray's remote object store and only send the object Ref to the Ray remote task.\n",
    "\n",
    "#### Takeaway Tip 4:\n",
    "Avoid sending the same large object to a Ray remote tasks. Instead, put it into the object store and only send the object ref."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.shutdown()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Summary\n",
    "\n",
    "In this short tutorial, we got a short glimpse at the Ray Core APIs. By no means it was comprehensive, but we touched upon some methods we \n",
    "have seen in the previous lessons. With those methods, we explored additional arguments to the `.remote()` call such as number of return\n",
    "statements as well as how to supply runtime environments and dependencies for your Ray cluster during `ray.init()` call. Note that some arguments to `ray.init()` \n",
    "can also be supplied to `ray.remote()` decorator, such as `num_cpus`, `num_gpus`, `runtime_env`, etc. \n",
    "\n",
    "More importantly, we walked through some tips and tricks that many developers new to Ray can easily stumble upon. Although the examples were short and simple,\n",
    "the lessons behind the cautionary tales are important part of the learning process."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Next Step\n",
    "\n",
    "Let's move on to our final module 3 and start with [Ray Datasets lesson](ex_07_ray_data.ipynb)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "📖 [Back to Table of Contents](./ex_00_tutorial_overview.ipynb)<br>\n",
    "➡ [Next notebook](./ex_07_ray_data.ipynb) <br>\n",
    "⬅️ [Previous notebook](./ex_05_multiprocess_pool.ipynb) <br>"
   ]
  }
 ],
 "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.13"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {
    "height": "calc(100% - 180px)",
    "left": "10px",
    "top": "150px",
    "width": "382.391px"
   },
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}