Merge branch 'master' into example-fix-314

Author: Jamie-Chang

tests/test_context.py

@@ -512,9 +512,10 @@ async def write_over():
                         proto.transport.write(b'q' * 16384)
                         count += 1
                 else:
-                    proto.transport.write(b'q' * 16384)
                     proto.transport.set_write_buffer_limits(high=256, low=128)
-                    count += 1
+                    while not proto.transport.get_write_buffer_size():
+                        proto.transport.write(b'q' * 16384)
+                        count += 1
                 return count
 
             s = self.loop.run_in_executor(None, accept)

tests/test_dns.py

@@ -217,6 +217,10 @@ def test_getaddrinfo_22(self):
         self._test_getaddrinfo(payload, 80)
         self._test_getaddrinfo(payload, 80, type=socket.SOCK_STREAM)
 
+    def test_getaddrinfo_broadcast(self):
+        self._test_getaddrinfo('<broadcast>', 80)
+        self._test_getaddrinfo('<broadcast>', 80, type=socket.SOCK_STREAM)
+
     ######
 
     def test_getnameinfo_1(self):

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)

tests/test_process.py

@@ -685,6 +685,12 @@ async def cancel_make_transport():
             self.loop.run_until_complete(cancel_make_transport())
 
     def test_cancel_post_init(self):
+        if sys.version_info >= (3, 13) and self.implementation == 'asyncio':
+            # https://github.com/python/cpython/issues/103847#issuecomment-3736561321
+            # This test started to flake on CPython 3.13 and later,
+            # so we skip it for asyncio tests until the issue is resolved.
+            self.skipTest('flaky test on CPython 3.13+')
+
         async def cancel_make_transport():
             coro = self.loop.subprocess_exec(asyncio.SubprocessProtocol,
                                              *self.PROGRAM_BLOCKED)

tests/test_tcp.py

@@ -405,7 +405,7 @@ async def client(addr):
             self.assertEqual(await reader.readexactly(2), b'OK')
 
             re = r'(a bytes-like object)|(must be byte-ish)'
-            if sys.version_info >= (3, 14):
+            if sys.version_info >= (3, 13, 9):
                 re += r'|(must be a bytes, bytearray, or memoryview object)'
             with self.assertRaisesRegex(TypeError, re):
                 writer.write('AAAA')
@@ -1226,21 +1226,16 @@ def resume_writing(self):
             t, p = await self.loop.create_connection(Protocol, *addr)
 
             t.write(b'q' * 512)
-            self.assertEqual(t.get_write_buffer_size(), 512)
-
             t.set_write_buffer_limits(low=16385)
-            self.assertFalse(paused)
             self.assertEqual(t.get_write_buffer_limits(), (16385, 65540))
 
             with self.assertRaisesRegex(ValueError, 'high.*must be >= low'):
                 t.set_write_buffer_limits(high=0, low=1)
 
             t.set_write_buffer_limits(high=1024, low=128)
-            self.assertFalse(paused)
             self.assertEqual(t.get_write_buffer_limits(), (128, 1024))
 
             t.set_write_buffer_limits(high=256, low=128)
-            self.assertTrue(paused)
             self.assertEqual(t.get_write_buffer_limits(), (128, 256))
 
             t.close()

tests/test_udp.py

@@ -378,6 +378,22 @@ def test_udp_sendto_dns(self):
         s_transport.close()
         self.loop.run_until_complete(asyncio.sleep(0.01))
 
+    def test_udp_sendto_broadcast(self):
+        coro = self.loop.create_datagram_endpoint(
+            asyncio.DatagramProtocol,
+            local_addr=('127.0.0.1', 0),
+            family=socket.AF_INET)
+
+        s_transport, server = self.loop.run_until_complete(coro)
+
+        try:
+            s_transport.sendto(b'aaaa', ('<broadcast>', 80))
+        except ValueError as exc:
+            raise AssertionError('sendto raises {}.'.format(exc))
+
+        s_transport.close()
+        self.loop.run_until_complete(asyncio.sleep(0.01))
+
     def test_send_after_close(self):
         coro = self.loop.create_datagram_endpoint(
             asyncio.DatagramProtocol,

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/handles/stream.pxd

@@ -1,11 +1,17 @@
+cdef enum ProtocolType:
+    SIMPLE = 0          # User Protocol doesn't support asyncio.BufferedProtocol
+    BUFFERED = 1        # User Protocol supports asyncio.BufferedProtocol
+    SSL_PROTOCOL = 2    # Our own SSLProtocol
+
+
 cdef class UVStream(UVBaseTransport):
     cdef:
         uv.uv_shutdown_t _shutdown_req
         bint __shutting_down
         bint __reading
         bint __read_error_close
 
-        bint __buffered
+        ProtocolType __protocol_type
         object _protocol_get_buffer
         object _protocol_buffer_updated
 
@@ -16,6 +22,8 @@ cdef class UVStream(UVBaseTransport):
         Py_buffer _read_pybuf
         bint _read_pybuf_acquired
 
+    cpdef write(self, object buf)
+
     # All "inline" methods are final
 
     cdef inline _init(self, Loop loop, object protocol, Server server,
@@ -39,8 +47,8 @@ cdef class UVStream(UVBaseTransport):
 
     # _exec_write() is the method that does the actual send, and _try_write()
     # is a fast-path used in _exec_write() to send a single chunk.
-    cdef inline _exec_write(self)
-    cdef inline _try_write(self, object data)
+    cdef inline bint _exec_write(self) except -1
+    cdef inline Py_ssize_t _try_write(self, object data) except -2
 
     cdef _close(self)