Sequences and Loops.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Sequences and Loops in Parsl"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook, we will be walking through examples of implementation of Parsl in different kinds of sequences and loops. Specifically, this notebook contains the implementations of the following Sequences:\n",
    "\n",
    "1. For loops, in a sequence.\n",
    "2. Nested Loops\n",
    "3. Synchronization\n",
    "4. Exclusive Routing (Type 1): Step by step evaluation\n",
    "5. Exclusive Routing (Type 2): Continuous evaluation\n",
    "6. Exclusive Routing (Type 3): Filtering and Aggregration "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "Note that Parsl is not effective if multiple CPU cores aren't available because Parsl's ability to execute tasks in parallel is depenedent on the availability multiple cores."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Cores available: 4\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing\n",
    "print('Cores available: {}'.format(multiprocessing.cpu_count()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Importing Libraries and Configuration"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll be using the htex configuration for Parsl. Read more [here.]( https://github.com/Parsl/parsl/blob/master/parsl/configs/htex_local.py)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<parsl.dataflow.dflow.DataFlowKernel at 0x11843fd90>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import numpy as np\n",
    "import parsl\n",
    "import os\n",
    "\n",
    "from parsl.app.app import python_app, bash_app\n",
    "from parsl.providers import LocalProvider\n",
    "from parsl.channels import LocalChannel\n",
    "\n",
    "from parsl.config import Config\n",
    "from parsl.executors import HighThroughputExecutor\n",
    "\n",
    "config = Config(\n",
    "    executors=[\n",
    "        HighThroughputExecutor(\n",
    "            label=\"htex_local\",\n",
    "            cores_per_worker=1,\n",
    "            provider=LocalProvider(\n",
    "                channel=LocalChannel(),\n",
    "                init_blocks=1,\n",
    "                max_blocks=1,\n",
    "            ),\n",
    "        )\n",
    "    ],\n",
    ")\n",
    "\n",
    "parsl.load(config)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### For loops in Sequence"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since sequences are executed chronologically, their steps can't be executed in parallel. For example, if f(x) is followed by g(x) which is followed by h(x), then we can't execute them in parallel since their inputs depend on the previous functions results. But we can execute multiple sequences at the same time."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For example, Let x be an input that has to be processed by f(x), g(x) and h(x). Then we can execute parallel sequences for 100 values of x as shown below. Note: f(x), g(x) and h(x) are all randomly selected functions can be changed as per the need of the experiment"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<img src=\"./images/parallel_sequences.png\"\n",
    "     style=\"float: left; size: 15px;\" />"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Building the Python Apps for f(x), g(x) and h(x)\n",
    "'''\n",
    "\n",
    "@python_app\n",
    "def f(x):\n",
    "    return (x**2 - 4)\n",
    "\n",
    "@python_app\n",
    "def g(x):\n",
    "    import numpy as np\n",
    "    return (np.sin(x) + np.cos(x))\n",
    "\n",
    "@python_app\n",
    "def h(x):\n",
    "    import numpy as np\n",
    "    return (np.exp(x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "This is out input list of first 100 natural number\n",
    "'''\n",
    "x = np.linspace(1,100,100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "We evaluate 3 For-loops for f(x), g(x) and h(x) on parallel.\n",
    "'''\n",
    "\n",
    "out_f = []\n",
    "out_g = []\n",
    "out_h = []\n",
    "\n",
    "for element in x:\n",
    "    out_f.append(f(element))\n",
    "\n",
    "for element2 in out_f:\n",
    "    out_g.append(g(element2))\n",
    "\n",
    "for element3 in out_g:\n",
    "    out_h.append(h(element3))\n",
    "    \n",
    "out_h = [i.result() for i in out_h]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "An important thing to note in the above example is that we didn't have to evaluate the results of f(x) or g(x) before h(x) using the .result() function. We simply parsed the AppFuture objects into the next python app without evaluation. Thus, it reduces the number of separate evaluations conducted and pushes everything to be conducted in parallel at the end."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "100\n",
      "100\n",
      "100\n"
     ]
    }
   ],
   "source": [
    "# Checking the length of the results\n",
    "\n",
    "print(len(out_f))\n",
    "print(len(out_g))\n",
    "print(len(out_h))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Nested Loops"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A nested loop is a loop within a loop. Here, one loop gives us a result which becomes the input of another loop. For example, The function 'natural numbers' generates a list of the 20 natural numbers after the given input 'x' and 'square roots' generates a list of a given number's square root"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./images/nested_looping.png\"\n",
    "     style=\"float: left; size: 15px;\" />"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "x = np.linspace(1,50,50)\n",
    "\n",
    "'''\n",
    "Building the python apps for initiating the \n",
    "intermediate lists and then evaluating them\n",
    "'''\n",
    "\n",
    "@python_app\n",
    "def natural_numbers(x):\n",
    "    import numpy as np\n",
    "    return np.linspace(x+1, (19+x),20)\n",
    "\n",
    "@python_app\n",
    "def square_roots(x):\n",
    "    return [+x**0.5,-x**0.5]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Using for loops to generate the intermediate lists \n",
    "and using the output to evaluate the second function\n",
    "'''\n",
    "\n",
    "final_array = []\n",
    "\n",
    "for element in x:\n",
    "    output_list = natural_numbers(element) # creating the first list\n",
    "\n",
    "    for element2 in output_list.result():\n",
    "        '''\n",
    "        We have to evaluate the first function.\n",
    "        We then evaluate the second function on the first output list\n",
    "        '''\n",
    "\n",
    "        final_array.append(square_roots(element2))\n",
    "final_array = [i.result() for i in final_array]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1000\n"
     ]
    }
   ],
   "source": [
    "print(len(final_array))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Synchronization"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Synchronization is the merge of existing parallel threads to compute one final result. For example, 100 different natural numbers are first squared and then their average is evaluated"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./images/synchronization.png\"\n",
    "     style=\"float: left; size: 10px;\" />"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Building a python funcion to evaluate the results and then take the final mean\n",
    "'''\n",
    "\n",
    "@python_app\n",
    "def square(x):\n",
    "    return x**2\n",
    "\n",
    "@python_app\n",
    "def average(inputs=[]):\n",
    "    return sum(inputs)/len(inputs)\n",
    "    \n",
    "output_list = []\n",
    "for i in range(1,5):\n",
    "    output_list.append(square(i))\n",
    "\n",
    "mean = average(inputs=output_list).result()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "7.5\n"
     ]
    }
   ],
   "source": [
    "print(mean)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exclusive Routing 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Exclusive Routing is the idea of parsing a result to a parallel branch if it meets a given condition"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./images/exclusive_routing1.png\"\n",
    "     style=\"float: left; size: 15px;\" />"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "inputs = np.linspace(1,100,100)\n",
    "\n",
    "'''\n",
    "Using a check function to route the input to either of the python apps\n",
    "'''\n",
    "\n",
    "def check(x):\n",
    "    if x%3 == 0 and x%5 == 0:\n",
    "        return True\n",
    "    else:\n",
    "        return False\n",
    "    \n",
    "'''\n",
    "Functions Below:\n",
    "\n",
    "square_root: Python app for taking the square root of the function\n",
    "squared: Python app for taking the square of the function\n",
    "'''\n",
    "@python_app\n",
    "def square_root(x):\n",
    "    return x**0.5\n",
    "\n",
    "@python_app\n",
    "def squared(x):\n",
    "    return x**2\n",
    "\n",
    "final_list = []\n",
    "\n",
    "for x in inputs:\n",
    "    if check(x):\n",
    "        final_list.append(square_root(x))\n",
    "    else:\n",
    "        final_list.append(squared(x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Evaluating final results\n",
    "final_list = [i.result() for i in final_list]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "100\n"
     ]
    }
   ],
   "source": [
    "print(len(final_list))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exclusive Routing 2 ( Continuous Evaluation )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This type of exclusive Routing is the idea of parsing a result to given function repeatedly until it meets a condition. For example, a function takes the square root of a number until it becomes less than 2. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./images/exclusive_routing_2.png\"\n",
    "     style=\"float: left; size: 15px;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's consider a simple version of continuous evaluation (as shown in the workflow above):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "@python_app\n",
    "def square_root(x):\n",
    "    '''\n",
    "    Python app for taking the square root of the input.\n",
    "    '''\n",
    "    return x**0.5\n",
    "\n",
    "x = 1000\n",
    "\n",
    "x = square_root(x)\n",
    "\n",
    "while x.result() > 2:\n",
    "    x = square_root(x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1.539926526059492"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "x.result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let us consider a restricted but more efficient and parallel version of continuous evaluation which compares the inputs as part of the parallel thread rather than the main thread:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "@python_app\n",
    "def compare_square_root(x):\n",
    "    # Python app for comparing the square root of the function\n",
    "    if x > 2:\n",
    "        return True\n",
    "    else:\n",
    "        False\n",
    "\n",
    "def convergence(x):\n",
    "    \n",
    "    x = square_root(x)\n",
    "    \n",
    "    for i in range(100):\n",
    "        if compare_square_root(x):\n",
    "            x = square_root(x)\n",
    "        else:\n",
    "            break\n",
    "    \n",
    "    if compare_square_root(x):\n",
    "        return True, x  ## Converged over 100 iterations\n",
    "    else:\n",
    "        return False, x ## Didn't converge over 100 iterations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "inputs = np.linspace(101,200,100)\n",
    "outputs = []\n",
    "\n",
    "for i in inputs: # Execution\n",
    "    state, output = convergence(i)\n",
    "    outputs.append(output)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "outputs = [i.result() for i in outputs]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exclusive Routing 3 ( Filtering and Aggregation)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This type of exclusive Routing is the idea of parsing a result to given function and evaluating if it would be part of the aggregate function (synchronization at the end). For example, a result(x) evaluates the sin(x) + cos(x). In the end, we want the mean of all the positive outputs."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./images/exclusive_routing_3.png\"\n",
    "     style=\"float: left; size: 15px;\" />"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "@python_app\n",
    "def result(x):\n",
    "    '''\n",
    "    Python app for evaluating the sum of sine and cosine of x.\n",
    "    '''\n",
    "    import numpy as np\n",
    "    return np.sin(x) + np.cos(x)\n",
    "\n",
    "@python_app\n",
    "def compare(x):\n",
    "    '''\n",
    "    Python app for comparing the input\n",
    "    '''\n",
    "    if x > 0:\n",
    "        return True\n",
    "    else:\n",
    "        return False\n",
    "\n",
    "@python_app\n",
    "def percentiles(inputs = []):\n",
    "    '''\n",
    "    Python app for computing the \n",
    "    confidence intervals of the inputs list.\n",
    "    '''\n",
    "    import numpy as np\n",
    "    return np.percentile(inputs,25), np.percentile(inputs,75)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "inputs = np.linspace(0,2*np.pi,100)\n",
    "filtered = []\n",
    "\n",
    "for element in inputs: # Computation and comparison of the results in parallel\n",
    "    output = result(element)\n",
    "    if compare(output):\n",
    "        filtered.append(output)\n",
    "        \n",
    "final_result = percentiles(inputs=filtered).result()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(-0.9797614795410088, 0.9999999999999998)\n"
     ]
    }
   ],
   "source": [
    "print(final_result)"
   ]
  }
 ],
 "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}