diff --git a/examples/bench/dns.py b/examples/bench/dns.py new file mode 100644 index 00000000..82ca6657 --- /dev/null +++ b/examples/bench/dns.py @@ -0,0 +1,196 @@ +"""Compare DNS throughput and executor isolation in two uvloop checkouts. + +Build both checkouts in place with matching optimized uvloop/libuv settings. +For example, on Linux: + + python examples/bench/dns.py --baseline /path/to/baseline \\ + --candidate /path/to/pr --affinity 0,1 --samples 15 + +Use --count 65536 for longer samples. The native_numeric_* cases specify +socktype=0 to bypass the numeric fast path without performing network DNS. +Use identical --baseline and --candidate paths to measure harness noise. +""" + +import argparse +import asyncio +import concurrent.futures +import json +import os +import pathlib +import platform +import socket +import statistics +import subprocess +import sys +import threading +import time + + +CASES = ( + 'forward_sequential', 'forward_concurrent', + 'reverse_sequential', 'reverse_concurrent', + 'numeric_fast_path', 'native_numeric_sequential', + 'native_numeric_concurrent', 'executor_saturated', +) + + +def run_child(args): + repo = args.repo.resolve() + sys.path.insert(0, str(repo)) + import uvloop + import uvloop.loop + + loop = uvloop.new_event_loop() + if loop._debug_cc: + raise RuntimeError('benchmark requires a non-debug build') + if not pathlib.Path(uvloop.loop.__file__).is_relative_to(repo): + raise RuntimeError('uvloop was imported from the wrong checkout') + asyncio.set_event_loop(loop) + native_numeric = args.case.startswith('native_numeric') + + async def lookup(reverse=False, numeric=False): + if native_numeric: + return await loop.getaddrinfo( + '127.0.0.1', 443, family=socket.AF_INET, + type=0, flags=socket.AI_NUMERICHOST) + if reverse: + return await loop.getnameinfo( + ('127.0.0.1', 443), + socket.NI_NUMERICHOST | socket.NI_NUMERICSERV) + return await loop.getaddrinfo( + '127.0.0.1' if numeric else 'localhost', 443, + family=socket.AF_INET, type=socket.SOCK_STREAM) + + async def run(): + if args.case == 'executor_saturated': + # Occupy every worker with sleeping application work. DNS should + # remain independent of this queue, as it is in native libuv. + loop.set_default_executor( + concurrent.futures.ThreadPoolExecutor(max_workers=4)) + release = threading.Event() + ready = [threading.Event() for _ in range(4)] + + def application_work(event): + event.set() + release.wait() + + tasks = [loop.run_in_executor(None, application_work, event) + for event in ready] + while not all(event.is_set() for event in ready): + await asyncio.sleep(0.001) + timer = loop.call_later(0.2, release.set) + start = time.perf_counter() + try: + await lookup() + elapsed = time.perf_counter() - start + finally: + release.set() + timer.cancel() + await asyncio.gather(*tasks) + return {'dns_latency_ms': elapsed * 1000} + + reverse = args.case.startswith('reverse') + numeric = args.case == 'numeric_fast_path' + concurrency = 32 if args.case.endswith('concurrent') else 1 + count = args.count or (32768 if numeric else + (4096 if reverse else 8192)) + count = ((count + concurrency - 1) // concurrency) * concurrency + for _ in range(100): + await lookup(reverse, numeric) + + async def worker(): + for _ in range(count // concurrency): + result = await lookup(reverse, numeric) + if not result: + raise RuntimeError('lookup returned no result') + + start = time.perf_counter() + await asyncio.gather(*(worker() for _ in range(concurrency))) + elapsed = time.perf_counter() - start + return {'lookups_per_second': count / elapsed, + 'microseconds_per_lookup': elapsed * 1e6 / count, + 'lookups': count, 'concurrency': concurrency} + + try: + result = loop.run_until_complete(run()) + loop.run_until_complete(loop.shutdown_default_executor()) + finally: + loop.close() + result.update(module=uvloop.loop.__file__, version=uvloop.__version__, + libuv_version=uvloop.loop.libuv_get_version()) + print(json.dumps(result)) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--repo', type=pathlib.Path, + help=argparse.SUPPRESS) + parser.add_argument('--case', choices=CASES, help=argparse.SUPPRESS) + parser.add_argument('--baseline', type=pathlib.Path) + parser.add_argument('--candidate', type=pathlib.Path) + parser.add_argument('--samples', type=int, default=7) + parser.add_argument('--count', type=int, default=0) + parser.add_argument('--cases', nargs='+', choices=CASES, default=CASES) + parser.add_argument('--affinity', help='CPU IDs, e.g. 0,1 (Linux only)') + parser.add_argument('--output', type=pathlib.Path, + default=pathlib.Path('dns-benchmark.json')) + args = parser.parse_args() + if args.samples < 1 or args.count < 0: + parser.error('--samples must be positive and --count nonnegative') + if args.repo: + if not args.case: + parser.error('--repo requires --case') + run_child(args) + return + if not args.baseline or not args.candidate: + parser.error('--baseline and --candidate are required') + if args.affinity: + if not hasattr(os, 'sched_setaffinity'): + parser.error('--affinity requires sched_setaffinity') + os.sched_setaffinity(0, {int(cpu) for cpu in args.affinity.split(',')}) + result = { + 'environment': { + 'python': sys.version, + 'platform': platform.platform(), + 'cpu_count': os.cpu_count(), + 'cpu_affinity': (sorted(os.sched_getaffinity(0)) + if hasattr(os, 'sched_getaffinity') else None), + 'uv_threadpool_size': 4, + 'python_hash_seed': 0, + 'samples': args.samples, + }, + 'cases': {}, + } + env = dict(os.environ) + env.pop('PYTHONASYNCIODEBUG', None) + env.pop('LD_PRELOAD', None) + env['UV_THREADPOOL_SIZE'] = '4' + env['PYTHONHASHSEED'] = '0' + for case in args.cases: + samples = {name: [] for name in ('baseline', 'candidate')} + for index in range(args.samples): + order = (('baseline', 'candidate') if index % 2 == 0 else + ('candidate', 'baseline')) + for name in order: + output = subprocess.check_output( + [sys.executable, str(pathlib.Path(__file__).resolve()), + '--repo', str(getattr(args, name)), '--case', case, + '--count', str(args.count)], + text=True, env=env, timeout=60) + samples[name].append(json.loads(output)) + key = ('dns_latency_ms' if case == 'executor_saturated' else + 'lookups_per_second') + medians = {name: statistics.median(sample[key] for sample in runs) + for name, runs in samples.items()} + result['cases'][case] = { + 'samples': samples, + 'median': medians, + 'candidate_over_baseline': (medians['candidate'] / + medians['baseline']), + } + print(case, json.dumps(medians), flush=True) + args.output.write_text(json.dumps(result, indent=2) + '\n') + + +if __name__ == '__main__': + main() diff --git a/tests/test_dns.py b/tests/test_dns.py index 106ef580..00f95234 100644 --- a/tests/test_dns.py +++ b/tests/test_dns.py @@ -1,5 +1,15 @@ import asyncio +import concurrent.futures +import os +import pathlib +import select +import shutil import socket +import subprocess +import sys +import tempfile +import textwrap +import threading import unittest from uvloop import _testbase as tb @@ -241,6 +251,29 @@ def test_getnameinfo_5(self): class Test_UV_DNS(BaseTestDNS, tb.UVTestCase): + def test_getaddrinfo_with_busy_default_executor(self): + started = threading.Event() + release = threading.Event() + + def application_work(): + started.set() + release.wait(10) + + executor = concurrent.futures.ThreadPoolExecutor(max_workers=1) + self.loop.set_default_executor(executor) + work = executor.submit(application_work) + try: + self.assertTrue(started.wait(5)) + result = self.loop.run_until_complete(asyncio.wait_for( + self.loop.getaddrinfo('localhost', 443, + family=socket.AF_INET, + type=socket.SOCK_STREAM), 5)) + self.assertTrue(result) + self.assertFalse(work.done()) + finally: + release.set() + executor.shutdown(wait=True) + def test_getaddrinfo_close_loop(self): # Test that we can close the loop with a running # DNS query. @@ -266,3 +299,295 @@ async def run(): class Test_AIO_DNS(BaseTestDNS, tb.AIOTestCase): pass + + +@unittest.skipUnless(sys.platform == 'linux', 'requires LD_PRELOAD') +class TestDNSClose(unittest.TestCase): + # Regression for MagicStack/uvloop#777. + # Block only our test hostname, without consulting the network. The worker + # signals that it has started, then waits until the parent releases it. + SHIM = r""" + #define _GNU_SOURCE + #include + #include + #include + #include + #include + #include + + static struct addrinfo *completed_result; + + static void notify(char byte) { + int started = atoi(getenv("DNS_STARTED_FD")); + ssize_t n; + do { + n = write(started, &byte, 1); + } while (n < 0 && errno == EINTR); + if (n != 1) + _exit(90); + } + + static void wait_for_release(void) { + int release = atoi(getenv("DNS_RELEASE_FD")); + char byte; + ssize_t n; + notify('S'); + do { + n = read(release, &byte, 1); + } while (n < 0 && errno == EINTR); + } + + int getnameinfo(const struct sockaddr *addr, socklen_t addrlen, + char *host, socklen_t hostlen, + char *service, socklen_t servicelen, int flags) { + int (*real_getnameinfo)(const struct sockaddr *, socklen_t, + char *, socklen_t, char *, socklen_t, int) = + dlsym(RTLD_NEXT, "getnameinfo"); + if (getenv("DNS_BLOCK_REVERSE")) { + wait_for_release(); + if (!getenv("DNS_SUCCESS")) + return EAI_NONAME; + } + return real_getnameinfo(addr, addrlen, host, hostlen, + service, servicelen, flags); + } + + int getaddrinfo(const char *node, const char *service, + const struct addrinfo *hints, struct addrinfo **res) { + int (*real_getaddrinfo)(const char *, const char *, + const struct addrinfo *, + struct addrinfo **) = + dlsym(RTLD_NEXT, "getaddrinfo"); + if (node && strcmp(node, "uvloop-close-test.invalid") == 0) { + wait_for_release(); + if (getenv("DNS_SUCCESS")) { + int err = real_getaddrinfo("127.0.0.1", service, + hints, res); + if (err == 0) + completed_result = *res; + return err; + } + return EAI_NONAME; + } + return real_getaddrinfo(node, service, hints, res); + } + + void freeaddrinfo(struct addrinfo *res) { + void (*real_freeaddrinfo)(struct addrinfo *) = + dlsym(RTLD_NEXT, "freeaddrinfo"); + if (res && res == completed_result) { + completed_result = NULL; + notify('F'); + } + real_freeaddrinfo(res); + } + """ + + CHILD = """ + import asyncio + import concurrent.futures + import gc + import os + import sys + import weakref + + if sys.argv[1] == 'uvloop': + import uvloop + loop = uvloop.new_event_loop() + else: + loop = asyncio.new_event_loop() + + executor = concurrent.futures.ThreadPoolExecutor(max_workers=1) + loop.set_default_executor(executor) + started = int(sys.argv[2]) + os.set_blocking(started, False) + + expected_started = int(os.environ.get('DNS_STARTED_COUNT', '1')) + drop_loop = bool(os.environ.get('DNS_DROP_LOOP')) + + async def wait_for_dns(): + count = 0 + while True: + try: + byte = os.read(started, 1) + if byte != b'S': + raise RuntimeError('DNS worker handshake failed') + count += 1 + if count == expected_started: + return + except BlockingIOError: + await asyncio.sleep(0.01) + + def query(): + if sys.argv[4] == 'reverse': + return loop.getnameinfo(('127.0.0.1', 443)) + return loop.getaddrinfo('uvloop-close-test.invalid', 443) + + tasks = [loop.create_task(query()) + for _ in range(int(os.environ.get('DNS_REQUESTS', '1')))] + loop.run_until_complete(asyncio.wait_for(wait_for_dns(), 5)) + assert all(not task.done() for task in tasks) + if drop_loop: + for task in tasks: + task.cancel() + loop.run_until_complete(asyncio.gather( + *tasks, return_exceptions=True)) + del task + if int(os.environ.get('DNS_REQUESTS', '1')) > expected_started: + # Both libuv DNS workers are blocked. A second loop's DNS is + # therefore entirely queued, and should close without a thread. + other = uvloop.new_event_loop() + queued = other.create_task(other.getaddrinfo( + 'uvloop-close-test.invalid', 443)) + other.run_until_complete(asyncio.sleep(0)) + queued.cancel() + other.run_until_complete(asyncio.gather( + queued, return_exceptions=True)) + other.close() + assert other._closed_loop_thread is None + print('CLOSE_ENTER', flush=True) + loop.close() + loop.close() # Idempotent even while native cleanup is pending. + timestamp = loop.time() + assert loop.time() >= timestamp + print('CLOSE_RETURN', flush=True) + cleanup = (loop._closed_loop_thread + if sys.argv[1] == 'uvloop' else None) + if drop_loop: + ref = weakref.ref(loop) + tasks.clear() + del loop + gc.collect() + assert ref() is not None, 'native work lost its loop owner' + os.write(int(sys.argv[3]), b'C') + if sys.argv[1] == 'asyncio' and sys.version_info < (3, 10): + # Python 3.9's executor bridge delivers to closed loops. This + # control checks close() only; uvloop always checks completion. + os._exit(0) + # The parent releases DNS only after close() returns. Wait for the + # worker and its completion callback to catch late-delivery errors. + executor.shutdown(wait=True) + if sys.argv[1] == 'uvloop': + if cleanup is not None: + cleanup.join(5) + assert not cleanup.is_alive(), 'native cleanup did not finish' + if drop_loop: + gc.collect() + assert ref() is None, 'closed loop leaked after native cleanup' + if os.environ.get('DNS_SUCCESS') and sys.argv[4] != 'reverse': + assert os.read(started, 1) == b'F', 'DNS result was not freed' + if int(os.environ.get('DNS_REQUESTS', '1')) > expected_started: + try: + os.read(started, 1) + except BlockingIOError: + pass + else: + raise AssertionError('queued DNS work was not cancelled') + print('CLEANUP_RETURN', flush=True) + # Avoid pending-task diagnostics during interpreter shutdown. + os._exit(0) + """ + + @classmethod + def setUpClass(cls): + cc = shutil.which('cc') + if cc is None: + raise unittest.SkipTest('C compiler required for DNS shim') + tmp = tempfile.TemporaryDirectory(prefix='uvloop-dns-close-') + cls.addClassCleanup(tmp.cleanup) + source = pathlib.Path(tmp.name) / 'dns.c' + cls.shim = pathlib.Path(tmp.name) / 'dns.so' + source.write_text(textwrap.dedent(cls.SHIM)) + subprocess.run( + [cc, '-shared', '-fPIC', '-Wall', '-Werror', str(source), + '-o', str(cls.shim), '-ldl'], + check=True, capture_output=True, text=True, timeout=30) + + def check_close(self, implementation, reverse=False, *, + success=False, queued=False, drop_loop=False): + started_read, started_write = os.pipe() + release_read, release_write = os.pipe() + closed_read, closed_write = os.pipe() + try: + env = dict(os.environ, + DNS_STARTED_FD=str(started_write), + DNS_RELEASE_FD=str(release_read)) + env['LD_PRELOAD'] = str(self.shim) + env.pop('DNS_BLOCK_REVERSE', None) + env.pop('DNS_SUCCESS', None) + env.pop('DNS_DROP_LOOP', None) + env['UV_THREADPOOL_SIZE'] = '4' + env['DNS_REQUESTS'] = '3' if queued else '1' + env['DNS_STARTED_COUNT'] = '2' if queued else '1' + if success: + env['DNS_SUCCESS'] = '1' + if drop_loop: + env['DNS_DROP_LOOP'] = '1' + if reverse: + env['DNS_BLOCK_REVERSE'] = '1' + with subprocess.Popen( + [sys.executable, '-c', textwrap.dedent(self.CHILD), + implementation, str(started_read), str(closed_write), + 'reverse' if reverse else 'forward'], + env=env, + pass_fds=(started_read, started_write, release_read, + closed_write), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) as proc: + try: + ready, _, _ = select.select([closed_read], [], [], 10) + if not ready: + raise subprocess.TimeoutExpired(proc.args, 10) + self.assertEqual(os.read(closed_read, 1), b'C') + os.write(release_write, b'RR' if queued else b'R') + stdout, stderr = proc.communicate(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + stdout, stderr = proc.communicate() + self.assertIn('CLOSE_ENTER', stdout, (stdout, stderr)) + self.fail( + 'DNS shutdown or late completion timed out ' + f'request ({implementation})\n{stdout}{stderr}') + self.assertEqual(proc.returncode, 0, (stdout, stderr)) + self.assertIn('CLOSE_RETURN', stdout) + if implementation == 'uvloop': + self.assertIn('CLEANUP_RETURN', stdout) + self.assertEqual(stderr, '') + finally: + # Close every pipe, including when the watchdog kills the child. + for fd in (started_read, started_write, + release_read, release_write, + closed_read, closed_write): + os.close(fd) + + def test_uvloop_close_during_getaddrinfo(self): + self.check_close('uvloop') + + def test_asyncio_close_during_getaddrinfo(self): + self.check_close('asyncio') + + def test_uvloop_close_during_getnameinfo(self): + self.check_close('uvloop', reverse=True) + + @unittest.skipIf( + sys.version_info < (3, 11), + 'stdlib getnameinfo holds the GIL before Python 3.11') + def test_asyncio_close_during_getnameinfo(self): + self.check_close('asyncio', reverse=True) + + def test_uvloop_close_before_successful_getaddrinfo(self): + self.check_close('uvloop', success=True) + + def test_uvloop_close_before_successful_getnameinfo(self): + self.check_close('uvloop', reverse=True, success=True) + + def test_uvloop_close_cancels_queued_dns(self): + self.check_close('uvloop', queued=True) + + def test_uvloop_close_retains_loop_until_dns_finishes(self): + self.check_close('uvloop', drop_loop=True) + + def test_uvloop_close_retains_loop_until_reverse_dns_finishes(self): + self.check_close('uvloop', reverse=True, drop_loop=True) diff --git a/tests/test_process.py b/tests/test_process.py index 7b06e76d..02b6c5e5 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -435,6 +435,7 @@ async def main(): self.loop.run_until_complete(main()) + @tb.silence_long_exec_warning() def test_process_streams_redirect(self): async def test(): prog = bR''' diff --git a/tests/test_signals.py b/tests/test_signals.py index 7e8ed220..bf852129 100644 --- a/tests/test_signals.py +++ b/tests/test_signals.py @@ -235,6 +235,7 @@ async def runner(): import asyncio import signal import uvloop +from uvloop import _testbase as tb srv = None @@ -260,12 +261,14 @@ def handler_hup(): loop.add_signal_handler(signal.SIGUSR2, handler2) loop.add_signal_handler(signal.SIGHUP, handler_hup) loop.create_task(worker()) -try: - loop.run_forever() -finally: - srv.close() - loop.run_until_complete(srv.wait_closed()) - loop.close() +# Debug timing warnings must not become part of the signal-test output. +with tb.silence_long_exec_warning(): + try: + loop.run_forever() + finally: + srv.close() + loop.run_until_complete(srv.wait_closed()) + loop.close() """ diff --git a/uvloop/dns.pyx b/uvloop/dns.pyx index 67aeb595..7c8992d9 100644 --- a/uvloop/dns.pyx +++ b/uvloop/dns.pyx @@ -330,11 +330,34 @@ cdef class AddrInfo: return type(other) is AddrInfo +cdef void _dns_request_start(Loop loop, DNSRequestLink *link, + uv.uv_req_t *request) noexcept: + # Intrusive links avoid Python allocations and ownership cycles on the + # normal resolver path. Each UVRequest already retains itself until done. + link.request = request + link.prev = NULL + link.next = loop._dns_requests + if link.next is not NULL: + link.next.prev = link + loop._dns_requests = link + + +cdef void _dns_request_done(Loop loop, DNSRequestLink *link) noexcept: + if link.prev is NULL: + loop._dns_requests = link.next + else: + link.prev.next = link.next + if link.next is not NULL: + link.next.prev = link.prev + link.prev = link.next = NULL + + cdef class AddrInfoRequest(UVRequest): cdef: system.addrinfo hints object callback uv.uv_getaddrinfo_t _req_data + DNSRequestLink _link def __cinit__(self, Loop loop, bytes host, bytes port, @@ -371,6 +394,7 @@ cdef class AddrInfoRequest(UVRequest): self.callback = callback self.request.data = self + _dns_request_start(loop, &self._link, self.request) err = uv.uv_getaddrinfo(loop.uvloop, self.request, __on_addrinfo_resolved, @@ -379,6 +403,7 @@ cdef class AddrInfoRequest(UVRequest): &self.hints) if err < 0: + _dns_request_done(self.loop, &self._link) self.on_done() try: if err == uv.UV_EINVAL: @@ -397,6 +422,7 @@ cdef class NameInfoRequest(UVRequest): cdef: object callback uv.uv_getnameinfo_t _req_data + DNSRequestLink _link def __cinit__(self, Loop loop, callback): self.request = &self._req_data @@ -405,12 +431,14 @@ cdef class NameInfoRequest(UVRequest): cdef query(self, system.sockaddr *addr, int flags): cdef int err + _dns_request_start(self.loop, &self._link, self.request) err = uv.uv_getnameinfo(self.loop.uvloop, self.request, __on_nameinfo_resolved, addr, flags) if err < 0: + _dns_request_done(self.loop, &self._link) self.on_done() self.callback(convert_error(err)) @@ -440,6 +468,9 @@ cdef void __on_addrinfo_resolved( AddrInfo ai try: + if loop._closed: + uv.uv_freeaddrinfo(res) + return if status < 0: callback(convert_error(status)) else: @@ -451,6 +482,7 @@ cdef void __on_addrinfo_resolved( except BaseException as ex: loop._handle_exception(ex) finally: + _dns_request_done(loop, &request._link) request.on_done() @@ -466,6 +498,8 @@ cdef void __on_nameinfo_resolved( object callback = request.callback try: + if loop._closed: + return if status < 0: callback(convert_error(status)) else: @@ -476,4 +510,5 @@ cdef void __on_nameinfo_resolved( except BaseException as ex: loop._handle_exception(ex) finally: + _dns_request_done(loop, &request._link) request.on_done() diff --git a/uvloop/includes/uv.pxd b/uvloop/includes/uv.pxd index 510b1498..b2d5f7b4 100644 --- a/uvloop/includes/uv.pxd +++ b/uvloop/includes/uv.pxd @@ -288,6 +288,7 @@ cdef extern from "uv.h" nogil: uint64_t uv_now(const uv_loop_t*) int uv_run(uv_loop_t*, uv_run_mode mode) nogil + uint64_t uv_hrtime() void uv_stop(uv_loop_t*) # Idle handler diff --git a/uvloop/loop.pxd b/uvloop/loop.pxd index 89566ea3..cafcf79e 100644 --- a/uvloop/loop.pxd +++ b/uvloop/loop.pxd @@ -29,9 +29,18 @@ ctypedef object (*method2_t)(object, object, object) ctypedef object (*method3_t)(object, object, object, object) +cdef struct DNSRequestLink: + uv.uv_req_t *request + DNSRequestLink *prev + DNSRequestLink *next + + cdef class Loop: cdef: uv.uv_loop_t *uvloop + bint _uvloop_closed + DNSRequestLink *_dns_requests + readonly object _closed_loop_thread bint _coroutine_debug_set int _coroutine_origin_tracking_saved_depth diff --git a/uvloop/loop.pyx b/uvloop/loop.pyx index 802cdb69..a7495dab 100644 --- a/uvloop/loop.pyx +++ b/uvloop/loop.pyx @@ -137,6 +137,9 @@ cdef class Loop: self.slow_callback_duration = 0.1 self._closed = 0 + self._uvloop_closed = 0 + self._dns_requests = NULL + self._closed_loop_thread = None self._debug = 0 self._thread_id = 0 self._running = 0 @@ -230,6 +233,11 @@ cdef class Loop: if self._closed == 0: aio_logger.error("deallocating an open event loop") return + if not self._uvloop_closed: + # A failed native close must never free memory still referenced + # by libuv requests. Normally the cleanup thread retains us. + aio_logger.error("deallocating an event loop with pending cleanup") + return PyMem_RawFree(self.uvloop) self.uvloop = NULL @@ -551,6 +559,7 @@ cdef class Loop: cdef _close(self): cdef int err + cdef DNSRequestLink *dns if self._running == 1: raise RuntimeError("Cannot close a running event loop") @@ -560,6 +569,13 @@ cdef class Loop: self._closed = 1 + # Cancel DNS work still queued in libuv's pool. Executing requests + # return UV_EBUSY and are retained until their callbacks run. + dns = self._dns_requests + while dns is not NULL: + uv.uv_cancel(dns.request) + dns = dns.next + for cb_handle in self._ready: cb_handle.cancel() self._ready.clear() @@ -580,9 +596,10 @@ cdef class Loop: self.handler_check__exec_writes._close() __close_all_handles(self) self._shutdown_signals() - # During this run there should be no open handles, - # so it should finish right away - self.__run(uv.UV_RUN_DEFAULT) + # Flush handle close callbacks without waiting for native DNS. A + # second iteration also handles an outstanding uv_stop flag. + self.__run(uv.UV_RUN_NOWAIT) + self.__run(uv.UV_RUN_NOWAIT) if self._fd_to_writer_fileobj: for fileobj in self._fd_to_writer_fileobj.values(): @@ -609,8 +626,19 @@ cdef class Loop: f"{self._ready}") err = uv.uv_loop_close(self.uvloop) - if err < 0: + if err == uv.UV_EBUSY and self._dns_requests is not NULL: + # Executing DNS requests cannot be cancelled. Keep their loop + # alive and drain it independently of synchronous close(). DNS + # callbacks discard results when the Python loop is closed. + thread = threading_Thread( + target=self._drain_closed_loop, name='uvloop-close', + daemon=True) + self._closed_loop_thread = thread + thread.start() + elif err < 0: raise convert_error(err) + else: + self._uvloop_closed = 1 self.handler_async = None self.handler_idle = None @@ -622,6 +650,15 @@ cdef class Loop: self._default_executor = None executor.shutdown(wait=False) + def _drain_closed_loop(self): + # The thread's bound method retains self (and thus uvloop) until all + # native callbacks have run. No normal loop callbacks are delivered. + self.__run(uv.UV_RUN_DEFAULT) + cdef int err = uv.uv_loop_close(self.uvloop) + if err < 0: + raise convert_error(err) + self._uvloop_closed = 1 + cdef uint64_t _time(self): # asyncio doesn't have a time cache, neither should uvloop. uv.uv_update_time(self.uvloop) # void @@ -887,6 +924,8 @@ cdef class Loop: int proto, int flags, int unpack): + self._check_closed() + if isinstance(port, str): port = port.encode() elif isinstance(port, int): @@ -926,6 +965,7 @@ cdef class Loop: cdef _getnameinfo(self, system.sockaddr *addr, int flags): cdef NameInfoRequest nr + self._check_closed() fut = self._new_future() def callback(result): @@ -1337,6 +1377,9 @@ cdef class Loop: epoch, precision, accuracy and drift are unspecified and may differ per event loop. """ + if self._closed: + # A cleanup thread may own the native loop after close(). + return uv.uv_hrtime() / 1e9 return self._time() / 1000 def stop(self):