05-event-loop-futures.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Futures"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A future is some object that represents an asynchronous computation. Like how a synchronous function returns a result, an asynchronous function may return a future. Rather than the function blocking when called, it can return the future immediately.\n",
    "\n",
    "```python\n",
    "future = asynchronous_function_call()\n",
    "print('Started async computation')  # runs immediately after function call\n",
    "```\n",
    "\n",
    "The future can then be passed around the code like a normal object, until a piece of code tries blocking on the result.\n",
    "\n",
    "```python\n",
    "real_value = future.get_result()  # Returns immediately if future is already done, otherwise blocks.\n",
    "```\n",
    "\n",
    "Often, it is possible to derive a new future from an existing future, again without having to immediately block. Instead, a callback determines how the new future derives its value from the old future.\n",
    "\n",
    "```python\n",
    "new_future = future.map_result(lambda real_value: (2 * real_value))\n",
    "```\n",
    "\n",
    "Like an event loop task, a future can have callbacks that get called when the result is available.\n",
    "\n",
    "```python\n",
    "future.register_callback(lambda real_value: print(\"Future completed with result\", real_value))\n",
    "```\n",
    "\n",
    "Finally, the code that owns the result of the callback can set the result whenever the computation is done.\n",
    "\n",
    "```python\n",
    "future.set_result(10)\n",
    "```\n",
    "\n",
    "In a threaded program, `asynchronous_function_call()` might be a function that sends a computation to another thread, and creates a future object that is shared by both threads. The future contains a `threading.Condition` that gets set when the result is set, and calling `future.get_result()` waits on that `Condition` to be set. For examples of this kind of future, see `concurrent.futures` and <https://pypi.python.org/pypi/aplus>."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Event Loop Futures"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Event loop futures are very similar to threading futures. The main difference is that the asynchronous computation is happening on the same thread, in the same event loop.\n",
    "\n",
    "This means that `get_result()` can be a little smarter. Instead of putting the thread to sleep (which would be disastrous for the event loop), the task gets unscheduled from the event loop. At the same time, a new callback is registered with the future. This callback will reschedule the task with the event loop. Thus, when the future completes, the callback will be called, the task will be rescheduled, and eventually the task will be resumed, and will be able to use the result of the future.\n",
    "\n",
    "And instead of the future being completed by another thread, it will be completed by another task on the event loop."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# `asyncio.Future`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So far we've been discussing event loops in general terms, and deriving them from first principles. At this point, it makes sense to start focusing on the particulars of the `asyncio` interface. This certainly isn't the only way to do event loop futures, but it is the way that was chosen for the standard library implementation.\n",
    "\n",
    "When a task calls `get_result()`, it wants to suspend itself if necessary, but it eventually wants the function to return with a value. What do we know of that can do that? That's right, a generator! And what kind of expression can extract the return value of a generator? `yield from`!\n",
    "\n",
    "Thus, `asyncio.Future` went with the approach that a task can wait for a future to be complete by doing\n",
    "\n",
    "```python\n",
    "real_value = yield from future\n",
    "```\n",
    "\n",
    "To do this, `asyncio.Future` needs to be an iterable, which yields to the event loop, but in the end returns the final result. Yielding needs to somehow signal to the event loop how to set up a relationship between the future and this task. It does so by yielding itself. The event loop then registers a callback on the future, such that when the future is completed, the task will be rescheduled with the event loop. Finally, the future is written such that when it is finished, all of its callbacks will be scheduled with the event loop."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting good task\n",
      "<Future done=False>\n",
      "Waiting on future\n",
      "Add done callback <TaskWakeupMethWrapper object at 0x10a8b93a8> to <Future done=False cb=[<TaskWakeupMethWrapper object at 0x10a8b93a8>()]>\n",
      "Set result 10 for <Future done=True result=10 cb=[<TaskWakeupMethWrapper object at 0x10a8b93a8>()]>\n",
      "Scheduling callback <TaskWakeupMethWrapper object at 0x10a8b93a8> for <Future done=True result=10 cb=[<TaskWakeupMethWrapper object at 0x10a8b93a8>()]>\n",
      "Task resumed from waiting on <generator object Future.__iter__ at 0x10a8aea40>\n",
      "Result: 10\n",
      "Didn't ever yield\n",
      "Result: None\n",
      "Yielding\n",
      "Resumed\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Traceback (most recent call last):\n",
      "  File \"<ipython-input-1-2819b86a00b7>\", line 114, in bad_task\n",
      "    result = (yield future)\n",
      "RuntimeError: yield was used instead of yield from in task <Task pending coro=<bad_task() running at <ipython-input-1-2819b86a00b7>:114>> with <Future done=False>\n"
     ]
    }
   ],
   "source": [
    "import asyncio\n",
    "from asyncio import base_futures, test_utils\n",
    "import traceback\n",
    "\n",
    "# Code sample from <https://github.com/python/cpython/tree/v3.6.1/Lib/asyncio>  \n",
    "# Copyright (c) 2001-2017 Python Software Foundation.  \n",
    "# All Rights Reserved.  \n",
    "# License: Python license, <https://www.python.org/3.6/license.html>  \n",
    "# Some modifications made so as to only highlight the interesting parts.\n",
    "class Future:\n",
    "    _done = False\n",
    "    _result = None\n",
    "    _loop = None\n",
    "    \n",
    "    # This field is used for a dual purpose:\n",
    "    # - Its presence is a marker to declare that a class implements\n",
    "    #   the Future protocol (i.e. is intended to be duck-type compatible).\n",
    "    #   The value must also be not-None, to enable a subclass to declare\n",
    "    #   that it is not compatible by setting this to None.\n",
    "    # - It is set by __iter__() below so that Task._step() can tell\n",
    "    #   the difference between `yield from Future()` (correct) vs.\n",
    "    #   `yield Future()` (incorrect).\n",
    "    _asyncio_future_blocking = False\n",
    "\n",
    "    def __init__(self, *, loop):\n",
    "        super().__init__()\n",
    "        self._loop = loop\n",
    "        self._callbacks = []\n",
    "        \n",
    "    def __repr__(self):\n",
    "        info = [f\"done={self._done!r}\"]\n",
    "        if self._done:\n",
    "            info.append('result={}'.format(self._result))\n",
    "        if self._callbacks:\n",
    "            info.append(base_futures._format_callbacks(self._callbacks))\n",
    "        return '<%s %s>' % (self.__class__.__name__, ' '.join(info))\n",
    "    \n",
    "    def _schedule_callbacks(self):\n",
    "        future_repr = repr(self)\n",
    "        callbacks = self._callbacks[:]\n",
    "        if not callbacks:\n",
    "            return\n",
    "\n",
    "        self._callbacks[:] = []\n",
    "        for callback in callbacks:\n",
    "            print('Scheduling callback', callback, 'for', future_repr)\n",
    "            self._loop.call_soon(callback, self)\n",
    "            \n",
    "    def done(self):\n",
    "        return self._done\n",
    "    \n",
    "    def result(self):\n",
    "        if not self._done:\n",
    "            raise base_futures.InvalidStateError('Result is not ready.')\n",
    "        return self._result\n",
    "    \n",
    "    def add_done_callback(self, fn):\n",
    "        if self._done:\n",
    "            print('Add done callback', fn, 'but', self, 'is already done, so calling soon')\n",
    "            self._loop.call_soon(fn, self)\n",
    "        else:\n",
    "            self._callbacks.append(fn)\n",
    "            print('Add done callback', fn, 'to', self)\n",
    "\n",
    "    def remove_done_callback(self, fn):\n",
    "        filtered_callbacks = [f for f in self._callbacks if f != fn]\n",
    "        removed_count = len(self._callbacks) - len(filtered_callbacks)\n",
    "        if removed_count:\n",
    "            self._callbacks[:] = filtered_callbacks\n",
    "        return removed_count\n",
    "    \n",
    "    def set_result(self, result):\n",
    "        if self._done:\n",
    "            raise base_futures.InvalidStateError('{!r}'.format(self))\n",
    "        self._result = result\n",
    "        self._done = True\n",
    "        print('Set result', result, 'for', self)\n",
    "        self._schedule_callbacks()\n",
    "        \n",
    "    def __iter__(self):\n",
    "        if not self.done():\n",
    "            self._asyncio_future_blocking = True\n",
    "            yield self  # This tells Task to wait for completion.\n",
    "        assert self.done(), \"yield from wasn't used with future\"\n",
    "        return self.result()\n",
    "\n",
    "\n",
    "loop = asyncio.new_event_loop()\n",
    "    \n",
    "def good_task(*, loop):\n",
    "    print('Starting good task')\n",
    "    future = Future(loop=loop)\n",
    "    print(future)\n",
    "    loop.call_soon(future.set_result, 10)\n",
    "    print('Waiting on future')\n",
    "    future = iter(future)\n",
    "    result = (yield from future)\n",
    "    print(\"Task resumed from waiting on\", future)\n",
    "    print(\"Result:\", result)\n",
    "    result2 = (yield from future)\n",
    "    print(\"Didn't ever yield\")\n",
    "    print(\"Result:\", result2)\n",
    "    print(\"Yielding\")\n",
    "    yield\n",
    "    print(\"Resumed\")\n",
    "    \n",
    "asyncio.ensure_future(good_task(loop=loop), loop=loop)\n",
    "for _ in range(100):\n",
    "    test_utils.run_once(loop=loop)\n",
    "    \n",
    "def bad_task(*, loop):\n",
    "    future = Future(loop=loop)\n",
    "    try:\n",
    "        result = (yield future)\n",
    "    except RuntimeError:\n",
    "        traceback.print_exc()\n",
    "\n",
    "print()\n",
    "asyncio.ensure_future(bad_task(loop=loop), loop=loop)\n",
    "for _ in range(100):\n",
    "    test_utils.run_once(loop=loop)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "This snippet shows the basic mechanics of `asyncio.Future`. For simplicity, I did leave out some other functionality it has:\n",
    "- A future can also be completed with the `set_exception()` method, which will cause that exception to be raised inside the task when `result()` is called.\n",
    "- A future can be cancelled. This causes any tasks waiting on the future to also be cancelled.\n",
    "- Callbacks can be un-registered, if the event loop or another task determines that the callback is no longer needed.\n",
    "\n",
    "Notice that `asyncio.Future` is not protected by any threading locks. This is one of the advantages of using an event loop instead of using threads. Since everything is running on the same thread, you don't get any surprise context switches. Every evaluation step is guaranteed to execute immediately after the previous evaulation step, unless the previous step is a `yield` or `yield from`. This drastically reduces the number of of ways in which your program to execute, and makes it much easier to reason about."
   ]
  }
 ],
 "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.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}