JuliaCon_Parallel_Workshop.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Concurrent and Parallel programming #\n",
    "```\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "```\n",
    "###### Amit Murthy, JuliaCon India 2015 ######\n",
    "###### https://github.com/amitmurthy ######"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Tasks in Julia ##\n",
    "\n",
    "- Coroutines / Cooperative multitasking\n",
    "  - Multi-threading is a WIP under development in the 0.5 branch\n",
    "- Useful for \n",
    "  - IO bound tasks - Network, File, etc\n",
    "  - Implementation of Timers, background tasks\n",
    "- High level API - `@async`, `@sync`, `@schedule`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "t0=time()\n",
    "# Start an async task\n",
    "t = @async begin\n",
    "    sleep(2.0)\n",
    "    return pi\n",
    "end\n",
    "\n",
    "# Control comes here immediately\n",
    "println(\"Started \", t)\n",
    "\n",
    "# wait for its completion\n",
    "resp = wait(t)\n",
    "println(\"Response : \", resp, \" after \", time()-t0, \" seconds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "# @sync waits for all enclosed async operations to complete\n",
    "t0=time()\n",
    "@sync begin\n",
    "    for i in 1:100\n",
    "        @async (sleep(1.0); print(\"$i..\"))\n",
    "    end\n",
    "end\n",
    "println(\"\\n\\n All tasks finished in \", time() - t0, \" seconds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "# compute-bound or blocking ccalls do not yield\n",
    "t0=time()\n",
    "@sync begin\n",
    "    for i in 1:5\n",
    "        @async begin\n",
    "            ccall(:sleep, Cint, (Cint,), 1.0)\n",
    "            print(\"$i..\")\n",
    "        end\n",
    "    end\n",
    "end\n",
    "println(\"\\n\\n All tasks finished in \", time() - t0, \" seconds\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "- `@schedule` schedules an expression to be run\n",
    "  - like `@async` except: \n",
    "    - launched task is not waited on by enclosing `@sync` blocks\n",
    "    - does not \"localize\" variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "a=1\n",
    "\n",
    "# @async localizes variables\n",
    "@async (sleep(2.0); println(\"@async : \", a))\n",
    "\n",
    "# @schedule does not\n",
    "@schedule (sleep(2.0); println(\"@schedule : \", a))\n",
    "\n",
    "a=\"Hello\";\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "##  Tasks - produce/consume ##\n",
    "\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"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "function producer(cnt)\n",
    "    @schedule begin\n",
    "        for i in 1:cnt\n",
    "            produce(i)\n",
    "        end\n",
    "        println(\"Producer DONE!\")\n",
    "    end\n",
    "end\n",
    "\n",
    "function consumer(t, id)\n",
    "    @schedule begin\n",
    "        sleep(1.0)\n",
    "        while !istaskdone(t)\n",
    "            v = consume(t)\n",
    "            (v != nothing) && println(\"consumer $id ===> $v\")\n",
    "        end\n",
    "        println(\"consumer $id DONE!\")\n",
    "    end\n",
    "end\n",
    "\n",
    "# start a producer task\n",
    "t=producer(6);\n",
    "\n",
    "# start consumers\n",
    "consumer(t, 1);\n",
    "consumer(t, 2);\n",
    "consumer(t, 3);"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "# start a producer task\n",
    "t=producer(6);\n",
    "\n",
    "for v in t\n",
    "    println(\"consumed $v\")\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Channels ##\n",
    "\n",
    "- Another means of inter-task communication\n",
    "- Is type-aware and has a size\n",
    "- API \n",
    "  - `put!`\n",
    "  - `take!`\n",
    "  - `fetch`\n",
    "  - `isready`\n",
    "  - `wait`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "c=Channel{Int64}(100)\n",
    "for i in 1:6\n",
    "    put!(c, i)\n",
    "end\n",
    "\n",
    "@schedule begin \n",
    "    for v in c\n",
    "        println(v)\n",
    "    end\n",
    "    println(\"Consumer task DONE!\")\n",
    "end;\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "close(c);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Multi-processing ##\n",
    "\n",
    "- Leverge multiple cores\n",
    "- Distribute across machines\n",
    "- Remote function execution as opposed to message passing\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## @parallel for ##\n",
    "- distrubutes the range across workers and executes in parallel "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "nprocs() > 1 && rmprocs(workers(); waitfor=0.5)\n",
    "addprocs(4)\n",
    "@parallel (+) for i in 1:100\n",
    "    rand(Bool)\n",
    "end\n",
    "@time @parallel (+) for i in 1:10^8\n",
    "    rand(Bool)\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## pmap ##\n",
    "- distributed `map` function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "nprocs() > 1 && rmprocs(workers(); waitfor=0.5)\n",
    "addprocs(4)\n",
    "files = filter(x->endswith(x, \".jl\"), readdir(\"/Users/amitm/Julia/julia/base\"))\n",
    "\n",
    "# compilation run\n",
    "pmap(x->begin\n",
    "        path=joinpath(\"/Users/amitm/Julia/julia/base\",x)\n",
    "        readall(`md5 $path`)\n",
    "    end, files);\n",
    "\n",
    "# timed run\n",
    "@time pmap(x->begin\n",
    "        path=joinpath(\"/Users/amitm/Julia/julia/base\",x)\n",
    "        # simulate compute time by calculating MD5 a few times\n",
    "        for i in 1:5\n",
    "            readall(`md5 $path`)\n",
    "        end\n",
    "        readall(`md5 $path`)\n",
    "    end, files)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## distributing work with remote channels ##"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "nprocs() > 1 && rmprocs(workers(); waitfor=0.5)\n",
    "addprocs(4)\n",
    "\n",
    "jobs_c = RemoteRef(()->Channel(100))\n",
    "results_c = RemoteRef(()->Channel(100))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# define function on all workers\n",
    "@everywhere function do_some_work(jobs_c, results_c)\n",
    "    println(\"Worker task started.\")\n",
    "    while true\n",
    "        try\n",
    "            rqst = take!(jobs_c)\n",
    "            if rqst == :EXIT\n",
    "                break\n",
    "            else\n",
    "                sleep(rqst)\n",
    "                put!(results_c, (myid(), rqst))\n",
    "            end\n",
    "        catch e\n",
    "            println(e)\n",
    "        end\n",
    "    end\n",
    "    println(\"DONE!\")\n",
    "end"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# start the worker tasks....\n",
    "for p in workers()\n",
    "    remotecall(p, do_some_work, jobs_c, results_c)\n",
    "end"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# adding jobs....\n",
    "jobs = rand(1:10, 30)\n",
    "for i in jobs\n",
    "    put!(jobs_c, i)\n",
    "end\n",
    "\n",
    "# Have the worker tasks exit at the end\n",
    "for i in 1:nworkers()\n",
    "    put!(jobs_c, :EXIT)\n",
    "end"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "for i in 1:length(jobs)\n",
    "    (wrkr, data) = take!(results_c) \n",
    "    println(\"worker $wrkr : \", data)\n",
    "end"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## More examples ##"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "nprocs() > 1 && rmprocs(workers(); waitfor=0.5)\n",
    "addprocs(4)\n",
    "\n",
    "# execute expression on all processors\n",
    "@everywhere println(myid())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "rr = remotecall(workers()[1], ()->(sleep(1.0); pi))\n",
    "isready(rr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "fetch(rr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "take!(rr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "isready(rr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Other calls ##\n",
    "- `remotecall_fetch` blocks and returns with the result of function execution\n",
    "- `put!(rr, v)` assigns a value to the reference. Blocks if reference is full\n",
    "- `@spawnat pid expr` executes expression on worker `pid`\n",
    "- `@spawn expr` executes expression on the next worker. Cycles through all workers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "nprocs() > 1 && rmprocs(workers(); waitfor=0.5)\n",
    "addprocs(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# @spawn cycles through all workers\n",
    "fetch(@spawn myid())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rr = RemoteRef(()->Channel{AbstractString}(2), workers()[1])\n",
    "put!(rr, \"Hello\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "put!(rr, 2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "put!(rr, \"World\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# should be ready\n",
    "isready(rr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# take the first element\n",
    "take!(rr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# and then the next\n",
    "take!(rr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# should not be ready\n",
    "isready(rr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- RemoteRefs can be serialized across workers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "p1 = workers()[1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "p2 = workers()[2]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rr = RemoteRef(p1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "put!(rr, \"put! from $(myid())\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "remotecall_fetch(p2, r->(println(take!(r)); put!(r, \"put! from $(myid())\"); nothing), rr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "fetch(rr)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Julia 0.4.0-rc3",
   "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
}