Tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Twisted up in a distributed tornado - a beginners guide to async frameworks in python\n",
    "\n",
    "Andrew Stretton\n",
    "\n",
    "Software Engineer @ https://zegami.com\n",
    "\n",
    "\n",
    "* astretton@zegami.com\n",
    "* https://github.com/strets123\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Setup:\n",
    "\n",
    "Install MongoDB and a few packages\n",
    "\n",
    "https://www.mongodb.com/download-center#community\n",
    "\n",
    "conda install distributed luigi\n",
    "\n",
    "pip install aiohttp  timeout-decorator motor\n",
    "\n",
    "Slides are here:\n",
    "\n",
    "https://github.com/strets123/beginners-async-python\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "\n",
    "# What are we going to cover???\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Use all of your processors when running a script\n",
    "\n",
    "We will demonstrate how to max out your processors and network by running an image processing algorithm over this dataset:\n",
    "\n",
    "https://demo.zegami.com/ashmolean%20paintings"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Consume social media data as it is created\n",
    "\n",
    "We will demonstrate how to read the wikipedia changes feed \n",
    "\n",
    "https://stream.wikimedia.org/v2/stream/recentchange\n",
    "\n",
    "## Write to a database and move on to the next task without waiting for confirmation\n",
    "\n",
    "We will save the events to MongoDB"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Create a super simple task scheduler without an ugly while True loop\n",
    "\n",
    "We will create a simple clock on the terminal and perhaps go on to make it work on a web page"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Call a function and throw an exception if it takes too long\n",
    "\n",
    "We will demonstrate how tricky this is and go through all the options"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Understand how Jupyter notebooks work\n",
    "\n",
    "We will look at the Jupyter architecture and then how we can reproduce parts of it with websockets and Tornado"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## First a little theory... definitions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Concurrency / Parallelism\n",
    "\n",
    "Running more than one piece of code at one time\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "\n",
    "### Asynchronous\n",
    "\n",
    "* Used to present the impression of concurrent or parallel tasking \n",
    "\n",
    "* For a process that needs to do work away from the current application \n",
    "\n",
    "* Where we don't want to wait and block our application awaiting the response.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "\n",
    "### Futures\n",
    " \n",
    "Traditional thread and process handling requires a \"join\" step to bring the processes or threads back together. Arguably more elegant is to use the concept of `Futures`.\n",
    "\n",
    "Called Promises in JavaScipt, Futures are an IOU for the result of a function.\n",
    "\n",
    "![](http://sharedstory.org/sites/default/files/stories/user-154/story_profile_pic.jpg)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Executors\n",
    "\n",
    "Executors are threads or processes that exist separately to the task that is being run. \n",
    "\n",
    "You can think of an executor as being like a road. \n",
    "\n",
    "![](https://encrypted-tbn0.gstatic.com/images?q=tbn:ANd9GcSgvO4FY_kEwF6-BmNH7BoZsordtULkTa-SIkznhhnXD14uGUd8Pg)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## It can carry all kinds of vehicule (task) to where they need to go. \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## It has numbers of lanes which determines its total capacity.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## It does not itself change state or move around, it is still there when the cars have all driven through\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Two types of Executor in the standard library \n",
    "\n",
    "* `ThreadPoolExecutor` which manages threads and `ProcessPoolExecutor` which manages processes.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Here we use a ThreadPoolExecutor to open some threads and then we run a single function on one of those threads."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "False\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ThreadPoolExecutor\n",
    "from time import sleep\n",
    "\n",
    "def return_after_5_secs(message):\n",
    "    sleep(5)\n",
    "    return message\n",
    "\n",
    "pool = ThreadPoolExecutor(3)  #The ThreadPoolExecutor manages the 3 open threads\n",
    "\n",
    "future = pool.submit(return_after_5_secs, (\"hello\"))\n",
    "print(future.done())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "True\n",
      "hello\n"
     ]
    }
   ],
   "source": [
    "print(future.done())\n",
    "print(future.result())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "After waiting for a bit the task is now done. We have not had to do anything in our business logic to say that processing is complete, `future.done()` is handled for us."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Executor.map for processing lists of data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Executor.map allows us to spread a list of data points over ProcessPool and run a function over them."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "112272535095293 is prime: True\n",
      "112582705942171 is prime: True\n",
      "2897 is prime: True\n",
      "112272535095293 is prime: True\n",
      "115280095190773 is prime: True\n",
      "115797848077099 is prime: True\n",
      "1099726899285419 is prime: False\n"
     ]
    }
   ],
   "source": [
    "import concurrent.futures\n",
    "import math\n",
    " \n",
    "PRIMES = [\n",
    "    112272535095293,\n",
    "    112582705942171,\n",
    "    112272535095293,\n",
    "    115280095190773,\n",
    "    115797848077099,\n",
    "    1099726899285419,\n",
    "    ]\n",
    " \n",
    "def is_prime(n):\n",
    "    if n % 2 == 0:\n",
    "        return False\n",
    " \n",
    "    sqrt_n = int(math.floor(math.sqrt(n)))\n",
    "    for i in range(3, sqrt_n + 1, 2):\n",
    "        if n % i == 0:\n",
    "            return False\n",
    "    return True\n",
    " \n",
    "def main():\n",
    "    with concurrent.futures.ProcessPoolExecutor() as executor:\n",
    "        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):\n",
    "            print('%d is prime: %s' % (number, prime))\n",
    "main()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### as_completed\n",
    "\n",
    "The `as_completed` function allows us to iterate a list of futures as they are completed.\n",
    "\n",
    "![](https://encrypted-tbn2.gstatic.com/images?q=tbn:ANd9GcRXO5DbttPr5NFffK9KQnNzOaxCKlbz0EVd0-2qG0hV9cQafkgB)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Return of 0\n",
      "Return of 1\n",
      "Return of 4\n",
      "Return of 3\n",
      "Return of 2\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ThreadPoolExecutor, wait, as_completed\n",
    "from time import sleep\n",
    "from random import randint\n",
    " \n",
    "def return_after_5_secs(num):\n",
    "    sleep(randint(1, 5))\n",
    "    return \"Return of {}\".format(num)\n",
    " \n",
    "pool = ThreadPoolExecutor(5)\n",
    "futures = []\n",
    "for x in range(5):\n",
    "    futures.append(pool.submit(return_after_5_secs, x))\n",
    " \n",
    "for x in as_completed(futures):\n",
    "    print(x.result())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Excersise 1: t-SNE Analysis of an art gallery dataset\n",
    "\n",
    "[Zegami](https://demo.zegami.com) is used to tag and analyse numerous image-centric data sets. \n",
    "\n",
    "In this example we use Zegami help museums to annotate and digitise paintings. One option for clusting images is the t-SNE (t-distributed stochastic neighbor embedding) algorithm as implemented in Scikit Learn. More about the algorithm here: https://www.youtube.com/watch?v=RJVL80Gg3lA. The steps involved in the data processing are:\n",
    "\n",
    "1. Download and parse a CSV file with the image IDs\n",
    "1. Download the images \n",
    "1. Resize the images and convert into linear arrays of pixels\n",
    "1. Perform t-SNE on the list of arrays to give an X and Y coordinate for each image\n",
    "\n",
    "At http://localhost:8888/edit/ashmolean.py is some single-threaded example code. In this tutorial we are going to first rewrite the code to run in a `ProcessPoolExecutor`. We will later optimise the http code using async python."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "==========================\n",
      "Appended x and y coordinates and created zegami_tsne.tab.\n",
      "==========================\n"
     ]
    }
   ],
   "source": [
    "TTF_URL = \"http://tutorial.zegami.net.s3.eu-west-2.amazonaws.com/ashmolean/data.tab\"\n",
    "\n",
    "IMAGE_URL = \"http://tutorial.zegami.net.s3.eu-west-2.amazonaws.com/ashmolean/{}.jpg\"\n",
    "\n",
    "# TSNE on images\n",
    "import os\n",
    "import io\n",
    "import requests\n",
    "import time\n",
    "from PIL import Image\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from sklearn.manifold import TSNE\n",
    "from distributed.client import Client,   wait\n",
    "from distributed import worker_client, LocalCluster\n",
    "from concurrent.futures import ThreadPoolExecutor, as_completed\n",
    "pd.set_option('display.max_columns', None)\n",
    "pd.set_option('display.max_rows', None)\n",
    "output_file = \"zegami_tsne.tab\"\n",
    "\n",
    "# setup a standard image size; this will distort some images but will get everything into the same shape\n",
    "STANDARD_SIZE = (100, 100)\n",
    "\n",
    "\n",
    "def img_data_to_matrix(image_data, verbose=False):\n",
    "    \"\"\"\n",
    "    takes a jpeg or oher image bytes object and turns it into a numpy array of RGB pixels\n",
    "    \"\"\"\n",
    "\n",
    "    bio = io.BytesIO(image_data)\n",
    "    bio.seek(0)\n",
    "    img = Image.open(bio)\n",
    "    img = img.resize(STANDARD_SIZE)\n",
    "    img = list(img.getdata())\n",
    "    img = np.array(img)\n",
    "\n",
    "    return img\n",
    "\n",
    "\n",
    "def flatten_image(img):\n",
    "    \"\"\"\n",
    "    takes in an (m, n) numpy array and flattens it\n",
    "    into an array of shape (1, m * n)\n",
    "    \"\"\"\n",
    "    s = img.shape[0] * img.shape[1]\n",
    "    img_wide = img.reshape(1, s)\n",
    "    return img_wide[0]\n",
    "\n",
    "\n",
    "def get_data():\n",
    "    \"\"\"\n",
    "    Requests the CSV data over the web and returns it as a cleaned pandas dataframe\n",
    "    \"\"\"\n",
    "    resp = requests.get(TTF_URL, stream=True)\n",
    "    data = pd.read_csv(io.StringIO(resp.content.decode(\"latin-1\")),\n",
    "                       sep=\"\\t\",\n",
    "                       header=0)\n",
    "    return data.dropna(subset=[\"id\"])\n",
    "\n",
    "\n",
    "def get_pca(data):\n",
    "    \"\"\"\n",
    "    Runs the TSNE algorithm over a set of image data\n",
    "    \"\"\"\n",
    "    pca = TSNE()\n",
    "\n",
    "    X = pca.fit_transform(data)\n",
    "    results = pd.DataFrame({\"x\": X[:, 0], \"y\": X[:, 1]})\n",
    "    return results\n",
    "\n",
    "\n",
    "def image_data_from_id(image_id):\n",
    "    \"\"\"\n",
    "    Take the image id and return the bytes content of the image via an http request\n",
    "    \"\"\"\n",
    "    if str(image_id) == \"nan\":\n",
    "        raise Exception(\"ID should not be nan\")\n",
    "    image_url = IMAGE_URL.format(image_id)\n",
    "    image_data = requests.get(image_url, timeout=60).content\n",
    "    return image_data\n",
    "\n",
    "\n",
    "def process_image_bytes(image_bytes):\n",
    "    \"\"\"\n",
    "    Reshape the image pixels into a flat list of tuples of pixel values. The list should be of equla size for all images\n",
    "    \"\"\"\n",
    "    matrix = img_data_to_matrix(image_bytes)\n",
    "    return flatten_image(matrix)\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "def chunks(l, n):\n",
    "    \"\"\"Yield successive n-sized chunks from l.\"\"\"\n",
    "    for i in range(0, len(l), n):\n",
    "        yield l[i:i + n]\n",
    "\n",
    "\n",
    "def get_tsne(df, data):\n",
    "    \"\"\"\n",
    "    Given a flat list of tuples of pixel values write out the TSNE CSV output file\n",
    "    \"\"\"\n",
    "    results = get_pca(data)\n",
    "    df['x'] = results['x']\n",
    "    df['y'] = results['y']\n",
    "    print(\"==========================\")\n",
    "    # print str(df)\n",
    "    print(\"Appended x and y coordinates and created \" + output_file + \".\")\n",
    "    df.to_csv(output_file, sep=\"\\t\", index=False)\n",
    "    print(\"==========================\")\n",
    "\n",
    "def run():\n",
    "    \"\"\"\n",
    "    main script runner for the synchronous version of this ashmolean data processing script\n",
    "    \"\"\"\n",
    "    df = get_data()[:10]\n",
    "    images = df[\"id\"]\n",
    "    data = []\n",
    "    labels = []\n",
    "    bytes_data = [image_data_from_id(image_id) for image_id in images]\n",
    "    data = [process_image_bytes(image_bytes) for image_bytes in bytes_data]\n",
    "\n",
    "    #pca = TSNE()\n",
    "    # or can do PCA\n",
    "    get_tsne(df, data)\n",
    "\n",
    "run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Review\n",
    "\n",
    "* We are able to run data processing across multiple machines\n",
    "* `concurrent.futures` allows us to do this with little need to change the code\n",
    "* We are able to open a large number of threads in a ThreadPoolExecutor to retrieve all of the images\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### What is asyncronous python???\n",
    "\n",
    "Asynchronous python represents a change to the flow control of a program. As this can be somewhat tricky to understand let's try an analogy..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "#### Frying eggs and making toast for all the family\n",
    "\n",
    "Let's break down the tasks... \n",
    " \n",
    "1. Toast bread\n",
    "1. Butter toast\n",
    "1. Heat up frying pan\n",
    "1. Add oil\n",
    "1. Break eggs into pan\n",
    "1. Wait until the right moment\n",
    "1. Flip the eggs carefully for a few seconds\n",
    "1. Put on more toast\n",
    "1. Make sure everyone is at the table\n",
    "1. Server the eggs on toast"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## What are the possible ways we could do these tasks?\n",
    "\n",
    "### 1) Procedural\n",
    "\n",
    "Let's say my toaster is one of these:\n",
    "\n",
    "![](https://www.picclickimg.com/00/s/MTA2NlgxNjAw/z/RGEAAOSwNSxVXQQw/$/Antique-Westinghouse-Flip-Up-Chrome-Toaster-WORKS-ORIG-CORD-_1.jpg)\n",
    "\n",
    "And my kitchen is really badly laid out - in fact there are no sockets in my kitchen and so the toaster is in the dining room.\n",
    "\n",
    "How am I going to get everyone's toast cooked as well as their eggs?\n",
    "\n",
    "One way to avoid burning anything would be:\n",
    "\n",
    "Half cook toast on each side then leave in the toaster to keep warm.\n",
    "\n",
    "Cook eggs\n",
    "\n",
    "Finish cooking toast\n",
    "\n",
    "Combine and serve\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "\n",
    "### 2) Multi process\n",
    "\n",
    "Another way of getting round these constraints is just to ask for help. In this case two of us can get everything done in a coordinated way but it relys on two people being happy to cook the breakfast.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "\n",
    "\n",
    "### 3) Multi threaded\n",
    "\n",
    "In this case I have my toaster next to my frying pan and I can keep my hands and eyes on both at once. In this way I get everything cooked and I can slow down or speed up the eggs to make sure the eggs and the toast are all just right. I use my multiple arms to acheive this. My brain and eyes switch between the two tasks to keep an eye on them.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "\n",
    "\n",
    "### 4) Asynchrony\n",
    "\n",
    "In this case I decide to throw away my old toaster and get a new one which pops up and also detects when the bread is just right. \n",
    "\n",
    "[http://news.bbc.co.uk/1/hi/england/cambridgeshire/4291816.stm]\n",
    "\n",
    "I also get myself the amazing \"Heston Blumenthal BEG800SIL One Degree Precision Poacher\"\n",
    "\n",
    "[https://www.go-electrical.co.uk/sage-heston-blumenthal-beg800sil-the-one-degree-precision-poacher.html?gclid=CJqwia-nk9MCFY8Q0wodLqcL4Q]\n",
    "\n",
    "Now I can simply set off my toast and eggs at the appropriate time and also have time to do some other things, perhaps get out my autonymous vacuum cleaner and send it to trip someone up."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Cooking Breakfast Programatically\n",
    "\n",
    "So what is the programatic equivalent of turning on the toaster then leaving it and waiting for the sound of it popping up?\n",
    "\n",
    "1. Send task to be run which requires no processing on my machine\n",
    "2. Get back an acknowledgement and wait for task completion (we wait using a Future object)\n",
    "3. Do some other stuff in the meantime\n",
    "4. Hear an event and go back to the task to do whatever needs finishing off."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### As discussed above `concurrent.futures` allows me to wait on tasks \n",
    "\n",
    "### Asyncio and Tornado take this one step further by doing things in a single thread"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Asyncio is the async library from the python standard library \n",
    "\n",
    "### Example: Sleep for 8 hours in 10 seconds using coroutine decorator:\n",
    "\n",
    "* Async programming has a different the execution model  \n",
    "* I can \"sleep\" without the CPU waiting\n",
    "* I can hand back control of the single thread while sleeping by using `yield from`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import time\n",
    "\n",
    "t = time.time()\n",
    "\n",
    "@asyncio.coroutine\n",
    "def bar():\n",
    "\n",
    "    yield from asyncio.sleep(8)   # Note that the sleep function myust be asyncronous\"\n",
    "\n",
    "ioloop = asyncio.get_event_loop()\n",
    "tasks = [ioloop.create_task(bar()) for i in range(0,3600)]\n",
    "wait_tasks = asyncio.wait(tasks)\n",
    "ioloop.run_until_complete(wait_tasks)\n",
    "ioloop.close()\n",
    "\n",
    "print(time.time()-t)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Example: Sleep for 8 hours in 10 seconds using the new `async` keyword:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import time\n",
    "\n",
    "t = time.time()\n",
    "\n",
    "\n",
    "async def  bar():\n",
    "\n",
    "    await asyncio.sleep(8)\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "ioloop = asyncio.get_event_loop()\n",
    "tasks = [ioloop.create_task(bar()) for i in range(0,3600)]\n",
    "wait_tasks = asyncio.wait(tasks)\n",
    "ioloop.run_until_complete(wait_tasks)\n",
    "ioloop.close()\n",
    "\n",
    "print(time.time()-t)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### What are the key features of these functions?\n",
    "\n",
    "* The function we execute is an asynchronous coroutine\n",
    "* It is wrapped in a \"Task\" which is a type of \"Future\" - this is a placeholder for the result of the function\n",
    "* The sleep function is also an asynchronous coroutine (time.sleep() would not work here)\n",
    "* The tasks are submitted to an event loop\n",
    "* `yield from` or `await` are used to send back control to the event loop so it can do other things.\n",
    "\n",
    "These feautures will crop up again when we use other asynchronous libraries"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Excersise 2: \n",
    "\n",
    "Implement the sleep for 8 hours function tornado and python 3\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import time\n",
    "from tornado.ioloop import IOLoop\n",
    "from tornado import    gen\n",
    "\n",
    "@gen.coroutine\n",
    "def task():\n",
    "    yield gen.sleep(8)\n",
    "\n",
    "\n",
    "@gen.coroutine\n",
    "def main():\n",
    "    yield [task() for i in range(0,3600)]\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    loop = IOLoop.instance()\n",
    "    loop.run_sync(main)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Excersise 3:\n",
    "\n",
    "Do the same thing with the ThreadPoolExecutor and concurrent.futures"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "from concurrent.futures import ThreadPoolExecutor, as_completed\n",
    "from time import sleep, time\n",
    "\n",
    "def return_after_8_secs(message):\n",
    "    sleep(8)\n",
    "    return message\n",
    "\n",
    "t = time()\n",
    "pool = ThreadPoolExecutor(3600)  #The ThreadPoolExecutor manages the 3 open threads\n",
    "\n",
    "\n",
    "futures = [pool.submit(return_after_8_secs, (i)) for i in range (0,500)]\n",
    "\n",
    "for fut in as_completed(futures):\n",
    "    print(fut.result())\n",
    "\n",
    "print(t - time())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Review\n",
    "\n",
    "* In python 2 we can only break the flow control of the application by using an exception. \n",
    "* Various new language features have been added in python 3 which give us more options.\n",
    "* On linux, creating threads can be faster than using an async function \n",
    "* There is not a large performance difference between waiting for many things in many threads and waiting in one thread asynchronously\n",
    "\n",
    "# Modern CPUs are really good at context switching between threads!\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Lots more tutorials from Yeray Diaz Diaz here:\n",
    "\n",
    "https://github.com/yeraydiazdiaz/asyncio-ftwpd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# We keep using asyncios `event_loop` and Tornado's `IOLoop`... \n",
    "\n",
    "What is an event loop???\n",
    "\n",
    "\"In computer science, the event loop, message dispatcher, message loop, message pump, or run loop is a programming construct that waits for and dispatches events or messages in a program.\"\n",
    "\n",
    "Events were introduced in Python 3 as we have seen, Python 2 does not have anything that can break the flow of a program apart from exceptions. \n",
    "\n",
    "If we first consider a loop that is waiting for only one kind of event, that is simpler to conceptualise... \n",
    " \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Reading Social Media data as it is created\n",
    "\n",
    "From the following URL I can open a connection and then follow every change to Wikipedia.\n",
    " \n",
    "https://stream.wikimedia.org/v2/stream/recentchange\n",
    "\n",
    "We can read data synchonously from the URL like this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import requests\n",
    "url = 'https://stream.wikimedia.org/v2/stream/recentchange'\n",
    "import json\n",
    "r = requests.get(url, stream=True)\n",
    "for line in r.iter_lines(decode_unicode=True):\n",
    "    if line:\n",
    "        if line.startswith(\"data:\"):\n",
    "            datapoint = line[5:]\n",
    "            data = json.loads(datapoint)\n",
    "            print(data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Question... \n",
    "\n",
    "Is this compatible with a Tornado/ Asyncio event loop? Can we use it to send events and have them processed?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Answer:\n",
    "\n",
    "No! The generator response.iter_lines is blocking. Therefore we need to read the http request in a nonblocking way OR use another process/thread."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Non-blocking read with `tornado`\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "from tornado import ioloop, gen, httpclient\n",
    "    \n",
    "@gen.coroutine\n",
    "def main():\n",
    "    http_client = httpclient.AsyncHTTPClient()\n",
    "\n",
    "    def streaming_callback(data):\n",
    "        print(data)\n",
    "\n",
    "    yield http_client.fetch('https://stream.wikimedia.org/v2/stream/recentchange', \n",
    "                            streaming_callback=streaming_callback, \n",
    "                            request_timeout=3600)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    ioloop.IOLoop.instance().run_sync(main)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "OK, so that works but now we are recieving packets that may be partial data points, therefore we need to store state between packets and collect them up."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "from tornado import ioloop, gen, httpclient\n",
    "import json\n",
    "\n",
    "class DataReciever(object):\n",
    "    def __init__(self):\n",
    "        self.partial_data = None\n",
    "\n",
    "    # def data_from_line(self, line):\n",
    "\n",
    "    def streaming_callback(self, input):\n",
    "        data = input.decode(\"utf-8\")\n",
    "        lines = data.splitlines()\n",
    "        for line in lines:\n",
    "            if self.partial_data is not None:\n",
    "                line = self.partial_data + line\n",
    "                self.partial_data = None\n",
    "            if len(line) > 0  and len(line) < 100 and \"data:\".startswith(line[:4]):\n",
    "                self.partial_data = line\n",
    "            if line.startswith(\"data:\"):\n",
    "                try:\n",
    "                    data = json.loads(line[5:])\n",
    "                    print(data)\n",
    "                except (ValueError, IndexError):\n",
    "                    self.partial_data = line\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "@gen.coroutine\n",
    "def main():\n",
    "    http_client = httpclient.AsyncHTTPClient()\n",
    "    dr = DataReciever()\n",
    "    yield http_client.fetch(\n",
    "        'https://stream.wikimedia.org/v2/stream/recentchange',\n",
    "        streaming_callback=dr.streaming_callback,\n",
    "        request_timeout=3600\n",
    "    )\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    ioloop.IOLoop.instance().run_sync(main)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Great! so now we are receiving JSON data in a tornado coroutine, \n",
    "\n",
    "How do we save these in a database?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "from tornado import ioloop, gen, httpclient\n",
    "import json\n",
    "import motor\n",
    "\n",
    "class DataReciever(object):\n",
    "    def __init__(self):\n",
    "        self.partial_data = None\n",
    "        self.partial_chars = None\n",
    "        self.client = motor.motor_tornado.MotorClient()\n",
    "        self.db = self.client['test_database']\n",
    "\n",
    "    # def data_from_line(self, line):\n",
    "    def my_callback(self, result, error):\n",
    "        print('result %s' % repr(result.inserted_id))\n",
    "\n",
    "    def streaming_callback(self, input):\n",
    "\n",
    "        data = input.decode(\"utf-8\")\n",
    "\n",
    "        lines = data.splitlines()\n",
    "        for line in lines:\n",
    "            if self.partial_data is not None:\n",
    "                line = self.partial_data + line\n",
    "                self.partial_data = None\n",
    "            if len(line) > 0  and len(line) < 100 and \"data:\".startswith(line[:4]):\n",
    "                self.partial_data = line\n",
    "            if line.startswith(\"data:\"):\n",
    "                try:\n",
    "                    data = json.loads(line[5:])\n",
    "                    self.db.test_collection.insert_one(data, callback=self.my_callback)\n",
    "                except (ValueError, IndexError):\n",
    "                    self.partial_data = line\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "@gen.coroutine\n",
    "def main():\n",
    "    http_client = httpclient.AsyncHTTPClient()\n",
    "    dr = DataReciever()\n",
    "    yield http_client.fetch(\n",
    "        'https://stream.wikimedia.org/v2/stream/recentchange',\n",
    "        streaming_callback=dr.streaming_callback,\n",
    "        request_timeout=3600\n",
    "    )\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    ioloop.IOLoop.instance().run_sync(main)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "This is all very well but more complex than the requests example.\n",
    "           \n",
    "There are also two callbacks! We were supposed to be writing cleaner code which avoids callbacks. We are also having to handle the raw http data.\n",
    "\n",
    "Additionally, it is harder to debug errors as they will not have a full stack trace, for example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "#tornado example with an error\n",
    "\n",
    "from tornado import ioloop, gen, httpclient\n",
    "import json\n",
    "import motor\n",
    "\n",
    "class DataReciever(object):\n",
    "    def __init__(self):\n",
    "        self.partial_data = None\n",
    "        self.client = motor.motor_tornado.MotorClient()\n",
    "        self.db = self.client['test_database']\n",
    "\n",
    "    # def data_from_line(self, line):\n",
    "    def my_callback(self, result, error):\n",
    "        print('result %s' % repr(result.inserted_id))\n",
    "\n",
    "    def streaming_callback(self, input):\n",
    "        data = input.decode(\"ascii\")\n",
    "        lines = data.splitlines()\n",
    "        for line in lines:\n",
    "            if self.partial_data is not None:\n",
    "                line = self.partial_data + line\n",
    "                self.partial_data = None\n",
    "            if len(line) > 0  and len(line) < 100 and \"data:\".startswith(line[:4]):\n",
    "                self.partial_data = line\n",
    "            if line.startswith(\"data:\"):\n",
    "                try:\n",
    "                    data = json.loads(line[5:])\n",
    "                    self.db.test_collection.insert_one(data, callback=self.my_callback)\n",
    "                except (ValueError, IndexError):\n",
    "                    self.partial_data = line\n",
    "\n",
    "\n",
    "@gen.coroutine\n",
    "def main():\n",
    "    http_client = httpclient.AsyncHTTPClient()\n",
    "    dr = DataReciever()\n",
    "    yield http_client.fetch(\n",
    "        'https://stream.wikimedia.org/v2/stream/recentchange',\n",
    "        streaming_callback=dr.streaming_callback,\n",
    "        request_timeout=3600\n",
    "    )\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    ioloop.IOLoop.instance().run_sync(main)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Enter aiohttp...\n",
    "\n",
    "Asyncio has an http library, we need to install it with pip and the docs are not as good as Tornado but we can make use of some interesting new language features.."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import aiohttp\n",
    "ioloop = asyncio.get_event_loop()\n",
    "async def getdata():\n",
    "     async with aiohttp.ClientSession() as session:\n",
    "         async with session.get('https://stream.wikimedia.org/v2/stream/recentchange') as resp:\n",
    "             async for line in resp.content:\n",
    "                 print(line)\n",
    "ioloop.run_until_complete(getdata())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Excersise 4: Use [the tornado asyncio bridge]( http://www.tornadoweb.org/en/stable/asyncio.html#tornado.platform.asyncio.AsyncIOMainLoop) to write the data to MongoDB using motor.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Async HTTP review\n",
    "\n",
    "We have learned that:\n",
    "\n",
    "* Asyncio and Tornado work well for external HTTP requests and other things where an event is waited for\n",
    "* A wrapper Task class is used to wrap the result of a function in a future\n",
    "* Where blocking code needs to be run, a ProcessPoolExecutor can be used\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Excersise 5: Take this terminal clock example and reimplement it using asyncio or tornado to provide the timer\n",
    "\n",
    "http://www.tornadoweb.org/en/stable/ioloop.html#tornado.ioloop.PeriodicCallback\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "def overprint(stringdata):\n",
    "    print(stringdata , end=\"\\r\")\n",
    "\n",
    "\n",
    "def get_time():\n",
    "    return time.strftime(\"%d-%m-%Y %H:%M:%S\")\n",
    "\n",
    "\n",
    "while True:\n",
    "    start = time.time()\n",
    "    overprint(get_time())\n",
    "    sleep_for = 1 - (start - time.time())\n",
    "    time.sleep(sleep_for)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Running a task and timing out if it takes too long\n",
    "\n",
    "This seemingly simple question is in fact pretty difficult in python. HTTP libraries like aiohttp, and tornado implement timeout functions, but to actually timeout a piece of blocking code there are further difficulties...\n",
    "\n",
    "### With ProcessPoolExecutor and asyncio\n",
    "\n",
    "The first port of call is to combine the `ProcessPoolExecutor` and asyncio event loop then cancel the future. This fails to work correctly as the future cancels and the task is left running."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import time\n",
    "from concurrent.futures import ProcessPoolExecutor\n",
    "\n",
    "def block_then_print():\n",
    "    time.sleep(10)\n",
    "    print(\"done\")\n",
    "\n",
    "@asyncio.coroutine\n",
    "def run_executor_in_loop(loop, ex):\n",
    "\n",
    "    yield from loop.run_in_executor(ex, block_then_print)  # This can be interrupted.\n",
    "\n",
    "ex = ProcessPoolExecutor(2)\n",
    "loop = asyncio.get_event_loop()\n",
    "loop.run_until_complete(asyncio.wait_for(coro(run_executor_in_loop, ex), 1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Timeout tasks with signals (Unix)\n",
    "\n",
    "\n",
    "Using raw processes it is possible to timeout a task by by using signals as demonstrated by this decorator:\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import time\n",
    "import timeout_decorator\n",
    "\n",
    "@timeout_decorator.timeout(5)\n",
    "def mytest():\n",
    "    print (\"Start\")\n",
    "    for i in range(1,10):\n",
    "        time.sleep(1)\n",
    "        print (\"%d seconds have passed\" % i)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    mytest()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Timeout tasks with `subprocess.call`\n",
    "\n",
    "Equally we can also add a timeout when we shell out to a subprocess using `subprocess.run` new in python 3.5.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "from subprocess import STDOUT, check_output, TimeoutExpired\n",
    "try:\n",
    "    output = check_output([\"ping\", \"google.com\"], stderr=STDOUT, timeout=2)\n",
    "except TimeoutExpired as e:\n",
    "    print(e.output.decode(\"utf-8\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Review\n",
    "\n",
    "This would work fine for small scripts but use of signals, shelling out and raw processes are not reccomended in larger projects due to:\n",
    "\n",
    "* Platform differences\n",
    "* Possibility of zombie processes\n",
    "* Lack of control over how many processes are open\n",
    "\n",
    "\n",
    "So how should we proceed???\n",
    "\n",
    "A few of the python task fromeworks suggest that they support task cancellation but upon closer inspection `Celery` and `Dask/Distributed` only support cancellation before execution starts not during execution.\n",
    "\n",
    "Often people may want to check if the task needs cancelling during execution, dask/distributed will check between subtasks but a timeout means killing the task dead!\n",
    "\n",
    "## Luigi to the rescue!\n",
    "\n",
    "Luigi also uses some tornado and event loops for communication but also helps keep track of subprocesses within workers. This gives all the flexibility of the subprocess example above along with the stability of a proper data processing framework.\n",
    "\n",
    "    import luigi\n",
    "    from time import sleep\n",
    "    class MyTask(luigi.Task):\n",
    "        worker_timeout = 5\n",
    "        def run(self):\n",
    "            sleep(10)\n",
    "            return luigi.LocalTarget(\"result\", \"done\")  \n",
    "            \n",
    "The task can be run like this:\n",
    "\n",
    "    PTHYONPATH=$(pwd) luigi --module luigi_timeout MyTask --local-scheduler --workers=\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Back to our image processing task\n",
    "\n",
    "We now seem to have the building blocks that we need to request our image data asynchonously and process it synchronously:\n",
    "\n",
    "1. Get the CSV file as a dataframe\n",
    "1. Start a tornado event loop\n",
    "1. Send HTTP requests to get the image data using tornado\n",
    "1. As the futures are completed, submit the data to a ProcessPoolExecutor\n",
    "\n",
    "BUT... we don't really want tornado to be driving the whole app, quite fiddly to do, perhaps there is a framwork that gives us access to the ioloop from a separate thread..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Learnings so far\n",
    "\n",
    "### Good things:\n",
    "\n",
    "* All of our attempts at running the image processing algorithm were faster than doing it synchronously\n",
    "* The simple combination of `ThreadPoolExecutor` for HTTP requests and `ProcessPoolExecutor` for data processing worked fine\n",
    "* Aiohttp or Requests can both be used succesfully for streaming endpoints\n",
    "* Asyncio and Tornado provide easy ways to create timers in python\n",
    "* Tornado can clearly be helpful for inter-process communication in projects like Luigi\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Less good things\n",
    "\n",
    "* ProcessPoolExecutor cannot scale beyond a single machine\n",
    "    * We need a way to use Tornado to communicate between processes or computers\n",
    "* The as_completed function blocks when we run it\n",
    "    * We want to be able to just declare how the functions should be called and then go for a cup of weak herbal tea while it runs on the cluster\n",
    "* Running everything from a central IOLoop can be fiddly..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Dask/Distributed a framework built on `Tornado` and `ThreadPoolExecutor`\n",
    "\n",
    "* Works similarly to the `ThreadPoolExecutor` \n",
    "* It is used to access either a set of local subprocesses or a cluster of worker machines\n",
    "* Access to a running IOLoop\n",
    "\n",
    "https://github.com/dask/distributed\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### \"Run later\" features of Dask/Distributed\n",
    "\n",
    "* Futures can be used as arguments to `distributed.Client().submit`\n",
    "* `worker_client` can be used to process data remotely on the worker\n",
    "* Data structures like [Dask.delayed, array, bag and DataFrame](http://dask.pydata.org/en/latest/index.html) help build up computations\n",
    "* [Queues] can be passed to `distributed.Client().map`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## How can we make this run on a cluster?\n",
    "\n",
    "Dask/Distributed allows us to run a workflow like this across multiple machines. The steps involved are:\n",
    "\n",
    "1. Start the Dask/Distributed cluster running locally\n",
    "1. Get the CSV file as a dataframe\n",
    "1. Send the image ids in batches to each of the distributed workers\n",
    "1. Use the IOLoop instance on the worker_client of distributed to pull down the images\n",
    "1. As the futures are completed process the images\n",
    "1. Return a list of pixels and run the TSNE plot\n",
    "\n",
    "see http://localhost:8888/edit/distrunner.py\n",
    "and http://localhost:8888/edit/distrunner_tornado.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## How Jupyter notebooks work\n",
    "\n",
    "Jupyter notebooks have a similar executor to the dask/distributed cluster. The flow of events looks something like this:\n",
    "    \n",
    "1. User gets HTML page\n",
    "1. Websockets opened for communication with browser\n",
    "1. When code is added and run a message is sent by websockets\n",
    "1. Tornado handles this message and passes the Executor over the 0mq protocol\n",
    "1. When the task is done a message is sent back to tornado and then in tunr back to the browser\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "![](http://jupyter.readthedocs.io/en/latest/_images/notebook_components.png)\n",
    "\n",
    "### Why Tornado for systems like this?\n",
    "\n",
    "The single threaded nature of Tornado makes it ideal for keeping multiple browser tabs up to date - it is impossible to append to a stream with multiple threads - we must use a single threaded server like Tornado or Node.js.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## A Web-based vidiprinter\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "See tornado_vidiprinter.py\n",
    "\n",
    "from tornado import gen, ioloop, web\n",
    "\n",
    "    @web.stream_request_body\n",
    "    class Handler(web.RequestHandler):\n",
    "        @gen.coroutine\n",
    "        def get(self):\n",
    "            self.write('start<br/>')\n",
    "            yield self.flush()\n",
    "            yield gen.sleep(5)\n",
    "            self.write('finish\\n')\n",
    "            yield self.flush()\n",
    "\n",
    "\n",
    "    application = web.Application([(r'/', Handler)])\n",
    "    application.listen(8889)\n",
    "    ioloop.IOLoop.current().start()\n",
    "\n",
    "We can start a streaming response, the real power though comes from using websockets with Tornado."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Excersise 5 Websockets task\n",
    "\n",
    "Use websockets to create an ascii art clock in the browser or a live visualisation of the countries of the most recent wikipedia contributors\n",
    "\n",
    "If doing the visualisation, you can use the answer to Excersise \n",
    "\n",
    "The repo https://github.com/albertobeta/UberSimpleWebsockets is a really great starting point."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Summary\n",
    "\n",
    "In this tutorial we have learned:\n",
    "\n",
    "* How to use `ProcessPoolExecutor` and `ThreadPoolExcecutor` to run tasks which we can watch the status of with `Futures`\n",
    "* How to make HTTP requests asynchronously\n",
    "* How to follow a streaming HTTP request asynchronously and save it to a database\n",
    "* How to timeout the running of a functiuon\n",
    "* How to combine asynchronous execution of HTTP requests with synchronous concurrent data processing and to distribute that across multiple machines."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Thanks\n",
    "\n",
    "* Zegami (Talk to me if you wnat to use it, work for us etc!!!)\n",
    "* Matthew Wrocklin - developer of dask/distributed\n",
    "* Everyone for listening\n",
    "\n",
    "\n",
    "## Andrew Stretton\n",
    "\n",
    "* astretton@zegami.com\n",
    "* https://github.com/strets123\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}