Merge branch 'master' into ssl_optimizations

Author: fantix

tests/test_fs_event.py

@@ -1,100 +1,87 @@
 import asyncio
+import contextlib
 import os.path
 import tempfile
 
 from uvloop import _testbase as tb
 from uvloop.loop import FileSystemEvent
 
 
-class Test_UV_FS_EVENT_CHANGE(tb.UVTestCase):
-    async def _file_writer(self):
-        f = await self.q.get()
-        while True:
-            f.write('hello uvloop\n')
-            f.flush()
-            x = await self.q.get()
-            if x is None:
-                return
-
-    def fs_event_setup(self):
-        self.change_event_count = 0
-        self.fname = ''
-        self.q = asyncio.Queue()
-
-    def event_cb(self, ev_fname: bytes, evt: FileSystemEvent):
-        _d, fn = os.path.split(self.fname)
-        self.assertEqual(ev_fname, fn)
-        self.assertEqual(evt, FileSystemEvent.CHANGE)
-        self.change_event_count += 1
-        if self.change_event_count < 4:
-            self.q.put_nowait(0)
-        else:
-            self.q.put_nowait(None)
+class Test_UV_FS_Event(tb.UVTestCase):
+    def setUp(self):
+        super().setUp()
+        self.exit_stack = contextlib.ExitStack()
+        self.tmp_dir = self.exit_stack.enter_context(
+            tempfile.TemporaryDirectory()
+        )
+
+    def tearDown(self):
+        self.exit_stack.close()
+        super().tearDown()
 
     def test_fs_event_change(self):
-        self.fs_event_setup()
-
-        async def run(write_task):
-            self.q.put_nowait(tf)
-            try:
-                await asyncio.wait_for(write_task, 4)
-            except asyncio.TimeoutError:
-                write_task.cancel()
-
-        with tempfile.NamedTemporaryFile('wt') as tf:
-            self.fname = tf.name.encode()
-            h = self.loop._monitor_fs(tf.name, self.event_cb)
+        change_event_count = 0
+        filename = "fs_event_change.txt"
+        path = os.path.join(self.tmp_dir, filename)
+        q = asyncio.Queue()
+
+        with open(path, 'wt') as f:
+            async def file_writer():
+                while True:
+                    f.write('hello uvloop\n')
+                    f.flush()
+                    x = await q.get()
+                    if x is None:
+                        return
+
+            def event_cb(ev_fname: bytes, evt: FileSystemEvent):
+                nonlocal change_event_count
+                self.assertEqual(ev_fname, filename.encode())
+                self.assertEqual(evt, FileSystemEvent.CHANGE)
+                change_event_count += 1
+                if change_event_count < 4:
+                    q.put_nowait(0)
+                else:
+                    q.put_nowait(None)
+
+            h = self.loop._monitor_fs(path, event_cb)
+            self.loop.run_until_complete(
+                asyncio.sleep(0.1)  # let monitor start
+            )
             self.assertFalse(h.cancelled())
 
-            self.loop.run_until_complete(run(
-                self.loop.create_task(self._file_writer())))
+            self.loop.run_until_complete(asyncio.wait_for(file_writer(), 4))
             h.cancel()
             self.assertTrue(h.cancelled())
 
-        self.assertEqual(self.change_event_count, 4)
-
-
-class Test_UV_FS_EVENT_RENAME(tb.UVTestCase):
-    async def _file_renamer(self):
-        await self.q.get()
-        os.rename(os.path.join(self.dname, self.changed_name),
-                  os.path.join(self.dname, self.changed_name + "-new"))
-        await self.q.get()
-
-    def fs_event_setup(self):
-        self.dname = ''
-        self.changed_name = "hello_fs_event.txt"
-        self.changed_set = {self.changed_name, self.changed_name + '-new'}
-        self.q = asyncio.Queue()
-
-    def event_cb(self, ev_fname: bytes, evt: FileSystemEvent):
-        ev_fname = ev_fname.decode()
-        self.assertEqual(evt, FileSystemEvent.RENAME)
-        self.changed_set.remove(ev_fname)
-        if len(self.changed_set) == 0:
-            self.q.put_nowait(None)
+        self.assertEqual(change_event_count, 4)
 
     def test_fs_event_rename(self):
