_0.3_Introduction_to_Parallel_Processing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### Parallel processing\n",
    "\n",
    "This is probably the hardest tutorial of the day. I have done the best I can to make it clear and simple as possible. It took me a very long time to understand it. I hope that this tutorial means you can understand parallel processing in a much shorter time than I can.\n",
    "\n",
    "Parallel processing is highly important in bioinformatics and something that is often taken for granted. You might think you know how to parallel process by adding '-t 20' to the options of your command. But that doesn't help grip what's going on under the hood.\n",
    "\n",
    "\"You only truly understand the benefits of parallel processing until you can no longer do it.\"\n",
    "\n",
    "This example generated here is taken with much influence from a fantastic stackover-flow [thread](http://stackoverflow.com/questions/14533458/python-threading-multiple-bash-subprocesses).\n",
    "\n",
    "An example of important parallel-processing in nanopore sequencing is extracting fastq information from a set of fast5 files. This can take up to a day...\n",
    "\n",
    "In our example today we will generate 200 commands that all take somewhere between 0 and 9 seconds to complete. We will run this in five parallel streams. This should roughly take one fifth of the time it would take if we were to do this with only one stream."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Import the libraries we will need\n",
    "import subprocess  # The library that will actually be talking to the shell and\n",
    "                   # tell it to what to run and when.\n",
    "from itertools import islice  # Important tool that will allow us to split up our\n",
    "                              # commands for each output.\n",
    "import random  # This will determine how long each process will take.\n",
    "random.seed(1)  # Feel free to change this, but useful in the notebook so the author can explain\n",
    "                # the output even if the output is 'random'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<open file 'output.file.0', mode 'w' at 0x7fdcd08ec150>\n",
      "<open file 'output.file.1', mode 'w' at 0x7fdcd08ec1e0>\n",
      "<open file 'output.file.2', mode 'w' at 0x7fdcd08ec270>\n",
      "<open file 'output.file.3', mode 'w' at 0x7fdcd08ec300>\n",
      "<open file 'output.file.4', mode 'w' at 0x7fdcd08ec390>\n"
     ]
    }
   ],
   "source": [
    "# Set the number of threads\n",
    "threads = 5  # Of the 200 commands, five will be running at any one time.\n",
    "\n",
    "# We need to have 5 separate output files to stop each running command from \n",
    "# over writing the work of a simultaneous command.\n",
    "output_files = [\"output.file.%d\" % i for i in range(0, threads)]  #output.file.0 to output.file.4\n",
    "\n",
    "file_handlers = [None]*threads  # Generates a list of NULL variables of length 5.\n",
    "\n",
    "# This assigns the file handler for each file.\n",
    "for index, output_file in enumerate(output_files):\n",
    "    file_handlers[index] = open(output_file, 'w')\n",
    "\n",
    "for handler in file_handlers:  # Print the file handler so we know what they look like.\n",
    "    print handler"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ugh! Looking kinda ugly, but what we can see that each of our output files are in an 'open' state\n",
    "and that we are 'writing' into them (rather than reading, or appending)\n",
    "\n",
    "Now we will make a list of random numbers and a list of associated commands.\n",
    "These commands will sleep for a 'random' number of seconds, then wake up and print which of the 200 commands they were, and how long they slept for."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "random_number_list = [random.randrange(1,10) for i in range(0,200)]\n",
    "commands = [\"sleep %d && echo Command number - %d. Slept for %d.\" % (j, i, j)\n",
    "            for i, j in enumerate(random_number_list)]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we're going to make some processes using the subprocess command. The subprocess command is able to talk and boss around the terminal, then pull in the output of the terminal."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Talk to the shell. Note these commands won't run just yet.\n",
    "processes = (subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)\n",
    "            for cmd in commands)\n",
    "\n",
    "# We use the islice command to split our list of 200 commands into five smaller lists.\n",
    "running_processes = list(islice(processes, threads))\n",
    "\n",
    "while running_processes:\n",
    "    for i, process in enumerate(running_processes):\n",
    "        if process.poll() is not None:  # Means that the process is complete!\n",
    "            stdout, stderr = process.communicate()  # Get the output of the completed process\n",
    "            file_handlers[i].write(str(stdout) + \"\\n\")  # Write the output to handler that\n",
    "            running_processes[i] = next(processes, None)\n",
    "            # Run the next number in the list.\n",
    "            if running_processes[i] is None:  # No more commands waiting to be processed.\n",
    "                del running_processes[i]  # Not a valid process.\n",
    "                break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Before we can see anything we need to close of the file handlers."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# By closing the file_handler this prints everything accumulated in the handler to the file.\n",
    "for handler in file_handlers:\n",
    "    handler.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "### output.file.0 ###\n",
      "['Command number - 0. Slept for 2.\\n', '\\n', 'Command number - 5. Slept for 5.\\n']\n",
      "### output.file.1 ###\n",
      "['Command number - 1. Slept for 8.\\n', '\\n', 'Command number - 10. Slept for 8.\\n']\n",
      "### output.file.2 ###\n",
      "['Command number - 2. Slept for 7.\\n', '\\n', 'Command number - 9. Slept for 1.\\n']\n",
      "### output.file.3 ###\n",
      "['Command number - 3. Slept for 3.\\n', '\\n', 'Command number - 6. Slept for 6.\\n']\n",
      "### output.file.4 ###\n",
      "['Command number - 4. Slept for 5.\\n', '\\n', 'Command number - 7. Slept for 8.\\n']\n"
     ]
    }
   ],
   "source": [
    "# Now let's have a look at the first few lines of each each file.\n",
    "number_of_lines = 3\n",
    "for output_file in output_files:\n",
    "    with open(output_file) as output_handler:\n",
    "        head = list(islice(output_handler, number_of_lines))\n",
    "    print \"### \" + output_file + \" ###\"\n",
    "    print head"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice how the first line of each file is in order. Then let's look at which of these first lines was the fastest to process. Notice how they got the next thread? Command 0 slept for the least amount of time, so it got command five, command 3 next and Therefore, it is highly likely that each of these output files have a different length as some will have processed some shorter reads faster. "
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2.0
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}