generator-based event loop impl

Author: iximiuz

README.md

@@ -1,6 +1,8 @@
-# Simple but workable event loop implementation in Python (< 100 LOC)
+## Simple but workable event loop in Python (< 100 LOC)
 
-Example:
+Check out the series about the event loop on my blog <a href="https://micromind.me/en/posts/explain-event-loop-in-100-lines-of-code/">Explain event loop in 100 lines of code</a> and <a href="https://micromind.me/en/posts/from-callback-hell-to-async-await-heaven/">From Callback Hell to async/await Heaven</a>.
+
+### Example using callback-style:
 ```python
 import socket as _socket
 
@@ -51,5 +53,38 @@ Give it a try:
 > python event_loop.py 53210
 ```
 
-Check out the article about this code on my blog <a href="https://micromind.me/en/posts/explain-event-loop-in-100-lines-of-code/">Explain event loop in 100 lines of code</a>.
+### Example using async-await-like style:
+```python
+import socket as _socket
+
+class EventLoop: pass
 
+class Context: pass
+
+class socket(Context): pass
+
+def http_get(sock):
+  try:
+    yield sock.sendall(b'GET / HTTP/1.1\r\nHost: t.co\r\n\r\n')
+    return sock.recv(1024)
+  finally:
+    sock.close()
+
+def main(serv_addr):
+  sock = socket(AF_INET, SOCK_STREAM)
+  yield sock.connect(serv_addr)
+  resp = yield http_get(sock)
+  print(resp)
+
+if __name__ == '__main__':
+    event_loop = EventLoop()
+    Context.set_event_loop(event_loop)
+
+    serv_addr = ('t.co', 80)
+    event_loop.run(main)
+```
+
+Give it a try:
+```bash
+> python event_loop_gen.py 53210
+```

event_loop.py

@@ -278,7 +278,7 @@ def on_account(err, acc=None):
     set_timer(random.randint(0, 10e6), on_timer)
 
 
-def main(serv_addr):
+def main1(serv_addr):
     def on_balance(err, balance=None):
         if err:
             print('ERROR', err)
@@ -289,10 +289,40 @@ def on_balance(err, balance=None):
         get_user_balance(serv_addr, i, on_balance)
 
 
+def main2(*args):
+  sock = socket(_socket.AF_INET, _socket.SOCK_STREAM)
+
+  def on_conn(err):
+    if err:
+      return print(err)
+
+    def on_sent(err):
+      if err:
+        sock.close()
+        return print(err)
+
+      def on_resp(err, resp=None):
+        sock.close()
+        if err:
+          return print(err)
+        print(resp)
+
+      sock.recv(1024, on_resp)
+
+    sock.sendall(b'GET / HTTP/1.1\r\nHost: t.co\r\n\r\n', on_sent)
+
+  sock.connect(('t.co', 80), on_conn)
+
+
 if __name__ == '__main__':
+    print('Run main1()')
     event_loop = EventLoop()
     Context.set_event_loop(event_loop)
 
     serv_addr = ('127.0.0.1', int(sys.argv[1]))
-    event_loop.run(main, serv_addr)
+    event_loop.run(main1, serv_addr)
 
+    print('\nRun main2()')
+    event_loop = EventLoop()
+    Context.set_event_loop(event_loop)
+    event_loop.run(main2)

event_loop_ng.py event_loop_gen.py

@@ -42,13 +42,11 @@ def set_timer(self, duration):
     def _execute(self, callback, *args):
         self._time = hrtime()
         try:
-            # unwind(callback(*args)).catch(
-            #        lambda e: print('Uncaught rejection:', e))
             ret = callback(*args)
             if is_generator(ret):
-                unwind2(ret, 
-                        ok=lambda *_: None, 
-                        fail=lambda e: print('Uncaught rejection:', e))
+                unwind(ret,
+                       ok=lambda *_: None,
+                       fail=lambda e: print('Uncaught rejection:', e))
 
         except Exception as err:
             print('Uncaught exception:', err)
@@ -119,7 +117,7 @@ def evloop(self):
         return self._event_loop
 
 
-def unwind2(gen, ok, fail, ret=None, method='send'):
+def unwind(gen, ok, fail, ret=None, method='send'):
     try:
         ret = getattr(gen, method)(ret)
     except StopIteration as stop:
@@ -128,16 +126,16 @@ def unwind2(gen, ok, fail, ret=None, method='send'):
         return fail(e)
 
     if is_generator(ret):
-        unwind2(ret,
-                ok=lambda x: unwind2(gen, ok, fail, x),
-                fail=lambda e: unwind2(gen, ok, fail, e, 'throw'))
+        unwind(ret,
+               ok=lambda x: unwind(gen, ok, fail, x),
+               fail=lambda e: unwind(gen, ok, fail, e, 'throw'))
     elif is_promise(ret):
-        ret.then(lambda x=None: unwind2(gen, ok, fail, x)) \
-           .catch(lambda e: unwind2(gen, ok, fail, e, 'throw'))
+        ret.then(lambda x=None: unwind(gen, ok, fail, x)) \
+           .catch(lambda e: unwind(gen, ok, fail, e, 'throw'))
     else:
