From db2fe1d1e1d0f7352167a2ff00c4296115810adc Mon Sep 17 00:00:00 2001
From: Yury Selivanov <yury@magic.io>
Date: Thu, 10 Nov 2016 18:54:03 -0500
Subject: [PATCH 1/8] Add asyncio.run() and asyncio.run_in_executor()
 functions.

---
 asyncio/__init__.py    |  1 +
 asyncio/base_events.py | 23 +++++++++++++-
 asyncio/run.py         | 68 ++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 91 insertions(+), 1 deletion(-)
 create mode 100644 asyncio/run.py

diff --git a/asyncio/__init__.py b/asyncio/__init__.py
index 011466b3..51f24be3 100644
--- a/asyncio/__init__.py
+++ b/asyncio/__init__.py
@@ -24,6 +24,7 @@
 from .futures import *
 from .locks import *
 from .protocols import *
+from .run import *
 from .queues import *
 from .streams import *
 from .subprocess import *
diff --git a/asyncio/base_events.py b/asyncio/base_events.py
index aa783671..c930a2fa 100644
--- a/asyncio/base_events.py
+++ b/asyncio/base_events.py
@@ -38,7 +38,7 @@
 from .log import logger
 
 
-__all__ = ['BaseEventLoop']
+__all__ = ['BaseEventLoop', 'run_in_executor']
 
 
 # Minimum number of _scheduled timer handles before cleanup of
@@ -55,6 +55,27 @@
                        ConnectionResetError, ConnectionAbortedError)
 
 
+@coroutine
+def run_in_executor(executor, func, *args):
+    """Run the function in a thread or a process pool.
+
+    Any concurrent.futures executor can be passed to the
+    function.
+
+    If executor is None, the function will be run in
+    the default threadpool of the event loop.
+
+    Example:
+
+        async def coro():
+            await asyncio.run_in_executor(
+                None, long_calculation, 42)
+    """
+    loop = events.get_event_loop()
+    result = yield from loop.run_in_executor(executor, func, *args)
+    return result
+
+
 def _format_handle(handle):
     cb = handle._callback
     if inspect.ismethod(cb) and isinstance(cb.__self__, tasks.Task):
diff --git a/asyncio/run.py b/asyncio/run.py
new file mode 100644
index 00000000..4f556a32
--- /dev/null
+++ b/asyncio/run.py
@@ -0,0 +1,68 @@
+"""asyncio.run() function."""
+
+__all__ = ['run']
+
+import threading
+
+from . import coroutines
+from . import events
+
+
+def run(coro, *, debug=False):
+    """Run a coroutine.
+
+    This function runs the passed coroutine, taking care of
+    managing the asyncio event loop and finalizing asynchronous
+    generators.
+
+    This function must be called from the main thread, and it
+    cannot be called when another asyncio event loop is running.
+
+    If debug is True, the event loop will be run in debug mode.
+
+    This function should be used as a main entry point for
+    asyncio programs, and should not be used to call asynchronous
+    APIs.
+
+    Example::
+
+        import asyncio
+
+        async def main():
+            await asyncio.sleep(1)
+            print('hello')
+
+        asyncio.run(main())
+    """
+    if events._get_running_loop() is not None:
+        raise RuntimeError(
+            "asyncio.run() cannot be called from a running event loop")
+    if not isinstance(threading.current_thread(), threading._MainThread):
+        raise RuntimeError(
+            "asyncio.run() must be called from the main thread")
+    if not coroutines.iscoroutine(coro):
+        raise ValueError("a coroutine was expected, got {!r}".format(coro))
+
+    loop = events.new_event_loop()
+    try:
+        events.set_event_loop(loop)
+
+        if debug:
+            loop.set_debug(True)
+
+        result = loop.run_until_complete(coro)
+
+        try:
+            # `shutdown_asyncgens` was added in Python 3.6; not all
+            # event loops might support it.
+            shutdown_asyncgens = loop.shutdown_asyncgens
+        except AttributeError:
+            pass
+        else:
+            loop.run_until_complete(shutdown_asyncgens())
+
+        return result
+
+    finally:
+        events.set_event_loop(None)
+        loop.close()

From 9acdcebe7f81acbecd86ed1d2247cdb4628a7455 Mon Sep 17 00:00:00 2001
From: Yury Selivanov <yury@magic.io>
Date: Mon, 14 Nov 2016 19:26:51 -0500
Subject: [PATCH 2/8] Add asyncio.forever()

---
 asyncio/__init__.py    |  3 ++-
 asyncio/base_events.py | 14 ++++++++++++++
 asyncio/events.py      |  3 +++
 asyncio/run.py         | 39 +++++++++++++++++++++++++++++++++++++--
 asyncio/tasks.py       |  2 +-
 5 files changed, 57 insertions(+), 4 deletions(-)

