3-parsl-workflows.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# **Parsl: Workflow Tutorial**\n",
    "\n",
    "In this tutorial, you will be able to first try a few Parsl dataflows (examples 1-4) on your local machine, to get a sense of the library. Then, in examples 5-7 you will run similar dataflows on any resource you may have access to, such as clouds (Amazon Web Services) and HPC systems,  to see how more complex workflows can be expressed with Parsl."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Tutorial Section One\n",
    "\n",
    "This section will provide a walk-through for getting a simple \"mock\" science application running using Parsl on your local machine. \n",
    "|\n",
    "Note: the mock science apps are included in the apps directory of the tutorial repository."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example 1: Run a single application using  Parsl\n",
    "\n",
    "The first Parsl script runs `simulate.sh` to generate a single random number. It writes the number to stdout. You can run the following script repeatedly to see different \"simulated\" results. \n",
    "\n",
    "![Part 1](figures/part01.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import parsl\n",
    "from parsl.app.app import python_app, bash_app\n",
    "from parsl.configs.local_threads import config\n",
    "\n",
    "parsl.load(config)\n",
    "\n",
    "@bash_app\n",
    "def mysim(stdout=(\"output/p1.out\", \"w\"), stderr=(\"output/p1.err\", \"w\")):\n",
    "    \"\"\"Set this example up as a bash app by returning the \n",
    "    command line app to be called, in this case simulate\"\"\"\n",
    "    return \"app/simulate\"\n",
    "\n",
    "# call the mysim app and wait for the result\n",
    "mysim().result()\n",
    "\n",
    "# open the output file and read the result\n",
    "with open('output/p1.out', 'r') as f:\n",
    "    print(f.read())\n",
    "\n",
    "parsl.clear()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example 2: Run a single Python function using Parsl\n",
    "\n",
    "The second example mirrors the first, however instead of using an external application it uses a Python function to simulate a science app again by generating a random number. In this case the Python app returns the simulated value as a Python object rather than using an external file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import parsl\n",
    "from parsl.app.app import python_app, bash_app\n",
    "from parsl.configs.local_threads import config\n",
    "\n",
    "parsl.load(config)\n",
    "\n",
    "@python_app\n",
    "def mysim():\n",
    "    from random import randint\n",
    "    \"\"\"Generate a random integer and return it\"\"\"\n",
    "    return randint(1,100)\n",
    "\n",
    "print(mysim().result())\n",
    "\n",
    "parsl.clear()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example 3:  Running an ensemble of many apps in parallel with a loop\n",
    "\n",
    "The third example shows how Parsl can be used to run many simulations in parallel. In this case, we define the same Parsl simulation app (`simulate`). The Python script then loops calling this app. Parsl ensures that each independent instance of the app will execute in parallel. Note: rather than use stdout the simulation app redirects the output to a specified file.\n",
    "\n",
    "![Part 3](figures/part03.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import parsl\n",
    "from parsl.app.app import python_app, bash_app\n",
    "from parsl.configs.local_threads import config\n",
    "from parsl.data_provider.files import File\n",
    "\n",
    "parsl.load(config)\n",
    "\n",
    "@bash_app\n",
    "def mysim(outputs=[], stdout=(\"output/p3.out\", \"w\"), stderr=(\"output/p3.err\", \"w\")):\n",
    "    \"\"\"Call simulate and return results in the output file\"\"\"\n",
    "    return 'app/simulate > {0}'.format(outputs[0].filepath)\n",
    "\n",
    "# loop to execute the simulation app 5 times\n",
    "results = []\n",
    "for i in range(5):\n",
    "    out_file = \"output/p3_sim_{0}\".format(i)\n",
    "    results.append(mysim(outputs=[File(out_file)]))\n",
    "\n",
    "# print each job status, initially all are running\n",
    "print (\"Job Status: {}\".format([r.done() for r in results]))\n",
    "\n",
    "# wait for all apps to complete\n",
    "[r.result() for r in results]\n",
    "\n",
    "# print each job status, they will now be finished\n",
    "print (\"Job Status: {}\".format([r.done() for r in results]))\n",
    "\n",
    "# collect up the output files and print their values\n",
    "outputs = [r.outputs[0] for r in results]\n",
    "for o in outputs:\n",
    "    with open(o.filename, 'r') as f:\n",
    "        print(f.read().strip())\n",
    "        \n",
    "parsl.clear()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example 4: Analyzing results of a parallel ensemble\n",
    "\n",
    "After all the parallel simulations in an ensemble run have completed, it is typically necessary to gather and analyze their results with some kind of post-processing analysis program or script. The fourth example introduces such a postprocessing step. In this case, the files created by all of the parallel runs of `simulate` will be averaged by the \"analysis application\" `stats`. \n",
    "\n",
    "Note: in this example we do not block on the outputs of the simulation app, rather the futures are passed directly to the analysis application. Parsl will manage these dependencies and only execute the analysis app when the simulation app completes.\n",
    "\n",
    "![Part 4](figures/part04.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import parsl\n",
    "from parsl.app.app import python_app, bash_app\n",
    "from parsl.configs.local_threads import config\n",
    "from parsl.data_provider.files import File\n",
    "\n",
    "parsl.load(config)\n",
    "\n",
    "@bash_app\n",
    "def mysim(outputs=[],\n",
    "          stdout=(\"output/p4_sim.out\", \"w\"),\n",
    "          stderr=(\"output/p4_sim.err\", \"w\")):\n",
    "    \"\"\"Call simulate and return results in the output file\"\"\"\n",
    "    return 'app/simulate > {0}'.format(outputs[0].filepath)\n",
    "    \n",
    "\n",
    "@bash_app\n",
    "def stats(inputs=[],\n",
    "          outputs=[],\n",
    "          stderr=('output/p4_stats.err', \"w\"),\n",
    "          stdout=('output/p4_stats.out', \"w\")):\n",
    "    \"\"\"call stats with all simulation results as inputs\"\"\"\n",
    "    return \"app/stats {0} > {1}\".format(\" \".join(i.filepath for i in inputs), outputs[0].filepath)\n",
    "\n",
    "\n",
    "# call the simulation app 5 times\n",
    "results = []\n",
    "for i in range(5):\n",
    "    out_file = \"output/p4_sim_{0}\".format(i)\n",
    "    results.append(mysim(outputs=[File(out_file)]))\n",
    "\n",
    "# collect the output data futures\n",
    "sim_outputs = [r.outputs[0] for r in results]\n",
    "\n",
    "# run the stats app\n",
    "s = stats(inputs=sim_outputs, outputs=[File(\"output/p4_stats.txt\")])\n",
    "\n",
    "# wait for the result\n",
    "s.result()\n",
    "\n",
    "# print the result\n",
    "with open('output/p4_stats.txt', 'r') as f:\n",
    "    print(f.read())\n",
    "\n",
    "parsl.clear()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Tutorial Section Two\n",
    "\n",
    "This section introduces execution of Parsl scripts on remote computational resources. \n",
    "\n",
    "Parsl supports a variety of resource providers as well as methods for submitting workload to those resources (e.g., pilot jobs). Example configurations for compute resources such as UChicago Midway, NERSC Cori, NCSA BlueWaters, and ANL Cooley are included in the config directory of the tutorial repository. \n",
    "\n",
    "A Parsl script can be executed from a login node by using these configuration files directly. Parsl also supports a more advanced submission model in which a SSH channel is used to submit wokloads from an external machine (e.g., your laptop). In this case the machine on which the script is executed must be accessible from the cluster (i.e., you will need to ensure that firewall rules allow pilot job connections back to your laptop). "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example 5: Running a simple app using pilot jobs\n",
    "\n",
    "In this example we show how one might develop a script to run on a remote resource. We first develop a Parsl script for sorting a file. We initially run this script using the local ThreadPoolExecutor on a laptop, we subsequently extend this example to submit the job via the HighThroughputExecutor pilot job model locally and then on a remote resource. We finally show how Parsl's SSH channel can be used to connect to the remote resource directly from your laptop. \n",
    "\n",
    "This script uses the `sort` application to sort the numbers from an unsorted file. \n",
    "\n",
    "### Example 5.a: Running the sort app locally using threads\n",
    "\n",
    "First we use the local thread executor to run the `sort` command locally."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import parsl\n",
    "from parsl.app.app import python_app, bash_app\n",
    "from parsl.configs.local_threads import config\n",
    "from parsl.data_provider.files import File\n",
    "\n",
    "parsl.load(config)\n",
    "\n",
    "@bash_app\n",
    "def sort(unsorted, outputs=[]):\n",
    "    \"\"\"Call sort executable on the input file\"\"\"\n",
    "    return \"sort -g {} > {}\".format(unsorted.filepath, outputs[0].filepath)\n",
    "\n",
    "# call the sort app on the unsorted.txt file\n",
    "# save the results to a_sorted.txt\n",
    "s = sort(File(\"input/unsorted.txt\"), outputs=[File(\"output/a_sorted.txt\")])\n",
    "\n",
    "# wait for the result\n",
    "output_file = s.outputs[0].result()\n",
    "\n",
    "# print the contents of the unsorted and sorted files\n",
    "print(\"Contents of the unsorted.txt file:\")\n",
    "with open('input/unsorted.txt', 'r') as f:\n",
    "    print(f.read().replace(\"\\n\",\",\"))\n",
    "    \n",
    "print(\"\\nContents of the sorted output file:\")\n",
    "with open(output_file, 'r') as f:\n",
    "    print(f.read().replace(\"\\n\",\",\"))\n",
    "    \n",
    "parsl.clear()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 5.b: Running the sort app locally using pilot jobs\n",
    "\n",
    "We now use IPyParallel to run the `sort` command using a pilot job model. In this case we use the `htex_local` configuration to tell Parsl to use local HighThroughputExecutor. \n",
    "\n",
    "You will notice that apart from the configuration and instantiation of the DataFlowKernal this script is identical to the previous script. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import parsl\n",
    "from parsl.app.app import python_app, bash_app\n",
    "from parsl.configs.htex_local import config\n",
    "from parsl.data_provider.files import File\n",
    "\n",
    "parsl.load(config)\n",
    "\n",
    "@bash_app\n",
    "def sort(unsorted, outputs=[]):\n",
    "    \"\"\"Call sort executable on the input file\"\"\"\n",
    "    return \"sort -g {} > {}\".format(unsorted.filepath, outputs[0].filepath)\n",
    "\n",
    "s = sort(File(\"input/unsorted.txt\"), outputs=[File(\"output/b_sorted.txt\")])\n",
    "\n",
    "output_file = s.outputs[0].result()\n",
    "\n",
    "print(\"Contents of the unsorted.txt file:\")\n",
    "with open('input/unsorted.txt', 'r') as f:\n",
    "    print(f.read().replace(\"\\n\",\",\"))\n",
    "    \n",
    "print(\"\\nContents of the sorted output file:\")\n",
    "with open(output_file, 'r') as f:\n",
    "    print(f.read().replace(\"\\n\",\",\"))\n",
    "    \n",
    "parsl.clear()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 5.c: Running the sort app on a cluster using pilot jobs\n",
    "\n",
    "We now take the previous example and run it on a cluster using HighThroughputExecutor. To run this example you will need to execute the script from a login node and uncomment the configuration needed for your cluster. \n",
    "\n",
    "The following example uses a basic configuration for TACC's Frontera. You will need to update the scheduler_options with your allocation and the conda environment specified by worker_init.\n",
    "\n",
    "Note: you will need to either install the Parsl library on that login node or use an existing environment. You will also need to download the parsl-tutorial repository and ensure that it is avaialble on the worker nodes via a shared file system."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import parsl\n",
    "from parsl.app.app import python_app, bash_app\n",
    "from parsl.data_provider.files import File\n",
    "import os\n",
    "from parsl.config import Config\n",
    "from parsl.channels import LocalChannel\n",
    "from parsl.providers import SlurmProvider\n",
    "from parsl.executors import HighThroughputExecutor\n",
    "from parsl.launchers import SrunLauncher\n",
    "from parsl.addresses import address_by_hostname\n",
    "\n",
    "\"\"\" This config assumes that it is used to launch parsl tasks from the login nodes\n",
    "of Frontera at TACC. Each job submitted to the scheduler will request 2 nodes for 10 minutes.\n",
    "\"\"\"\n",
    "config = Config(\n",
    "    executors=[\n",
    "        HighThroughputExecutor(\n",
    "            label=\"frontera_htex\",\n",
    "            address=address_by_hostname(),\n",
    "            max_workers=56,\n",
    "            provider=SlurmProvider(\n",
    "                cmd_timeout=60,    \n",
    "                channel=LocalChannel(),\n",
    "                nodes_per_block=1,\n",
    "                partition='development',\n",
    "                scheduler_options='#SBATCH -A ALLOCATION',   # Enter allocation\n",
    "                worker_init='conda activate CONDA_ENV', # Enter conda environment\n",
    "                walltime='00:10:00',\n",
    "                launcher=SrunLauncher(),\n",
    "            ),\n",
    "        )\n",
    "    ],\n",
    ")\n",
    "\n",
    "parsl.load(config)\n",
    "\n",
    "@bash_app\n",
    "def sort(unsorted, outputs=[]):\n",
    "    \"\"\"Call sort executable on the input file\"\"\"\n",
    "    return \"sleep 30; sort -g {} > {}\".format(unsorted.filepath, outputs[0].filepath)\n",
    "\n",
    "s = sort(File(os.path.abspath(\"input/unsorted.txt\")), \n",
    "         outputs=[File(os.path.abspath(\"output/sorted_c.txt\"))])\n",
    "\n",
    "output_file = s.outputs[0].result()\n",
    "\n",
    "print(\"Contents of the unsorted.txt file:\")\n",
    "with open('input/unsorted.txt', 'r') as f:\n",
    "    print(f.read().replace(\"\\n\",\",\"))\n",
    "    \n",
    "print(\"\\nContents of the sorted output file:\")\n",
    "with open(output_file, 'r') as f:\n",
    "    print(f.read().replace(\"\\n\",\",\"))\n",
    "\n",
    "parsl.clear()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example 6: MPI Hello\n",
    "\n",
    "The final example is a basic \"Hello World!\" example that shows you how to run MPI applications. Here we have a simple MPI code `mpi_hello.c` that has each MPI rank sleep for a user-specified duration and then print the processor name on which the rank is executing followed by \"Hello World!\". \n",
    "\n",
    "The following script is designed to be run from the login node. Like the previous examples you will need to update the shared directory. You may optionally use an SSH channel by following the previous instructions. If running from the login node you will not need the remote configuration details. \n",
    "\n",
    "Given the range of MPI libraries installed on each site you will need to specify the site-specific MPI compiler (e.g., mpicc, cc) and the way to execute the MPI job (e.g., mpirun, srun). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import parsl\n",
    "from parsl.app.app import python_app, bash_app\n",
    "\n",
    "# Update to import config for your machine\n",
    "from config.midway import config\n",
    "\n",
    "import os\n",
    "\n",
    "parsl.load(config)\n",
    "\n",
    "remote = False\n",
    "shared_dir = 'SHARED_DIRECTORY' # path to the tutorial repository\n",
    "\n",
    "@bash_app\n",
    "def compile_app(dirpath, stdout=None, stderr=None, compiler=\"mpicc\"):\n",
    "    \"\"\"Compile mpi app using site-specific compiler.\n",
    "    E.g.,  midway compiler = mpicc, Cori compiler= cc\n",
    "    \"\"\"\n",
    "    return '''cd {0}; make clean; make CC={1} '''.format(dirpath, compiler)\n",
    "\n",
    "@bash_app\n",
    "def mpi_hello(dirpath, launcher=\"mpirun\", app=\"mpi_hello\", nproc=20, outputs=[]):\n",
    "    \"\"\"Call compiled mpi executable with local mpilib.\n",
    "    Works natively for openmpi mpiexec, mpirun, orterun, oshrun, shmerun\n",
    "    mpiexec is default\"\"\"\n",
    "    import os\n",
    "    if launcher == \"mpirun\" :\n",
    "        return \"mpirun -np {} {} &> {};\".format(nproc, os.path.join(dirpath,app), outputs[0])\n",
    "    elif launcher == \"srun\" :\n",
    "        return \"srun -n {} ./{} &> {};\".format(nproc, os.path.join(dirpath,app), outputs[0])\n",
    "\n",
    "# complile the app and wait for it to complete (.result())\n",
    "compile_app(dirpath=os.path.join(shared_dir, \"mpi_apps\"),\n",
    "            stdout=os.path.join(shared_dir, \"mpi_apps.compile.out\"),\n",
    "            stderr=os.path.join(shared_dir, \"mpi_apps.compile.err\",),\n",
    "            compiler='mpicc'\n",
    "           ).result()\n",
    "\n",
    "# run the mpi app\n",
    "hello = mpi_hello(os.path.join(shared_dir, \"mpi_apps\"),\n",
    "                  launcher=\"mpirun\",\n",
    "                  outputs=[os.path.join(shared_dir, \"mpi_apps\", \"hello.txt\")])\n",
    "\n",
    "output_file = hello.outputs[0].result()\n",
    "\n",
    "# if running remotely using SSH, copy the file back to the host\n",
    "if remote:\n",
    "    dfk.executor.execution_provider.channel.pull_file(output_file, '.')\n",
    "\n",
    "# read the result file\n",
    "with open(output_file, 'r') as f:\n",
    "     print(f.read())\n",
    "\n",
    "parsl.clear()"
   ]
  }
 ],
 "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.9.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}