Classification_mpi4py.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction to MPI and `mpi4py`\n",
    "\n",
    "MPI stands for Message Passing Interface. It is a **library** that allows to:\n",
    "- **spawn** several processes \n",
    "- **adress** them individually\n",
    "- have them **communicate** between them\n",
    "\n",
    "MPI can be used in many languages (C, C++, Fortran), and is extensively used in High-Performance Computing.  \n",
    "`mpi4py` is the Python interface to MPI.\n",
    "\n",
    "### Installation\n",
    "\n",
    "```\n",
    "conda install -c conda-forge mpi4py\n",
    "```\n",
    "(The standard Anaconda channel for mpi4py is [broken](https://github.com/conda/conda/issues/2277). It is necessary to use the [conda-forge](https://conda-forge.github.io/) channel instead.)\n",
    "\n",
    "### Example\n",
    "\n",
    "Let us try to get a feeling on how `mpi4py` works by looking at the example below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%file example.py\n",
    "\n",
    "from mpi4py.MPI import COMM_WORLD as communicator\n",
    "import random\n",
    "\n",
    "# Draw one random integer between 0 and 100\n",
    "i = random.randint(0, 100)\n",
    "print('Rank %d' %communicator.rank + ' drew a random integer: %d' %i )\n",
    "\n",
    "# Gather the results\n",
    "integer_list = communicator.gather( i, root=0 )\n",
    "if communicator.rank == 0:\n",
    "    print('\\nRank 0 gathered the results:')\n",
    "    print(integer_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "! mpirun -np 3 python example.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### What happened?\n",
    "\n",
    "- \"`mpirun -np 3`\" spawns 3 processes.\n",
    "\n",
    "- **All processes execute the same code.** (In this case, they all execute the same Python script: `example.py`.)\n",
    "\n",
    "- Each process gets a **unique identification number** (`communicator.rank`).\n",
    "\n",
    "- **Based on this identifier** and e.g. **based on `if` statements**, the different processes can be **addressed individually**, and perform different work.\n",
    "\n",
    "- MPI provides functions (like `communicator.gather`) that allow processes to **communicate** data (even between **different nodes**).\n",
    "\n",
    "NB: There are many other communication functions, e.g.:\n",
    "- one-to-one communication (`send`, `receive`, `isend`, `ireceive`)\n",
    "- all-to-one communication (`gather`, `reduce`)\n",
    "- one-to-all communication (`scatter`, `broadcast`)\n",
    "- all-to-all (`allgather`, `allreduce`)\n",
    "\n",
    "See the [mpi4py documentation](https://mpi4py.readthedocs.io/en/stable/) for more information.\n",
    "\n",
    "![MPI processes](./tutorial_images/MPI.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "----\n",
    "# Digit classification with `mpi4py`\n",
    "\n",
    "Let us now apply `mpi4py` to our problem: digit classification.  \n",
    "\n",
    "As mentioned earlier, the parallelization, in this case, is conceptually trivial: the process should **split** the test data among themselves and each process should perform the prediction only its share of the data.\n",
    "\n",
    "## On two processes\n",
    "\n",
    "Let start with only with only two processes. In the script below, the data `test_images` is split into a smaller array `small_test_images`, which is different for each process."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%file parallel_script.py\n",
    "\n",
    "from classification import nearest_neighbor_prediction\n",
    "import numpy as np\n",
    "from mpi4py.MPI import COMM_WORLD as communicator\n",
    "\n",
    "# Load 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",
    "# Use only the data that this rank needs\n",
    "N_test = len(test_images)\n",
    "if communicator.rank == 0:\n",
    "    i_start = 0\n",
    "    i_end = N_test/2\n",
    "elif communicator.rank == 1:\n",
    "    i_start = N_test/2\n",
    "    i_end = N_test    \n",
    "small_test_images = test_images[i_start:i_end]\n",
    "\n",
    "# Predict the results\n",
    "small_test_labels = nearest_neighbor_prediction(small_test_images, train_images, train_labels)\n",
    "\n",
    "# Assignement: gather the labels on one process and have it write it to a file\n",
    "# Hint: you can use np.hstack to merge a list of arrays into a single array, \n",
    "# and np.save to save an array to a file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "! mpirun -np 2 python parallel_script.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The code **executes faster than the serial example**, because each process has a smaller amount of work, and the two processes execute this work in parallel.\n",
    "\n",
    "However, at the end of the script, each process has the corresponding label array `small_test_labels`. But these arrays still need to be concatenated together and written to a single file.\n",
    "\n",
    "**Assignement: based on the previous script (`example.py`), use the functionalities of `mpi4py` to gather the labels on one rank, and have this rank write the data to a single file `data/test_labels_parallel.npy`**."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## On more processes\n",
    "\n",
    "The above code works for two processes, but does not generalize easily to an arbitrary number of processes.\n",
    "\n",
    "In order to **split** the initial array `test_images` into an arbitrary number of arrays (one per process), we can use the function `np.array_split`, which splits an array and returns a **list** of smaller arrays.\n",
    "\n",
    "Note: Below, the number 784 corresponds to 28x28, i.e. the number of pixels for each image."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Load and split the set of test images\n",
    "test_images = np.load('data/test_images.npy')\n",
    "split_arrays_list = np.array_split( test_images, 4 )\n",
    "\n",
    "# Print the corresponding shape\n",
    "print( 'Shape of the original array:' )\n",
    "print( test_images.shape )\n",
    "print('Shape of the splitted arrays:')\n",
    "for array in split_arrays_list:\n",
    "    print( array.shape )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Assignement: in the code below, use the function `array_split` to split `test_images` between an arbitrary number of processes, and have each process pick their own small array.**\n",
    "\n",
    "Note: Within the script, `communicator.size` gives the number of processes that have been spawn by `mpirun`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%file parallel_script.py\n",
    "\n",
    "from classification import nearest_neighbor_prediction\n",
    "import numpy as np\n",
    "from mpi4py.MPI import COMM_WORLD as communicator\n",
    "\n",
    "# Load 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",
    "# Assignement: use the function np.array_split the data `test_images` among the processes\n",
    "# Have each process select their own small array.\n",
    "small_test_images = #.....\n",
    "\n",
    "# Predict the results and gather it on rank 0\n",
    "small_test_labels = nearest_neighbor_prediction(small_test_images, train_images, train_labels)\n",
    "\n",
    "# Assignement: gather the labels on one process and have it write it to a file\n",
    "# Hint: you can use np.hstack to merge a list of arrays into a single array, \n",
    "# and np.save to save an array to a file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "! mpirun -np 4 python parallel_script.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "# Check the results\n",
    "\n",
    "Finally we can check that the results are valid."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Load the data from the file\n",
    "test_images = np.load('data/test_images.npy')\n",
    "test_labels_parallel = np.load('data/test_labels_parallel.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": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "show_random_digit( test_images, test_labels_parallel )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Next tutorial\n",
    "\n",
    "Let us now see how to [perform the same tasks with `concurrent.futures`](./Classification_futures.ipynb)."
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "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
}