PYTHON-INTERMEDIATE2-3-CONCURRENCY.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## GIL (Global Interpreter Lock)\n",
    "*파이썬의 아킬레스건*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "- CPython이 단일 바이트코드를 수행할 때\n",
    "- Thread가 interrupt하지 못하도록 전역 인터프리터 잠금을 수행합니다.\n",
    "- Thread가 여러개 돌고 있는 상황에서 인터프리터가 올바르게 동작하도록"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "<span style='color:blue'>**그 결과.**</span>\n",
    "- <u>Multithreading 상황에서도 GIL은 한 번에 한 스레드에서만 동작하게 합니다.</u>\n",
    "- Multithreading의 본래의 장점이 잘 발휘되지 않습니다.\n",
    "- Multiprocess는 물론 영향을 받지 않습니다. 다만 *multithread보다는 다루기 어렵습니다.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Took 1.828 seconds\n"
     ]
    }
   ],
   "source": [
    "def factorize(number):\n",
    "    return [i for i in range(1, number+1) if number % i == 0]\n",
    "\n",
    "import time\n",
    "\n",
    "numbers = [2139079, 1214759, 1516637, 1852285] * 4 # 값을 바꿔서 heavy하게 돌려보세요.\n",
    "start = time.time()\n",
    "for number in numbers:\n",
    "    factorize(number)\n",
    "end = time.time()\n",
    "print('Took %.3f seconds' % (end-start))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Took 2.217 seconds\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ThreadPoolExecutor\n",
    "\n",
    "pool = ThreadPoolExecutor(max_workers=4)\n",
    "start = time.time()\n",
    "results = list(pool.map(factorize, numbers))\n",
    "end = time.time()\n",
    "\n",
    "print('Took %.3f seconds' % (end-start))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Took 0.576 seconds\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ProcessPoolExecutor\n",
    "\n",
    "pool = ProcessPoolExecutor(max_workers=4)\n",
    "start = time.time()\n",
    "results = list(pool.map(factorize, numbers))\n",
    "end = time.time()\n",
    "\n",
    "print('Took %.3f seconds' % (end-start))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## 그렇다면 만약 IO를 다룰때는 어떨까요?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Took 0.507 seconds\n"
     ]
    }
   ],
   "source": [
    "import select\n",
    "import time\n",
    "def slow_systemcall():\n",
    "    select.select([], [], [], 0.1)\n",
    "\n",
    "start = time.time()\n",
    "for _ in range(5):\n",
    "    slow_systemcall()\n",
    "end = time.time()\n",
    "\n",
    "print('Took %.3f seconds' % (end-start))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Took 0.209 seconds\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ThreadPoolExecutor\n",
    "\n",
    "start = time.time()\n",
    "with ThreadPoolExecutor(max_workers=4) as e:\n",
    "    for _ in range(5):\n",
    "        e.submit(slow_systemcall)\n",
    "end = time.time()\n",
    "\n",
    "print('Took %.3f seconds' % (end-start))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Took 0.223 seconds\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ProcessPoolExecutor\n",
    "\n",
    "start = time.time()\n",
    "with ProcessPoolExecutor(max_workers=4) as e:\n",
    "    for _ in range(5):\n",
    "        e.submit(slow_systemcall)\n",
    "end = time.time()\n",
    "\n",
    "print('Took %.3f seconds' % (end-start))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "IO에 의한 대기는 GIL이 영향을 받지 않으므로, IO부하가 큰 작업은 Thread로도 충분합니다. 상황에 맞게 Concurrency 방법을 선택하세요. 시스템 프로그래머가 아니면 thread를 heavy하게 쓸 일은 거의 없고, 위에 소개된 concurrent.futures는 쉽고 강력한 cuncurrency를 구현할 수 있습니다."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## EXERCISE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "CRITICAL:root:downloads/cn.gif\n",
      "CRITICAL:root:downloads/id.gif\n",
      "CRITICAL:root:downloads/us.gif\n",
      "CRITICAL:root:downloads/br.gif\n",
      "CRITICAL:root:downloads/in.gif\n",
      "CRITICAL:root:downloads/pk.gif\n",
      "CRITICAL:root:downloads/ng.gif\n",
      "CRITICAL:root:downloads/bd.gif\n",
      "CRITICAL:root:downloads/ru.gif\n",
      "CRITICAL:root:downloads/jp.gif\n",
      "CRITICAL:root:downloads/kr.gif\n",
      "CRITICAL:root:downloads/mx.gif\n",
      "CRITICAL:root:downloads/vn.gif\n",
      "CRITICAL:root:downloads/ph.gif\n",
      "CRITICAL:root:downloads/et.gif\n",
      "CRITICAL:root:downloads/eg.gif\n",
      "CRITICAL:root:downloads/de.gif\n",
      "CRITICAL:root:downloads/tr.gif\n",
      "CRITICAL:root:downloads/ir.gif\n",
      "CRITICAL:root:downloads/fr.gif\n",
      "CRITICAL:root:downloads/cd.gif\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "elapsed in 2.40s\n"
     ]
    }
   ],
   "source": [
    "from IPython.display import display, Image\n",
    "import time, os.path, logging\n",
    "from urllib.request import urlopen\n",
    "from concurrent.futures import ProcessPoolExecutor\n",
    "from concurrent.futures import ThreadPoolExecutor\n",
    "\n",
    "POP20_CC = (\n",
    "    ' CN IN US ID BR PK NG BD RU JP KR '\n",
    "    'MX PH VN ET EG DE IR TR CD FR'\n",
    ").split()\n",
    "BASE_URL = 'http://flupy.org/data/flags'\n",
    "DEST_DIR = 'downloads/'\n",
    "\n",
    "def get_flag(cc):\n",
    "    url = '{}/{cc}/{cc}.gif'.format(BASE_URL, cc = cc.lower())\n",
    "    # resp = requests.get(url)\n",
    "    resp = urlopen(url)\n",
    "    path = os.path.join(DEST_DIR, f'{cc.lower()}.gif')\n",
    "    with open(path, 'wb') as fp:\n",
    "        fp.write(resp.read())\n",
    "    path = os.path.join(DEST_DIR, f'{cc.lower()}.gif')\n",
    "#     display(Image(url=path))  # thread-safe하지 않은듯.\n",
    "    logging.critical(path)\n",
    "\n",
    "def proc():\n",
    "    t0 = time.time()\n",
    "    ######################################################\n",
    "    # 1. cc별로 get_flag를 수행해서 download받는 로직을 구현해주세요.\n",
    "    # 2. threading으로 처리해보세요.\n",
    "    # 3. processing으로 처리해보세요.\n",
    "    map_ = map  # 8sec\n",
    "    map_ = ProcessPoolExecutor(max_workers=4).map  # 2.40\n",
    "#     map_ = ThreadPoolExecutor(max_workers=4).map  # 1.81\n",
    "\n",
    "    list(map_(get_flag, POP20_CC))  # 2.36sec\n",
    "    ######################################################\n",
    "    elapsed = time.time() - t0\n",
    "    print('elapsed in {:.2f}s'.format(elapsed))\n",
    "    \n",
    "os.makedirs(DEST_DIR, exist_ok=True)\n",
    "proc()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "def worker():\n",
    "    import time\n",
    "    import random\n",
    "    wait_seconds = random.randrange(3)\n",
    "    time.sleep(wait_seconds)\n",
    "    #if random.randrange(10) > 6:\n",
    "    #    raise Exception(f'{wait_seconds}')\n",
    "    return wait_seconds\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "worker()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Thread work done(0s): Thread-19Thread work done(0s): Thread-20\n",
      "\n",
      "Thread work done(1s): Thread-17\n",
      "Thread work done(2s): Thread-16Thread work done(2s): Thread-18\n",
      "\n",
      "coordinating...\n",
      "coordinating... done..\n",
      "all done.\n"
     ]
    }
   ],
   "source": [
    "import threading\n",
    "import time\n",
    "\n",
    "#coordinator\n",
    "c_event_start = threading.Event()\n",
    "\n",
    "threads = []\n",
    "workdone_events = []\n",
    "\n",
    "def or_set(self):\n",
    "    self._set()\n",
    "    self.changed()\n",
    "\n",
    "def or_clear(self):\n",
    "    self._clear()\n",
    "    self.changed()\n",
    "\n",
    "def orify(e, changed_callback):\n",
    "    e._set = e.set\n",
    "    e._clear = e.clear\n",
    "    e.changed = changed_callback\n",
    "    e.set = lambda: or_set(e)\n",
    "    e.clear = lambda: or_clear(e)\n",
    "\n",
    "def AllEvents(*events):\n",
    "    or_event = threading.Event()\n",
    "    def changed():\n",
    "        bools = [e.is_set() for e in events]\n",
    "        if all(bools):\n",
    "            or_event.set()\n",
    "        else:\n",
    "            or_event.clear()\n",
    "    for e in events:\n",
    "        orify(e, changed)\n",
    "    changed()\n",
    "    return or_event\n",
    "\n",
    "def runner(fin_event, wait_event, value):\n",
    "    tname = threading.current_thread().name\n",
    "\n",
    "    wait_event.wait()\n",
    "    k = worker()\n",
    "    print('Thread work done(%ds): %s' % (k, tname))\n",
    "    fin_event.set()\n",
    "\n",
    "def coordinator(all_events, workdone_events, notify_event):\n",
    "    # order\n",
    "    notify_event.set()\n",
    "    notify_event.clear()\n",
    "\n",
    "    # waiting..\n",
    "    all_events.wait()\n",
    "    \n",
    "    # job completing.. coordinating..\n",
    "    print('coordinating...')\n",
    "    time.sleep(0.3)\n",
    "    print('coordinating... done..')\n",
    "    # all_events.set()\n",
    "    for e in workdone_events:\n",
    "        e.clear()\n",
    "    all_events.clear()\n",
    "\n",
    "\n",
    "# events\n",
    "for t in range(5):\n",
    "    e = threading.Event()\n",
    "    workdone_events.append(e)\n",
    "all_e = AllEvents(*workdone_events)\n",
    "\n",
    "# threads\n",
    "threads = [\n",
    "    threading.Thread(\n",
    "        target=runner,\n",
    "        args=(workdone_events[t], c_event_start, t)\n",
    "    )\n",
    "    for t in range(5)\n",
    "]\n",
    "cthread = threading.Thread(\n",
    "    target=coordinator,\n",
    "    args=(all_e, workdone_events, c_event_start,)\n",
    ")\n",
    "\n",
    "[\n",
    "    thread.start()\n",
    "    for thread in threads\n",
    "]\n",
    "cthread.start()\n",
    "\n",
    "[\n",
    "    thread.join()\n",
    "    for thread in threads\n",
    "]\n",
    "cthread.join()\n",
    "print('all done.')"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}