diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst index ebdf40a09..a51c6d6dd 100644 --- a/docs/source/changelog.rst +++ b/docs/source/changelog.rst @@ -6,6 +6,7 @@ Changelog Fixes +- Make async loop lock initialization thread-safe (#1783) - FTP: preserve filenames containing whitespace in _mlsd2 (#2043) - Prevent attribute error for 'forced' before flushing cache (#2042) - Reflect async _walk correctly (#2040) diff --git a/fsspec/asyn.py b/fsspec/asyn.py index 32ad3d35d..d896d6fe2 100644 --- a/fsspec/asyn.py +++ b/fsspec/asyn.py @@ -20,32 +20,22 @@ private = re.compile("_[^_]") iothread = [None] # dedicated fsspec IO thread loop = [None] # global event loop for any non-async instance -_lock = None # global lock placeholder +_lock = threading.Lock() get_running_loop = asyncio.get_running_loop def get_lock(): - """Allocate or return a threading lock. - - The lock is allocated on first use to allow setting one lock per forked process. - """ - global _lock - if not _lock: - _lock = threading.Lock() + """Return the process-local threading lock.""" return _lock def reset_lock(): - """Reset the global lock. - - This should be called only on the init of a forked process to reset the lock to - None, enabling the new forked process to get a new lock. - """ + """Reset the global loop and lock after forking.""" global _lock iothread[0] = None loop[0] = None - _lock = None + _lock = threading.Lock() async def _runner(event, coro, result, timeout=None): @@ -155,10 +145,7 @@ def get_loop(): def reset_after_fork(): - global lock - loop[0] = None - iothread[0] = None - lock = None + reset_lock() if hasattr(os, "register_at_fork"): diff --git a/fsspec/tests/test_async.py b/fsspec/tests/test_async.py index bd0593347..f0bf971ad 100644 --- a/fsspec/tests/test_async.py +++ b/fsspec/tests/test_async.py @@ -1,7 +1,10 @@ import asyncio import inspect import io +import threading import time +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace import pytest @@ -10,6 +13,27 @@ from fsspec.asyn import _run_coros_in_chunks +def test_get_lock_is_thread_safe(monkeypatch): + barrier = threading.Barrier(2) + real_threading = fsspec.asyn.threading + + def make_lock(): + candidate = threading.Lock() + barrier.wait() + return candidate + + fsspec.asyn.reset_lock() + try: + monkeypatch.setattr(fsspec.asyn, "threading", SimpleNamespace(Lock=make_lock)) + with ThreadPoolExecutor(max_workers=2) as executor: + locks = list(executor.map(lambda _: fsspec.asyn.get_lock(), range(2))) + finally: + monkeypatch.setattr(fsspec.asyn, "threading", real_threading) + fsspec.asyn.reset_lock() + + assert locks[0] is locks[1] + + def test_sync_methods(): inst = fsspec.asyn.AsyncFileSystem() assert inspect.iscoroutinefunction(inst._info)