4 - Workflow Management Systems and Tools.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Ch. 4. Workflow Management Systems and Tools"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## I. What is a Workflow?\n",
    "A _workflow_ is a generalization of the concept of a dataflow, which was introduced in chapter 3, section VI. If you have not read that section, I suggest you go back there and read it now. For the stubborn, a dataflow is essentially a set of different algorithms, which may or may not be parallel, that all fit together in some way, with the output of some parts leading to the input of other parts. Many, many times, people will want to package the things they are doing on an HPC system into a workflow, because workflows are self contained, portable, and have a very clear input and output. You can view a workflow as a \"black box\" which will run on provided input and provide (hopefully correct) output. Here is an example of a representation of a workflow:\n",
    "![basic dataflow](http://www.digitaleng.news/de/wp-content/uploads/2016/10/HPC-Workflow.jpg)\n",
    "In this case, data is first generated, then the quality of that data is assured, and then the data gets computed on, creating some output, and finally, that data gets visualized and consolidated to a human-readable format. This is a fairly common type of dataflow. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 4.1 - My First Workflow\n",
    "The workflow presented in the previous section is a relatively typical dataflow, but it's also fairly complex. To give you a more gentle introduction to working with workflows, we are going to design a dataflow called `sleep_fail`. This workflow consists of a data generation layer, in which we define a function which sleeps for _n_ (user input) seconds before failing with a probability _p_ (also user input). Then, it writes whether it failed or succeeded to a file. Then, we have a data consolidation layer, which reads that file and prints out how many times it ran, succeeded, and failed. The resulting workflow looks a bit like this:\n",
    "```\n",
    "    sleep1  sleep2 ...sleepN\n",
    "      |       |        |\n",
    "      V       V        v\n",
    "       \\      |       /\n",
    "        \\     |      /\n",
    "          sleep_Final\n",
    "```\n",
    "Note that this workflow is the same as the example from Sec. 3.6, except we have parallelized it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "import multiprocessing\n",
    "import time\n",
    "import random\n",
    "\n",
    "# Define sleep_fail function\n",
    "def sleep_fail(n, p, path):\n",
    "    time.sleep(n)\n",
    "    if random.random() < p:\n",
    "        file = open(path, \"a\")\n",
    "        file.write(\"Exception\\n\")\n",
    "        file.close()\n",
    "        raise Exception\n",
    "    else:\n",
    "        file = open(path, \"a\")\n",
    "        file.write(\"Success\\n\")\n",
    "        file.close()\n",
    "        \n",
    "# Define summarize_sleeps\n",
    "def summarize_sleeps(path):\n",
    "    file = open(path, \"r\")\n",
    "    succeed = 0\n",
    "    exceptio = 0\n",
    "    for line in file.readlines():\n",
    "        if \"Exception\" in line:\n",
    "            exceptio +=1\n",
    "        elif \"Success\" in line:\n",
    "            succeed +=1\n",
    "    print(\"Failed {} Times\".format(exceptio))\n",
    "    print(\"Passed {} Times\".format(succeed))\n",
    "    print(\"Ran {} Times\".format(succeed + exceptio))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Failed 28 Times\n",
      "Passed 72 Times\n",
      "Ran 100 Times\n"
     ]
    }
   ],
   "source": [
    "# Parallel Dataflow\n",
    "\n",
    "import os\n",
    "import multiprocessing\n",
    "\n",
    "path = \"/home/users/glick/intro-to-hpc/data/sleep_fail.out\"\n",
    "\n",
    "if os.path.isfile(path):\n",
    "    os.remove(path)\n",
    "\n",
    "def sleepfail_runner():\n",
    "    try:\n",
    "        sleep_fail(0, 0.22, path)\n",
    "    except Exception as e:\n",
    "        pass\n",
    "\n",
    "processes = [multiprocessing.Process(target=sleepfail_runner) for _ in range(100)]\n",
    "[process.start() for process in processes]\n",
    "[process.join() for process in processes]\n",
    "\n",
    "summarize_sleeps(path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## II. Task Flows\n",
    "A workflow is a nice abstraction because it allows you to concisely describe both very high level concepts about how your data pipeline works and low level information about how each individual park of the workflow runs. A good technique to use when designing a workflow is to use \"black box thinking\" to first design the way the various pieces of the workflow interact (some would call this the \"architecture\", or the \"interfaces\"), and then later, design the way that each specific piece of the workflow (or \"black box\") is engineered. Let's say that you, for example, are in the field of bioinformatics. You want to run `BLAST` on a bunch of proteins, and then perform some comparisons for each BLAST-ed protein, and then, perform some summary statistics. Your workflow may look like this:\n",
    "![blast-workflow](https://i.imgur.com/ABREkfc.png)\n",
    "You could, at least in theory, run each part individually and manually make sure the right output data from the right places goes to the right inputs of the next stage, but that's both mentally exhausting and also, not very interesting. Instead, you should apply the model I mentioned in the previous paragraph, and first think about what information each phase is going to need from the previous phase, and then design one script that keeps track of it. This script represents the architecture of your workflow. Then, after that, you would start thinking about how you're going to specifically call the blast pipeline from the script. Learning to do this is a very important tool for HPC specifically, because often there is a limit on both how you can interact with your workflow once it's running, and also how many times you can run it."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 4.2 - Generating lots of random numbers\n",
    "In this example, we're going to simulate the kind of workflow you see in the diagram above. However, we're not going to do it exactly the same way. Running BLAST is just unnecessarily complex, so we're going to replace `BLAST` with generating a random number, `BLASTED` with adding a constant to that random number, and the `summarize` stage will be averaging, as follows:\n",
    "\n",
    "![rng workflow diagram](https://i.imgur.com/8s1CkQ2.png)\n",
    "\n",
    "It is true that this workflow is not really that useful for anything, but it's an interesting learning exercise, and as I said before, the hard part of writing HPC workflows is making the different bits of the workflow interact, and not designing each bit on its own. As an exercise, you could easily replace the randomization bits of this with any other application."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "# RNG workflow\n",
    "import random\n",
    "import os\n",
    "\n",
    "rngfile = \"/home/users/glick/intro-to-hpc/data/rng_add.out\"\n",
    "addfile = \"/home/users/glick/intro-to-hpc/data/rng_add.out\"\n",
    "\n",
    "if os.path.isfile(rngfile):\n",
    "    os.remove(rngfile)\n",
    "if os.path.isfile(addfile):\n",
    "    os.remove(addfile)\n",
    "\n",
    "!touch $rngfile $addfile\n",
    "    \n",
    "def gen_rand_num(uprange, addconst, addfile, rngfile):\n",
    "    num = random.randint(0,uprange)\n",
    "    with open(rngfile, \"a\") as f:\n",
    "        f.write(\"{}\\n\".format(num))\n",
    "    num += addconst\n",
    "    with open(addfile, \"a\") as f:\n",
    "        f.write(\"{}\\n\".format(num))\n",
    "        \n",
    "def summary_stats(addfile):\n",
    "    with open(addfile) as f:\n",
    "        lines = f.readlines()\n",
    "        return sum([int(line.strip()) for line in lines])/ len(lines)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "59.33846153846154\n",
      "CPU times: user 59.1 ms, sys: 247 ms, total: 306 ms\n",
      "Wall time: 290 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "import multiprocessing\n",
    "\n",
    "processes = [multiprocessing.Process(target=gen_rand_num, args=(100, 23, addfile, rngfile)) for _ in range(100)]\n",
    "[process.start() for process in processes]\n",
    "[process.join() for process in processes]\n",
    "\n",
    "print(summary_stats(addfile))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## III. Dataflows With Files\n",
    "As seen before, a file can be used as an intermediate step in a workflow from which to help make your various black boxes interface properly. Because files are permanent and easier to access than memory (albeit much slower), they can be used so that processes that do not share memory space can access information from other processes, without complex inter-process communication, like MPI, except file access is much slower. \n",
    "\n",
    "However, for some processes, the performance penalty is worth it because of how much easier it makes writing the code. Files can also be used as checkpoints. Imagine that our workflow from above actually took about an hour for each of the `RNG_n` tasks. Also, imagine that our workflow failed immediately after the `RNG_n` phase was over. If we had been keeping all of our output in memory, we would have just wasted _n_ hours of time on an HPC system. We may not have been able to get a larger allocation or more compute time for a while, so we would have to sit and wait for more compute space on our cluster. However, we remember that we were writing all of that output to files. Therefore, because the hard part of the workflow was over, we can run a small subset of the workflow, which may take just minutes to run, or may even be so small that we can download our intermediate checkpoints to our laptops and finish running the workflow there. \n",
    "\n",
    "Checkpointing is always a good idea. It's true that it takes some processing power to copy out intermediate result, but if parts of your workflow fail (and this _will_ happen from time to time, even if your workflow is perfect, which it's not, there are still hardware failures, HPC os failures, unscheduled downtime and a host of other reasons), you will be undoubtedly saving time overall. I propose the following diagram as an improved version of our workflow from earlier:\n",
    "![workflow-with-checkpointing](https://i.imgur.com/pF6kyYK.png)\n",
    "This way, if any of our workflow fails, we can still salvage some of the intermediate output."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 4.3 - Using SGE for monte carlo pi\n",
    "In this example, we're going to use SGE to run a large number of random number generators on all of our worker nodes in order to create a workflow using our old friend, the monte carlo pi simulation, to a high degree of accuracy and with high performance. We will also inherently be using checkpointing, because at each stage, we will be writing to a file which we can then later access."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Your job 3596 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3597 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3598 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3599 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3600 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3601 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3602 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3603 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3604 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3605 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3606 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3607 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3608 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3609 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3610 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3611 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3612 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3613 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3614 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3615 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3616 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3617 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3618 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3619 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3620 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3621 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3622 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3623 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3624 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3625 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3626 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3627 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3628 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3629 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3630 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3631 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3632 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3633 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3634 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3635 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3636 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3637 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3638 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3639 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3640 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3641 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3642 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3643 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3644 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3645 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3646 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3647 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3648 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3649 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3650 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3651 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3652 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3653 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3654 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3655 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3656 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3657 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3658 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3659 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3660 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3661 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3662 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3663 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3664 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3665 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3666 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3667 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3668 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3669 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3670 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3671 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3672 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3673 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3674 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3675 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3676 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3677 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3678 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3679 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3680 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3681 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3682 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3683 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3684 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3685 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3686 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3687 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3688 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3689 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3690 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3691 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3692 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3693 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3694 (\"pi_SGE_script.sh\") has been submitted\n",
      "Your job 3695 (\"pi_SGE_script.sh\") has been submitted\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing\n",
    "import os\n",
    "os.environ['SGE_ROOT'] = '/local/cluster/sge'\n",
    "\n",
    "path = \"/home/users/glick/intro-to-hpc/data/pi_SGE_script.sh\"\n",
    "outpath = \"/home/users/glick/intro-to-hpc/data/pi_SGE.out\"\n",
    "\n",
    "if os.path.isfile(path):\n",
    "    os.remove(path)\n",
    "    \n",
    "with open(path, \"w\") as file:\n",
    "     # I know this looks horrible. We need to use str(str object) because of the way that jupyter\n",
    "     # does string interpolation with the literal `$` character.\n",
    "     file.write(str('bc -l <<< \"scale=23 ; ${RANDOM}/32767\" >> /home/users/glick/intro-to-hpc/data/pi_SGE.out\\n'))\n",
    "     file.write(str('bc -l <<< \"scale=23 ; ${RANDOM}/32767\" >> /home/users/glick/intro-to-hpc/data/pi_SGE.out\\n'))\n",
    "\n",
    "for _ in range(100):\n",
    "    !qsub $path -e /dev/null\n",
    "\n",
    "# we have now run a lot of those jobs generating random points. In the next phase of the workflow\n",
    "# we will read from that file and construct an estimate of pi from them."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Esitmated value of Pi: 3.144215530903328 \n"
     ]
    }
   ],
   "source": [
    "# Just for reference, this is actually quite an idiotic way of running this workflow.\n",
    "# in real life, we would never submit this many scripts to the queue scheduling system\n",
    "# We would break up the random number generation into much larger chunks, with each chunk\n",
    "# generating maybe a thousand or a million numbers in serial, because it's so fast.\n",
    "# in any case, we should now have about 2000 random points in our output file/checkpoint file\n",
    "import math as m\n",
    "\n",
    "inside = 0\n",
    "total = 0\n",
    "with open(outpath) as file:\n",
    "    lines = [i.strip() for i in file.readlines()]\n",
    "    for i in range(0, len(lines), 2):\n",
    "        # handle malformed lines\n",
    "        try:\n",
    "            if m.sqrt(float(lines[i])**2 + float(lines[i+1])**2) < 1.0:\n",
    "                inside += 1\n",
    "            total += 1\n",
    "        except Exception as e:\n",
    "            pass\n",
    "# inside / total = pi / 4\n",
    "pi = (float(inside) / total) * 4\n",
    "\n",
    "print(\"Esitmated value of Pi: {} \".format(pi))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## IV. What is a Workflow Management System and Introduction to Parsl.\n",
    "If you think that what we just did is _waaaaay_ to complex to do every day with every workflow you have, you are absolutely right. We had to manually wrangle the resource manager, worry about how the files are structured, think about how the asynchronous nature of resource manager jobs may affect the computation, and many other things that we just don't want to think about, all on top of thinking about the design of the workflow as a whole. Luckily, there are projects that are specifically designed to help HPC users and other parallelism enthusiasts keep track of their workflows in an easy and convenient way. These programs are called _workflow management systems_. A few examples of these are [Pegasus](https://pegasus.isi.edu/), [Fireworks](https://materialsproject.github.io/fireworks/), [Swift-lang(not the apple one)](http://swift-lang.org/main/), and [Parsl](http://parsl-project.org/). Of these, my favorite is easily Parsl. I find that Parsl is by far the easiest to learn, easiest to understand and use, and its performance is quite impressive. Additionally, Parsl is extremely easy to get up and running, because it is a pure Python package with minimal requirements and can be installed with a simple `pip3 install parsl`.\n",
    "\n",
    "\n",
    "So, we know that the author of this document is a Parsl fanboy, but what exactly is Parsl? Well, Parsl is a parallel scripting library for Python. It provides a model by which complex workflows can be represented in an intuitive Python-based control application. It facilitates transparent parallel execution of workflow components (apps) on any distributed or parallel computing system. What does this mean? It means that Parsl handles all of the \"workflowey\" parts of your HPC workflow. It handles the way that all of the different bits of your workflow interact, and it handles the scheduling between them. It has an understanding of what different parts of your apps need the output of other parts, and because of this, Parsl is really good at making sure your workflow is performant, portable, flexible, and readable. Not to mention, all of the Parsl code that you write is very similar to regular Python, which means it's easily read and written by even people who don't really understand HPC or workflows. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sidenote: How Does Parsl Work? It's So Cool...\n",
    "\n",
    "#### If you're not interested in how Parsl works, I won't take it personally. Feel free to skip this section. If you are, then read on!\n",
    "\n",
    "Parsl is a Python-based parallel scripting library that supports development and execution of asynchronous and parallel data-oriented workflows (dataflows). These workflows glue together existing executables (called Apps) and Python functions with control logic written in Python. Parsl brings implicit parallel execution to standard Python scripts. Rather than explicitly defining a graph and/or modifying data structures, instead developers simply annotate Python functions and Apps. Parsl constructs a dynamic, parallel execution graph derived from the implicit linkage between Apps based on shared input/output data objects. Parsl then executes these components when dependencies are met. Parsl is resource-independent, that is, the same Parsl script can be executed on your laptop through to clusters, clouds, and supercomputers.\n",
    "\n",
    "In Parsl, the execution of an App yields futures. These futures can be passed to other Apps as inputs, establishing a dependency. These dependencies are assembled implicitly into directed acyclic graphs, although these are never explicitly expressed. It is important to note that this graph is dynamically built and then update while the Parsl script executes. That is, the graph is not computed in advanced and is only complete when the script finishes executing. Apps that have all their dependencies met are slated for execution (in parallel). This allows Parsl to exploit all parallelism to the fullest extent and at the granularity expressed by the user.\n",
    "\n",
    "Let's parse that a little bit. Parsl allows you to write functions (regular python functions or special ones that make command line calls), and automatically keeps track of them and schedules them to run in a special order, which is calculated by what data is required as inputs for various functions. This kind of parallelism is called _data dependency parallelism_, and it will be described in more detail in the Algorithm Analysis section (Section 6).\n",
    "\n",
    "The way that parsl schedules jobs is described really well by this animated gif: \n",
    "![parsl scheduling alg](http://parsl.readthedocs.io/en/latest/_images/parsl_parallelism.gif)\n",
    "Parsl first calculates the order in which jobs need to run, then, it scales out \"sites\" (parsl has the ability to run different jobs in different places, for example, run some jobs in the cloud, some on your laptop, and some on the supercomputer). Then, it takes those \"sites\". and fills them up with jobs. Each of these sites could have many nodes, each with many cores or one core. You can also define your own custom sites with an easy interface called `Libsubmit`. This extensibility allows Parsl workflows to run literally anywhere, as long as you're willing to write a new `Libsubmit` execution provider for it.\n",
    "\n",
    "I could talk for hours about Parsl and how cool it is, but that's not really the point of this course. If you want to understand more about how Parsl works \"under the hood\", visit the [Parsl docs](https://parsl.readthedocs.io/en/stable/), read [Parsl papers](http://parsl-project.org/publications), or join the [Parsl slack team](http://parsl-project.slack.com). You can also feel free to [contact me](https://glick.cloud/contact.html), and I will be happy to attempt to answer any questions you have about Parsl or any other aspect of this course."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 4.4 - Parsl for simple parallel workflows\n",
    "\n",
    "In this example, we are going to define a very simple workflow using Parsl, and run it on the local machine. The nice thing about Parsl (and other workflow management systems) is that you can write an app designed to be run on a supercomputer, and then run a small version of it on your local computer for testing, and then with only a configuration change, and not any code change, run it on an HPC system. Note how much easier this is than rewriting everything into an SGE script that must be submitted through qsub. You can run arbitrary HPC jobs from python directly without involving Bash at all. In this example, we're going to do a simple \"Hello World\" dataflow with parsl.\n",
    "\n",
    "Parsl’s DataFlowKernel acts as a layer over any pool of execution resources, in our case a pool of [threads](https://en.wikipedia.org/wiki/Thread_(computing).\n",
    "\n",
    "Let’s define a simple python function that returns the string ‘Hello World!’. This function is made an App using the **@App** decorator. The decorator itself takes the type of app (‘python’|’bash’) and the DataFlowKernel object as arguments.\n",
    "\n",
    "Unlike a regular python function, when an App is called, it immediately returns an AppFuture. Futures act as a proxy to the results or exceptions that the App will produce once its execution completes. You can ask a future object its status with future.done() or ask it to wait for its result with the result() call. It is important to note that while the done() call just gives you the current status, the result() call blocks execution until the App is complete and the result is available."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Status:  True\n",
      "Result:  Hello World!\n"
     ]
    }
   ],
   "source": [
    "from parsl import App, ThreadPoolExecutor, DataFlowKernel\n",
    "\n",
    "# Let's create a pool of threads to execute our functions\n",
    "\n",
    "workers = ThreadPoolExecutor(max_workers=4)\n",
    "# We pass the workers to the DataFlowKernel which will execute our Apps over the workers.\n",
    "dfk = DataFlowKernel(executors=[workers])\n",
    "# Here we define our first App function, a simple python app that returns a string\n",
    "@App('python', dfk)\n",
    "def hello ():\n",
    "    return 'Hello World!'\n",
    "\n",
    "app_future = hello()\n",
    "\n",
    "# Check status\n",
    "print(\"Status: \", app_future.done())\n",
    "\n",
    "# Get result\n",
    "print(\"Result: \", app_future.result())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## V. Resource Manager Systems and Resources\n",
    "The cool thing about Parsl isn't just the Future, it's what the Future enables. The fact that Parsl Futures return immediately allows Parsl to compute _Data Dependencies._ When a future created by an App is passed as inputs to another, a data dependency is created. Parsl ensures that Apps are executed as their dependencies are resolved.\n",
    "\n",
    "Parsl is designed to enable the straightforward orchestration of asynchronous tasks into dataflow-based workflows in Python. Parsl manages the parallel execution of these tasks across computation resources when dependencies (e.g., input data dependencies) are met.\n",
    "\n",
    "\n",
    "There are limitations on what Python functions can be converted to apps:\n",
    "\n",
    "- Functions should act only on defined input arguments.\n",
    "- Functions must explicitly import any required modules.\n",
    "- Functions should not use script-level or global variables.\n",
    "- Parsl uses cloudpickle and pickle to serialize Python constructs, such as inputs and outputs to functions. - - \n",
    "- Therefore, Python apps can only use inputs and outputs that can be serialized by cloudpickle or pickle.\n",
    "\n",
    "\n",
    "Parsl is designed to support arbitrary execution providers (e.g., PCs, clusters, supercomputers) as well as arbitrary execution models (e.g., threads, pilot jobs, etc.). That is, Parsl scripts are independent of execution provider or executor. Instead, the configuration used to run the script tells Parsl how to execute apps on the desired environment. Parsl provides a high level abstraction, called a Block, for providing a uniform description of a resource configuration for a particular app or script.\n",
    "\n",
    "This image describes how Parsl \"sites\" scale automatically: ![parsl autoscale](https://parsl.readthedocs.io/en/stable/_images/parsl_scaling.gif)\n",
    "\n",
    "Again, I could write a whole course just about Parsl and how cool it is, and if you agree with me on that, you should check out their docs and tutorials, or contact them or me."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 4.5 - Monte carlo pi with Parsl\n",
    "Let’s see an example of this using our good friend the monte-carlo method to calculate pi. We call 3 iterations of this slow function, and take the average. The dependency chain looks like this :\n",
    "```\n",
    "App Calls    pi()  pi()   pi()\n",
    "              \\     |     /\n",
    "Futures        a    b    c\n",
    "                \\   |   /\n",
    "App Call         mysum()\n",
    "                    |\n",
    "Future            avg_pi\n",
    "```\n",
    "\n",
    "Here we call the function pi() _n_ times, each of which run independently in parallel. We then call the next app mysum() with the three app futures that were returned from the pi() calls. Since mysum() is also a parsl app, it returns an app future immediately, but defers execution (blocks) until all the futures passed to it as inputs have resolved.\n",
    "\n",
    "In this example, we're going to use one of those _Execution Providers_ that I mentioned earlier. We're going to use it to interact with `SGE` on the machine to run distributed code without running `qsub`. Let's see how that works. First, we have to define a configuration specifying how we will talk to our site, and then, we will have to submit apps to it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from parsl import *\n",
    "import logging\n",
    "\n",
    "# Define config and set variables\n",
    "\n",
    "ipp_config = {\n",
    "    \"sites\": [{\n",
    "        \"site\": \"LC_Cluster\",\n",
    "        \"auth\": {\n",
    "            \"channel\": \"local\"\n",
    "        },\n",
    "        \"execution\": {\n",
    "            \"executor\": \"ipp\",\n",
    "            \"provider\": \"sge\",\n",
    "            \"script_dir\": \".scripts\",\n",
    "            \"scriptDir\": \".scripts\",\n",
    "            \"block\": {\n",
    "                \"nodes\": 1,\n",
    "                \"taskBlocks\": 1,\n",
    "                \"walltime\": \"00:05:00\",\n",
    "                \"initBlocks\": 1,\n",
    "                \"minBlocks\": 0,\n",
    "                \"maxBlocks\": 10,\n",
    "                \"scriptDir\": \".\",\n",
    "                \"options\": {\n",
    "                    \"partition\": \"debug\"\n",
    "                }\n",
    "            }\n",
    "        }\n",
    "    }],\n",
    "    \"globals\": {\"lazyErrors\": True},\n",
    "    \"controller\": {\"profile\": \"default\"},\n",
    "}\n",
    "\n",
    "import os\n",
    "os.environ['SGE_ROOT'] = '/local/cluster/sge'\n",
    "\n",
    "dfk = DataFlowKernel(config=ipp_config)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Workflow defined here:\n",
    "\n",
    "\n",
    "@App('python', dfk)\n",
    "def pi(total):\n",
    "    # App functions have to import modules they will use.\n",
    "    import random\n",
    "    # Set the size of the box (edge length) in which we drop random points\n",
    "    edge_length = 10000\n",
    "    center = edge_length / 2\n",
    "    c2 = center ** 2\n",
    "    count = 0\n",
    "\n",
    "    for i in range(total):\n",
    "        # Drop a random point in the box.\n",
    "        x, y = random.randint(1, edge_length), random.randint(1, edge_length)\n",
    "        # Count points within the circle\n",
    "        if (x - center)**2 + (y - center)**2 < c2:\n",
    "            count += 1\n",
    "\n",
    "    return (count * 4 / total)\n",
    "\n",
    "\n",
    "@App('python', dfk)\n",
    "def avg_n(inputs=[]):\n",
    "    return sum(inputs) / len(inputs)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Call the workflow:\n",
    "N = 10\n",
    "sims = [pi(10**6) for i in range(N)]\n",
    "avg_pi = avg_n([task.result() for task in sims])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Print the results\n",
    "print(\"Average: {0:.63f}\".format(avg_pi.result()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Clean up parsl resources\n",
    "dfk.cleanup()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that this takes a long time. There is some overhead involved with Parsl, because it does things like mathematical calculations and automatic submissions of jobs to resource managers. We're also running this to an extreme amount of accuracy. Additionally, Parsl keeps resources running until the workflow exits, which doesn't happen until you tell it to on Jupyter, so re-runs of the workflow will be faster. Feel free to rerun the \"Call the workflow\" cell and the \"Print the results\" cell without any other ones. It'll be faster. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise 4. Simple Dataflow With Parsl"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.6.4"
  },
  "latex_envs": {
   "LaTeX_envs_menu_present": true,
   "autoclose": false,
   "autocomplete": true,
   "bibliofile": "biblio.bib",
   "cite_by": "apalike",
   "current_citInitial": 1,
   "eqLabelWithNumbers": true,
   "eqNumInitial": 1,
   "hotkeys": {
    "equation": "Ctrl-E",
    "itemize": "Ctrl-I"
   },
   "labels_anchors": false,
   "latex_user_defs": false,
   "report_style_numbering": false,
   "user_envs_cfg": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}