ray_tutorial_and_setup.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "WARNING: Not monitoring node memory since `psutil` is not installed. Install this with `pip install psutil` (or ray[debug]) to enable debugging of memory-related crashes.\n"
     ]
    }
   ],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import ray\n",
    "import time\n",
    "import numpy as np\n",
    "import pickle"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2019-04-15 00:19:17,980\tWARNING worker.py:1406 -- 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-04-15 00:19:17,983\tINFO node.py:423 -- Process STDOUT and STDERR is being redirected to /tmp/ray/session_2019-04-15_00-19-17_9021/logs.\n",
      "2019-04-15 00:19:18,096\tINFO services.py:363 -- Waiting for redis server at 127.0.0.1:64023 to respond...\n",
      "2019-04-15 00:19:18,231\tINFO services.py:363 -- Waiting for redis server at 127.0.0.1:36089 to respond...\n",
      "2019-04-15 00:19:18,236\tINFO services.py:760 -- Starting Redis shard with 1.0 GB max memory.\n",
      "2019-04-15 00:19:18,278\tINFO services.py:1384 -- Starting the Plasma object store with 10.0 GB memory using /dev/shm.\n",
      "2019-04-15 00:19:18,299\tWARNING services.py:863 -- Failed to start the reporter. The reporter requires 'pip install psutil'.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'node_ip_address': None,\n",
       " 'redis_address': '10.9.1.17:64023',\n",
       " 'object_store_address': '/tmp/ray/session_2019-04-15_00-19-17_9021/sockets/plasma_store',\n",
       " 'webui_url': None,\n",
       " 'raylet_socket_name': '/tmp/ray/session_2019-04-15_00-19-17_9021/sockets/raylet'}"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ray.init(num_cpus=4, include_webui=False, ignore_reinit_error=True, redis_max_memory=1000000000, object_store_memory=10000000000)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def slow_function(i):\n",
    "    time.sleep(1)\n",
    "    return i"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The results are [0, 1, 2, 3]. This took 1.006211280822754 seconds. Run the next cell to see if the exercise was done correctly.\n"
     ]
    }
   ],
   "source": [
    "time.sleep(10.0)\n",
    "start_time = time.time()\n",
    "\n",
    "results = [slow_function.remote(i) for i in range(4)]\n",
    "results = ray.get(results)\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))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Success! The example took 1.006211280822754 seconds.\n"
     ]
    }
   ],
   "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": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def load_data(filename):\n",
    "    time.sleep(0.1)\n",
    "    return np.ones((1000, 100))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def normalize_data(data):\n",
    "    time.sleep(0.1)\n",
    "    return data - np.mean(data, axis=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def extract_features(normalized_data):\n",
    "    time.sleep(0.1)\n",
    "    return np.hstack([normalized_data, normalized_data ** 2])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\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)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "\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": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "time.sleep(2.0)\n",
    "start_time = time.time()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "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.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The losses are [1000.0, 1000.0, 1000.0, 1000.0].\n",
      "\n",
      "The loss is 4000.0. This took 0.5383255481719971 seconds. Run the next cell to see if the exercise was done correctly.\n"
     ]
    }
   ],
   "source": [
    "losses = ray.get(losses)\n",
    "print('The losses are {}.'.format(losses) + '\\n')\n",
    "loss = sum(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": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Success! The example took 0.5383255481719971 seconds.\n"
     ]
    }
   ],
   "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": "code",
   "execution_count": 15,
   "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"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert hasattr(Foo, 'remote'), 'You need to turn \"Foo\" into an actor with @ray.remote.'\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "f1 = Foo.remote()\n",
    "f2 = Foo.remote()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "time.sleep(2.0)\n",
    "start_time = time.time()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ObjectID(01000000e27d741a8030458ed0a9f5a4aea7daf7)"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "f1.reset.remote()\n",
    "f2.reset.remote()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = []\n",
    "for _ in range(5):\n",
    "    results.append(f1.increment.remote())\n",
    "    results.append(f2.increment.remote())\n",
    "results = ray.get(results)\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "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": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Success! The example took 2.5290303230285645 seconds.\n"
     ]
    }
   ],
   "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": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def f(i):\n",
    "    np.random.seed(5 + i)\n",
    "    x = np.random.uniform(0, 4)\n",
    "    time.sleep(x)\n",
    "    return i, time.time()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "time.sleep(2.0)\n",
    "start_time = time.time()\n",
    "result_ids = [f.remote(i) for i in range(6)]\n",
    "initial_results,remaining_results = ray.wait(result_ids,num_returns=3,timeout=None)\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time\n",
    "initial_results = ray.get(initial_results)\n",
    "remaining_results = ray.get(remaining_results)\n",
    "assert len(initial_results) == 3\n",
    "assert len(remaining_results) == 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Success! The example took 1.1988871097564697 seconds.\n"
     ]
    }
   ],
   "source": [
    "initial_indices = [result[0] for result in initial_results]\n",
    "initial_times = [result[1] for result in initial_results]\n",
    "remaining_indices = [result[0] for result in remaining_results]\n",
    "remaining_times = [result[1] for result in remaining_results]\n",
    "\n",
    "assert set(initial_indices + remaining_indices) == set(range(6))\n",
    "\n",
    "assert duration < 1.5, ('The initial batch of ten tasks was retrieved in '\n",
    "                        '{} seconds. This is too slow.'.format(duration))\n",
    "\n",
    "assert duration > 0.8, ('The initial batch of ten tasks was retrieved in '\n",
    "                        '{} seconds. This is too slow.'.format(duration))\n",
    "\n",
    "# Make sure the initial results actually completed first.\n",
    "assert max(initial_times) < min(remaining_times)\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "neural_net_weights = {'variable{}'.format(i): np.random.normal(size=1000000)\n",
    "                      for i in range(50)}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Ray - serializing\n",
      "time:  0.1253950595855713\n",
      "\n",
      "Ray - deserializing\n",
      "time:  0.000698089599609375\n",
      "\n",
      "pickle - serializing\n",
      "time:  0.37827324867248535\n",
      "\n",
      "pickle - deserializing\n",
      "time:  0.17782354354858398\n"
     ]
    }
   ],
   "source": [
    "print('Ray - serializing')\n",
    "start = time.time()\n",
    "x_id = ray.put(neural_net_weights)\n",
    "print('time: ', time.time() - start)\n",
    "print('\\nRay - deserializing')\n",
    "start = time.time()\n",
    "x_val = ray.get(x_id)\n",
    "print('time: ', time.time() - start)\n",
    "\n",
    "print('\\npickle - serializing')\n",
    "start = time.time()\n",
    "serialized = pickle.dumps(neural_net_weights)\n",
    "print('time: ', time.time() - start)\n",
    "print('\\npickle - deserializing')\n",
    "start = time.time()\n",
    "deserialized = pickle.loads(serialized)\n",
    "print('time: ', time.time() - start)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def use_weights(weights, i):\n",
    "    return i"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "time.sleep(2.0)\n",
    "start_time = time.time()\n",
    "\n",
    "neural_net_weights_id = ray.put(neural_net_weights)\n",
    "results = ray.get([use_weights.remote(neural_net_weights_id, i)\n",
    "                   for i in range(20)])\n",
    "\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Success! The example took 0.11236715316772461 seconds.\n"
     ]
    }
   ],
   "source": [
    "assert results == list(range(20))\n",
    "assert duration < 1, ('The experiments ran in {} seconds. This is too '\n",
    "                      'slow.'.format(duration))\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (Intel, 2019 update 2)",
   "language": "python",
   "name": "c009-intel_distribution_of_python_3_2019u2"
  },
  "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}