{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Concurent futures\n",
    "\n",
    "- [A QUICK INTRODUCTION TO THE CONCURRENT.FUTURES MODULE](http://masnun.com/2016/03/29/python-a-quick-introduction-to-the-concurrent-futures-module.html)\n",
    "- [RealPython - Speed Up Your Python Program With Concurrency](https://realpython.com/python-concurrency/)\n",
    "- [python3 非同步處理：thread, process 與 concurrent.futures 的兩三事](https://medium.com/@felixie/python3-%E9%9D%9E%E5%90%8C%E6%AD%A5%E8%99%95%E7%90%86-thread-process-%E8%88%87concurrent-futures%E5%85%A9%E4%B8%89%E4%BA%8B-f9d61fc7ccbf)\n",
    "\n",
    "複習一下作業系統 program/process/thread 觀念，program 執行被 load 到記憶體以一個或多個 process 的形式存在。\n",
    "\n",
    "- process 是 thread 的容器，同一個 process 間的 thread 共用資源而 process 間的資源彼此獨立。因此，同一個 process 的 thread 也共用同一個 GIL，要避開 GIL 就使用多個 process 就好了。\n",
    "- multiprocess 比起 multithread 最大的優點是不受 GIL 限制，cpu bound 的 task 執行會很快。而最大的缺點就是 process 間彼此無法共享資源、溝通會受到限制。\n",
    "- 不管是 multithread 或是 multiprocess，都需要耗費資源去產生 thread/process，而產生 process 的成本又比產生 thread 的成本高。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Executors\n",
    "\n",
    "This module features the Executor class which is an abstract class and it can not be used directly. However it has two very useful concrete subclasses – `ThreadPoolExecutor` and `ProcessPoolExecutor`. As their names suggest, one uses multi threading and the other one uses multi-processing. In both case, we get a pool of threads or processes and we can submit tasks to this pool. The pool would assign tasks to the available resources (threads or processes) and schedule them to run."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ThreadPoolExecutor\n",
    "\n",
    "[**Futures**](https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.Future)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "False\n",
      "True\n",
      "hello\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ThreadPoolExecutor\n",
    "from time import sleep\n",
    " \n",
    "def return_after_5_secs(message):\n",
    "    sleep(5)\n",
    "    return message\n",
    " \n",
    "pool = ThreadPoolExecutor(3)\n",
    " \n",
    "future = pool.submit(return_after_5_secs, (\"hello\"))\n",
    "print(future.done())\n",
    "sleep(5)\n",
    "print(future.done())\n",
    "print(future.result())\n",
    "\n",
    "# result() Return the value returned by the call.\n",
    "# If the call hasn’t yet completed then this method will wait up to timeout seconds."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ProcessPoolExecutor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "False\n",
      "False\n",
      "Result: hello\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ProcessPoolExecutor\n",
    "from time import sleep\n",
    "\n",
    "def return_after_5_secs(message):\n",
    "    sleep(5)\n",
    "    return message\n",
    "\n",
    "pool = ProcessPoolExecutor(3)\n",
    "\n",
    "future = pool.submit(return_after_5_secs, (\"hello\"))\n",
    "print(future.done())\n",
    "sleep(5)\n",
    "print(future.done())\n",
    "print(\"Result: \" + future.result())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Executor.map()\n",
    "\n",
    "Both executors have a common method – `map()`. Like the built in function, the map method allows multiple calls to a provided function, passing each of the items in an iterable to that function. Except, in this case, the functions are called concurrently. For multiprocessing, this iterable is broken into chunks and each of these chunks is passed to the function in separate processes. We can control the chunk size by passing a third parameter, `chunk_size`. By default the chunk size is 1."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "'http://www.foxnews.com/' page is 243520 bytes\n",
      "'http://some-made-up-domain.com/' page is 475 bytes\n",
      "'http://europe.wsj.com/' page is 954801 bytes\n",
      "'http://www.bbc.co.uk/' page is 327624 bytes\n",
      "'http://www.cnn.com/' page is 993570 bytes\n"
     ]
    }
   ],
   "source": [
    "import concurrent.futures\n",
    "import urllib.request\n",
    "\n",
    "URLS = ['http://www.foxnews.com/',\n",
    "        'http://www.cnn.com/',\n",
    "        'http://europe.wsj.com/',\n",
    "        'http://www.bbc.co.uk/',\n",
    "        'http://some-made-up-domain.com/']\n",
    "\n",
    "# Retrieve a single page and report the url and contents\n",
    "def load_url(url, timeout):\n",
    "    with urllib.request.urlopen(url, timeout=timeout) as conn:\n",
    "        return conn.read()\n",
    "\n",
    "\n",
    "# We can use a with statement to ensure threads are cleaned up promptly\n",
    "with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:\n",
    "    # Start the load operations and mark each future with its URL\n",
    "    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}\n",
    "    for future in concurrent.futures.as_completed(future_to_url):\n",
    "        url = future_to_url[future]\n",
    "        try:\n",
    "            data = future.result()\n",
    "        except Exception as exc:\n",
    "            print('%r generated an exception: %s' % (url, exc))\n",
    "        else:\n",
    "            print('%r page is %d bytes' % (url, len(data)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "112272535095293 is prime: True\n",
      "112582705942171 is prime: True\n",
      "112272535095293 is prime: True\n",
      "115280095190773 is prime: True\n",
      "115797848077099 is prime: True\n",
      "1099726899285419 is prime: False\n"
     ]
    }
   ],
   "source": [
    "import concurrent.futures\n",
    "import math\n",
    "\n",
    "PRIMES = [\n",
    "    112272535095293,\n",
    "    112582705942171,\n",
    "    112272535095293,\n",
    "    115280095190773,\n",
    "    115797848077099,\n",
    "    1099726899285419]\n",
    "\n",
    "def is_prime(n):\n",
    "    if n % 2 == 0:\n",
    "        return False\n",
    " \n",
    "    sqrt_n = int(math.floor(math.sqrt(n)))\n",
    "    for i in range(3, sqrt_n + 1, 2):\n",
    "        if n % i == 0:\n",
    "            return False\n",
    "    return True\n",
    "\n",
    "def main():\n",
    "    with concurrent.futures.ProcessPoolExecutor() as executor:\n",
    "        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):\n",
    "            print('%d is prime: %s' % (number, prime))\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### as_completed() & wait()\n",
    "\n",
    "The `concurrent.futures` module has two functions for dealing with the futures returned by the executors. One is `as_completed()` and the other one is `wait()`.\n",
    "\n",
    "The `as_completed()` function takes an *iterable* of Future objects and starts yielding values as soon as the futures start resolving. The main difference between the aforementioned `map` method with `as_completed` is that `map` returns the results *in the order* in which we pass the iterables. That is the first result from the map method is the result for the first item. On the other hand, the first result from the `as_completed` function is from whichever future completed first."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Return of 0\n",
      "Return of 4\n",
      "Return of 1\n",
      "Return of 2\n",
      "Return of 3\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ThreadPoolExecutor, wait, as_completed\n",
    "from time import sleep\n",
    "from random import randint\n",
    "\n",
    "def return_after_5_secs(num):\n",
    "    sleep(randint(1, 5))\n",
    "    return \"Return of {}\".format(num)\n",
    "\n",
    "pool = ThreadPoolExecutor(5)\n",
    "futures = []\n",
    "for x in range(5):\n",
    "    futures.append(pool.submit(return_after_5_secs, x))\n",
    "\n",
    "for x in as_completed(futures):\n",
    "    print(x.result())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `wait()` function would return a **named tuple** which contains two set – one set contains the futures which **completed** `done` (either got result or exception) and the other set containing the ones which **didn’t complete** `not_done`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DoneAndNotDoneFutures(done={<Future at 0x108d038d0 state=finished returned str>}, not_done={<Future at 0x10af48a20 state=running>, <Future at 0x10af485c0 state=running>, <Future at 0x10af48dd8 state=running>, <Future at 0x10af48278 state=running>})\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import ThreadPoolExecutor, wait, as_completed, FIRST_COMPLETED\n",
    "from time import sleep\n",
    "from random import randint\n",
    "\n",
    "def return_after_5_secs(num):\n",
    "    sleep(randint(1, 5))\n",
    "    return \"Return of {}\".format(num)\n",
    "\n",
    "pool = ThreadPoolExecutor(5)\n",
    "futures = []\n",
    "for x in range(5):\n",
    "    futures.append(pool.submit(return_after_5_secs, x))\n",
    "\n",
    "print(wait(futures, return_when=FIRST_COMPLETED))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}