-        self.fs_event_setup()
-
-        async def run(write_task):
-            self.q.put_nowait(0)
-            try:
-                await asyncio.wait_for(write_task, 4)
-            except asyncio.TimeoutError:
-                write_task.cancel()
-
-        with tempfile.TemporaryDirectory() as td_name:
-            self.dname = td_name
-            f = open(os.path.join(td_name, self.changed_name), 'wt')
+        orig_name = "hello_fs_event.txt"
+        new_name = "hello_fs_event_rename.txt"
+        changed_set = {orig_name, new_name}
+        event = asyncio.Event()
+
+        async def file_renamer():
+            os.rename(os.path.join(self.tmp_dir, orig_name),
+                      os.path.join(self.tmp_dir, new_name))
+            await event.wait()
+
+        def event_cb(ev_fname: bytes, evt: FileSystemEvent):
+            ev_fname = ev_fname.decode()
+            self.assertEqual(evt, FileSystemEvent.RENAME)
+            changed_set.discard(ev_fname)
+            if len(changed_set) == 0:
+                event.set()
+
+        with open(os.path.join(self.tmp_dir, orig_name), 'wt') as f:
             f.write('hello!')
-            f.close()
-            h = self.loop._monitor_fs(td_name, self.event_cb)
-            self.assertFalse(h.cancelled())
+        h = self.loop._monitor_fs(self.tmp_dir, event_cb)
+        self.loop.run_until_complete(asyncio.sleep(0.5))  # let monitor start
+        self.assertFalse(h.cancelled())
 
-            self.loop.run_until_complete(run(
-                self.loop.create_task(self._file_renamer())))
-            h.cancel()
-            self.assertTrue(h.cancelled())
+        self.loop.run_until_complete(asyncio.wait_for(file_renamer(), 4))
+        h.cancel()
+        self.assertTrue(h.cancelled())
 
-        self.assertEqual(len(self.changed_set), 0)
+        self.assertEqual(len(changed_set), 0)

uvloop/cbhandles.pyx

@@ -9,8 +9,10 @@ cdef class Handle:
     cdef inline _set_loop(self, Loop loop):
         self.loop = loop
         if UVLOOP_DEBUG:
-            loop._debug_cb_handles_total += 1
-            loop._debug_cb_handles_count += 1
+            system.__atomic_fetch_add(
+                &loop._debug_cb_handles_total, 1, system.__ATOMIC_RELAXED)
+            system.__atomic_fetch_add(
+                &loop._debug_cb_handles_count, 1, system.__ATOMIC_RELAXED)
         if loop._debug:
             self._source_traceback = extract_stack()
 
@@ -21,7 +23,8 @@ cdef class Handle:
 
     def __dealloc__(self):
         if UVLOOP_DEBUG and self.loop is not None:
-            self.loop._debug_cb_handles_count -= 1
+            system.__atomic_fetch_sub(
+                &self.loop._debug_cb_handles_count, 1, system.__ATOMIC_RELAXED)
         if self.loop is None:
             raise RuntimeError('Handle.loop is None in Handle.__dealloc__')
 
@@ -174,8 +177,10 @@ cdef class TimerHandle:
         self._cancelled = 0
 
         if UVLOOP_DEBUG:
-            self.loop._debug_cb_timer_handles_total += 1
-            self.loop._debug_cb_timer_handles_count += 1
+            system.__atomic_fetch_add(
+                &self.loop._debug_cb_timer_handles_total, 1, system.__ATOMIC_RELAXED)
+            system.__atomic_fetch_add(
+                &self.loop._debug_cb_timer_handles_count, 1, system.__ATOMIC_RELAXED)
 
         if context is None:
             context = Context_CopyCurrent()
@@ -205,7 +210,8 @@ cdef class TimerHandle:
 
     def __dealloc__(self):
         if UVLOOP_DEBUG:
-            self.loop._debug_cb_timer_handles_count -= 1
+            system.__atomic_fetch_sub(
+                &self.loop._debug_cb_timer_handles_count, 1, system.__ATOMIC_RELAXED)
         if self.timer is not None:
             raise RuntimeError('active TimerHandle is deallacating')
 

