5 - Distributed Algorithms.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Ch. 5. Distributed Algorithms"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## I. What is a Distributed Algorithm\n",
    "A distributed algorithm is very similar in concept to a parallel algorithm, and since we discussed parallel algorithms at length in chapter 3, this chapter will be fairly brief. However, we will try and apply some of the concepts we learned in ch.3 to distributed systems, and talk about the main principles of distributed systems and distributed computing as opposed to parallel computing.\n",
    "\n",
    "The first and most obvious thing to do is to define what we mean by distributed. In this case, distributed refers to a system or pool of systems made up of many interconnected computers. These computers could be interconnected by anything as fast as the fastest InfiniBand interconnect, or something as slow as dial-up. All that matters for our definition of a distributed system is that it has different components which cannot natively access each others' memory pools, but all of the components of the system still work together towards common tasks. Based on our definition, a cluster is a distributed system in its own right, but others would not consider a cluster a distributed system, and would describe a distributed system as a more loosely connected system, usually connected by the internet rather than a specialized network interconnect. In any case, arguing either side of this debate is not the point of this course, but giving you the tools to make your own conclusions is. The image below represents a distributed system, by our definition.\n",
    "\n",
    "![distributed system architecture](http://www.ejbtutorial.com/wp-content/uploads/2013/09/distributed-system-1024x441.png)\n",
    "\n",
    "We define a cluster as a distributed system because when writing code for a cluster, you need to make the same considerations as you would when writing more code for a \"traditional\" distributed system - you need to worry about how to make sure your code is fault tolerant, scalable, and most importantly, you need to make sure your code can make use of multiple nodes at once, otherwise, your code is not really cluster-optimized. \n",
    "\n",
    "So, now that we know what a distributed system is, briefly, what is a distributed algorithm? Well, a distributed algorithm is, rather boringly, an algorithm designed to be run on a distributed system. As mentioned earlier, when writing a distributed algorithm, there are a lot of new things you need to worry about that you wouldn't on a single system algorithm. \n",
    "\n",
    "You need to worry about what parts of your algorithm can run concurrently, as often, things can be offloaded to remote machines if they can be run concurrently. You need to worry about which parts of your process need to access things from other parts, because you can't depend on all of the remote parts of your code having access to the same memory pool as any other part. Because of this, you need to worry about how you can have the processes communicate with each other. Because that usually depends on some mixture of files and inter-process communication, you need to think about how to minimize the number of times those things happen, because they're slow. As you can see in the graphic above, applications A and C are not true distributed algorithms, because they only make use of one computer at a time, while application B is a distributed application.\n",
    "\n",
    "Writing good distributed algorithms is one of the hardest parts of all of computer science. The distributed algorithms that perform well at scale and accomplish interesting goals are often some of the most complex programs in the world. This complexity is yet another reason why workflow management systems are useful and helpful. We will be using Parsl to help us work with distributed algorithms, because it makes our lives much easier and more fun."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 5.1 - SOMETHING\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Code me up"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## II. Cluster Architecture\n",
    "Remember, in the first chapter, when we talked about networks and I said there would be more network discussion? Well, here's more network discussion. First, recall what we talked about before. You have a bunch of machines and you want then to all talk to each other. The way to do this that would be fastest is by plugging all of them directly into each other. In our diagram below, this represents a \"fully connected\" topology. The main downside of this topology is that it's really expensive and impractical. See that with just six computers, it takes twenty one wires. Most server units do not have very many network ports, so you run into problems there as well. \n",
    "\n",
    "\n",
    "A better way to network machines is to use some clever [Graph Theory](https://en.wikipedia.org/wiki/Graph_theory) to minimize the number of hops it takes to get from any node to any other node. Some popular ones include star, mesh, and tree architectures. For small clusters, a bus is a very good compromise, because while it takes longer to talk through a bus (network switch) than it would to talk straight over the network (call talking through a switch about a hop and a half), it is very fast to talk to any machine on the same switch. Therefore, for small clusters, ones that can fit on just a couple of switches, the bus architecture can rival the fully connected architecture in speed, without the cost or logistical challenges.The cluster this course is hosted on has a bus architecture, because it is quite a small cluster. There are also a number of more complex topologies, such as a torus, which is like a multi-dimensional version of a ring, and is illustrated by this image:\n",
    "![torus architecture](https://upload.wikimedia.org/wikipedia/commons/thumb/3/3f/2x2x2torus.svg/485px-2x2x2torus.svg.png)\n",
    "\n",
    "In a torus like this one, which is a 3D torus, each node is plugged into its three nearest neighbors, and in each direction, it's a loop, meaning the last one is plugged into the first one. Toruses are an attempt to minimize the number of hops it takes for two arbitrary nodes to communicate. In general, each node of an _n_ Dimensional torus will be plugged into its _n_ nearest neighbors.\n",
    "\n",
    "There are even some proprietary architectures, like Cray's dragonfly interconnect, which I do not know much about, other than that it powers many of the largest and most powerful systems in the world.\n",
    "\n",
    "Image for reference:\n",
    "![Network Topologies](https://static9.depositphotos.com/1389325/1074/v/950/depositphotos_10743280-stock-illustration-illustration-of-network-topology-computer.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example - 5.2 Connecting to and Requesting Data From All Hosts\n",
    "In this example, we're going to use Parsl to request information about the nodes we have access to. Recall that we can run (almost) arbitrary Python code from within Parsl apps, and recall that we can easily use Parsl to submit tasks through our resource manager to run lots of jobs on lots of nodes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Parsl config\n",
    "from parsl import *\n",
    "import logging\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": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define app for sys info\n",
    "@App('python', dfk)\n",
    "def sys_info():\n",
    "    outputs = [{}]\n",
    "    import platform\n",
    "    outputs[0].update({\"machine\": platform.machine()})\n",
    "    outputs[0].update({\"platform\": platform.platform()})\n",
    "    outputs[0].update({\"uname\": platform.uname()})\n",
    "    outputs[0].update({\"processor\": platform.processor()})\n",
    "    outputs[0].update({\"system\": platform.system()})\n",
    "    return outputs\n",
    "\n",
    "\n",
    "def info_processor(inputs=[]):\n",
    "    import json\n",
    "    for lis in inputs:\n",
    "        for dictio in lis:\n",
    "            print(json.dumps(dictio, indent=4))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n",
      "{\n",
      "    \"machine\": \"x86_64\",\n",
      "    \"platform\": \"Linux-3.10.0-327.el7.x86_64-x86_64-with-centos-7.2.1511-Core\",\n",
      "    \"uname\": [\n",
      "        \"Linux\",\n",
      "        \"bacon.blt.lclark.local\",\n",
      "        \"3.10.0-327.el7.x86_64\",\n",
      "        \"#1 SMP Thu Nov 19 22:10:57 UTC 2015\",\n",
      "        \"x86_64\",\n",
      "        \"x86_64\"\n",
      "    ],\n",
      "    \"processor\": \"x86_64\",\n",
      "    \"system\": \"Linux\"\n",
      "}\n"
     ]
    }
   ],
   "source": [
    "# Call worfklow\n",
    "\n",
    "infos = [sys_info().result() for _ in range(50)]\n",
    "info_processor(infos)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## III. Messages Between Systems\n",
    "As we mentioned before, it is extremely important for processes to be able to communicate with each other. This is to avoid situations where either parts of a workflow get run too many times or not enough times. Situations like this mean that the workflow is either broken or not performing as fast as it could be. This image shows a distributed system in which processes can communicate over the Internet. ![inter process communication grid](https://upload.wikimedia.org/wikipedia/commons/e/ef/ArchitectureCloudLinksSameSite.png)\n",
    "\n",
    "So, what exactly is interprocess communication? There are many situations where having processes communicate can increase speed, convenience and modularity. Inter process communication is a mechanism which allows processes to communicate each other and synchronize their actions. The communication between these processes can be seen as a method of cooperation between them. Processes can communicate in a number of ways, including shared memory, through files, through pipes, with signaling, over network sockets, and when programmed very specifically, through an incredibly fast and powerful API called MPI - the Message Passing Interface. We don't get into MPI too much in this  course, but it is a very useful skill to learn if you want to be serious about HPC. \n",
    "\n",
    "So, hopefully I've convinced you of the usefulness of inter-process communication, and now I can tell you a bit more about how they work. In the simplest example, shared memory, two (or more) processes can simple use the same data. This (obviously) only works when the two processes are on the same computer with access to the same memory pool. This is how the `multiprocessing` module which we have been using works. Sometimes, processes can write data to files. Then, with proper permissions, any other compute core on the network can reach information from that process. The main downside of that is that both network IO and file IO are very slow and will cause processes to idle. Another option is to have a server-client relationship. This is how Parsl works. The Parsl host (the computer running the Parsl script) acts as a server, telling each client, or worker, what it needs to do. This way, the inter process communication all comes from one server. In the image below, this represents a \"scatter\" or \"broadcast\" type of communication. Then, at the end of the work, Parsl requests the data back from all of the worker machines, representing a \"reduction\" or a \"gather\" method of communication. Server-client communication can often combine the speed of message passing, to an extent, with the ease of other forms of inter process communication. This is why we're going to primarily use server-client communication through Parsl in this course.\n",
    "\n",
    "![Message Passing Diagram](https://computing.llnl.gov/tutorials/mpi/images/collective_comm.gif)\n",
    "\n",
    "\n",
    "### Sidenote on MPI\n",
    "\n",
    "I mentioned MPI earlier, and it's a really important topic in HPC, so I think I should at least explain a little about what it is. MPI is a standardized and portable message-passing interface designed by a group of researchers from academia and industry to function on a wide variety of parallel computing architectures. It's designed to be able to ensure that processes can share data, metadata, and other related information with each other. Some example uses of this are making sure that processes are synchronized, providing work to specific processes, and ensuring that the right data ends up in the right places. MPI has a heirarchical organization of processes, where tasks with a higher priority can give instructions to tasks with a lower priority. While MPI is a bit out of the scope of this course, it's a really good skill to have if you want to do HPC in the real world."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 5.3 - Parsl IPyParallel Tasks\n",
    "As we mentioned before, though Parsl supports MPI through one of its executors, the way we are using it is in the client-server communication model. Parsl uses IPyParallel as a backend for this client-server behavior. IPyParallel is a way to start remote IPython kernels and run Python code through them on remote machines. Using IPyParallel, Parsl is able to create a virtual cluster within our physical cluster and push and pull work and data to each virtual node. Each virtual node takes up one core of the physical machine, and so we can make a large virtual cluster if we need to. In this example, we're going to do some distributed work across the computer. We're going to perform a basic parallel dataflow through Parsl. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Parsl config\n",
    "from parsl import *\n",
    "import logging\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": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "79838\n"
     ]
    }
   ],
   "source": [
    "# App that generates a random number\n",
    "@App('bash', dfk)\n",
    "def generate(outputs=[]):\n",
    "    return \"echo $(( RANDOM )) &> {outputs[0]}\"\n",
    "\n",
    "# App that concatenates input files into a single output file\n",
    "@App('bash', dfk)\n",
    "def concat(inputs=[], outputs=[], stdout=\"data/stdout.txt\", stderr='data/stderr.txt'):\n",
    "    return \"cat {0} > {1}\".format(\" \".join(inputs), outputs[0])\n",
    "\n",
    "# App that calculates the sum of values in a list of input files\n",
    "@App('python', dfk)\n",
    "def total(inputs=[]):\n",
    "    total = 0\n",
    "    with open(inputs[0], 'r') as f:\n",
    "        for l in f:\n",
    "            total += int(l)\n",
    "    return total\n",
    "\n",
    "# Create 5 files with random numbers\n",
    "output_files = []\n",
    "for i in range (5):\n",
    "     output_files.append(generate(outputs=['data/random-%s.txt' % i]))\n",
    "\n",
    "# Concatenate the files into a single file\n",
    "cc = concat(inputs=[i.outputs[0] for i in output_files], outputs=[\"data/all.txt\"])\n",
    "\n",
    "# Calculate the sum of the random numbers\n",
    "total = total(inputs=[cc.outputs[0]])\n",
    "print (total.result())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## IV. Distributed Dataflows\n",
    "When we introduced dataflows before, we mentioned that they are often parallelizable as different \"black boxes\" can run on different cores without any interaction, and without different black boxes blocking each other. In general, but not always, this is also true of dataflows in terms of distributability. If a black box in a dataflow is completely independent of another black box, you can not only parallelize them, but also you can distribute them and run them on different nodes of a machine.\n",
    "\n",
    "Consider the workflow we were working with in Ch. 4: ![heirarchical workflow](https://i.imgur.com/8s1CkQ2.png)\n",
    "\n",
    "Because of the way the arrows are set up, you can see that any black box in the same \"layer\" as another can be run in parallel, as we discussed earlier. In addition, because they are not dependent on each other, they can be easily distributed. The rng_add phase must be run after the rng section, but it can be distributed as well."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 5.4 - Doing Distributed Math With Parsl\n",
    "In this example, we're going to use Parsl to distribute the workflow from Ch. 4. It is still 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": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Parsl config\n",
    "from parsl import *\n",
    "import logging\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": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "86919\n"
     ]
    }
   ],
   "source": [
    "# App that generates a random number\n",
    "@App('bash', dfk)\n",
    "def generate(outputs=[]):\n",
    "    return \"echo $(( RANDOM )) &> {outputs[0]}\"\n",
    "\n",
    "# App that concatenates input files into a single output file\n",
    "@App('bash', dfk)\n",
    "def concat(inputs=[], outputs=[], stdout=\"data/stdout.txt\", stderr='data/stderr.txt'):\n",
    "    return \"cat {0} > {1}\".format(\" \".join(inputs), outputs[0])\n",
    "\n",
    "# App that calculates the sum of values in a list of input files\n",
    "@App('python', dfk)\n",
    "def total(inputs=[]):\n",
    "    total = 0\n",
    "    with open(inputs[0], 'r') as f:\n",
    "        for l in f:\n",
    "            total += int(l)\n",
    "    return total\n",
    "\n",
    "@App('python', dfk)\n",
    "def rng_add(add_const, inputs=[]):\n",
    "    total = 0\n",
    "    lines = []\n",
    "    with open(inputs[0], 'r') as f:\n",
    "        lines = file.readlines()\n",
    "        for line in lines:\n",
    "            line = str(int(line) + add_const)\n",
    "    with open(inputs[0], 'w') as f:\n",
    "        f.writelines(lines)\n",
    "\n",
    "# Create 5 files with random numbers\n",
    "output_files = []\n",
    "for i in range (5):\n",
    "     output_files.append(generate(outputs=['data/random-%s.txt' % i]))\n",
    "\n",
    "# Concatenate the files into a single file\n",
    "cc = concat(inputs=[i.outputs[0] for i in output_files], outputs=[\"data/all.txt\"])\n",
    "\n",
    "# Calculate the sum of the random numbers\n",
    "rng_add(200, inputs=[cc.outputs[0]])\n",
    "\n",
    "total = total(inputs=[cc.outputs[0]])\n",
    "print (total.result())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## V. Large Scale Problems\n",
    "Large scale HPC problems often involve more than Python code. Luckily, that other code is usually callable from the UNIX command line. Parsl supports calling arbitrary \"black box\" applications over the command line through bash apps.  Parsl’s Bash app allows you to wrap execution of external applications from the command-line as you would in a Bash shell. It can also be used to execute Bash scripts directly. To define a Bash app the wrapped Python function must return the command-line string to be executed. Any command-line invocation represented by an arbitrarily long string, can be returned by a function decorated within an `@App` of type bash to be executed. Unlike the Python app, Bash apps communicate by passing files. The decorated bash function provides the same special keyword arguments to manage input and output files."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 5.5 - Basic HPC Dataflow\n",
    "In this example, we're going to use these bash apps to create a simulated HPC workflow calling things from the command line. In a Parsl bash app function, there are a few special reserved keyword arguments:\n",
    "\n",
    "- inputs (List) : A list of strings or DataFutures\n",
    "- outputs (List) : A list of output file paths\n",
    "- stdout (str) : redirects STDOUT to string filename\n",
    "- stderr (str) : redirects STDERR to string filename\n",
    "\n",
    "In addition, if a list of output filenames are provided via the outputs=[], a list of futures corresponding to each filename in the outputs list is made available via the outputs attribute of the app.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Parsl config\n",
    "from parsl import *\n",
    "import logging\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": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Contents of the unsorted.txt file:\n",
      "36,40,87,34,97,50,39,11,11,57,64,49,76,67,89,20,75,80,82,98,10,73,50,70,8,41,75,1,7,55,4,9,6,99,69,75,81,67,58,47,32,77,13,32,64,28,89,78,27,68,11,2,65,70,75,47,31,96,7,32,20,84,3,92,33,96,81,61,43,10,13,16,64,56,89,39,81,4,31,6,93,14,34,47,27,23,56,49,71,85,56,65,68,16,11,54,10,28,64,97,\n",
      "\n",
      "Contents of the sorted output file:\n",
      "1,2,3,4,4,6,6,7,7,8,9,10,10,10,11,11,11,11,13,13,14,16,16,20,20,23,27,27,28,28,31,31,32,32,32,33,34,34,36,39,39,40,41,43,47,47,47,49,49,50,50,54,55,56,56,56,57,58,61,64,64,64,64,65,65,67,67,68,68,69,70,70,71,73,75,75,75,75,76,77,78,80,81,81,81,82,84,85,87,89,89,89,92,93,96,96,97,97,98,99,\n"
     ]
    }
   ],
   "source": [
    "@App('bash', dfk)\n",
    "def sort(unsorted, outputs=[]):\n",
    "    \"\"\"Call sort executable on file `unsorted`\"\"\"\n",
    "    return \"sort -g {} > {}\".format(unsorted, outputs[0])\n",
    "\n",
    "s = sort(os.path.abspath(\"data/unsorted.txt\"), \n",
    "         outputs=[os.path.abspath(\"data/sorted_c.txt\")])\n",
    "\n",
    "output_file = s.outputs[0].result()\n",
    "\n",
    "print(\"Contents of the unsorted.txt file:\")\n",
    "with open('data/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\",\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise 5. Real Live Workflow!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code goes here"
   ]
  }
 ],
 "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
}