diff --git a/asyncio/__init__.py b/asyncio/__init__.py
index 51f24be3..846afc7b 100644
--- a/asyncio/__init__.py
+++ b/asyncio/__init__.py
@@ -41,7 +41,8 @@
            streams.__all__ +
            subprocess.__all__ +
            tasks.__all__ +
-           transports.__all__)
+           transports.__all__ +
+           ['run', 'forever'])  # Will fix this later.
 
 if sys.platform == 'win32':  # pragma: no cover
     from .windows_events import *
diff --git a/asyncio/base_events.py b/asyncio/base_events.py
index c930a2fa..a324aad5 100644
--- a/asyncio/base_events.py
+++ b/asyncio/base_events.py
@@ -294,6 +294,9 @@ def __init__(self):
         # Set to True when `loop.shutdown_asyncgens` is called.
         self._asyncgens_shutdown_called = False
 
+        # Future that isn't resolved while the loop is running.
+        self._forever_fut = None
+
     def __repr__(self):
         return ('<%s running=%s closed=%s debug=%s>'
                 % (self.__class__.__name__, self.is_running(),
@@ -430,8 +433,12 @@ def shutdown_asyncgens(self):
                     'asyncgen': agen
                 })
 
+    def get_forever_future(self):
+        return self._forever_fut
+
     def run_forever(self):
         """Run until stop() is called."""
+        self._forever_fut = self.create_future()
         self._check_closed()
         if self.is_running():
             raise RuntimeError('This event loop is already running')
@@ -450,7 +457,14 @@ def run_forever(self):
                 self._run_once()
                 if self._stopping:
                     break
+        except BaseException as ex:
+            self._forever_fut.set_exception(ex)
+            self._forever_fut._log_traceback = False
+            raise ex
+        else:
+            self._forever_fut.set_result(None)
         finally:
+            self._forever_fut = None
             self._stopping = False
             self._thread_id = None
             events._set_running_loop(None)
diff --git a/asyncio/events.py b/asyncio/events.py
index 28a45fc3..d4bee48b 100644
--- a/asyncio/events.py
+++ b/asyncio/events.py
@@ -512,6 +512,9 @@ def get_debug(self):
     def set_debug(self, enabled):
         raise NotImplementedError
 
+    def get_forever_future(self):
+        raise NotImplementedError
+
 
 class AbstractEventLoopPolicy:
     """Abstract policy for accessing the event loop."""
diff --git a/asyncio/run.py b/asyncio/run.py
index 4f556a32..44ae4ab3 100644
--- a/asyncio/run.py
+++ b/asyncio/run.py
@@ -1,6 +1,6 @@
 """asyncio.run() function."""
 
-__all__ = ['run']
+__all__ = ['run', 'forever']
 
 import threading
 
@@ -8,6 +8,33 @@
 from . import events
 
 