uvloop/includes/stdlib.pxi

@@ -37,7 +37,6 @@ cdef aio_wait = asyncio.wait
 cdef aio_wrap_future = asyncio.wrap_future
 cdef aio_logger = asyncio.log.logger
 cdef aio_iscoroutine = asyncio.iscoroutine
-cdef aio_iscoroutinefunction = asyncio.iscoroutinefunction
 cdef aio_BaseProtocol = asyncio.BaseProtocol
 cdef aio_Protocol = asyncio.Protocol
 cdef aio_isfuture = getattr(asyncio, 'isfuture', None)
@@ -65,6 +64,7 @@ cdef gc_disable = gc.disable
 
 cdef iter_chain = itertools.chain
 cdef inspect_isgenerator = inspect.isgenerator
+cdef inspect_iscoroutinefunction = inspect.iscoroutinefunction
 
 cdef int has_IPV6_V6ONLY = hasattr(socket, 'IPV6_V6ONLY')
 cdef int IPV6_V6ONLY = getattr(socket, 'IPV6_V6ONLY', -1)

uvloop/includes/system.pxd

@@ -94,3 +94,11 @@ cdef extern from "includes/fork_handler.h":
     void setForkHandler(OnForkHandler handler)
     void resetForkHandler()
     void setMainThreadID(uint64_t id)
+
+
+cdef extern from * nogil:
+    uint64_t __atomic_fetch_add(uint64_t *ptr, uint64_t val, int memorder)
+    uint64_t __atomic_fetch_sub(uint64_t *ptr, uint64_t val, int memorder)
+
+    cdef enum:
+        __ATOMIC_RELAXED

uvloop/loop.pxd

@@ -50,7 +50,6 @@ cdef class Loop:
         object _default_executor
         object _ready
         set _queued_streams, _executing_streams
-        Py_ssize_t _ready_len
 
         set _servers
 

uvloop/loop.pyx

@@ -181,7 +181,6 @@ cdef class Loop:
         self._queued_streams = set()
         self._executing_streams = set()
         self._ready = col_deque()
-        self._ready_len = 0
 
         self.handler_async = UVAsync.new(
             self, <method_t>self._on_wake, self)
@@ -440,7 +439,7 @@ cdef class Loop:
             self.handler_async.send()
 
     cdef _on_wake(self):
-        if ((self._ready_len > 0 or self._stopping) and
+        if ((len(self._ready) > 0 or self._stopping) and
                 not self.handler_idle.running):
             self.handler_idle.start()
 
@@ -481,8 +480,7 @@ cdef class Loop:
         if len(self._queued_streams):
             self._exec_queued_writes()
 
-        self._ready_len = len(self._ready)
-        if self._ready_len == 0 and self.handler_idle.running:
+        if len(self._ready) == 0 and self.handler_idle.running:
             self.handler_idle.stop()
 
         if self._stopping:
@@ -570,7 +568,6 @@ cdef class Loop:
         for cb_handle in self._ready:
             cb_handle.cancel()
         self._ready.clear()
-        self._ready_len = 0
 
         if self._polls:
             for poll_handle in self._polls.values():
@@ -672,7 +669,6 @@ cdef class Loop:
     cdef inline _append_ready_handle(self, Handle handle):
         self._check_closed()
         self._ready.append(handle)
-        self._ready_len += 1
 
     cdef inline _call_soon_handle(self, Handle handle):
         self._append_ready_handle(handle)
@@ -2731,7 +2727,7 @@ cdef class Loop:
             return transport, protocol
 
     def run_in_executor(self, executor, func, *args):
-        if aio_iscoroutine(func) or aio_iscoroutinefunction(func):
+        if aio_iscoroutine(func) or inspect_iscoroutinefunction(func):
             raise TypeError("coroutines cannot be used with run_in_executor()")
 
         self._check_closed()
@@ -2910,7 +2906,7 @@ cdef class Loop:
                 'the main thread')
 
         if (aio_iscoroutine(callback)
-                or aio_iscoroutinefunction(callback)):
+                or inspect_iscoroutinefunction(callback)):
             raise TypeError(
                 "coroutines cannot be used with add_signal_handler()")