From 3edb36bcfad22ce0961d7d60fdce57c8968b7eee Mon Sep 17 00:00:00 2001 From: Dsssyc Date: Thu, 29 Oct 2026 23:51:29 +0800 Subject: [PATCH 1/2] Release standalone native memory deterministically in the Python binding The Python binding never owned databases returned by the standalone native factories, so their layers, per-row caches and file backing stayed allocated. Each truncate/fill/save/close cycle on a 15,498,593-row F32 table retained about 361 MiB, and each load/read/close cycle about 302 MiB. Numeric column lookup also built per-row feature and geometry caches through tryGetFeature(0). Factories now return owning proxies. load_xbuffer takes its buffer export in the same C call that constructs the database; the native release callback drops the export, and closes a shared mapping, after the layers are destroyed. Chunk arrays set their chunk as NumPy base in C, and numeric columns use the table address and field offset directly. Tables share a backing state with their cached accessors and views, so nothing refers back to a table; tables keep their engine alive and the engine registers them weakly. RecordEngine.close() releases engine references, and later writes that need the engine raise a clear error. Core validates standalone envelopes, table bounds and legacy string tables before constructing layers. With cyclic GC disabled, both cycles stay flat after the first. Validation on macOS ARM64: 403 Python tests on CPython 3.10 and 3.13 wheels, compileall, uv build, 45 native ASan/UBSan tests, Rust and TypeScript/WASM tests, both 117-symbol ABI checks, and subprocess tests for interpreter shutdown and cyclic collection. See docs/reports/0.2.4-native-memory.md. --- docs/reports/0.2.4-native-memory.md | 128 ++++++++ fastcarto/fastdb/include/fastdb.h | 5 +- fastcarto/fastdb/src/FastVectorDb.cpp | 82 ++++- fastcarto/fastdb/src/FastVectorDbLayer.cpp | 8 + fastcarto/fastdb/src/FastVectorDbLayer_p.h | 1 + fastcarto/fastdb/src/FastVectorDb_p.h | 1 - fastcarto/fastdb/swig/fastdb4py.i | 144 +++++++-- python/README.md | 21 ++ python/fastdb4py/column_view.py | 12 +- python/fastdb4py/object_engine.py | 17 +- python/fastdb4py/orm/table.py | 189 +++++++---- python/fastdb4py/record_engine.py | 273 +++++++++------- python/fastdb4py/serializer.py | 14 +- python/fastdb4py/string_column.py | 46 +-- tests/cpp/CMakeLists.txt | 21 ++ tests/cpp/test_legacy_lifetime.cpp | 191 +++++++++++ tests/python/test_native_lifetime.py | 342 ++++++++++++++++++++ tests/python/test_native_shutdown.py | 210 ++++++++++++ tests/python/test_object_engine_lifetime.py | 37 +++ tests/python/test_push.py | 9 +- tests/python/test_reader.py | 7 - tests/python/test_record_engine.py | 4 +- tests/python/test_string_column.py | 16 +- tests/python/test_table_ownership_cache.py | 166 ++++++++++ tools/probe_standalone_memory.py | 109 +++++++ 25 files changed, 1767 insertions(+), 286 deletions(-) create mode 100644 docs/reports/0.2.4-native-memory.md create mode 100644 tests/cpp/test_legacy_lifetime.cpp create mode 100644 tests/python/test_native_lifetime.py create mode 100644 tests/python/test_native_shutdown.py create mode 100644 tests/python/test_object_engine_lifetime.py create mode 100644 tests/python/test_table_ownership_cache.py create mode 100644 tools/probe_standalone_memory.py diff --git a/docs/reports/0.2.4-native-memory.md b/docs/reports/0.2.4-native-memory.md new file mode 100644 index 0000000..ec38d7b --- /dev/null +++ b/docs/reports/0.2.4-native-memory.md @@ -0,0 +1,128 @@ +# Standalone native memory ownership fix for 0.2.4 + +## Root causes + +The Python projection did not own newly allocated databases returned by the +native factories. Their destructors never released layers, cached features or +file backing. Numeric column lookup also requested row zero through +`tryGetFeature`, allocating a feature-pointer vector and a geometry-pointer +vector for every row. Bulk columns now use the table base, field offset and +row stride directly. + +`Table` and its cached column accessor retained each other. That cycle kept +one complete backing allocation alive per read/write cycle until cyclic GC. +Build-only references also retained native builders after finalization. + +## Lifetime design + +The buffer loader acquires an independent memoryview and `Py_buffer` in the +same C call that constructs the native database. Its cookie holds that export +and, on successful shared-memory loading, the mapping owner. The real native +destructor first destroys its layers and features; its callback then releases +the export and closes the mapping. Failed loads release the export and use one factory error; +high-level shared-memory loaders close mappings they opened on failure. + +Factory results are owning SWIG proxies. Borrowed layers, features, references +and chunks retain their parents. Chunk arrays set their actual chunk as base +with `PyArray_SetBaseObject` in C; the owner argument is supplied internally by +SWIG and cannot be overridden by a caller. Numeric column arrays retain their +layer through the array interface. Raw arrays, slices, rows, tables and column +accessors keep backing alive until their last reference is released. + +Tables strongly cache their column accessors and arrays. These projections +share a backing-state object that never references a table, accessor or view. +The state and write callbacks retain their RecordEngine, which maintains one +registry of weak state and table references with direct lookup by name. +Registrations survive while a state is retained, so string views still follow +remaps after the table object is released. Write callbacks receive the +backing state as an argument. Snapshot remapping updates shared state and +replaces the table's accessor cache; retained checked accessors reject the old generation, including +fields first accessed after remapping. There is no ownership cycle. + +Builders and push dispatch caches are discarded when their build phase ends. +Mutable-build states contain native builders and row counts, without engine +references; they are separate from the weak registry of public tables. +`ObjectEngine` holds no duplicate backing reference and relies on reference +counting. + +`RecordEngine.close()` is idempotent and drops engine-owned references. It does not +invalidate a caller's `FdbViewOwner`; callers use `fdb.invalidate` when a lease +ends. Engine-dependent writes after closure raise `RuntimeError` with +`RecordEngine is closed.` Direct writable array access remains valid. +Retained checked views still enforce owner validity. Buffer resize or +mapping close raises `BufferError` while views pin its export. These contracts +are documented in the [Python binding README](../../python/README.md). + +Unused unsafe borrowing interfaces are excluded from the Python projection. +Invalid standalone envelope, descriptor, table and legacy string-table bounds +are checked in Core before constructing layers. Optional column-section +handling retains its existing behavior. + +## Memory measurements + +Measurements use macOS ARM64, CPython 3.10.17 and NumPy 2.2.6. The published +0.2.3 wheel and a locally built 0.2.4 wheel run the same scripts. Each large case +has 15,498,593 F32 rows, six complete write or read cycles, and cyclic GC +disabled throughout. Closing and deleting the last reference releases native +objects immediately; no `gc.collect()` is used in these measurements. + +| Path | Published 0.2.3 | 0.2.4 | +| --- | ---: | ---: | +| truncate, fill, save, close, delete | 361.18 MiB/cycle | 0.00 MiB/cycle | +| file load, column copy, close, delete | 301.50 MiB/cycle | 0.00 MiB/cycle | + +Values are mean increments over cycles 2–6, measured using macOS physical +footprint and rounded to 0.1 MiB by the script. All six 0.2.4 write samples +are 257.5 MiB; all six read samples are 257.1 MiB. There is no observed +post-warm-up growth at this resolution. First-cycle allocator +warm-up is allowed; stable reuse is the criterion, rather than an immediate +return to the import baseline. No cyclic collection or allocator relief occurs +between samples. + +Automated memory regressions use 1,000,000 rows and ten cycles with GC disabled. +After two warm-up cycles, all remaining samples must stay within 16 MiB above +that baseline. The 0.2.3 native cache and backing leaks alone retain about +16–20 MiB per cycle at this row count. Separate weak-reference and mapping assertions check +immediate release independently of allocator RSS behavior. + +## Conversion timings + +The same benchmark takes the best of five runs using the same Python/NumPy +versions and data. Timing is reported here; regression tests use deterministic +cache-identity and weak-reference assertions rather than timing thresholds. + +| Operation | Published 0.2.3 | 0.2.4 | +| --- | ---: | ---: | +| `StringColumn.to_pylist`, 200,000 rows | 747.6 ms | 723.3 ms | +| `chunk.as_array`, 200,000 calls | 71.7 ms | 70.1 ms | +| `table.column.value`, 100,000 reads | 66.4 ms | 54.2 ms | +| `table[i].value`, 100,000 reads | 321.4 ms | 318.8 ms | + +Column accessors and their arrays remain cached per table. Each measured path +has latency at or below the published version in this run; repeated runs +vary by a few percent in either direction. + +## Verification scope + +The regression suite covers caller-owned bytes, bytearray and mmap exports, +shared mappings, invalid inputs, retained accessors/rows/columns, shared checked +owners, and immediate builder release. Shutdown subprocesses cover daemon +readers, module destructors and cyclic destructors; mapping closure is asserted +on native destruction and on the cyclic case. Native ASan/UBSan tests revoke +backing access with `PROT_NONE` before destruction and check release callbacks +exactly once. Tests write file fixtures only in their assigned temporary or +build directory. + +The source suite passes 403 Python tests on CPython 3.13; installed wheels +each pass all 403 tests on CPython 3.10 and 3.13. Compileall and local +sdist/wheel builds also pass, together with +45 native ASan/UBSan tests, 21 Rust unit/integration tests plus one compile-fail +doctest, 58 TypeScript/WASM tests and 117-symbol native and WASM ABI checks. +The three shutdown scripts each return zero twice in both Python runtimes. +The six retained-view/release scenarios and chained writes pass on CPython +3.10. Release policy tests pass 105 cases and 98 subtests. +Windows, Linux, macOS x86_64 and free-threaded Python have +not been executed for this fix. Registry artifacts and the complete release +matrix remain subject to the [publishing process](../releases/publishing.md). + +Raw logs, measurements and build archives are kept outside the source repository. diff --git a/fastcarto/fastdb/include/fastdb.h b/fastcarto/fastdb/include/fastdb.h index 7d0d1c7..83c77a6 100644 --- a/fastcarto/fastdb/include/fastdb.h +++ b/fastcarto/fastdb/include/fastdb.h @@ -224,7 +224,7 @@ namespace wx void setTableBufferMaterialized(bool enabled); size_t tableBufferBytes(); void setField(unsigned ix, const FastVectorDbFeatureRef* ref); - FastVectorDbFeatureRef* createFeatureRef(u32 ix=-1); + FastVectorDbFeatureRef* createFeatureRef(u32 ix=0xFFFFFFFF); void freeFeatureRef(FastVectorDbFeatureRef* ref); void addFeatureEnd(); // List column build API @@ -370,6 +370,9 @@ namespace wx const char* getFieldDefn(unsigned ix, FieldTypeEnum &ft, double &vmin, double &vmax); size_t getFieldOffset(unsigned ix); size_t getFeatureByteSize(); + // Borrowed base of the row table. Rows are separated by getFeatureByteSize(). + // The address remains valid only while the database and its backing buffer live. + void* getTableDataAddress(); void getExtent(double &minx, double &miny, double &maxx, double &maxy); u32 getFeatureCount(); void rewind(); diff --git a/fastcarto/fastdb/src/FastVectorDb.cpp b/fastcarto/fastdb/src/FastVectorDb.cpp index bf519e8..8c18277 100644 --- a/fastcarto/fastdb/src/FastVectorDb.cpp +++ b/fastcarto/fastdb/src/FastVectorDb.cpp @@ -1,6 +1,7 @@ #include "FastVectorDb_p.h" #include "FastVectorDbLayer_p.h" #include +#include #include #ifdef _WIN32 @@ -24,15 +25,61 @@ namespace wx std::memcpy(&value, source, sizeof(value)); return value; } + + bool valid_layer_bounds(const u8* layer, const layer_header_t& header) + { + const size_t tail = size_t(header.total_size) - sizeof(layer_header_t); + if (header.field_count > tail / sizeof(field_desc_ex_t)) + return false; + const size_t descriptors = size_t(header.field_count) * sizeof(field_desc_ex_t); + const u8* data = layer + sizeof(layer_header_t) + descriptors; + const size_t size = tail - descriptors; + if (header.offset_table > size) + return false; + if (header.field_count) { + const auto last = load_unaligned( + layer + sizeof(layer_header_t) + descriptors - sizeof(field_desc_ex_t)); + if (last.offset > std::numeric_limits::max() - last.size) + return false; + const u64 stride = last.offset + last.size; + if (stride && header.feature_count > (size - header.offset_table) / stride) + return false; + for (u16 i = 0; i < header.field_count; ++i) { + const auto field = load_unaligned( + layer + sizeof(layer_header_t) + size_t(i) * sizeof(field_desc_ex_t)); + if (field.offset > stride || field.size > stride - field.offset) + return false; + } + } + // Legacy string tables are read during construction, before the + // bounded optional column sections. Check their terminators here. + for (unsigned width : {1u, 2u}) { + u64 offset = width == 1 ? header.offset_strings : header.offset_wstrings; + if (offset > size || size - offset < sizeof(u32)) + return false; + const u32 count = load_unaligned(data + offset); + offset += sizeof(u32); + if (count > (size - offset) / width) + return false; + for (u32 i = 0; i < count; ++i) { + for (;;) { + if (size - offset < width) + return false; + const bool terminated = width == 1 ? data[offset] == 0 : + load_unaligned(data + offset) == 0; + offset += width; + if (terminated) break; + } + } + } + return true; + } } FastVectorDb::Impl::Impl(void *pdata, size_t size, fnFreeDbBuffer fnFreeBuffer, void *cookie) : m_pdata(pdata), m_size(size), m_fnFreeBuffer(fnFreeBuffer), m_cookie(cookie) { u8 *ptr = (u8 *)pdata; - bool check_mask = strcmp((const char *)ptr, "FASTVectorDB0.1") == 0; - assert(check_mask); - m_mask_check_ok=check_mask; ptr += 16; const u32 count = load_unaligned(ptr); ptr += sizeof(u32); @@ -113,17 +160,32 @@ namespace wx FastVectorDb *FastVectorDb::load(void *pdata, size_t size, fnFreeDbBuffer fnFreeBuffer, void *cookie) { - FastVectorDb::Impl *impl = new FastVectorDb::Impl(pdata, size, fnFreeBuffer, cookie); - if(impl->m_mask_check_ok) - { - auto db = new FastVectorDb(impl); - return db; + // Validate the standalone envelope once before reading its headers. + // The backing callback also releases invalid inputs. + bool valid = pdata && size >= 20 && + std::memcmp(pdata, "FASTVectorDB0.1", sizeof("FASTVectorDB0.1")) == 0; + if (valid) { + const u8* ptr = static_cast(pdata) + 20; + size_t remaining = size - 20; + const u32 count = load_unaligned(static_cast(pdata) + 16); + for (u32 i = 0; i < count; ++i) { + if (remaining < sizeof(layer_header_t)) { valid = false; break; } + const auto header = load_unaligned(ptr); + if (header.total_size < sizeof(layer_header_t) || header.total_size > remaining || + !valid_layer_bounds(ptr, header)) { + valid = false; + break; + } + ptr += header.total_size; + remaining -= header.total_size; + } } - else + if (!valid) { - delete impl; + if (fnFreeBuffer) fnFreeBuffer(pdata, size, cookie); return nullptr; } + return new FastVectorDb(new Impl(pdata, size, fnFreeBuffer, cookie)); } FastVectorDbFeature* FastVectorDb::tryGetFeature(wx::FastVectorDbFeatureRef* ref) diff --git a/fastcarto/fastdb/src/FastVectorDbLayer.cpp b/fastcarto/fastdb/src/FastVectorDbLayer.cpp index 5262421..2cd934a 100644 --- a/fastcarto/fastdb/src/FastVectorDbLayer.cpp +++ b/fastcarto/fastdb/src/FastVectorDbLayer.cpp @@ -727,6 +727,10 @@ namespace wx { return m_table_line_size; } + void* FastVectorDbLayer::Impl::getTableDataAddress() + { + return m_table_data_ptr0; + } /////////////////////////////////////////// FastVectorDbLayer::FastVectorDbLayer(Impl *_impl) : impl(_impl) {} @@ -842,6 +846,10 @@ namespace wx { return impl->getFeatureByteSize(); } + void* FastVectorDbLayer::getTableDataAddress() + { + return impl->getTableDataAddress(); + } // ---- List column read helpers ---- chunk_data_t FastVectorDbLayer::Impl::getFieldAsListView_internal(u32 ifeature, u32 ix) diff --git a/fastcarto/fastdb/src/FastVectorDbLayer_p.h b/fastcarto/fastdb/src/FastVectorDbLayer_p.h index 76d635d..f89bab8 100644 --- a/fastcarto/fastdb/src/FastVectorDbLayer_p.h +++ b/fastcarto/fastdb/src/FastVectorDbLayer_p.h @@ -65,6 +65,7 @@ namespace wx size_t getFieldOffset(unsigned ix); size_t getFeatureByteSize(); + void* getTableDataAddress(); private: void move_next_geometry_ptr(); size_t get_geometry_like_size(const u8* pdata); diff --git a/fastcarto/fastdb/src/FastVectorDb_p.h b/fastcarto/fastdb/src/FastVectorDb_p.h index 15817c7..200a010 100644 --- a/fastcarto/fastdb/src/FastVectorDb_p.h +++ b/fastcarto/fastdb/src/FastVectorDb_p.h @@ -20,7 +20,6 @@ namespace wx{ size_t m_size; fnFreeDbBuffer m_fnFreeBuffer; void* m_cookie; - bool m_mask_check_ok; friend class FastVectorDb; }; } diff --git a/fastcarto/fastdb/swig/fastdb4py.i b/fastcarto/fastdb/swig/fastdb4py.i index f6f8b9d..21913f1 100644 --- a/fastcarto/fastdb/swig/fastdb4py.i +++ b/fastcarto/fastdb/swig/fastdb4py.i @@ -4,11 +4,75 @@ #include "fastdb.h" #include "fastdb-geometry-utils.h" using namespace wx; + + struct PythonDbBacking { + Py_buffer buffer; + PyObject* mapping; + }; + + static void release_python_db_backing(void*, size_t, void* cookie) { + auto* backing = static_cast(cookie); + PyGILState_STATE state = PyGILState_Ensure(); + PyBuffer_Release(&backing->buffer); + if (backing->mapping) { + PyObject* result = PyObject_CallMethod(backing->mapping, "close", NULL); + if (result) + Py_DECREF(result); + else + PyErr_WriteUnraisable(backing->mapping); + Py_DECREF(backing->mapping); + } + PyMem_Free(backing); + PyGILState_Release(state); + } %} // Tell SWIG to ignore fastdb_api when parsing headers #define fastdb_api +// These factories transfer a newly allocated object to their caller. Layers, +// cached features and builder-created refs remain owned by their parent. +%newobject wx::FastVectorDb::load(const char *filename); +%newobject wx::FastVectorDb::load_python_buffer; +%rename(load_xbuffer) wx::FastVectorDb::load_python_buffer; +%newobject wx::FastVectorDbFeatureRef::make_ref; + +%define BORROWED_RESULT(method) +%pythonappend method %{ + if val is not None: + val._fdb_parent = self +%} +%enddef +BORROWED_RESULT(wx::FastVectorDbBuild::createLayerBegin) +BORROWED_RESULT(wx::FastVectorDbLayerBuild::createFeatureRef) +BORROWED_RESULT(wx::FastVectorDb::getLayer) +BORROWED_RESULT(wx::FastVectorDb::tryGetFeature) +BORROWED_RESULT(wx::FastVectorDb::buffer) +BORROWED_RESULT(wx::FastVectorDbLayer::tryGetFeatureAt) +BORROWED_RESULT(wx::FastVectorDbLayer::getGeometryLikeChunk) +BORROWED_RESULT(wx::FastVectorDbLayer::getFieldAsStringView) +BORROWED_RESULT(wx::FastVectorDbLayer::getStringColumnOffsets) +BORROWED_RESULT(wx::FastVectorDbLayer::getStringColumnData) +BORROWED_RESULT(wx::FastVectorDbLayer::getFieldAsFeatureRef) +BORROWED_RESULT(wx::FastVectorDbFeature::layer) +BORROWED_RESULT(wx::FastVectorDbFeature::getGeometryLikeChunk) +BORROWED_RESULT(wx::FastVectorDbFeature::getFieldAsStringView) +BORROWED_RESULT(wx::FastVectorDbFeature::getFieldAsListView) +BORROWED_RESULT(wx::FastVectorDbFeature::getFieldAsFeatureRef) +BORROWED_RESULT(wx::FastVectorDbFeature::getFieldListRefAt) +BORROWED_RESULT(wx::MemoryStream::data) + +%define DATABASE_RESULT(method) +%exception method { + $action + if (PyErr_Occurred()) SWIG_fail; + if (!result) + SWIG_exception_fail(SWIG_ValueError, "Invalid FastDB database buffer."); +} +%enddef +DATABASE_RESULT(wx::FastVectorDb::load) +DATABASE_RESULT(wx::FastVectorDb::load_python_buffer) + // Release the GIL for pure C++ operations that do not touch Python objects. // This benefits both standard Python (better multi-threaded throughput) and // free-threaded Python 3.13+ (PEP 703). @@ -32,7 +96,33 @@ // When built on free-threaded Python, SWIG automatically declares Py_MOD_GIL_NOT_USED. %} -//%array_class(point2_t, LineString); +%extend wx::FastVectorDb { + static wx::FastVectorDb* load_python_buffer(PyObject* pdata, PyObject* mapping = Py_None) { + // Independent view: callers may release their view after this call. + PyObject* view = PyMemoryView_FromObject(pdata); + if (!view) return NULL; + auto* backing = static_cast(PyMem_Calloc(1, sizeof(PythonDbBacking))); + if (!backing) { + Py_DECREF(view); + PyErr_NoMemory(); + return NULL; + } + int status = PyObject_GetBuffer(view, &backing->buffer, PyBUF_SIMPLE); + Py_DECREF(view); + if (status != 0) { + PyMem_Free(backing); + return NULL; + } + wx::FastVectorDb* db = wx::FastVectorDb::load( + backing->buffer.buf, (size_t)backing->buffer.len, + release_python_db_backing, backing); + if (db && mapping != Py_None) { + Py_INCREF(mapping); + backing->mapping = mapping; + } + return db; + } +} %typemap(in) const wchar_t * (wchar_t *temp = NULL) { if ($input == Py_None) { @@ -101,26 +191,19 @@ } %typemap(default) (const u8* data,size_t size) {}; -// 在你的 .i 文件中对目标函数应用此类型映射 + +// Copying stream writes borrow their input only for this C call. Retained +// database buffers use load_python_buffer's native lifetime cookie instead. %typemap(in) (void* pdata, size_t size) (Py_buffer view) { - // 检查输入对象是否支持缓冲区接口 - if (PyObject_CheckBuffer($input) == 0) { - SWIG_exception_fail(SWIG_TypeError, "Expected a bufferable object (e.g., bytes, bytearray)"); - } - // 获取缓冲区视图，PyBUF_SIMPLE 请求一个简单的连续字节缓冲区 - if (PyObject_GetBuffer($input, &view, PyBUF_SIMPLE) != 0) { - SWIG_exception_fail(SWIG_TypeError, "Failed to get buffer from object."); + memset(&view, 0, sizeof(view)); + if (PyObject_GetBuffer($input, &view, PyBUF_SIMPLE) < 0) { + SWIG_fail; } - // 将获取到的缓冲区地址和长度赋给参数 - $1 = view.buf; // void* pdata 指向数据 - $2 = view.len; // size_t size 设置数据长度 + $1 = view.buf; + $2 = (size_t)view.len; } - -// 定义 freearg 类型映射，在函数调用后释放缓冲区 %typemap(freearg) (void* pdata, size_t size) { - if (view$argnum.obj) { - PyBuffer_Release(&view$argnum); - } + if (view$argnum.obj) PyBuffer_Release(&view$argnum); } // Typemap for set_field_list_numeric: accepts any buffer object; nbytes = byte length of buffer @@ -191,6 +274,13 @@ %ignore wx::FastVectorDbFeature::FastVectorDbFeature(); %ignore wx::FastVectorDbFeature::~FastVectorDbFeature(); %ignore wx::FastVectorDb::load(void *pdata, size_t size, fnFreeDbBuffer fnFreeBuffer, void *cookie); +%ignore wx::FastVectorDb::load_xbuffer; +%ignore wx::FixedBufferWriteStream::FixedBufferWriteStream; +%nodefaultctor wx::FixedBufferWriteStream; +%ignore wx::aabbox_t::minEdge; +%ignore wx::aabbox_t::maxEdge; +%ignore wx::GeometryUtils::get_point_at; +%ignore wx::GeometryUtils::part_data; %ignore wx::FastVectorDbBuild::postToBuffer; %ignore wx::FastVectorDbLayerBuild::setNumericColumnBulk; %nodefaultctor FastVectorDbLayerBuild; @@ -228,6 +318,7 @@ %rename(get_field_defn) getFieldDefn_p; %rename(get_extent) getExtent_p; %rename(get_feature_count) getFeatureCount; +%rename(get_table_data_address) getTableDataAddress; %rename(get_geometry_like_chunk) getGeometryLikeChunk; %rename(get_field_as_float) getFieldAsFloat; %rename(get_field_as_int) getFieldAsInt; @@ -261,11 +352,10 @@ %rename(get_field_list_ref_at) getFieldListRefAt; %extend wx::chunk_data_t { - PyObject *as_array(PyObject* npType) { + PyObject *as_array(PyObject* npType, PyObject** PYTHON_SELF) { int typenum; PyArray_Descr* descr = PyArray_DescrFromTypeObject(npType); if (descr == NULL) { - PyErr_Clear(); // 或进行其他错误处理 return NULL; } else { typenum = descr->type_num; @@ -281,21 +371,22 @@ Py_DECREF(descr); - npy_intp num_elements = $self->size / item_size; - - if ($self->size % item_size != 0) { + if (item_size == 0 || $self->size % item_size != 0) { PyErr_SetString(PyExc_ValueError, "Buffer size does not match an integer number of elements for the given data type."); return NULL; } + npy_intp num_elements = $self->size / item_size; npy_intp dims[1] = {num_elements}; PyObject *array = PyArray_SimpleNewFromData(1, dims, typenum, (void*)($self->pdata)); if (!array) { PyErr_SetString(PyExc_RuntimeError, "Failed to create NumPy array"); return NULL; } - PyArray_ENABLEFLAGS((PyArrayObject*)array, NPY_ARRAY_C_CONTIGUOUS); - PyArray_ENABLEFLAGS((PyArrayObject*)array, NPY_ARRAY_WRITEABLE); - PyArray_CLEARFLAGS((PyArrayObject*)array, NPY_ARRAY_OWNDATA); + Py_INCREF(*PYTHON_SELF); + if (PyArray_SetBaseObject((PyArrayObject*)array, *PYTHON_SELF) < 0) { + Py_DECREF(array); + return NULL; + } return array; } @@ -333,6 +424,7 @@ %pythoncode %{ import numpy as np + %} %extend wx::FastVectorDbLayer { %pythoncode %{ @@ -379,7 +471,7 @@ if(not tystr): return None - ptr = _fastdb4py.get_swig_ptr_as_long(table.tryGetFeature(0).get_address()) + ptr = _fastdb4py.get_swig_ptr_as_long(table.get_table_data_address()) offset = table.get_field_offset(index) ptr+=offset return __column_np_interface__( diff --git a/python/README.md b/python/README.md index c79060f..c8ba15f 100644 --- a/python/README.md +++ b/python/README.md @@ -129,6 +129,27 @@ print(table.column.x.mean()) ## Backed View Lifetimes +`RecordEngine.close()` releases the engine's own references and may be called +repeatedly. Tables retain their engine, and cached column accessors and views +share backing state that does not reference the table. Their last reference +releases the native database immediately, without waiting for cyclic garbage +collection. `ObjectEngine` releases its references when the engine itself is +no longer referenced; it has no `close()` method. Closing a RecordEngine does +not invalidate a caller's `FdbViewOwner`; the caller ends a lease with +`fdb.invalidate(owner_or_view)`. + +After `RecordEngine.close()`, writes that require the engine, including +`Table.fill()` and string-column fills, raise `RuntimeError: RecordEngine is +closed.` Direct writes to retained writable NumPy views still update the +memory those views retain. + +`core.WxDatabase.load_xbuffer(buffer)` pins a contiguous buffer export until +the native database is destroyed. This includes borrowed `bytes`, `bytearray`, +NumPy and `mmap` storage. Resizing or closing an exported buffer raises +`BufferError` while a table, row, or column still keeps the database alive. +Shared-memory mappings close after native layers and the buffer export are +released. + FastDB Python values have two lifetime modes. A plain `@feature` instance is owned Python data: its fields live in `__dict__`, it can be kept indefinitely, and assigning `point.x = 1.0` is ordinary Python object mutation. A row returned from a mapped `Table` is a backed feature view: schema fields read from the native table row, scalar numeric writes go back to native storage when the view is writeable, and the view is tied to a `FdbViewOwner`. Standalone FastDB tables use a trusted unchecked owner by default, so existing high-performance workflows keep raw NumPy column access: diff --git a/python/fastdb4py/column_view.py b/python/fastdb4py/column_view.py index f7b44cd..211bf38 100644 --- a/python/fastdb4py/column_view.py +++ b/python/fastdb4py/column_view.py @@ -7,20 +7,20 @@ from .view_owner import FdbViewOwner, FdbViewWriteError if TYPE_CHECKING: - from .orm.table import Table + from .orm.table import _TableState class NumericColumnView: """Owner-checked view over a numeric native column.""" - def __init__(self, table: 'Table', field_index: int, field_name: str, array: np.ndarray): - self._table = table + def __init__(self, state: '_TableState', field_index: int, field_name: str, array: np.ndarray): + self._state = state self._field_index = field_index self._field_name = field_name self._array = array - self._owner: FdbViewOwner = table._fdb_owner - self._writeable = table._fdb_writeable - self._owner_generation = table._fdb_owner.generation + self._owner: FdbViewOwner = state._fdb_owner + self._writeable = state._fdb_writeable + self._owner_generation = state._fdb_owner.generation def _checked_array(self) -> np.ndarray: self._owner.assert_alive(self._owner_generation) diff --git a/python/fastdb4py/object_engine.py b/python/fastdb4py/object_engine.py index 4eec18e..dc8a9e4 100644 --- a/python/fastdb4py/object_engine.py +++ b/python/fastdb4py/object_engine.py @@ -132,7 +132,6 @@ class ObjectEngine: def __init__(self): self._db_build: Optional[core.WxDatabaseBuild] = None self._db: Optional[core.WxDatabase] = None - self._buffer: Optional[bytes] = None self._layers: Dict[Type, LayerState] = {} self._layer_order: List[Type] = [] self._built = False @@ -170,10 +169,10 @@ def truncate(cls, layouts) -> 'ObjectEngine': engine._db_build.post(mem) buf = mem.data().as_array(np.uint8).tobytes() engine._db = core.WxDatabase.load_xbuffer(buf) - engine._db._buffer = buf - engine._buffer = buf engine._built = True engine._db_build = None + for state in engine._layers.values(): + state.build = None return engine def push(self, obj: Any) -> None: @@ -237,10 +236,10 @@ def combine(self): self._db_build.post(mem) buf = mem.data().as_array(np.uint8).tobytes() self._db = core.WxDatabase.load_xbuffer(buf) - self._db._buffer = buf - self._buffer = buf self._built = True self._db_build = None + for state in self._layers.values(): + state.build = None self._pending.clear() self._pushed_ids.clear() self._pending_counts.clear() @@ -393,9 +392,9 @@ def share(self, name: str): if not self._built: raise RuntimeError("Call combine() before sharing") import multiprocessing.shared_memory as shm_mod - data = self._buffer - seg = shm_mod.SharedMemory(name=name, create=True, size=len(data)) - seg.buf[:len(data)] = data + chunk = self._db.buffer() + seg = shm_mod.SharedMemory(name=name, create=True, size=chunk.size) + chunk.copy_to_buffer(seg.buf) seg.close() @classmethod @@ -408,8 +407,6 @@ def load(cls, name: str) -> 'ObjectEngine': orm = cls() orm._db = core.WxDatabase.load_xbuffer(buf) - orm._db._buffer = buf - orm._buffer = buf orm._built = True for i in range(orm._db.get_layer_count()): diff --git a/python/fastdb4py/orm/table.py b/python/fastdb4py/orm/table.py index 0792941..01ac91b 100644 --- a/python/fastdb4py/orm/table.py +++ b/python/fastdb4py/orm/table.py @@ -70,7 +70,7 @@ def _normalize_bool_fill_ndarray(field_name: str, values: np.ndarray) -> np.ndar ) -def _create_column_accessor(feature_type: Type[T], table) -> T: +def _create_column_accessor(feature_type: Type[T], state) -> T: """ Create a column accessor that provides numpy array access with proper type hints. @@ -81,7 +81,7 @@ def _create_column_accessor(feature_type: Type[T], table) -> T: schema = get_schema(feature_type) ColumnAccessorClass = schema.column_accessor_class if ColumnAccessorClass is not None: - return ColumnAccessorClass(table, feature_type) + return ColumnAccessorClass(state) # Get original annotations from feature_type original_annotations = getattr(feature_type, '__annotations__', {}).copy() @@ -96,10 +96,10 @@ class ColumnAccessor: # Set the new annotations __annotations__ = original_annotations - def __init__(self, table, feature_type): - # Don't call parent __init__ to avoid initializing cache - # Just set internal references - object.__setattr__(self, '_table', table) + def __init__(self, state): + object.__setattr__(self, '_state', state) + object.__setattr__(self, '_fdb_owner', state._fdb_owner) + object.__setattr__(self, '_owner_generation', state._fdb_owner.generation) object.__setattr__(self, '_field_index_map', _field_index_map) object.__setattr__(self, '_name_cache', {}) object.__setattr__(self, '_cache_lock', Lock()) @@ -108,8 +108,9 @@ def __getattr__(self, name: str) -> np.ndarray: """Override to return numpy array instead of single value""" # Hot path: name → array directly (1× getattribute + 1× dict.get). # Cached arrays are stable: fixed-scale tables never reallocate columns. - table = object.__getattribute__(self, '_table') - table._assert_alive() + owner = object.__getattribute__(self, '_fdb_owner') + if owner.checked: + owner.assert_alive(object.__getattribute__(self, '_owner_generation')) name_cache = object.__getattribute__(self, '_name_cache') arr = name_cache.get(name) if arr is not None: @@ -127,12 +128,13 @@ def __getattr__(self, name: str) -> np.ndarray: if idx is None: raise AttributeError(f'Field "{name}" not found in the table.') - table_origin = table._origin + state = object.__getattribute__(self, '_state') + table_origin = state._origin fd = schema.fields[idx] if fd.field_type in {OriginFieldType.str, OriginFieldType.wstr}: - arr = StringColumn(table, idx, name, wide=fd.field_type == OriginFieldType.wstr) + arr = StringColumn(state, idx, name, wide=fd.field_type == OriginFieldType.wstr) elif fd.field_type == OriginFieldType.bytes: - arr = BytesColumn(table, idx, name) + arr = BytesColumn(state, idx, name) else: column = table_origin.get_column(idx) if column is None: @@ -140,8 +142,8 @@ def __getattr__(self, name: str) -> np.ndarray: f'Field "{name}" does not expose a fixed native column.', ) arr = column.as_nparray() - if table._fdb_owner.checked or not table._fdb_writeable: - arr = NumericColumnView(table, idx, name, arr) + if state._fdb_owner.checked or not state._fdb_writeable: + arr = NumericColumnView(state, idx, name, arr) name_cache[name] = arr return arr @@ -157,21 +159,40 @@ def __setattr__(self, name: str, value): # Store the class in ClassSchema (replaces _column_accessor_cache WeakKeyDict). schema.column_accessor_class = ColumnAccessor - return ColumnAccessor(table, feature_type) + return ColumnAccessor(state) + + +class _TableState: + """Backing shared by table projections, with no references to projections.""" -class Table(Generic[T]): def __init__(self): self._fc: np.ndarray = np.zeros(1, dtype=np.int64) - self._column: T | None = None self._feature_type: Type[T] | None = None self._db: core.WxDatabase | core.WxDatabaseBuild = None self._origin: core.WxLayerTable | core.WxLayerTableBuild | None = None + self._engine = None + self._table_name = '' self._fixed_fill_handler = None self._read_lock = RLock() self._fdb_owner: FdbViewOwner = trusted_view_owner(writeable=True) - self._fdb_checked = False self._fdb_writeable = True + def __len__(self) -> int: + self._assert_alive() + return self._origin.get_feature_count() + + def _assert_alive(self, generation: int | None = None) -> None: + self._fdb_owner.assert_alive(generation) + + def _assert_engine_open(self) -> None: + if self._engine is not None: + self._engine._assert_open() + + def _remap(self, origin, db) -> None: + self._db = db + self._origin = origin + self.feature_count = origin.get_feature_count() + @property def feature_count(self) -> int: return int(self._fc[0]) @@ -180,41 +201,110 @@ def feature_count(self) -> int: def feature_count(self, value: int) -> None: self._fc[0] = value + +class Table(Generic[T]): + def __init__(self): + self._state = _TableState() + self._column: T | None = None + + @staticmethod + def _from_state(state: _TableState) -> 'Table[T]': + table = Table[T]() + table._state = state + table._reset_columns() + return table + + def _reset_columns(self) -> None: + self._column = ( + _create_column_accessor(self._feature_type, self._state) + if self.fixed and self._feature_type is not None else None + ) + + @property + def _db(self): + return self._state._db + + @property + def _origin(self): + return self._state._origin + + @_origin.setter + def _origin(self, value): + self._state._origin = value + + @property + def _feature_type(self): + return self._state._feature_type + + @_feature_type.setter + def _feature_type(self, value): + self._state._feature_type = value + + @property + def _read_lock(self): + return self._state._read_lock + + @_read_lock.setter + def _read_lock(self, value): + self._state._read_lock = value + + @property + def _fdb_owner(self): + return self._state._fdb_owner + + @property + def _fdb_writeable(self): + return self._state._fdb_writeable + + @property + def _fixed_fill_handler(self): + return self._state._fixed_fill_handler + + @property + def feature_count(self) -> int: + return self._state.feature_count + + @feature_count.setter + def feature_count(self, value: int) -> None: + self._state.feature_count = value + def __len__(self) -> int: self._assert_alive() return self._origin.get_feature_count() def __getitem__(self, index: int) -> T: - with self._read_lock: - self._assert_alive() - count = self._origin.get_feature_count() + state = self._state + with state._read_lock: + state._assert_alive() + count = state._origin.get_feature_count() if index < 0: index = count + index if index < 0 or index >= count: raise IndexError(f'Feature index {index} out of range [0, {count}).') return bind_feature( - self._feature_type, - self._db, - self._origin, + state._feature_type, + state._db, + state._origin, index, - owner=self._fdb_owner, - writeable=self._fdb_writeable, + owner=state._fdb_owner, + writeable=state._fdb_writeable, table=self, ) def __iter__(self) -> Generator[T, None, None]: self._assert_alive() for i in range(self._origin.get_feature_count()): - with self._read_lock: - self._assert_alive() + state = self._state + with state._read_lock: + state._assert_alive() item = bind_feature( - self._feature_type, - self._db, - self._origin, + state._feature_type, + state._db, + state._origin, i, - owner=self._fdb_owner, - writeable=self._fdb_writeable, + owner=state._fdb_owner, + writeable=state._fdb_writeable, table=self, ) yield item @@ -287,7 +377,7 @@ def map_from( writeable: bool | None = None, ) -> 'Table[T]': table = Table[T]() - table._db = db + table._state._db = db table._origin = origin table._feature_type = feature_type table._set_view_owner(owner, writeable=writeable) @@ -295,8 +385,6 @@ def map_from( # Get feature count if the fastdb table has fixed scale if table.fixed: table.feature_count = origin.get_feature_count() - # Create column accessor that pretends to be T but returns numpy arrays - table._column = _create_column_accessor(feature_type, table) if feature_type is not None else None return table @@ -305,17 +393,8 @@ def _remap( origin: core.WxLayerTable | core.WxLayerTableBuild, db: core.WxDatabase | core.WxDatabaseBuild, ) -> None: - self._db = db - self._origin = origin - self._fdb_owner.bump_generation() - if self.fixed: - self.feature_count = origin.get_feature_count() - self._column = ( - _create_column_accessor(self._feature_type, self) - if self._feature_type is not None else None - ) - else: - self._column = None + self._state._remap(origin, db) + self._reset_columns() def rewind(self): self._assert_alive() @@ -328,6 +407,7 @@ def fill(self, **col_arrays) -> None: ``tbl.column.`` is stale and must not be reused. Always re-fetch ``tbl.column.`` after fill(). """ + self._state._assert_engine_open() if not self.fixed: raise RuntimeError('fill() only supports fixed-scale tables.') self._assert_writeable() @@ -338,7 +418,7 @@ def fill(self, **col_arrays) -> None: expected = len(self) writes = {} - col = self._column + col = self.column schema = get_schema(self._feature_type) for field_name, values in col_arrays.items(): field_def = schema.get(field_name) @@ -375,7 +455,7 @@ def fill(self, **col_arrays) -> None: ) writes[field_name] = arr - self._fixed_fill_handler(writes) + self._fixed_fill_handler(self._state, writes) def _set_view_owner( self, @@ -385,17 +465,12 @@ def _set_view_owner( ) -> None: if owner is None: owner = trusted_view_owner(writeable=True if writeable is None else writeable) - self._fdb_owner = owner - self._fdb_checked = owner.checked - self._fdb_writeable = owner.writeable if writeable is None else writeable - if self.fixed: - self._column = ( - _create_column_accessor(self._feature_type, self) - if self._feature_type is not None else None - ) + self._state._fdb_owner = owner + self._state._fdb_writeable = owner.writeable if writeable is None else writeable + self._reset_columns() def _assert_alive(self, generation: int | None = None) -> None: - self._fdb_owner.assert_alive(generation) + self._state._fdb_owner.assert_alive(generation) def _assert_writeable(self, generation: int | None = None) -> None: self._fdb_owner.assert_alive(generation) diff --git a/python/fastdb4py/record_engine.py b/python/fastdb4py/record_engine.py index e6e9a18..cb0bb42 100644 --- a/python/fastdb4py/record_engine.py +++ b/python/fastdb4py/record_engine.py @@ -6,6 +6,7 @@ from pathlib import Path from typing import List, TypeVar, Type from multiprocessing import shared_memory +from weakref import ref from . import core from .registry import ( @@ -16,7 +17,7 @@ raw_payload_storage_diagnostics, ) from .layout import Layout -from .orm.table import Table +from .orm.table import Table, _TableState from .type import OriginFieldType from .view_owner import FdbViewOwner from .push_compiler import ( @@ -28,6 +29,14 @@ T = TypeVar('T') +def _load_shared_database(shm: shared_memory.SharedMemory): + try: + return core.WxDatabase.load_xbuffer(shm.buf, shm) + except BaseException: + shm.close() + raise + + def _post_build_to_bytearray(build: object) -> bytearray: size = build.byte_length() buffer = bytearray(size) @@ -95,10 +104,28 @@ def _reject_unsupported_schema(schema: LayerSchema, cls_name: str) -> None: _reject_non_native_lists(schema, cls_name) +class _TableRegistration: + """Weak projections of one backing state, retained until that state dies.""" + + def __init__(self, engine, key, table: Table): + engine_ref = ref(engine) + + def forget_state(state_ref): + parent = engine_ref() + if parent is not None: + entry = parent._tables.get(key) + if entry is not None and entry.state is state_ref: + del parent._tables[key] + + self.state = ref(table._state, forget_state) + self.table = ref(table) + + class RecordEngine: def __init__(self): self._shm: shared_memory.SharedMemory | None = None - self._table_map: dict[str, Table] = {} + self._tables: dict[tuple[str, int], _TableRegistration] = {} + self._build_states: dict[str, _TableState] = {} self._table_feature_types: dict[str, Type] = {} self._origin: core.WxDatabase | core.WxDatabaseBuild | None = None self._is_mutable: bool = False @@ -111,6 +138,11 @@ def __init__(self): self._push_batch_fn: dict = {} self._push_dispatch: dict = {} self._bool_push_schemas: dict = {} + self._closed = False + + def _assert_open(self) -> None: + if self._closed: + raise RuntimeError('RecordEngine is closed.') @property def fixed(self) -> bool: @@ -155,7 +187,7 @@ def _prepare_truncate( table_name = layout.name or ft_cls.__name__ engine._table_feature_types[table_name] = ft_cls - existing = engine._table_map.get(table_name) + existing = engine._fixed_layer_builds.get(table_name) if existing is not None: warnings.warn( f'Table "{table_name}" already exists, truncate ' @@ -163,22 +195,17 @@ def _prepare_truncate( UserWarning, ) else: - new_table = Table.map_from( - ft_cls, - _get_default_table_build(engine._fixed_build, table_name), - engine._fixed_build, - ) + layer = _get_default_table_build(engine._fixed_build, table_name) field_ids = {} for fd in schema.fields: field_ids[fd.name] = len(field_ids) if fd.field_type == OriginFieldType.list: - new_table._origin.add_list_field(fd.name, fd.cpp_type) + layer.add_list_field(fd.name, fd.cpp_type) else: - new_table._origin.add_field(fd.name, fd.field_type.value) + layer.add_field(fd.name, fd.field_type.value) if not materialize_table_buffer: - new_table._origin._set_table_buffer_materialized(False) - engine._table_map[table_name] = new_table - engine._fixed_layer_builds[table_name] = new_table._origin + layer._set_table_buffer_materialized(False) + engine._fixed_layer_builds[table_name] = layer engine._fixed_table_fields[table_name] = field_ids engine._fixed_build.truncate(table_name, layout.capacity) @@ -202,6 +229,7 @@ def truncate(layouts: List[Layout]) -> 'RecordEngine': def push(self, feature, table_name: str = '') -> None: """Push a single @feature instance to the database.""" + self._assert_open() cls = feature.__class__ # Fast path: batchable buffer buf = self._push_buf.get(cls) @@ -242,25 +270,7 @@ def _push_full(self, feature, table_name: str = '') -> None: cache = normalize_bool_cache(feature.__dict__, schema) feat_table_name = table_name if table_name else feature_type.__name__ - t_obj: Table = self._table_map.get(feat_table_name) - if t_obj is None: - new_table = Table.map_from( - feature_type, - _get_default_table_build( - self._origin, - feat_table_name, - raw_payload=bool(schema.bytes_plan), - ), - self._origin, - ) - for fd in schema.fields: - if fd.field_type == OriginFieldType.list: - new_table._origin.add_list_field(fd.name, fd.cpp_type) - else: - new_table._origin.add_field(fd.name, fd.field_type.value) - self._table_map[feat_table_name] = new_table - self._table_feature_types[feat_table_name] = feature_type - t_obj = new_table + t_obj = self._get_build_state(feature_type, feat_table_name, schema) # Ensure compiled push_fn exists if schema.push_fn is None: @@ -294,6 +304,7 @@ def _push_full(self, feature, table_name: str = '') -> None: def push_many(self, features: list, table_name: str = '') -> None: """Push multiple @feature instances of the same type efficiently.""" + self._assert_open() if not self._is_mutable or not features: return feature_type = features[0].__class__ @@ -320,25 +331,7 @@ def push_many(self, features: list, table_name: str = '') -> None: for feat in features ] feat_table_name = table_name if table_name else feature_type.__name__ - t_obj: Table = self._table_map.get(feat_table_name) - if t_obj is None: - new_table = Table.map_from( - feature_type, - _get_default_table_build( - self._origin, - feat_table_name, - raw_payload=bool(schema.bytes_plan), - ), - self._origin, - ) - for fd in schema.fields: - if fd.field_type == OriginFieldType.list: - new_table._origin.add_list_field(fd.name, fd.cpp_type) - else: - new_table._origin.add_field(fd.name, fd.field_type.value) - self._table_map[feat_table_name] = new_table - self._table_feature_types[feat_table_name] = feature_type - t_obj = new_table + t_obj = self._get_build_state(feature_type, feat_table_name, schema) t_origin = t_obj._origin fc = t_obj.feature_count @@ -352,6 +345,22 @@ def push_many(self, features: list, table_name: str = '') -> None: fc += 1 t_obj.feature_count = fc + def _get_build_state(self, feature_type: Type, name: str, schema: LayerSchema) -> _TableState: + state = self._build_states.get(name) + if state is None: + state = _TableState() + state._db = self._origin + state._origin = _get_default_table_build(self._origin, name, raw_payload=bool(schema.bytes_plan)) + state._feature_type = feature_type + for fd in schema.fields: + if fd.field_type == OriginFieldType.list: + state._origin.add_list_field(fd.name, fd.cpp_type) + else: + state._origin.add_field(fd.name, fd.field_type.value) + self._build_states[name] = state + self._table_feature_types[name] = feature_type + return state + def _flush_push_batches(self): """Flush all pending batch push buffers to C++.""" for cls, buf in self._push_buf.items(): @@ -365,6 +374,7 @@ def _flush_push_batches(self): def combine(self): """Combine memory from all tables into a single continuous block.""" + self._assert_open() if self._origin is None: warnings.warn('Database is empty, cannot combine.', UserWarning) return @@ -376,12 +386,15 @@ def combine(self): buffer = _post_build_to_bytearray(self._origin) self._origin = core.WxDatabase.load_xbuffer(buffer) - self._origin._buffer = buffer - if self._shm: - self._shm.close() - self._shm = None - self._table_map = {} + self._shm = None + self._tables.clear() + self._build_states.clear() + self._is_mutable = False + self._push_buf.clear() + self._push_batch_fn.clear() + self._push_dispatch.clear() + self._bool_push_schemas.clear() def table( self, @@ -392,28 +405,33 @@ def table( writeable: bool | None = None, ) -> Table: """Get a Table for the given @feature class.""" + self._assert_open() if not is_feature(feature_type): raise TypeError( f'{feature_type.__name__} is not a @feature class. ' f'Use @feature decorator.' ) table_name = name or feature_type.__name__ - cached = self._table_map.get(table_name) - if cached is not None: - self._table_feature_types[table_name] = feature_type - if owner is not None or writeable is not None: - tbl = Table.map_from( - feature_type, - cached._origin, - cached._db, - owner=owner, - writeable=writeable, - ) - self._attach_owner_fixed_fill_handler(tbl, table_name) - return tbl - self._attach_fixed_fill_handler(cached, table_name) - return cached + alias = owner is not None or writeable is not None + if not alias: + entry = self._tables.get((table_name, 0)) + if entry is not None: + state = entry.state() + if state is not None: + cached = entry.table() + if cached is None: + cached = Table._from_state(state) + entry.table = ref(cached) + return cached db = self._origin + if isinstance(db, core.WxDatabaseBuild): + state = self._build_states.get(table_name) + if state is None: + raise KeyError(f'Table "{table_name}" not found') + tbl = Table.map_from(feature_type, state._origin, db, owner=owner, writeable=writeable) + tbl._state._fc = state._fc + self._register_table(tbl, table_name, alias) + return tbl fallback_names = list(self._table_feature_types) for i in range(db.get_layer_count()): layer = db.get_layer(i) @@ -427,46 +445,38 @@ def table( writeable=writeable, ) self._table_feature_types[table_name] = feature_type - if owner is not None or writeable is not None: - self._attach_owner_fixed_fill_handler(tbl, table_name) - return tbl - self._attach_fixed_fill_handler(tbl, table_name) - self._table_map[table_name] = tbl + self._register_table(tbl, table_name, alias) return tbl raise KeyError(f'Table "{table_name}" not found') - def _attach_owner_fixed_fill_handler(self, table: Table, table_name: str) -> None: - if table.fixed and self._fixed_direct_fill: - self._attach_fixed_fill_handler(table, table_name) - return - if table.fixed and self._fixed_build is not None and self._shm is None: - table._fixed_fill_handler = ( - lambda writes, _table=table: self._fill_mapped_table_and_drop_fixed_build(_table, writes) - ) - return - self._attach_fixed_fill_handler(table, table_name) - - def _attach_fixed_fill_handler(self, table: Table, table_name: str) -> None: - if table.fixed and self._fixed_direct_fill: - table._fixed_fill_handler = ( - lambda writes, _table=table: self._fill_mapped_table_in_place(_table, writes) - ) - elif table.fixed and self._fixed_build is not None and self._shm is None: - table._fixed_fill_handler = ( - lambda writes: self._fill_fixed_table(table_name, writes) - ) + def _register_table(self, table: Table, name: str, alias: bool) -> None: + key = (name, id(table._state) if alias else 0) + self._tables[key] = _TableRegistration(self, key, table) + table._state._engine = self + self._attach_fixed_fill_handler(table._state, name, alias) + + def _attach_fixed_fill_handler(self, state: _TableState, table_name: str, alias: bool = False) -> None: + state._table_name = table_name + fixed = isinstance(state._origin, core.WxLayerTable) + if fixed and self._fixed_direct_fill: + state._fixed_fill_handler = self._fill_mapped_table_in_place + elif fixed and self._fixed_build is not None and self._shm is None: + state._fixed_fill_handler = self._fill_mapped_table_and_drop_fixed_build if alias else self._fill_fixed_table else: - table._fixed_fill_handler = None + state._fixed_fill_handler = None def _invalidate_fixed_writes(self) -> None: self._fixed_build = None self._fixed_layer_builds = {} self._fixed_table_fields = {} self._fixed_direct_fill = False - for table_name, table in self._table_map.items(): - self._attach_fixed_fill_handler(table, table_name) + for (table_name, alias_id), entry in list(self._tables.items()): + state = entry.state() + if state is not None: + self._attach_fixed_fill_handler(state, table_name, bool(alias_id)) - def _fill_fixed_table(self, table_name: str, writes: dict[str, object]) -> None: + def _fill_fixed_table(self, state: _TableState, writes: dict[str, object]) -> None: + table_name = state._table_name layer_build = self._fixed_layer_builds[table_name] field_ids = self._fixed_table_fields[table_name] try: @@ -520,26 +530,38 @@ def _publish_fixed_snapshot_to_buffer(self, buffer: object, *, direct_fill: bool self._publish_fixed_buffer(writable[:size], direct_fill=direct_fill) def _publish_fixed_buffer(self, buffer: object, *, direct_fill: bool) -> None: + previous_db = self._origin self._fixed_direct_fill = direct_fill self._origin = core.WxDatabase.load_xbuffer(buffer) - self._origin._buffer = buffer - for table_name, table in list(self._table_map.items()): + remaps = [] + for (table_name, alias_id), entry in list(self._tables.items()): + state = entry.state() + if state is None or state._db is not previous_db: + continue new_layer = self._find_layer(table_name) if new_layer is None: continue - table._remap(new_layer, self._origin) - self._attach_fixed_fill_handler(table, table_name) + remaps.append((state, entry.table(), new_layer, table_name, bool(alias_id))) + for owner in {state._fdb_owner for state, *_ in remaps}: + owner.bump_generation() + for state, table, layer, table_name, alias in remaps: + if table is None: + state._remap(layer, self._origin) + else: + table._remap(layer, self._origin) + self._attach_fixed_fill_handler(state, table_name, alias) - def _fill_mapped_table_in_place(self, table: Table, writes: dict[str, object]) -> None: + def _fill_mapped_table_in_place(self, state: _TableState, writes: dict[str, object]) -> None: + schema = get_schema(state._feature_type) for field_name, payload in writes.items(): if isinstance(payload, _StringSequencePayload) or isinstance(payload, tuple): raise RuntimeError('direct fixed backing writes do not support variable-size string payloads.') - column = getattr(table.column, field_name) + column = state._origin.get_column(schema.field_index_map[field_name]).as_nparray() column[:] = payload - def _fill_mapped_table_and_drop_fixed_build(self, table: Table, writes: dict[str, object]) -> None: - self._fill_mapped_table_in_place(table, writes) + def _fill_mapped_table_and_drop_fixed_build(self, state: _TableState, writes: dict[str, object]) -> None: + self._fill_mapped_table_in_place(state, writes) self._invalidate_fixed_writes() # ------------------------------------------------------------------ @@ -560,13 +582,14 @@ def load(name: str, from_file: bool = False) -> 'RecordEngine': name = _normalize_shm_name(name) try: engine._shm = shared_memory.SharedMemory(name=name) - engine._origin = core.WxDatabase.load_xbuffer(engine._shm.buf) + engine._origin = _load_shared_database(engine._shm) except FileNotFoundError: raise FileNotFoundError(f"Database '{name}' not found in shared memory.") return engine def share(self, shm_name: str, close_after: bool = False): """Share the database in shared memory.""" + self._assert_open() if self._origin is None: raise RuntimeError('Database is empty, cannot share.') if isinstance(self._origin, core.WxDatabaseBuild): @@ -575,17 +598,17 @@ def share(self, shm_name: str, close_after: bool = False): shm_name = _normalize_shm_name(shm_name) chunk = self._origin.buffer() self._shm = shared_memory.SharedMemory(create=True, size=chunk.size, name=shm_name) - dest = np.ndarray(chunk.size, dtype=np.uint8, buffer=self._shm.buf) - dest[:] = chunk.as_array(np.uint8) - self._origin._buffer = None - self._origin = core.WxDatabase.load_xbuffer(self._shm.buf) + chunk.copy_to_buffer(self._shm.buf) + self._origin = _load_shared_database(self._shm) self._invalidate_fixed_writes() + self._tables.clear() if close_after and platform.system() != 'Windows': self.close() def save(self, path: str): """Save the database to a file.""" + self._assert_open() if self._origin is None: raise RuntimeError('Database is empty, cannot save.') if isinstance(self._origin, core.WxDatabaseBuild): @@ -596,18 +619,28 @@ def save(self, path: str): f.write(chunk.to_bytes()) def close(self): - """Close the database and release resources.""" - if self._shm: - self._shm.close() - self._shm = None - self._origin = None + """Release engine references; existing views retain their native backing.""" + self._closed = True + self._tables.clear() + self._build_states.clear() + self._table_feature_types.clear() + self._origin = None + self._shm = None + self._fixed_build = None + self._fixed_layer_builds.clear() + self._fixed_table_fields.clear() + self._fixed_direct_fill = False + self._is_mutable = False + self._push_buf.clear() + self._push_batch_fn.clear() + self._push_dispatch.clear() + self._bool_push_schemas.clear() def unlink(self): """Unlink the shared memory database.""" if self._shm: self._shm.unlink() - self._shm = None - self._origin = None + self.close() def __len__(self): if self._origin is None: diff --git a/python/fastdb4py/serializer.py b/python/fastdb4py/serializer.py index 5322c84..83b310f 100644 --- a/python/fastdb4py/serializer.py +++ b/python/fastdb4py/serializer.py @@ -294,6 +294,7 @@ def loads_shm(shm_name: str, length: int, offset: int, root_type: type): from multiprocessing import shared_memory as _shm_mod shm = _shm_mod.SharedMemory(name=shm_name, create=False) + db = None try: if offset + length > shm.size: raise ValueError( @@ -301,13 +302,9 @@ def loads_shm(shm_name: str, length: int, offset: int, root_type: type): f"exceeds shared memory size ({shm.size})" ) - # Create a memoryview slice and hand it to the C++ loader. - # load_xbuffer's SWIG typemap acquires/releases its own Py_buffer, - # so we can release *buf* immediately after the call. The C++ db - # holds a raw pointer into the still-alive mmap (owned by *shm*). - buf = shm.buf[offset:offset + length] - db = core.WxDatabase.load_xbuffer(buf) - buf.release() + # Native destruction releases the export before closing the mapping. + with shm.buf[offset:offset + length] as buf: + db = core.WxDatabase.load_xbuffer(buf, shm) if db.get_layer_count() == 0: return None @@ -335,7 +332,8 @@ def loads_shm(shm_name: str, length: int, offset: int, root_type: type): return result finally: - shm.close() + if db is None: + shm.close() # --- Internal Helpers --- diff --git a/python/fastdb4py/string_column.py b/python/fastdb4py/string_column.py index 2166d36..5abecbc 100644 --- a/python/fastdb4py/string_column.py +++ b/python/fastdb4py/string_column.py @@ -7,7 +7,7 @@ from .view_owner import FdbViewOwner, FdbViewWriteError if TYPE_CHECKING: - from .orm.table import Table + from .orm.table import _TableState _UINT32_MAX = np.iinfo(np.uint32).max @@ -69,14 +69,14 @@ def _normalize_string_values( class StringColumn: - def __init__(self, table: 'Table', field_index: int, field_name: str, *, wide: bool = False): - self._table = table + def __init__(self, state: '_TableState', field_index: int, field_name: str, *, wide: bool = False): + self._state = state self._field_index = field_index self._field_name = field_name self._wide = wide - self._owner: FdbViewOwner = table._fdb_owner - self._writeable = table._fdb_writeable - self._owner_generation = table._fdb_owner.generation + self._owner: FdbViewOwner = state._fdb_owner + self._writeable = state._fdb_writeable + self._owner_generation = state._fdb_owner.generation def _assert_alive(self) -> None: self._owner.assert_alive(self._owner_generation) @@ -89,18 +89,18 @@ def _assert_writeable(self) -> None: def __len__(self) -> int: self._assert_alive() - return len(self._table) + return len(self._state) def _offsets(self) -> np.ndarray | None: self._assert_alive() - chunk = self._table._origin.get_string_column_offsets(self._field_index) + chunk = self._state._origin.get_string_column_offsets(self._field_index) if chunk is None or chunk.size == 0: return None return chunk.as_array(np.uint32) def _data(self) -> np.ndarray | None: self._assert_alive() - chunk = self._table._origin.get_string_column_data(self._field_index) + chunk = self._state._origin.get_string_column_data(self._field_index) if chunk is None or chunk.size == 0: return np.empty(0, dtype=np.uint8) return chunk.as_array(np.uint8) @@ -120,8 +120,8 @@ def get(self, index: int) -> str: end = int(offsets[index + 1]) return bytes(data[start:end]).decode('utf-8') - with self._table._read_lock: - feat = self._table._origin.tryGetFeature(index) + with self._state._read_lock: + feat = self._state._origin.tryGetFeature(index) if self._wide: return feat.get_field_as_wstring(self._field_index) or '' raw = feat.get_field_as_string_view(self._field_index) @@ -145,6 +145,7 @@ def to_owned(self) -> list[str]: return self.to_pylist() def fill(self, strings: Iterable[Optional[str]]) -> None: + self._state._assert_engine_open() self._assert_writeable() payload = _normalize_string_values(strings, len(self)) self._dispatch_string_payload(payload) @@ -158,9 +159,9 @@ def _normalize_fill_values( return self._validate_utf8_payload(offsets_arr, data_arr, expected_len) def _dispatch_string_payload(self, payload: _StringSequencePayload) -> None: - fill_handler = self._table._fixed_fill_handler + fill_handler = self._state._fixed_fill_handler if fill_handler is not None: - fill_handler({self._field_name: payload}) + fill_handler(self._state, {self._field_name: payload}) return raise RuntimeError( @@ -194,6 +195,7 @@ def _validate_utf8_payload( return offsets_arr, data_arr def fill_utf8(self, offsets: np.ndarray, data: np.ndarray) -> None: + self._state._assert_engine_open() self._assert_writeable() offsets_arr, data_arr = self._validate_utf8_payload(offsets, data, len(self)) self._dispatch_utf8_payload(offsets_arr, data_arr) @@ -203,9 +205,9 @@ def _dispatch_utf8_payload( offsets_arr: np.ndarray, data_arr: np.ndarray, ) -> None: - fill_handler = self._table._fixed_fill_handler + fill_handler = self._state._fixed_fill_handler if fill_handler is not None: - fill_handler({self._field_name: (offsets_arr, data_arr)}) + fill_handler(self._state, {self._field_name: (offsets_arr, data_arr)}) return raise RuntimeError( @@ -215,19 +217,19 @@ def _dispatch_utf8_payload( class BytesColumn: - def __init__(self, table: 'Table', field_index: int, field_name: str): - self._table = table + def __init__(self, state: '_TableState', field_index: int, field_name: str): + self._state = state self._field_index = field_index self._field_name = field_name - self._owner: FdbViewOwner = table._fdb_owner - self._owner_generation = table._fdb_owner.generation + self._owner: FdbViewOwner = state._fdb_owner + self._owner_generation = state._fdb_owner.generation def _assert_alive(self) -> None: self._owner.assert_alive(self._owner_generation) def __len__(self) -> int: self._assert_alive() - return len(self._table) + return len(self._state) def get(self, index: int) -> bytes: self._assert_alive() @@ -237,8 +239,8 @@ def get(self, index: int) -> bytes: if index < 0 or index >= length: raise IndexError(f'Bytes index {index} out of range [0, {length}).') - with self._table._read_lock: - feat = self._table._origin.tryGetFeature(index) + with self._state._read_lock: + feat = self._state._origin.tryGetFeature(index) raw = feat.get_geometry_like_chunk() if raw is None: return b'' diff --git a/tests/cpp/CMakeLists.txt b/tests/cpp/CMakeLists.txt index 518df91..593dace 100644 --- a/tests/cpp/CMakeLists.txt +++ b/tests/cpp/CMakeLists.txt @@ -85,6 +85,27 @@ add_test( COMMAND fastdb_legacy_descriptor_determinism ) +add_executable(fastdb_legacy_lifetime test_legacy_lifetime.cpp) +target_include_directories( + fastdb_legacy_lifetime + PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR}/../../fastcarto/fastdb/include + ${CMAKE_CURRENT_SOURCE_DIR}/../../fastcarto/fastdb/src + ${CMAKE_CURRENT_SOURCE_DIR}/payload +) +target_link_libraries(fastdb_legacy_lifetime PRIVATE fastdb) +target_compile_options( + fastdb_legacy_lifetime + PRIVATE + $<$:/Zc:preprocessor> +) +fastdb_apply_sanitizers(fastdb_legacy_lifetime) +add_test( + NAME legacy.lifetime + COMMAND fastdb_legacy_lifetime + "$/legacy_lifetime.fdb" +) + set(FASTDB_PAYLOAD_FUZZ_MATCHING_SPEC ${CMAKE_CURRENT_SOURCE_DIR}/../golden/payload/v1/binary/spec/component-list-composition.source.json ) diff --git a/tests/cpp/test_legacy_lifetime.cpp b/tests/cpp/test_legacy_lifetime.cpp new file mode 100644 index 0000000..4471299 --- /dev/null +++ b/tests/cpp/test_legacy_lifetime.cpp @@ -0,0 +1,191 @@ +#include "TestSupport.hpp" +#include "fastdb.h" +#include "FastVectorDbLayerBuild_p.h" + +#include +#include +#include +#include +#include +#include + +#ifdef _WIN32 +#include +#else +#include +#endif + +namespace { + +void* map_buffer(size_t size) { +#ifdef _WIN32 + return VirtualAlloc(nullptr, size, MEM_COMMIT | MEM_RESERVE, + PAGE_READWRITE); +#else + void* data = mmap(nullptr, size, PROT_READ | PROT_WRITE, + MAP_PRIVATE | MAP_ANONYMOUS, -1, 0); + return data == MAP_FAILED ? nullptr : data; +#endif +} + +bool protect_buffer(void* data, size_t size, bool readable) { +#ifdef _WIN32 + DWORD previous = 0; + return VirtualProtect(data, size, readable ? PAGE_READWRITE : PAGE_NOACCESS, + &previous) != 0; +#else + return mprotect(data, size, readable ? PROT_READ | PROT_WRITE : PROT_NONE) == 0; +#endif +} + +bool unmap_buffer(void* data, size_t size) { +#ifdef _WIN32 + (void)size; + return VirtualFree(data, 0, MEM_RELEASE) != 0; +#else + return munmap(data, size) == 0; +#endif +} + +std::vector build_database() { + wx::FastVectorDbBuild build; + build.begin("{}"); + build.createLayerBegin("values"); + build.addField("tag", wx::ftU8); + build.addField("value", wx::ftF64); + for (int row = 0; row < 2; ++row) { + build.addFeatureBegin(); + wx::point2_t point{double(row), double(row)}; + build.setGeometry(&point, sizeof(point), wx::ginPoint2); + build.setField(0, row + 1); + build.setField(1, 2.5 + row); + build.addFeatureEnd(); + } + build.createLayerEnd(); + std::vector bytes(build.byteLength()); + if (build.postToBuffer(bytes.data(), bytes.size()) != bytes.size()) + return {}; + return bytes; +} + +struct CallbackState { + void* expected_data = nullptr; + size_t expected_size = 0; + int calls = 0; + bool matched = false; + bool restored = false; + bool unmapped = false; +}; + +void release_buffer(void* data, size_t size, void* cookie) { + auto* state = static_cast(cookie); + state->matched = data == state->expected_data && + size == state->expected_size; + ++state->calls; + state->restored = protect_buffer(data, size, true); + state->unmapped = unmap_buffer(data, size); +} + +int check_database(wx::FastVectorDb* db) { + require(db != nullptr); + require(db->getLayerCount() == 1); + auto* layer = db->getLayer(0); + require(layer != nullptr); + require(layer == db->getLayer(0)); + require(layer->getFeatureCount() == 2); + require(layer->getFieldOffset(0) == 0); + require(layer->getFieldOffset(1) == sizeof(wx::u8)); + require(layer->getFeatureByteSize() >= sizeof(wx::u8) + sizeof(double)); + + auto* base = static_cast(layer->getTableDataAddress()); + require(base != nullptr); + const size_t stride = layer->getFeatureByteSize(); + const size_t value_offset = layer->getFieldOffset(1); + for (int row = 0; row < 2; ++row) { + require(base[row * stride] == row + 1); + double value = 0; + std::memcpy(&value, base + row * stride + value_offset, + sizeof(value)); + require(value == 2.5 + row); + auto* feature = layer->tryGetFeatureAt(row); + require(feature != nullptr); + require(feature == layer->tryGetFeatureAt(row)); + require(feature->getFieldAsFloat(1) == value); + } + require(layer->tryGetFeatureAt(2) == nullptr); + return EXIT_SUCCESS; +} + +} // namespace + +int main(int argc, char** argv) { + require(argc == 2); + // A complete layer header does not imply its descriptors or string counts + // exist. Reject these envelopes before the layer constructor reads them. + std::vector truncated(20 + sizeof(wx::layer_header_t), 0); + std::memcpy(truncated.data(), "FASTVectorDB0.1", sizeof("FASTVectorDB0.1")); + const wx::u32 layer_count = 1; + std::memcpy(truncated.data() + 16, &layer_count, sizeof(layer_count)); + wx::layer_header_t header{}; + header.total_size = sizeof(header); + for (wx::u16 fields : {wx::u16(1), wx::u16(0)}) { + header.field_count = fields; + std::memcpy(truncated.data() + 20, &header, sizeof(header)); + require(wx::FastVectorDb::load(truncated.data(), truncated.size(), nullptr, nullptr) == nullptr); + } + void* invalid = map_buffer(truncated.size()); + require(invalid != nullptr); + std::memcpy(invalid, truncated.data(), truncated.size()); + CallbackState rejected{invalid, truncated.size()}; + require(wx::FastVectorDb::load(invalid, truncated.size(), release_buffer, &rejected) == nullptr); + require(rejected.calls == 1 && rejected.restored && rejected.unmapped); + + auto bytes = build_database(); + require(!bytes.empty()); + + // Revoking reads after caching children catches destructor access to backing. + void* borrowed = map_buffer(bytes.size()); + require(borrowed != nullptr); + std::memcpy(borrowed, bytes.data(), bytes.size()); + { + std::unique_ptr db( + wx::FastVectorDb::load(borrowed, bytes.size(), nullptr, nullptr)); + require(check_database(db.get()) == EXIT_SUCCESS); + require(protect_buffer(borrowed, bytes.size(), false)); + } + require(protect_buffer(borrowed, bytes.size(), true)); + require(static_cast(borrowed)[0] == 'F'); + require(unmap_buffer(borrowed, bytes.size())); + + // The owned callback restores and releases a protected mapping exactly once. + void* owned = map_buffer(bytes.size()); + require(owned != nullptr); + std::memcpy(owned, bytes.data(), bytes.size()); + CallbackState state{owned, bytes.size()}; + { + std::unique_ptr db(wx::FastVectorDb::load( + owned, bytes.size(), release_buffer, &state)); + require(check_database(db.get()) == EXIT_SUCCESS); + require(state.calls == 0); + require(protect_buffer(owned, bytes.size(), false)); + } + require(state.calls == 1); + require(state.matched); + require(state.restored); + require(state.unmapped); + + // CTest supplies a path in the build directory, outside the launch cwd. + const char* filename = argv[1]; + { + std::ofstream file(filename, std::ios::binary); + require(file.good()); + file.write(reinterpret_cast(bytes.data()), bytes.size()); + require(file.good()); + } + { + std::unique_ptr db(wx::FastVectorDb::load(filename)); + require(check_database(db.get()) == EXIT_SUCCESS); + } + require(std::remove(filename) == 0); + return EXIT_SUCCESS; +} diff --git a/tests/python/test_native_lifetime.py b/tests/python/test_native_lifetime.py new file mode 100644 index 0000000..9d5773b --- /dev/null +++ b/tests/python/test_native_lifetime.py @@ -0,0 +1,342 @@ +"""Native ownership, borrowed view safety, and bounded long-running retention.""" +import gc +import json +from pathlib import Path +import subprocess +import sys +import weakref + +import numpy as np +import pytest + +import fastdb4py as fdb +from fastdb4py import core + + +@fdb.feature +class NativeValue: + value: fdb.F32 + + +def _engine(kind): + engine_type = getattr(fdb, kind) + engine = engine_type.truncate([fdb.Layout(NativeValue, 3)]) + engine.table(NativeValue).column.value[:] = [1.25, 2.5, 3.75] + return engine + + +@pytest.fixture(autouse=True) +def _without_gc(): + enabled = gc.isenabled() + gc.disable() + try: + yield + finally: + if enabled: + gc.enable() + + +@pytest.mark.parametrize('kind', ['RecordEngine', 'ObjectEngine']) +@pytest.mark.parametrize('view_kind', ['table', 'row', 'column', 'slice', 'asarray', 'reuse']) +def test_trusted_views_keep_native_database_after_engine_close(kind, view_kind): + engine = _engine(kind) + table = engine.table(NativeValue) + native = table._db + assert native.thisown + reference = weakref.ref(native) + if view_kind == 'table': + view = table + elif view_kind == 'row': + view = table[1] + elif view_kind == 'reuse': + view = next(table.iter_reuse()) + elif view_kind == 'slice': + view = table.column.value[1:] + elif view_kind == 'asarray': + view = np.asarray(table.column.value)[1:] + else: + view = table.column.value + if kind == 'RecordEngine': + engine.close() + engine.close() + del native, table, engine + assert reference() is not None + if view_kind == 'table': + assert view[1].value == 2.5 + elif view_kind in ('row', 'reuse'): + assert view.value == (1.25 if view_kind == 'reuse' else 2.5) + else: + expected = [2.5, 3.75] if view_kind in ('slice', 'asarray') else [1.25, 2.5, 3.75] + np.testing.assert_array_equal(view, expected) + view[-1] = 9 + assert view[-1] == 9 + del view + assert reference() is None + + +@pytest.mark.parametrize('kind', ['RecordEngine', 'ObjectEngine']) +def test_checked_views_survive_engine_close_until_caller_invalidates_owner(kind): + engine = _engine(kind) + owner = fdb.FdbViewOwner(checked=True, writeable=True) + table = engine.table(NativeValue, owner=owner) + row = table[0] + column = table.column.value + detached = column.to_numpy() + if kind == 'RecordEngine': + engine.close() + del engine + assert row.value == 1.25 + assert column[0] == 1.25 + fdb.invalidate(owner) + for read in (lambda: len(table), lambda: row.value, lambda: column[0]): + with pytest.raises(fdb.FdbViewInvalidatedError): + read() + np.testing.assert_array_equal(detached, [1.25, 2.5, 3.75]) + + +def test_file_loaded_database_and_its_raw_column_release_after_last_reference(tmp_path): + engine = _engine('RecordEngine') + path = tmp_path / 'value.fdb' + engine.save(str(path)) + engine.close() + loaded = fdb.RecordEngine.load(str(path), from_file=True) + native = loaded._origin + reference = weakref.ref(native) + assert native.thisown + column = loaded.table(NativeValue).column.value + loaded.close() + assert loaded._origin is None + del native, loaded + np.testing.assert_array_equal(column, [1.25, 2.5, 3.75]) + assert reference() is not None + del column + assert reference() is None + + +def test_borrowed_buffer_remains_exported_and_alive_until_native_database_release(): + engine = _engine('RecordEngine') + backing = bytearray(engine._origin.buffer().to_bytes()) + engine.close() + native = core.WxDatabase.load_xbuffer(backing) + assert native.thisown + reference = weakref.ref(native) + layer = native.get_layer(0) + column = layer.get_column(0).as_nparray() + with pytest.raises(BufferError): + backing.extend(b'x') + del native, layer + np.testing.assert_array_equal(column, [1.25, 2.5, 3.75]) + assert reference() is not None + del column + assert reference() is None + backing.extend(b'x') + + +def test_raw_chunk_array_keeps_database_alive(): + engine = _engine('RecordEngine') + expected = engine._origin.buffer().to_bytes() + reference = weakref.ref(engine._origin) + array = engine._origin.buffer().as_array(np.uint8) + engine.close() + del engine + assert reference() is not None + assert array.tobytes() == expected + del array + assert reference() is None + + +def test_stream_copy_input_and_borrowed_array_lifetime(): + stream = core.WxMemoryStream() + stream.write(b'copied input') + reference = weakref.ref(stream) + array = stream.data().as_array(np.uint8) + del stream + assert reference() is not None + assert array.tobytes() == b'copied input' + del array + assert reference() is None + + +def test_shared_mapping_closes_after_last_raw_view(): + import secrets + + engine = _engine('RecordEngine') + old_table = engine.table(NativeValue) + engine.share('fdb_lifetime_' + secrets.token_hex(8)) + with pytest.raises(RuntimeError, match='read-only fixed tables'): + old_table.fill(value=[4, 5, 6]) + mapping = engine._shm._mmap + column = engine.table(NativeValue).column.value + engine.unlink() + del engine + assert not mapping.closed + np.testing.assert_array_equal(column, [1.25, 2.5, 3.75]) + del column + assert mapping.closed + + +@pytest.mark.skipif(sys.platform == 'win32', reason='close_after releases the publisher on POSIX only') +def test_shared_close_after_publish_releases_mapping(monkeypatch): + import secrets + from multiprocessing import shared_memory + import fastdb4py.record_engine as record_module + + mappings = [] + load_shared = record_module._load_shared_database + + def capture_mapping(shm): + mappings.append(shm._mmap) + return load_shared(shm) + + monkeypatch.setattr(record_module, '_load_shared_database', capture_mapping) + + name = 'fdb_close_' + secrets.token_hex(8) + engine = _engine('RecordEngine') + engine.share(name, close_after=True) + assert mappings[0].closed + engine.close() + loaded = fdb.RecordEngine.load(name) + np.testing.assert_array_equal(loaded.table(NativeValue).column.value, [1.25, 2.5, 3.75]) + loaded.close() + assert mappings[1].closed + segment = shared_memory.SharedMemory(name=name) + segment.unlink() + segment.close() + + +def test_object_get_row_survives_engine_release(): + engine = _engine('ObjectEngine') + row = engine.get(NativeValue, 1) + reference = weakref.ref(engine._db) + del engine + assert row.value == 2.5 + assert reference() is not None + del row + assert reference() is None + + +def test_factory_reference_ownership_matches_native_parent(): + standalone = core.WxFeatureRef.make_ref(0, 1) + assert standalone.thisown + build = core.WxDatabaseBuild() + build.begin('') + layer = build.create_layer_begin('values') + layer.add_field('value', core.ftF32) + layer.add_feature_begin() + layer.set_field(0, 2.5) + layer.add_feature_end() + borrowed = layer.create_feature_ref() + assert not borrowed.thisown + reference = weakref.ref(build) + del build, layer + assert reference() is not None + assert borrowed.ilayer == 0 + del borrowed + assert reference() is None + + +@pytest.mark.parametrize('kind', ['RecordEngine', 'ObjectEngine']) +def test_retained_column_accessor_keeps_native_backing_without_gc(kind): + engine = _engine(kind) + native = engine._origin if kind == 'RecordEngine' else engine._db + reference = weakref.ref(native) + accessor = engine.table(NativeValue).column + if kind == 'RecordEngine': + engine.close() + del engine, native + np.testing.assert_array_equal(accessor.value, [1.25, 2.5, 3.75]) + assert reference() is not None + del accessor + assert reference() is None + + +def test_table_retains_engine_without_engine_retaining_table(): + engine = _engine('RecordEngine') + table = engine.table(NativeValue) + engine_ref = weakref.ref(engine) + native_ref = weakref.ref(engine._origin) + del engine + assert engine_ref() is not None + assert table[1].value == 2.5 + del table + assert engine_ref() is None + assert native_ref() is None + + +def test_record_combine_releases_builder_and_push_dispatch_without_gc(): + engine = fdb.RecordEngine.create() + engine.push(NativeValue(value=1.25)) + builder_ref = weakref.ref(engine._origin) + engine.combine() + assert builder_ref() is None + assert not engine._push_buf + assert not engine._push_batch_fn + assert not engine._push_dispatch + engine.close() + + +def test_object_combine_releases_layer_and_database_builders_without_gc(): + engine = fdb.ObjectEngine.create() + state = engine._ensure_layer(NativeValue) + builder_ref = weakref.ref(engine._db_build) + layer_ref = weakref.ref(state.build) + engine.push(NativeValue(value=1.25)) + engine.combine() + assert state.build is None + assert builder_ref() is None + assert layer_ref() is None + del engine + + +def test_snapshot_remap_updates_checked_alias_and_invalidates_its_old_views(): + engine = _engine('RecordEngine') + table = engine.table(NativeValue) + owner = fdb.FdbViewOwner(checked=True, writeable=True) + alias = engine.table(NativeValue, owner=owner) + old_row = alias[0] + old_column = alias.column.value + raw = table.column.value + table.fill(value=[4, 5, 6]) + np.testing.assert_array_equal(alias.column.value.to_numpy(), [4, 5, 6]) + np.testing.assert_array_equal(raw, [1.25, 2.5, 3.75]) + with pytest.raises(fdb.FdbViewInvalidatedError): + _ = old_row.value + with pytest.raises(fdb.FdbViewInvalidatedError): + _ = old_column[0] + engine.close() + + +def test_empty_numeric_column(): + engine = fdb.RecordEngine.truncate([fdb.Layout(NativeValue, 0)]) + assert engine.table(NativeValue).column.value.shape == (0,) + engine.close() + + +def test_bulk_columns_do_not_request_per_row_native_features(monkeypatch): + def reject_row_lookup(*args): + raise AssertionError('bulk columns must not populate the feature cache') + + monkeypatch.setattr(core.WxLayerTable, 'tryGetFeature', reject_row_lookup) + engine = fdb.RecordEngine.truncate([fdb.Layout(NativeValue, 1_000_000)]) + table = engine.table(NativeValue) + table.fill(value=np.full(1_000_000, 7, dtype=np.float32)) + assert table.column.value[0] == 7 + assert table.column.value[-1] == 7 + engine.close() + + +@pytest.mark.parametrize('mode', ['write', 'read']) +def test_million_row_cycles_have_bounded_retained_memory(tmp_path, mode): + probe = Path(__file__).resolve().parents[2] / 'tools' / 'probe_standalone_memory.py' + result = subprocess.run( + [sys.executable, str(probe), '--mode', mode, '--rows', '1000000', + '--cycles', '10', '--path', str(tmp_path / 'frame.fdb')], + capture_output=True, text=True, check=True, + ) + report = json.loads(result.stdout.splitlines()[-1]) + # Allow two allocator warm-up cycles, then at most 16 MiB growth above + # that baseline across eight cycles (2 MiB/cycle). The old leak retains 16-20 MiB per + # cycle at this row count, so this tolerates allocator noise while rejecting + # linear retention by at least an order of magnitude. + samples = report['retained_bytes'][2:] + assert max(samples) - report['retained_bytes'][1] < 16 * 1024**2, result.stdout diff --git a/tests/python/test_native_shutdown.py b/tests/python/test_native_shutdown.py new file mode 100644 index 0000000..fc4c66e --- /dev/null +++ b/tests/python/test_native_shutdown.py @@ -0,0 +1,210 @@ +"""Borrowed buffers survive finalization and release through native destruction.""" +import gc +import mmap +from multiprocessing import shared_memory +import subprocess +import sys +import textwrap +import weakref + +import numpy as np +import pytest + +import fastdb4py as fdb +from fastdb4py import core + + +@fdb.feature +class ShutdownValue: + value: fdb.F32 + + +def _bytes(): + engine = fdb.RecordEngine.truncate([fdb.Layout(ShutdownValue, 4)]) + engine.table(ShutdownValue).fill(value=np.arange(4, dtype=np.float32)) + data = engine._origin.buffer().to_bytes() + engine.close() + return data + + +@pytest.mark.parametrize('kind', ['bytes', 'bytearray', 'mmap', 'shared_memory']) +def test_native_buffer_pin_and_mapping_release_without_cyclic_gc(kind): + enabled = gc.isenabled() + gc.disable() + data = _bytes() + mapping = None + segment = None + if kind == 'mmap': + mapping = mmap.mmap(-1, len(data)) + mapping[:] = data + backing = mapping + elif kind == 'shared_memory': + segment = shared_memory.SharedMemory(create=True, size=len(data)) + segment.buf[:len(data)] = data + mapping = segment._mmap + backing = segment.buf + else: + backing = bytes(data) if kind == 'bytes' else bytearray(data) + try: + native = core.WxDatabase.load_xbuffer(backing, segment) + reference = weakref.ref(native) + column = native.get_layer(0).get_column(0).as_nparray() + if kind == 'bytearray': + with pytest.raises(BufferError): + backing.extend(b'x') + elif kind in ('mmap', 'shared_memory'): + with pytest.raises(BufferError): + mapping.close() + del native + assert reference() is not None + np.testing.assert_array_equal(column, [0, 1, 2, 3]) + del column + assert reference() is None + if kind == 'shared_memory': + assert mapping.closed + elif kind == 'mmap': + mapping.close() + assert mapping.closed + elif kind == 'bytearray': + backing.extend(b'x') + finally: + if segment is not None: + segment.unlink() + segment.close() + elif mapping is not None: + mapping.close() + if enabled: + gc.enable() + + +@pytest.mark.parametrize('data', [b'', b'wrong database', b'FASTVectorDB0.1\0' + b'\0' * 3]) +def test_invalid_buffers_have_one_error_and_release_their_export(data): + backing = bytearray(data) + with pytest.raises(ValueError, match=r'^Invalid FastDB database buffer\.$'): + core.WxDatabase.load_xbuffer(backing) + backing.extend(b'x') + + +def test_array_base_always_retains_the_actual_chunk(): + native = core.WxDatabase.load_xbuffer(_bytes()) + chunk = native.buffer() + array = chunk.as_array(np.uint8) + assert array.base is chunk + with pytest.raises(TypeError): + chunk.as_array(np.uint8, None) + reference = weakref.ref(native) + del native, chunk + assert reference() is not None + assert array.tobytes().startswith(b'FASTVectorDB0.1') + del array + assert reference() is None + + +def test_failed_shared_load_closes_its_mapping(): + from fastdb4py.record_engine import _load_shared_database + + segment = shared_memory.SharedMemory(create=True, size=64) + mapping = segment._mmap + try: + with pytest.raises(ValueError, match=r'^Invalid FastDB database buffer\.$'): + _load_shared_database(segment) + assert mapping.closed + finally: + segment.unlink() + segment.close() + + +def test_invalid_file_uses_the_same_native_factory_error(tmp_path): + path = tmp_path / 'invalid.fdb' + path.write_bytes(b'not a database') + with pytest.raises(ValueError, match=r'^Invalid FastDB database buffer\.$'): + fdb.RecordEngine.load(str(path), from_file=True) + + +def test_close_does_not_invalidate_an_owner_shared_by_two_engines(): + owner = fdb.FdbViewOwner(checked=True) + first = fdb.RecordEngine.truncate([fdb.Layout(ShutdownValue, 4)]) + second = fdb.RecordEngine.truncate([fdb.Layout(ShutdownValue, 4)]) + left = first.table(ShutdownValue, owner=owner) + right = second.table(ShutdownValue, owner=owner) + first.close() + assert owner.alive + assert left[0].value == right[0].value == 0 + fdb.invalidate(owner) + with pytest.raises(fdb.FdbViewInvalidatedError): + _ = right[0].value + second.close() + + +_SHUTDOWN_SETUP = ''' +import gc, secrets, threading, time +import numpy as np +import fastdb4py as fdb +@fdb.feature +class V: + value: fdb.F32 +name = 'fdb_shutdown_' + secrets.token_hex(8) +publisher = fdb.RecordEngine.truncate([fdb.Layout(V, 2000000)]) +publisher.table(V).fill(value=np.ones(2000000, dtype=np.float32)) +publisher.share(name) +''' + +_SHUTDOWN_CASES = { + 'daemon': ''' +reader = fdb.RecordEngine.load(name) +column = reader.table(V).column.value +publisher.unlink() +ready = threading.Event() +def worker(): + ready.set() + while True: + float(column.sum()) +threading.Thread(target=worker, daemon=True).start() +ready.wait() +time.sleep(0.1) +print('daemon reader exits safely', flush=True) +''', + 'module_del': ''' +class Reporter: + def __init__(self, column): + self.column = column + def __del__(self): + assert float(self.column.sum()) == 2000000 + print('module destructor read passed', flush=True) +reader = fdb.RecordEngine.load(name) +reporter = Reporter(reader.table(V).column.value) +publisher.unlink() +''', + 'gc_cycle': ''' +class Service: + def __init__(self): + self.engine = fdb.RecordEngine.load(name) + self.table = self.engine.table(V) + self.cycle = self + def __del__(self): + assert self.table[3].value == 1 + print('cycle destructor read passed', flush=True) +service = Service() +mapping = service.engine._shm._mmap +del service +gc.collect() +assert mapping.closed +publisher.unlink() +print('cycle mapping closed', flush=True) +''', +} + + +@pytest.mark.parametrize('case', _SHUTDOWN_CASES) +def test_shared_memory_shutdown_subprocess(case, tmp_path): + result = subprocess.run( + [sys.executable, '-X', 'faulthandler', '-c', + textwrap.dedent(_SHUTDOWN_SETUP + _SHUTDOWN_CASES[case])], + cwd=tmp_path, capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert 'Exception ignored' not in result.stderr, result.stderr + if case == 'gc_cycle': + assert 'cycle mapping closed' in result.stdout + elif case == 'module_del': + assert 'module destructor read passed' in result.stdout diff --git a/tests/python/test_object_engine_lifetime.py b/tests/python/test_object_engine_lifetime.py new file mode 100644 index 0000000..010103a --- /dev/null +++ b/tests/python/test_object_engine_lifetime.py @@ -0,0 +1,37 @@ +"""Reference-counted lifetime of standalone ObjectEngine views.""" + +import weakref + +import numpy as np + +import fastdb4py as fdb + + +@fdb.feature +class ObjectLifetimeValue: + value: fdb.F32 + + +def test_raw_column_retains_native_database_after_engine_is_released(): + engine = fdb.ObjectEngine.create() + engine.push(ObjectLifetimeValue(value=1.25)) + engine.push(ObjectLifetimeValue(value=2.5)) + engine.combine() + + assert engine._db_build is None + assert all(state.build is None for state in engine._layers.values()) + + table = engine.table(ObjectLifetimeValue) + column = table.column.value + native_ref = weakref.ref(table._db) + engine_ref = weakref.ref(engine) + del table, engine + + assert engine_ref() is None + assert native_ref() is not None + np.testing.assert_array_equal(column, [1.25, 2.5]) + column[1] = 3.75 + assert column[1] == 3.75 + + del column + assert native_ref() is None diff --git a/tests/python/test_push.py b/tests/python/test_push.py index eddc589..bfb7824 100644 --- a/tests/python/test_push.py +++ b/tests/python/test_push.py @@ -45,7 +45,6 @@ def test_push_scalar_fields(): db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf layer = rdb.get_layer(0) feat = layer.tryGetFeature(0) assert abs(feat.get_field_as_float(0) - 42.0) < 1e-9 @@ -67,7 +66,6 @@ class Named: db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf layer = rdb.get_layer(0) feat = layer.tryGetFeature(0) def test_push_list_field(): @@ -85,7 +83,6 @@ class WithList: db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf layer = rdb.get_layer(0) feat = layer.tryGetFeature(0) arr = feat.get_field_as_list_view(0).as_array(np.float64) @@ -102,7 +99,6 @@ def test_push_default_values(): db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf layer = rdb.get_layer(0) feat = layer.tryGetFeature(0) assert feat.get_field_as_float(0) == 0.0 @@ -124,7 +120,6 @@ class WithEmptyList: db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf layer = rdb.get_layer(0) feat = layer.tryGetFeature(0) arr = feat.get_field_as_list_view(0).as_array(np.float64) @@ -144,7 +139,6 @@ class WithNumpyList: db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf layer = rdb.get_layer(0) feat = layer.tryGetFeature(0) arr = feat.get_field_as_list_view(0).as_array(np.float64) @@ -176,7 +170,6 @@ class Mixed: db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf layer = rdb.get_layer(0) feat = layer.tryGetFeature(0) @@ -231,4 +224,4 @@ def test_push_ref_list_via_orm(): tag0 = orm.get(Tag, 0, mode='copy') tag1 = orm.get(Tag, 1, mode='copy') assert tag0.name == "python" - assert tag1.name == "fastdb" \ No newline at end of file + assert tag1.name == "fastdb" diff --git a/tests/python/test_reader.py b/tests/python/test_reader.py index ea307b6..76c9940 100644 --- a/tests/python/test_reader.py +++ b/tests/python/test_reader.py @@ -41,7 +41,6 @@ def _build_db_with_points(): db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf return rdb @@ -188,7 +187,6 @@ class WithBytes: db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf layer = rdb.get_layer(0) obj = copy_feature(WithBytes, layer, 0) @@ -199,7 +197,6 @@ class WithBytes: def test_varlen_string_column_reader_exposes_raw_buffers(): buf, offsets, data = _build_varlen_string_db_bytes(["a", "bé", "中"]) rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf layer = rdb.get_layer(0) np.testing.assert_array_equal( @@ -220,7 +217,6 @@ def test_varlen_string_column_reader_rejects_truncated_section_payload(): corrupted_buf = _corrupt_first_varlen_string_byte_count(valid_buf, data.nbytes + 8) rdb = core.WxDatabase.load_xbuffer(corrupted_buf) - rdb._buffer = corrupted_buf layer = rdb.get_layer(0) assert layer.get_string_column_offsets(0).size == 0 @@ -242,7 +238,6 @@ def test_truncate_layer_build_set_numeric_column_bulk_round_trips(): db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf out = rdb.get_layer(0).get_column(0).as_nparray() np.testing.assert_allclose(out, np.array([1.0, 2.5, 3.5], dtype=np.float64)) @@ -278,7 +273,6 @@ def test_truncate_layer_build_set_numeric_column_bulk_silently_ignores_wrong_siz db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf out = rdb.get_layer(0).get_column(0).as_nparray() np.testing.assert_allclose(out, np.array([1.0, 2.5, 3.5], dtype=np.float64)) @@ -297,7 +291,6 @@ def test_truncate_layer_build_set_numeric_column_bulk_zero_rows_round_trips(): db.post(mem) buf = mem.data().as_array(np.uint8).tobytes() rdb = core.WxDatabase.load_xbuffer(buf) - rdb._buffer = buf out_layer = rdb.get_layer(0) assert out_layer.get_feature_count() == 0 assert out_layer.get_field_count() == 1 diff --git a/tests/python/test_record_engine.py b/tests/python/test_record_engine.py index 825666d..090cd15 100644 --- a/tests/python/test_record_engine.py +++ b/tests/python/test_record_engine.py @@ -260,7 +260,7 @@ def test_record_engine_push_many_bool_fields_reject_before_table_creation(): CEBoolPoint(active='maybe'), ]) - assert CEBoolPoint.__name__ not in engine._table_map + assert CEBoolPoint.__name__ not in engine._table_feature_types assert CEBoolPoint.__name__ not in engine._table_feature_types @@ -447,7 +447,7 @@ def test_table_fill_coerces_float32_into_f64_columns(): def test_table_fill_rejects_non_fixed_tables(): engine = RecordEngine.create() engine.push(CEPoint(x=1.0, y=2.0)) - tbl = engine._table_map[CEPoint.__name__] + tbl = engine.table(CEPoint) with pytest.raises(RuntimeError, match='fixed-scale tables'): tbl.fill(x=np.array([1.0], dtype=np.float64)) diff --git a/tests/python/test_string_column.py b/tests/python/test_string_column.py index d3311ef..03117ad 100644 --- a/tests/python/test_string_column.py +++ b/tests/python/test_string_column.py @@ -95,15 +95,15 @@ def test_normalize_string_values_wraps_raw_sequence_payload(): def test_string_column_fill_routes_sequence_payload_to_fixed_handler(monkeypatch): engine = RecordEngine.truncate([Layout(CEStringPoint, 2)]) - tbl = engine.table(CEStringPoint) captured = [] original = engine._fill_fixed_table - def capture_fill(table_name, writes): - captured.append((table_name, writes)) - return original(table_name, writes) + def capture_fill(state, writes): + captured.append((state._table_name, writes)) + return original(state, writes) monkeypatch.setattr(engine, '_fill_fixed_table', capture_fill) + tbl = engine.table(CEStringPoint) tbl.column.name.fill(["hi", None]) @@ -118,16 +118,16 @@ def capture_fill(table_name, writes): def test_string_column_fill_utf8_uses_unified_fixed_writer(monkeypatch): engine = RecordEngine.truncate([Layout(CEStringPoint, 2)]) - tbl = engine.table(CEStringPoint) captured = [] original = engine._fill_fixed_table - def capture_fill(table_name, writes): - captured.append((table_name, writes)) - return original(table_name, writes) + def capture_fill(state, writes): + captured.append((state._table_name, writes)) + return original(state, writes) monkeypatch.setattr(engine, '_fill_fixed_table', capture_fill) + tbl = engine.table(CEStringPoint) offsets, data = _pack_utf8(["hi", "中"]) tbl.column.name.fill_utf8(offsets, data) diff --git a/tests/python/test_table_ownership_cache.py b/tests/python/test_table_ownership_cache.py new file mode 100644 index 0000000..a53626f --- /dev/null +++ b/tests/python/test_table_ownership_cache.py @@ -0,0 +1,166 @@ +"""Cached table projections retain parents without requiring cyclic GC.""" +import gc +import weakref + +import numpy as np +import pytest + +import fastdb4py as fdb + + +@fdb.feature +class CachedValue: + value: fdb.F32 + name: fdb.STR + + +@pytest.fixture(autouse=True) +def without_gc(): + enabled = gc.isenabled() + gc.disable() + try: + yield + finally: + if enabled: + gc.enable() + + +@pytest.mark.parametrize('checked', [False, True]) +def test_accessor_and_columns_are_cached_without_retaining_table(checked): + engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)]) + owner = fdb.FdbViewOwner(checked=checked, writeable=True) + table = engine.table(CachedValue, owner=owner) + accessor = table.column + numeric, strings = accessor.value, accessor.name + assert table.column is accessor + assert table.column.value is numeric + assert table.column.name is strings + assert weakref.ref(table.column)() is accessor + table_ref = weakref.ref(table) + engine_ref = weakref.ref(engine) + native_ref = weakref.ref(table._db) + engine.close() + del engine, table + assert table_ref() is None + assert engine_ref() is not None + assert len(numeric) == len(strings) == 3 + del accessor, numeric, strings + assert engine_ref() is None + assert native_ref() is None + + +def test_table_retains_engine_for_chained_fill_and_releases_immediately(): + table = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)]).table(CachedValue) + table.fill(value=[1, 2, 3], name=['a', 'b', 'c']) + np.testing.assert_array_equal(table.column.value, [1, 2, 3]) + assert table.column.name.to_pylist() == ['a', 'b', 'c'] + native_ref = weakref.ref(table._db) + del table + assert native_ref() is None + + +def test_table_strongly_caches_accessor_between_independent_reads(): + engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)]) + table = engine.table(CachedValue) + cached = weakref.ref(table.column) + assert cached() is not None + assert table.column is cached() + column = table.column.value + assert table.column.value is column + table_ref = weakref.ref(table) + del table + assert table_ref() is None + assert cached() is None + + +def test_one_weak_registry_keeps_default_table_identity_and_releases_aliases(): + engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)]) + table = engine.table(CachedValue) + assert engine.table(CachedValue) is table + owner = fdb.FdbViewOwner(checked=True, writeable=True) + alias = engine.table(CachedValue, owner=owner) + assert len(engine._tables) == 2 + table_ref, alias_ref = weakref.ref(table), weakref.ref(alias) + del table, alias + assert table_ref() is alias_ref() is None + assert not engine._tables + engine_ref = weakref.ref(engine) + del engine + assert engine_ref() is None + + +def test_remap_replaces_cached_columns_and_releases_old_database(): + engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)]) + table = engine.table(CachedValue) + accessor = table.column + column = accessor.value + previous = weakref.ref(table._db) + table.fill(value=[4, 5, 6], name=['x', 'y', 'z']) + assert table.column is not accessor + assert table.column.value is table.column.value + np.testing.assert_array_equal(table.column.value, [4, 5, 6]) + np.testing.assert_array_equal(column, [0, 0, 0]) + assert previous() is not None + del accessor, column + assert previous() is None + current = weakref.ref(table._db) + del table, engine + assert current() is None + + +def test_checked_accessor_cannot_create_columns_on_stale_backing_after_remap(): + engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)]) + owner = fdb.FdbViewOwner(checked=True, writeable=True) + table = engine.table(CachedValue, owner=owner) + accessor = table.column + engine.table(CachedValue).fill(value=[1, 2, 3], name=['a', 'b', 'c']) + for field in ('value', 'name'): + with pytest.raises(fdb.FdbViewInvalidatedError, match='stale after owner remap'): + getattr(accessor, field) + np.testing.assert_array_equal(table.column.value.to_numpy(), [1, 2, 3]) + assert table.column.name.to_pylist() == ['a', 'b', 'c'] + + +def test_retained_string_column_follows_snapshot_without_retaining_table(): + engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)]) + table = engine.table(CachedValue) + column = table.column.name + table_ref = weakref.ref(table) + del table + assert table_ref() is None + column.fill(['a', 'b', 'c']) + assert column.to_pylist() == ['a', 'b', 'c'] + engine.table(CachedValue).fill(name=['x', 'y', 'z']) + assert column.to_pylist() == ['x', 'y', 'z'] + native_ref = weakref.ref(engine._origin) + del engine + assert column.to_pylist() == ['x', 'y', 'z'] + del column + assert native_ref() is None + + +def test_remap_refreshes_aliases_that_share_a_checked_owner(): + engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)]) + owner = fdb.FdbViewOwner(checked=True, writeable=True) + aliases = [engine.table(CachedValue, owner=owner) for _ in range(2)] + engine.table(CachedValue).fill(value=[1, 2, 3]) + for table in aliases: + np.testing.assert_array_equal(table.column.value.to_numpy(), [1, 2, 3]) + + +@pytest.mark.parametrize('method', ['table', 'string', 'utf8']) +def test_engine_writes_after_close_fail_clearly_but_raw_array_writes_work(method): + engine = fdb.RecordEngine.truncate([fdb.Layout(CachedValue, 3)]) + table = engine.table(CachedValue) + strings = table.column.name + array = table.column.value + engine.close() + with pytest.raises(RuntimeError, match='RecordEngine is closed'): + if method == 'table': + table.fill(value=[1, 2, 3]) + elif method == 'string': + strings.fill(['a', 'b', 'c']) + else: + strings.fill_utf8(*fdb.pack_utf8_column(['a', 'b', 'c'])) + array[:] = [7, 8, 9] + np.testing.assert_array_equal(table.column.value, [7, 8, 9]) diff --git a/tools/probe_standalone_memory.py b/tools/probe_standalone_memory.py new file mode 100644 index 0000000..5b07a44 --- /dev/null +++ b/tools/probe_standalone_memory.py @@ -0,0 +1,109 @@ +"""Measure current process memory after complete standalone database cycles. + +Runs in its own process so allocator warm-up and the import baseline are explicit. +""" +import argparse +import ctypes +import json +import os +from pathlib import Path +import sys + +import numpy as np +import fastdb4py as fdb + +# ctypes libraries cache function objects that reference their library. Build +# these once so the monitor itself does not add cycles with GC disabled. +if sys.platform == 'darwin': + _proc_pidinfo = ctypes.CDLL('/usr/lib/libproc.dylib').proc_pidinfo + _proc_pidinfo.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_uint64, ctypes.c_void_p, ctypes.c_int] +elif sys.platform == 'win32': + from ctypes import wintypes + + class Counters(ctypes.Structure): + _fields_ = [('cb', wintypes.DWORD), ('faults', wintypes.DWORD)] + [ + (name, ctypes.c_size_t) for name in ( + 'peak_working', 'working', 'peak_paged', 'paged', + 'peak_nonpaged', 'nonpaged', 'pagefile', 'peak_pagefile', 'private', + ) + ] + + _kernel = ctypes.WinDLL('kernel32', use_last_error=True) + _kernel.GetCurrentProcess.restype = wintypes.HANDLE + _psapi = ctypes.WinDLL('psapi', use_last_error=True) + _psapi.GetProcessMemoryInfo.argtypes = [wintypes.HANDLE, ctypes.c_void_p, wintypes.DWORD] + + +@fdb.feature +class MemoryValue: + value: fdb.F32 + + +def resident_bytes(): + if sys.platform == 'darwin': + info = ctypes.create_string_buffer(96) + count = _proc_pidinfo(os.getpid(), 4, 0, info, len(info)) + if count != len(info): + raise OSError('proc_pidinfo(PROC_PIDTASKINFO) failed') + return ctypes.c_uint64.from_buffer(info, 8).value + if sys.platform == 'win32': + counters = Counters() + counters.cb = ctypes.sizeof(counters) + if not _psapi.GetProcessMemoryInfo(_kernel.GetCurrentProcess(), ctypes.byref(counters), counters.cb): + raise ctypes.WinError(ctypes.get_last_error()) + return counters.private + return int(Path('/proc/self/statm').read_text().split()[1]) * os.sysconf('SC_PAGE_SIZE') + + +def run(path, rows, cycles, mode): + import gc + gc.disable() + values = np.random.default_rng(0).random(rows, dtype=np.float32) + if mode == 'read': + engine = fdb.RecordEngine.truncate([fdb.Layout(MemoryValue, rows)]) + engine.table(MemoryValue).fill(value=values) + engine.save(str(path)) + engine.close() + del engine + results = [] + for iteration in range(cycles): + steps = [('start', resident_bytes())] + if mode == 'write': + engine = fdb.RecordEngine.truncate([fdb.Layout(MemoryValue, rows)]) + steps.append(('truncate', resident_bytes())) + table = engine.table(MemoryValue) + table.fill(value=values) + steps.append(('fill', resident_bytes())) + engine.save(str(path)) + steps.append(('save', resident_bytes())) + del table + else: + engine = fdb.RecordEngine.load(str(path), from_file=True) + steps.append(('load', resident_bytes())) + copied = np.array(engine.table(MemoryValue).column.value, copy=True) + np.testing.assert_array_equal(copied, values) + steps.append(('column_copy', resident_bytes())) + del copied + owns = engine._origin.thisown + engine.close() + steps.append(('close', resident_bytes())) + del engine + steps.append(('released', resident_bytes())) + entry = {'iteration': iteration + 1, 'thisown': owns, 'bytes': dict(steps)} + entry['net_bytes'] = steps[-1][1] - steps[0][1] + results.append(entry) + print(json.dumps(entry), flush=True) + summary = {'mode': mode, 'rows': rows, 'cycles': cycles, + 'retained_bytes': [r['bytes']['released'] for r in results]} + print(json.dumps(summary), flush=True) + return summary + + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument('--path', type=Path, required=True) + parser.add_argument('--rows', type=int, default=1_000_000) + parser.add_argument('--cycles', type=int, default=8) + parser.add_argument('--mode', choices=('write', 'read'), required=True) + args = parser.parse_args() + run(args.path, args.rows, args.cycles, args.mode) From 5cc10458acf5631c8ca250dd1168f2f232853321 Mon Sep 17 00:00:00 2001 From: Dsssyc Date: Fri, 30 Oct 2026 00:11:18 +0800 Subject: [PATCH 2/2] Read shared memory without imports in the teardown test On Linux CPython 3.12 and 3.13t, the module-destructor case reached ndarray.sum() after finalization had disabled imports, and NumPy's runtime import of its reduction helper raised ImportError. The mapped column itself was still readable. Verify the whole column with tobytes() and bytes.count, and report through a pre-bound os.write, so the destructor needs no imports. --- tests/python/test_native_shutdown.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/python/test_native_shutdown.py b/tests/python/test_native_shutdown.py index fc4c66e..cc7c3dd 100644 --- a/tests/python/test_native_shutdown.py +++ b/tests/python/test_native_shutdown.py @@ -165,12 +165,16 @@ def worker(): print('daemon reader exits safely', flush=True) ''', 'module_del': ''' +import os class Reporter: + # Late teardown cannot import, so read and report through C-level calls only. def __init__(self, column): self.column = column + self.one = np.float32(1).tobytes() + self.write = os.write def __del__(self): - assert float(self.column.sum()) == 2000000 - print('module destructor read passed', flush=True) + assert self.column.tobytes().count(self.one) == 2000000 + self.write(1, b'module destructor read passed\\n') reader = fdb.RecordEngine.load(name) reporter = Reporter(reader.table(V).column.value) publisher.unlink()