+@coroutines.coroutine
+def forever():
+    """Wait until the current event loop stops running.
+
+    The coroutine will return None if the loop is stopped by
+    calling the `loop.stop()` method.
+
+    The coroutine will propagate any exception that caused
+    the loop to stop;
+
+    It is recommended to use this coroutine with the asyncio.run()
+    function:
+
+        async def coro():
+            print('hi')
+            try:
+                await asyncio.forever()
+            except KeyboardInterrupt:
+                await asyncio.sleep(1)
+                print('bye')
+
+        asyncio.run(coro())
+    """
+    loop = events.get_event_loop()
+    return (yield from loop.get_forever_future())
+
+
 def run(coro, *, debug=False):
     """Run a coroutine.
 
@@ -50,7 +77,15 @@ async def main():
         if debug:
             loop.set_debug(True)
 
-        result = loop.run_until_complete(coro)
+        task = loop.create_task(coro)
+        task.add_done_callback(lambda task: loop.stop())
+
+        try:
+            loop.run_forever()
+        except BaseException as ex:
+            result = loop.run_until_complete(task)
+        else:
+            result = task.result()
 
         try:
             # `shutdown_asyncgens` was added in Python 3.6; not all
diff --git a/asyncio/tasks.py b/asyncio/tasks.py
index 8852aa5a..af43ba2d 100644
--- a/asyncio/tasks.py
+++ b/asyncio/tasks.py
@@ -302,7 +302,7 @@ def _step(self, exc=None):
     def _wakeup(self, future):
         try:
             future.result()
-        except Exception as exc:
+        except BaseException as exc:
             # This may also be a cancellation.
             self._step(exc)
         else:

From f24ff3056ce9506390594d94fd627b77964b97e0 Mon Sep 17 00:00:00 2001
From: Yury Selivanov <yury@magic.io>
Date: Tue, 15 Nov 2016 16:39:52 -0500
Subject: [PATCH 3/8] try async generators

---
 asyncio/base_events.py | 14 --------------
 asyncio/events.py      |  3 ---
 asyncio/run.py         | 38 +++++++++++++++++++++++++++++---------
 3 files changed, 29 insertions(+), 26 deletions(-)

diff --git a/asyncio/base_events.py b/asyncio/base_events.py
index a324aad5..c930a2fa 100644
--- a/asyncio/base_events.py
+++ b/asyncio/base_events.py
@@ -294,9 +294,6 @@ def __init__(self):
         # Set to True when `loop.shutdown_asyncgens` is called.
         self._asyncgens_shutdown_called = False
 
-        # Future that isn't resolved while the loop is running.
-        self._forever_fut = None
-
     def __repr__(self):
         return ('<%s running=%s closed=%s debug=%s>'
                 % (self.__class__.__name__, self.is_running(),
@@ -433,12 +430,8 @@ def shutdown_asyncgens(self):
                     'asyncgen': agen
                 })
 
-    def get_forever_future(self):
-        return self._forever_fut
-
     def run_forever(self):
         """Run until stop() is called."""
-        self._forever_fut = self.create_future()
         self._check_closed()
         if self.is_running():
             raise RuntimeError('This event loop is already running')
@@ -457,14 +450,7 @@ def run_forever(self):
                 self._run_once()
                 if self._stopping:
                     break
-        except BaseException as ex:
-            self._forever_fut.set_exception(ex)
-            self._forever_fut._log_traceback = False
-            raise ex
-        else:
-            self._forever_fut.set_result(None)
         finally:
-            self._forever_fut = None
             self._stopping = False
             self._thread_id = None
             events._set_running_loop(None)
diff --git a/asyncio/events.py b/asyncio/events.py
index d4bee48b..28a45fc3 100644
--- a/asyncio/events.py
+++ b/asyncio/events.py
@@ -512,9 +512,6 @@ def get_debug(self):
     def set_debug(self, enabled):
         raise NotImplementedError
 
-    def get_forever_future(self):
-        raise NotImplementedError
-
 
 class AbstractEventLoopPolicy:
     """Abstract policy for accessing the event loop."""
diff --git a/asyncio/run.py b/asyncio/run.py
index 44ae4ab3..0db2a589 100644
--- a/asyncio/run.py
+++ b/asyncio/run.py
@@ -2,12 +2,19 @@
 
 __all__ = ['run', 'forever']
 
+import inspect
 import threading
 
 from . import coroutines
 from . import events
 
 
+def _isasyncgen(obj):
+    if hasattr(inspect, 'isasyncgen'):
+        return inspect.isasyncgen(obj)
+    return False
+
+
 @coroutines.coroutine
 def forever():
     """Wait until the current event loop stops running.
@@ -67,8 +74,10 @@ async def main():
     if not isinstance(threading.current_thread(), threading._MainThread):
         raise RuntimeError(
             "asyncio.run() must be called from the main thread")
-    if not coroutines.iscoroutine(coro):
-        raise ValueError("a coroutine was expected, got {!r}".format(coro))
+    if not coroutines.iscoroutine(coro) and not _isasyncgen(coro):
+        raise ValueError(
+            "a coroutine or an asynchronous generator was expected, "
+            "got {!r}".format(coro))
 
     loop = events.new_event_loop()
     try:
@@ -77,15 +86,26 @@ async def main():
         if debug:
             loop.set_debug(True)
 
-        task = loop.create_task(coro)
-        task.add_done_callback(lambda task: loop.stop())
+        if _isasyncgen(coro):
+            result = None
+            loop.run_until_complete(coro.asend(None))
+            try:
+                loop.run_forever()
+            except BaseException as ex:
+                try:
+                    loop.run_until_complete(coro.athrow(ex))
+                except StopAsyncIteration as ex:
+                    if ex.args:
+                        result = ex.args[0]
+            else:
+                try:
+                    loop.run_until_complete(coro.asend(None))
+                except StopAsyncIteration as ex:
+                    if ex.args:
+                        result = ex.args[0]
 
-        try:
-            loop.run_forever()
-        except BaseException as ex:
-            result = loop.run_until_complete(task)
         else:
-            result = task.result()
+            result = loop.run_until_complete(coro)
 
         try:
             # `shutdown_asyncgens` was added in Python 3.6; not all

From fa721ee80e17bcbdde8f5f5431538300436b2301 Mon Sep 17 00:00:00 2001
From: Yury Selivanov <yury@magic.io>
Date: Tue, 15 Nov 2016 17:26:52 -0500
Subject: [PATCH 4/8] Rollback all changes to tasks.py; drop forever

---
 asyncio/__init__.py |  2 +-
 asyncio/run.py      | 29 +----------------------------
 asyncio/tasks.py    |  2 +-
 3 files changed, 3 insertions(+), 30 deletions(-)

