Make async lock initialization thread-safe

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)

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"):

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)