tests/ipyparallel-tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Parallelization in `ipyrad` using `ipyparallel`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "One of the real strenghts of `ipyrad` is the advanced parallelization methods that it uses to distribute work across arbitrarily large computing clusters, and to be able to do so when working interactively and remotely. This is done through use of the `ipyparallel` package, which is tightly linked to `ipython` and `jupyter`. When you run the command-line `ipyrad` program all of the work of `ipyparallel` is hidden under the hood, which we've done to make the program very user-friendly. However, when using the `ipyrad` API, we've taken the alternative approach of instructing users to become intimate with the `ipyparallel` library to better understand how work is being distributed on their system. This has the benefit of allowing more flexible parallelization setups, and also makes it easier for users to take advantage of `ipyparallel` for parallelizing downstream analyses, which we have many examples of in the `analysis-tools` section of the ipyrad documentation. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Required software\n",
    "All software required for this tutorial is installed during the ipyrad conda installation. \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "## conda install ipyrad -c ipyrad"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Starting an `ipcluster` instance\n",
    "\n",
    "The tricky aspect of using `ipyparallel` inside of Python (i.e., in a jupyter-notebook) is that you need to first start a cluster instance by running a command-line program called ``ipcluster`` (alternatively, you can also install an extension that makes it possible to start ipcluster from a tab in jupyter notebooks, but I feel the command line tool is simpler). This command will start separate python \"kernels\" (instances) running on the cluster/computer and ensure that they can all talk to each other. Using advanced options you can even connect kernels across multiple computers or nodes on a HPC cluster, which we'll demonstrate. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "## ipcluster start --n=4"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Start a `jupyter-notebook`\n",
    "If you are working on your laptop of a workstation then I typically open up two terminals, one to start a notebook and one to start an ipcluster instance. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "## jupyter-notebook"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<HTML>\n",
    "  <body>\n",
    "    <img src=\"https://eaton-lab.github.io/images/start-jup-ipc.gif\">\n",
    "  </body>\n",
    "</HTML>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Open a notebook\n",
    "Running `jupyter-notebook` will launch a server that will open a dashboard view in your browser, usually at the address is at `localhost:8888`. From the dashboard go to the menu and select `new/notebook/Python` to open a new notebook. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Connect to `ipcluster` in your notebook\n",
    "Now from inside a notebook you can connect to the cluster using the `ipyparallel` library. Below we will connect to the client by providing no additional arguments, which is sufficient in this case sine we are using a very basic `ipcluster` setup. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "6.0.2\n"
     ]
    }
   ],
   "source": [
    "import ipyparallel as ipp\n",
    "print ipp.__version__"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4 cores\n"
     ]
    }
   ],
   "source": [
    "## connect to ipcluster using default arguments\n",
    "ipyclient = ipp.Client()\n",
    "\n",
    "## count how many engines are connected\n",
    "print len(ipyclient), 'cores'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `Profiles` in ipcluster and how ipyrad uses them\n",
    "Below we show an example of a common error caused when the Client cannot find the `ipcluster` instance, in this case because it has a differnt profile name. When you start an ipcluster instance it keeps track of itself by using a specific name (its profile). The default profile is an empty string (\"\") and so this is the default profile that the `ipp.Client()` command will look for (and similarly the default profile that `ipyrad` will look for). If you change the name of the profile then you have to indicate this, like below. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Waiting for connection file: ~/.ipython/profile_MPI/security/ipcontroller-client.json\n"
     ]
    },
    {
     "ename": "IOError",
     "evalue": "Connection file '~/.ipython/profile_MPI/security/ipcontroller-client.json' not found.\nYou have attempted to connect to an IPython Cluster but no Controller could be found.\nPlease double-check your configuration and ensure that a cluster is running.",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mIOError\u001b[0m                                   Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-22-89ba45db61df>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;31m## example connecting to a named profile\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 2\u001b[0;31m \u001b[0mmpi\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mipp\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mClient\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mprofile\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m\"MPI\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;32m/home/deren/miniconda2/lib/python2.7/site-packages/ipyparallel/client/client.pyc\u001b[0m in \u001b[0;36m__init__\u001b[0;34m(self, url_file, profile, profile_dir, ipython_dir, context, debug, sshserver, sshkey, password, paramiko, timeout, cluster_id, **extra_args)\u001b[0m\n\u001b[1;32m    395\u001b[0m                         \u001b[0mno_file_msg\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    396\u001b[0m                     ])\n\u001b[0;32m--> 397\u001b[0;31m                     \u001b[0;32mraise\u001b[0m \u001b[0mIOError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mmsg\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    398\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0murl_file\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0mNone\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    399\u001b[0m             \u001b[0;32mraise\u001b[0m \u001b[0mIOError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mno_file_msg\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mIOError\u001b[0m: Connection file '~/.ipython/profile_MPI/security/ipcontroller-client.json' not found.\nYou have attempted to connect to an IPython Cluster but no Controller could be found.\nPlease double-check your configuration and ensure that a cluster is running."
     ]
    }
   ],
   "source": [
    "## example connecting to a named profile\n",
    "mpi = ipp.Client(profile=\"MPI\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Start a second `ipcluster` instance with a specific profile\n",
    "By using separate profiles you can have multiple `ipcluster` instances running at the same time (and possibly using different options or connected to different nodes of an HPC cluster) and you ensure that you connect to each one dictinctly. Start a new instance and then connect to it like below. Here we give it a profile name and we also tell it to initiate the engines using the `MPI` initiator by using the `--engines` flag. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "## ipcluster start --n=4 --engines=MPI --profile=MPI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4 cores\n"
     ]
    }
   ],
   "source": [
    "## now you should be able to connect to the MPI profile\n",
    "mpi = ipp.Client(profile=\"MPI\")\n",
    "\n",
    "## print mpi info\n",
    "print len(mpi), 'cores'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using an `client()` object to distribute jobs\n",
    "For full details of how this works you can read the `ipyparallel` documentation. Here I will focus on the tools in `ipyrad` that we have developed to utilize `Client` objects to distribute work. In general, all you have to do is provide the ipyclient object to the `.run()` function and `ipyrad` will take care of the rest. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<ipyparallel.client.client.Client at 0x7fec0c31f910>"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "## the ipyclient object is simply a view to the engines\n",
    "ipyclient"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "import ipyrad as ip\n",
    "import ipyrad.analysis as ipa"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example of using a Client in an ipyrad assembly\n",
    "Here we create an Assembly and when we call the `.run()` command we provide a specific `ipyclient` object as the target to distribute work on. If you do not provide this option then by default `ipyrad` will look for an `ipcluster` instance running on the default profile (\"\"). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "New Assembly: example\n",
      "Assembly: example\n",
      "[####################] 100%  loading reads         | 0:00:12 | s1 | \n"
     ]
    }
   ],
   "source": [
    "## run steps of an ipyrad assembly on a specific ipcluster instance\n",
    "data = ip.Assembly(\"example\")\n",
    "data.set_params(\"sorted_fastq_path\", \"example_empirical_rad/*.fastq.gz\")\n",
    "data.run(\"1\", ipyclient=mpi)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example of using a Client in an analysis (`tetrad`)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "loading seq array [13 taxa x 14159 bp]\n",
      "max unlinked SNPs per quartet (nloci): 2777\n",
      "inferring 715 quartet tree sets\n",
      "host compute node: [4 cores] on oud\n",
      "[####################] 100% generating q-sets | 0:00:02 |  \n",
      "[####################] 100% initial tree      | 0:00:06 |  \n",
      "[####################] 100% bootstrap trees   | 0:00:07 |  \n",
      "[####################] 100% calculating stats | 0:00:00 |  \n"
     ]
    }
   ],
   "source": [
    "## run ipyrad analysis tools on a specific ipcluster instance\n",
    "tet = ipa.tetrad(\n",
    "    name=\"test\",\n",
    "    data=\"./analysis-ipyrad/pedic-full_outfiles/pedic-full.snps.phy\",\n",
    "    mapfile=\"./analysis-ipyrad/pedic-full_outfiles/pedic-full.snps.map\",\n",
    "    nboots=20);\n",
    "\n",
    "tet.run(ipyclient=mpi)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Parallelizing any arbitrary function with ipyparallel\n",
    "A very simple example..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "## get load-balanced view of our ipyclient\n",
    "lbview = ipyclient.load_balanced_view()\n",
    "\n",
    "## a dict to store results\n",
    "res = {}\n",
    "\n",
    "## define your func\n",
    "def my_sum_func(x, y):\n",
    "    return x, y, sum([x, y])\n",
    "\n",
    "## submit jobs to cluster with arguments\n",
    "import random\n",
    "for job in range(10):\n",
    "    x = random.randint(0, 10)\n",
    "    y = random.randint(0, 10)\n",
    "    \n",
    "    ## submitting a job returns an async object\n",
    "    async = lbview.apply(my_sum_func, x, y)\n",
    "    \n",
    "    ## store results\n",
    "    res[job] = async\n",
    "    \n",
    "## block until all jobs finish\n",
    "ipyclient.wait()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{0: <AsyncResult: my_sum_func:finished>,\n",
       " 1: <AsyncResult: my_sum_func:finished>,\n",
       " 2: <AsyncResult: my_sum_func:finished>,\n",
       " 3: <AsyncResult: my_sum_func:finished>,\n",
       " 4: <AsyncResult: my_sum_func:finished>,\n",
       " 5: <AsyncResult: my_sum_func:finished>,\n",
       " 6: <AsyncResult: my_sum_func:finished>,\n",
       " 7: <AsyncResult: my_sum_func:finished>,\n",
       " 8: <AsyncResult: my_sum_func:finished>,\n",
       " 9: <AsyncResult: my_sum_func:finished>}"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "## the results objects\n",
    "res"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "job: 0, result: (10, 3, 13)\n",
      "job: 1, result: (6, 7, 13)\n",
      "job: 2, result: (6, 10, 16)\n",
      "job: 3, result: (6, 9, 15)\n",
      "job: 4, result: (6, 9, 15)\n",
      "job: 5, result: (3, 4, 7)\n",
      "job: 6, result: (0, 2, 2)\n",
      "job: 7, result: (1, 7, 8)\n",
      "job: 8, result: (0, 8, 8)\n",
      "job: 9, result: (3, 0, 3)\n"
     ]
    }
   ],
   "source": [
    "for ridx in range(10):\n",
    "    print \"job: {}, result: {}\".format(ridx, res[ridx].get())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Starting an ipcluster instance on a HPC cluster\n",
    "See the multi-node setup in the HPC tunneling tutorial for instructions. http://ipyrad.readthedocs.io/HPC_Tunnel.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "## an example command to connect to 80 cores across multiple nodes using MPI\n",
    "## ipcluster start --n=80 --engines=MPI --ip='*' --profile='MPI80'"
   ]
  }
 ],
 "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}