diff --git a/asyncio/__init__.py b/asyncio/__init__.py
index 846afc7b..3434cb05 100644
--- a/asyncio/__init__.py
+++ b/asyncio/__init__.py
@@ -42,7 +42,7 @@
            subprocess.__all__ +
            tasks.__all__ +
            transports.__all__ +
-           ['run', 'forever'])  # Will fix this later.
+           ['run'])  # Will fix this later.
 
 if sys.platform == 'win32':  # pragma: no cover
     from .windows_events import *
diff --git a/asyncio/run.py b/asyncio/run.py
index 0db2a589..f75a567f 100644
--- a/asyncio/run.py
+++ b/asyncio/run.py
@@ -1,6 +1,6 @@
 """asyncio.run() function."""
 
-__all__ = ['run', 'forever']
+__all__ = ['run']
 
 import inspect
 import threading
@@ -15,33 +15,6 @@ def _isasyncgen(obj):
     return False
 
 
-@coroutines.coroutine
-def forever():
-    """Wait until the current event loop stops running.
-
-    The coroutine will return None if the loop is stopped by
-    calling the `loop.stop()` method.
-
-    The coroutine will propagate any exception that caused
-    the loop to stop;
-
-    It is recommended to use this coroutine with the asyncio.run()
-    function:
-
-        async def coro():
-            print('hi')
-            try:
-                await asyncio.forever()
-            except KeyboardInterrupt:
-                await asyncio.sleep(1)
-                print('bye')
-
-        asyncio.run(coro())
-    """
-    loop = events.get_event_loop()
-    return (yield from loop.get_forever_future())
-
-
 def run(coro, *, debug=False):
     """Run a coroutine.
 
diff --git a/asyncio/tasks.py b/asyncio/tasks.py
index af43ba2d..8852aa5a 100644
--- a/asyncio/tasks.py
+++ b/asyncio/tasks.py
@@ -302,7 +302,7 @@ def _step(self, exc=None):
     def _wakeup(self, future):
         try:
             future.result()
-        except BaseException as exc:
+        except Exception as exc:
             # This may also be a cancellation.
             self._step(exc)
         else:

From b8b0fa0e1aede796c827dd0d4764eab0e34848a7 Mon Sep 17 00:00:00 2001
From: Yury Selivanov <yury@magic.io>
Date: Tue, 15 Nov 2016 17:28:50 -0500
Subject: [PATCH 5/8] Drop run_in_executor

---
 asyncio/base_events.py | 23 +----------------------
 1 file changed, 1 insertion(+), 22 deletions(-)

diff --git a/asyncio/base_events.py b/asyncio/base_events.py
index c930a2fa..aa783671 100644
--- a/asyncio/base_events.py
+++ b/asyncio/base_events.py
@@ -38,7 +38,7 @@
 from .log import logger
 
 
-__all__ = ['BaseEventLoop', 'run_in_executor']
+__all__ = ['BaseEventLoop']
 
 
 # Minimum number of _scheduled timer handles before cleanup of
@@ -55,27 +55,6 @@
                        ConnectionResetError, ConnectionAbortedError)
 
 
-@coroutine
-def run_in_executor(executor, func, *args):
-    """Run the function in a thread or a process pool.
-
-    Any concurrent.futures executor can be passed to the
-    function.
-
-    If executor is None, the function will be run in
-    the default threadpool of the event loop.
-
-    Example:
-
-        async def coro():
-            await asyncio.run_in_executor(
-                None, long_calculation, 42)
-    """
-    loop = events.get_event_loop()
-    result = yield from loop.run_in_executor(executor, func, *args)
-    return result
-
-
 def _format_handle(handle):
     cb = handle._callback
     if inspect.ismethod(cb) and isinstance(cb.__self__, tasks.Task):

From 3c903649af196084eb20854b8bc5c10302c5f59d Mon Sep 17 00:00:00 2001
From: Yury Selivanov <yury@magic.io>
Date: Wed, 16 Nov 2016 18:50:48 -0500
Subject: [PATCH 6/8] Add asyncio.run_forever(); add tests.

---
 asyncio/__init__.py  |   6 +-
 asyncio/run.py       |  96 --------------------
 asyncio/runners.py   | 146 ++++++++++++++++++++++++++++++
 runtests.py          |   4 +
 tests/test_runner.py | 205 +++++++++++++++++++++++++++++++++++++++++++
 5 files changed, 358 insertions(+), 99 deletions(-)
 delete mode 100644 asyncio/run.py
 create mode 100644 asyncio/runners.py
 create mode 100644 tests/test_runner.py

diff --git a/asyncio/__init__.py b/asyncio/__init__.py
index 3434cb05..30cfbdce 100644
--- a/asyncio/__init__.py
+++ b/asyncio/__init__.py
@@ -24,7 +24,7 @@
 from .futures import *
 from .locks import *
 from .protocols import *
-from .run import *
+from .runners import *
 from .queues import *
 from .streams import *
 from .subprocess import *
@@ -37,12 +37,12 @@
            futures.__all__ +
            locks.__all__ +
            protocols.__all__ +
+           runners.__all__ +
            queues.__all__ +
            streams.__all__ +
            subprocess.__all__ +
            tasks.__all__ +
-           transports.__all__ +
-           ['run'])  # Will fix this later.
+           transports.__all__)
 
 if sys.platform == 'win32':  # pragma: no cover
     from .windows_events import *
diff --git a/asyncio/run.py b/asyncio/run.py
deleted file mode 100644
index f75a567f..00000000
--- a/asyncio/run.py
+++ /dev/null
@@ -1,96 +0,0 @@
-"""asyncio.run() function."""
-
-__all__ = ['run']
-
-import inspect
-import threading
-
-from . import coroutines
-from . import events
-
-
-def _isasyncgen(obj):
-    if hasattr(inspect, 'isasyncgen'):
-        return inspect.isasyncgen(obj)
-    return False
-
-
-def run(coro, *, debug=False):
-    """Run a coroutine.
-
-    This function runs the passed coroutine, taking care of
-    managing the asyncio event loop and finalizing asynchronous
-    generators.
-
-    This function must be called from the main thread, and it
-    cannot be called when another asyncio event loop is running.
-
-    If debug is True, the event loop will be run in debug mode.
-
-    This function should be used as a main entry point for
-    asyncio programs, and should not be used to call asynchronous
-    APIs.
-
-    Example::
-
-        import asyncio
-
-        async def main():
-            await asyncio.sleep(1)
-            print('hello')
-
-        asyncio.run(main())
-    """
-    if events._get_running_loop() is not None:
-        raise RuntimeError(
-            "asyncio.run() cannot be called from a running event loop")
-    if not isinstance(threading.current_thread(), threading._MainThread):
-        raise RuntimeError(
-            "asyncio.run() must be called from the main thread")
-    if not coroutines.iscoroutine(coro) and not _isasyncgen(coro):
-        raise ValueError(
-            "a coroutine or an asynchronous generator was expected, "
-            "got {!r}".format(coro))
-
-    loop = events.new_event_loop()
-    try:
-        events.set_event_loop(loop)
-
-        if debug:
-            loop.set_debug(True)
-
-        if _isasyncgen(coro):
-            result = None
-            loop.run_until_complete(coro.asend(None))
-            try:
-                loop.run_forever()
-            except BaseException as ex:
-                try:
-                    loop.run_until_complete(coro.athrow(ex))
-                except StopAsyncIteration as ex:
-                    if ex.args:
-                        result = ex.args[0]
-            else:
-                try:
-                    loop.run_until_complete(coro.asend(None))
-                except StopAsyncIteration as ex:
-                    if ex.args:
-                        result = ex.args[0]
-
-        else:
-            result = loop.run_until_complete(coro)
-
-        try:
-            # `shutdown_asyncgens` was added in Python 3.6; not all
-            # event loops might support it.
-            shutdown_asyncgens = loop.shutdown_asyncgens
-        except AttributeError:
-            pass
-        else:
-            loop.run_until_complete(shutdown_asyncgens())
-
-        return result
-
-    finally:
-        events.set_event_loop(None)
-        loop.close()
diff --git a/asyncio/runners.py b/asyncio/runners.py
new file mode 100644
index 00000000..e8fa22dd
--- /dev/null
+++ b/asyncio/runners.py
@@ -0,0 +1,146 @@
+"""asyncio.run() and asyncio.run_forever() functions."""
+
+__all__ = ['run', 'run_forever']
+
+import inspect
+import threading
+
+from . import coroutines
+from . import events
+
+
+def _cleanup(loop):
+    try:
+        # `shutdown_asyncgens` was added in Python 3.6; not all
+        # event loops might support it.
+        shutdown_asyncgens = loop.shutdown_asyncgens
+    except AttributeError:
+        pass
+    else:
+        loop.run_until_complete(shutdown_asyncgens())
+    finally:
+        events.set_event_loop(None)
+        loop.close()
+
+
+def run(main, *, debug=False):
+    """Run a coroutine.
+
+    This function runs the passed coroutine, taking care of
+    managing the asyncio event loop and finalizing asynchronous
+    generators.
+
+    This function must be called from the main thread, and it
+    cannot be called when another asyncio event loop is running.
+
+    If debug is True, the event loop will be run in debug mode.
+
+    This function should be used as a main entry point for
+    asyncio programs, and should not be used to call asynchronous
+    APIs.
+
+    Example::
+
+        async def main():
+            await asyncio.sleep(1)
+            print('hello')
+
+        asyncio.run(main())
+    """
+    if events._get_running_loop() is not None:
+        raise RuntimeError(
+            "asyncio.run() cannot be called from a running event loop")
+    if not isinstance(threading.current_thread(), threading._MainThread):
+        raise RuntimeError(
+            "asyncio.run() must be called from the main thread")
+    if not coroutines.iscoroutine(main):
+        raise ValueError("a coroutine was expected, got {!r}".format(main))
+
+    loop = events.new_event_loop()
+    try:
+        events.set_event_loop(loop)
+
+        if debug:
+            loop.set_debug(True)
+
+        return loop.run_until_complete(main)
+    finally:
+        _cleanup(loop)
+
+
+def run_forever(main, *, debug=False):
+    """Run asyncio loop.
+
+    main must be an asynchronous generator with one yield, separating
+    program initialization from cleanup logic.
+
+    If debug is True, the event loop will be run in debug mode.
+
+    This function should be used as a main entry point for
+    asyncio programs, and should not be used to call asynchronous
+    APIs.
+
+    Example:
+
+        async def main():
+            server = await asyncio.start_server(...)
+            try:
+                yield  # <- Let event loop run forever.
+            except KeyboardInterrupt:
+                print('^C received; exiting.')
+            finally:
+                server.close()
+                await server.wait_closed()
+
+        asyncio.run_forever(main())
+    """
+    if not hasattr(inspect, 'isasyncgen'):
+        raise NotImplementedError
+
+    if events._get_running_loop() is not None:
+        raise RuntimeError(
+            "asyncio.run_forever() cannot be called from a running event loop")
+    if not isinstance(threading.current_thread(), threading._MainThread):
+        raise RuntimeError(
+            "asyncio.run() must be called from the main thread")
+    if not inspect.isasyncgen(main):
+        raise ValueError(
+            "an asynchronous generator was expected, got {!r}".format(main))
+
+    loop = events.new_event_loop()
+    try:
+        events.set_event_loop(loop)
+        if debug:
+            loop.set_debug(True)
+
+        ret = None
+        try:
+            ret = loop.run_until_complete(main.asend(None))
+        except StopAsyncIteration as ex:
+            return
+        if ret is not None:
+            raise RuntimeError("only empty yield is supported")
+
+        yielded_twice = False
+        try:
+            loop.run_forever()
+        except BaseException as ex:
+            try:
+                loop.run_until_complete(main.athrow(ex))
+            except StopAsyncIteration as ex:
+                pass
+            else:
+                yielded_twice = True
+        else:
+            try:
+                loop.run_until_complete(main.asend(None))
+            except StopAsyncIteration as ex:
+                pass
+            else:
+                yielded_twice = True
+
+        if yielded_twice:
+            raise RuntimeError("only one yield is supported")
+
+    finally:
+        _cleanup(loop)
diff --git a/runtests.py b/runtests.py
index c4074624..8fa2db93 100644
--- a/runtests.py
+++ b/runtests.py
@@ -112,6 +112,10 @@ def list_dir(prefix, dir):
             print("Skipping '{0}': need at least Python 3.5".format(modname),
                   file=sys.stderr)
             continue
+        if modname == 'test_runner' and (sys.version_info < (3, 6)):
+            print("Skipping '{0}': need at least Python 3.6".format(modname),
+                  file=sys.stderr)
+            continue
         try:
             loader = importlib.machinery.SourceFileLoader(modname, sourcefile)
             mods.append((loader.load_module(), sourcefile))
diff --git a/tests/test_runner.py b/tests/test_runner.py
new file mode 100644
index 00000000..e5d9244b
--- /dev/null
+++ b/tests/test_runner.py
@@ -0,0 +1,205 @@
+"""Tests asyncio.run() and asyncio.run_forever()."""
+
+import asyncio
+import unittest
+import sys
+
+from unittest import mock
+
+
+class TestPolicy(asyncio.AbstractEventLoopPolicy):
+
+    def __init__(self, loop_factory):
+        self.loop_factory = loop_factory
+        self.loop = None
+
+    def get_event_loop(self):
+        # shouldn't ever be called by asyncio.run()
+        # or asyncio.run_forever()
+        raise RuntimeError
+
+    def new_event_loop(self):
+        return self.loop_factory()
+
+    def set_event_loop(self, loop):
+        if loop is not None:
+            # we want to check if the loop is closed
+            # in BaseTest.tearDown
+            self.loop = loop
+
+
+class BaseTest(unittest.TestCase):
+
+    def new_loop(self):
+        loop = asyncio.BaseEventLoop()
+        loop._process_events = mock.Mock()
+        loop._selector = mock.Mock()
+        loop._selector.select.return_value = ()
+        loop.shutdown_ag_run = False
+
+        async def shutdown_asyncgens():
+            loop.shutdown_ag_run = True
+        loop.shutdown_asyncgens = shutdown_asyncgens
+
+        return loop
+
+    def setUp(self):
+        super().setUp()
+
+        policy = TestPolicy(self.new_loop)
+        asyncio.set_event_loop_policy(policy)
+
+    def tearDown(self):
+        policy = asyncio.get_event_loop_policy()
+        if policy.loop is not None:
+            self.assertTrue(policy.loop.is_closed())
+            self.assertTrue(policy.loop.shutdown_ag_run)
+
+        asyncio.set_event_loop_policy(None)
+        super().tearDown()
+
+
+class RunTests(BaseTest):
+
+    def test_asyncio_run_return(self):
+        async def main():
+            await asyncio.sleep(0)
+            return 42
+
+        self.assertEqual(asyncio.run(main()), 42)
+
+    def test_asyncio_run_raises(self):
+        async def main():
+            await asyncio.sleep(0)
+            raise ValueError('spam')
+
+        with self.assertRaisesRegex(ValueError, 'spam'):
+            asyncio.run(main())
+
+    def test_asyncio_run_only_coro(self):
+        for o in {1, lambda: None}:
+            with self.subTest(obj=o), \
+                    self.assertRaisesRegex(ValueError,
+                                           'a coroutine was expected'):
+                asyncio.run(o)
+
+    def test_asyncio_run_debug(self):
+        async def main(expected):
+            loop = asyncio.get_event_loop()
+            self.assertIs(loop.get_debug(), expected)
+
+        asyncio.run(main(False))
+        asyncio.run(main(True), debug=True)
+
+    def test_asyncio_run_from_running_loop(self):
+        async def main():
+            asyncio.run(main())
+
+        with self.assertRaisesRegex(RuntimeError,
+                                    'cannot be called from a running'):
+            asyncio.run(main())
+
+
+class RunForeverTests(BaseTest):
+
+    def stop_soon(self, *, exc=None):
+        loop = asyncio.get_event_loop()
+
+        if exc:
+            def throw():
+                raise exc
+            loop.call_later(0.01, throw)
+        else:
+            loop.call_later(0.01, loop.stop)
+
+    def test_asyncio_run_forever_return(self):
+        async def main():
+            if 0:
+                yield
+            return
+
+        self.assertIsNone(asyncio.run_forever(main()))
+
+    def test_asyncio_run_forever_non_none_yield(self):
+        async def main():
+            yield 1
+
+        with self.assertRaisesRegex(RuntimeError, 'only empty'):
+            self.assertIsNone(asyncio.run_forever(main()))
+
+    def test_asyncio_run_forever_raises_before_yield(self):
+        async def main():
+            await asyncio.sleep(0)
+            raise ValueError('spam')
+            yield
+
+        with self.assertRaisesRegex(ValueError, 'spam'):
+            asyncio.run_forever(main())
+
+    def test_asyncio_run_forever_raises_after_yield(self):
+        async def main():
+            self.stop_soon()
+            yield
+            raise ValueError('spam')
+
+        with self.assertRaisesRegex(ValueError, 'spam'):
+            asyncio.run_forever(main())
+
+    def test_asyncio_run_forever_two_yields(self):
+        async def main():
+            self.stop_soon()
+            yield
+            yield
+            raise ValueError('spam')
+
+        with self.assertRaisesRegex(RuntimeError, 'only one yield'):
+            asyncio.run_forever(main())
+
+    def test_asyncio_run_forever_only_ag(self):
+        async def coro():
+            pass
+
+        for o in {1, lambda: None, coro()}:
+            with self.subTest(obj=o), \
+                    self.assertRaisesRegex(ValueError,
+                                           'an asynchronous.*was expected'):
+                asyncio.run_forever(o)
+
+    def test_asyncio_run_forever_debug(self):
+        async def main(expected):
+            loop = asyncio.get_event_loop()
+            self.assertIs(loop.get_debug(), expected)
+            if 0:
+                yield
+
+        asyncio.run_forever(main(False))
+        asyncio.run_forever(main(True), debug=True)
+
+    def test_asyncio_run_forever_from_running_loop(self):
+        async def main():
+            asyncio.run_forever(main())
+            if 0:
+                yield
+
+        with self.assertRaisesRegex(RuntimeError,
+                                    'cannot be called from a running'):
+            asyncio.run_forever(main())
+
+    def test_asyncio_run_forever_base_exception(self):
+        vi = sys.version_info
+        if vi[:2] != (3, 6) or vi.releaselevel == 'beta' and vi.serial < 4:
+            # See http://bugs.python.org/issue28721 for details.
+            raise unittest.SkipTest(
+                'this test requires Python 3.6b4 or greater')
+
+        class MyExc(BaseException):
+            pass
+
+        async def main():
+            self.stop_soon(exc=MyExc)
+            try:
+                yield
+            except MyExc:
+                pass
+
+        asyncio.run_forever(main())

From 7e67b4859bf3e9f8e8ebb4c246dece916a9e8a98 Mon Sep 17 00:00:00 2001
From: Yury Selivanov <yury@magic.io>
Date: Thu, 17 Nov 2016 11:33:44 -0500
Subject: [PATCH 7/8] Address Ivan's review

---
 asyncio/runners.py   | 8 +++++---
 tests/test_runner.py | 4 ++--
 2 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/asyncio/runners.py b/asyncio/runners.py
index e8fa22dd..4d34663c 100644
--- a/asyncio/runners.py
+++ b/asyncio/runners.py
@@ -102,11 +102,13 @@ async def main():
             "asyncio.run_forever() cannot be called from a running event loop")
     if not isinstance(threading.current_thread(), threading._MainThread):
         raise RuntimeError(
-            "asyncio.run() must be called from the main thread")
+            "asyncio.run_forever() must be called from the main thread")
     if not inspect.isasyncgen(main):
         raise ValueError(
             "an asynchronous generator was expected, got {!r}".format(main))
 
+    one_yield_msg = ("asyncio.run_forever() supports only "
+                     "asynchronous generators with one empty yield")
     loop = events.new_event_loop()
     try:
         events.set_event_loop(loop)
@@ -119,7 +121,7 @@ async def main():
         except StopAsyncIteration as ex:
             return
         if ret is not None:
-            raise RuntimeError("only empty yield is supported")
+            raise RuntimeError(one_yield_msg)
 
         yielded_twice = False
         try:
@@ -140,7 +142,7 @@ async def main():
                 yielded_twice = True
 
         if yielded_twice:
-            raise RuntimeError("only one yield is supported")
+            raise RuntimeError(one_yield_msg)
 
     finally:
         _cleanup(loop)
diff --git a/tests/test_runner.py b/tests/test_runner.py
index e5d9244b..778d69a6 100644
--- a/tests/test_runner.py
+++ b/tests/test_runner.py
@@ -124,7 +124,7 @@ def test_asyncio_run_forever_non_none_yield(self):
         async def main():
             yield 1
 
-        with self.assertRaisesRegex(RuntimeError, 'only empty'):
+        with self.assertRaisesRegex(RuntimeError, 'one empty yield'):
             self.assertIsNone(asyncio.run_forever(main()))
 
     def test_asyncio_run_forever_raises_before_yield(self):
@@ -152,7 +152,7 @@ async def main():
             yield
             raise ValueError('spam')
 
-        with self.assertRaisesRegex(RuntimeError, 'only one yield'):
+        with self.assertRaisesRegex(RuntimeError, 'one empty yield'):
             asyncio.run_forever(main())
 
     def test_asyncio_run_forever_only_ag(self):

From 275072a2fbae0c98619597536d85a65bd72d706b Mon Sep 17 00:00:00 2001
From: Yury Selivanov <yury@magic.io>
Date: Thu, 17 Nov 2016 11:38:26 -0500
Subject: [PATCH 8/8] Fix few nits; add a test.

---
 asyncio/runners.py   |  2 +-
 tests/test_runner.py | 20 +++++++++++++++++++-
 2 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/asyncio/runners.py b/asyncio/runners.py
index 4d34663c..fb70ac57 100644
--- a/asyncio/runners.py
+++ b/asyncio/runners.py
@@ -39,7 +39,7 @@ def run(main, *, debug=False):
     asyncio programs, and should not be used to call asynchronous
     APIs.
 
-    Example::
+    Example:
 
         async def main():
             await asyncio.sleep(1)
diff --git a/tests/test_runner.py b/tests/test_runner.py
index 778d69a6..d439353f 100644
--- a/tests/test_runner.py
+++ b/tests/test_runner.py
@@ -127,6 +127,20 @@ async def main():
         with self.assertRaisesRegex(RuntimeError, 'one empty yield'):
             self.assertIsNone(asyncio.run_forever(main()))
 
+    def test_asyncio_run_forever_try_finally(self):
+        DONE = 0
+
+        async def main():
+            nonlocal DONE
+            self.stop_soon()
+            try:
+                yield
+            finally:
+                DONE += 1
+
+        asyncio.run_forever(main())
+        self.assertEqual(DONE, 1)
+
     def test_asyncio_run_forever_raises_before_yield(self):
         async def main():
             await asyncio.sleep(0)
@@ -192,14 +206,18 @@ def test_asyncio_run_forever_base_exception(self):
             raise unittest.SkipTest(
                 'this test requires Python 3.6b4 or greater')
 
+        DONE = 0
+
         class MyExc(BaseException):
             pass
 
         async def main():
+            nonlocal DONE
             self.stop_soon(exc=MyExc)
             try:
                 yield
             except MyExc:
-                pass
+                DONE += 1
 
         asyncio.run_forever(main())
+        self.assertEqual(DONE, 1)