distributed_computing_with_julia.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# <span style=\"color:red\">Distributed Computing with Julia</span>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "# Master/Worker architecture\n",
    "\n",
    "Julia relies on the Master/Worker paradigm. While the master process controls the distributed computation, the **workers performs all the computation**.\n",
    "\n",
    "<img src=\"figures/master-worker.png\" width=\"400\"height=\"400\" />\n",
    "\n",
    "\n",
    "* Each process has an associated identifier\n",
    "    * **Master** PID = 1\n",
    "    * **Workers** PID >= 2 \n",
    "* Only workers execute distributed calls\n",
    "    * Except in a single-process environment\n",
    "        * Process 1 is both master and worker\n",
    "        * I.e., Julia interactive prompt\n",
    "\n",
    "\n",
    "Use [nprocs()](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.nprocs) to know the number of process in use:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "nprocs()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can either add more processes via **SSH** or add more **local processors** to take advantage of multi-core computers with [addprocs(...)](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.addprocs):\n",
    "\n",
    "_Remark_: _\"Julia 0.4 supports \"auto\\*host\" which will launch as many workers on host as cores\"_. Se [discusion at users mailing list](https://groups.google.com/forum/#!msg/julia-users/XH80LzxCERY/jMhbw5YpjvEJ) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2-element Array{Int64,1}:\n",
       " 2\n",
       " 3"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "addprocs(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then list again to see the just-added processes:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "nprocs()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can get the list of processes with [procs()](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.procs):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3-element Array{Int64,1}:\n",
       " 1\n",
       " 2\n",
       " 3"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "procs()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "On the other hand, you can list only the [workers()](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.workers):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2-element Array{Int64,1}:\n",
       " 2\n",
       " 3"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "workers()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's run some simple function on all workers. The following code calls the [myid()](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.myid) function on all workers:\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Hi, it's the Master (process ID=1) listing the workers...\n",
      "Listing worker #2\n",
      "Listing worker #3\n"
     ]
    }
   ],
   "source": [
    "println(\"Hi, it's the Master (process ID=$(myid())) listing the workers...\")\n",
    "for pid in workers()\n",
    "    println(\"Listing worker #\", pid)\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we will remove both workers (the processes we previously added) with [rmprocs(pids...)](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.rmprocs):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       ":ok"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rmprocs(3,2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1-element Array{Int64,1}:\n",
       " 1"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "procs()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# [Clusters](http://docs.julialang.org/en/release-0.3/manual/parallel-computing/?highlight=cluster#clustermanagers)\n",
    "\n",
    "\n",
    "## Basics\n",
    "\n",
    "* Julia processes run on Julia [Clusters](http://docs.julialang.org/en/release-0.3/manual/parallel-computing/?highlight=cluster#clustermanagers)\n",
    "    * **Local** cluster\n",
    "        * Leverage local multi-cores \n",
    "    * **Distributed** cluster\n",
    "        * On top of SSH/TCP/IP connections\n",
    "* Clusters can be customized\n",
    "\n",
    "* Clusters can be created as Julia starts. For instance, **to create a 2-core local cluster**:\n",
    "\n",
    "```\n",
    "$ julia -p 2\n",
    "```\n",
    "\n",
    "* A **distributed SSH cluster** can be created as follows:\n",
    "\n",
    "```\n",
    "julia --machinefile HOSTS_FILE common_code.jl\n",
    "```\n",
    "\n",
    "* Julia will start worker processes remotely through a password-less SSH login on the computers listed at the HOSTS_FILE file. In addition, it will deploy the *common_code.jl* file on all computers.\n",
    "\n",
    "* It is important to remark that these computers can be any computer reachable by password-less SSH logins. For example, LAN computers, Amazon EC2 instances, [Docker](http://docker.com) containers, and do forth.\n",
    "\n",
    "## Advanced\n",
    "\n",
    "* See [Cluster Manager Interface](http://docs.julialang.org/en/latest/stdlib/parallel/#cluster-manager-interface)\n",
    "    <!--\n",
    "    * [launch](http://docs.julialang.org/en/latest/stdlib/parallel/#Base.launch)\n",
    "     -->\n",
    "    \n",
    "<!--\n",
    "# Cloud support\n",
    "\n",
    "## Virtual machines (VMs)\n",
    "\n",
    "* [`Amazon Web Services (AWS)`](https://github.com/amitmurthy/AWS.jl)\n",
    "* [`OpenStack.jl`](https://github.com/Keno/OpenStack.jl/blob/master/src/OpenStack.jl)\n",
    "    * Uses [OpenStack Restful API](http://developer.openstack.org/api-ref-compute-v2.html)\n",
    "\n",
    "## Containers\n",
    "\n",
    "* [Infra.jl](https://github.com/gsd-ufal/CloudArray.jl/blob/master/src/Infra.jl)\n",
    "    * Set of **management functions for managing Julia workers on Docker**\n",
    "* [Docker.jl](https://github.com/Keno/Docker.jl)\n",
    "    * Wrapper to [Docker HTTP API](https://docs.docker.com/reference/api/docker_remote_api/)\n",
    "\n",
    "## Higher-level support \n",
    "-->"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# [<span style=\"color:red\">Julia support for distributed communication</span>](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#general-parallel-computing-support)\n",
    "\n",
    "## Overview\n",
    "\n",
    "Let's read the first paragraphs of the [Julia Official Documentation on Parallel Computing](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#general-parallel-computing-support)\n",
    "\n",
    "* \"Julia’s implementation of message passing is **different from other environments such as MPI**. \n",
    "* **Communication in Julia is generally “one-sided”**\n",
    "    * The programmer needs **to explicitly manage only one process in a two-process operation**. \n",
    "    * These operations typically **do not look like “message send” and “message receive”** but rather resemble **higher-level operations like calls to user functions**.\n",
    "\n",
    "* Parallel programming in Julia is built on two primitives: **remote references** and **remote calls**. \n",
    "    * A **remote reference** is an object that can be used from any process to **refer to an object stored on a particular process.**\n",
    "    * A **remote call** is a **request by one process to call a certain function** on certain arguments on another (possibly the same) process. \n",
    "        * A remote call **returns a remote reference** to its result. \n",
    "        * Remote calls return immediately [**Asynchronous**]; the process that made the call proceeds to its next operation while the remote call happens somewhere else. \n",
    "        * You can wait for a remote call to finish by calling wait on its remote reference[**Synchronous**].\n",
    "        * You can obtain the full value of the result using **fetch**. \n",
    "        * You can store a value to a remote reference using **put**.\n",
    "\"\n",
    "\n",
    "#### In other words...\n",
    "\n",
    "* It is **not a send-receive paradigm**\n",
    "    * There is no receive: it assumes that communication is managed by the calling process\n",
    "    * The calling process calls remote functions or expressions\n",
    "    \n",
    "#### Further supports\n",
    "\n",
    "* **Macros** to make communication easier\n",
    "* **Tasks**\n",
    "    * Coroutines, green threads that run on a single actual thread\n",
    "    * **Tasks can be combined with remote calls for parallel programming**\n",
    "    * Julia multithreading is being developed at Julia 0.5     \n",
    "    \n",
    "## Programming model\n",
    "\n",
    "\n",
    "* Programming model\n",
    "    * **Primitive: remote calls**\n",
    "        * A remote request similiar to _send_\n",
    "            * However, remote calls specify the operation to be performed\n",
    "            * Similar to embed the _receive_ operation into _send_\n",
    "        * [`remotecall(id, func, args...)`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.remotecall)\n",
    "            * `id` is the process that will perform the computation\n",
    "            * `func` is the function or expression to be executed\n",
    "            * `args` ar the function arguments\n",
    "    * **Abstraction: remote reference**\n",
    "        * Remote calls return a remote reference (`RemoteRef`)\n",
    "        * `RemoteRef` can be used by any processes\n",
    "    \n",
    "\n",
    "\n",
    "## [Remote calls](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.remotecall)\n",
    "\n",
    "* Julia low-level communication primitives\n",
    "* Remote calls can be performed **Synchronously** or **Asynchronously**.\n",
    "\n",
    "In order to put workers to execute some code, let's add two processes since we currently have a single (master) process:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2-element Array{Int64,1}:\n",
       " 4\n",
       " 5"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "addprocs(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As processes have unique ID, remark that the new processes' IDs are 4 and 5.\n",
    "\n",
    "\n",
    "\n",
    "### Asynchronous remote calls\n",
    "\n",
    "* The calling process **does not wait** the worker computation\n",
    "    * **Non-blocking call**\n",
    "* [`remotecall(id, func, args...)`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.remotecall)\n",
    "    * E.g., let's put *worker 4* to calculate the *square root of 144*:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 261,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RemoteRef{Channel{Any}}(4,1,7925)"
      ]
     },
     "execution_count": 261,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "remotecall(4, sqrt, 144)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 262,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RemoteRef{Channel{Any}}(4,1,7926)"
      ]
     },
     "execution_count": 262,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "async_sqtr = remotecall(4, sqrt, 144)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 263,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "12.0"
      ]
     },
     "execution_count": 263,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fetch(async_sqtr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Synchronous remote calls\n",
    "\n",
    "* **Synchronous** communications block the sender till they receive the response\n",
    "    * Makes the caller **wait**\n",
    "* [`remotecall_wait(id, func, args...)`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.remotecall_wait)\n",
    "    * The following example puts *worker 2* to *sleep 3 seconds* and waits:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 264,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "This message waited worker 5\n"
     ]
    }
   ],
   "source": [
    "wait_till_worker_2_sleeps_5_secs = remotecall_wait(2, sleep, 3)\n",
    "\n",
    "println(\"This message waited for worker 2\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This call blocked the caller process (master, process whose ID is 1) for 3 seconds. \n",
    "\n",
    "Remark that the result from the Output appeared 3 seconds later with the RemoteRef.\n",
    "\n",
    "Now let's execute the same command **asynchronously** (non-blocking call):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 266,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "This message did NOT wait for worker 2\n"
     ]
    }
   ],
   "source": [
    "dont_wait_till_worker_2_sleeps_5_secs = remotecall(2, sleep, 3)\n",
    "\n",
    "println(\"This message did NOT wait for worker 2\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### @sync and @async\n",
    "\n",
    "* Macros that ease the construction of synchronous and asynchronous calls.\n",
    "* These macros return [`Tasks`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#tasks).\n",
    "* [**`@sync`**](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.@sync)\n",
    "    * Makes the calling process **wait** for a function/expression execution\n",
    "        * _Wait until all dynamically-enclosed uses of @async, @spawn, @spawnat and @parallel are complete._\n",
    "    * Similiar to [monitors](https://en.wikipedia.org/wiki/Monitor_%28synchronization%29)\n",
    "    * Optionally, we can use  `wait` instead:\n",
    "        * [`wait([x])`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.wait)\n",
    "            * _Block the current task until some event occurs, depending on the type of the argument._\n",
    "        * [`timedwait(testcb, secs; pollint)`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.timedwait)\n",
    "            * _Waits till `testcb` returns true or for `secs` seconds, whichever is earlier._\n",
    "            * _`testcb` is polled every `pollint` seconds._\n",
    "            \n",
    "* [**`@async`**](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.@async)\n",
    "    * Free the calling process (Task) from a blocked operation\n",
    "    * Similar to running a command in background in Bash\n",
    "   \n",
    "\n",
    "### Remark on Asynchronous vs. Synchronous calls\n",
    "\n",
    "* Usage depends on the distributed application\n",
    "* Use **Synchronous** calls when you need to wait a result to carry on the computation:\n",
    "    * Workflows\n",
    "    * Mutual exclusion: consistency, integrity\n",
    "* Otherwise, use **Asynchronous** calls since blocking communications degrade the system performance and can eventually imply [deadlocks](https://en.wikipedia.org/wiki/Deadlock)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## [Remote reference](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.RemoteRef)\n",
    "\n",
    "* A [`RemoteRef`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.RemoteRef) is a **reference to an object stored at a remote process**.\n",
    "    * It is NOT in the sense of the ~~Object-oriented Programming (OOP)~~\n",
    "* All `RemoteRef` are **visible by all processes**\n",
    "* **You can put to and get values from it**\n",
    "    * It's like a data/function/expression recipient\n",
    "\n",
    "\n",
    "### `RemoteRef` operations:\n",
    "\n",
    "* [`RemoteRef()`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.RemoteRef)\n",
    "* [`RemoteRef(p)`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.RemoteRef)\n",
    "* [`fetch(RemoteRef)`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.fetch)\n",
    "    * _Wait for and get the value of a remote reference._\n",
    "* [`put!(RemoteRef, value)`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.put!)\n",
    "    * _Store a value to a remote reference. Implements **“shared queue of length 1”** semantics: if a value is already present, **blocks** until the value is removed with take!. Returns its first argument._\n",
    "    * Example: let's create a RemoteRef that will run asynchronously (in background) till we put some value to it:\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 267,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Task (waiting) @0x000000010f07b340"
      ]
     },
     "execution_count": 267,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r = RemoteRef()\n",
    "@async (fetch(r), println(\"hi\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Now let's put some value to `r`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 268,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hi"
     ]
    },
    {
     "data": {
      "text/plain": [
       "RemoteRef{Channel{Any}}(1,1,7932)"
      ]
     },
     "execution_count": 268,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "put!(r,0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* [`take!(RemoteRef)`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.take!)\n",
    "    * _Fetch the value of a remote reference, removing it so that the reference is empty again._\n",
    "    * **Blocks** the caller if the `RemoteRef` is empty\n",
    "* [`isready(r::RemoteRef)`](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.isready)\n",
    "    * _Determine whether a `RemoteRef` has a value stored to it._\n",
    "    * **Blocks** the caller\n",
    "    * Should be used **only on `RemoteRef`s that are assigned once**.\n",
    "        * `isready(..)` might cause race conditions -> consistency issues\n",
    "    * Example: checking whether a `put!` was performed:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 275,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "false"
      ]
     },
     "execution_count": 275,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rr = RemoteRef()\n",
    "@async put!(rr, remotecall_fetch(4, sleep, 10))\n",
    "isready(rr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you wait 10 seconds, the condition will be set to true:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 274,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "true"
      ]
     },
     "execution_count": 274,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "isready(rr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## @spawn and @spawnat\n",
    "\n",
    "* The macro [@spawn](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.@spawn) spawns (runs) an expression/function at a worker.\n",
    "* All function computation is done at worker side"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 276,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RemoteRef{Channel{Any}}(2,1,7941)"
      ]
     },
     "execution_count": 276,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "remote_matrix = @spawn rand(2,2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we fetch the `remote_matrix` variable with fetch(...):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 277,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2x2 Array{Float64,2}:\n",
       " 0.341608  0.745106\n",
       " 0.992225  0.202961"
      ]
     },
     "execution_count": 277,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fetch(remote_matrix)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you need to execute an expression over the `remote_matrix`, you can still use @spawn combined with fetch(...):\n",
    "\n",
    "For instance:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 278,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RemoteRef{Channel{Any}}(3,1,7943)"
      ]
     },
     "execution_count": 278,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "remote_matrix_plus_one = @spawn 1 .+ fetch(remote_matrix)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we fetch the result:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 284,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3"
      ]
     },
     "execution_count": 284,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fetch(remote_matrix_plus_one)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you are curious about where is `remote_matrix_plus_one`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 287,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "remote_matrix_plus_one.where = 3"
     ]
    },
    {
     "data": {
      "text/plain": [
       "3"
      ]
     },
     "execution_count": 287,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "remote_matrix_plus_one.where"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you need to have more control of the distributed computation, you may use the macro [@spawnat](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.@spawnat) which allows to spawn an expression/function at a specific worker.\n",
    "\n",
    "In the following example, we execute the sqtr(...) function at the worker whose ID is 4:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 280,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RemoteRef{Channel{Any}}(4,1,7945)"
      ]
     },
     "execution_count": 280,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sqrt_at_4 = @spawnat 4 sqrt(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we fetch the `sqrt_at_4` variable with `fetch` and get the worker where `sqrt_at_4` is stored:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 289,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 289,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fetch(sqrt_at_4)\n",
    "\n",
    "sqrt_at_4.where"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## [@fetch](http://docs.julialang.org/en/latest/stdlib/parallel/#Base.@fetch) and [@fetchfrom](http://docs.julialang.org/en/latest/stdlib/parallel/#Base.@fetch)\n",
    "\n",
    "We can also use spawn and fetch at once with macro [@fetch(...)](http://docs.julialang.org/en/latest/stdlib/parallel/#Base.@fetch) which is equivalent to **`fetch(@spawn expr)`**:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 297,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8"
      ]
     },
     "execution_count": 297,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@fetch 2 + 6"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For more control, use the macro [@fetchfrom(...)](http://docs.julialang.org/en/latest/stdlib/parallel/#Base.@fetchfrom) which is equivalent to **`fetch(@spawnat p expr)`**:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5159780352"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@fetchfrom 5 12^9"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## [@everywhere](http://julia.readthedocs.org/en/latest/stdlib/parallel/#Base.@everywhere)\n",
    "\n",
    "* The [@everywhere](http://julia.readthedocs.org/en/latest/stdlib/parallel/#Base.@everywhere) macro forces a command/expression to run on all processes.\n",
    "* Example: let's print everywhere (at every workers) processes' IDs:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n",
      "\tFrom worker 2:\t2\n",
      "\tFrom worker 3:\t3\n"
     ]
    }
   ],
   "source": [
    "addprocs(2)\n",
    "@everywhere println(myid())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For instance, let's define on all processes a function that prints workers' IDs:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "@everywhere who_are_you() = println(\"I'm your father... oops, I'm worker $(myid())\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we call it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RemoteRef{Channel{Any}}(4,1,9)"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 4:\tI'm your father... oops, I'm worker 4\n"
     ]
    }
   ],
   "source": [
    "@spawn who_are_you()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* @everywhere can also be used to load modules on all processes:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 303,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#Pkg.add(\"Calculus\")\n",
    "@everywhere using Calculus"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 301,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "12.000000000099304"
      ]
     },
     "execution_count": 301,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@fetchfrom 4 derivative(x-> x^3, 2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* For loading source files, use [`require(..)`](http://julia.readthedocs.org/en/latest/stdlib/base/#Base.require) instead."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "ename": "LoadError",
     "evalue": "LoadError: UndefVarError: MyCode not defined\nwhile loading In[45], in expression starting on line 1",
     "output_type": "error",
     "traceback": [
      "LoadError: UndefVarError: MyCode not defined\nwhile loading In[45], in expression starting on line 1",
      ""
     ]
    }
   ],
   "source": [
    "require(MyCode.jl)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##[@parallel](http://docs.julialang.org/en/latest/stdlib/parallel/#Base.@parallel)\n",
    "\n",
    "* Parallelizes a for loop or a comprenhesion automatically among the available workers.\n",
    "    * Split the range\n",
    "    * Distribute the range to each process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 305,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4-element Array{Any,1}:\n",
       " RemoteRef{Channel{Any}}(4,1,7983)\n",
       " RemoteRef{Channel{Any}}(5,1,7984)\n",
       " RemoteRef{Channel{Any}}(2,1,7985)\n",
       " RemoteRef{Channel{Any}}(3,1,7986)"
      ]
     },
     "execution_count": 305,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 4:\t104\n",
      "\tFrom worker 5:\t105\n",
      "\tFrom worker 4:\t104\n",
      "\tFrom worker 5:\t105\n",
      "\tFrom worker 4:\t104\n",
      "\tFrom worker 5:\t105\n",
      "\tFrom worker 2:\t102\n",
      "\tFrom worker 2:\t102\n",
      "\tFrom worker 3:\t103\n",
      "\tFrom worker 3:\t103\n"
     ]
    }
   ],
   "source": [
    "@everywhere plus_my_id(x) = x + myid()\n",
    "\n",
    "a = 100\n",
    "@parallel for i=1:10\n",
    "    println(plus_my_id(a))\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* **`@parallel`** may also performs a **reduce function on each worker and the final reduction at the calling process** (e.g., Master):\n",
    "    * Optionally, @parallel may take a “reducer” as its first argument\n",
    "    * The results from each remote reducer will be aggregated using the reducer locally\n",
    "    * Tasks are assigned asynchronously when no reducer operation is used\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 306,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1033"
      ]
     },
     "execution_count": 306,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@parallel (+) for i=1:10\n",
    "   plus_my_id(a)\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "# [<span style=\"color:red\">Tasks (coroutines)</span>](http://docs.julialang.org/en/latest/stdlib/parallel/#tasks)\n",
    "\n",
    "\n",
    "* There is **no thread**.\n",
    "    * Multi-threading is begin developed at Julia 0.5\n",
    "    * Julia Tasks are **not thread-safe**.\n",
    "* Julia tasks are **green threads (coroutines)**\n",
    "    * Tasks do **not** spam over different CPUs\n",
    "    * Julia scheduler decides which tasks runs at a given time\n",
    "    * Tasks can yield, notify, schedule, etc.\n",
    "* Useful when you need concurrency programming with no actual parallelization (**pseudo-parallelization** instead)\n",
    "    * Context change is very fast as tasks belongs to a single kernel thread\n",
    "* Julia relies on [\"**remote function execution** as opposed to message passing\"](https://github.com/ViralBShah/JuliaConIndia2015/blob/master/JuliaCon_Parallel_Workshop.ipynb)\n",
    "    * “... Our current approach to concurrent computation is a **shared-nothing multi-process model** that also works across machines.” [Stefan Karpinski at julia-users mailing list](https://groups.google.com/forum/#!searchin/julia-users/kernel$20thread/julia-users/Y9XYxI5Bgi4/yUSszl1ajioJ)\n",
    "\n",
    "\n",
    "<img src=\"figures/julia-tasks.png\" width=\"300\" />\n",
    "\n",
    "\n",
    "<!--\n",
    "## Task as a programming abstraction \n",
    "\n",
    "* States\n",
    " * Runnable (“created”)\n",
    " * Running\n",
    " * Done\n",
    " * ...\n",
    "\n",
    "* Communication **primitives**:\n",
    " * **produce**(..)\n",
    " * **consume**(..) -> synchronous call\n",
    " * From [Amit Murthy ipynb](https://github.com/ViralBShah/JuliaConIndia2015/blob/master/JuliaCon_Parallel_Workshop.ipynb)\n",
    "     * \"produce and consume allow a producer task to \"feed\" one or more consumers\n",
    "     * produce blocks till a consumer removes a value\n",
    "     * consume blocks till the producer adds a value or exits\"\n",
    "* High-level API (macros)\n",
    "    * @task\n",
    "    * @sync\n",
    "    * @async\n",
    "    * @schedule\n",
    "        * \"@schedule schedules an expression to be run like @async except: \n",
    "            * launched task is not waited on by enclosing @sync blocks\n",
    "            * does not \"localize\" variables\"\n",
    "    * wait\n",
    "    * \n",
    "    \n",
    "## Channels \n",
    "# See: https://github.com/ViralBShah/JuliaConIndia2015/blob/master/JuliaCon_Parallel_Workshop.ipynb\n",
    "-->"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Examples\n",
    "\n",
    "####Using the **`produce`** function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 171,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Task (runnable) @0x000000010f0782b0"
      ]
     },
     "execution_count": 171,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "t = Task( () -> produce(5) )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 172,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5"
      ]
     },
     "execution_count": 172,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "consume(t)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Implementing a **function with [produce](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.produce)**:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "producer (generic function with 1 method)"
      ]
     },
     "execution_count": 149,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "function producer(p)\n",
    "    for i in 1:p\n",
    "        produce(i)\n",
    "    end\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Create a task with the implemented function with [Task](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.Task)\n",
    "    * We have to use an anonymous function (lambda) for that: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 173,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Task (runnable) @0x000000010f07bb20"
      ]
     },
     "execution_count": 173,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "t2 = Task( () -> producer(4) )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Run the task with [consume](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.consume)\n",
    "    * Let's perform a single consume:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 174,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 174,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "consume(t2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Now, let's consume the last products from function `producer` and calculates their square root in a `for` loop.\n",
    "    * Each iteration performs a `consume(tsk1)` call:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 175,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "t=2; t^2=4\n",
      "t=3; t^2=9\n",
      "t=4; t^2=16\n"
     ]
    }
   ],
   "source": [
    "for t in t2\n",
    "    println(\"t=\", t, \"; t^2=\", t^2)\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Macro @task \n",
    "\n",
    "* Optionally, you can create a task with macro [@task](http://docs.julialang.org/en/release-0.3/stdlib/parallel/#Base.@task)\n",
    "* Let's define a function which produces [Fibonacci numbers](https://en.wikipedia.org/wiki/Fibonacci_number):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 176,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "fib_producer (generic function with 1 method)"
      ]
     },
     "execution_count": 176,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "function fib_producer(n)\n",
    "    a, b = (0, 1)\n",
    "    for i = 1:n\n",
    "        produce(b)\n",
    "        a, b = (b, a + b)\n",
    "    end\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Now we create a 6-element Fibonacci series:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 195,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Task (runnable) @0x000000010a80a0e0"
      ]
     },
     "execution_count": 195,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "t3 = @task fib_producer(6)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* We can perform a multiple consume in a for loop\n",
    "* For instance, `println(t)` implicitly calls `consume(t)` as we iterate over `tsk2`:\n",
    "\n",
    "<!--\n",
    "a = @async 1+2\n",
    "println(\"non-blocking call\")\n",
    "consume(a)\n",
    "-->"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 196,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n",
      "1\n",
      "2\n",
      "3\n",
      "5\n",
      "8\n"
     ]
    }
   ],
   "source": [
    "for t in t3; println(t); end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example: concurrent (parallel) clients accessing an authentication server\n",
    "\n",
    "### Goal\n",
    "\n",
    "Show how we can **spawn and manage parallel codes wrapped as tasks**\n",
    "\n",
    "### Defining type, variables, and functions\n",
    "\n",
    "* `type User`\n",
    "* The list of users\n",
    "* Functions specific to the authentication server (Worker 2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 221,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#\n",
    "# workers 2:n\n",
    "#\n",
    "\n",
    "@everywhere type User\n",
    "    username\n",
    "    password\n",
    "end\n",
    "\n",
    "#\n",
    "# worker 2\n",
    "#\n",
    "\n",
    "@everywhere list_of_users = User[]\n",
    "\n",
    "@everywhere function auth(user::User)\n",
    "    for u in list_of_users\n",
    "        if user.username == u.username && user.password == u.password\n",
    "            return true\n",
    "        end\n",
    "    end\n",
    "    return false\n",
    "end\n",
    "\n",
    "@everywhere function add_user(user::User)\n",
    "    push!(list_of_users,user)\n",
    "end\n",
    "\n",
    "@everywhere function list_users()\n",
    "    for u in list_of_users\n",
    "        @show u.username\n",
    "    end\n",
    "end\n",
    "\n",
    "@everywhere function remove_user(username::AbstractString )\n",
    "    i=1\n",
    "    for u in list_of_users\n",
    "        if username == u.username \n",
    "            deleteat!(list_of_users,i)\n",
    "            return true\n",
    "        end\n",
    "        i=i+1\n",
    "    end\n",
    "    return false\n",
    "end\n",
    "\n",
    "@everywhere function remove_all_users()\n",
    "    empty!(list_of_users)\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Defining client `sign_up` function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 243,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#\n",
    "# workers 3:n\n",
    "#\n",
    "\n",
    "@everywhere function sign_up()\n",
    "    u = User(\"client-$(myid())\",\"$(myid())\")\n",
    "    remotecall(2,add_user,u)\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Each Worker signs up once"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 244,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "for w in workers()\n",
    "    if w != 2 # skipping Worker 2   \n",
    "        remotecall(w,sign_up)\n",
    "    end\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Listing user at authentication server (at Worker 2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 245,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RemoteRef{Channel{Any}}(2,1,7868)"
      ]
     },
     "execution_count": 245,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 2:\tu.username = \"client-3\"\n",
      "\tFrom worker 2:\tu.username = \"client-4\"\n",
      "\tFrom worker 2:\tu.username = \"client-5\"\n"
     ]
    }
   ],
   "source": [
    "remotecall(2,list_users)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Cleaning up list of users"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 255,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RemoteRef{Channel{Any}}(2,1,7917)"
      ]
     },
     "execution_count": 255,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "remotecall(2, remove_all_users)\n",
    "remotecall(2,list_users)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using `Task` to define sign up functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 247,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "@everywhere function launch_sing_ups(n_of_sign_ups)\n",
    "#    users_task = Task( () -> launch_sing_ups_task(n_of_sign_ups) )\n",
    "    users_task = @task launch_sing_ups_task(n_of_sign_ups)\n",
    "    for u in users_task\n",
    "        remotecall_wait(2,add_user,u)\n",
    "    end\n",
    "end\n",
    "\n",
    "@everywhere function launch_sing_ups_task(n_of_sign_ups)\n",
    "    for n=1:n_of_sign_ups\n",
    "        u = User(\"client-$(myid())-$n\",\"$(myid())-$n\")\n",
    "        println(\"User created at process $(myid())\")\n",
    "        produce(u)\n",
    "    end\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Launching sing up Tasks then listing users"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 256,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 3:\tUser created at process 3\n",
      "\tFrom worker 4:\tUser created at process 4\n",
      "\tFrom worker 5:\tUser created at process 5\n",
      "\tFrom worker 3:\tUser created at process 3\n",
      "\tFrom worker 4:\tUser created at process 4\n",
      "\tFrom worker 5:\tUser created at process 5\n"
     ]
    }
   ],
   "source": [
    "for w in workers()\n",
    "    if w != 2 # skipping Worker 2 \n",
    "        remotecall(w,launch_sing_ups, 2)\n",
    "    end\n",
    "end"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 257,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RemoteRef{Channel{Any}}(2,1,7921)"
      ]
     },
     "execution_count": 257,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 2:\tu.username = \"client-3-1\"\n",
      "\tFrom worker 2:\tu.username = \"client-4-1\"\n",
      "\tFrom worker 2:\tu.username = \"client-5-1\"\n",
      "\tFrom worker 2:\tu.username = \"client-3-2\"\n",
      "\tFrom worker 2:\tu.username = \"client-5-2\"\n",
      "\tFrom worker 2:\tu.username = \"client-4-2\"\n"
     ]
    }
   ],
   "source": [
    "remotecall(2,list_users)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# [<span style=\"color:red\">Distributed Arrays</span>](https://github.com/JuliaParallel/DistributedArrays.jl)\n",
    "\n",
    "\n",
    "## Introduction\n",
    "\n",
    "* **Distributed Memory** model\n",
    "    * A multi-dimensional array stored at distributed processes\n",
    "    * No shared memory\n",
    "* **Dense Arrays**\n",
    "    * E.g., non-sparse matrices\n",
    "* All processes know about DistributedArrays\n",
    "* **Only metadata is exchanged**\n",
    "    * Each process holds a part of data the whole DistributedArray\n",
    "\n",
    "```\n",
    "2 2 2 2\n",
    "3 3 3 3\n",
    "4 4 4 4\n",
    "5 5 5 5\n",
    "```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4-element Array{Int64,1}:\n",
       " 2\n",
       " 3\n",
       " 4\n",
       " 5"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "addprocs(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "@everywhere using DistributedArrays"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating DistributedArrays\n",
    "\n",
    "Use `Array` constructors with **d** at the beginning:\n",
    "\n",
    "    dzeros(100,100,10)\n",
    "    dones(100,100,10)\n",
    "    drand(100,100,10)\n",
    "    drandn(100,100,10)\n",
    "    dfill(x,100,100,10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10-element DistributedArrays.DArray{Float64,1,Array{Float64,1}}:\n",
       " 0.0\n",
       " 0.0\n",
       " 0.0\n",
       " 0.0\n",
       " 0.0\n",
       " 0.0\n",
       " 0.0\n",
       " 0.0\n",
       " 0.0\n",
       " 0.0"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "d = dzeros(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.0"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "d[4]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Float64"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "i = d[4]\n",
    "typeof(i)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating from an `Array`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10-element DistributedArrays.DArray{Float64,1,Array{Float64,1}}:\n",
       " 0.725223 \n",
       " 0.660856 \n",
       " 0.858211 \n",
       " 0.0869098\n",
       " 0.652438 \n",
       " 0.656942 \n",
       " 0.253129 \n",
       " 0.978451 \n",
       " 0.295076 \n",
       " 0.529512 "
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "arr = rand(10)\n",
    "darr = distribute(arr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Converting `DistributedArrays` to `Array`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10-element Array{Float64,1}:\n",
       " 0.112362\n",
       " 0.319637\n",
       " 0.639904\n",
       " 0.65593 \n",
       " 0.985051\n",
       " 0.854342\n",
       " 0.39156 \n",
       " 0.571651\n",
       " 0.908335\n",
       " 0.624201"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "arr2 = convert(Array, darr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array{Float64,1}"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "typeof(arr2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Getting information on where DistributedArrays are stored\n",
    "\n",
    "* **`localpart(a::DArray)`** obtains the locally-stored portion of a  DArray.\n",
    "\n",
    "* **`localindexes(a::DArray)`** gives a tuple of the index ranges owned by the local process."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4-element Array{Tuple{UnitRange{Int64}},1}:\n",
       " (1:3,) \n",
       " (4:5,) \n",
       " (6:7,) \n",
       " (8:10,)"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "localpart(darr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 2:\t(1:3,)\n",
      "\tFrom worker 3:\t(4:5,)\n",
      "\tFrom worker 5:\t(8:10,)\n",
      "\tFrom worker 4:\t(6:7,)\n"
     ]
    }
   ],
   "source": [
    "for w in workers()\n",
    "    @spawnat w println(localindexes(darr))\n",
    "end"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 2:\t[0.7252225541990764,0.6608562929625534,0.8582114993759962]\n",
      "\tFrom worker 3:\t[0.08690975874742035,0.6524378568671487]\n",
      "\tFrom worker 5:\t[0.9784513095442966,0.29507647226882905,0.5295115322543209]\n",
      "\tFrom worker 4:\t[0.6569416593382784,0.25312886183129724]\n"
     ]
    }
   ],
   "source": [
    "for w in workers()\n",
    "    @spawnat w println(localpart(darr))\n",
    "end"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "d.chunks = RemoteRef[RemoteRef{Channel{Any}}(2,1,1140),RemoteRef{Channel{Any}}(3,1,1141),RemoteRef{Channel{Any}}(4,1,1142),RemoteRef{Channel{Any}}(5,1,1143)]"
     ]
    },
    {
     "data": {
      "text/plain": [
       "4-element Array{Int64,1}:\n",
       " 2\n",
       " 3\n",
       " 4\n",
       " 5"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "d.cuts = [[1,4,6,8,11]]\n",
      "d.dims = (10,)\n",
      "d.indexes = [(1:3,),(4:5,),(6:7,),(8:10,)]\n",
      "d.pids = [2,3,4,5]\n"
     ]
    }
   ],
   "source": [
    "@show d.chunks\n",
    "@show d.cuts\n",
    "@show d.dims\n",
    "@show d.indexes\n",
    "@show d.pids"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 406,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "search: "
     ]
    },
    {
     "data": {
      "text/latex": [
       "\\section{DistributedArrays.jl}\n",
       "\\href{https://travis-ci.org/JuliaParallel/DistributedArrays.jl}{\\begin{figure}\n",
       "\\centering\n",
       "\\includegraphics{https://travis-ci.org/JuliaParallel/DistributedArrays.jl.svg?branch=master}\n",
       "\\caption{Build Status}\n",
       "\\end{figure}\n",
       "}\n",
       "Distributed Arrays for Julia\n",
       "***NOTE*** Distributed Arrays will only work on the latest development version of Julia (v0.4.0-dev).\n",
       "\\texttt{DArray}s have been removed from Julia Base library in v0.4 so it is now necessary to import the \\texttt{DistributedArrays} package on all spawned processes.\n",
       "\\begin{verbatim}\n",
       "@everywhere using DistributedArrays\n",
       "\\end{verbatim}\n",
       "\\subsection{Distributed Arrays}\n",
       "Large computations are often organized around large arrays of data. In these cases, a particularly natural way to obtain parallelism is to distribute arrays among several processes. This combines the memory resources of multiple machines, allowing use of arrays too large to fit on one machine. Each process operates on the part of the array it owns, providing a ready answer to the question of how a program should be divided among machines.\n",
       "Julia distributed arrays are implemented by the \\texttt{DArray} type. A \\texttt{DArray} has an element type and dimensions just like an \\texttt{Array}. A \\texttt{DArray} can also use arbitrary array-like types to represent the local chunks that store actual data. The data in a \\texttt{DArray} is distributed by dividing the index space into some number of blocks in each dimension.\n",
       "Common kinds of arrays can be constructed with functions beginning with \\texttt{d}:\n",
       "\\begin{verbatim}\n",
       "    dzeros(100,100,10)\n",
       "    dones(100,100,10)\n",
       "    drand(100,100,10)\n",
       "    drandn(100,100,10)\n",
       "    dfill(x,100,100,10)\n",
       "\\end{verbatim}\n",
       "In the last case, each element will be initialized to the specified value \\texttt{x}. These functions automatically pick a distribution for you. For more control, you can specify which processes to use, and how the data should be distributed:\n",
       "\\begin{verbatim}\n",
       "    dzeros((100,100), workers()[1:4], [1,4])\n",
       "\\end{verbatim}\n",
       "The second argument specifies that the array should be created on the first four workers. When dividing data among a large number of processes, one often sees diminishing returns in performance. Placing \\texttt{DArray}{\\textbackslash} s on a subset of processes allows multiple \\texttt{DArray} computations to happen at once, with a higher ratio of work to communication on each process.\n",
       "The third argument specifies a distribution; the nth element of this array specifies how many pieces dimension n should be divided into. In this example the first dimension will not be divided, and the second dimension will be divided into 4 pieces. Therefore each local chunk will be of size \\texttt{(100,25)}. Note that the product of the distribution array must equal the number of processes.\n",
       "\\begin{itemize}\n",
       "\\item \\texttt{distribute(a::Array)} converts a local array to a distributed array.\n",
       "\\end{itemize}\n",
       "\\begin{itemize}\n",
       "\\item \\texttt{localpart(a::DArray)} obtains the locally-stored portion of a  \\texttt{DArray}.\n",
       "\\end{itemize}\n",
       "\\begin{itemize}\n",
       "\\item \\texttt{localindexes(a::DArray)} gives a tuple of the index ranges owned by the local process.\n",
       "\\end{itemize}\n",
       "\\begin{itemize}\n",
       "\\item \\texttt{convert(Array, a::DArray)} brings all the data to the local process.\n",
       "\\end{itemize}\n",
       "Indexing a \\texttt{DArray} (square brackets) with ranges of indexes always creates a \\texttt{SubArray}, not copying any data.\n",
       "\\subsection{Constructing Distributed Arrays}\n",
       "The primitive \\texttt{DArray} constructor has the following somewhat elaborate signature:\n",
       "\\begin{verbatim}\n",
       "    DArray(init, dims[, procs, dist])\n",
       "\\end{verbatim}\n",
       "\\texttt{init} is a function that accepts a tuple of index ranges. This function should allocate a local chunk of the distributed array and initialize it for the specified indices. \\texttt{dims} is the overall size of the distributed array. \\texttt{procs} optionally specifies a vector of process IDs to use. \\texttt{dist} is an integer vector specifying how many chunks the distributed array should be divided into in each dimension.\n",
       "The last two arguments are optional, and defaults will be used if they are omitted.\n",
       "As an example, here is how to turn the local array constructor \\texttt{fill} into a distributed array constructor:\n",
       "\\begin{verbatim}\n",
       "    dfill(v, args...) = DArray(I->fill(v, map(length,I)), args...)\n",
       "\\end{verbatim}\n",
       "In this case the \\texttt{init} function only needs to call \\texttt{fill} with the dimensions of the local piece it is creating.\n",
       "\\texttt{DArray}s can also be constructed from multidimensional \\texttt{Array} comprehensions with the \\texttt{@DArray} macro syntax.  This syntax is just sugar for the primitive \\texttt{DArray} constructor:\n",
       "\\begin{verbatim}\n",
       "julia> [i+j for i = 1:5, j = 1:5]\n",
       "5x5 Array{Int64,2}:\n",
       " 2  3  4  5   6\n",
       " 3  4  5  6   7\n",
       " 4  5  6  7   8\n",
       " 5  6  7  8   9\n",
       " 6  7  8  9  10\n",
       "\n",
       "julia> @DArray [i+j for i = 1:5, j = 1:5]\n",
       "5x5 DistributedArrays.DArray{Int64,2,Array{Int64,2}}:\n",
       " 2  3  4  5   6\n",
       " 3  4  5  6   7\n",
       " 4  5  6  7   8\n",
       " 5  6  7  8   9\n",
       " 6  7  8  9  10\n",
       "\\end{verbatim}\n",
       "\\subsection{Distributed Array Operations}\n",
       "At this time, distributed arrays do not have much functionality. Their major utility is allowing communication to be done via array indexing, which is convenient for many problems. As an example, consider implementing the \"life\" cellular automaton, where each cell in a grid is updated according to its neighboring cells. To compute a chunk of the result of one iteration, each process needs the immediate neighbor cells of its local chunk. The following code accomplishes this::\n",
       "\\begin{verbatim}\n",
       "    function life_step(d::DArray)\n",
       "        DArray(size(d),procs(d)) do I\n",
       "            top   = mod(first(I[1])-2,size(d,1))+1\n",
       "            bot   = mod( last(I[1])  ,size(d,1))+1\n",
       "            left  = mod(first(I[2])-2,size(d,2))+1\n",
       "            right = mod( last(I[2])  ,size(d,2))+1\n",
       "\n",
       "            old = Array(Bool, length(I[1])+2, length(I[2])+2)\n",
       "            old[1      , 1      ] = d[top , left]   # left side\n",
       "            old[2:end-1, 1      ] = d[I[1], left]\n",
       "            old[end    , 1      ] = d[bot , left]\n",
       "            old[1      , 2:end-1] = d[top , I[2]]\n",
       "            old[2:end-1, 2:end-1] = d[I[1], I[2]]   # middle\n",
       "            old[end    , 2:end-1] = d[bot , I[2]]\n",
       "            old[1      , end    ] = d[top , right]  # right side\n",
       "            old[2:end-1, end    ] = d[I[1], right]\n",
       "            old[end    , end    ] = d[bot , right]\n",
       "\n",
       "            life_rule(old)\n",
       "        end\n",
       "    end\n",
       "\\end{verbatim}\n",
       "As you can see, we use a series of indexing expressions to fetch data into a local array \\texttt{old}. Note that the \\texttt{do} block syntax is convenient for passing \\texttt{init} functions to the \\texttt{DArray} constructor. Next, the serial function \\texttt{life_rule} is called to apply the update rules to the data, yielding the needed \\texttt{DArray} chunk. Nothing about \\texttt{life_rule} is \\texttt{DArray}{\\textbackslash} -specific, but we list it here for completeness::\n",
       "\\begin{verbatim}\n",
       "    function life_rule(old)\n",
       "        m, n = size(old)\n",
       "        new = similar(old, m-2, n-2)\n",
       "        for j = 2:n-1\n",
       "            for i = 2:m-1\n",
       "                nc = +(old[i-1,j-1], old[i-1,j], old[i-1,j+1],\n",
       "                       old[i  ,j-1],             old[i  ,j+1],\n",
       "                       old[i+1,j-1], old[i+1,j], old[i+1,j+1])\n",
       "                new[i-1,j-1] = (nc == 3 || nc == 2 && old[i,j])\n",
       "            end\n",
       "        end\n",
       "        new\n",
       "    end\n",
       "\\end{verbatim}\n",
       "\\subsection{Numerical Results of Distributed Computations}\n",
       "Floating point arithmetic is not associative and this comes up when performing distributed computations over \\texttt{DArray}s.  All \\texttt{DArray} operations are performed over the \\texttt{localpart} chunks and then aggregated. The change in ordering of the operations will change the numeric result as seen in this simple example:\n",
       "\\begin{verbatim}\n",
       "julia> addprocs(8);\n",
       "\n",
       "julia> @everywhere using DistributedArrays\n",
       "\n",
       "julia> A = fill(1.1, (100,100));\n",
       "\n",
       "julia> sum(A)\n",
       "11000.000000000013\n",
       "\n",
       "julia> DA = distribute(A);\n",
       "\n",
       "julia> sum(DA)\n",
       "11000.000000000127\n",
       "\n",
       "julia> sum(A) == sum(DA)\n",
       "false\n",
       "\\end{verbatim}\n",
       "The ultimate ordering of operations will be dependent on how the Array is distributed.\n"
      ],
      "text/markdown": [
       "# DistributedArrays.jl\n",
       "\n",
       "[![Build Status](https://travis-ci.org/JuliaParallel/DistributedArrays.jl.svg?branch=master)](https://travis-ci.org/JuliaParallel/DistributedArrays.jl)\n",
       "\n",
       "Distributed Arrays for Julia\n",
       "\n",
       "***NOTE*** Distributed Arrays will only work on the latest development version of Julia (v0.4.0-dev).\n",
       "\n",
       "`DArray`s have been removed from Julia Base library in v0.4 so it is now necessary to import the `DistributedArrays` package on all spawned processes.\n",
       "\n",
       "```julia\n",
       "@everywhere using DistributedArrays\n",
       "```\n",
       "\n",
       "## Distributed Arrays\n",
       "\n",
       "Large computations are often organized around large arrays of data. In these cases, a particularly natural way to obtain parallelism is to distribute arrays among several processes. This combines the memory resources of multiple machines, allowing use of arrays too large to fit on one machine. Each process operates on the part of the array it owns, providing a ready answer to the question of how a program should be divided among machines.\n",
       "\n",
       "Julia distributed arrays are implemented by the `DArray` type. A `DArray` has an element type and dimensions just like an `Array`. A `DArray` can also use arbitrary array-like types to represent the local chunks that store actual data. The data in a `DArray` is distributed by dividing the index space into some number of blocks in each dimension.\n",
       "\n",
       "Common kinds of arrays can be constructed with functions beginning with `d`:\n",
       "\n",
       "```julia\n",
       "    dzeros(100,100,10)\n",
       "    dones(100,100,10)\n",
       "    drand(100,100,10)\n",
       "    drandn(100,100,10)\n",
       "    dfill(x,100,100,10)\n",
       "```\n",
       "\n",
       "In the last case, each element will be initialized to the specified value `x`. These functions automatically pick a distribution for you. For more control, you can specify which processes to use, and how the data should be distributed:\n",
       "\n",
       "```julia\n",
       "    dzeros((100,100), workers()[1:4], [1,4])\n",
       "```\n",
       "\n",
       "The second argument specifies that the array should be created on the first four workers. When dividing data among a large number of processes, one often sees diminishing returns in performance. Placing `DArray`\\ s on a subset of processes allows multiple `DArray` computations to happen at once, with a higher ratio of work to communication on each process.\n",
       "\n",
       "The third argument specifies a distribution; the nth element of this array specifies how many pieces dimension n should be divided into. In this example the first dimension will not be divided, and the second dimension will be divided into 4 pieces. Therefore each local chunk will be of size `(100,25)`. Note that the product of the distribution array must equal the number of processes.\n",
       "\n",
       "  * `distribute(a::Array)` converts a local array to a distributed array.\n",
       "\n",
       "  * `localpart(a::DArray)` obtains the locally-stored portion of a  `DArray`.\n",
       "\n",
       "  * `localindexes(a::DArray)` gives a tuple of the index ranges owned by the local process.\n",
       "\n",
       "  * `convert(Array, a::DArray)` brings all the data to the local process.\n",
       "\n",
       "Indexing a `DArray` (square brackets) with ranges of indexes always creates a `SubArray`, not copying any data.\n",
       "\n",
       "## Constructing Distributed Arrays\n",
       "\n",
       "The primitive `DArray` constructor has the following somewhat elaborate signature:\n",
       "\n",
       "```julia\n",
       "    DArray(init, dims[, procs, dist])\n",
       "```\n",
       "\n",
       "`init` is a function that accepts a tuple of index ranges. This function should allocate a local chunk of the distributed array and initialize it for the specified indices. `dims` is the overall size of the distributed array. `procs` optionally specifies a vector of process IDs to use. `dist` is an integer vector specifying how many chunks the distributed array should be divided into in each dimension.\n",
       "\n",
       "The last two arguments are optional, and defaults will be used if they are omitted.\n",
       "\n",
       "As an example, here is how to turn the local array constructor `fill` into a distributed array constructor:\n",
       "\n",
       "```julia\n",
       "    dfill(v, args...) = DArray(I->fill(v, map(length,I)), args...)\n",
       "```\n",
       "\n",
       "In this case the `init` function only needs to call `fill` with the dimensions of the local piece it is creating.\n",
       "\n",
       "`DArray`s can also be constructed from multidimensional `Array` comprehensions with the `@DArray` macro syntax.  This syntax is just sugar for the primitive `DArray` constructor:\n",
       "\n",
       "```julia\n",
       "julia> [i+j for i = 1:5, j = 1:5]\n",
       "5x5 Array{Int64,2}:\n",
       " 2  3  4  5   6\n",
       " 3  4  5  6   7\n",
       " 4  5  6  7   8\n",
       " 5  6  7  8   9\n",
       " 6  7  8  9  10\n",
       "\n",
       "julia> @DArray [i+j for i = 1:5, j = 1:5]\n",
       "5x5 DistributedArrays.DArray{Int64,2,Array{Int64,2}}:\n",
       " 2  3  4  5   6\n",
       " 3  4  5  6   7\n",
       " 4  5  6  7   8\n",
       " 5  6  7  8   9\n",
       " 6  7  8  9  10\n",
       "```\n",
       "\n",
       "## Distributed Array Operations\n",
       "\n",
       "At this time, distributed arrays do not have much functionality. Their major utility is allowing communication to be done via array indexing, which is convenient for many problems. As an example, consider implementing the \"life\" cellular automaton, where each cell in a grid is updated according to its neighboring cells. To compute a chunk of the result of one iteration, each process needs the immediate neighbor cells of its local chunk. The following code accomplishes this::\n",
       "\n",
       "```julia\n",
       "    function life_step(d::DArray)\n",
       "        DArray(size(d),procs(d)) do I\n",
       "            top   = mod(first(I[1])-2,size(d,1))+1\n",
       "            bot   = mod( last(I[1])  ,size(d,1))+1\n",
       "            left  = mod(first(I[2])-2,size(d,2))+1\n",
       "            right = mod( last(I[2])  ,size(d,2))+1\n",
       "\n",
       "            old = Array(Bool, length(I[1])+2, length(I[2])+2)\n",
       "            old[1      , 1      ] = d[top , left]   # left side\n",
       "            old[2:end-1, 1      ] = d[I[1], left]\n",
       "            old[end    , 1      ] = d[bot , left]\n",
       "            old[1      , 2:end-1] = d[top , I[2]]\n",
       "            old[2:end-1, 2:end-1] = d[I[1], I[2]]   # middle\n",
       "            old[end    , 2:end-1] = d[bot , I[2]]\n",
       "            old[1      , end    ] = d[top , right]  # right side\n",
       "            old[2:end-1, end    ] = d[I[1], right]\n",
       "            old[end    , end    ] = d[bot , right]\n",
       "\n",
       "            life_rule(old)\n",
       "        end\n",
       "    end\n",
       "```\n",
       "\n",
       "As you can see, we use a series of indexing expressions to fetch data into a local array `old`. Note that the `do` block syntax is convenient for passing `init` functions to the `DArray` constructor. Next, the serial function `life_rule` is called to apply the update rules to the data, yielding the needed `DArray` chunk. Nothing about `life_rule` is `DArray`\\ -specific, but we list it here for completeness::\n",
       "\n",
       "```julia\n",
       "    function life_rule(old)\n",
       "        m, n = size(old)\n",
       "        new = similar(old, m-2, n-2)\n",
       "        for j = 2:n-1\n",
       "            for i = 2:m-1\n",
       "                nc = +(old[i-1,j-1], old[i-1,j], old[i-1,j+1],\n",
       "                       old[i  ,j-1],             old[i  ,j+1],\n",
       "                       old[i+1,j-1], old[i+1,j], old[i+1,j+1])\n",
       "                new[i-1,j-1] = (nc == 3 || nc == 2 && old[i,j])\n",
       "            end\n",
       "        end\n",
       "        new\n",
       "    end\n",
       "```\n",
       "\n",
       "## Numerical Results of Distributed Computations\n",
       "\n",
       "Floating point arithmetic is not associative and this comes up when performing distributed computations over `DArray`s.  All `DArray` operations are performed over the `localpart` chunks and then aggregated. The change in ordering of the operations will change the numeric result as seen in this simple example:\n",
       "\n",
       "```julia\n",
       "julia> addprocs(8);\n",
       "\n",
       "julia> @everywhere using DistributedArrays\n",
       "\n",
       "julia> A = fill(1.1, (100,100));\n",
       "\n",
       "julia> sum(A)\n",
       "11000.000000000013\n",
       "\n",
       "julia> DA = distribute(A);\n",
       "\n",
       "julia> sum(DA)\n",
       "11000.000000000127\n",
       "\n",
       "julia> sum(A) == sum(DA)\n",
       "false\n",
       "```\n",
       "\n",
       "The ultimate ordering of operations will be dependent on how the Array is distributed.\n"
      ],
      "text/plain": [
       "# DistributedArrays.jl\n",
       "\n",
       "[![Build Status](https://travis-ci.org/JuliaParallel/DistributedArrays.jl.svg?branch=master)](https://travis-ci.org/JuliaParallel/DistributedArrays.jl)\n",
       "\n",
       "Distributed Arrays for Julia\n",
       "\n",
       "***NOTE*** Distributed Arrays will only work on the latest development version of Julia (v0.4.0-dev).\n",
       "\n",
       "`DArray`s have been removed from Julia Base library in v0.4 so it is now necessary to import the `DistributedArrays` package on all spawned processes.\n",
       "\n",
       "```julia\n",
       "@everywhere using DistributedArrays\n",
       "```\n",
       "\n",
       "## Distributed Arrays\n",
       "\n",
       "Large computations are often organized around large arrays of data. In these cases, a particularly natural way to obtain parallelism is to distribute arrays among several processes. This combines the memory resources of multiple machines, allowing use of arrays too large to fit on one machine. Each process operates on the part of the array it owns, providing a ready answer to the question of how a program should be divided among machines.\n",
       "\n",
       "Julia distributed arrays are implemented by the `DArray` type. A `DArray` has an element type and dimensions just like an `Array`. A `DArray` can also use arbitrary array-like types to represent the local chunks that store actual data. The data in a `DArray` is distributed by dividing the index space into some number of blocks in each dimension.\n",
       "\n",
       "Common kinds of arrays can be constructed with functions beginning with `d`:\n",
       "\n",
       "```julia\n",
       "    dzeros(100,100,10)\n",
       "    dones(100,100,10)\n",
       "    drand(100,100,10)\n",
       "    drandn(100,100,10)\n",
       "    dfill(x,100,100,10)\n",
       "```\n",
       "\n",
       "In the last case, each element will be initialized to the specified value `x`. These functions automatically pick a distribution for you. For more control, you can specify which processes to use, and how the data should be distributed:\n",
       "\n",
       "```julia\n",
       "    dzeros((100,100), workers()[1:4], [1,4])\n",
       "```\n",
       "\n",
       "The second argument specifies that the array should be created on the first four workers. When dividing data among a large number of processes, one often sees diminishing returns in performance. Placing `DArray`\\ s on a subset of processes allows multiple `DArray` computations to happen at once, with a higher ratio of work to communication on each process.\n",
       "\n",
       "The third argument specifies a distribution; the nth element of this array specifies how many pieces dimension n should be divided into. In this example the first dimension will not be divided, and the second dimension will be divided into 4 pieces. Therefore each local chunk will be of size `(100,25)`. Note that the product of the distribution array must equal the number of processes.\n",
       "\n",
       "  * `distribute(a::Array)` converts a local array to a distributed array.\n",
       "\n",
       "  * `localpart(a::DArray)` obtains the locally-stored portion of a  `DArray`.\n",
       "\n",
       "  * `localindexes(a::DArray)` gives a tuple of the index ranges owned by the local process.\n",
       "\n",
       "  * `convert(Array, a::DArray)` brings all the data to the local process.\n",
       "\n",
       "Indexing a `DArray` (square brackets) with ranges of indexes always creates a `SubArray`, not copying any data.\n",
       "\n",
       "## Constructing Distributed Arrays\n",
       "\n",
       "The primitive `DArray` constructor has the following somewhat elaborate signature:\n",
       "\n",
       "```julia\n",
       "    DArray(init, dims[, procs, dist])\n",
       "```\n",
       "\n",
       "`init` is a function that accepts a tuple of index ranges. This function should allocate a local chunk of the distributed array and initialize it for the specified indices. `dims` is the overall size of the distributed array. `procs` optionally specifies a vector of process IDs to use. `dist` is an integer vector specifying how many chunks the distributed array should be divided into in each dimension.\n",
       "\n",
       "The last two arguments are optional, and defaults will be used if they are omitted.\n",
       "\n",
       "As an example, here is how to turn the local array constructor `fill` into a distributed array constructor:\n",
       "\n",
       "```julia\n",
       "    dfill(v, args...) = DArray(I->fill(v, map(length,I)), args...)\n",
       "```\n",
       "\n",
       "In this case the `init` function only needs to call `fill` with the dimensions of the local piece it is creating.\n",
       "\n",
       "`DArray`s can also be constructed from multidimensional `Array` comprehensions with the `@DArray` macro syntax.  This syntax is just sugar for the primitive `DArray` constructor:\n",
       "\n",
       "```julia\n",
       "julia> [i+j for i = 1:5, j = 1:5]\n",
       "5x5 Array{Int64,2}:\n",
       " 2  3  4  5   6\n",
       " 3  4  5  6   7\n",
       " 4  5  6  7   8\n",
       " 5  6  7  8   9\n",
       " 6  7  8  9  10\n",
       "\n",
       "julia> @DArray [i+j for i = 1:5, j = 1:5]\n",
       "5x5 DistributedArrays.DArray{Int64,2,Array{Int64,2}}:\n",
       " 2  3  4  5   6\n",
       " 3  4  5  6   7\n",
       " 4  5  6  7   8\n",
       " 5  6  7  8   9\n",
       " 6  7  8  9  10\n",
       "```\n",
       "\n",
       "## Distributed Array Operations\n",
       "\n",
       "At this time, distributed arrays do not have much functionality. Their major utility is allowing communication to be done via array indexing, which is convenient for many problems. As an example, consider implementing the \"life\" cellular automaton, where each cell in a grid is updated according to its neighboring cells. To compute a chunk of the result of one iteration, each process needs the immediate neighbor cells of its local chunk. The following code accomplishes this::\n",
       "\n",
       "```julia\n",
       "    function life_step(d::DArray)\n",
       "        DArray(size(d),procs(d)) do I\n",
       "            top   = mod(first(I[1])-2,size(d,1))+1\n",
       "            bot   = mod( last(I[1])  ,size(d,1))+1\n",
       "            left  = mod(first(I[2])-2,size(d,2))+1\n",
       "            right = mod( last(I[2])  ,size(d,2))+1\n",
       "\n",
       "            old = Array(Bool, length(I[1])+2, length(I[2])+2)\n",
       "            old[1      , 1      ] = d[top , left]   # left side\n",
       "            old[2:end-1, 1      ] = d[I[1], left]\n",
       "            old[end    , 1      ] = d[bot , left]\n",
       "            old[1      , 2:end-1] = d[top , I[2]]\n",
       "            old[2:end-1, 2:end-1] = d[I[1], I[2]]   # middle\n",
       "            old[end    , 2:end-1] = d[bot , I[2]]\n",
       "            old[1      , end    ] = d[top , right]  # right side\n",
       "            old[2:end-1, end    ] = d[I[1], right]\n",
       "            old[end    , end    ] = d[bot , right]\n",
       "\n",
       "            life_rule(old)\n",
       "        end\n",
       "    end\n",
       "```\n",
       "\n",
       "As you can see, we use a series of indexing expressions to fetch data into a local array `old`. Note that the `do` block syntax is convenient for passing `init` functions to the `DArray` constructor. Next, the serial function `life_rule` is called to apply the update rules to the data, yielding the needed `DArray` chunk. Nothing about `life_rule` is `DArray`\\ -specific, but we list it here for completeness::\n",
       "\n",
       "```julia\n",
       "    function life_rule(old)\n",
       "        m, n = size(old)\n",
       "        new = similar(old, m-2, n-2)\n",
       "        for j = 2:n-1\n",
       "            for i = 2:m-1\n",
       "                nc = +(old[i-1,j-1], old[i-1,j], old[i-1,j+1],\n",
       "                       old[i  ,j-1],             old[i  ,j+1],\n",
       "                       old[i+1,j-1], old[i+1,j], old[i+1,j+1])\n",
       "                new[i-1,j-1] = (nc == 3 || nc == 2 && old[i,j])\n",
       "            end\n",
       "        end\n",
       "        new\n",
       "    end\n",
       "```\n",
       "\n",
       "## Numerical Results of Distributed Computations\n",
       "\n",
       "Floating point arithmetic is not associative and this comes up when performing distributed computations over `DArray`s.  All `DArray` operations are performed over the `localpart` chunks and then aggregated. The change in ordering of the operations will change the numeric result as seen in this simple example:\n",
       "\n",
       "```julia\n",
       "julia> addprocs(8);\n",
       "\n",
       "julia> @everywhere using DistributedArrays\n",
       "\n",
       "julia> A = fill(1.1, (100,100));\n",
       "\n",
       "julia> sum(A)\n",
       "11000.000000000013\n",
       "\n",
       "julia> DA = distribute(A);\n",
       "\n",
       "julia> sum(DA)\n",
       "11000.000000000127\n",
       "\n",
       "julia> sum(A) == sum(DA)\n",
       "false\n",
       "```\n",
       "\n",
       "The ultimate ordering of operations will be dependent on how the Array is distributed.\n"
      ]
     },
     "execution_count": 406,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DistributedArrays\n",
      "\n"
     ]
    }
   ],
   "source": [
    "?DistributedArrays"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Advanced topics in DArray\n",
    "\n",
    "### Customizing DistributedArrays creation\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10x4 DistributedArrays.DArray{Float64,2,Array{Float64,2}}:\n",
       " 0.68117   0.421907  0.595136  0.0870558\n",
       " 0.877746  0.889965  0.25493   0.116047 \n",
       " 0.188807  0.571747  0.797157  0.708954 \n",
       " 0.446026  0.156147  0.416299  0.539737 \n",
       " 0.325877  0.339183  0.685777  0.569284 \n",
       " 0.19773   0.53126   0.985314  0.0775015\n",
       " 0.995182  0.856847  0.256815  0.670528 \n",
       " 0.480575  0.865361  0.245227  0.455537 \n",
       " 0.316386  0.448017  0.292674  0.863689 \n",
       " 0.763345  0.674715  0.563915  0.148122 "
      ]
     },
     "execution_count": 102,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "darray = drand((10,4), workers()[1:4], [1,4])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 2:\t(1:10,1:1)\n",
      "\tFrom worker 3:\t(1:10,2:2)\n",
      "\tFrom worker 5:\t(1:10,4:4)\n",
      "\tFrom worker 4:\t(1:10,3:3)\n"
     ]
    }
   ],
   "source": [
    "for w in workers()\n",
    "    @spawnat w println(localindexes(darray))\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Now let's know which `RemoteRef`s holds the chunks of `darray`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1x4 Array{RemoteRef{RemoteStore},2}:\n",
       " RemoteRef{Channel{Any}}(2,1,7147)  …  RemoteRef{Channel{Any}}(5,1,7150)"
      ]
     },
     "execution_count": 91,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "darray.chunks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* We can also know which processes hold `darray`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 143,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1x4 Array{Int64,2}:\n",
       " 2  3  4  5"
      ]
     },
     "execution_count": 143,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "darray.pids"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Or we can know the splitting indexes of `darray`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 142,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1x4 Array{Tuple{UnitRange{Int64},UnitRange{Int64}},2}:\n",
       " (1:10,1:1)  (1:10,2:2)  (1:10,3:3)  (1:10,4:4)"
      ]
     },
     "execution_count": 142,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "darray.indexes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* From a process, we can know the indexes it holds:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 139,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 2:\tlocalindexes(darray) = (1:10,1:1)\n",
      "\tFrom worker 3:\tlocalindexes(darray) = (1:10,2:2)\n",
      "\tFrom worker 4:\tlocalindexes(darray) = (1:10,3:3)\n",
      "\tFrom worker 5:\tlocalindexes(darray) = (1:10,4:4)\n"
     ]
    }
   ],
   "source": [
    "for w in workers()\n",
    "    @spawnat w @show localindexes(darray)\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* We can get any data of `darray` from any process, for instance, from the master:\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 141,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.32587670612620934"
      ]
     },
     "execution_count": 141,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "darray[5]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* We can also access `darray` from any process such as worker 4:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 140,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.32587670612620934"
      ]
     },
     "execution_count": 140,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fetch(@spawnat 4 darray[5])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Distributed reduce using DistributedArrays\n",
    "* From Jeff Bezanson's Tutorial on [Parallel and Distributed Computing with Julia](https://www.youtube.com/watch?v=JoRn4ryMclc)\n",
    "\n",
    "* If we wanted to sum `darray` locally, we can simply call `sum(darray)`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "20.357692287307746"
      ]
     },
     "execution_count": 104,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sum(darray)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* The previous command was performed **locally**, at the master process\n",
    "\n",
    "* **Let's take advantage of parallelism** by putting all workers to execute the sum on the portion of `darray` each worker holds in order to not imply data movement\n",
    "* It's similiar to a distributed `reduce` operation:\n",
    "    * The way you think and prgram in Julia is natural, i.e., you can read the following line as you would say to some people to do some task:\n",
    "    * _\"spwan at process `p` the sum of its `darray` local part for all processes\"_\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4-element Array{Any,1}:\n",
       " RemoteRef{Channel{Any}}(2,1,7522)\n",
       " RemoteRef{Channel{Any}}(3,1,7523)\n",
       " RemoteRef{Channel{Any}}(4,1,7524)\n",
       " RemoteRef{Channel{Any}}(5,1,7525)"
      ]
     },
     "execution_count": 108,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "local_sums = [ @spawnat w sum(localpart(darray)) for w in workers()]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Remark: `@everywhere` could not be use in the previous example since it does not return any value/RemoteRef.\n",
    "\n",
    "* Now let's see the contents of each `RemoteRef`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Local sum of RemoteRef{Channel{Any}}(2,1,7522) is 5.272844798025005\n",
      "Local sum of RemoteRef{Channel{Any}}(3,1,7523) is 5.755148538138882\n",
      "Local sum of RemoteRef{Channel{Any}}(4,1,7524) is 5.093243829209424\n",
      "Local sum of RemoteRef{Channel{Any}}(5,1,7525) is 4.2364551219344335\n"
     ]
    }
   ],
   "source": [
    "for s in local_sums println(\"Local sum of $s is $(fetch(s))\") end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* So basically `local_sums` is an array with the results of sums which were executed **at each worker**\n",
    "* In other words, it reduced the values of `darray` each worker holds by executing the sum remotely.\n",
    "* **Now lets merge all the reduces to generate the final output**:\n",
    "    * First we `fetch` the results with `map`\n",
    "    * Then we merge them (i.e. sum) localy, at the master process\n",
    "        * This is similar to merging the _Output files_ of the MapReduce framework ([c.f. Figure 1](http://static.googleusercontent.com/media/research.google.com/pt-BR//archive/mapreduce-osdi04.pdf))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "20.357692287307746"
      ]
     },
     "execution_count": 112,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sum(map(fetch, [ @spawnat p sum(localpart(darray)) for p in workers()] ))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Indeed, **we can directly use the `reduce` function** instead of `sum`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "20.357692287307746"
      ]
     },
     "execution_count": 113,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "reduce(+, map(fetch, [ @spawnat p reduce(+, localpart(darray)) for p in workers()] ))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Finallly, we can define a **parallel reduce** function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "parallel_reduce (generic function with 1 method)"
      ]
     },
     "execution_count": 116,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "parallel_reduce(f,darray) = reduce(f, map(fetch, [ @spawnat p reduce(f, localpart(darray)) for p in workers()] ))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Now we use the **`parallel_reduce` function to operate over other `DistributedArray`s**:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4.142936730385074"
      ]
     },
     "execution_count": 117,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "d = drand(10)\n",
    "parallel_reduce(+, d)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* You can test with further function since it reduces, i.e., it's a binary operator:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.8151657589779049"
      ]
     },
     "execution_count": 118,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "parallel_reduce(*, d)\n",
    "parallel_reduce(max, d)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### More on distributed reduce\n",
    "\n",
    "* Intersting [discussion at Julia mailing list on ```pmap_reduce```](https://groups.google.com/forum/#!msg/julia-users/WJBIAYzrZgg/3xzjoMfpKVMJ)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### What if <span style=\"color:green\">I don't know the size of the data I need to load</span>  and <span style=\"color:gree\">I don't want to deal with low-level details</span> on DistributedArrays?\n",
    "\n",
    "#### [<span style=\"color:red\">CloudArray</span> solves this transparently!](cloudarray.ipynb)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "# <span style=\"color:red\">Distributed Maps and Reduces</span>\n",
    "\n",
    "\n",
    "* Map, Filter and Reduce are common [**functional programming**](https://en.wikipedia.org/wiki/Functional_programming) features\n",
    "* [**Google MapReduce framework**](http://research.google.com/archive/mapreduce-osdi04.pdf) leveraged Map and Reduce concepts and proposed a _distributed framewrok for data-intensive applications_\n",
    "    * [**Hadoop**](http://hadoop.apache.org) is an open-source implementation of Google MapReduce framework\n",
    "    * [**Rhipe**](https://www.datadr.org) allows to transparently execute R codes on top of Hadoop (no need to deal with Hadoop details)\n",
    "* Julia functional features include:\n",
    "    * **Map, Filter, Reduce, and MapReduce**\n",
    "    * However, they are not distributed\n",
    "    * E.g., `mapreduce` uses the same worker to execute both map and reduce\n",
    "* Native **Distribute MapReduce** in Julia is done in different ways\n",
    "    * Spawning `map, reduce, mapreduce` calls on workers\n",
    "        * @parallel, @spawn, remotecall, etc.\n",
    "    * **Distributed map** called [**pmap**](http://docs.julialang.org/en/latest/stdlib/parallel/#Base.pmap)\n",
    "\n",
    "<!--\n",
    "    * pmap\n",
    "        * https://groups.google.com/forum/#!msg/julia-users/WJBIAYzrZgg/_UPc8vhkCAAJ\n",
    "    * Discussion on MapReduce, ShredArrays, etc. at [julia-users maling list](https://groups.google.com/forum/#!msg/julia-users/1sNXYtIbS1Q/1sNd0h92AQAJ)\n",
    "    * distributed mapreduce by using @spawn\n",
    "    * @distribute\n",
    "    * @dmap\n",
    "-->\n",
    "\n",
    "## Distributed `reduce`\n",
    "\n",
    "See example with `DistributedArrays`.\n",
    "\n",
    "## `pmap`\n",
    "\n",
    "* Context (motivation)\n",
    "    * No reduction operator is needed\n",
    "    * Any variables used inside the @parallel loop will be copied (broadcasted) to each process.\n",
    "* `pmap` function\n",
    "    * Applies a function to all elements in some collection\n",
    "* `pmap(f, coll)`\n",
    "    * `f: function`\n",
    "    * `coll: collection`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 322,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3-element Array{Any,1}:\n",
       " 1.02\n",
       " 3.14\n",
       " 5.63"
      ]
     },
     "execution_count": 322,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "M = [-1.02, 3.14, -5.63]\n",
    "\n",
    "pmap(abs,M)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Where `pmap` distributed my processing?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 345,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 2:\tWhere am I?\n",
      "\tFrom worker 3:\tWhere am I?\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "5-element Array{Any,1}:\n",
       " nothing\n",
       " nothing\n",
       " nothing\n",
       " nothing\n",
       " nothing"
      ]
     },
     "execution_count": 345,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\tFrom worker 4:\tWhere am I?\n",
      "\tFrom worker 5:\tWhere am I?\n",
      "\tFrom worker 2:\tWhere am I?\n"
     ]
    }
   ],
   "source": [
    "M = [\"Where am I?\", \"Where am I?\", \"Where am I?\", \"Where am I?\", \"Where am I?\"]\n",
    "\n",
    "pmap(println,M)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 405,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10000-element Array{Any,1}:\n",
       " 1.10063\n",
       " 1.62362\n",
       " 1.43912\n",
       " 1.24644\n",
       " 1.21631\n",
       " 1.79434\n",
       " 1.02257\n",
       " 1.67295\n",
       " 1.66239\n",
       " 1.6951 \n",
       " 1.2725 \n",
       " 1.48917\n",
       " 1.11586\n",
       " ⋮      \n",
       " 1.68033\n",
       " 1.52385\n",
       " 1.71493\n",
       " 1.3362 \n",
       " 1.43307\n",
       " 1.57744\n",
       " 1.47357\n",
       " 1.51247\n",
       " 1.85764\n",
       " 1.68235\n",
       " 1.02524\n",
       " 1.27252"
      ]
     },
     "execution_count": 405,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@everywhere f(x) = x.+1\n",
    "\n",
    "M = rand(10^4)\n",
    "pmap(f, M)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, suppose that we have to calculate the rank of a number of large matrices"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rank_marray (generic function with 1 method)"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "function rank_marray()\n",
    "   marr = [rand(1000,1000) for i=1:4]\n",
    "   for arr in marr\n",
    "      println(rank(arr))\n",
    "   end\n",
    "end"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1000\n",
      "1000\n",
      "1000\n",
      "1000\n",
      "  5.118474 seconds (1.38 k allocations: 63.449 MB, 0.19% gc time)\n"
     ]
    }
   ],
   "source": [
    "@time rank_marray()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Any[1000,1000,1000,1000]\n",
      "  2.963267 seconds (1.03 k allocations: 70.031 KB)\n"
     ]
    }
   ],
   "source": [
    "marr = [rand(1000,1000) for i=1:4]\n",
    "\n",
    "@time println(pmap(rank, marr))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "search: "
     ]
    },
    {
     "data": {
      "text/latex": [
       "\\begin{verbatim}\n",
       "..  pmap(f, lsts...; err_retry=true, err_stop=false, pids=workers())\n",
       "\n",
       "Transform collections ``lsts`` by applying ``f`` to each element in parallel.\n",
       "(Note that ``f`` must be made available to all worker processes; see :ref:`Code Availability and Loading Packages <man-parallel-computing-code-availability>` for details.)\n",
       "If ``nprocs() > 1``, the calling process will be dedicated to assigning tasks.\n",
       "All other available processes will be used as parallel workers, or on the processes specified by ``pids``.\n",
       "\n",
       "If ``err_retry`` is ``true``, it retries a failed application of ``f`` on a different worker.\n",
       "If ``err_stop`` is ``true``, it takes precedence over the value of ``err_retry`` and ``pmap`` stops execution on the first error.\n",
       "\\end{verbatim}\n"
      ],
      "text/markdown": [
       "```rst\n",
       "..  pmap(f, lsts...; err_retry=true, err_stop=false, pids=workers())\n",
       "\n",
       "Transform collections ``lsts`` by applying ``f`` to each element in parallel.\n",
       "(Note that ``f`` must be made available to all worker processes; see :ref:`Code Availability and Loading Packages <man-parallel-computing-code-availability>` for details.)\n",
       "If ``nprocs() > 1``, the calling process will be dedicated to assigning tasks.\n",
       "All other available processes will be used as parallel workers, or on the processes specified by ``pids``.\n",
       "\n",
       "If ``err_retry`` is ``true``, it retries a failed application of ``f`` on a different worker.\n",
       "If ``err_stop`` is ``true``, it takes precedence over the value of ``err_retry`` and ``pmap`` stops execution on the first error.\n",
       "```\n"
      ],
      "text/plain": [
       "```rst\n",
       "..  pmap(f, lsts...; err_retry=true, err_stop=false, pids=workers())\n",
       "\n",
       "Transform collections ``lsts`` by applying ``f`` to each element in parallel.\n",
       "(Note that ``f`` must be made available to all worker processes; see :ref:`Code Availability and Loading Packages <man-parallel-computing-code-availability>` for details.)\n",
       "If ``nprocs() > 1``, the calling process will be dedicated to assigning tasks.\n",
       "All other available processes will be used as parallel workers, or on the processes specified by ``pids``.\n",
       "\n",
       "If ``err_retry`` is ``true``, it retries a failed application of ``f`` on a different worker.\n",
       "If ``err_stop`` is ``true``, it takes precedence over the value of ``err_retry`` and ``pmap`` stops execution on the first error.\n",
       "```\n"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "pmap promote_shape repmat typemax permutations SparseMatrix\n",
      "\n"
     ]
    }
   ],
   "source": [
    "?pmap"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# More on Distributed/Parallel Computing in Julia\n",
    "\n",
    "* JuliaCon 2015\n",
    "    * [JuliaCon2015 Workshop](https://github.com/JuliaParallel/JuliaCon2015_Workshop)    \n",
    "    * [JuliaCon 2015 | Amit Murthy: Cluster Managers and parallel julia](https://www.youtube.com/watch?v=XJAQ24NS458) \n",
    "        * [IJulia Notebbok on Parallel Julia (internals)](https://github.com/JuliaParallel/JuliaCon2015_Workshop/blob/master/JuliaCon%202015.ipynb) \n",
    "    * [JuliaCon India 2015 - Concurrent and Parallel programming](https://github.com/ViralBShah/JuliaConIndia2015/blob/master/JuliaCon_Parallel_Workshop.ipynb)\n",
    "    * [JuliaCon 2015 - Julia Parallel Workshop](https://github.com/JuliaParallel/JuliaCon2015_Workshop/blob/master/Workshop.ipynb)\n",
    "* [Julia Oficcial Documentation on Parallel Computing](http://docs.julialang.org/en/latest/manual/parallel-computing/)\n",
    "* [Official Julia packages on Parallel Computing](https://github.com/JuliaParallel)\n",
    "* [Julia Lightning Round (Alan Edelman, Viral B. Shah)](https://www.youtube.com/watch?v=37L1OMk_3FU&list=PLP8iPy9hna6Si2sjMkrPY-wt2mEouZgaZ)\n",
    "* [Stefan Karpinski - Julia: Fast Performance, Distributed Computing & Multiple Dispatch](https://www.youtube.com/watch?v=rUczbQ6ZPd8)\n",
    "* [Viral's IJulia Notebooks](https://github.com/ViralBShah?tab=repositories)\n",
    "* [Distributed DataFrame](https://github.com/JuliaParallel/Blocks.jl)\n",
    "    * Discontinued but there are plans to carry on its implementation\n",
    "    * See [this topic at julia-dev mailing list](https://groups.google.com/forum/#!msg/julia-dev/ja3ienKR0-g/wZg1NMZvB_QJ)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Julia 0.4.0",
   "language": "julia",
   "name": "julia-0.4"
  },
  "language_info": {
   "file_extension": ".jl",
   "mimetype": "application/julia",
   "name": "julia",
   "version": "0.4.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}