From f5ea6386357432abd00ac4928c5d75834a4add7e Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 12 Nov 2026 16:13:10 +0800 Subject: [PATCH 1/7] feat(cpp/python): expose the parallel read/write switches Add tsfile_set_parallel_read_enabled, tsfile_set_parallel_write_enabled and tsfile_set_thread_count to the C API, and the matching set_parallel_read_enabled / set_parallel_write_enabled / set_thread_count to the Python binding. The switches already existed as C++ globals; only the bindings were missing. They matter for benchmarking and for single-table workloads, where running the read path with the pool disabled measured ~20% faster than the default because the worker pool is mostly idle. --- cpp/src/cwrapper/tsfile_cwrapper.cc | 14 ++++++++++++++ cpp/src/cwrapper/tsfile_cwrapper.h | 26 ++++++++++++++++++++++++++ python/tsfile/__init__.py | 3 +++ python/tsfile/tsfile_cpp.pxd | 3 +++ python/tsfile/tsfile_py_cpp.pxd | 3 +++ python/tsfile/tsfile_py_cpp.pyx | 12 ++++++++++++ 6 files changed, 61 insertions(+) diff --git a/cpp/src/cwrapper/tsfile_cwrapper.cc b/cpp/src/cwrapper/tsfile_cwrapper.cc index 6eaffff55..8bb9cd39f 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.cc +++ b/cpp/src/cwrapper/tsfile_cwrapper.cc @@ -169,6 +169,20 @@ TsFileReadBackend tsfile_get_file_read_backend() { return static_cast(common::get_file_read_backend()); } +ERRNO tsfile_set_parallel_read_enabled(int enabled) { + common::set_parallel_read_enabled(enabled != 0); + return common::E_OK; +} + +ERRNO tsfile_set_parallel_write_enabled(int enabled) { + common::set_parallel_write_enabled(enabled != 0); + return common::E_OK; +} + +ERRNO tsfile_set_thread_count(int32_t count) { + return common::set_thread_count(count); +} + WriteFile write_file_new(const char* pathname, ERRNO* err_code) { if (err_code == nullptr) { return nullptr; diff --git a/cpp/src/cwrapper/tsfile_cwrapper.h b/cpp/src/cwrapper/tsfile_cwrapper.h index baa662142..5451b5a01 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.h +++ b/cpp/src/cwrapper/tsfile_cwrapper.h @@ -357,6 +357,32 @@ ERRNO tsfile_set_file_read_backend(int32_t backend); /** @return The backend configured for subsequently opened readers. */ TsFileReadBackend tsfile_get_file_read_backend(void); +/** + * @brief Enable/disable the parallel read path for subsequently opened + * readers. + * + * @param enabled Non-zero to enable, zero to disable. + * @return RET_OK on success. + */ +ERRNO tsfile_set_parallel_read_enabled(int enabled); + +/** + * @brief Enable/disable the parallel write path for subsequently opened + * writers. + * + * @param enabled Non-zero to enable, zero to disable. + * @return RET_OK on success. + */ +ERRNO tsfile_set_parallel_write_enabled(int enabled); + +/** + * @brief Set the size of the global worker thread pool. + * + * @param count Number of worker threads, in [1, 64]. + * @return RET_OK on success, or RET_INVALID_ARG for an out-of-range count. + */ +ERRNO tsfile_set_thread_count(int32_t count); + /** * @brief Get the encoding type for global time column * diff --git a/python/tsfile/__init__.py b/python/tsfile/__init__.py index c0bf8fb85..4afa123dd 100644 --- a/python/tsfile/__init__.py +++ b/python/tsfile/__init__.py @@ -91,6 +91,9 @@ def _preload_dll(path): get_file_read_backend, get_tsfile_config, set_file_read_backend, + set_parallel_read_enabled, + set_parallel_write_enabled, + set_thread_count, set_tsfile_config, ) from .tsfile_table_writer import TsFileTableWriter diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd index e4f75b868..80e5946f9 100644 --- a/python/tsfile/tsfile_cpp.pxd +++ b/python/tsfile/tsfile_cpp.pxd @@ -230,6 +230,9 @@ cdef extern from "cwrapper/tsfile_cwrapper.h": ErrorCode tsfile_set_file_read_backend(int32_t backend) TsFileReadBackend tsfile_get_file_read_backend() + ErrorCode tsfile_set_parallel_read_enabled(int32_t enabled) + ErrorCode tsfile_set_parallel_write_enabled(int32_t enabled) + ErrorCode tsfile_set_thread_count(int32_t count) ctypedef void * TagFilterHandle diff --git a/python/tsfile/tsfile_py_cpp.pxd b/python/tsfile/tsfile_py_cpp.pxd index b7106a03e..c3e4c66eb 100644 --- a/python/tsfile/tsfile_py_cpp.pxd +++ b/python/tsfile/tsfile_py_cpp.pxd @@ -92,3 +92,6 @@ cpdef public api object get_tsfile_config() cpdef public api void set_tsfile_config(dict new_config) cpdef public api object get_file_read_backend() cpdef public api void set_file_read_backend(object backend) +cpdef public api void set_parallel_read_enabled(bint enabled) +cpdef public api void set_parallel_write_enabled(bint enabled) +cpdef public api void set_thread_count(int32_t count) diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx index ae1083b4d..c67923fe3 100644 --- a/python/tsfile/tsfile_py_cpp.pyx +++ b/python/tsfile/tsfile_py_cpp.pyx @@ -927,6 +927,18 @@ cpdef void set_file_read_backend(object backend): raise TypeError(f"Unsupported FileReadBackend: {backend}") check_error(tsfile_set_file_read_backend( int(backend.value))) +cpdef void set_parallel_read_enabled(bint enabled): + """Enable/disable the parallel read path for subsequently opened readers.""" + check_error(tsfile_set_parallel_read_enabled( enabled)) + +cpdef void set_parallel_write_enabled(bint enabled): + """Enable/disable the parallel write path for subsequently opened writers.""" + check_error(tsfile_set_parallel_write_enabled( enabled)) + +cpdef void set_thread_count(int32_t count): + """Set the size of the global worker thread pool (1..64).""" + check_error(tsfile_set_thread_count(count)) + cpdef void set_tsfile_config(dict new_config): if "file_read_backend_" in new_config and not isinstance( new_config["file_read_backend_"], FileReadBackendPy From d3c96f155a5d7c0ac576196b43dc700d96972a1d Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 12 Nov 2026 17:03:38 +0800 Subject: [PATCH 2/7] test(cpp/python): cover the parallel read/write switches C API: verify the setters take effect on the shared config, that thread counts outside [1, 64] are rejected without touching the configured value, and that flipping the parallel read switch over the same file returns identical rows. Python: exercise the new bindings, including the InvalidArgumentError raised for an out-of-range thread count, and compare reads with the parallel path on and off. --- cpp/test/cwrapper/cwrapper_test.cc | 143 ++++++++++++++++++++++++++++ python/tests/test_write_and_read.py | 79 +++++++++++++++ 2 files changed, 222 insertions(+) diff --git a/cpp/test/cwrapper/cwrapper_test.cc b/cpp/test/cwrapper/cwrapper_test.cc index 0dda7797c..290b182a0 100644 --- a/cpp/test/cwrapper/cwrapper_test.cc +++ b/cpp/test/cwrapper/cwrapper_test.cc @@ -25,11 +25,18 @@ #include #include +#include +#include +#include +#include "common/global.h" #include "common/row_record.h" +#include "common/tablet.h" #include "cwrapper/tsfile_cwrapper.h" +#include "file/write_file.h" #include "reader/result_set.h" #include "reader/tsfile_reader.h" +#include "writer/tsfile_table_writer.h" #include "writer/tsfile_writer.h" namespace storage { @@ -66,6 +73,142 @@ TEST_F(CWrapperTest, FileReadBackendConfigurationRoundTripsAndValidates) { EXPECT_EQ(tsfile_set_file_read_backend(original), RET_OK); } +TEST_F(CWrapperTest, ParallelTogglesAndThreadCountValidate) { + const bool read_enabled = common::get_parallel_read_enabled(); + const bool write_enabled = common::get_parallel_write_enabled(); + const int32_t thread_count = common::g_config_value_.thread_count_; + + EXPECT_EQ(tsfile_set_parallel_read_enabled(0), RET_OK); + EXPECT_FALSE(common::get_parallel_read_enabled()); + EXPECT_EQ(tsfile_set_parallel_read_enabled(1), RET_OK); + EXPECT_TRUE(common::get_parallel_read_enabled()); + // Any non-zero value enables the path, matching the bool parameter. + EXPECT_EQ(tsfile_set_parallel_read_enabled(42), RET_OK); + EXPECT_TRUE(common::get_parallel_read_enabled()); + + EXPECT_EQ(tsfile_set_parallel_write_enabled(0), RET_OK); + EXPECT_FALSE(common::get_parallel_write_enabled()); + EXPECT_EQ(tsfile_set_parallel_write_enabled(1), RET_OK); + EXPECT_TRUE(common::get_parallel_write_enabled()); + + // Out-of-range counts are rejected and leave the configured size alone. + for (int32_t invalid : {0, -1, 65, 1000}) { + EXPECT_EQ(tsfile_set_thread_count(invalid), RET_INVALID_ARG) << invalid; + EXPECT_EQ(common::g_config_value_.thread_count_, thread_count) + << invalid; + } + for (int32_t valid : {1, 32, 64}) { + EXPECT_EQ(tsfile_set_thread_count(valid), RET_OK) << valid; + EXPECT_EQ(common::g_config_value_.thread_count_, valid) << valid; + } + + EXPECT_EQ(tsfile_set_thread_count(thread_count), RET_OK); + EXPECT_EQ(tsfile_set_parallel_read_enabled(read_enabled), RET_OK); + EXPECT_EQ(tsfile_set_parallel_write_enabled(write_enabled), RET_OK); +} + +// Flipping the parallel read switch must not change what a query returns; it +// only selects whether the aligned multi-column read uses the worker pool. +TEST_F(CWrapperTest, ParallelReadToggleReturnsIdenticalRows) { + storage::libtsfile_init(); + const std::string path = "cwrapper_parallel_read_toggle.tsfile"; + remove(path.c_str()); + constexpr int kDevices = 2; + constexpr int kRowsPerDevice = 500; + constexpr int kFields = 5; + { + storage::WriteFile file; + int flags = O_WRONLY | O_CREAT | O_TRUNC; +#ifdef _WIN32 + flags |= O_BINARY; +#endif + ASSERT_EQ(file.create(path, flags, 0666), common::E_OK); + std::vector columns; + columns.emplace_back("device", common::STRING, common::UNCOMPRESSED, + common::PLAIN, common::ColumnCategory::TAG); + for (int c = 0; c < kFields; ++c) { + columns.emplace_back("s" + std::to_string(c), common::INT64, + common::UNCOMPRESSED, common::PLAIN, + common::ColumnCategory::FIELD); + } + storage::TableSchema schema("t", columns); + storage::TsFileTableWriter writer(&file, &schema); + storage::Tablet tablet( + "t", schema.get_measurement_names(), schema.get_data_types(), + schema.get_column_categories(), kDevices * kRowsPerDevice); + int index = 0; + for (int device = 0; device < kDevices; ++device) { + const std::string tag = "d" + std::to_string(device); + for (int row = 0; row < kRowsPerDevice; ++row, ++index) { + ASSERT_EQ(tablet.add_timestamp(index, row), common::E_OK); + ASSERT_EQ(tablet.add_value(index, "device", tag.c_str()), + common::E_OK); + for (int c = 0; c < kFields; ++c) { + const std::string name = "s" + std::to_string(c); + ASSERT_EQ( + tablet.add_value(index, name, int64_t(row) * 10 + c), + common::E_OK); + } + } + } + ASSERT_EQ(writer.get_error(), common::E_OK); + ASSERT_EQ(writer.write_table(tablet), common::E_OK); + ASSERT_EQ(writer.flush(), common::E_OK); + ASSERT_EQ(writer.close(), common::E_OK); + } + + std::vector names{"device", "s0", "s1", "s2", "s3", "s4"}; + std::vector name_ptrs; + for (auto& name : names) name_ptrs.push_back(&name[0]); + auto read_all = [&]() { + ERRNO code = RET_OK; + TsFileReader reader = tsfile_reader_new(path.c_str(), &code); + EXPECT_EQ(RET_OK, code); + ResultSet result_set = + tsfile_query_table(reader, "t", name_ptrs.data(), + static_cast(name_ptrs.size()), 0, + std::numeric_limits::max(), &code); + EXPECT_EQ(RET_OK, code); + std::vector rows; + while (tsfile_result_set_next(result_set, &code) && code == RET_OK) { + char* tag = tsfile_result_set_get_value_by_name_string(result_set, + "device"); + std::string line = + tag == nullptr ? std::string() : std::string(tag); + free(tag); + line += ":" + + std::to_string(tsfile_result_set_get_value_by_name_int64_t( + result_set, "time")); + for (int c = 0; c < kFields; ++c) { + const std::string name = "s" + std::to_string(c); + line += ":" + std::to_string( + tsfile_result_set_get_value_by_name_int64_t( + result_set, name.c_str())); + } + rows.push_back(line); + } + EXPECT_EQ(RET_OK, code); + tsfile_reader_close(reader); + free_tsfile_result_set(&result_set); + return rows; + }; + + const bool saved = common::get_parallel_read_enabled(); + EXPECT_EQ(tsfile_set_parallel_read_enabled(0), RET_OK); + const std::vector serial_rows = read_all(); + EXPECT_EQ(tsfile_set_parallel_read_enabled(1), RET_OK); + const std::vector parallel_rows = read_all(); + EXPECT_EQ(tsfile_set_parallel_read_enabled(saved), RET_OK); + + ASSERT_EQ(static_cast(kDevices * kRowsPerDevice), + serial_rows.size()); + EXPECT_EQ(serial_rows, parallel_rows); + EXPECT_EQ("d0:0:0:1:2:3:4", serial_rows.front()); + EXPECT_EQ("d1:499:4990:4991:4992:4993:4994", serial_rows.back()); + remove(path.c_str()); + storage::libtsfile_destroy(); +} + TEST_F(CWrapperTest, CodecAndCompressionConfigIncludesJavaIds) { uint8_t old_int32_encoding = get_datatype_encoding(TS_DATATYPE_INT32); uint8_t old_int64_encoding = get_datatype_encoding(TS_DATATYPE_INT64); diff --git a/python/tests/test_write_and_read.py b/python/tests/test_write_and_read.py index da5288478..4f50b730c 100644 --- a/python/tests/test_write_and_read.py +++ b/python/tests/test_write_and_read.py @@ -669,6 +669,85 @@ def read_rows(backend): os.remove(file_name) +def test_parallel_switches_and_thread_count(): + from tsfile import ( + InvalidArgumentError, + set_parallel_read_enabled, + set_parallel_write_enabled, + set_thread_count, + ) + + file_name = "parallel_switches_python_test.tsfile" + table = TableSchema( + "parallel_table", + [ + ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG), + ColumnSchema("value", TSDataType.INT64, ColumnCategory.FIELD), + ], + ) + devices = 4 + rows_per_device = 500 + + def read_rows(): + rows = [] + with TsFileReader(file_name) as reader: + with reader.query_table( + "parallel_table", ["device", "value"], 0, rows_per_device - 1 + ) as result: + while result.next(): + rows.append( + ( + result.get_value_by_name(TIME_COLUMN), + result.get_value_by_name("device"), + result.get_value_by_name("value"), + ) + ) + return rows + + try: + for invalid in (0, -1, 65): + with pytest.raises(InvalidArgumentError): + set_thread_count(invalid) + set_thread_count(4) + + expected = [ + (row, f"device{device}", row * 3) + for device in range(devices) + for row in range(rows_per_device) + ] + for write_parallel in (False, True): + set_parallel_write_enabled(write_parallel) + if os.path.exists(file_name): + os.remove(file_name) + with TsFileTableWriter(file_name, table) as writer: + tablet = Tablet( + ["device", "value"], + [TSDataType.STRING, TSDataType.INT64], + devices * rows_per_device, + ) + index = 0 + for device in range(devices): + for row in range(rows_per_device): + tablet.add_timestamp(index, row) + tablet.add_value_by_name("device", index, f"device{device}") + tablet.add_value_by_name("value", index, row * 3) + index += 1 + writer.write_table(tablet) + + set_parallel_read_enabled(False) + serial_rows = read_rows() + set_parallel_read_enabled(True) + parallel_rows = read_rows() + assert parallel_rows == serial_rows + assert serial_rows == expected + finally: + set_parallel_read_enabled(True) + set_parallel_write_enabled(True) + set_thread_count(6) + if os.path.exists(file_name): + os.remove(file_name) + + def test_tsfile_to_df(): table = TableSchema( "test_table", From 6d4aebe8670f4211d7c5bdd8711064c9dac0cd11 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 12 Nov 2026 17:39:02 +0800 Subject: [PATCH 3/7] feat(cpp/python): make the parallel switches observable and init-safe Follow-ups from reviewing the new switches: * Setting a parallel flag before the first reader/writer was silently undone: init_config_value() assigned both flags true during the lazy initialization performed by tsfile_reader_new/write_file_new. Give them in-class defaults (as thread_count_ already has) and stop assigning them at init time. Regression test sets both to 0, creates a file handle, and checks they stay 0. * set_thread_count() deleted the live pool before allocating its replacement, so a failed thread creation left g_thread_pool_ dangling and let the exception escape the extern "C" boundary. Build the new pool first, swap it in, then delete the old one; map failures to RET_OOM and keep the previous pool. Skip the join/respawn when the count is unchanged. * Add tsfile_get_parallel_read_enabled/..._write_enabled/ tsfile_get_thread_count and the matching Python getters, and surface all three through get_tsfile_config()/set_tsfile_config(). The setter docs now state that the flags are read per operation and that no reader, writer or result set may be active when resizing the pool. * Update cpp/README.md and python/README.md, which documented non-existent entry points (storage::set_write_thread_count) and described auto-disable on single-core machines. --- cpp/README.md | 20 ++++++++--- cpp/src/common/config/config.h | 8 +++-- cpp/src/common/global.cc | 36 +++++++++++++------ cpp/src/cwrapper/tsfile_cwrapper.cc | 16 +++++++-- cpp/src/cwrapper/tsfile_cwrapper.h | 36 +++++++++++++++---- cpp/test/cwrapper/cwrapper_test.cc | 39 +++++++++++++++++++- python/README.md | 22 ++++++++++++ python/tests/test_write_and_read.py | 55 +++++++++++++++++++++++++++-- python/tsfile/__init__.py | 3 ++ python/tsfile/tsfile_cpp.pxd | 6 ++++ python/tsfile/tsfile_py_cpp.pxd | 3 ++ python/tsfile/tsfile_py_cpp.pyx | 47 ++++++++++++++++++++++-- 12 files changed, 259 insertions(+), 32 deletions(-) diff --git a/cpp/README.md b/cpp/README.md index 8089ac1cd..09a80a887 100644 --- a/cpp/README.md +++ b/cpp/README.md @@ -340,14 +340,26 @@ cmake .. -DENABLE_THREADS=OFF # disable — all thread code is stripped at comp ```cpp #include "common/global.h" -// Enable or disable parallel write at runtime (auto-disabled on single-core machines) +// Enable or disable the parallel write / read paths at runtime. The flags are +// consulted on every write and when a query opens its iterators. storage::set_parallel_write_enabled(true); +storage::set_parallel_read_enabled(true); -// Set the number of worker threads (must be called before creating TsFileWriter) -storage::set_write_thread_count(4); +// Resize the global worker pool (1..64). Takes effect immediately when the +// library has been initialized; do not call it while any reader, writer or +// result set is in use. +storage::set_thread_count(4); ``` -By default, parallel write is enabled when the machine has more than one CPU core, and the thread count is set to the number of hardware cores (capped at 64). +The C API exposes the same switches as +`tsfile_set_parallel_read_enabled()`, `tsfile_set_parallel_write_enabled()` and +`tsfile_set_thread_count()` (plus the matching getters); the Python binding +has `set_parallel_read_enabled()`, `set_parallel_write_enabled()` and +`set_thread_count()`. + +Parallel read and write are enabled by default with a 6-thread pool. With +`ENABLE_THREADS=OFF` the setters only record the values and the parallel paths +are compiled out. ### Local File Read Backend diff --git a/cpp/src/common/config/config.h b/cpp/src/common/config/config.h index 8230614a3..19ff2dd2c 100644 --- a/cpp/src/common/config/config.h +++ b/cpp/src/common/config/config.h @@ -53,8 +53,12 @@ typedef struct ConfigValue { TSEncoding double_encoding_type_; TSEncoding string_encoding_type_; CompressionType default_compression_type_; - bool parallel_read_enabled_; - bool parallel_write_enabled_; + // Unlike the fields above, these two keep their in-class default instead + // of being assigned by init_config_value(): a caller may configure the + // parallel paths before the first reader/writer triggers lazy library + // initialization, and the init-time assignment would silently reset them. + bool parallel_read_enabled_ = true; + bool parallel_write_enabled_ = true; // Size of the single global worker pool (common::g_thread_pool_) shared by // the parallel write and parallel read paths. The pool is (re)created from // this value in init_common(). Like sync_on_close_/encrypt_flag_ it keeps diff --git a/cpp/src/common/global.cc b/cpp/src/common/global.cc index bfc713037..09c685a62 100644 --- a/cpp/src/common/global.cc +++ b/cpp/src/common/global.cc @@ -20,6 +20,8 @@ #include "global.h" #include +#include +#include #ifdef ENABLE_THREADS #include "common/thread_pool.h" @@ -86,10 +88,9 @@ void init_config_value() { #else g_config_value_.default_compression_type_ = UNCOMPRESSED; #endif - g_config_value_.parallel_read_enabled_ = true; - g_config_value_.parallel_write_enabled_ = true; - // thread_count_ keeps its in-class default (see config.h) so a - // set_thread_count() before libtsfile_init() is not reset here. + // parallel_read_enabled_ / parallel_write_enabled_ / thread_count_ keep + // their in-class defaults (see config.h) so setters called before + // libtsfile_init() are not reset here. } extern TSEncoding get_value_encoder(TSDataType data_type) { @@ -178,19 +179,34 @@ int init_common() { int set_thread_count(int32_t count) { if (count < 1 || count > 64) return E_INVALID_ARG; - g_config_value_.thread_count_ = count; #ifdef ENABLE_THREADS // If the global pool already exists (libtsfile_init has run) rebuild it at // the new size so the change takes effect immediately instead of only at // the next libtsfile_init(). This joins all current workers and recreates - // them, so the caller must ensure no read/write is concurrently using the - // pool — intended for setup / benchmark reconfiguration, not mid-operation - // resizing. + // them, so the caller must ensure no reader, writer or result set is using + // the pool — intended for setup / benchmark reconfiguration, not + // mid-operation resizing. if (g_thread_pool_ != nullptr) { - delete g_thread_pool_; - g_thread_pool_ = new ThreadPool(static_cast(count)); + if (count == g_config_value_.thread_count_) { + return E_OK; // already this size; do not join and respawn workers + } + // Build the replacement before touching the old pool so a failure + // (thread creation or allocation) leaves the old one intact, and never + // propagate an exception across the C API boundary. + std::unique_ptr new_pool; + try { + new_pool.reset(new ThreadPool(static_cast(count))); + } catch (const std::exception&) { + return E_OOM; + } + ThreadPool* old_pool = g_thread_pool_; + g_thread_pool_ = new_pool.release(); + g_config_value_.thread_count_ = count; + delete old_pool; + return E_OK; } #endif + g_config_value_.thread_count_ = count; return E_OK; } diff --git a/cpp/src/cwrapper/tsfile_cwrapper.cc b/cpp/src/cwrapper/tsfile_cwrapper.cc index 8bb9cd39f..588561e3e 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.cc +++ b/cpp/src/cwrapper/tsfile_cwrapper.cc @@ -169,20 +169,32 @@ TsFileReadBackend tsfile_get_file_read_backend() { return static_cast(common::get_file_read_backend()); } -ERRNO tsfile_set_parallel_read_enabled(int enabled) { +ERRNO tsfile_set_parallel_read_enabled(int32_t enabled) { common::set_parallel_read_enabled(enabled != 0); return common::E_OK; } -ERRNO tsfile_set_parallel_write_enabled(int enabled) { +int tsfile_get_parallel_read_enabled() { + return common::get_parallel_read_enabled() ? 1 : 0; +} + +ERRNO tsfile_set_parallel_write_enabled(int32_t enabled) { common::set_parallel_write_enabled(enabled != 0); return common::E_OK; } +int tsfile_get_parallel_write_enabled() { + return common::get_parallel_write_enabled() ? 1 : 0; +} + ERRNO tsfile_set_thread_count(int32_t count) { return common::set_thread_count(count); } +int32_t tsfile_get_thread_count() { + return common::g_config_value_.thread_count_; +} + WriteFile write_file_new(const char* pathname, ERRNO* err_code) { if (err_code == nullptr) { return nullptr; diff --git a/cpp/src/cwrapper/tsfile_cwrapper.h b/cpp/src/cwrapper/tsfile_cwrapper.h index 5451b5a01..4012fc9b6 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.h +++ b/cpp/src/cwrapper/tsfile_cwrapper.h @@ -358,31 +358,53 @@ ERRNO tsfile_set_file_read_backend(int32_t backend); TsFileReadBackend tsfile_get_file_read_backend(void); /** - * @brief Enable/disable the parallel read path for subsequently opened - * readers. + * @brief Enable/disable the parallel read path. + * + * The switch is consulted when a query opens its series iterators, so it + * applies to every subsequent read, not only to readers opened afterwards. + * The flags and the pool are read without synchronization: set them before + * starting queries/writes, not while another thread is reading or writing. * * @param enabled Non-zero to enable, zero to disable. * @return RET_OK on success. */ -ERRNO tsfile_set_parallel_read_enabled(int enabled); +ERRNO tsfile_set_parallel_read_enabled(int32_t enabled); + +/** @return Whether the parallel read path is enabled. */ +int tsfile_get_parallel_read_enabled(void); /** - * @brief Enable/disable the parallel write path for subsequently opened - * writers. + * @brief Enable/disable the parallel write path. + * + * Consulted on every write, not only by writers opened afterwards. Same + * concurrency rule as tsfile_set_parallel_read_enabled(). * * @param enabled Non-zero to enable, zero to disable. * @return RET_OK on success. */ -ERRNO tsfile_set_parallel_write_enabled(int enabled); +ERRNO tsfile_set_parallel_write_enabled(int32_t enabled); + +/** @return Whether the parallel write path is enabled. */ +int tsfile_get_parallel_write_enabled(void); /** * @brief Set the size of the global worker thread pool. * + * Takes effect immediately when the library has been initialized (an existing + * pool is joined and recreated). Readers and writers borrow the pool, so this + * must not be called while any reader, writer or result set is in use — it is + * meant for setup / benchmark reconfiguration. With ENABLE_THREADS off this + * only records the value. + * * @param count Number of worker threads, in [1, 64]. - * @return RET_OK on success, or RET_INVALID_ARG for an out-of-range count. + * @return RET_OK on success, RET_INVALID_ARG for an out-of-range count, or + * RET_OOM if the new pool could not be created (the old one is kept). */ ERRNO tsfile_set_thread_count(int32_t count); +/** @return The configured worker thread count (1..64). */ +int32_t tsfile_get_thread_count(void); + /** * @brief Get the encoding type for global time column * diff --git a/cpp/test/cwrapper/cwrapper_test.cc b/cpp/test/cwrapper/cwrapper_test.cc index 290b182a0..dfe13e101 100644 --- a/cpp/test/cwrapper/cwrapper_test.cc +++ b/cpp/test/cwrapper/cwrapper_test.cc @@ -31,7 +31,6 @@ #include "common/global.h" #include "common/row_record.h" -#include "common/tablet.h" #include "cwrapper/tsfile_cwrapper.h" #include "file/write_file.h" #include "reader/result_set.h" @@ -80,26 +79,32 @@ TEST_F(CWrapperTest, ParallelTogglesAndThreadCountValidate) { EXPECT_EQ(tsfile_set_parallel_read_enabled(0), RET_OK); EXPECT_FALSE(common::get_parallel_read_enabled()); + EXPECT_EQ(0, tsfile_get_parallel_read_enabled()); EXPECT_EQ(tsfile_set_parallel_read_enabled(1), RET_OK); EXPECT_TRUE(common::get_parallel_read_enabled()); + EXPECT_EQ(1, tsfile_get_parallel_read_enabled()); // Any non-zero value enables the path, matching the bool parameter. EXPECT_EQ(tsfile_set_parallel_read_enabled(42), RET_OK); EXPECT_TRUE(common::get_parallel_read_enabled()); EXPECT_EQ(tsfile_set_parallel_write_enabled(0), RET_OK); EXPECT_FALSE(common::get_parallel_write_enabled()); + EXPECT_EQ(0, tsfile_get_parallel_write_enabled()); EXPECT_EQ(tsfile_set_parallel_write_enabled(1), RET_OK); EXPECT_TRUE(common::get_parallel_write_enabled()); + EXPECT_EQ(1, tsfile_get_parallel_write_enabled()); // Out-of-range counts are rejected and leave the configured size alone. for (int32_t invalid : {0, -1, 65, 1000}) { EXPECT_EQ(tsfile_set_thread_count(invalid), RET_INVALID_ARG) << invalid; EXPECT_EQ(common::g_config_value_.thread_count_, thread_count) << invalid; + EXPECT_EQ(tsfile_get_thread_count(), thread_count) << invalid; } for (int32_t valid : {1, 32, 64}) { EXPECT_EQ(tsfile_set_thread_count(valid), RET_OK) << valid; EXPECT_EQ(common::g_config_value_.thread_count_, valid) << valid; + EXPECT_EQ(tsfile_get_thread_count(), valid) << valid; } EXPECT_EQ(tsfile_set_thread_count(thread_count), RET_OK); @@ -107,6 +112,31 @@ TEST_F(CWrapperTest, ParallelTogglesAndThreadCountValidate) { EXPECT_EQ(tsfile_set_parallel_write_enabled(write_enabled), RET_OK); } +// The switches are read by the library's lazy initialization. Setting them +// before the first reader/writer used to be undone: init_config_value() +// assigned both flags true, silently re-enabling the parallel paths. +TEST_F(CWrapperTest, ParallelTogglesSurviveLazyInitialization) { + storage::libtsfile_destroy(); + const std::string path = "cwrapper_toggles_before_init.tsfile"; + remove(path.c_str()); + + ASSERT_EQ(tsfile_set_parallel_read_enabled(0), RET_OK); + ASSERT_EQ(tsfile_set_parallel_write_enabled(0), RET_OK); + + ERRNO code = RET_OK; + WriteFile file = write_file_new(path.c_str(), &code); + ASSERT_EQ(RET_OK, code); + ASSERT_NE(nullptr, file); + + EXPECT_EQ(0, tsfile_get_parallel_read_enabled()); + EXPECT_EQ(0, tsfile_get_parallel_write_enabled()); + + free_write_file(&file); + EXPECT_EQ(tsfile_set_parallel_read_enabled(1), RET_OK); + EXPECT_EQ(tsfile_set_parallel_write_enabled(1), RET_OK); + remove(path.c_str()); +} + // Flipping the parallel read switch must not change what a query returns; it // only selects whether the aligned multi-column read uses the worker pool. TEST_F(CWrapperTest, ParallelReadToggleReturnsIdenticalRows) { @@ -164,12 +194,19 @@ TEST_F(CWrapperTest, ParallelReadToggleReturnsIdenticalRows) { ERRNO code = RET_OK; TsFileReader reader = tsfile_reader_new(path.c_str(), &code); EXPECT_EQ(RET_OK, code); + if (reader == nullptr) { + return std::vector(); + } ResultSet result_set = tsfile_query_table(reader, "t", name_ptrs.data(), static_cast(name_ptrs.size()), 0, std::numeric_limits::max(), &code); EXPECT_EQ(RET_OK, code); std::vector rows; + if (result_set == nullptr) { + tsfile_reader_close(reader); + return rows; + } while (tsfile_result_set_next(result_set, &code) && code == RET_OK) { char* tag = tsfile_result_set_get_value_by_name_string(result_set, "device"); diff --git a/python/README.md b/python/README.md index c99d11f9d..7980e91db 100644 --- a/python/README.md +++ b/python/README.md @@ -178,6 +178,28 @@ set_tsfile_config({"file_read_backend_": FileReadBackend.AUTO}) The setting only affects readers opened afterward. Do not modify or truncate a file while it is open through the memory-mapped backend. +## Parallel read, write and worker pool + +```python +from tsfile import ( + get_parallel_read_enabled, + set_parallel_read_enabled, + set_parallel_write_enabled, + set_thread_count, +) + +set_parallel_read_enabled(False) # applies to subsequent reads +set_parallel_write_enabled(True) +set_thread_count(4) # 1..64; applies immediately +print(get_parallel_read_enabled()) +``` + +The switches and the worker pool are read without synchronization, so set them +before starting queries or writes. `set_thread_count()` must not be called while +a reader, writer or result set is in use. All three values are also available +through `get_tsfile_config()` / `set_tsfile_config()` as +`parallel_read_enabled_`, `parallel_write_enabled_` and `thread_count_`. + ## Seekable binary file objects `TsFileReader` also accepts a seekable binary file object. This allows remote diff --git a/python/tests/test_write_and_read.py b/python/tests/test_write_and_read.py index 4f50b730c..747954739 100644 --- a/python/tests/test_write_and_read.py +++ b/python/tests/test_write_and_read.py @@ -17,6 +17,8 @@ # import os +import subprocess +import sys from datetime import date import numpy as np @@ -672,11 +674,20 @@ def read_rows(backend): def test_parallel_switches_and_thread_count(): from tsfile import ( InvalidArgumentError, + get_parallel_read_enabled, + get_parallel_write_enabled, + get_thread_count, + get_tsfile_config, set_parallel_read_enabled, set_parallel_write_enabled, set_thread_count, + set_tsfile_config, ) + original_read = get_parallel_read_enabled() + original_write = get_parallel_write_enabled() + original_threads = get_thread_count() + file_name = "parallel_switches_python_test.tsfile" table = TableSchema( "parallel_table", @@ -709,6 +720,16 @@ def read_rows(): with pytest.raises(InvalidArgumentError): set_thread_count(invalid) set_thread_count(4) + assert get_thread_count() == 4 + + # The switches are readable and also reachable through the config dict. + set_parallel_read_enabled(False) + assert get_parallel_read_enabled() is False + assert get_tsfile_config()["parallel_read_enabled_"] is False + set_tsfile_config({"parallel_read_enabled_": True}) + assert get_parallel_read_enabled() is True + with pytest.raises(TypeError): + set_tsfile_config({"thread_count_": True}) expected = [ (row, f"device{device}", row * 3) @@ -741,13 +762,41 @@ def read_rows(): assert parallel_rows == serial_rows assert serial_rows == expected finally: - set_parallel_read_enabled(True) - set_parallel_write_enabled(True) - set_thread_count(6) + set_parallel_read_enabled(original_read) + set_parallel_write_enabled(original_write) + set_thread_count(original_threads) if os.path.exists(file_name): os.remove(file_name) +def test_parallel_toggles_apply_before_first_use(): + # Lazy initialization used to assign both switches to true, undoing any + # setting made before the first reader/writer existed. A fresh process is + # the only way to exercise that ordering from Python. + script = """ +import tsfile +tsfile.set_parallel_read_enabled(False) +tsfile.set_parallel_write_enabled(False) +tsfile.set_thread_count(3) +try: + tsfile.TsFileReader("missing_before_init.tsfile") +except Exception: + pass +print( + tsfile.get_parallel_read_enabled(), + tsfile.get_parallel_write_enabled(), + tsfile.get_thread_count(), +) +""" + out = subprocess.run( + [sys.executable, "-c", script], + check=True, + capture_output=True, + text=True, + ).stdout.split() + assert out == ["False", "False", "3"] + + def test_tsfile_to_df(): table = TableSchema( "test_table", diff --git a/python/tsfile/__init__.py b/python/tsfile/__init__.py index 4afa123dd..fec2d2e85 100644 --- a/python/tsfile/__init__.py +++ b/python/tsfile/__init__.py @@ -91,6 +91,9 @@ def _preload_dll(path): get_file_read_backend, get_tsfile_config, set_file_read_backend, + get_parallel_read_enabled, + get_parallel_write_enabled, + get_thread_count, set_parallel_read_enabled, set_parallel_write_enabled, set_thread_count, diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd index 80e5946f9..76feba2df 100644 --- a/python/tsfile/tsfile_cpp.pxd +++ b/python/tsfile/tsfile_cpp.pxd @@ -231,8 +231,11 @@ cdef extern from "cwrapper/tsfile_cwrapper.h": ErrorCode tsfile_set_file_read_backend(int32_t backend) TsFileReadBackend tsfile_get_file_read_backend() ErrorCode tsfile_set_parallel_read_enabled(int32_t enabled) + int tsfile_get_parallel_read_enabled() ErrorCode tsfile_set_parallel_write_enabled(int32_t enabled) + int tsfile_get_parallel_write_enabled() ErrorCode tsfile_set_thread_count(int32_t count) + int32_t tsfile_get_thread_count() ctypedef void * TagFilterHandle @@ -530,6 +533,9 @@ cdef extern from "common/config/config.h" namespace "common": uint8_t string_encoding_type_; uint8_t default_compression_type_; bint allow_non_decreasing_timestamps_; + bint parallel_read_enabled_; + bint parallel_write_enabled_; + int32_t thread_count_; cdef extern from "common/global.h" namespace "common": ConfigValue g_config_value_ diff --git a/python/tsfile/tsfile_py_cpp.pxd b/python/tsfile/tsfile_py_cpp.pxd index c3e4c66eb..9e6c60c87 100644 --- a/python/tsfile/tsfile_py_cpp.pxd +++ b/python/tsfile/tsfile_py_cpp.pxd @@ -93,5 +93,8 @@ cpdef public api void set_tsfile_config(dict new_config) cpdef public api object get_file_read_backend() cpdef public api void set_file_read_backend(object backend) cpdef public api void set_parallel_read_enabled(bint enabled) +cpdef public api bint get_parallel_read_enabled() cpdef public api void set_parallel_write_enabled(bint enabled) +cpdef public api bint get_parallel_write_enabled() cpdef public api void set_thread_count(int32_t count) +cpdef public api int32_t get_thread_count() diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx index c67923fe3..67531866b 100644 --- a/python/tsfile/tsfile_py_cpp.pyx +++ b/python/tsfile/tsfile_py_cpp.pyx @@ -915,6 +915,9 @@ cpdef object get_tsfile_config(): "default_compression_type_": CompressorPy(int(g_config_value_.default_compression_type_)), "file_read_backend_": FileReadBackendPy(int(tsfile_get_file_read_backend())), "allow_non_decreasing_timestamps_": g_config_value_.allow_non_decreasing_timestamps_, + "parallel_read_enabled_": bool(g_config_value_.parallel_read_enabled_), + "parallel_write_enabled_": bool(g_config_value_.parallel_write_enabled_), + "thread_count_": g_config_value_.thread_count_, } cpdef object get_file_read_backend(): @@ -928,17 +931,41 @@ cpdef void set_file_read_backend(object backend): check_error(tsfile_set_file_read_backend( int(backend.value))) cpdef void set_parallel_read_enabled(bint enabled): - """Enable/disable the parallel read path for subsequently opened readers.""" + """Enable/disable the parallel read path. + + Applies to every subsequent read. Set it before starting queries; the + flag is read without synchronization. + """ check_error(tsfile_set_parallel_read_enabled( enabled)) +cpdef bint get_parallel_read_enabled(): + """Return whether the parallel read path is enabled.""" + return tsfile_get_parallel_read_enabled() != 0 + cpdef void set_parallel_write_enabled(bint enabled): - """Enable/disable the parallel write path for subsequently opened writers.""" + """Enable/disable the parallel write path. + + Applies to every subsequent write. Same concurrency rule as + set_parallel_read_enabled(). + """ check_error(tsfile_set_parallel_write_enabled( enabled)) +cpdef bint get_parallel_write_enabled(): + """Return whether the parallel write path is enabled.""" + return tsfile_get_parallel_write_enabled() != 0 + cpdef void set_thread_count(int32_t count): - """Set the size of the global worker thread pool (1..64).""" + """Set the size of the global worker thread pool (1..64). + + Takes effect immediately. Must not be called while any reader, writer or + result set is in use. + """ check_error(tsfile_set_thread_count(count)) +cpdef int32_t get_thread_count(): + """Return the configured worker thread count.""" + return tsfile_get_thread_count() + cpdef void set_tsfile_config(dict new_config): if "file_read_backend_" in new_config and not isinstance( new_config["file_read_backend_"], FileReadBackendPy @@ -946,6 +973,15 @@ cpdef void set_tsfile_config(dict new_config): raise TypeError( f"Unsupported FileReadBackend: {new_config['file_read_backend_']}" ) + if "parallel_read_enabled_" in new_config: + _check_bool(new_config["parallel_read_enabled_"]) + set_parallel_read_enabled(new_config["parallel_read_enabled_"]) + if "parallel_write_enabled_" in new_config: + _check_bool(new_config["parallel_write_enabled_"]) + set_parallel_write_enabled(new_config["parallel_write_enabled_"]) + if "thread_count_" in new_config: + _check_int(new_config["thread_count_"]) + set_thread_count(new_config["thread_count_"]) if "tsblock_mem_inc_step_size_" in new_config: _check_uint32(new_config["tsblock_mem_inc_step_size_"]) g_config_value_.tsblock_max_memory_ = new_config["tsblock_mem_inc_step_size_"] @@ -1044,6 +1080,11 @@ cdef _check_bool(value): if not isinstance(value, bool): raise TypeError(f"Expected bool, got {type(value)}") +cdef _check_int(value): + # bool is an int subclass, but it is not a meaningful thread count. + if isinstance(value, bool) or not isinstance(value, int): + raise TypeError(f"Expected int, got {type(value)}") + # Register table and device cdef ErrorCode tsfile_writer_register_device_py_cpp(TsFileWriter writer, DeviceSchema *schema): cdef ErrorCode errno From fd0fda710bf25f8893b0f7e9a74f6389071ef3c2 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 12 Nov 2026 23:11:50 +0800 Subject: [PATCH 4/7] fix(cpp/python): harden the thread-count setter and its Python check Catch (...) when building the replacement pool so no exception can cross the C API boundary, and accept integer-like values (numpy integers, IntEnum) in the Python config setter while still rejecting bool. --- cpp/src/common/global.cc | 3 ++- python/tsfile/tsfile_py_cpp.pyx | 6 ++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/cpp/src/common/global.cc b/cpp/src/common/global.cc index 09c685a62..991ec9097 100644 --- a/cpp/src/common/global.cc +++ b/cpp/src/common/global.cc @@ -196,7 +196,8 @@ int set_thread_count(int32_t count) { std::unique_ptr new_pool; try { new_pool.reset(new ThreadPool(static_cast(count))); - } catch (const std::exception&) { + } catch (...) { + // Nothing may escape into the C API; keep the old pool intact. return E_OOM; } ThreadPool* old_pool = g_thread_pool_; diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx index 67531866b..27dbcb1d6 100644 --- a/python/tsfile/tsfile_py_cpp.pyx +++ b/python/tsfile/tsfile_py_cpp.pyx @@ -20,6 +20,7 @@ from datetime import date as date_type from .date_utils import parse_date_to_int from .tsfile_cpp cimport * +import numbers import os import pandas as pd import numpy as np @@ -1081,8 +1082,9 @@ cdef _check_bool(value): raise TypeError(f"Expected bool, got {type(value)}") cdef _check_int(value): - # bool is an int subclass, but it is not a meaningful thread count. - if isinstance(value, bool) or not isinstance(value, int): + # Accept integer-like values (numpy integers, IntEnum) but not bool, which + # is an int subclass and not a meaningful thread count. + if isinstance(value, bool) or not isinstance(value, numbers.Integral): raise TypeError(f"Expected int, got {type(value)}") # Register table and device From d00c6f190a9beb1315fd516e79006fa35048d2db Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 12 Nov 2026 23:16:58 +0800 Subject: [PATCH 5/7] fix(python): reject huge thread counts with InvalidArgumentError set_thread_count() took int32_t, so Python-int-to-int32 conversion raised OverflowError before the library could return RET_INVALID_ARG. Accept an object, validate it (integer-like, excluding bool) and the [1, 64] range, then convert. Also declare ConfigValue's C++ bool fields as libcpp bool in the pxd instead of bint, so Cython sees the real field type. --- python/tests/test_write_and_read.py | 2 +- python/tsfile/tsfile_cpp.pxd | 9 +++++---- python/tsfile/tsfile_py_cpp.pxd | 2 +- python/tsfile/tsfile_py_cpp.pyx | 13 ++++++++++--- 4 files changed, 17 insertions(+), 9 deletions(-) diff --git a/python/tests/test_write_and_read.py b/python/tests/test_write_and_read.py index 747954739..151a8e17e 100644 --- a/python/tests/test_write_and_read.py +++ b/python/tests/test_write_and_read.py @@ -716,7 +716,7 @@ def read_rows(): return rows try: - for invalid in (0, -1, 65): + for invalid in (0, -1, 65, 2**40): with pytest.raises(InvalidArgumentError): set_thread_count(invalid) set_thread_count(4) diff --git a/python/tsfile/tsfile_cpp.pxd b/python/tsfile/tsfile_cpp.pxd index 76feba2df..24c21475f 100644 --- a/python/tsfile/tsfile_cpp.pxd +++ b/python/tsfile/tsfile_cpp.pxd @@ -18,6 +18,7 @@ #cython: language_level=3 from libc.stdint cimport uint16_t, uint32_t, int32_t, int64_t, uint64_t, uint8_t +from libcpp cimport bool as cppbool ctypedef int32_t ErrorCode @@ -524,7 +525,7 @@ cdef extern from "common/config/config.h" namespace "common": uint8_t time_compress_type_ int32_t chunk_group_size_threshold_ int32_t record_count_for_next_mem_check_ - bint encrypt_flag_ + cppbool encrypt_flag_ uint8_t boolean_encoding_type_; uint8_t int32_encoding_type_; uint8_t int64_encoding_type_; @@ -532,9 +533,9 @@ cdef extern from "common/config/config.h" namespace "common": uint8_t double_encoding_type_; uint8_t string_encoding_type_; uint8_t default_compression_type_; - bint allow_non_decreasing_timestamps_; - bint parallel_read_enabled_; - bint parallel_write_enabled_; + cppbool allow_non_decreasing_timestamps_; + cppbool parallel_read_enabled_; + cppbool parallel_write_enabled_; int32_t thread_count_; cdef extern from "common/global.h" namespace "common": diff --git a/python/tsfile/tsfile_py_cpp.pxd b/python/tsfile/tsfile_py_cpp.pxd index 9e6c60c87..3fad5ab5d 100644 --- a/python/tsfile/tsfile_py_cpp.pxd +++ b/python/tsfile/tsfile_py_cpp.pxd @@ -96,5 +96,5 @@ cpdef public api void set_parallel_read_enabled(bint enabled) cpdef public api bint get_parallel_read_enabled() cpdef public api void set_parallel_write_enabled(bint enabled) cpdef public api bint get_parallel_write_enabled() -cpdef public api void set_thread_count(int32_t count) +cpdef public api void set_thread_count(object count) cpdef public api int32_t get_thread_count() diff --git a/python/tsfile/tsfile_py_cpp.pyx b/python/tsfile/tsfile_py_cpp.pyx index 27dbcb1d6..933af75b2 100644 --- a/python/tsfile/tsfile_py_cpp.pyx +++ b/python/tsfile/tsfile_py_cpp.pyx @@ -955,13 +955,20 @@ cpdef bint get_parallel_write_enabled(): """Return whether the parallel write path is enabled.""" return tsfile_get_parallel_write_enabled() != 0 -cpdef void set_thread_count(int32_t count): +cpdef void set_thread_count(object count): """Set the size of the global worker thread pool (1..64). Takes effect immediately. Must not be called while any reader, writer or - result set is in use. + result set is in use. Accepts any Python integer-like value and raises + InvalidArgumentError outside [1, 64] (instead of OverflowError from the + int32 conversion). """ - check_error(tsfile_set_thread_count(count)) + _check_int(count) + if count < 1 or count > 64: + raise InvalidArgumentError( + context=f"thread count must be in [1, 64], got {count}" + ) + check_error(tsfile_set_thread_count( count)) cpdef int32_t get_thread_count(): """Return the configured worker thread count.""" From eb8d1bf44f3acd9b80088417259c07fbcdc0add6 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Fri, 13 Nov 2026 01:37:29 +0800 Subject: [PATCH 6/7] test(cpp): give reader-test files a per-process unique name Parallel CTest processes can seed the name generator in the same clock tick; one then removes the file another is opening (E_ALREADY_EXIST at the fixture's open, observed as SEGFAULT / test failure on CI). Append pid and a per-process counter, as the writer tests already do. --- cpp/test/reader/tsfile_reader_test.cc | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/cpp/test/reader/tsfile_reader_test.cc b/cpp/test/reader/tsfile_reader_test.cc index 060c7a541..1fe277dde 100644 --- a/cpp/test/reader/tsfile_reader_test.cc +++ b/cpp/test/reader/tsfile_reader_test.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -43,6 +44,12 @@ #include "reader/block/single_device_tsblock_reader.h" #include "reader/filter/time_operator.h" #include "reader/qds_without_timegenerator.h" + +#ifdef _WIN32 +#include +#else +#include +#endif #include "reader/tsfile_series_scan_iterator.h" #include "writer/chunk_writer.h" #include "writer/tsfile_writer.h" @@ -118,6 +125,18 @@ class TsFileReaderTest : public ::testing::Test { random_string += chars[dis(gen)]; } + // CTest runs test cases in parallel processes, and two of them can + // seed the generator in the same clock tick. Without a per-process + // unique suffix one process can remove the file another is using, + // which surfaced as E_ALREADY_EXIST from TsFileWriter::open. +#ifdef _WIN32 + const auto process_id = static_cast(_getpid()); +#else + const auto process_id = static_cast(getpid()); +#endif + static std::atomic counter{0}; + random_string += "_" + std::to_string(process_id) + "_" + + std::to_string(counter.fetch_add(1)); return random_string; } From efc71e26d3c5e514ecb2c4970571f76f2e0887e4 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Fri, 13 Nov 2026 14:53:17 +0800 Subject: [PATCH 7/7] test(cpp): give clock-seeded test files unique names Several fixtures build their .tsfile name from a clock count alone. CTest runs cases in parallel processes, so two can read the same tick and then one removes the file the other is opening or reopening (E_ALREADY_EXIST / E_FILE_OPEN_ERR, seen as SEGFAULT on macOS). Append the process id and a per-process counter, the pattern the writer tests already use. Covers table_read_failure, table_field_filter, chunk_reader_resource, aligned_chunk_reader_bulk, aligned_field_filter_regression, masked_bulk_copy and timeseries_schema_codec. --- .../reader/aligned_chunk_reader_bulk_test.cc | 18 ++++++++++++++++++ .../aligned_field_filter_regression_test.cc | 17 ++++++++++++++++- cpp/test/reader/block/masked_bulk_copy_test.cc | 17 ++++++++++++++++- cpp/test/reader/chunk_reader_resource_test.cc | 17 +++++++++++++++++ cpp/test/reader/table_field_filter_test.cc | 17 ++++++++++++++++- cpp/test/reader/table_read_failure_test.cc | 17 ++++++++++++++++- .../reader/timeseries_schema_codec_test.cc | 17 +++++++++++++++++ 7 files changed, 116 insertions(+), 4 deletions(-) diff --git a/cpp/test/reader/aligned_chunk_reader_bulk_test.cc b/cpp/test/reader/aligned_chunk_reader_bulk_test.cc index 0be62def8..d3efd8796 100644 --- a/cpp/test/reader/aligned_chunk_reader_bulk_test.cc +++ b/cpp/test/reader/aligned_chunk_reader_bulk_test.cc @@ -20,6 +20,7 @@ #include #include +#include #include #include #include @@ -41,6 +42,12 @@ #include "writer/tsfile_table_writer.h" #ifdef ENABLE_THREADS #include "common/thread_pool.h" + +#ifdef _WIN32 +#include +#else +#include +#endif #endif using namespace common; @@ -93,6 +100,17 @@ class AlignedBulkCopyTest : public ::testing::Test { "_" + std::to_string( std::chrono::steady_clock::now().time_since_epoch().count()) + + "_" + std::to_string([] { +#ifdef _WIN32 + return static_cast(_getpid()); +#else + return static_cast(getpid()); +#endif + }()) + + "_" + std::to_string([] { + static std::atomic counter{0}; + return counter.fetch_add(1); + }()) + ".tsfile"; std::remove(path_.c_str()); { diff --git a/cpp/test/reader/aligned_field_filter_regression_test.cc b/cpp/test/reader/aligned_field_filter_regression_test.cc index c2b7506d1..265a075aa 100644 --- a/cpp/test/reader/aligned_field_filter_regression_test.cc +++ b/cpp/test/reader/aligned_field_filter_regression_test.cc @@ -19,6 +19,7 @@ #include #include +#include #include #include #include @@ -37,6 +38,12 @@ #include "reader/table_query_executor.h" #include "writer/tsfile_table_writer.h" +#ifdef _WIN32 +#include +#else +#include +#endif + namespace { using namespace common; using namespace storage; @@ -70,11 +77,19 @@ class AlignedFieldFilterRegressionTest saved_ = g_config_value_; g_config_value_.page_writer_max_point_num_ = 3; g_config_value_.parallel_read_enabled_ = std::get<0>(GetParam()); + // Parallel CTest processes can seed from the same clock tick. +#ifdef _WIN32 + const auto process_id = static_cast(_getpid()); +#else + const auto process_id = static_cast(getpid()); +#endif + static std::atomic filename_counter{0}; filename_ = "aligned_field_filter_regression_" + std::to_string( std::chrono::steady_clock::now().time_since_epoch().count()) + - ".tsfile"; + "_" + std::to_string(process_id) + "_" + + std::to_string(filename_counter.fetch_add(1)) + ".tsfile"; } void TearDown() override { std::remove(filename_.c_str()); diff --git a/cpp/test/reader/block/masked_bulk_copy_test.cc b/cpp/test/reader/block/masked_bulk_copy_test.cc index 21a9e1b50..d2a467a31 100644 --- a/cpp/test/reader/block/masked_bulk_copy_test.cc +++ b/cpp/test/reader/block/masked_bulk_copy_test.cc @@ -19,6 +19,7 @@ #include #include +#include #include #include #include @@ -33,6 +34,12 @@ #include "reader/block/single_device_tsblock_reader.h" #include "writer/tsfile_table_writer.h" +#ifdef _WIN32 +#include +#else +#include +#endif + namespace { using namespace common; using namespace storage; @@ -47,11 +54,19 @@ class MaskedBulkCopyTest : public ::testing::TestWithParam { // Source tuple: time + v + n + s + hidden + empty = 72 bytes/row. // Each source TsBlock has 137 rows, crossing two bitmap words. g_config_value_.tsblock_max_memory_ = 72 * 137; + // Parallel CTest processes can seed from the same clock tick. +#ifdef _WIN32 + const auto process_id = static_cast(_getpid()); +#else + const auto process_id = static_cast(getpid()); +#endif + static std::atomic filename_counter{0}; filename_ = "masked_bulk_copy_" + std::to_string( std::chrono::steady_clock::now().time_since_epoch().count()) + - ".tsfile"; + "_" + std::to_string(process_id) + "_" + + std::to_string(filename_counter.fetch_add(1)) + ".tsfile"; const std::vector columns{ {"v", INT64, UNCOMPRESSED, PLAIN, ColumnCategory::FIELD}, {"n", INT64, UNCOMPRESSED, PLAIN, ColumnCategory::FIELD}, diff --git a/cpp/test/reader/chunk_reader_resource_test.cc b/cpp/test/reader/chunk_reader_resource_test.cc index 7e0a0af73..159db638c 100644 --- a/cpp/test/reader/chunk_reader_resource_test.cc +++ b/cpp/test/reader/chunk_reader_resource_test.cc @@ -41,6 +41,12 @@ #include "writer/tsfile_table_writer.h" #ifdef ENABLE_THREADS #include "common/thread_pool.h" + +#ifdef _WIN32 +#include +#else +#include +#endif #endif namespace storage { @@ -180,6 +186,17 @@ class AlignedReaderFailureTest : public ::testing::Test { "_" + std::to_string( std::chrono::steady_clock::now().time_since_epoch().count()) + + "_" + std::to_string([] { +#ifdef _WIN32 + return static_cast(_getpid()); +#else + return static_cast(getpid()); +#endif + }()) + + "_" + std::to_string([] { + static std::atomic counter{0}; + return counter.fetch_add(1); + }()) + ".tsfile"; WriteFile file; int flags = O_WRONLY | O_CREAT | O_TRUNC; diff --git a/cpp/test/reader/table_field_filter_test.cc b/cpp/test/reader/table_field_filter_test.cc index 62b495a83..a7c9aebc0 100644 --- a/cpp/test/reader/table_field_filter_test.cc +++ b/cpp/test/reader/table_field_filter_test.cc @@ -20,6 +20,7 @@ #include +#include #include #include #include @@ -36,6 +37,12 @@ #include "reader/table_query_executor.h" #include "writer/tsfile_table_writer.h" +#ifdef _WIN32 +#include +#else +#include +#endif + namespace { using namespace common; using namespace storage; @@ -164,11 +171,19 @@ class TableFieldFilterTest saved_ = g_config_value_; g_config_value_.page_writer_max_point_num_ = 3; g_config_value_.parallel_read_enabled_ = std::get<0>(GetParam()); + // Parallel CTest processes can seed from the same clock tick. +#ifdef _WIN32 + const auto process_id = static_cast(_getpid()); +#else + const auto process_id = static_cast(getpid()); +#endif + static std::atomic filename_counter{0}; filename_ = "table_field_filter_" + std::to_string( std::chrono::steady_clock::now().time_since_epoch().count()) + - ".tsfile"; + "_" + std::to_string(process_id) + "_" + + std::to_string(filename_counter.fetch_add(1)) + ".tsfile"; WriteFile file; int flags = O_WRONLY | O_CREAT | O_TRUNC; #ifdef _WIN32 diff --git a/cpp/test/reader/table_read_failure_test.cc b/cpp/test/reader/table_read_failure_test.cc index 9e8076612..99d02669b 100644 --- a/cpp/test/reader/table_read_failure_test.cc +++ b/cpp/test/reader/table_read_failure_test.cc @@ -19,6 +19,7 @@ #include #include +#include #include #include #include @@ -42,6 +43,12 @@ #include "utils/injection.h" #include "writer/tsfile_table_writer.h" +#ifdef _WIN32 +#include +#else +#include +#endif + namespace { class InjectionGuard { @@ -106,11 +113,19 @@ class TableReadFailureTest storage::libtsfile_init(); saved_index_degree_ = common::g_config_value_.max_degree_of_index_node_; ASSERT_EQ(storage::set_max_degree_of_index_node(2), common::E_OK); + // Parallel CTest processes can seed from the same clock tick. +#ifdef _WIN32 + const auto process_id = static_cast(_getpid()); +#else + const auto process_id = static_cast(getpid()); +#endif + static std::atomic filename_counter{0}; filename_ = "table_read_failure_" + std::to_string( std::chrono::steady_clock::now().time_since_epoch().count()) + - ".tsfile"; + "_" + std::to_string(process_id) + "_" + + std::to_string(filename_counter.fetch_add(1)) + ".tsfile"; storage::WriteFile file; int flags = O_WRONLY | O_CREAT | O_TRUNC; #ifdef _WIN32 diff --git a/cpp/test/reader/timeseries_schema_codec_test.cc b/cpp/test/reader/timeseries_schema_codec_test.cc index cebb51541..613dda28a 100644 --- a/cpp/test/reader/timeseries_schema_codec_test.cc +++ b/cpp/test/reader/timeseries_schema_codec_test.cc @@ -19,6 +19,7 @@ #include +#include #include #include #include @@ -31,6 +32,12 @@ #include "reader/tsfile_reader.h" #include "writer/tsfile_writer.h" +#ifdef _WIN32 +#include +#else +#include +#endif + namespace storage { // get_timeseries_schema() reports the encoding and the compressor of every @@ -77,6 +84,16 @@ class TimeseriesSchemaCodecTest : public ::testing::Test { for (int i = 0; i < length; ++i) { s += chars[static_cast(dis(gen))]; } + // CTest runs cases in parallel processes; two can seed from the same + // clock tick and then collide on the file name. +#ifdef _WIN32 + const auto process_id = static_cast(_getpid()); +#else + const auto process_id = static_cast(getpid()); +#endif + static std::atomic filename_counter{0}; + s += "_" + std::to_string(process_id) + "_" + + std::to_string(filename_counter.fetch_add(1)); return s; }