asyncio_talk.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "fcd40906-a2f5-4bb0-bb86-2b70ec1d893c",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "#### meetup python toulouse\n",
    "\n",
    "# asyncio\n",
    "## Pourquoi c'est nécessaire et comment ça marche ?"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "af1a2cb3-7e40-4554-ab6e-8fb75dde838e",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "notes"
    },
    "tags": []
   },
   "source": [
    "Retours sur l'histoire et le besoin de la concurrence en developpement, puis sur l'implementation en python de la librairie de reference d'IO asynchrone en python"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "1ab967d0-40ed-4965-a6c7-9603443db4af",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "skip"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "import time\n",
    "import asyncio\n",
    "import socket\n",
    "from itertools import islice\n",
    "from datetime import datetime, timedelta\n",
    "import select\n",
    "from functools import wraps\n",
    "import warnings\n",
    "import httpx\n",
    "import requests\n",
    "httpx = httpx.AsyncClient()\n",
    "warnings.filterwarnings(\"ignore\")\n",
    "\n",
    "now = datetime.now\n",
    "class Timer:\n",
    "    def __init__(self, prefix=None):\n",
    "        self.start = datetime.now()\n",
    "        self.prefix = prefix\n",
    "    \n",
    "    def prefixfmt(slef):\n",
    "        if p := slef.prefix:\n",
    "            return f'[{p}]'\n",
    "        return ''\n",
    "        \n",
    "    def log(self, message):\n",
    "        delta = datetime.now() - self.start\n",
    "        print(f'{self.prefixfmt()}[{delta}] {message}')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2baac051-b288-4dae-9a4b-4d803374f76a",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Asyncio\n",
    "\n",
    "- what it means\n",
    "- what it does"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d854f3ac-6a57-404d-82c3-7b3778f0aa4b",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# what is syncIO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "ae743af5-ba82-44ad-953a-f92e34f8cc7e",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0:00:00.000252] starting\n",
      "[0:00:04.006964] fetched hourly\n",
      "[0:00:08.011324] fetched daily\n",
      "[0:00:08.011638] finished\n"
     ]
    }
   ],
   "source": [
    "t = Timer()\n",
    "def weather_hourly():\n",
    "    hourly = requests.get(\"http://localhost:8081\")\n",
    "    t.log('fetched hourly')\n",
    "    return hourly\n",
    "\n",
    "def weather_daily():\n",
    "    daily = requests.get(\"http://localhost:8081\")\n",
    "    t.log('fetched daily')\n",
    "    return daily\n",
    "    \n",
    "def dashboard_data():\n",
    "    return {\n",
    "        \"hourly\": weather_hourly(),\n",
    "        \"daily\": weather_daily(),\n",
    "    }\n",
    "\n",
    "t.log(\"starting\")\n",
    "dashboard_data()\n",
    "t.log(\"finished\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7adb3b74-96c0-44b3-9b98-9cf807a5b84a",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "# what is asyncIO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "869f0586-f299-412e-b4d2-63a1c945610c",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0:00:00.000308] starting\n",
      "[0:00:04.016179] fetched daily\n",
      "[0:00:04.016740] fetched hourly\n",
      "[0:00:04.016820] finished\n"
     ]
    }
   ],
   "source": [
    "t = Timer()\n",
    "async def weather_hourly():\n",
    "    hourly = await httpx.get(\"http://localhost:8081\")\n",
    "    t.log('fetched hourly')\n",
    "    return hourly\n",
    "\n",
    "async def weather_daily():\n",
    "    daily = await httpx.get(\"http://localhost:8081\")\n",
    "    t.log('fetched daily')\n",
    "    return daily\n",
    "    \n",
    "\n",
    "async def dashboard_data():\n",
    "    hourly = asyncio.create_task(weather_hourly())\n",
    "    daily = asyncio.create_task(weather_daily())\n",
    "    return {\n",
    "        \"hourly\": await hourly,\n",
    "        \"daily\": await daily,\n",
    "    }\n",
    "\n",
    "t.log(\"starting\")\n",
    "await dashboard_data()\n",
    "t.log(\"finished\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b52935d4-e3a9-468a-9926-0320465f4ea0",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# What is IO ?"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "03c737fa-9e8a-4fa3-ba03-6d0c7a9ebb26",
   "metadata": {
    "editable": true,
    "raw_mimetype": "",
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Blocking IO vs non blocking IO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "bab14e18-ff4c-4e10-88cb-c73ae39e43cb",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0:00:00.000002] starting function\n",
      "[0:00:00.000252] connected\n",
      "[0:00:00.000285] sent payload\n",
      "[0:00:02.001922] received response\n",
      "response: \"Hello, world\"\n"
     ]
    }
   ],
   "source": [
    "def blocking_io():\n",
    "    t = Timer()\n",
    "    t.log('starting function')\n",
    "    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n",
    "    s.connect(('127.0.0.1', 8080))\n",
    "    t.log('connected')\n",
    "    s.sendall(f'Hello, world'.encode())\n",
    "    t.log('sent payload')\n",
    "    data = s.recv(1024)\n",
    "    t.log('received response')\n",
    "    s.close()\n",
    "    return data.decode()\n",
    "\n",
    "print(f'response: \"{blocking_io()}\"')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "afebe4e7-abef-4ca1-9cf0-8d044ab5a947",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def non_blocking_io():\n",
    "    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n",
    "    s.connect(('127.0.0.1', 8080))\n",
    "\n",
    "    s.setblocking(False)\n",
    "    \n",
    "    s.sendall(f'Hello, world'.encode()) # ???\n",
    "    data = s.recv(1024) # ???\n",
    "    s.close()\n",
    "    return data.decode()\n",
    "\n",
    "# print(f'response: \"{non_blocking_io()}\"')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "21720f1d-e5bb-46e9-8faf-865609665799",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "# Extracting socket setup for readability"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "dabec00f-4b20-44b6-83f7-ec891328d5bb",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def setup_non_blocking_socket():\n",
    "    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n",
    "    s.connect(('127.0.0.1', 8080))\n",
    "    s.setblocking(False)\n",
    "    s.sendall('Hello, world'.encode())\n",
    "    return s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "0a947a5d-ac41-4456-a282-c3648d89bd25",
   "metadata": {
    "editable": true,
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0:00:00.000231] socket not ready to read\n",
      "[0:00:00.500450] socket not ready to read\n",
      "[0:00:01.000774] socket not ready to read\n",
      "[0:00:01.501074] socket not ready to read\n",
      "[0:00:02.001347] socket not ready to read\n",
      "response: \"Hello, world\"\n"
     ]
    }
   ],
   "source": [
    "def non_blocking_io():\n",
    "    t = Timer()\n",
    "    s = setup_non_blocking_socket()\n",
    "    \n",
    "    while True:\n",
    "        r,_,_ = select.select([s.fileno()], [], [], 0)\n",
    "        t.log('socket not ready to read')\n",
    "        if len(r) > 0: break\n",
    "        time.sleep(0.5) # still blocking\n",
    "        \n",
    "    data = s.recv(1024)\n",
    "    s.close()\n",
    "    return data.decode()\n",
    "\n",
    "print(f'response: \"{non_blocking_io()}\"')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bfce8aff-ce07-4e83-bdfa-1fa7ad933e94",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Generators are coroutines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "160914c6-c47e-4779-9740-1f7b05ed4321",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "inside coroutine 2\n",
      "2\n",
      "inside coroutine 3\n",
      "3\n",
      "inside coroutine 4\n",
      "4\n",
      "inside coroutine 5\n",
      "5\n",
      "inside coroutine 6\n",
      "6\n"
     ]
    }
   ],
   "source": [
    "def coroutine():\n",
    "    a = 1\n",
    "    while True:\n",
    "        a += 1\n",
    "        print('inside coroutine', a)\n",
    "        yield a\n",
    "\n",
    "gen = coroutine()\n",
    "print(next(gen))\n",
    "print(next(gen))\n",
    "print(next(gen))\n",
    "print(next(gen))\n",
    "print(next(gen))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "88ed6541-1afa-40a6-acb9-8ce4890b9f97",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def really_non_blocking_io():\n",
    "    s = setup_non_blocking_socket()\n",
    "    \n",
    "    while True:\n",
    "        r,_,_ = select.select([s.fileno()], [], [], 0)\n",
    "        if len(r) > 0: break\n",
    "        yield\n",
    "        \n",
    "    data = s.recv(1024)\n",
    "    s.close()\n",
    "    return data.decode()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "cf9e5712-ff37-4162-ae7e-8296dcdaa04d",
   "metadata": {
    "editable": true,
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "waiting...\n",
      "waiting...\n",
      "waiting...\n",
      "waiting...\n",
      "response: \"Hello, world\"\n"
     ]
    }
   ],
   "source": [
    "gen = really_non_blocking_io()\n",
    "try:\n",
    "    while True:\n",
    "        next(gen)\n",
    "        print('waiting...')\n",
    "        time.sleep(.5)\n",
    "except StopIteration as response:\n",
    "    print(f'response: \"{response.value}\"')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0b4e1dda-c942-4f8d-8290-df2fcb37039f",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Async executor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "4a7d049b-145c-484b-ae1f-8b3e14d50388",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def run_until_complete(task):\n",
    "    try:\n",
    "        while True:\n",
    "            next(task)\n",
    "            time.sleep(.5)\n",
    "    except StopIteration as response:\n",
    "        return response.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "5fb97f1e-21bb-48aa-b732-66b7c1c39f38",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "skip"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "# cheeky redefinition for slightly more features\n",
    "def run_until_complete(task, wait=.05, debug=None):\n",
    "    try:\n",
    "        while True:\n",
    "            next(task)\n",
    "            if debug:\n",
    "                if type(debug) == bool:\n",
    "                    print('executor waiting...')\n",
    "                else:\n",
    "                    debug()\n",
    "            time.sleep(wait)\n",
    "    except StopIteration as response:\n",
    "        return response.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "7ea0fd9c-5dab-4334-b9e0-4cbf7388b0a1",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0:00:00.000051] running task\n",
      "[0:00:02.005236] ret='Hello, world'\n"
     ]
    }
   ],
   "source": [
    "t = Timer()\n",
    "task = really_non_blocking_io()\n",
    "t.log('running task')\n",
    "ret = run_until_complete(task)\n",
    "t.log(f'{ret=}')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e319c66b-494b-44f6-a675-8ec9883f549f",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# async IO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "540d2802-0a0f-4bab-9fea-738820457878",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "class NonBlockingSocket:\n",
    "    def __init__(self, address=('127.0.0.1', 8080)):\n",
    "        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n",
    "        self.socket.connect(address)\n",
    "        self.socket.setblocking(False)\n",
    "\n",
    "    def sendall(self, value):\n",
    "        while len(select.select([], [self.socket.fileno()], [], 0)[1]) == 0:\n",
    "            yield\n",
    "        return self.socket.sendall(value)\n",
    "\n",
    "    def recv(self, amount):\n",
    "        while len(select.select([self.socket.fileno()], [], [], 0)[0]) == 0:\n",
    "            yield\n",
    "        return self.socket.recv(amount)\n",
    "\n",
    "    def close(self):\n",
    "        return self.socket.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "dff171d7-6804-4b24-8821-5e99df1e3207",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def really_non_blocking_io():\n",
    "    s = NonBlockingSocket()\n",
    "    s.sendall(\"hello world\")\n",
    "    data = s.recv(1024)\n",
    "    s.close()\n",
    "    return data.decode()\n",
    "    \n",
    "# ret = really_non_blocking_io()\n",
    "# print(f'response: \"{ret}\"')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "c5ff7ce6-923d-412c-b38d-e7438a938ea2",
   "metadata": {
    "editable": true,
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0:00:00.000204] send\n",
      "[0:00:00.000342] recv\n",
      "[0:00:02.007581] close\n",
      "Hello world\n"
     ]
    }
   ],
   "source": [
    "def not_really_non_blocking_io():\n",
    "    t = Timer()\n",
    "    s = NonBlockingSocket()\n",
    "    t.log('send')\n",
    "    run_until_complete(s.sendall('Hello world'.encode()))\n",
    "    t.log('recv')\n",
    "    data = run_until_complete(s.recv(1024))\n",
    "    t.log('close')\n",
    "    s.close()\n",
    "    return data.decode()\n",
    "\n",
    "print(not_really_non_blocking_io())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0edd8419-5fe8-4ad3-8ea1-9b5b079d32d4",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Generators all the way down"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "e1953df1-0357-4d6a-8e50-073e8f500275",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0:00:00.000135] trying to write\n",
      "[0:00:00.000361] trying to read\n",
      "[0:00:00.500480] trying to read\n",
      "[0:00:01.000948] trying to read\n",
      "[0:00:01.501164] trying to read\n",
      "[0:00:02.001448] trying to read\n",
      "Hello world\n"
     ]
    }
   ],
   "source": [
    "def really_really_non_blocking_io():\n",
    "    t = Timer()\n",
    "    s = NonBlockingSocket()\n",
    "\n",
    "    gen = s.sendall('Hello world'.encode())\n",
    "    try:\n",
    "        while True:\n",
    "            t.log('trying to write')\n",
    "            yield next(gen)\n",
    "    except StopIteration as response:\n",
    "        pass\n",
    "    \n",
    "    gen = s.recv(1024)\n",
    "    try:\n",
    "        while True:\n",
    "            t.log('trying to read')\n",
    "            yield next(gen)\n",
    "    except StopIteration as response:\n",
    "        s.close()\n",
    "        return response.value.decode()\n",
    "\n",
    "print(run_until_complete(really_really_non_blocking_io(), wait=.5))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e0403043-341c-4358-9131-5e3706129d44",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Yield from"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "c456e4eb-4cb6-4ad2-8908-2c563e0114fc",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "executor waiting...\n",
      "executor waiting...\n",
      "executor waiting...\n",
      "executor waiting...\n",
      "Hello world\n"
     ]
    }
   ],
   "source": [
    "def really_really_non_blocking_io():\n",
    "    s = NonBlockingSocket()\n",
    "    yield from s.sendall('Hello world'.encode())\n",
    "    data = yield from s.recv(1024)\n",
    "    s.close()\n",
    "    return data.decode()\n",
    "\n",
    "print(run_until_complete(really_really_non_blocking_io(), wait=.5, debug=True))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ccea0224-8360-4b99-aff0-7346c5166288",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Awaitable"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "b1d64b34-6ec7-4442-8c86-bac92038c12d",
   "metadata": {
    "editable": true,
    "scrolled": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "# how to write :\n",
    "async def func():\n",
    "    await really_non_blocking_io(1)\n",
    "# await func()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "2efcc1ce-fe39-46a2-b193-ee9281f612a2",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "class Awaitable:\n",
    "    def __init__(self, gen):\n",
    "        self._gen = gen\n",
    "        \n",
    "    def __await__(self):\n",
    "        ret = yield from self._gen\n",
    "        return ret\n",
    "        \n",
    "    __iter__ = __await__ # still a generator\n",
    "\n",
    "    def __next__(self):\n",
    "        next(self._gen)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "db8a3d51-77ec-4887-ba3d-c8e342c907c7",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Hello world'"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def make_awaitable(f):\n",
    "    def wrapper(*args, **kwds):\n",
    "        gen = f(*args, **kwds)\n",
    "        return Awaitable(gen)\n",
    "    return wrapper\n",
    "\n",
    "awaitable_non_blocking_io = make_awaitable(really_really_non_blocking_io)\n",
    "await awaitable_non_blocking_io()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9694f751-9d9c-4861-8c4b-c8086c9b4f11",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Actual event loop\n",
    "\n",
    "```python\n",
    "loop.create_task(meteo_hourly()) #  ??\n",
    "loop.create_task(meteo_daily()) #  ??\n",
    "loop.run() #  ??\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "dd77e1be-345d-4e44-9775-a615a90b7f42",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "skip"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "@make_awaitable\n",
    "def asleep(s):\n",
    "    target = now() + timedelta(seconds=s)\n",
    "    while now() < target:\n",
    "        yield\n",
    "    return"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "377d2805-3c3c-45e3-9919-f13371e9dee6",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "class Task:\n",
    "    '''\n",
    "    Lets not talk about this too much for now\n",
    "    '''\n",
    "    def __init__(self, coro):\n",
    "        self._coro = coro\n",
    "        \n",
    "    def _step(self):\n",
    "        try:\n",
    "            return self._coro.send(None)\n",
    "        except StopIteration as result:\n",
    "            raise result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "4887f546-0a77-45cb-bec5-c528f69b5676",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "skip"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "# sneaky rewrite to add name and repr\n",
    "\n",
    "class BaseTask:\n",
    "    def __init__(self, coro, name=None):\n",
    "        self._coro = coro\n",
    "        self._name = name\n",
    "\n",
    "    def __repr__(self):\n",
    "        return f'<{self._name}>' if self._name is not None else super().__repr__()\n",
    "        \n",
    "    def _step(self):\n",
    "        try:\n",
    "            return self._coro.send(None)\n",
    "        except StopIteration as result:\n",
    "            raise result\n",
    "\n",
    "class Task(BaseTask):\n",
    "    pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "9d0c3011-7929-4f92-829b-b854de231a80",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "class BaseLoop:\n",
    "    tasks = []\n",
    "\n",
    "    def create_task(self, coro):\n",
    "        task = Task(coro)\n",
    "        self._schedule_task(task)\n",
    "        return task\n",
    "\n",
    "    def _schedule_task(self, task):\n",
    "        self.tasks.append(task)\n",
    "\n",
    "    def run_until_complete(self):\n",
    "        while len(self.tasks) != 0:\n",
    "            print('loop queue', self.tasks)\n",
    "            self.advance()\n",
    "            time.sleep(0.5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "cb98d6c8-c46d-40a4-8f50-e612216ff06c",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "skip"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "# sneaky rewrite to add the immediate arg and not use a global tasks list\n",
    "\n",
    "class BaseLoop:\n",
    "    def __init__(self):\n",
    "        self.tasks = []\n",
    "\n",
    "    def create_task(self, coro, name=None):\n",
    "        task = Task(coro, name)\n",
    "        self._schedule_task(task)\n",
    "        return task\n",
    "\n",
    "    def _schedule_task(self, task, immediately=False):\n",
    "        if immediately:\n",
    "            self.tasks.insert(0, task)\n",
    "        else:\n",
    "            self.tasks.append(task)\n",
    "\n",
    "    def run_until_complete(self):\n",
    "        while len(self.tasks) != 0:\n",
    "            print('loop queue', self.tasks)\n",
    "            self.advance()\n",
    "            time.sleep(0.5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "bf13ccd5-db92-4f09-aa3d-e8e94035f9c4",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "class Loop(BaseLoop):\n",
    "    def advance(self):\n",
    "        task = self.tasks.pop(0)\n",
    "        try:\n",
    "            task._step()\n",
    "            self._schedule_task(task)\n",
    "        except StopIteration as result:\n",
    "            print(f'task {task!r} result: \"{result.value}\"')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "a0250ec0-63ff-4a4a-9aca-ba0af42ef3e0",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "async def awaitable_io(n):\n",
    "    await asleep(2)\n",
    "    return f'Hello world {n}'\n",
    "\n",
    "async def wait_and_print(n):\n",
    "    result = await awaitable_io(n)\n",
    "    print(f' -> received \"{result}\"')\n",
    "    return result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "b8f5b434-6e6a-45d0-bfb2-0c2134378727",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "loop queue [<task1>, <task2>]\n",
      "loop queue [<task2>, <task1>]\n",
      "loop queue [<task1>, <task2>]\n",
      "loop queue [<task2>, <task1>]\n",
      "loop queue [<task1>, <task2>]\n",
      " -> received \"Hello world 1\"\n",
      "task <task1> result: \"Hello world 1\"\n",
      "loop queue [<task2>]\n",
      " -> received \"Hello world 2\"\n",
      "task <task2> result: \"Hello world 2\"\n"
     ]
    }
   ],
   "source": [
    "l = Loop()\n",
    "l.create_task(wait_and_print(1), 'task1')\n",
    "l.create_task(wait_and_print(2), 'task2')\n",
    "l.run_until_complete()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "856caf91-5a85-4321-8e30-770bf8437b42",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "loop queue [<parent_task>]\n",
      "loop queue [<background_task>, <parent_task>]\n",
      "loop queue [<parent_task>, <background_task>]\n",
      "task <parent_task> result: \"None\"\n",
      "loop queue [<background_task>]\n",
      "loop queue [<background_task>]\n",
      "loop queue [<background_task>]\n",
      "finished background task\n",
      "task <background_task> result: \"None\"\n"
     ]
    }
   ],
   "source": [
    "async def background_task():\n",
    "    await asleep(2)\n",
    "    print('finished background task')\n",
    "\n",
    "async def parent_task():\n",
    "    task = l.create_task(background_task(), 'background_task')\n",
    "    await asleep(1)\n",
    "    # await task ?\n",
    "    # print('finished waiting for background tasks')\n",
    "\n",
    "l.create_task(parent_task(), 'parent_task')\n",
    "l.run_until_complete()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7c1c70af-107f-455c-a5db-033c5bca9797",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Futures"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "942723a5-0339-4905-a1e6-59d3a062e668",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "skip"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "from asyncio.futures import _PENDING, _CANCELLED, _FINISHED"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "6cb00118-6fb3-4513-aedb-e8382934dff6",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "class NaiveFuture:\n",
    "    _state = _PENDING\n",
    "    _result = None\n",
    "    \n",
    "    def __await__(self):\n",
    "        while not self.done():\n",
    "            yield self\n",
    "        return self._result\n",
    "\n",
    "    def done(self): return self._state != _PENDING\n",
    "\n",
    "    def set_result(self, result):\n",
    "        self._result = result\n",
    "        self._state = _FINISHED"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "610b8242-9692-437b-9833-0e2bde0b71ef",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "class Task(BaseTask, NaiveFuture):\n",
    "    def _step(self):\n",
    "        try:\n",
    "            return self._coro.send(None)\n",
    "        except StopIteration as result:\n",
    "            self.set_result(result)\n",
    "            raise result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "0241c519-c06a-48b8-aa5c-89fd53f405fa",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "loop queue [<parent_task>]\n",
      "loop queue [<background_task>, <parent_task>]\n",
      "loop queue [<parent_task>, <background_task>]\n",
      "waiting for subtask task\n",
      "loop queue [<background_task>, <parent_task>]\n",
      "loop queue [<parent_task>, <background_task>]\n",
      "loop queue [<background_task>, <parent_task>]\n",
      "task <background_task> result: \"None\"\n",
      "loop queue [<parent_task>]\n",
      "finishing parent task\n",
      "task <parent_task> result: \"None\"\n"
     ]
    }
   ],
   "source": [
    "l = Loop()\n",
    "async def background_task():\n",
    "    await asleep(2)\n",
    "\n",
    "async def parent_task():\n",
    "    task = l.create_task(background_task(), 'background_task')\n",
    "    await asleep(1)\n",
    "    print('waiting for subtask task')\n",
    "    await task\n",
    "    print('finishing parent task')\n",
    "\n",
    "l.create_task(parent_task(), 'parent_task')\n",
    "l.run_until_complete()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ff19b24f-6860-48e5-aa5f-8ad5e94904c2",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Actual futures are not coroutines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "e45e24cf-2b7e-4fad-b31d-f699b682b672",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "class Future(NaiveFuture):\n",
    "    _asyncio_future_blocking = False\n",
    "\n",
    "    def __init__(self, *args, **kwargs):\n",
    "        super().__init__(*args, *kwargs)\n",
    "        self._callbacks = []\n",
    "\n",
    "    def __await__(self):\n",
    "        yield self\n",
    "        if not self.done(): raise Exception(\"Future is not ready\")\n",
    "        return self._result\n",
    "\n",
    "    def set_result(self, result):\n",
    "        super().set_result(result)\n",
    "        print(f'result set for future {self}')\n",
    "        for c in self._callbacks: c(self)\n",
    "\n",
    "    def register_callback(self, cb):\n",
    "        self._callbacks.append(cb)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "b52073a2-cb8a-459b-b2d7-31d08dab3462",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "skip"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "class Task(Future, BaseTask):\n",
    "    def _step(self):\n",
    "        try:\n",
    "            return self._coro.send(None)\n",
    "        except StopIteration as result:\n",
    "            self.set_result(result)\n",
    "            raise result"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9e9c30dd-2018-4aa0-9f64-b3a65cca5054",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# New event loop"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "a09ccda5-1ef3-4f77-bb67-9793a0ee68a4",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "class Loop(BaseLoop):\n",
    "    def advance(self):\n",
    "        task = self.tasks.pop(0)\n",
    "        try:\n",
    "            step = task._step()\n",
    "            # both schedule later execution\n",
    "            if asyncio.isfuture(step):\n",
    "                step.register_callback(lambda _: self._schedule_task(task))\n",
    "            else:\n",
    "                self._schedule_task(task)\n",
    "        except StopIteration as result:\n",
    "            print(f'task {task!r} result: {result.value}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "b219028c-e5d2-4a02-be15-5dfb7e149bc1",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "loop queue [<parent_task>]\n",
      "loop queue [<background_task>, <parent_task>]\n",
      "loop queue [<parent_task>, <background_task>]\n",
      "waiting for subtask task\n",
      "loop queue [<background_task>]\n",
      "loop queue [<background_task>]\n",
      "loop queue [<background_task>]\n",
      "result set for future <background_task>\n",
      "task <background_task> result: None\n",
      "loop queue [<parent_task>]\n",
      "finishing parent task\n",
      "result set for future <parent_task>\n",
      "task <parent_task> result: 1\n"
     ]
    }
   ],
   "source": [
    "l = Loop()\n",
    "async def background_task():\n",
    "    await asleep(2)\n",
    "\n",
    "async def parent_task():\n",
    "    task = l.create_task(background_task(), 'background_task')\n",
    "    await asleep(1)\n",
    "    print('waiting for subtask task')\n",
    "    await task\n",
    "    print('finishing parent task')\n",
    "    return 1\n",
    "\n",
    "l.create_task(parent_task(), 'parent_task')\n",
    "l.run_until_complete()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cc4446b8-0508-4075-973c-072b06c4b68b",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Why you NEVER want sync code"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "a92bb4b0-ce88-44e4-b932-40774b8effed",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "loop queue [<parent_task>]\n",
      "loop queue [<task1>, <task2>, <parent_task>]\n",
      "[0:00:02.500738] background task finished\n",
      "result set for future <task1>\n",
      "task <task1> result: None\n",
      "loop queue [<task2>, <parent_task>]\n",
      "[0:00:05.001394] background task finished\n",
      "result set for future <task2>\n",
      "task <task2> result: None\n",
      "loop queue [<parent_task>]\n",
      "[0:00:05.502062] finished main task\n"
     ]
    }
   ],
   "source": [
    "t = Timer()\n",
    "\n",
    "async def background_task():\n",
    "    time.sleep(2)\n",
    "    t.log('background task finished')\n",
    "\n",
    "async def parent_task():\n",
    "    task1 = l.create_task(background_task(), 'task1')\n",
    "    task2 = l.create_task(background_task(), 'task2')\n",
    "    await asleep(1)\n",
    "    t.log('finished main task')\n",
    "    await task1\n",
    "    await task2\n",
    "\n",
    "l.create_task(parent_task(), 'parent_task')\n",
    "l.run_until_complete()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9bb01c08-727b-47af-9039-650e8c8335c8",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Final example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "9ed6e75c-215c-4bc9-8b96-d533d90706c2",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "loop queue [<parent_task>]\n",
      "loop queue [<task1>, <task2>, <parent_task>]\n",
      "loop queue [<task2>, <parent_task>, <task1>]\n",
      "loop queue [<parent_task>, <task1>, <task2>]\n",
      "finished main task\n",
      "loop queue [<task1>, <task2>]\n",
      "loop queue [<task2>, <task1>]\n",
      "loop queue [<task1>, <task2>]\n",
      "finished background task\n",
      "result set for future <task1>\n",
      "task <task1> result: None\n",
      "loop queue [<task2>, <parent_task>]\n",
      "finished background task\n",
      "result set for future <task2>\n",
      "task <task2> result: None\n",
      "loop queue [<parent_task>]\n"
     ]
    }
   ],
   "source": [
    "l = Loop()\n",
    "async def background_task():\n",
    "    await asleep(2)\n",
    "    print('finished background task')\n",
    "\n",
    "async def parent_task():\n",
    "    task1 = l.create_task(background_task(), 'task1')\n",
    "    task2 = l.create_task(background_task(), 'task2')\n",
    "\n",
    "    await asleep(1)\n",
    "    print('finished main task')\n",
    "    await task1\n",
    "    await task2\n",
    "    print('finished waiting for background tasks')\n",
    "\n",
    "l.create_task(parent_task(), 'parent_task')\n",
    "l.run_until_complete()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3e92a82c-df0a-4b69-b668-258899c2a154",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# asyncio + async / await:\n",
    "\n",
    "- generators based coroutines\n",
    "- syntax to make talking to coroutines easier\n",
    "- run every \"waiting\" taks elsewhere and wait on future   \n",
    "- event loop to run coroutines\n",
    "- Futures = unscheduled until callback\n",
    "  - (future is not scheduled, not the underlying task)\n",
    "- non blocking io with setblocking(False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c6c47943-fe18-457e-a164-f7b8053129c9",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "\n",
    "# references:\n",
    "\n",
    "- https://github.com/python/cpython/tree/main/Lib/asyncio\n",
    "- https://docs.python.org/3/library/asyncio-task.html\n",
    "- https://docs.python.org/3/library/socket.html\n",
    "- https://bbc.github.io/cloudfit-public-docs/asyncio/asyncio-part-2\n",
    "- https://medium.com/vaidikkapoor/understanding-non-blocking-i-o-with-python-part-1-ec31a2e2db9b\n",
    "- https://stackoverflow.com/questions/49005651/how-does-asyncio-actually-work"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.12.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}