From b3e6ac750e7b70d521703af8647df6765800a8f8 Mon Sep 17 00:00:00 2001 From: Darek Date: Wed, 7 Oct 2026 22:23:30 +0200 Subject: [PATCH] fix(pybind): load each import-cache entry once, under a lock The driver loads the Python objects it needs (datetime, uuid, pandas, ...) into a process-wide import cache on first use. PythonCachedItem::operator() checked and set its loaded flag without synchronization, and addToCache appended to a shared vector, relying on the GIL. On free-threaded CPython with the GIL disabled, threads making their first calls at the same time loaded entries concurrently: the vector reallocated under another thread's push_back and corrupted the heap, which crashed later in an unrelated allocation (SIGSEGV, or "free(): invalid pointer"). Each entry now loads once. The flag is atomic, and loading and addToCache run under a mutex that waiters take with the GIL released (detached on free-threaded builds), so the thread holding it can always finish its import; the parent entry is loaded before locking. Creating the cache in the Database constructor is serialized as well. PandasAnalyzer took a copy of the cached pandas entry, so it loaded into the copy and added to the cache on every call; it now uses the cache's own entry. This does not declare free-threading support; it removes this race. Co-Authored-By: Claude Opus 5.5 --- src_cpp/cached_import/py_cached_import.cpp | 7 +++ src_cpp/cached_import/py_cached_item.cpp | 16 ++--- .../include/cached_import/py_cached_import.h | 4 ++ .../include/cached_import/py_cached_item.h | 7 ++- src_cpp/pandas/pandas_analyzer.cpp | 2 +- src_cpp/py_database.cpp | 4 ++ test/test_import_cache_threading.py | 60 +++++++++++++++++++ 7 files changed, 89 insertions(+), 11 deletions(-) create mode 100644 test/test_import_cache_threading.py diff --git a/src_cpp/cached_import/py_cached_import.cpp b/src_cpp/cached_import/py_cached_import.cpp index 33a409d..1266ee6 100644 --- a/src_cpp/cached_import/py_cached_import.cpp +++ b/src_cpp/cached_import/py_cached_import.cpp @@ -11,6 +11,13 @@ PythonCachedImport::~PythonCachedImport() { allObjects.clear(); } +std::unique_lock PythonCachedImport::lockForLoading() { + // Wait with the GIL released (detached on free-threaded builds): the holder needs the GIL to + // finish its import, and a detached waiter does not stall stop-the-world pauses. + py::gil_scoped_release release; + return std::unique_lock(loadMutex); +} + py::handle PythonCachedImport::addToCache(py::object obj) { auto ptr = obj.ptr(); allObjects.push_back(obj); diff --git a/src_cpp/cached_import/py_cached_item.cpp b/src_cpp/cached_import/py_cached_item.cpp index ea16b3a..7385ff6 100644 --- a/src_cpp/cached_import/py_cached_item.cpp +++ b/src_cpp/cached_import/py_cached_item.cpp @@ -7,16 +7,18 @@ namespace lbug { py::handle PythonCachedItem::operator()() { assert((bool)PyGILState_Check()); - // load if unloaded, return cached object if already loaded - if (loaded) { + if (isLoaded()) { return object; } - if (parent == nullptr) { - object = importCache->addToCache(std::move(py::module::import(name.c_str()))); - } else { - object = importCache->addToCache(std::move((*parent)().attr(name.c_str()))); + // The parent takes the same lock, so load it before locking. + py::handle parentObject = parent == nullptr ? py::handle() : (*parent)(); + auto lock = importCache->lockForLoading(); + if (!isLoaded()) { + auto obj = parent == nullptr ? py::object(py::module::import(name.c_str())) : + py::object(parentObject.attr(name.c_str())); + object = importCache->addToCache(std::move(obj)); + loaded.store(true, std::memory_order_release); } - loaded = true; return object; } diff --git a/src_cpp/include/cached_import/py_cached_import.h b/src_cpp/include/cached_import/py_cached_import.h index 734550b..59ce8d4 100644 --- a/src_cpp/include/cached_import/py_cached_import.h +++ b/src_cpp/include/cached_import/py_cached_import.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include @@ -15,7 +16,9 @@ class PythonCachedImport { PythonCachedImport() = default; ~PythonCachedImport(); + // Call with the lock from lockForLoading() held. py::handle addToCache(py::object obj); + std::unique_lock lockForLoading(); DateTimeCachedItem datetime; DecimalCachedItem decimal; @@ -28,6 +31,7 @@ class PythonCachedImport { UUIDCachedItem uuid; private: + std::mutex loadMutex; std::vector allObjects; }; diff --git a/src_cpp/include/cached_import/py_cached_item.h b/src_cpp/include/cached_import/py_cached_item.h index 6840ebf..e00ce48 100644 --- a/src_cpp/include/cached_import/py_cached_item.h +++ b/src_cpp/include/cached_import/py_cached_item.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include @@ -10,16 +11,16 @@ namespace lbug { class PythonCachedItem { public: explicit PythonCachedItem(std::string name, PythonCachedItem* parent = nullptr) - : name(std::move(name)), parent(parent), loaded(false) {} + : name(std::move(name)), parent(parent) {} virtual ~PythonCachedItem() = default; - bool isLoaded() const { return loaded; } + bool isLoaded() const { return loaded.load(std::memory_order_acquire); } py::handle operator()(); protected: std::string name; PythonCachedItem* parent; - bool loaded; + std::atomic loaded{false}; py::handle object; }; diff --git a/src_cpp/pandas/pandas_analyzer.cpp b/src_cpp/pandas/pandas_analyzer.cpp index 04300c8..8f6707a 100644 --- a/src_cpp/pandas/pandas_analyzer.cpp +++ b/src_cpp/pandas/pandas_analyzer.cpp @@ -193,7 +193,7 @@ static py::object findFirstNonNull(const py::handle& row, uint64_t numRows) { common::LogicalType PandasAnalyzer::innerAnalyze(py::object column, bool& canConvert) { auto numRows = py::len(column); - auto pandasModule = importCache->pandas; + auto& pandasModule = importCache->pandas; auto pandasSeries = pandasModule.core.series.Series(); if (py::isinstance(column, pandasSeries)) { diff --git a/src_cpp/py_database.cpp b/src_cpp/py_database.cpp index 8a5c1ae..8281610 100644 --- a/src_cpp/py_database.cpp +++ b/src_cpp/py_database.cpp @@ -1,6 +1,7 @@ #include "include/py_database.h" #include +#include #include "common/exception/runtime.h" #include "extension/extension.h" @@ -65,6 +66,9 @@ PyDatabase::PyDatabase(const std::string& databasePath, uint64_t bufferPoolSize, lbug::extension::ExtensionUtils::addTableFunc(*state->database); state->storageDriver = std::make_unique(state->database.get()); py::gil_scoped_acquire acquire; + // Taken with the GIL held: nothing inside may call into Python. + static std::mutex importCacheMutex; + std::lock_guard lock(importCacheMutex); if (lbug::importCache.get() == nullptr) { lbug::importCache = std::make_shared(); } diff --git a/test/test_import_cache_threading.py b/test/test_import_cache_threading.py new file mode 100644 index 0000000..41e11fd --- /dev/null +++ b/test/test_import_cache_threading.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import os +import subprocess +import sys +import sysconfig +from textwrap import dedent +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from pathlib import Path + +# The driver fills its import cache on first use, and the cache cannot be reset within a +# process, so every attempt runs in a fresh interpreter. Only a free-threaded build with the GIL +# disabled can expose a race; on other builds this is a smoke test. +FREE_THREADED = bool(sysconfig.get_config_var("Py_GIL_DISABLED")) + + +@pytest.mark.parametrize("attempt", range(10 if FREE_THREADED else 1)) +def test_concurrent_first_use_of_the_import_cache( + attempt: int, build_dir: Path +) -> None: + code = dedent(f""" + import sys + import threading + + sys.path.append(r"{build_dir!s}") + + import ladybug as lb + + db = lb.Database(":memory:") + conns = [lb.Connection(db) for _ in range(16)] + start = threading.Barrier(len(conns)) + errors = [] + + def run(conn, n): + start.wait() + try: + assert conn.execute("RETURN $n", {{"n": n}}).get_next() == [n] + except BaseException as e: + errors.append(e) + + threads = [threading.Thread(target=run, args=(c, i)) for i, c in enumerate(conns)] + for t in threads: + t.start() + for t in threads: + t.join() + assert not errors, errors + """) + env = {**os.environ, "PYTHON_GIL": "0"} if FREE_THREADED else None + result = subprocess.run( + [sys.executable, "-c", code], + env=env, + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, (result.returncode, result.stderr)