Classification_futures.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "from classification import nearest_neighbor_prediction\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction to `concurrent.futures`\n",
    "\n",
    "`concurrent.futures` is a high-level interface for **asynchronous**, **parallel** calculations.   \n",
    "It is a new interface that [replaces](https://www.python.org/dev/peps/pep-3148/) the **`multiprocessing`** and **`threading`** interfaces.\n",
    "\n",
    "As a reminder, here is comparison between `mpi4py` and `concurrent.futures`:\n",
    "\n",
    "`mpi4py` | `concurrent.futures`\n",
    "------ | -----\n",
    "Can run on **several nodes** | Runs only on **one node** \n",
    "Only handles **processes** | Can handle **processes** and **threads** \n",
    "No interactivity | Some interactivity   (e.g. integrates in Jupyter notebook)\n",
    "Allows elaborate communications between processes | No communication between processes or threads\n",
    "\n",
    "In this notebook, the feature that will be most visible is the fact that `concurrent.futures` integrates well into **interactive python environments** (e.g. Jupyter notebook, ipython). In particular, there is no need to execute code in a separate script, no need to save the data in a file and then to read it in the interactive python environment. The results are directly accessible in the interactive python environment.\n",
    "\n",
    "### Installation\n",
    "\n",
    "- Python 3: already installed by default \n",
    "\n",
    "- Python 2: `conda install futures`\n",
    "\n",
    "### Import statement"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example\n",
    "\n",
    "In the example, we determine which integers in a list of integers are even.\n",
    "\n",
    "- The statement **`with ProcessPoolExecutor(max_workers=4) as e`** spawns 4 processes and suppresses them once they have finished the calculation. \n",
    "\n",
    "- The function **`map`** distributes the integers of the list among the processes.\n",
    "\n",
    "- The returned object **`result`** is an **iterator**, with each element corresponding to the elements of **`list_integers`**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Define function\n",
    "def is_even( i ):\n",
    "    \"\"\"Determine if the integer i is even\"\"\"\n",
    "    even = (i%2==0)\n",
    "    return( even )\n",
    "\n",
    "list_integers = [ 4, 3, 6, 7, 9, 10, 124, 325 ]\n",
    "\n",
    "# Spawn 4 processes\n",
    "with ProcessPoolExecutor(max_workers=4) as e:\n",
    "    result = e.map( is_even, list_integers )\n",
    "    \n",
    "for answer in result:\n",
    "    print(answer)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that, because `concurrent.futures` integrates in the Jupyter Notebook, the object **`result`** is directly accessible in the notebook. We did not have to save it to a file and to load it in the Jupyter notebook. We also did not need to gather the results on one processor."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "# Digit classification\n",
    "\n",
    "Let us again load the data, and define a function to look at it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Load the data\n",
    "train_images = np.load('./data/train_images.npy')\n",
    "train_labels = np.load('./data/train_labels.npy')\n",
    "test_images = np.load('./data/test_images.npy')\n",
    "\n",
    "# Define function to have a look at the data\n",
    "def show_random_digit( images, labels=None ):\n",
    "    \"\"\"\"Show a random image out of `images`, \n",
    "    with the corresponding label if available\"\"\"\n",
    "    i = np.random.randint(len(images))\n",
    "    image = images[i].reshape((28, 28))\n",
    "    plt.imshow( image, cmap='Greys' )\n",
    "    if labels is not None:\n",
    "        plt.title('Label: %d' %labels[i])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In addition, we need to define a function that takes only one argument, in order to pass it to the function `map`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def predict( test_images ):\n",
    "    return( nearest_neighbor_prediction( test_images, train_images, train_labels ) )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Serial execution\n",
    "\n",
    "Let us again run the serial code, but now **directly from the Jupyter notebook** (i.e. not from a script that we then run from the terminal using the `!` character)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%time \n",
    "test_labels_serial = predict( test_images )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## With processes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Choose the number of processes and split the data\n",
    "N_processes = 4\n",
    "split_arrays = np.array_split( test_images, N_processes )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "with ProcessPoolExecutor(max_workers=N_processes) as e:\n",
    "    result = e.map( predict, split_arrays )\n",
    "\n",
    "# Merge the result from each process into a single array\n",
    "test_labels_proc = np.hstack( ( small_test_labels for small_test_labels in result ) )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "show_random_digit( test_images, test_labels_proc )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## With threads"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Choose the number of threads and split the data\n",
    "N_threads = 4\n",
    "split_arrays = np.array_split( test_images, N_threads )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "with ThreadPoolExecutor(max_workers=N_threads) as e:\n",
    "    result = e.map( predict, split_arrays )\n",
    "    \n",
    "# Merge the result from each thread into a single array\n",
    "test_labels_threads = np.hstack( ( small_test_labels for small_test_labels in result ) )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "show_random_digit( test_images, test_labels_threads )"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}