Ray_Tutorial.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Ray Tutorial.ipynb",
      "version": "0.3.2",
      "provenance": [],
      "collapsed_sections": [],
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "accelerator": "GPU"
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/denklewer/yandex-rl-tasks/blob/master/Ray_Tutorial.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "LaStUkbRS7jm",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# # in google colab uncomment this\n",
        "\n",
        "import os"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "kRl2cO9eTFBe",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "90b53a79-9f9d-4955-fbc9-48f849bbbb2e"
      },
      "source": [
        "os.system('pip install ray')"
      ],
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "0"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 2
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "TG2xe-d8WZ0c",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 255
        },
        "outputId": "824d5d1e-96fe-4c7b-aa0a-09746ffd69e2"
      },
      "source": [
        "import ray\n",
        "ray.init(num_gpus=8)"
      ],
      "execution_count": 3,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "2019-05-12 16:10:56,129\tWARNING worker.py:1337 -- WARNING: Not updating worker name since `setproctitle` is not installed. Install this with `pip install setproctitle` (or ray[debug]) to enable monitoring of worker processes.\n",
            "2019-05-12 16:10:56,131\tINFO node.py:469 -- Process STDOUT and STDERR is being redirected to /tmp/ray/session_2019-05-12_16-10-56_1940/logs.\n",
            "2019-05-12 16:10:56,243\tINFO services.py:407 -- Waiting for redis server at 127.0.0.1:63370 to respond...\n",
            "2019-05-12 16:10:56,384\tINFO services.py:407 -- Waiting for redis server at 127.0.0.1:48214 to respond...\n",
            "2019-05-12 16:10:56,391\tINFO services.py:804 -- Starting Redis shard with 2.52 GB max memory.\n",
            "2019-05-12 16:10:56,433\tINFO node.py:483 -- Process STDOUT and STDERR is being redirected to /tmp/ray/session_2019-05-12_16-10-56_1940/logs.\n",
            "2019-05-12 16:10:56,438\tINFO services.py:1427 -- Starting the Plasma object store with 3.78 GB memory using /dev/shm.\n"
          ],
          "name": "stderr"
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "{'node_ip_address': '172.28.0.2',\n",
              " 'object_store_address': '/tmp/ray/session_2019-05-12_16-10-56_1940/sockets/plasma_store',\n",
              " 'raylet_socket_name': '/tmp/ray/session_2019-05-12_16-10-56_1940/sockets/raylet',\n",
              " 'redis_address': '172.28.0.2:63370',\n",
              " 'webui_url': None}"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 3
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "cQEpKXOsaqH6",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "a20f6602-acfb-43bd-ca88-60adbcb6863e"
      },
      "source": [
        "x = \"example\"\n",
        "ray.put(x)  # ObjectID(b49a32d72057bdcfc4dda35584b3d838aad89f5d)"
      ],
      "execution_count": 4,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "ObjectID(ffffffff11d2dfcaf2d3e429e56d8b95f47d0ef7)"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 4
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "eQuqAU6Aa6pY",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "27c9d28a-e58f-4747-c543-1376f8742685"
      },
      "source": [
        "x_id = ray.put(\"example\")\n",
        "ray.get(x_id)  # \"example\""
      ],
      "execution_count": 5,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "'example'"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 5
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "fNyKGCGebIXE",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "68121de2-c863-4eae-dcc8-132b25adc731"
      },
      "source": [
        "result_ids = [ray.put(i) for i in range(10)]\n",
        "ray.get(result_ids)  # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]"
      ],
      "execution_count": 6,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 6
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GeiD9mZjbOIG",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def add1(a, b):\n",
        "    return a + b"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "gP5wJJwDbSp1",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "@ray.remote\n",
        "def add2(a, b):\n",
        "    return a + b"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "lxh5czj9bU92",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "22b7ba3e-21c9-4dae-a3ea-647ec061bc4e"
      },
      "source": [
        "x_id = add2.remote(1, 2)\n",
        "ray.get(x_id)  # 3"
      ],
      "execution_count": 9,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "3"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 9
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "WRFGKWU_bZ7S",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "6f19298f-e827-4187-f656-0381da8f7f69"
      },
      "source": [
        "import time\n",
        "\n",
        "def f1():\n",
        "    time.sleep(1)\n",
        "\n",
        "@ray.remote\n",
        "def f2():\n",
        "    time.sleep(1)\n",
        "\n",
        "# The following takes ten seconds.\n",
        "[f1() for _ in range(10)]\n",
        "\n",
        "# The following takes one second (assuming the system has at least ten CPUs).\n",
        "ray.get([f2.remote() for _ in range(10)])"
      ],
      "execution_count": 10,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[None, None, None, None, None, None, None, None, None, None]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 10
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ErrFy8wkbdka",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "7c065735-bb44-4cfa-cb50-25a2373025ac"
      },
      "source": [
        "add2.remote(1, 2)\n",
        "add2.remote(1, ray.put(2))\n",
        "add2.remote(ray.put(1), ray.put(2))"
      ],
      "execution_count": 11,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "ObjectID(01000000839c2f5495c7538f46b31d89744d7a0c)"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 11
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "HzhPf9qHbryM",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "@ray.remote(num_return_vals=3)\n",
        "def return_multiple():\n",
        "    return 1, 2, 3\n",
        "\n",
        "a_id, b_id, c_id = return_multiple.remote()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "oKGo-Besb2vp",
        "colab_type": "text"
      },
      "source": [
        "## Expressing dependencies between tasks"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "fIKQ8PeGbxXB",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "cbc4e7a8-6554-43e8-a63b-0b4c04fda309"
      },
      "source": [
        "@ray.remote\n",
        "def f(x):\n",
        "    return x + 1\n",
        "\n",
        "x = f.remote(0)\n",
        "y = f.remote(x)\n",
        "z = f.remote(y)\n",
        "ray.get(z) # 3"
      ],
      "execution_count": 13,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "3"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 13
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "n2rcQK5McGkT",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import numpy as np\n",
        "\n",
        "@ray.remote\n",
        "def generate_data():\n",
        "    return np.random.normal(size=1000)\n",
        "\n",
        "@ray.remote\n",
        "def aggregate_data(x, y):\n",
        "    return x + y\n",
        "\n",
        "# Generate some random data. This launches 100 tasks that will be scheduled on\n",
        "# various nodes. The resulting data will be distributed around the cluster.\n",
        "data = [generate_data.remote() for _ in range(100)]\n",
        "\n",
        "# Perform a tree reduce.\n",
        "while len(data) > 1:\n",
        "    data.append(aggregate_data.remote(data.pop(0), data.pop(0)))\n",
        "\n",
        "# Fetch the result.\n",
        "ray.get(data)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "8vIaGDmxcRmM",
        "colab_type": "text"
      },
      "source": [
        "## Remote Functions Within Remote Functions"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "MUDceyD9cUiM",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 92
        },
        "outputId": "6fadb406-3753-449c-a896-15a7ca7d66ec"
      },
      "source": [
        "@ray.remote\n",
        "def sub_experiment(i, j):\n",
        "    # Run the jth sub-experiment for the ith experiment.\n",
        "    return i + j\n",
        "\n",
        "@ray.remote\n",
        "def run_experiment(i):\n",
        "    sub_results = []\n",
        "    # Launch tasks to perform 10 sub-experiments in parallel.\n",
        "    for j in range(10):\n",
        "        sub_results.append(sub_experiment.remote(i, j))\n",
        "    # Return the sum of the results of the sub-experiments.\n",
        "    return sum(ray.get(sub_results))\n",
        "\n",
        "results = [run_experiment.remote(i) for i in range(5)]\n",
        "ray.get(results) # [45, 55, 65, 75, 85]"
      ],
      "execution_count": 15,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "2019-05-12 14:56:09,395\tERROR worker.py:1672 -- WARNING: 6 workers have been started. This could be a result of using a large number of actors, or it could be a consequence of using nested tasks (see https://github.com/ray-project/ray/issues/3644) for some a discussion of workarounds.\n",
            "2019-05-12 14:56:10,855\tERROR worker.py:1672 -- WARNING: 8 workers have been started. This could be a result of using a large number of actors, or it could be a consequence of using nested tasks (see https://github.com/ray-project/ray/issues/3644) for some a discussion of workarounds.\n"
          ],
          "name": "stderr"
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[45, 55, 65, 75, 85]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 15
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "wcRCFse2dnet",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "70bdfd06-3002-47be-f146-7c359db42eb3"
      },
      "source": [
        "ray.get_gpu_ids()\n"
      ],
      "execution_count": 16,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 16
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7F848iOHcu6G",
        "colab_type": "text"
      },
      "source": [
        "# The Ray API"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Mr6f6HzXcWId",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        ""
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Gfx75U1tc-JO",
        "colab_type": "text"
      },
      "source": [
        "*\n",
        "\n",
        "*   **ray.init()** - Connect to an existing Ray cluster or start one and connect to it.\n",
        "*   **ray.remote()** - Define a remote function or an actor class.\n",
        "*  **ray.get()** - Get a remote object or a list of remote objects from the object store.\n",
        "*  **ray.wait()** - Return a list of IDs that are ready and a list of IDs that are not.\n",
        "*  **ray.put()** - Store an object in the object store.\n",
        "*  **ray.get_gpu_ids()** - Get the IDs of the GPUs that are available to the worker.\n",
        "*  **ray.get_resource_ids()** - Get the IDs of the resources that are available to the worker.\n",
        "*  **ray.get_webui_url()** - \n",
        " Get the URL to access the web UI.\n",
        "* **ray.shutdown()** - Disconnect the worker, and terminate processes started by ray.init().\n",
        "*  **ray.register_custom_serializer()** -- Enable serialization and deserialization for a particular class.\n",
        "*  **ray.profile()** - Profile a span of time so that it appears in the timeline visualization.\n",
        "*  **ray.method()** - Annotate an actor method.|\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "v-5jIOoyeTRc",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        ""
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "PYoS62Saelpv",
        "colab_type": "text"
      },
      "source": [
        "# Actors"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zcQQCVa5eraP",
        "colab_type": "text"
      },
      "source": [
        "## Defining and creating an actor"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "wOgW7zvkenmy",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "@ray.remote\n",
        "class Counter(object):\n",
        "    def __init__(self):\n",
        "        self.value = 0\n",
        "\n",
        "    def increment(self):\n",
        "        self.value += 1\n",
        "        return self.value"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "i6fsaDWXexlK",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "a1 = Counter.remote()\n",
        "a2 = Counter.remote()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "m0HGviNye3sj",
        "colab_type": "text"
      },
      "source": [
        "## Using an actor"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "gzTjhPBje5tl",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "a9732d0d-c299-43a0-f762-4c011609b8f8"
      },
      "source": [
        "a1.increment.remote()  # ray.get returns 1\n",
        "a2.increment.remote()  # ray.get returns 1"
      ],
      "execution_count": 19,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "ObjectID(01000000645ffd24f9db37e98af8eb11de018325)"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 19
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IAKV-iWwgQjn",
        "colab_type": "text"
      },
      "source": [
        "We can then call ray.get on the object ID to retrieve the actual value.\n",
        "\n",
        "Similarly, the call to a2.increment.remote() generates a task that is scheduled on the second Counter actor. Since these two tasks run on different actors, they can be executed in parallel (note that only actor methods will be scheduled on actor workers, regular remote functions will not be).\n",
        "\n",
        "On the other hand, methods called on the same Counter actor are executed serially in the order that they are called. They can thus share state with one another, as shown below."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "nFUGg-C3e7EE",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 128
        },
        "outputId": "8b73f391-17fd-4d10-ffee-059b90b48263"
      },
      "source": [
        "# Create ten Counter actors.\n",
        "counters = [Counter.remote() for _ in range(10)]\n",
        "\n",
        "# Increment each Counter once and get the results. These tasks all happen in\n",
        "# parallel.\n",
        "results = ray.get([c.increment.remote() for c in counters])\n",
        "print(results)  # prints [1, 1, 1, 1, 1, 1, 1, 1, 1, 1]\n",
        "\n",
        "# Increment the first Counter five times. These tasks are executed serially\n",
        "# and share state.\n",
        "results = ray.get([counters[0].increment.remote() for _ in range(5)])\n",
        "print(results)  # prints [2, 3, 4, 5, 6]"
      ],
      "execution_count": 20,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "2019-05-12 14:56:39,797\tERROR worker.py:1672 -- WARNING: 10 workers have been started. This could be a result of using a large number of actors, or it could be a consequence of using nested tasks (see https://github.com/ray-project/ray/issues/3644) for some a discussion of workarounds.\n",
            "2019-05-12 14:56:40,216\tERROR worker.py:1672 -- WARNING: 12 workers have been started. This could be a result of using a large number of actors, or it could be a consequence of using nested tasks (see https://github.com/ray-project/ray/issues/3644) for some a discussion of workarounds.\n",
            "2019-05-12 14:56:40,946\tERROR worker.py:1672 -- WARNING: 14 workers have been started. This could be a result of using a large number of actors, or it could be a consequence of using nested tasks (see https://github.com/ray-project/ray/issues/3644) for some a discussion of workarounds.\n"
          ],
          "name": "stderr"
        },
        {
          "output_type": "stream",
          "text": [
            "[1, 1, 1, 1, 1, 1, 1, 1, 1, 1]\n",
            "[2, 3, 4, 5, 6]\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Es0sbWc0gnMN",
        "colab_type": "text"
      },
      "source": [
        "## A More Interesting Actor Example"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "W87MhNbRgl9z",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import gym\n",
        "\n",
        "@ray.remote\n",
        "class GymEnvironment(object):\n",
        "    def __init__(self, name):\n",
        "        self.env = gym.make(name)\n",
        "        self.env.reset()\n",
        "\n",
        "    def step(self, action):\n",
        "        return self.env.step(action)\n",
        "\n",
        "    def reset(self):\n",
        "        self.env.reset()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "g6nu4HEVhCQi",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "6f0bf154-2c11-44be-f57d-4a4d88d6a8ef"
      },
      "source": [
        "pong = GymEnvironment.remote(\"Pong-v0\")\n",
        "pong.step.remote(0)  # Take action 0 in the simulator."
      ],
      "execution_count": 22,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "ObjectID(010000005e4a676537735cc0c1cad35f732c372f)"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 22
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0j4j6khyhNUh",
        "colab_type": "text"
      },
      "source": [
        "## Using GPUs on actors"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "oWRRSqXBhMhp",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import tensorflow as tf\n",
        "\n",
        "def construct_network():\n",
        "    x = tf.placeholder(tf.float32, [None, 784])\n",
        "    y_ = tf.placeholder(tf.float32, [None, 10])\n",
        "\n",
        "    W = tf.Variable(tf.zeros([784, 10]))\n",
        "    b = tf.Variable(tf.zeros([10]))\n",
        "    y = tf.nn.softmax(tf.matmul(x, W) + b)\n",
        "\n",
        "    cross_entropy = tf.reduce_mean(-tf.reduce_sum(y_ * tf.log(y), reduction_indices=[1]))\n",
        "    train_step = tf.train.GradientDescentOptimizer(0.5).minimize(cross_entropy)\n",
        "    correct_prediction = tf.equal(tf.argmax(y,1), tf.argmax(y_,1))\n",
        "    accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))\n",
        "\n",
        "    return x, y_, train_step, accuracy"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "c09nhhGlhM2q",
        "colab_type": "text"
      },
      "source": [
        "We can then define an actor for this network as follows."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "bZQV-KBbhJeq",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "@ray.remote(num_gpus=1)\n",
        "class NeuralNetOnGPU(object):\n",
        "    def __init__(self, mnist_data):\n",
        "        self.mnist = mnist_data\n",
        "        # Set an environment variable to tell TensorFlow which GPUs to use. Note\n",
        "        # that this must be done before the call to tf.Session.\n",
        "        os.environ[\"CUDA_VISIBLE_DEVICES\"] = \",\".join([str(i) for i in ray.get_gpu_ids()])\n",
        "        with tf.Graph().as_default():\n",
        "            with tf.device(\"/gpu:0\"):\n",
        "                self.x, self.y_, self.train_step, self.accuracy = construct_network()\n",
        "                # Allow this to run on CPUs if there aren't any GPUs.\n",
        "                config = tf.ConfigProto(allow_soft_placement=True)\n",
        "                self.sess = tf.Session(config=config)\n",
        "                # Initialize the network.\n",
        "                init = tf.global_variables_initializer()\n",
        "                self.sess.run(init)\n",
        "\n",
        "    def train(self, num_steps):\n",
        "        for _ in range(num_steps):\n",
        "            batch_xs, batch_ys = self.mnist.train.next_batch(100)\n",
        "            self.sess.run(self.train_step, feed_dict={self.x: batch_xs, self.y_: batch_ys})\n",
        "\n",
        "    def get_accuracy(self):\n",
        "        return self.sess.run(self.accuracy, feed_dict={self.x: self.mnist.test.images,\n",
        "                                                       self.y_: self.mnist.test.labels})"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BKEBQMvniQPz",
        "colab_type": "text"
      },
      "source": [
        "To indicate that an actor requires one GPU, we pass in num_gpus=1 to ray.remote. Note that in order for this to work, Ray must have been started with some GPUs, e.g., via ray.init(num_gpus=2). Otherwise, when you try to instantiate the GPU version with NeuralNetOnGPU.remote(), an exception will be thrown saying that there aren’t enough GPUs in the system.\n",
        "\n",
        "When the actor is created, it will have access to a list of the IDs of the GPUs that it is allowed to use via ray.get_gpu_ids(). This is a list of integers, like [], or [1], or [2, 5, 6]. Since we passed in ray.remote(num_gpus=1), this list will have length one."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "1pxkFmjzhjJP",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 855
        },
        "outputId": "58554e52-b034-45f3-b852-3a4688ec4501"
      },
      "source": [
        "import os\n",
        "import ray\n",
        "import tensorflow as tf\n",
        "from tensorflow.examples.tutorials.mnist import input_data\n",
        "\n",
        "# Load the MNIST dataset and tell Ray how to serialize the custom classes.\n",
        "mnist = input_data.read_data_sets(\"MNIST_data\", one_hot=True)\n",
        "\n",
        "# Create the actor.\n",
        "nn = NeuralNetOnGPU.remote(mnist)\n",
        "\n",
        "# Run a few steps of training and print the accuracy.\n",
        "nn.train.remote(100)\n",
        "accuracy = ray.get(nn.get_accuracy.remote())\n",
        "print(\"Accuracy is {}.\".format(accuracy))\n",
        "\n"
      ],
      "execution_count": 26,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "WARNING:tensorflow:From <ipython-input-26-31eeb2b0c429>:7: read_data_sets (from tensorflow.contrib.learn.python.learn.datasets.mnist) is deprecated and will be removed in a future version.\n",
            "Instructions for updating:\n",
            "Please use alternatives such as official/mnist/dataset.py from tensorflow/models.\n",
            "WARNING:tensorflow:From /usr/local/lib/python3.6/dist-packages/tensorflow/contrib/learn/python/learn/datasets/mnist.py:260: maybe_download (from tensorflow.contrib.learn.python.learn.datasets.base) is deprecated and will be removed in a future version.\n",
            "Instructions for updating:\n",
            "Please write your own downloading logic.\n",
            "WARNING:tensorflow:From /usr/local/lib/python3.6/dist-packages/tensorflow/contrib/learn/python/learn/datasets/base.py:252: _internal_retry.<locals>.wrap.<locals>.wrapped_fn (from tensorflow.contrib.learn.python.learn.datasets.base) is deprecated and will be removed in a future version.\n",
            "Instructions for updating:\n",
            "Please use urllib or similar directly.\n",
            "Successfully downloaded train-images-idx3-ubyte.gz 9912422 bytes.\n",
            "WARNING:tensorflow:From /usr/local/lib/python3.6/dist-packages/tensorflow/contrib/learn/python/learn/datasets/mnist.py:262: extract_images (from tensorflow.contrib.learn.python.learn.datasets.mnist) is deprecated and will be removed in a future version.\n",
            "Instructions for updating:\n",
            "Please use tf.data to implement this functionality.\n",
            "Extracting MNIST_data/train-images-idx3-ubyte.gz\n",
            "Successfully downloaded train-labels-idx1-ubyte.gz 28881 bytes.\n",
            "WARNING:tensorflow:From /usr/local/lib/python3.6/dist-packages/tensorflow/contrib/learn/python/learn/datasets/mnist.py:267: extract_labels (from tensorflow.contrib.learn.python.learn.datasets.mnist) is deprecated and will be removed in a future version.\n",
            "Instructions for updating:\n",
            "Please use tf.data to implement this functionality.\n",
            "Extracting MNIST_data/train-labels-idx1-ubyte.gz\n",
            "WARNING:tensorflow:From /usr/local/lib/python3.6/dist-packages/tensorflow/contrib/learn/python/learn/datasets/mnist.py:110: dense_to_one_hot (from tensorflow.contrib.learn.python.learn.datasets.mnist) is deprecated and will be removed in a future version.\n",
            "Instructions for updating:\n",
            "Please use tf.one_hot on tensors.\n",
            "Successfully downloaded t10k-images-idx3-ubyte.gz 1648877 bytes.\n",
            "Extracting MNIST_data/t10k-images-idx3-ubyte.gz\n",
            "Successfully downloaded t10k-labels-idx1-ubyte.gz 4542 bytes.\n",
            "Extracting MNIST_data/t10k-labels-idx1-ubyte.gz\n",
            "WARNING:tensorflow:From /usr/local/lib/python3.6/dist-packages/tensorflow/contrib/learn/python/learn/datasets/mnist.py:290: DataSet.__init__ (from tensorflow.contrib.learn.python.learn.datasets.mnist) is deprecated and will be removed in a future version.\n",
            "Instructions for updating:\n",
            "Please use alternatives such as official/mnist/dataset.py from tensorflow/models.\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m WARNING:tensorflow:From /usr/local/lib/python3.6/dist-packages/tensorflow/python/framework/op_def_library.py:263: colocate_with (from tensorflow.python.framework.ops) is deprecated and will be removed in a future version.\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m Instructions for updating:\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m Colocations handled automatically by placer.\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m WARNING:tensorflow:From /usr/local/lib/python3.6/dist-packages/tensorflow/python/ops/math_ops.py:3066: to_int32 (from tensorflow.python.ops.math_ops) is deprecated and will be removed in a future version.\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m Instructions for updating:\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m Use tf.cast instead.\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m 2019-05-12 14:59:07.008038: I tensorflow/core/platform/profile_utils/cpu_utils.cc:94] CPU Frequency: 2200000000 Hz\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m 2019-05-12 14:59:07.008388: I tensorflow/compiler/xla/service/service.cc:150] XLA service 0x724b1e0 executing computations on platform Host. Devices:\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m 2019-05-12 14:59:07.008423: I tensorflow/compiler/xla/service/service.cc:158]   StreamExecutor device (0): <undefined>, <undefined>\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m 2019-05-12 14:59:07.036873: E tensorflow/stream_executor/cuda/cuda_driver.cc:300] failed call to cuInit: CUDA_ERROR_NO_DEVICE: no CUDA-capable device is detected\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m 2019-05-12 14:59:07.036959: I tensorflow/stream_executor/cuda/cuda_diagnostics.cc:161] retrieving CUDA diagnostic information for host: c2a8030faeba\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m 2019-05-12 14:59:07.036972: I tensorflow/stream_executor/cuda/cuda_diagnostics.cc:168] hostname: c2a8030faeba\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m 2019-05-12 14:59:07.037039: I tensorflow/stream_executor/cuda/cuda_diagnostics.cc:192] libcuda reported version is: 410.79.0\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m 2019-05-12 14:59:07.037074: I tensorflow/stream_executor/cuda/cuda_diagnostics.cc:196] kernel reported version is: 410.79.0\n",
            "\u001b[2m\u001b[36m(pid=545)\u001b[0m 2019-05-12 14:59:07.037083: I tensorflow/stream_executor/cuda/cuda_diagnostics.cc:303] kernel version seems to match DSO: 410.79.0\n",
            "Accuracy is 0.8944000005722046.\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "VqdTPyPziceU",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "@ray.remote(num_gpus=1)\n",
        "class GPUActor(object):\n",
        "    def __init__(self):\n",
        "        return \"This actor is allowed to use GPUs {}.\".format(ray.get_gpu_ids())\n",
        "    def test(self):\n",
        "        return \"This actor is allowed to use GPUs {}.\".format(ray.get_gpu_ids())\n"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GydAYJxxjmKz",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "@ray.remote(num_gpus=1)\n",
        "class GPUActor(object):\n",
        "    def __init__(self):\n",
        "        self.gpu_ids = ray.get_gpu_ids()\n",
        "        os.environ[\"CUDA_VISIBLE_DEVICES\"] = \",\".join(map(str, self.gpu_ids))\n",
        "        # The call to tf.Session() will restrict TensorFlow to use the GPUs\n",
        "        # specified in the CUDA_VISIBLE_DEVICES environment variable.\n",
        "        self.sess = tf.Session()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "LA23FUnpljlE",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        ""
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "r1zNI0I3kejD",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "actor = GPUActor.remote()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "cijGHhAvlkY2",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 35
        },
        "outputId": "5c56ec70-297f-4eb7-c033-5978e16f8b5b"
      },
      "source": [
        "ray.get(actor.test.remote())"
      ],
      "execution_count": 7,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "'This actor is allowed to use GPUs [7].'"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 7
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "zDa_ZZMstktB",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        ""
      ],
      "execution_count": 0,
      "outputs": []
    }
  ]
}