Day 10 - Parallel Computing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Parallel Computing with cluster-helper\n",
    "\n",
    "<img src=\"https://computing.llnl.gov/tutorials/parallel_comp/images/nodesNetwork.gif\">\n",
    "\n",
    "A node is a like a computer within a much bigger computer!\n",
    "\n",
    "**Install ipython-cluster-helper**:\n",
    "\n",
    "* Clone repo from https://github.com/roryk/ipython-cluster-helper to your home directory\n",
    "* Activate your CML Anaconda environment (e.g. <code>source activate environmentname</code>)\n",
    "* Navigate to the ipython-cluster-helper directory and type <code>python setup.py install</code>\n",
    "\n",
    "(For some reason, this can take a while, so it may be advisable to have students do it at the end of the prior day's session)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Basic cluster helper usage works as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10 Engines running\n",
      "Sending a shutdown signal to the controller and engines.\n"
     ]
    }
   ],
   "source": [
    "# Basic cluster helper usage.\n",
    "\n",
    "import cluster_helper.cluster\n",
    "\n",
    "# We define a function to run as a job on each node.\n",
    "# This function takes a parameter from an iterable of parameter inputs.\n",
    "# NOTE!  Every instance of this function starts with an empty variable space,\n",
    "# as each run on each node is starting in its own Python instance.\n",
    "# This means that you need to import any needed libraries INSIDE the function,\n",
    "# and any data needed from the launching Python program needs to either be\n",
    "# loaded from a file inside the function, or passed in as part of the parameter.\n",
    "def squared(x):\n",
    "    return x**2\n",
    "\n",
    "from pathlib import Path\n",
    "myhomedir = str(Path.home())\n",
    "# If you are on a cluster other than Rhino, then parameters such as\n",
    "# scheduler and queue will likely need adjusting to the values for that cluster.\n",
    "with cluster_helper.cluster.cluster_view(scheduler=\"sge\", queue=\"RAM.q\", num_jobs=10,\n",
    "      cores_per_job=1, profile=myhomedir + '/.ipython/') as view:\n",
    "    \n",
    "    # 'map' applies a function to each value within an interable.\n",
    "    res = view.map(squared, range(0, 10))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## cluster-helper tips\n",
    "\n",
    "* cluster-helper will act as if a fresh python notebook were started for each job, so it will not inherit your workspace's variables or import statements. **Give each job everything it needs to complete!**\n",
    "* Jobs are still subject to memory limitations, so you may need to **break up large processes into smaller chunks.** For example, each job could correspond to analyzing one session, instead of one subject. \n",
    "* The cluster-helper memory parameter does not work. Supposedly, this will be fixed eventually. If absolutely necessary, you may specify more cores per job to functionally increase the allotted memory. But mind your total core count!\n",
    "* It is often useful to save the output of each job in a dedicated directory, and sometimes useful to save intermediate values to aid in debugging or later nonparallel analyses. The Python \"os\" library can be helpful here. \n",
    "* **Be respectful!** There are only so many cores available to the entire Kahana lab and our collaborators across the country. \n",
    "* **Limit typical jobs to 100 cores or less**. Heavy usage means fewer resources for other users to use, and due to shared disk resources might actually slow down all jobs overall. Please ask for permission before using more.\n",
    "* You can always use the '**qdel**' command in Terminal, followed by your job number, to kill any of your old jobs that may be wasting rhino's resources. \n",
    "* Use the '**qstat**' command in Terminal to see cluster usage information.\n",
    "* Each rhino2 node has ~128 GB of memory and ~40 cores. \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Simpler Cluster Helper usage"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "See the separate ClusterRun.py file in this repository for example code for more conveniently calling cluster-helper.  It will be imported directly from that file in the following example.  As long as you follow the polite usage etiquette described above, you should feel free to customize this to your own needs.  It is helpful to follow the general principles shown here, such as saving computational results for each job directly to disk and logging exceptions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10 Engines running\n",
      "Sending a shutdown signal to the controller and engines.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[True, True, True, True, True, True, True, True, True, True]"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from ClusterRun import ClusterRun\n",
    "\n",
    "def squared(x):\n",
    "    try:\n",
    "        import numpy as np\n",
    "    \n",
    "        res = x**2\n",
    "        \n",
    "        # You can directly return results, but saving them in separate files is\n",
    "        # better practice for large jobs.\n",
    "        np.save('squared_result_'+str(x)+'.npy', res)\n",
    "        \n",
    "        # We return True for success\n",
    "        return True\n",
    "    \n",
    "    except Exception as e:\n",
    "        # To diagnose programming errors or data issues, you will probably want\n",
    "        # to write unhandled exceptions to a log file.\n",
    "        # Try to handle more exceptions closer to where they happen, so that few\n",
    "        # get here to the top level.\n",
    "        \n",
    "        import traceback\n",
    "        np.savetxt('squared_error_'+str(x)+'.txt', \\\n",
    "                   [str(x), traceback.format_exc()], \\\n",
    "                   fmt='%s')\n",
    "        return False\n",
    "    \n",
    "\n",
    "parameters = range(0, 10)\n",
    "ClusterRun(squared, parameters)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0\n",
      "1\n",
      "4\n",
      "9\n",
      "16\n",
      "25\n",
      "36\n",
      "49\n",
      "64\n",
      "81\n"
     ]
    }
   ],
   "source": [
    "# To reload the saved results of the parallel run.\n",
    "import numpy as np\n",
    "for x in parameters:\n",
    "    print(np.load('squared_result_'+str(x)+'.npy', allow_pickle=True))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A slightly more convenient way to integrate parallel executions with a notebook is to use ClusterChecked, which provides a diagnostic and an exception if there is a problem with one or more inputs.  This will helpfully prevent your notebook from continuing on with execution if output data from your parallel jobs was not successfully created or updated."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2 Engines running\n",
      "Sending a shutdown signal to the controller and engines.\n",
      "Error on job parameters:\n",
      "  Ricanttypeohno\n"
     ]
    },
    {
     "ename": "RuntimeError",
     "evalue": "1 of 2 jobs failed!",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mRuntimeError\u001b[0m                              Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-4-6b1d900189c7>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m     28\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     29\u001b[0m \u001b[0msub_list\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m[\u001b[0m\u001b[0;34m'R1111M'\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m'Ricanttypeohno'\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 30\u001b[0;31m \u001b[0mClusterChecked\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mTryToLoadSubject\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msub_list\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;32m~/PythonBootcamp/ClusterRun.py\u001b[0m in \u001b[0;36mClusterChecked\u001b[0;34m(function, parameter_list, *args, **kwargs)\u001b[0m\n\u001b[1;32m     53\u001b[0m           '\\n  '.join(str(parameter_list[i]) for i in range(len(res))\n\u001b[1;32m     54\u001b[0m             if not bool(res[i])))\n\u001b[0;32m---> 55\u001b[0;31m       \u001b[0;32mraise\u001b[0m \u001b[0mRuntimeError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mstr\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfailed\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m+\u001b[0m\u001b[0;34m' of '\u001b[0m\u001b[0;34m+\u001b[0m\u001b[0mstr\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mlen\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mres\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m+\u001b[0m\u001b[0;34m' jobs failed!'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     56\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     57\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mRuntimeError\u001b[0m: 1 of 2 jobs failed!"
     ]
    }
   ],
   "source": [
    "from ClusterRun import ClusterChecked\n",
    "\n",
    "def TryToLoadSubject(sub):\n",
    "  try:\n",
    "    from CMLLoad import CMLLoad\n",
    "\n",
    "    # Remember to update this to where your CMLExamples data is located.\n",
    "    load = CMLLoad('./CMLExamples')\n",
    "    df = load.Index()\n",
    "    df_select = df[(df['subject']==sub)]\n",
    "    df_sess = df_select.iloc[0]\n",
    "    eeg_ptsa = load.LoadPTSA(df_sess, 0, 1600)\n",
    "\n",
    "    return True\n",
    "  \n",
    "  except Exception as e:\n",
    "    # To diagnose programming errors or data issues, you will probably want\n",
    "    # to write unhandled exceptions to a log file.\n",
    "    # Try to handle more exceptions closer to where they happen, so that few\n",
    "    # get here to the top level.\n",
    "\n",
    "    import traceback\n",
    "    import numpy as np\n",
    "    np.savetxt('trytoload_error_'+str(sub)+'.txt', \\\n",
    "               [str(sub), traceback.format_exc()], \\\n",
    "               fmt='%s')\n",
    "    return False\n",
    "\n",
    "sub_list = ['R1111M', 'Ricanttypeohno']\n",
    "ClusterChecked(TryToLoadSubject, sub_list)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If the input parameter is not sufficient to diagnose the problem, which it usually isn't, then inspecting the log file saved as trytoload_error_\\*.txt will reveal where the issue happened.  In this case it is an IndexError from attempting to choose index 0 of a df_select which is empty, because there are no matching frames."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Solving the challenges of parallel execution\n",
    "\n",
    "To pass large amounts of data around, np.save and np.load are the most convenient functions to use if all nodes share a file system.  This works for both passing data into a parallel function and for bringing result data out.  For other types of data a convenient settings class is provided in the ClusterRun file.  The following illustrates a good structure for a typical real-world analysis run in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "20 Engines running\n",
      "Sending a shutdown signal to the controller and engines.\n",
      "All 20 jobs successful.\n"
     ]
    }
   ],
   "source": [
    "from ClusterRun import ClusterChecked,Settings\n",
    "settings = Settings()\n",
    "settings.exp_list = ['FR1']\n",
    "settings.Save()\n",
    "\n",
    "def RecallRate(sub):\n",
    "  try:\n",
    "    from ClusterRun import Settings,LogErr,DFRLabel\n",
    "    settings = Settings.Load()\n",
    "    \n",
    "    import numpy as np\n",
    "    from CMLLoad import CMLLoad\n",
    "    # Remember to update this to where your CMLExamples data is located.\n",
    "    load = CMLLoad('./CMLExamples')\n",
    "    df = load.Index()\n",
    "    df = df[df['subject']==sub]\n",
    "    \n",
    "    total_words = 0\n",
    "    total_recalled = 0\n",
    "    sess_count = 0\n",
    "    for df_sess in df.itertuples():\n",
    "      try:\n",
    "        if df_sess._asdict()['experiment'] not in settings.exp_list:\n",
    "          continue\n",
    "        evs = load.Load(df_sess, 'events')\n",
    "        word_evs = evs[evs['type']=='WORD']\n",
    "        total_recalled += sum(word_evs['recalled'])\n",
    "        total_words += len(word_evs)\n",
    "        sess_count += 1\n",
    "      except Exception as e:\n",
    "        # Log the exception to a subject-labeled filename,\n",
    "        # along with a label of subject, experiment, and session.\n",
    "        LogErr(e, DFRLabel(df_sess), suffix=sub)\n",
    "    \n",
    "    # Save the result.\n",
    "    np.save('recall_'+sub+'.npy', [total_recalled, total_words, sess_count])\n",
    "    return True\n",
    "  except Exception as e:\n",
    "    LogErr(e, suffix=sub)\n",
    "    return False\n",
    "\n",
    "from CMLLoad import CMLLoad\n",
    "# Remember to update this to where your CMLExamples data is located.\n",
    "load = CMLLoad('./CMLExamples')\n",
    "df = load.Index()\n",
    "# Run on all subjects.\n",
    "sub_list = sorted(set(df['subject']))\n",
    "ClusterChecked(RecallRate, sub_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "R1060M: 30.4% recall\n",
      "R1065J: 33.6% recall\n",
      "R1108J had no sessions from ['FR1']\n",
      "R1111M: 53.9% recall\n",
      "R1189M: 37.2% recall\n",
      "R1236J had no sessions from ['FR1']\n",
      "R1292E: 29.6% recall\n",
      "R1332M: 38.0% recall\n",
      "R1350D: 30.6% recall\n",
      "R1354E: 29.8% recall\n",
      "R1361C: 30.2% recall\n",
      "R1375C: 27.6% recall\n",
      "R1377M: 47.1% recall\n",
      "R1378T: 29.2% recall\n",
      "R1380D: 42.9% recall\n",
      "R1383J: 29.5% recall\n",
      "R1385E: 33.8% recall\n",
      "R1390M: 22.4% recall\n",
      "R1391T: 30.3% recall\n",
      "R1401J: 19.9% recall\n"
     ]
    }
   ],
   "source": [
    "import numpy as np\n",
    "for sub in sub_list:\n",
    "  total_recalled,total_words,sess_count = np.load('recall_'+sub+'.npy')\n",
    "  if sess_count==0:\n",
    "    print(f'{sub} had no sessions from {settings.exp_list}')\n",
    "  else:\n",
    "    perc = 100.0*total_recalled / total_words\n",
    "    print(f'{sub}: {perc:.1f}% recall')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Exercise: Write a parallel function that returns the number of (bipolar) electrodes for every subject in the RAM example dataset. Run with 5 jobs and 1 core per job.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Wrap Up/Q&A\n",
    "\n",
    "#### What did we learn over the past 2 weeks?\n",
    "\n",
    "* Loading experimental info & EEG\n",
    "* Spectral decomposition & time-frequency analysis\n",
    "* Statistics: T-tests, multiple comparisons, permutations\n",
    "* Phase-based functional connectivity\n",
    "* Machine learning: linear, logisitic regression, feature selection\n",
    "\n",
    "#### What did we NOT learn?\n",
    "\n",
    "* Cognitive modeling\n",
    "* Complex behavioral analyses\n",
    "* Single-unit activity\n",
    "* Spatial memory tasks\n",
    "* Brain stimulation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "environmentname",
   "language": "python",
   "name": "environmentname"
  },
  "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}