Ray_exercises.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 1 - Simple Data Parallel Example\n",
    "\n",
    "**GOAL:** The goal of this exercise is to show how to run simple tasks in parallel.\n",
    "\n",
    "This script is too slow, and the computation is embarrassingly parallel. In this exercise, you will use Ray to execute the functions in parallel to speed it up.\n",
    "\n",
    "### Concept for this Exercise - Remote Functions\n",
    "\n",
    "The standard way to turn a Python function into a remote function is to add the `@ray.remote` decorator. Here is an example.\n",
    "\n",
    "```python\n",
    "# A regular Python function.\n",
    "def regular_function():\n",
    "    return 1\n",
    "\n",
    "# A Ray remote function.\n",
    "@ray.remote\n",
    "def remote_function():\n",
    "    return 1\n",
    "```\n",
    "\n",
    "The differences are the following:\n",
    "\n",
    "1. **Invocation:** The regular version is called with `regular_function()`, whereas the remote version is called with `remote_function.remote()`.\n",
    "2. **Return values:** `regular_function` immediately executes and returns `1`, whereas `remote_function` immediately returns an object ID (a future) and then creates a task that will be executed on a worker process. The result can be obtained with `ray.get`.\n",
    "    ```python\n",
    "    >>> regular_function()\n",
    "    1\n",
    "    \n",
    "    >>> remote_function.remote()\n",
    "    ObjectID(1c80d6937802cd7786ad25e50caf2f023c95e350)\n",
    "    \n",
    "    >>> ray.get(remote_function.remote())\n",
    "    1\n",
    "    ```\n",
    "3. **Parallelism:** Invocations of `regular_function` happen **serially**, for example\n",
    "    ```python\n",
    "    # These happen serially.\n",
    "    for _ in range(4):\n",
    "        regular_function()\n",
    "    ```\n",
    "    whereas invocations of `remote_function` happen in **parallel**, for example\n",
    "    ```python\n",
    "    # These happen in parallel.\n",
    "    for _ in range(4):\n",
    "        remote_function.remote()\n",
    "    ```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-11-30T17:48:30.874373Z",
     "start_time": "2018-11-30T17:48:26.995195Z"
    }
   },
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Start Ray. By default, Ray does not schedule more tasks concurrently than there are CPUs. This example requires four tasks to run concurrently, so we tell Ray that there are four CPUs. Usually this is not done and Ray computes the number of CPUs using `psutil.cpu_count()`. The argument `ignore_reinit_error=True` just ignores errors if the cell is run multiple times.\n",
    "\n",
    "The call to `ray.init` starts a number of processes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-11-30T17:48:58.177079Z",
     "start_time": "2018-11-30T17:48:58.159632Z"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Calling ray.init() again after it has already been called.\n"
     ]
    }
   ],
   "source": [
    "ray.init(num_cpus=3, include_webui=False, ignore_reinit_error=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** The function below is slow. Turn it into a remote function using the `@ray.remote` decorator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-11-30T17:37:39.092666Z",
     "start_time": "2018-11-30T17:37:39.080189Z"
    }
   },
   "outputs": [],
   "source": [
    "# This function is a proxy for a more interesting and computationally\n",
    "# intensive function.\n",
    "@ray.remote\n",
    "def slow_function(i):\n",
    "    time.sleep(1)\n",
    "    return i"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** The loop below takes too long. The four function calls could be executed in parallel. Instead of four seconds, it should only take one second. Once `slow_function` has been made a remote function, execute these four tasks in parallel by calling `slow_function.remote()`. Then obtain the results by calling `ray.get` on a list of the resulting object IDs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-11-30T17:37:46.519365Z",
     "start_time": "2018-11-30T17:37:43.247638Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The results are [0, 1, 2]. This took 1.2578864097595215 seconds. Run the next cell to see if the exercise was done correctly.\n"
     ]
    }
   ],
   "source": [
    "# Sleep a little to improve the accuracy of the timing measurements below.\n",
    "# We do this because workers may still be starting up in the background.\n",
    "time.sleep(2.0)\n",
    "start_time = time.time()\n",
    "\n",
    "results = ray.get([slow_function.remote(i) for i in range(3)])\n",
    "\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time\n",
    "\n",
    "print('The results are {}. This took {} seconds. Run the next cell to see '\n",
    "      'if the exercise was done correctly.'.format(results, duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-11-30T17:37:52.227402Z",
     "start_time": "2018-11-30T17:37:52.207635Z"
    }
   },
   "outputs": [
    {
     "ename": "AssertionError",
     "evalue": "The loop took 1.2578864097595215 seconds. This is too slow.",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mAssertionError\u001b[0m                            Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-6-93102aa7ee15>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;32massert\u001b[0m \u001b[0mresults\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0;34m[\u001b[0m\u001b[0;36m0\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;36m1\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;36m2\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m'Did you remember to call ray.get?'\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      2\u001b[0m assert duration < 1.1, ('The loop took {} seconds. This is too slow.'\n\u001b[0;32m----> 3\u001b[0;31m                         .format(duration))\n\u001b[0m\u001b[1;32m      4\u001b[0m assert duration > 1, ('The loop took {} seconds. This is too fast.'\n\u001b[1;32m      5\u001b[0m                       .format(duration))\n",
      "\u001b[0;31mAssertionError\u001b[0m: The loop took 1.2578864097595215 seconds. This is too slow."
     ]
    }
   ],
   "source": [
    "assert results == [0, 1, 2, 3], 'Did you remember to call ray.get?'\n",
    "assert duration < 1.1, ('The loop took {} seconds. This is too slow.'\n",
    "                        .format(duration))\n",
    "assert duration > 1, ('The loop took {} seconds. This is too fast.'\n",
    "                      .format(duration))\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Use the UI to view the task timeline and to verify that the four tasks were executed in parallel. After running the cell below, you'll need to click on **View task timeline**\".\n",
    "- Using the **second** button, you can click and drag to **move** the timeline.\n",
    "- Using the **third** button, you can click and drag to **zoom**. You can also zoom by holding \"alt\" and scrolling.\n",
    "\n",
    "**NOTE:** Normally our UI is used as a separate Jupyter notebook. However, for simplicity we embedded the relevant feature here in this notebook.\n",
    "\n",
    "**NOTE:** The first time you click **View task timeline** it may take **several minutes** to start up. This will change.\n",
    "\n",
    "**NOTE:** If you run more tasks and want to regenerate the UI, you need to move the slider bar a little bit and then click **View task timeline** again.\n",
    "\n",
    "**NOTE:** The timeline visualization may only work in **Chrome**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-11-30T17:38:01.538090Z",
     "start_time": "2018-11-30T17:38:01.312821Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "d2e8edcb203b4b1094b23329266a86cd",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "Dropdown(description='Selection Options:', index=1, options=('% total time', '% total tasks'), value='% total …"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "24e251f10497401fb6cb9c9dc5379093",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(IntRangeSlider(value=(0, 100), continuous_update=False, description='%:'), VBox(children=(Float…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "43a1efb6b12a4754b5668b7a7598d3d0",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(Checkbox(value=True, layout=Layout(width='20px')), Label(value='Task submissions', layout=Layou…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e3c7c96789244de2967a5c6a9ae713bd",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(Label(value='View options:', layout=Layout(width='100px')), Dropdown(index=1, options=('Basic',…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "0e58418aa8d44f28a93be46080219459",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "Button(description='View task timeline', style=ButtonStyle())"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import ray.experimental.ui as ui\n",
    "ui.task_timeline()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 2 - Parallel Data Processing with Task Dependencies\n",
    "\n",
    "**GOAL:** The goal of this exercise is to show how to pass object IDs into remote functions to encode dependencies between tasks.\n",
    "\n",
    "In this exercise, we construct a sequence of tasks each of which depends on the previous mimicking a data parallel application. Within each sequence, tasks are executed serially, but multiple sequences can be executed in parallel.\n",
    "\n",
    "In this exercise, you will use Ray to parallelize the computation below and speed it up.\n",
    "\n",
    "### Concept for this Exercise - Task Dependencies\n",
    "\n",
    "Suppose we have two remote functions defined as follows.\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "def f(x):\n",
    "    return x\n",
    "```\n",
    "\n",
    "Arguments can be passed into remote functions as usual.\n",
    "\n",
    "```python\n",
    ">>> x1_id = f.remote(1)\n",
    ">>> ray.get(x1_id)\n",
    "1\n",
    "\n",
    ">>> x2_id = f.remote([1, 2, 3])\n",
    ">>> ray.get(x2_id)\n",
    "[1, 2, 3]\n",
    "```\n",
    "\n",
    "**Object IDs** can also be passed into remote functions. When the function actually gets executed, **the argument will be a retrieved as a regular Python object**.\n",
    "\n",
    "```python\n",
    ">>> y1_id = f.remote(x1_id)\n",
    ">>> ray.get(y1_id)\n",
    "1\n",
    "\n",
    ">>> y2_id = f.remote(x2_id)\n",
    ">>> ray.get(y2_id)\n",
    "[1, 2, 3]\n",
    "```\n",
    "\n",
    "So when implementing a remote function, the function should expect a regular Python object regardless of whether the caller passes in a regular Python object or an object ID.\n",
    "\n",
    "**Task dependencies affect scheduling.** In the example above, the task that creates `y1_id` depends on the task that creates `x1_id`. This has the following implications.\n",
    "\n",
    "- The second task will not be executed until the first task has finished executing.\n",
    "- If the two tasks are scheduled on different machines, the output of the first task (the value corresponding to `x1_id`) will be copied over the network to the machine where the second task is scheduled."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-11-30T14:03:40.579691Z",
     "start_time": "2018-11-30T14:03:40.569563Z"
    }
   },
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import numpy as np\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-11-30T14:03:46.747576Z",
     "start_time": "2018-11-30T14:03:46.740152Z"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Calling ray.init() again after it has already been called.\n"
     ]
    }
   ],
   "source": [
    "ray.init(num_cpus=4, include_webui=False, ignore_reinit_error=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These are some helper functions that mimic an example pattern of a data parallel application.\n",
    "\n",
    "**EXERCISE:** You will need to turn all of these functions into remote functions. When you turn these functions into remote function, you do not have to worry about whether the caller passes in an object ID or a regular object. In both cases, the arguments will be regular objects when the function executes. This means that even if you pass in an object ID, you **do not need to call `ray.get`** inside of these remote functions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_data(filename):\n",
    "    time.sleep(0.1)\n",
    "    return np.ones((1000, 100))\n",
    "\n",
    "def normalize_data(data):\n",
    "    time.sleep(0.1)\n",
    "    return data - np.mean(data, axis=0)\n",
    "\n",
    "def extract_features(normalized_data):\n",
    "    time.sleep(0.1)\n",
    "    return np.hstack([normalized_data, normalized_data ** 2])\n",
    "\n",
    "def compute_loss(features):\n",
    "    num_data, dim = features.shape\n",
    "    time.sleep(0.1)\n",
    "    return np.sum((np.dot(features, np.ones(dim)) - np.ones(num_data)) ** 2)\n",
    "\n",
    "assert hasattr(load_data, 'remote'), 'load_data must be a remote function'\n",
    "assert hasattr(normalize_data, 'remote'), 'normalize_data must be a remote function'\n",
    "assert hasattr(extract_features, 'remote'), 'extract_features must be a remote function'\n",
    "assert hasattr(compute_loss, 'remote'), 'compute_loss must be a remote function'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** The loop below takes too long. Parallelize the four passes through the loop by turning `load_data`, `normalize_data`, `extract_features`, and `compute_loss` into remote functions and then retrieving the losses with `ray.get`.\n",
    "\n",
    "**NOTE:** You should only use **ONE** call to `ray.get`. For example, the object ID returned by `load_data` should be passed directly into `normalize_data` without needing to be retrieved by the driver."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sleep a little to improve the accuracy of the timing measurements below.\n",
    "time.sleep(2.0)\n",
    "start_time = time.time()\n",
    "\n",
    "losses = []\n",
    "for filename in ['file1', 'file2', 'file3', 'file4']:\n",
    "    inner_start = time.time()\n",
    "\n",
    "    data = load_data.remote(filename)\n",
    "    normalized_data = normalize_data.remote(data)\n",
    "    features = extract_features.remote(normalized_data)\n",
    "    loss = compute_loss.remote(features)\n",
    "    losses.append(loss)\n",
    "    \n",
    "    inner_end = time.time()\n",
    "    \n",
    "    if inner_end - inner_start >= 0.1:\n",
    "        raise Exception('You may be calling ray.get inside of the for loop! '\n",
    "                        'Doing this will prevent parallelism from being exposed. '\n",
    "                        'Make sure to only call ray.get once outside of the for loop.')\n",
    "\n",
    "print('The losses are {}.'.format(losses) + '\\n')\n",
    "loss = sum(ray.get(losses))\n",
    "\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time\n",
    "\n",
    "print('The loss is {}. This took {} seconds. Run the next cell to see '\n",
    "      'if the exercise was done correctly.'.format(loss, duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert loss == 4000\n",
    "assert duration < 0.8, ('The loop took {} seconds. This is too slow.'\n",
    "                        .format(duration))\n",
    "assert duration > 0.4, ('The loop took {} seconds. This is too fast.'\n",
    "                        .format(duration))\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Use the UI to view the task timeline and to verify that the relevant tasks were executed in parallel. After running the cell below, you'll need to click on **View task timeline**\".\n",
    "- Using the **second** button, you can click and drag to **move** the timeline.\n",
    "- Using the **third** button, you can click and drag to **zoom**. You can also zoom by holding \"alt\" and scrolling.\n",
    "\n",
    "In the timeline, click on **View Options** and select **Flow Events** to visualize tasks dependencies."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import ray.experimental.ui as ui\n",
    "ui.task_timeline()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 3 - Nested Parallelism\n",
    "\n",
    "**GOAL:** The goal of this exercise is to show how to create nested tasks by calling a remote function inside of another remote function.\n",
    "\n",
    "In this exercise, you will implement the structure of a parallel hyperparameter sweep which trains a number of models in parallel. Each model will be trained using parallel gradient computations.\n",
    "\n",
    "### Concepts for this Exercise - Nested Remote Functions\n",
    "\n",
    "Remote functions can call other functions. For example, consider the following.\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "def f():\n",
    "    return 1\n",
    "\n",
    "@ray.remote\n",
    "def g():\n",
    "    # Call f 4 times and return the resulting object IDs.\n",
    "    return [f.remote() for _ in range(4)]\n",
    "\n",
    "@ray.remote\n",
    "def h():\n",
    "    # Call f 4 times, block until those 4 tasks finish,\n",
    "    # retrieve the results, and return the values.\n",
    "    return ray.get([f.remote() for _ in range(4)])\n",
    "```\n",
    "\n",
    "Then calling `g` and `h` produces the following behavior.\n",
    "\n",
    "```python\n",
    ">>> ray.get(g.remote())\n",
    "[ObjectID(b1457ba0911ae84989aae86f89409e953dd9a80e),\n",
    " ObjectID(7c14a1d13a56d8dc01e800761a66f09201104275),\n",
    " ObjectID(99763728ffc1a2c0766a2000ebabded52514e9a6),\n",
    " ObjectID(9c2f372e1933b04b2936bb6f58161285829b9914)]\n",
    "\n",
    ">>> ray.get(h.remote())\n",
    "[1, 1, 1, 1]\n",
    "```\n",
    "\n",
    "**One limitation** is that the definition of `f` must come before the definitions of `g` and `h` because as soon as `g` is defined, it will be pickled and shipped to the workers, and so if `f` hasn't been defined yet, the definition will be incomplete."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import numpy as np\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.init(num_cpus=9, include_webui=False, ignore_reinit_error=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This example represents a hyperparameter sweep in which multiple models are trained in parallel. Each model training task also performs data parallel gradient computations.\n",
    "\n",
    "**EXERCISE:** Turn `compute_gradient` and `train_model` into remote functions so that they can be executed in parallel. Inside of `train_model`, do the calls to `compute_gradient` in parallel and fetch the results using `ray.get`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def compute_gradient(data, current_model):\n",
    "    time.sleep(0.03)\n",
    "    return 1\n",
    "\n",
    "@ray.remote\n",
    "def train_model(hyperparameters):\n",
    "    current_model = 0\n",
    "    # Iteratively improve the current model. This outer loop cannot be parallelized.\n",
    "    for _ in range(10):\n",
    "        # EXERCISE: Parallelize the list comprehension in the line below. After you\n",
    "        # turn \"compute_gradient\" into a remote function, you will need to call it\n",
    "        # with \".remote\". The results must be retrieved with \"ray.get\" before \"sum\"\n",
    "        # is called.\n",
    "        total_gradient = sum([compute_gradient.remote(j, current_model) for j in range(2)])\n",
    "        current_model += total_gradient\n",
    "\n",
    "    return ray.get(current_model)\n",
    "\n",
    "assert hasattr(compute_gradient, 'remote'), 'compute_gradient must be a remote function'\n",
    "assert hasattr(train_model, 'remote'), 'train_model must be a remote function'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** The code below runs 3 hyperparameter experiments. Change this to run the experiments in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sleep a little to improve the accuracy of the timing measurements below.\n",
    "time.sleep(2.0)\n",
    "start_time = time.time()\n",
    "\n",
    "# Run some hyperparaameter experiments.\n",
    "results = []\n",
    "for hyperparameters in [{'learning_rate': 1e-1, 'batch_size': 100},\n",
    "                        {'learning_rate': 1e-2, 'batch_size': 100},\n",
    "                        {'learning_rate': 1e-3, 'batch_size': 100}]:\n",
    "    results.append(train_model.remote(hyperparameters))\n",
    "\n",
    "# EXERCISE: Once you've turned \"results\" into a list of Ray ObjectIDs\n",
    "# by calling train_model.remote, you will need to turn \"results\" back\n",
    "# into a list of integers, e.g., by doing \"results = ray.get(results)\".\n",
    "results = ray.get(results)\n",
    "\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time\n",
    "\n",
    "assert all([isinstance(x, int) for x in results]), 'Looks like \"results\" is {}. You may have forgotten to call ray.get.'.format(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert results == [20, 20, 20]\n",
    "assert duration < 0.5, ('The experiments ran in {} seconds. This is too '\n",
    "                         'slow.'.format(duration))\n",
    "assert duration > 0.3, ('The experiments ran in {} seconds. This is too '\n",
    "                        'fast.'.format(duration))\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Use the UI to view the task timeline and to verify that the pattern makes sense."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import ray.experimental.ui as ui\n",
    "ui.task_timeline()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 4 - Introducing Actors\n",
    "\n",
    "**Goal:** The goal of this exercise is to show how to create an actor and how to call actor methods.\n",
    "\n",
    "See the documentation on actors at http://ray.readthedocs.io/en/latest/actors.html.\n",
    "\n",
    "Sometimes you need a \"worker\" process to have \"state\". For example, that state might be a neural network, a simulator environment, a counter, or something else entirely. However, remote functions are side-effect free. That is, they operate on inputs and produce outputs, but they don't change the state of the worker they execute on.\n",
    "\n",
    "Actors are different. When we instantiate an actor, a brand new worker is created, and all methods that are called on that actor are executed on the newly created worker.\n",
    "\n",
    "This means that with a single actor, no parallelism can be achieved because calls to the actor's methods will be executed one at a time. However, multiple actors can be created and methods can be executed on them in parallel.\n",
    "\n",
    "### Concepts for this Exercise - Actors\n",
    "\n",
    "To create an actor, decorate Python class with the `@ray.remote` decorator.\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "class Example(object):\n",
    "    def __init__(self, x):\n",
    "        self.x = x\n",
    "    \n",
    "    def set(self, x):\n",
    "        self.x = x\n",
    "    \n",
    "    def get(self):\n",
    "        return self.x\n",
    "```\n",
    "\n",
    "Like regular Python classes, **actors encapsulate state that is shared across actor method invocations**.\n",
    "\n",
    "Actor classes differ from regular Python classes in the following ways.\n",
    "1. **Instantiation:** A regular class would be instantiated via `e = Example(1)`. Actors are instantiated via\n",
    "    ```python\n",
    "    e = Example.remote(1)\n",
    "    ```\n",
    "    When an actor is instantiated, a **new worker process** is created by a local scheduler somewhere in the cluster.\n",
    "2. **Method Invocation:** Methods of a regular class would be invoked via `e.set(2)` or `e.get()`. Actor methods are invoked differently.\n",
    "    ```python\n",
    "    >>> e.set.remote(2)\n",
    "    ObjectID(d966aa9b6486331dc2257522734a69ff603e5a1c)\n",
    "    \n",
    "    >>> e.get.remote()\n",
    "    ObjectID(7c432c085864ed4c7c18cf112377a608676afbc3)\n",
    "    ```\n",
    "3. **Return Values:** Actor methods are non-blocking. They immediately return an object ID and **they create a task which is scheduled on the actor worker**. The result can be retrieved with `ray.get`.\n",
    "    ```python\n",
    "    >>> ray.get(e.set.remote(2))\n",
    "    None\n",
    "    \n",
    "    >>> ray.get(e.get.remote())\n",
    "    2\n",
    "    ```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import numpy as np\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.init(num_cpus=4, include_webui=False, ignore_reinit_error=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Change the `Foo` class to be an actor class by using the `@ray.remote` decorator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "class Foo(object):\n",
    "    def __init__(self):\n",
    "        self.counter = 0\n",
    "\n",
    "    def reset(self):\n",
    "        self.counter = 0\n",
    "\n",
    "    def increment(self):\n",
    "        time.sleep(0.5)\n",
    "        self.counter += 1\n",
    "        return self.counter\n",
    "\n",
    "assert hasattr(Foo, 'remote'), 'You need to turn \"Foo\" into an actor with @ray.remote.'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-11-30T16:42:57.032942Z",
     "start_time": "2018-11-30T16:42:57.014336Z"
    }
   },
   "source": [
    "**EXERCISE:** Change the intantiations below to create two actors by calling `Foo.remote()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create two Foo objects.\n",
    "f1 = Foo.remote()\n",
    "f2 = Foo.remote()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Parallelize the code below. The two actors can execute methods in parallel (though each actor can only execute one method at a time)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sleep a little to improve the accuracy of the timing measurements below.\n",
    "time.sleep(2.0)\n",
    "start_time = time.time()\n",
    "\n",
    "# Reset the actor state so that we can run this cell multiple times without\n",
    "# changing the results.\n",
    "f1.reset.remote()\n",
    "f2.reset.remote()\n",
    "\n",
    "# We want to parallelize this code. However, it is not straightforward to\n",
    "# make \"increment\" a remote function, because state is shared (the value of\n",
    "# \"self.counter\") between subsequent calls to \"increment\". In this case, it\n",
    "# makes sense to use actors.\n",
    "results = []\n",
    "for _ in range(5):\n",
    "    results.append(f1.increment.remote())\n",
    "    results.append(f2.increment.remote())\n",
    "\n",
    "results = ray.get(results)\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time\n",
    "\n",
    "assert not any([isinstance(result, ray.ObjectID) for result in results]), 'Looks like \"results\" is {}. You may have forgotten to call ray.get.'.format(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert results == [1, 1, 2, 2, 3, 3, 4, 4, 5, 5]\n",
    "\n",
    "assert duration < 3, ('The experiments ran in {} seconds. This is too '\n",
    "                      'slow.'.format(duration))\n",
    "assert duration > 2.5, ('The experiments ran in {} seconds. This is too '\n",
    "                        'fast.'.format(duration))\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 5 - Actor Handles\n",
    "\n",
    "**GOAL:** The goal of this exercise is to show how to pass around actor handles.\n",
    "\n",
    "Suppose we wish to have multiple tasks invoke methods on the same actor. For example, we may have a single actor that records logging information from a number of tasks. We can achieve this by passing a handle to the actor as an argument into the relevant tasks.\n",
    "\n",
    "### Concepts for this Exercise - Actor  Handles\n",
    "\n",
    "First of all, suppose we've created an actor as follows.\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "class Actor(object):\n",
    "    def method(self):\n",
    "        pass\n",
    "\n",
    "# Create the actor\n",
    "actor = Actor.remote()\n",
    "```\n",
    "\n",
    "Then we can define a remote function (or another actor) that takes an actor handle as an argument.\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "def f(actor):\n",
    "    # We can invoke methods on the actor.\n",
    "    x_id = actor.method.remote()\n",
    "    # We can block and get the results.\n",
    "    return ray.get(x_id)\n",
    "```\n",
    "\n",
    "Then we can invoke the remote function a few times and pass in the actor handle.\n",
    "\n",
    "```python\n",
    "# Each of the three tasks created below will invoke methods on the same actor.\n",
    "f.remote(actor)\n",
    "f.remote(actor)\n",
    "f.remote(actor)\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "from collections import defaultdict\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.init(num_cpus=4, include_webui=False, ignore_reinit_error=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this exercise, we're going to write some code that runs several \"experiments\" in parallel and has each experiment log its results to an actor. The driver script can then periodically pull the results from the logging actor.\n",
    "\n",
    "**EXERCISE:** Turn this `LoggingActor` class into an actor class."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "class LoggingActor(object):\n",
    "    def __init__(self):\n",
    "        self.logs = defaultdict(lambda: [])\n",
    "    \n",
    "    def log(self, index, message):\n",
    "        self.logs[index].append(message)\n",
    "    \n",
    "    def get_logs(self):\n",
    "        return dict(self.logs)\n",
    "\n",
    "\n",
    "assert hasattr(LoggingActor, 'remote'), ('You need to turn LoggingActor into an '\n",
    "                                         'actor (by using the ray.remote keyword).')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Instantiate the actor."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "logging_actor = LoggingActor.remote()\n",
    "\n",
    "# Some checks to make sure this was done correctly.\n",
    "assert hasattr(logging_actor, 'get_logs')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we define a remote function that runs and pushes its logs to the `LoggingActor`.\n",
    "\n",
    "**EXERCISE:** Modify this function so that it invokes methods correctly on `logging_actor` (you need to change the way you call the `log` method)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def run_experiment(experiment_index, logging_actor):\n",
    "    for i in range(60):\n",
    "        time.sleep(1)\n",
    "        # Push a logging message to the actor.\n",
    "        logging_actor.log.remote(experiment_index, 'On iteration {}'.format(i))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we create several tasks that use the logging actor."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "experiment_ids = [run_experiment.remote(i, logging_actor) for i in range(3)]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "While the experiments are running in the background, the driver process (that is, this Jupyter notebook) can query the actor to read the logs.\n",
    "\n",
    "**EXERCISE:** Modify the code below to dispatch methods to the `LoggingActor`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "logs = ray.get(logging_actor.get_logs.remote())\n",
    "print(logs)\n",
    "\n",
    "assert isinstance(logs, dict), (\"Make sure that you dispatch tasks to the \"\n",
    "                                \"actor using the .remote keyword and get the results using ray.get.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Try running the above box multiple times and see how the results change (while the experiments are still running in the background). You can also try running more of the experiment tasks and see what happens."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "cell_style": "center"
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.6.2"
  },
  "nbTranslate": {
   "displayLangs": [
    "*"
   ],
   "hotkey": "alt-t",
   "langInMainMenu": true,
   "sourceLang": "en",
   "targetLang": "fr",
   "useGoogleTranslate": true
  },
  "toc": {
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": "block",
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}