0_4_python_parallel_computing.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "executionInfo": {
     "elapsed": 2234,
     "status": "ok",
     "timestamp": 1613994284296,
     "user": {
      "displayName": "Sem Vijverberg",
      "photoUrl": "",
      "userId": "13865882758081973772"
     },
     "user_tz": -60
    },
    "id": "I762wWOVjdyC",
    "outputId": "c751ad2d-6a2a-45e9-e45d-435813109c01"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4 cpu's detected\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor\n",
    "from threading import Thread\n",
    "max_cpu = mp.cpu_count()\n",
    "print(f'{max_cpu} cpu\\'s detected')\n",
    "import numpy as np\n",
    "\n",
    "from math import sqrt\n",
    "from time import sleep, time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "mheQaMhnjdyJ"
   },
   "source": [
    "# Global Interpreter Lock\n",
    "Please keep in mind the [Python Global Interpreter Lock (GIL)](https://realpython.com/python-gil/#what-problem-did-the-gil-solve-for-python), this means that - on a single thread - no two lines of Pyhton code can execute at the same time (ever). Parallel computing is achieved by sending the script to multiple cores."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Two different worker functions\n",
    "Below, we create a worker function that does some numpy computation and a working function that simply waits a few seconds. The reason we create these two is that they very widely in their potential for a speed-up once parallelized. This is because numpy is already relying C++ code in the back-end. I do not know the specifics, but I belief C++ makes use of vectorization allowing it to already compute multiple things at the same time. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "id": "KatsZtIXjdyJ"
   },
   "outputs": [],
   "source": [
    "def poolworker(i, comp_cost_per_job):\n",
    "    data = np.random.uniform(size=(1024, 1024))\n",
    "    for i in range(comp_cost_per_job*30):\n",
    "        np.matmul(data, data) # already done in parallel in by Numpy, use other function\n",
    "    return i"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "id": "OH-CtTZIjdyM"
   },
   "outputs": [],
   "source": [
    "def poolworker(i, comp_cost_per_job):\n",
    "    sleep(comp_cost_per_job)\n",
    "    return "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "AcMTxMZ9jdyM"
   },
   "source": [
    "# Using (built-in) Concurrent futures"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "id": "gHpemVYDjdyM"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Sequential 5.018466949462891\n",
      "Concurrent Thread Pool 1.005444049835205\n",
      "Concurrent Process Pool 1.0631179809570312\n"
     ]
    }
   ],
   "source": [
    "n_jobs = 5\n",
    "comp_cost_per_job = 1\n",
    "t0 = time()\n",
    "results = [poolworker(i, comp_cost_per_job) for i in range(n_jobs)]\n",
    "print(\"Sequential\", time() - t0)\n",
    "\n",
    "# Thread Pool\n",
    "t0 = time()\n",
    "pool = ThreadPoolExecutor(8) # can crash when using interactively\n",
    "futures = [pool.submit(poolworker, i, comp_cost_per_job) for i in range(n_jobs)]\n",
    "results = [future.result() for future in futures]\n",
    "print(\"Concurrent Thread Pool\", time() - t0)\n",
    "\n",
    "# Process Pool\n",
    "t0 = time()\n",
    "pool = ProcessPoolExecutor(8) # can crash when using interactively\n",
    "futures = [pool.submit(poolworker, i, comp_cost_per_job) for i in range(n_jobs)]\n",
    "results = [future.result() for future in futures]\n",
    "print(\"Concurrent Process Pool\", time() - t0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "HE6Cx3r8jdyN"
   },
   "source": [
    "Here Thread Pools don't give any speedup, whereas Process Pools do\n",
    "\n",
    "Please Keep in mind that Process Pools use different address spaces, you can't use any variables defined outside the function's scope, whereas with Thread Pools, this is possible! :)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "1Sn0uzz9jdyN"
   },
   "source": [
    "# multiprocessing package"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "id": "ME-qnNHfjdyO"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Sequential 5.016125917434692\n",
      "MP Process Pool 5.0\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing as mp\n",
    "\n",
    "\n",
    "# Sequential\n",
    "t0 = time()\n",
    "results = [poolworker(i, comp_cost_per_job) for i in range(n_jobs)]\n",
    "print(\"Sequential\", time() - t0)\n",
    "\n",
    "\n",
    "t0 = time()\n",
    "# Step 1: Init multiprocessing.Pool()\n",
    "pool = mp.Pool(mp.cpu_count()) # can crash when using interactively\n",
    "# Step 2: `pool.apply` the `howmany_within_range()`\n",
    "results = [pool.apply(poolworker, args=(i, comp_cost_per_job)) for i in range(n_jobs)]\n",
    "# Step 3: Don't forget to close\n",
    "pool.close()    \n",
    "print(\"MP Process Pool\", round(time() - t0, 1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "08MHDzG-jdyP"
   },
   "source": [
    "# Joblib"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "id": "ZVuhCW3jjdyP",
    "outputId": "ed1ed002-6e7b-4649-ecc3-8c2204c5db04"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Sequential 40.039032220840454\n",
      "Joblib multiprocessing Pool 12.2\n",
      "Joblib loky Pool 13.1\n"
     ]
    }
   ],
   "source": [
    "from joblib import Parallel, delayed\n",
    "\n",
    "# n_jobs = 10\n",
    "# comp_cost_per_job = 4\n",
    "\n",
    "# Sequential\n",
    "t0 = time()\n",
    "results = [poolworker(i, comp_cost_per_job) for i in range(n_jobs)]\n",
    "print(\"Sequential\", time() - t0)\n",
    "\n",
    "# Joblib\n",
    "t0 = time()\n",
    "backend = 'multiprocessing' # can crash when using interactively\n",
    "results = Parallel(n_jobs=-1, backend=backend, batch_size=1)(delayed(\n",
    "    poolworker)(i, comp_cost_per_job) for i in range(n_jobs))\n",
    "print(f\"Joblib {backend} Pool\", round(time() - t0, 1))\n",
    "\n",
    "# Joblib\n",
    "t0 = time()\n",
    "backend = 'loky' # so far robust when using interactively\n",
    "results = Parallel(n_jobs=-1, backend=backend)(delayed(\n",
    "    poolworker)(i, comp_cost_per_job) for i in range(n_jobs))\n",
    "print(f\"Joblib {backend} Pool\", round(time() - t0, 1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "AEnM4hOUjdyP"
   },
   "source": [
    "# Dask"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "executionInfo": {
     "elapsed": 2951,
     "status": "ok",
     "timestamp": 1613994328644,
     "user": {
      "displayName": "Sem Vijverberg",
      "photoUrl": "",
      "userId": "13865882758081973772"
     },
     "user_tz": -60
    },
    "id": "GMAzMr3tjdyQ",
    "outputId": "d22babb6-e052-441f-cbdf-d0117163f68b"
   },
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "import bokeh\n",
    "\n",
    "if 'client' not in globals():\n",
    "    client = Client(processes=True, n_workers=3)  # start local workers as processes\n",
    "# or\n",
    "# clientthr = Client(processes=False)  # start local workers as threads"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 1000
    },
    "executionInfo": {
     "elapsed": 56718,
     "status": "error",
     "timestamp": 1613994414105,
     "user": {
      "displayName": "Sem Vijverberg",
      "photoUrl": "",
      "userId": "13865882758081973772"
     },
     "user_tz": -60
    },
    "id": "4Lmb4xjSjdyQ",
    "outputId": "049c5f63-09ae-4de2-8b09-b4449945cfab"
   },
   "outputs": [],
   "source": [
    "n_jobs = 10\n",
    "comp_cost_per_job = 4\n",
    "# sequestial\n",
    "t0 = time()\n",
    "results = [poolworker(i, comp_cost_per_job) for i in range(n_jobs)]\n",
    "print(\"  Sequential\",round(time() - t0, 1))\n",
    "# Dask futures\n",
    "# so far robust when using interactively\n",
    "t0 = time() \n",
    "# results = Parallel(n_jobs=-1, backend=backend)(delayed(\n",
    "#     poolworker)(i, comp_cost_per_job) for i in range(n_jobs))\n",
    "futures = [client.submit(poolworker, i, comp_cost_per_job) for i in range(n_jobs)]\n",
    "results = client.gather(futures)\n",
    "print(\"Dask Process Pool\", round(time() - t0, 1))"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "name": "parallel_computing.ipynb",
   "provenance": []
  },
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}