-        wait_all(ret, 
-                lambda x=None: unwind2(gen, ok, fail, x),
-                lambda e: unwind2(gen, ok, fail, e, 'throw'))
+        wait_all(ret,
+                lambda x=None: unwind(gen, ok, fail, x),
+                lambda e: unwind(gen, ok, fail, e, 'throw'))
 
 
 def wait_all(col, ok, fail):
@@ -157,7 +155,7 @@ def _do_resolve(val):
 
     for i, c in enumerate(col):
         if is_generator(c):
-            unwind2(c, ok=_resolve_single(i), fail=fail)
+            unwind(c, ok=_resolve_single(i), fail=fail)
             continue
 
         if is_promise(c):
@@ -168,59 +166,6 @@ def _do_resolve(val):
                         'can be yielded to event loop')
 
 
-def unwind(gen):
-    p = Promise()
-    if not is_generator(gen):
-        p._resolve(gen)
-        return p
-
-    def _on_fulfilled(res=None):
-        ret = None
-        try:
-            ret = gen.send(res)
-        except StopIteration as stop:
-            return p._resolve(stop.value)
-        except Exception as exc:
-            return p._reject(exc)
-        _next(ret)
-
-    def _on_rejected(err):
-        ret = None
-        try:
-            ret = gen.throw(err)
-        except StopIteration as stop:
-            return p._resolve(stop.value)
-        except Exception as exc:
-            return p._reject(exc)
-        _next(ret)
-
-    def _next(ret):
-        nextp = to_promise(ret)
-        if not nextp:
-            return _on_rejected(Exception('Only promise or generator '
-                                          'can be yielded to event loop'))
-        nextp.then(_on_fulfilled)
-        nextp.catch(_on_rejected)
-
-    _on_fulfilled()
-    return p
-
-
-def to_promise(val):
-    if is_promise(val):
-        return val
-
-    if is_generator(val):
-        return unwind(val)
-
-    if isinstance(val, list):
-        promises = [to_promise(x) for x in val]
-        if len(val) == len(list(filter(None, promises))):
-            return Promise.all(promises)
-
-    return None
-
-
 def is_generator(val):
     return isinstance(val, types.GeneratorType)
 
@@ -293,7 +238,7 @@ def _reject(self, err):
         self._value = err
         for cb in self._on_reject:
             cb(err)
-    
+
 
 class IOError(Exception):
     def __init__(self, message, errorno, errorcode):
@@ -322,7 +267,7 @@ def __init__(self, *args):
         # 1 - connecting
         # 2 - connected
         # 3 - closed
-        self._state = 0 
+        self._state = 0
         self._callbacks = {}
 
     def connect(self, addr):
@@ -365,7 +310,7 @@ def _on_write_ready(err):
             nonlocal data
             if err:
                 return p._reject(err)
-            
+
             n = self._sock.send(data)
             if n < len(data):
                 data = data[n:]
@@ -408,11 +353,11 @@ def _on_event(self, mask):
                 cb(err)
 
     def _get_sock_error(self):
-        err = self._sock.getsockopt(_socket.SOL_SOCKET, 
+        err = self._sock.getsockopt(_socket.SOL_SOCKET,
                                     _socket.SO_ERROR)
         if not err:
             return None
-        return IOError('connection failed', 
+        return IOError('connection failed',
                        err, errno.errorcode[err])
 
 ###############################################################################
@@ -457,7 +402,7 @@ def print_balance(serv_addr, user_id):
         print('Catched:', exc)
 
 
-def main(serv_addr):
+def main1(serv_addr):
     def on_sleep():
         b = yield get_user_balance(serv_addr, 1)
         print('side flow:', b)
@@ -469,10 +414,27 @@ def on_sleep():
     yield tasks
 
 
+def main2(*args):
+  sock = socket(_socket.AF_INET, _socket.SOCK_STREAM)
+  yield sock.connect(('t.co', 80))
+
+  try:
+    yield sock.sendall(b'GET / HTTP/1.1\r\nHost: t.co\r\n\r\n')
+    val = yield sock.recv(1024)
+    print(val)
+  finally:
+    sock.close()
+
+
 if __name__ == '__main__':
+    print('Run main1()')
     event_loop = EventLoop()
     Context.set_event_loop(event_loop)
 
     serv_addr = ('127.0.0.1', int(sys.argv[1]))
-    event_loop.run(main, serv_addr)
+    event_loop.run(main1, serv_addr)
 
+    print('\nRun main2()')
+    event_loop = EventLoop()
+    Context.set_event_loop(event_loop)
+    event_loop.run(main2)

server.py

@@ -24,8 +24,8 @@ def handle(self):
             raise Exception('Max request length exceeded')
 
         method, entity_kind, entity_id = req[:-1].split(' ', 3)
-        if (method != 'GET' 
-           or entity_kind not in ('user', 'account') 
+        if (method != 'GET'
+           or entity_kind not in ('user', 'account')
            or not entity_id.isdigit()):
             raise Exception('Bad request')
 
@@ -38,7 +38,7 @@ def handle(self):
 
             if 'account_id' not in user:
                 account_id = str(len(self.accounts) + 1)
-                account = {'id': account_id, 
+                account = {'id': account_id,
                            'balance': random.randint(0, 100)}
                 self.accounts[account_id] = account
                 user['account_id'] = account_id