Ch7-1_Concurrency.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Async programming"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## I/O 대기\n",
    "+ CPU 클럭의 속도가 1초라고 가정했을 때\n",
    "+ RAM은 20초, SSD는 4일, HDD 6개월, 네트워크 전송은?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 동시성\n",
    "+ 여러 개의 요청을 동시에 다룰 수 있는 시스템을 구현하는 방식\n",
    "+ 아래는 동기식으로 구현된 기본 함수"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time \n",
    "import functools\n",
    "\n",
    "\n",
    "DEFAULT_FMT = '[{elapsed:0.8f}s] {name}({args}, {kwargs}) -> {result}'\n",
    "\n",
    "def clock(fmt=DEFAULT_FMT): \n",
    "    def decorate(func): \n",
    "        @functools.wraps(func)\n",
    "        def clocked(*_args, **_kwargs): # clocked에서 *, ** 키워드를 통해 설정된 인수를 변수화\n",
    "            t0 = time.time()\n",
    "            _result = func(*_args)\n",
    "            elapsed = time.time() - t0\n",
    "            name = func.__name__\n",
    "            args = ', '.join(repr(arg) for arg in _args)\n",
    "            pairs = ['%s=%r' % (k, w) for k, w in sorted(_kwargs.items())]\n",
    "            kwargs = ', '.join(pairs)\n",
    "            result = repr(_result)\n",
    "            print(fmt.format(**locals()))\n",
    "            return _result # clocked()는 데커레이트된 함수를 대체하므로, 원래 함수가 반환하는 값을 반환해야 한다.\n",
    "        return clocked     # decorate()는 clocked()를 반환한다. \n",
    "    return decorate        # clock()은 decorate()를 반환한다. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1.00051713s] fetch_square(0, ) -> 0\n",
      "[1.00127792s] fetch_square(1, ) -> 1\n",
      "[1.00123954s] fetch_square(2, ) -> 4\n",
      "[1.00055385s] fetch_square(3, ) -> 9\n",
      "[1.00105548s] fetch_square(4, ) -> 16\n",
      "[1.00137401s] fetch_square(5, ) -> 25\n",
      "[1.00112104s] fetch_square(6, ) -> 36\n",
      "[1.00127673s] fetch_square(7, ) -> 49\n",
      "[1.00112653s] fetch_square(8, ) -> 64\n",
      "[1.00130653s] fetch_square(9, ) -> 81\n",
      "Total: 10.016206741333008 sec\n"
     ]
    }
   ],
   "source": [
    "def network_request(number):\n",
    "    time.sleep(1.0)\n",
    "    return {\"success\": True, \"result\": number ** 2}\n",
    "\n",
    "\n",
    "@clock()\n",
    "def fetch_square(number):\n",
    "    response = network_request(number)\n",
    "    if response[\"success\"]:\n",
    "        return response[\"result\"]\n",
    "\n",
    "        \n",
    "if __name__ == \"__main__\":\n",
    "    t0 = time.time()\n",
    "\n",
    "    returns = [fetch_square(i) for i in range(10)]\n",
    "    \n",
    "    elapsed = time.time() - t0\n",
    "    print(\"Total:\", elapsed, \"sec\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Callback\n",
    "\n",
    "+ 원리만 이해하고 복잡하니 쓰지 말자"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "\n",
    "\n",
    "def network_request_async(number, on_done):\n",
    "    \n",
    "    def timer_done():\n",
    "        time.sleep(1.0)\n",
    "        on_done({\"success\": True, \"result\": number ** 2})\n",
    "\n",
    "    timer = threading.Thread(target=timer_done, args=[])\n",
    "    timer.start()\n",
    "    \n",
    "    \n",
    "@clock()\n",
    "def fetch_square_async(number):\n",
    "    \n",
    "    def on_done(response):\n",
    "        if response[\"success\"]:\n",
    "            return response[\"result\"]\n",
    "        \n",
    "    network_request_async(number, on_done)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.00027442s] fetch_square_async(0, ) -> None\n",
      "[0.00050879s] fetch_square_async(1, ) -> None\n",
      "[0.00332260s] fetch_square_async(2, ) -> None\n",
      "[0.00051570s] fetch_square_async(3, ) -> None\n",
      "[0.00053382s] fetch_square_async(4, ) -> None\n",
      "[0.00039411s] fetch_square_async(5, ) -> None\n",
      "[0.00046110s] fetch_square_async(6, ) -> None\n",
      "[0.00057125s] fetch_square_async(7, ) -> None\n",
      "[0.00052476s] fetch_square_async(8, ) -> None\n",
      "[0.00072122s] fetch_square_async(9, ) -> None\n",
      "Total: 0.009291410446166992 sec\n"
     ]
    }
   ],
   "source": [
    "if __name__ == \"__main__\":\n",
    "    t0 = time.time()\n",
    "\n",
    "    returns = [fetch_square_async(i) for i in range(10)]\n",
    "    \n",
    "    elapsed = time.time() - t0\n",
    "    print(\"Total:\", elapsed, \"sec\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Future\n",
    "\n",
    "+ 요청한 자원을 추적하고 가용하게 될 때 까지 대기하는 데 도움이 되는 추상화"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from concurrent.futures import Future\n",
    "\n",
    "def callback(future):\n",
    "    print(future.result()[::-1])\n",
    "\n",
    "fut = Future()\n",
    "fut.add_done_callback(callback)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'_condition': <Condition(<unlocked _thread.RLock object owner=0 count=0 at 0x7f94dc577210>, 0)>,\n",
       " '_state': 'PENDING',\n",
       " '_result': None,\n",
       " '_exception': None,\n",
       " '_waiters': [],\n",
       " '_done_callbacks': [<function __main__.callback(future)>]}"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "vars(fut)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "OLLEH\n"
     ]
    }
   ],
   "source": [
    "fut.set_result(\"HELLO\") ## 퓨쳐를 리턴함"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'_condition': <Condition(<unlocked _thread.RLock object owner=0 count=0 at 0x7f94dc577210>, 0)>,\n",
       " '_state': 'FINISHED',\n",
       " '_result': 'HELLO',\n",
       " '_exception': None,\n",
       " '_waiters': [],\n",
       " '_done_callbacks': [<function __main__.callback(future)>]}"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "vars(fut)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "+ Network_request_async 를 쓰레드 방식으로 변환"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "def network_reqeust_async(number):\n",
    "    future = Future()\n",
    "    result = {\"success\": True, \"result\": number ** 2}\n",
    "    timer = threading.Timer(1.0, lambda: future.set_result(result))\n",
    "    timer.start()\n",
    "    return future\n",
    "\n",
    "\n",
    "@clock()\n",
    "def fetch_square_async(number):\n",
    "    fut = network_reqeust_async(number)\n",
    "\n",
    "    def on_done_future(future):\n",
    "        response = future.result()\n",
    "        if response[\"success\"]:\n",
    "#             return response[\"result\"]\n",
    "            print(response[\"result\"], flush=True)        \n",
    "        \n",
    "    fut.add_done_callback(on_done_future)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.00061083s] fetch_square_async(0, ) -> None\n",
      "[0.00058317s] fetch_square_async(1, ) -> None\n",
      "[0.00023699s] fetch_square_async(2, ) -> None\n",
      "[0.00046134s] fetch_square_async(3, ) -> None\n",
      "[0.00120807s] fetch_square_async(4, ) -> None\n",
      "[0.00201869s] fetch_square_async(5, ) -> None\n",
      "[0.00043750s] fetch_square_async(6, ) -> None\n",
      "[0.00081587s] fetch_square_async(7, ) -> None\n",
      "[0.00057101s] fetch_square_async(8, ) -> None\n",
      "[0.00039887s] fetch_square_async(9, ) -> None\n",
      "Total: 0.008257627487182617 sec\n",
      "0\n",
      "1\n",
      "49\n",
      "\n",
      "16\n",
      "2536\n",
      "\n",
      "4964\n",
      "\n",
      "81\n"
     ]
    }
   ],
   "source": [
    "if __name__ == \"__main__\":\n",
    "    t0 = time.time()\n",
    "\n",
    "    returns = [fetch_square_async(i) for i in range(10)]\n",
    "    \n",
    "    elapsed = time.time() - t0\n",
    "    print(\"Total:\", elapsed, \"sec\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## asyncio 프레임워크\n",
    "+ loop = asyncio.get_event_loop() 함수를 호출해 asyncio 루프를 얻을 수 있음\n",
    "+ loop.call_later() 를 사용해 콜백 실행을 예약할 수 있음\n",
    "+ loop.stop 메소드를 사용해서 루프 종료\n",
    "+ loop.run_forever() 로 루프 시작 가능\n",
    "+ loop.run_until_complete()을 통해 개별적인 실행 가능\n",
    "+ ensure_future()는 Task 인스턴스를 반환해서 실행하고 싶은 코루틴을 이벤트 루프로 전달 가능"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import nest_asyncio\n",
    "nest_asyncio.apply() # This module patches asyncio to allow nested (due to jupyter notebook)\n",
    "\n",
    "async def network_request(number):\n",
    "    await asyncio.sleep(1.0)\n",
    "    return {\"success\": True, \"result\": number ** 2}\n",
    "\n",
    "@clock()\n",
    "async def fetch_square(number):\n",
    "    response = await network_request(number)\n",
    "    if response[\"success\"]:\n",
    "        print(response[\"result\"], flush=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.00000095s] fetch_square(0, ) -> <coroutine object fetch_square at 0x7f94dc58fb00>\n",
      "0\n",
      "[0.00000262s] fetch_square(1, ) -> <coroutine object fetch_square at 0x7f94dc58fb00>\n",
      "1\n",
      "[0.00000310s] fetch_square(2, ) -> <coroutine object fetch_square at 0x7f94dc58fb00>\n",
      "4\n",
      "Total: 3.0104825496673584 sec\n"
     ]
    }
   ],
   "source": [
    "if __name__ == \"__main__\":\n",
    "    loop = asyncio.get_event_loop()\n",
    "    t0 = time.time()\n",
    "\n",
    "    loop.run_until_complete(fetch_square(0))\n",
    "    loop.run_until_complete(fetch_square(1))\n",
    "    loop.run_until_complete(fetch_square(2))\n",
    "    \n",
    "    elapsed = time.time() - t0\n",
    "    print(\"Total:\", elapsed, \"sec\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.00000095s] fetch_square(0, ) -> <coroutine object fetch_square at 0x7f94dc58f560>\n",
      "[0.00000119s] fetch_square(1, ) -> <coroutine object fetch_square at 0x7f94dc5265f0>\n",
      "[0.00000048s] fetch_square(2, ) -> <coroutine object fetch_square at 0x7f94dc526320>\n",
      "0\n",
      "1\n",
      "4\n",
      "Total: 1.0085382461547852 sec\n"
     ]
    }
   ],
   "source": [
    "if __name__ == \"__main__\":\n",
    "    loop = asyncio.get_event_loop()\n",
    "    t0 = time.time()\n",
    "\n",
    "    loop.run_until_complete(asyncio.gather(fetch_square(0), fetch_square(1), fetch_square(2)))\n",
    "    \n",
    "    elapsed = time.time() - t0\n",
    "    print(\"Total:\", elapsed, \"sec\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.00000143s] fetch_square(0, ) -> <coroutine object fetch_square at 0x7f94dc58f950>\n",
      "[0.00000191s] fetch_square(1, ) -> <coroutine object fetch_square at 0x7f94dc58f8c0>\n",
      "[0.00002360s] fetch_square(2, ) -> <coroutine object fetch_square at 0x7f94dc4d2050>\n",
      "Total: 0.0013251304626464844 sec\n",
      "0\n",
      "1\n",
      "4\n"
     ]
    }
   ],
   "source": [
    "if __name__ == \"__main__\":\n",
    "    loop = asyncio.get_event_loop()\n",
    "    t0 = time.time()\n",
    "\n",
    "    asyncio.ensure_future(fetch_square(0))\n",
    "    asyncio.ensure_future(fetch_square(1))\n",
    "    asyncio.ensure_future(fetch_square(2))\n",
    "    \n",
    "#     loop.run_forever()\n",
    "\n",
    "    elapsed = time.time() - t0\n",
    "    print(\"Total:\", elapsed, \"sec\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "0it [00:00, ?it/s]\n",
      "0it [00:00, ?it/s]\u001b[A\n",
      "0it [00:00, ?it/s]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Empty DataFrame\n",
      "Columns: []\n",
      "Index: []\n",
      "[0.00909424s] main(, ) -> None\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "import asyncio\n",
    "import json\n",
    "from pathlib import Path\n",
    "\n",
    "from collections import defaultdict\n",
    "import pandas as pd\n",
    "from tqdm import tqdm\n",
    "\n",
    "import time\n",
    "\n",
    "\n",
    "path = Path(\"/home/shyeon/workspace/python/SparkDefinitiveGuide/data/activity-data/\")\n",
    "\n",
    "\n",
    "def load_json_byline(path):\n",
    "    with open(path, \"r\", encoding=\"UTF-8\") as f:\n",
    "        for line in f:\n",
    "            yield line\n",
    "\n",
    "\n",
    "def load_json_byfile(fps):\n",
    "    for fp in tqdm(fps):\n",
    "        yield from load_json_byline(fp)\n",
    "\n",
    "\n",
    "@clock()\n",
    "def main():\n",
    "    dd = defaultdict(list)\n",
    "    files = path.glob(\"*.json\")\n",
    "    \n",
    "    for lines in tqdm(load_json_byfile(files)):\n",
    "        for key, value in json.loads(lines).items():\n",
    "            dd[key].append(value)\n",
    "                \n",
    "    result = pd.DataFrame(dd)\n",
    "    print(result.head())    \n",
    "    \n",
    "    \n",
    "if __name__ == \"__main__\":\n",
    "    main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "0it [00:00, ?it/s]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Empty DataFrame\n",
      "Columns: []\n",
      "Index: []\n",
      "[0.00340915s] main(, ) -> None\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "import asyncio\n",
    "import json\n",
    "from pathlib import Path\n",
    "\n",
    "from collections import defaultdict\n",
    "import pandas as pd\n",
    "from tqdm import tqdm\n",
    "\n",
    "import time\n",
    "\n",
    "\n",
    "path = Path(\"/home/shyeon/workspace/python/SparkDefinitiveGuide/data/activity-data/\")\n",
    "\n",
    "\n",
    "@asyncio.coroutine\n",
    "def load_json_byline(path):\n",
    "    with open(path, \"r\", encoding=\"UTF-8\") as fp:\n",
    "        yield from fp\n",
    "\n",
    "\n",
    "@asyncio.coroutine        \n",
    "def load_json_byfile(fps):\n",
    "    for fp in fps:\n",
    "        yield from load_json_byline(fp)\n",
    "\n",
    "\n",
    "@clock()\n",
    "def main():\n",
    "    dd = defaultdict(list)\n",
    "    files = path.glob(\"*.json\")\n",
    "    \n",
    "    for lines in tqdm(load_json_byfile(files)):\n",
    "        for key, value in json.loads(lines).items():\n",
    "            dd[key].append(value)\n",
    "                \n",
    "    result = pd.DataFrame(dd)\n",
    "    print(result.head())    \n",
    "    \n",
    "    \n",
    "if __name__ == \"__main__\":\n",
    "    main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "0it [00:00, ?it/s]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.00000119s] main(, ) -> <generator object main at 0x7f947e01a6d0>\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "\n",
    "import asyncio\n",
    "import nest_asyncio\n",
    "nest_asyncio.apply() # This module patches asyncio to allow nested (due to jupyter notebook)\n",
    "import aiofile\n",
    "\n",
    "import json\n",
    "import pandas as pd\n",
    "from collections import defaultdict\n",
    "from pathlib import Path\n",
    "from tqdm import tqdm\n",
    "\n",
    "\n",
    "path = Path(\"/home/shyeon/workspace/python/SparkDefinitiveGuide/data/activity-data/\")\n",
    "\n",
    "\n",
    "@asyncio.coroutine\n",
    "def load_json_byline(path):\n",
    "    with open(path, \"r\", encoding=\"UTF-8\") as fp:\n",
    "        yield from fp\n",
    "\n",
    "\n",
    "@asyncio.coroutine        \n",
    "def load_json_byfile(fps):\n",
    "    for fp in fps:\n",
    "        yield from load_json_byline(fp)\n",
    "\n",
    "\n",
    "def convert_json2dict(json_str:str, dic:dict):\n",
    "    for key, value in json.loads(json_str).items():\n",
    "        dic[key].append(value)\n",
    "        \n",
    "        \n",
    "@clock()\n",
    "@asyncio.coroutine   \n",
    "def main():\n",
    "    dd = defaultdict(list)\n",
    "    files = path.glob(\"*.json\")\n",
    "    \n",
    "    task = [convert_json2dict(lines, dd) for lines in tqdm(load_json_byfile(files))]\n",
    "    \n",
    "    return task\n",
    "                \n",
    "#     result = pd.DataFrame(dd)\n",
    "#     print(result.head())\n",
    "    \n",
    "    \n",
    "if __name__ == \"__main__\":\n",
    "    \n",
    "    loop = asyncio.get_event_loop()\n",
    "    loop.run_until_complete(asyncio.gather(*main()))\n",
    "#     loop.close() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.00005937s] main(, ) -> []\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "\n",
    "import asyncio\n",
    "import nest_asyncio\n",
    "nest_asyncio.apply() # This module patches asyncio to allow nested (due to jupyter notebook)\n",
    "from aiofile import AIOFile, Reader, Writer\n",
    "\n",
    "import json\n",
    "import pandas as pd\n",
    "from collections import defaultdict\n",
    "from pathlib import Path\n",
    "from tqdm import tqdm\n",
    "\n",
    "\n",
    "path = Path(\"/home/shyeon/workspace/python/SparkDefinitiveGuide/data/activity-data/\")\n",
    "\n",
    "\n",
    "async def load_json(path):\n",
    "    async with AIOFile(path, \"r\", encoding=\"UTF-8\") as afp:\n",
    "        print(await afp.read())\n",
    "\n",
    "\n",
    "\n",
    "def convert_json2dict(json_str:str, dic:dict):\n",
    "    for key, value in json.loads(json_str).items():\n",
    "        dic[key].append(value)\n",
    "        \n",
    "        \n",
    "@clock()\n",
    "def main():\n",
    "    dd = defaultdict(list)\n",
    "    files = path.glob(\"*.json\")\n",
    "    \n",
    "    tasks = [load_json(file) for file in files]\n",
    "    \n",
    "    return tasks\n",
    "                \n",
    "#     result = pd.DataFrame(dd)\n",
    "#     print(result.head())\n",
    "    \n",
    "    \n",
    "if __name__ == \"__main__\":\n",
    "    \n",
    "    loop = asyncio.get_event_loop()\n",
    "    loop.run_until_complete(asyncio.gather(*main()))\n",
    "#     loop.close() "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}