This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new 46cf0a126 Fix remaining C++ memory leak paths (#881)
46cf0a126 is described below
commit 46cf0a126548e7d9292fa5953a0adb53a7eeff26
Author: Colin Lee <[email protected]>
AuthorDate: Thu Jul 30 09:44:30 2026 +0800
Fix remaining C++ memory leak paths (#881)
* fix(cpp): close remaining memory leak paths
* test(cpp): make reader leak tests MSVC-safe
---
cpp/CMakeLists.txt | 3 +
cpp/src/common/allocator/alloc_base.h | 3 +
cpp/src/common/allocator/mem_alloc.cc | 18 +++-
cpp/src/common/container/array.h | 10 ++-
cpp/src/common/container/byte_buffer.h | 23 +++--
cpp/src/common/container/sorted_array.h | 10 ++-
cpp/src/common/tablet.cc | 19 ++--
cpp/src/common/tablet.h | 20 ++++-
cpp/src/compress/lzo_compressor.cc | 82 +++++++++--------
cpp/src/compress/snappy_compressor.cc | 41 ++++-----
cpp/src/reader/aligned_chunk_reader.cc | 119 ++++++++++++++-----------
cpp/src/reader/chunk_reader.cc | 9 +-
cpp/test/common/container/array_test.cc | 14 ++-
cpp/test/common/container/byte_buffer_test.cc | 19 +++-
cpp/test/common/container/sorted_array_test.cc | 13 +++
cpp/test/common/tablet_test.cc | 25 +++++-
cpp/test/compress/lzo_compressor_test.cc | 68 ++++++++++++++
cpp/test/compress/snappy_compressor_test.cc | 20 +++++
cpp/test/reader/chunk_reader_resource_test.cc | 117 ++++++++++++++++++++++++
19 files changed, 492 insertions(+), 141 deletions(-)
diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt
index 5b5fe5249..efdc5ecfb 100755
--- a/cpp/CMakeLists.txt
+++ b/cpp/CMakeLists.txt
@@ -211,6 +211,9 @@ endif ()
option(BUILD_TEST "Build tests" ON)
message("cmake using: BUILD_TEST=${BUILD_TEST}")
+if (BUILD_TEST)
+ add_definitions(-DENABLE_TEST)
+endif ()
option(BUILD_TOOLS "Build the tsfile command-line tools" ON)
message("cmake using: BUILD_TOOLS=${BUILD_TOOLS}")
diff --git a/cpp/src/common/allocator/alloc_base.h
b/cpp/src/common/allocator/alloc_base.h
index dd2e0ab61..2b97c8f83 100644
--- a/cpp/src/common/allocator/alloc_base.h
+++ b/cpp/src/common/allocator/alloc_base.h
@@ -65,6 +65,9 @@ extern TSFILE_API const char* g_mod_names[__LAST_MOD_ID];
void* mem_alloc(uint32_t size, AllocModID mid);
void mem_free(void* ptr);
void* mem_realloc(void* ptr, uint32_t size);
+#ifdef ENABLE_TEST
+TSFILE_API void TEST_fail_next_mem_realloc();
+#endif
class ModStat {
public:
diff --git a/cpp/src/common/allocator/mem_alloc.cc
b/cpp/src/common/allocator/mem_alloc.cc
index b7c5c09c1..deda0b8d8 100644
--- a/cpp/src/common/allocator/mem_alloc.cc
+++ b/cpp/src/common/allocator/mem_alloc.cc
@@ -22,6 +22,7 @@
#endif
#include <string.h>
+#include <atomic>
#include <iomanip>
#include <iostream>
@@ -33,6 +34,16 @@
namespace common {
+#ifdef ENABLE_TEST
+namespace {
+std::atomic<bool> g_fail_next_mem_realloc(false);
+}
+
+void TEST_fail_next_mem_realloc() {
+ g_fail_next_mem_realloc.store(true, std::memory_order_release);
+}
+#endif
+
const char* g_mod_names[__LAST_MOD_ID] = {
/* 0 */ "DEFAULT",
/* 1 */ "TVLIST_DATA",
@@ -139,6 +150,11 @@ void mem_free(void* ptr) {
}
void* mem_realloc(void* ptr, uint32_t size) {
+#ifdef ENABLE_TEST
+ if (g_fail_next_mem_realloc.exchange(false, std::memory_order_acq_rel)) {
+ return nullptr;
+ }
+#endif
char* p = static_cast<char*>(ptr);
char* raw_ptr = p - ALIGNMENT;
const uint64_t header =
@@ -222,4 +238,4 @@ void ModStat::print_stat() {
BaseAllocator g_base_allocator;
-} // end namespace common
\ No newline at end of file
+} // end namespace common
diff --git a/cpp/src/common/container/array.h b/cpp/src/common/container/array.h
index 7a055808f..9bb05d4ad 100644
--- a/cpp/src/common/container/array.h
+++ b/cpp/src/common/container/array.h
@@ -136,12 +136,13 @@ class Array {
}
size_t new_capacity = (size_t)tmp;
- array_ =
+ ValueType* new_array =
(ValueType*)mem_realloc(array_, new_capacity * sizeof(*(array_)));
- if (UNLIKELY(nullptr == array_)) {
+ if (UNLIKELY(nullptr == new_array)) {
// log_err("realloc failed.");
return E_OOM;
}
+ array_ = new_array;
capacity_ = new_capacity;
return E_OK;
}
@@ -150,12 +151,13 @@ class Array {
size_t new_capacity = (size_t)(capacity_ / 2);
// if the size passed to realloc is smaller than before, OS will
// automatically release the rest memory
- array_ =
+ ValueType* new_array =
(ValueType*)mem_realloc(array_, new_capacity * sizeof(*(array_)));
- if (UNLIKELY(nullptr == array_)) {
+ if (UNLIKELY(nullptr == new_array)) {
// log_err("malloc failed.");
return E_OOM;
}
+ array_ = new_array;
capacity_ = new_capacity;
return E_OK;
}
diff --git a/cpp/src/common/container/byte_buffer.h
b/cpp/src/common/container/byte_buffer.h
index 4e2dfab15..0f313acdc 100644
--- a/cpp/src/common/container/byte_buffer.h
+++ b/cpp/src/common/container/byte_buffer.h
@@ -52,13 +52,18 @@ class ByteBuffer {
FORCE_INLINE void reset() { real_data_size_ = 0; }
- FORCE_INLINE void extend_memory(uint32_t new_size) {
+ FORCE_INLINE int extend_memory(uint32_t new_size) {
ASSERT(new_size > reserved_size_);
- data_ = static_cast<char*>(mem_realloc(data_, new_size));
+ char* new_data = static_cast<char*>(mem_realloc(data_, new_size));
+ if (UNLIKELY(new_data == nullptr)) {
+ return E_OOM;
+ }
+ data_ = new_data;
reserved_size_ = new_size;
+ return E_OK;
}
- FORCE_INLINE void append_variable_value(const char* value, uint32_t len) {
+ FORCE_INLINE int append_variable_value(const char* value, uint32_t len) {
// dynamic growth
if (UNLIKELY((real_data_size_ + len + variable_type_len_) >
reserved_size_)) {
@@ -67,7 +72,9 @@ class ByteBuffer {
g_config_value_.tsblock_mem_inc_step_size_ > len
? g_config_value_.tsblock_mem_inc_step_size_
: (len + 1);
- extend_memory(reserved_size_ + growth_size);
+ if (UNLIKELY(extend_memory(reserved_size_ + growth_size) != E_OK))
{
+ return E_OOM;
+ }
}
ASSERT(data_);
@@ -80,9 +87,10 @@ class ByteBuffer {
memcpy(&data_[real_data_size_], value, len);
real_data_size_ += len;
}
+ return E_OK;
}
- FORCE_INLINE void append_fixed_value(const char* value, uint32_t len) {
+ FORCE_INLINE int append_fixed_value(const char* value, uint32_t len) {
// dynamic growth
if (UNLIKELY(real_data_size_ + len > reserved_size_)) {
// extreme scenarios, when encountering very long string
@@ -90,12 +98,15 @@ class ByteBuffer {
g_config_value_.tsblock_mem_inc_step_size_ > len
? g_config_value_.tsblock_mem_inc_step_size_
: (len + 1);
- extend_memory(reserved_size_ + growth_size);
+ if (UNLIKELY(extend_memory(reserved_size_ + growth_size) != E_OK))
{
+ return E_OOM;
+ }
}
ASSERT(data_);
memcpy(&data_[real_data_size_], value, len);
real_data_size_ += len;
+ return E_OK;
}
// for fixed len value
diff --git a/cpp/src/common/container/sorted_array.h
b/cpp/src/common/container/sorted_array.h
index 0713449dd..3e3b29289 100644
--- a/cpp/src/common/container/sorted_array.h
+++ b/cpp/src/common/container/sorted_array.h
@@ -139,12 +139,13 @@ class SortedArray {
}
size_t new_capacity = (size_t)tmp;
- array_ =
+ ValueType* new_array =
(ValueType*)mem_realloc(array_, new_capacity * sizeof(*(array_)));
- if (UNLIKELY(nullptr == array_)) {
+ if (UNLIKELY(nullptr == new_array)) {
// log_err("realloc failed.");
return E_OOM;
}
+ array_ = new_array;
capacity_ = new_capacity;
return E_OK;
}
@@ -153,14 +154,15 @@ class SortedArray {
size_t new_capacity = (size_t)(capacity_ / 2);
// if the size passed to realloc is smaller than before, OS will
// automatically release the rest memory
- array_ = (ValueType*)mem_realloc(
+ ValueType* new_array = (ValueType*)mem_realloc(
array_,
new_capacity *
sizeof(*(array_))); // TODO: user ourself's mem_alloc()
- if (UNLIKELY(nullptr == array_)) {
+ if (UNLIKELY(nullptr == new_array)) {
// log_err("malloc failed.");
return E_OOM;
}
+ array_ = new_array;
capacity_ = new_capacity;
return E_OK;
}
diff --git a/cpp/src/common/tablet.cc b/cpp/src/common/tablet.cc
index 7c8fec561..3bfc54bb9 100644
--- a/cpp/src/common/tablet.cc
+++ b/cpp/src/common/tablet.cc
@@ -423,15 +423,19 @@ void* Tablet::get_value(int row_index, uint32_t
schema_index,
}
template <>
-void Tablet::process_val(uint32_t row_index, uint32_t schema_index,
- common::String str) {
- value_matrix_[schema_index].string_col->append(row_index, str.buf_,
- str.len_);
+int Tablet::process_val(uint32_t row_index, uint32_t schema_index,
+ common::String str) {
+ int ret = value_matrix_[schema_index].string_col->append(
+ row_index, str.buf_, str.len_);
+ if (ret != E_OK) {
+ return ret;
+ }
bitmaps_[schema_index].clear(row_index); /* mark as non-null */
+ return E_OK;
}
template <typename T>
-void Tablet::process_val(uint32_t row_index, uint32_t schema_index, T val) {
+int Tablet::process_val(uint32_t row_index, uint32_t schema_index, T val) {
switch (schema_vec_->at(schema_index).data_type_) {
case common::BOOLEAN:
(value_matrix_[schema_index].bool_data)[row_index] =
@@ -459,6 +463,7 @@ void Tablet::process_val(uint32_t row_index, uint32_t
schema_index, T val) {
ASSERT(false);
}
bitmaps_[schema_index].clear(row_index); /* mark as non-null */
+ return E_OK;
}
template <typename T>
@@ -475,7 +480,7 @@ int Tablet::add_value(uint32_t row_index, uint32_t
schema_index, T val) {
if (UNLIKELY(!TypeMatch<T>(schema.data_type_))) {
return E_TYPE_NOT_MATCH;
}
- process_val(row_index, schema_index, val);
+ ret = process_val(row_index, schema_index, val);
}
return ret;
}
@@ -492,7 +497,7 @@ int Tablet::add_value(uint32_t row_index, uint32_t
schema_index, std::tm val) {
}
int32_t date_int;
if (RET_SUCC(common::DateConverter::date_to_int(val, date_int))) {
- process_val(row_index, schema_index, date_int);
+ ret = process_val(row_index, schema_index, date_int);
}
return ret;
}
diff --git a/cpp/src/common/tablet.h b/cpp/src/common/tablet.h
index c34408494..095157ac7 100644
--- a/cpp/src/common/tablet.h
+++ b/cpp/src/common/tablet.h
@@ -80,16 +80,28 @@ class Tablet {
if (offsets) offsets[0] = 0;
}
- void append(uint32_t row, const char* data, uint32_t len) {
+ int append(uint32_t row, const char* data, uint32_t len) {
// Grow buffer if needed
if (buf_used + len > buf_capacity) {
- buf_capacity = buf_capacity * 2 + len;
- buffer = (char*)common::mem_realloc(buffer, buf_capacity);
+ uint64_t new_capacity_64 =
+ static_cast<uint64_t>(buf_capacity) * 2 + len;
+ if (UNLIKELY(new_capacity_64 > UINT32_MAX)) {
+ return common::E_OVERFLOW;
+ }
+ uint32_t new_capacity = static_cast<uint32_t>(new_capacity_64);
+ char* new_buffer =
+ (char*)common::mem_realloc(buffer, new_capacity);
+ if (UNLIKELY(new_buffer == nullptr)) {
+ return common::E_OOM;
+ }
+ buffer = new_buffer;
+ buf_capacity = new_capacity;
}
memcpy(buffer + buf_used, data, len);
offsets[row] = static_cast<int32_t>(buf_used);
offsets[row + 1] = static_cast<int32_t>(buf_used + len);
buf_used += len;
+ return common::E_OK;
}
const char* get_str(uint32_t row) const {
@@ -393,7 +405,7 @@ class Tablet {
private:
template <typename T>
- void process_val(uint32_t row_index, uint32_t schema_index, T val);
+ int process_val(uint32_t row_index, uint32_t schema_index, T val);
uint32_t max_row_num_;
uint32_t cur_row_size_;
std::string insert_target_name_;
diff --git a/cpp/src/compress/lzo_compressor.cc
b/cpp/src/compress/lzo_compressor.cc
index 0400b039c..8cf049428 100644
--- a/cpp/src/compress/lzo_compressor.cc
+++ b/cpp/src/compress/lzo_compressor.cc
@@ -45,46 +45,53 @@ int LZOCompressor::compress(char* uncompressed_buf,
uint32_t uncompressed_buf_len,
char*& compressed_buf,
uint32_t& compressed_buf_len) {
- int ret = E_OK;
+ compressed_buf = nullptr;
+ compressed_buf_len = 0;
size_t max_dst_size = lzokay::compress_worst_size(uncompressed_buf_len);
- compressed_buf = (char*)mem_alloc(max_dst_size, MOD_COMPRESSOR_OBJ);
- if (compressed_buf == nullptr) {
- ret = E_OOM;
- } else {
- size_t compressed_len = 0;
- uint8_t* srcUint8 = reinterpret_cast<uint8_t*>(uncompressed_buf);
- uint8_t* dstUint8 = reinterpret_cast<uint8_t*>(compressed_buf);
- lzokay::EResult compress_result =
- lzokay::compress(srcUint8, uncompressed_buf_len, dstUint8,
- max_dst_size, compressed_len);
- if (compress_result == lzokay::EResult::Success) {
- char* compress_data = (char*)mem_realloc(dstUint8, compressed_len);
- if (compress_data == nullptr) {
- ret = E_OOM;
- } else {
- compressed_buf = compress_data;
- compressed_buf_ = compress_data;
- compressed_buf_len = compressed_len;
- }
- } else {
- ret = E_COMPRESS_ERR;
- }
+ char* allocated_buf = (char*)mem_alloc(max_dst_size, MOD_COMPRESSOR_OBJ);
+ if (allocated_buf == nullptr) {
+ return E_OOM;
}
- return ret;
+
+ size_t compressed_len = 0;
+ lzokay::EResult compress_result = lzokay::compress(
+ reinterpret_cast<uint8_t*>(uncompressed_buf), uncompressed_buf_len,
+ reinterpret_cast<uint8_t*>(allocated_buf), max_dst_size,
+ compressed_len);
+ if (compress_result != lzokay::EResult::Success) {
+ mem_free(allocated_buf);
+ return E_COMPRESS_ERR;
+ }
+
+ char* resized_buf = (char*)mem_realloc(allocated_buf, compressed_len);
+ if (resized_buf == nullptr) {
+ mem_free(allocated_buf);
+ return E_OOM;
+ }
+
+ compressed_buf = resized_buf;
+ compressed_buf_ = resized_buf;
+ compressed_buf_len = compressed_len;
+ return E_OK;
}
void LZOCompressor::after_compress(char* compressed_buf) {
if (compressed_buf != nullptr) {
mem_free(compressed_buf);
+ if (compressed_buf_ == compressed_buf) {
+ compressed_buf_ = nullptr;
+ }
}
}
int LZOCompressor::uncompress(char* compressed_buf, uint32_t
compressed_buf_len,
char*& uncompressed_buf,
uint32_t& uncompressed_buf_len) {
- int ret = E_OK;
+ uncompressed_buf = nullptr;
+ uncompressed_buf_len = 0;
+ int ret = E_COMPRESS_ERR;
char* regen_buffer = nullptr;
- size_t ulength;
+ size_t ulength = 0;
constexpr float ratio[] = {1.5, 2.5, 3.5, 4.5, 255};
for (uint8_t i = 0; i < UNCOMPRESSED_TIME; ++i) {
regen_buffer =
@@ -101,16 +108,16 @@ int LZOCompressor::uncompress(char* compressed_buf,
uint32_t compressed_buf_len,
regen_buffer = nullptr;
ret = E_COMPRESS_ERR;
} else {
- char* compress_data = (char*)mem_realloc(regen_buffer,
ulength);
- if (regen_buffer == nullptr) {
- ret = E_OOM;
- } else {
- ret = E_OK;
- uncompressed_buf_len = ulength;
- uncompressed_buf_ = compress_data;
- uncompressed_buf = compress_data;
- break;
+ char* resized_buf = (char*)mem_realloc(regen_buffer, ulength);
+ if (resized_buf == nullptr) {
+ mem_free(regen_buffer);
+ return E_OOM;
}
+ ret = E_OK;
+ uncompressed_buf_len = ulength;
+ uncompressed_buf_ = resized_buf;
+ uncompressed_buf = resized_buf;
+ break;
}
}
}
@@ -120,7 +127,10 @@ int LZOCompressor::uncompress(char* compressed_buf,
uint32_t compressed_buf_len,
void LZOCompressor::after_uncompress(char* uncompressed_buf) {
if (uncompressed_buf != nullptr) {
mem_free(uncompressed_buf);
+ if (uncompressed_buf_ == uncompressed_buf) {
+ uncompressed_buf_ = nullptr;
+ }
}
}
-} // end namespace storage
\ No newline at end of file
+} // end namespace storage
diff --git a/cpp/src/compress/snappy_compressor.cc
b/cpp/src/compress/snappy_compressor.cc
index e78a67ac3..c825f2232 100644
--- a/cpp/src/compress/snappy_compressor.cc
+++ b/cpp/src/compress/snappy_compressor.cc
@@ -45,31 +45,28 @@ int SnappyCompressor::compress(char* uncompressed_buf,
uint32_t uncompressed_buf_len,
char*& compressed_buf,
uint32_t& compressed_buf_len) {
- int ret = E_OK;
+ compressed_buf = nullptr;
+ compressed_buf_len = 0;
size_t max_dst_size = snappy::MaxCompressedLength(uncompressed_buf_len);
- compressed_buf = (char*)mem_alloc(max_dst_size, MOD_COMPRESSOR_OBJ);
- if (compressed_buf == nullptr) {
- ret = E_OOM;
- } else {
- size_t compressed_len = 0;
- snappy::RawCompress(uncompressed_buf, uncompressed_buf_len,
- compressed_buf, &compressed_len);
- if (compressed_buf == nullptr) {
- ret = E_COMPRESS_ERR;
- } else {
- char* compressed_data = (char*)mem_realloc(
- compressed_buf, static_cast<uint32_t>(compressed_len));
- if (compressed_data == nullptr) {
- ret = E_OOM;
- } else {
- compressed_buf = compressed_data;
- compressed_buf_ = compressed_data;
- compressed_buf_len = compressed_len;
- }
- }
+ char* allocated_buf = (char*)mem_alloc(max_dst_size, MOD_COMPRESSOR_OBJ);
+ if (allocated_buf == nullptr) {
+ return E_OOM;
}
- return ret;
+ size_t compressed_len = 0;
+ snappy::RawCompress(uncompressed_buf, uncompressed_buf_len, allocated_buf,
+ &compressed_len);
+ char* compressed_data = (char*)mem_realloc(
+ allocated_buf, static_cast<uint32_t>(compressed_len));
+ if (compressed_data == nullptr) {
+ mem_free(allocated_buf);
+ return E_OOM;
+ }
+
+ compressed_buf = compressed_data;
+ compressed_buf_ = compressed_data;
+ compressed_buf_len = compressed_len;
+ return E_OK;
}
void SnappyCompressor::after_compress(char* compressed_buf) {
diff --git a/cpp/src/reader/aligned_chunk_reader.cc
b/cpp/src/reader/aligned_chunk_reader.cc
index 4b89e659b..795ec6b80 100644
--- a/cpp/src/reader/aligned_chunk_reader.cc
+++ b/cpp/src/reader/aligned_chunk_reader.cc
@@ -87,6 +87,12 @@ void AlignedChunkReader::reset() {
time_compressor_->after_uncompress(time_uncompressed_buf_);
time_uncompressed_buf_ = nullptr;
}
+ if (value_uncompressed_buf_ != nullptr && value_compressor_ != nullptr) {
+ value_compressor_->after_uncompress(value_uncompressed_buf_);
+ value_uncompressed_buf_ = nullptr;
+ }
+ time_in_.reset();
+ value_in_.reset();
// Multi-value reset
for (auto* col : value_columns_) {
@@ -250,20 +256,23 @@ int AlignedChunkReader::load_by_aligned_meta(ChunkMeta*
time_chunk_meta,
ret = read_file_->read(time_chunk_meta_->offset_of_chunk_header_,
time_file_data_buf, file_data_time_buf_size_,
ret_read_len);
- if (IS_SUCC(ret) && ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
+ if (!IS_SUCC(ret)) {
+ mem_free(time_file_data_buf);
+ return ret;
+ }
+ if (ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
ret = E_TSFILE_CORRUPTED;
LOGE("file corrupted, ret=" << ret << ", offset="
<<
time_chunk_meta_->offset_of_chunk_header_
<< "read_len=" << ret_read_len);
mem_free(time_file_data_buf);
+ return ret;
}
- if (IS_SUCC(ret)) {
- time_in_stream_.wrap_from(time_file_data_buf, ret_read_len);
- if (RET_FAIL(time_chunk_header_.deserialize_from(time_in_stream_))) {
- } else {
- time_chunk_visit_offset_ = time_in_stream_.read_pos();
- }
+ time_in_stream_.wrap_from(time_file_data_buf, ret_read_len);
+ if (RET_FAIL(time_chunk_header_.deserialize_from(time_in_stream_))) {
+ return ret;
}
+ time_chunk_visit_offset_ = time_in_stream_.read_pos();
/* ================ deserialize value_chunk_header ================*/
ret_read_len = 0;
char* value_file_data_buf =
@@ -274,35 +283,38 @@ int AlignedChunkReader::load_by_aligned_meta(ChunkMeta*
time_chunk_meta,
ret = read_file_->read(value_chunk_meta_->offset_of_chunk_header_,
value_file_data_buf, file_data_value_buf_size_,
ret_read_len);
- if (IS_SUCC(ret) && ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
+ if (!IS_SUCC(ret)) {
+ mem_free(value_file_data_buf);
+ return ret;
+ }
+ if (ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
ret = E_TSFILE_CORRUPTED;
LOGE("file corrupted, ret="
<< ret << ", offset=" <<
value_chunk_meta_->offset_of_chunk_header_
<< "read_len=" << ret_read_len);
mem_free(value_file_data_buf);
+ return ret;
}
- if (IS_SUCC(ret)) {
- value_in_stream_.wrap_from(value_file_data_buf, ret_read_len);
- if (RET_FAIL(value_chunk_header_.deserialize_from(value_in_stream_))) {
- } else if (RET_FAIL(alloc_compressor_and_decoder(
- time_decoder_, time_compressor_,
- time_chunk_header_.encoding_type_,
- time_chunk_header_.data_type_,
- time_chunk_header_.compression_type_))) {
- } else if (RET_FAIL(alloc_compressor_and_decoder(
- value_decoder_, value_compressor_,
- value_chunk_header_.encoding_type_,
- value_chunk_header_.data_type_,
- value_chunk_header_.compression_type_))) {
- } else {
- value_chunk_visit_offset_ = value_in_stream_.read_pos();
+ value_in_stream_.wrap_from(value_file_data_buf, ret_read_len);
+ if (RET_FAIL(value_chunk_header_.deserialize_from(value_in_stream_))) {
+ } else if (RET_FAIL(alloc_compressor_and_decoder(
+ time_decoder_, time_compressor_,
+ time_chunk_header_.encoding_type_,
+ time_chunk_header_.data_type_,
+ time_chunk_header_.compression_type_))) {
+ } else if (RET_FAIL(alloc_compressor_and_decoder(
+ value_decoder_, value_compressor_,
+ value_chunk_header_.encoding_type_,
+ value_chunk_header_.data_type_,
+ value_chunk_header_.compression_type_))) {
+ } else {
+ value_chunk_visit_offset_ = value_in_stream_.read_pos();
#if DEBUG_SE
- std::cout << "AlignedChunkReader::load_by_meta, time_chunk_header="
- << time_chunk_header_
- << ", value_chunk_header=" << value_chunk_header_
- << std::endl;
+ std::cout << "AlignedChunkReader::load_by_meta, time_chunk_header="
+ << time_chunk_header_
+ << ", value_chunk_header=" << value_chunk_header_
+ << std::endl;
#endif
- }
}
return ret;
}
@@ -445,11 +457,11 @@ int AlignedChunkReader::read_from_file_and_rewrap(
(want_size < DEFAULT_READ_SIZE ? DEFAULT_READ_SIZE : want_size);
if (file_data_buf_size < read_size ||
(may_shrink && read_size < file_data_buf_size / 10)) {
- file_data_buf = (char*)mem_realloc(file_data_buf, read_size);
- if (IS_NULL(file_data_buf)) {
- in_stream_.clear_wrapped_buf();
+ char* resized_buf = (char*)mem_realloc(file_data_buf, read_size);
+ if (IS_NULL(resized_buf)) {
return E_OOM;
}
+ file_data_buf = resized_buf;
file_data_buf_size = read_size;
// Update stream pointer immediately so it stays valid even if
// the subsequent read fails and the caller frees via destroy().
@@ -1190,18 +1202,20 @@ int AlignedChunkReader::load_by_aligned_meta_multi(
ret = read_file_->read(time_chunk_meta_->offset_of_chunk_header_,
time_file_data_buf, file_data_time_buf_size_,
ret_read_len);
- if (IS_SUCC(ret) && ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
+ if (!IS_SUCC(ret)) {
+ mem_free(time_file_data_buf);
+ return ret;
+ }
+ if (ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
ret = E_TSFILE_CORRUPTED;
mem_free(time_file_data_buf);
return ret;
}
- if (IS_SUCC(ret)) {
- time_in_stream_.wrap_from(time_file_data_buf, ret_read_len);
- if (RET_FAIL(time_chunk_header_.deserialize_from(time_in_stream_))) {
- return ret;
- }
- time_chunk_visit_offset_ = time_in_stream_.read_pos();
+ time_in_stream_.wrap_from(time_file_data_buf, ret_read_len);
+ if (RET_FAIL(time_chunk_header_.deserialize_from(time_in_stream_))) {
+ return ret;
}
+ time_chunk_visit_offset_ = time_in_stream_.read_pos();
// Alloc time decoder/compressor
if (IS_SUCC(ret)) {
@@ -1236,24 +1250,25 @@ int AlignedChunkReader::load_by_aligned_meta_multi(
ret = read_file_->read(col->chunk_meta->offset_of_chunk_header_, vbuf,
col->file_data_buf_size, ret_read_len);
- if (IS_SUCC(ret) && ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
+ if (!IS_SUCC(ret)) {
+ mem_free(vbuf);
+ return ret;
+ }
+ if (ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
ret = E_TSFILE_CORRUPTED;
mem_free(vbuf);
+ return ret;
+ }
+ col->in_stream.wrap_from(vbuf, ret_read_len);
+ if (RET_FAIL(col->chunk_header.deserialize_from(col->in_stream))) {
break;
}
- if (IS_SUCC(ret)) {
- col->in_stream.wrap_from(vbuf, ret_read_len);
- if (RET_FAIL(col->chunk_header.deserialize_from(col->in_stream))) {
- break;
- }
- col->chunk_visit_offset = col->in_stream.read_pos();
- if (RET_FAIL(alloc_compressor_and_decoder(
- col->decoder, col->compressor,
- col->chunk_header.encoding_type_,
- col->chunk_header.data_type_,
- col->chunk_header.compression_type_))) {
- break;
- }
+ col->chunk_visit_offset = col->in_stream.read_pos();
+ if (RET_FAIL(alloc_compressor_and_decoder(
+ col->decoder, col->compressor,
col->chunk_header.encoding_type_,
+ col->chunk_header.data_type_,
+ col->chunk_header.compression_type_))) {
+ break;
}
}
diff --git a/cpp/src/reader/chunk_reader.cc b/cpp/src/reader/chunk_reader.cc
index f127ee2ee..ce8d87a58 100644
--- a/cpp/src/reader/chunk_reader.cc
+++ b/cpp/src/reader/chunk_reader.cc
@@ -254,11 +254,16 @@ int ChunkReader::read_from_file_and_rewrap(int want_size)
{
(want_size < DEFAULT_READ_SIZE ? DEFAULT_READ_SIZE : want_size);
if (file_data_buf_size_ < read_size ||
read_size < file_data_buf_size_ / 10) {
- file_data_buf = (char*)mem_realloc(file_data_buf, read_size);
- if (IS_NULL(file_data_buf)) {
+ char* resized_buf = (char*)mem_realloc(file_data_buf, read_size);
+ if (IS_NULL(resized_buf)) {
return E_OOM;
}
+ file_data_buf = resized_buf;
file_data_buf_size_ = read_size;
+ // mem_realloc() may move the allocation. Keep the stream's external
+ // pointer synchronized even if the subsequent file read fails, so
+ // reset()/destroy() can still release the live buffer.
+ in_stream_.wrap_from(file_data_buf, read_size);
}
int ret_read_len = 0;
if (RET_FAIL(
diff --git a/cpp/test/common/container/array_test.cc
b/cpp/test/common/container/array_test.cc
index 661ac2c2c..cda5b596d 100644
--- a/cpp/test/common/container/array_test.cc
+++ b/cpp/test/common/container/array_test.cc
@@ -165,4 +165,16 @@ TEST_F(ArrayTest, CapacityShrink) {
EXPECT_EQ(arr.size(), 0);
}
-} // namespace common
\ No newline at end of file
+TEST_F(ArrayTest, ReallocFailurePreservesAllocation) {
+ common::Array<int> arr(1);
+ ASSERT_EQ(arr.init(), E_OK);
+ ASSERT_EQ(arr.append(7), E_OK);
+
+ common::TEST_fail_next_mem_realloc();
+ EXPECT_EQ(arr.append(8), E_OOM);
+ EXPECT_EQ(arr.size(), 1);
+ EXPECT_EQ(arr.capacity(), 1);
+ EXPECT_EQ(arr[0], 7);
+}
+
+} // namespace common
diff --git a/cpp/test/common/container/byte_buffer_test.cc
b/cpp/test/common/container/byte_buffer_test.cc
index 4f9c63f06..a28c3c650 100644
--- a/cpp/test/common/container/byte_buffer_test.cc
+++ b/cpp/test/common/container/byte_buffer_test.cc
@@ -66,4 +66,21 @@ TEST_F(ByteBufferTest, ExtendMemory) {
EXPECT_STREQ(read_value, value);
}
-} // namespace
\ No newline at end of file
+TEST_F(ByteBufferTest, ReallocFailurePreservesAllocation) {
+ common::ByteBuffer byte_buffer;
+ byte_buffer.init(4);
+ const char first[] = {'a', 'b', 'c', 'd'};
+ const char second[] = {'e', 'f', 'g', 'h'};
+ ASSERT_EQ(byte_buffer.append_fixed_value(first, sizeof(first)),
+ common::E_OK);
+ char* original_data = byte_buffer.get_data();
+
+ common::TEST_fail_next_mem_realloc();
+ EXPECT_EQ(byte_buffer.append_fixed_value(second, sizeof(second)),
+ common::E_OOM);
+ EXPECT_EQ(byte_buffer.get_data(), original_data);
+ EXPECT_EQ(byte_buffer.get_data_size(), sizeof(first));
+ EXPECT_EQ(memcmp(byte_buffer.get_data(), first, sizeof(first)), 0);
+}
+
+} // namespace
diff --git a/cpp/test/common/container/sorted_array_test.cc
b/cpp/test/common/container/sorted_array_test.cc
index 85e55dd46..fce8f806a 100644
--- a/cpp/test/common/container/sorted_array_test.cc
+++ b/cpp/test/common/container/sorted_array_test.cc
@@ -154,4 +154,17 @@ TEST(SortedArrayTest, ShrinkArray) {
array.destroy();
}
+TEST(SortedArrayTest, ReallocFailurePreservesAllocation) {
+ SortedArray<int> array(1);
+ ASSERT_EQ(array.init(), E_OK);
+ ASSERT_EQ(array.insert(7), E_OK);
+
+ common::TEST_fail_next_mem_realloc();
+ EXPECT_EQ(array.insert(8), E_OOM);
+ EXPECT_EQ(array.size(), 1);
+ EXPECT_EQ(array.capacity(), 1);
+ EXPECT_EQ(array[0], 7);
+ array.destroy();
+}
+
} // namespace common
diff --git a/cpp/test/common/tablet_test.cc b/cpp/test/common/tablet_test.cc
index 11dfa485f..ce75ca2ee 100644
--- a/cpp/test/common/tablet_test.cc
+++ b/cpp/test/common/tablet_test.cc
@@ -148,6 +148,29 @@ TEST(TabletTest, StringRepeatedTotalBytesOverflowRejected)
{
common::E_OVERFLOW);
}
+TEST(TabletTest, StringReallocFailureReturnsOomAndPreservesColumn) {
+ std::vector<MeasurementSchema> schema_vec;
+ schema_vec.push_back(MeasurementSchema(
+ "m_str", common::TSDataType::STRING, common::TSEncoding::PLAIN,
+ common::CompressionType::UNCOMPRESSED));
+ Tablet tablet("dev",
+ std::make_shared<std::vector<MeasurementSchema>>(schema_vec),
+ 1u);
+
+ std::string oversized_value(64, 'x');
+ common::TEST_fail_next_mem_realloc();
+ EXPECT_EQ(tablet.add_value(0u, 0u, common::String(oversized_value)),
+ common::E_OOM);
+
+ common::String value("ok", 2);
+ ASSERT_EQ(tablet.add_value(0u, 0u, value), common::E_OK);
+ common::TSDataType type;
+ auto* stored = static_cast<common::String*>(tablet.get_value(0u, 0u,
type));
+ ASSERT_NE(stored, nullptr);
+ EXPECT_EQ(stored->len_, 2u);
+ EXPECT_EQ(memcmp(stored->buf_, "ok", 2), 0);
+}
+
// Regression: set_column_string_values only checked offsets[count] before;
// non-monotonic / negative / non-zero-start offsets would underflow the
// downstream `offsets[i+1] - offsets[i]` length calc and trigger wild
@@ -199,4 +222,4 @@ TEST(TabletTest, LargeQuantities) {
EXPECT_EQ(tablet.get_column_count(), schema_vec.size());
}
-} // namespace storage
\ No newline at end of file
+} // namespace storage
diff --git a/cpp/test/compress/lzo_compressor_test.cc
b/cpp/test/compress/lzo_compressor_test.cc
index efc5534dc..88179b573 100644
--- a/cpp/test/compress/lzo_compressor_test.cc
+++ b/cpp/test/compress/lzo_compressor_test.cc
@@ -126,4 +126,72 @@ TEST_F(LZOTest, TestBytes2) {
compressor.after_compress(compressed_buf);
compressor.after_uncompress(decompressed_buf);
}
+
+TEST_F(LZOTest, DestroyAfterReleasedBuffersIsSafe) {
+ storage::LZOCompressor compressor;
+ std::string input(4096, 'L');
+ char* compressed = nullptr;
+ uint32_t compressed_len = 0;
+ ASSERT_EQ(compressor.compress(&input[0], input.size(), compressed,
+ compressed_len),
+ common::E_OK);
+
+ char* uncompressed = nullptr;
+ uint32_t uncompressed_len = 0;
+ ASSERT_EQ(compressor.uncompress(compressed, compressed_len, uncompressed,
+ uncompressed_len),
+ common::E_OK);
+ compressor.after_compress(compressed);
+ compressor.after_uncompress(uncompressed);
+
+ compressor.destroy();
+ compressor.destroy();
+}
+
+TEST_F(LZOTest, CompressReallocFailureReleasesTemporaryBuffer) {
+ storage::LZOCompressor compressor;
+ std::string input(1024, 'L');
+ char* compressed = nullptr;
+ uint32_t compressed_len = 0;
+ int64_t memory_before =
+ common::ModStat::get_instance().get_stat(common::MOD_COMPRESSOR_OBJ);
+
+ common::TEST_fail_next_mem_realloc();
+ EXPECT_EQ(compressor.compress(&input[0], input.size(), compressed,
+ compressed_len),
+ common::E_OOM);
+ EXPECT_EQ(compressed, nullptr);
+ EXPECT_EQ(compressed_len, 0u);
+ EXPECT_EQ(
+ common::ModStat::get_instance().get_stat(common::MOD_COMPRESSOR_OBJ),
+ memory_before);
+ compressor.destroy();
+}
+
+TEST_F(LZOTest, UncompressReallocFailureReleasesTemporaryBuffer) {
+ storage::LZOCompressor compressor;
+ std::string input(1024, 'L');
+ char* compressed = nullptr;
+ uint32_t compressed_len = 0;
+ ASSERT_EQ(compressor.compress(&input[0], input.size(), compressed,
+ compressed_len),
+ common::E_OK);
+
+ char* uncompressed = nullptr;
+ uint32_t uncompressed_len = 0;
+ int64_t memory_before =
+ common::ModStat::get_instance().get_stat(common::MOD_COMPRESSOR_OBJ);
+ common::TEST_fail_next_mem_realloc();
+ EXPECT_EQ(compressor.uncompress(compressed, compressed_len, uncompressed,
+ uncompressed_len),
+ common::E_OOM);
+ EXPECT_EQ(uncompressed, nullptr);
+ EXPECT_EQ(uncompressed_len, 0u);
+ EXPECT_EQ(
+ common::ModStat::get_instance().get_stat(common::MOD_COMPRESSOR_OBJ),
+ memory_before);
+
+ compressor.after_compress(compressed);
+ compressor.destroy();
+}
} // namespace
diff --git a/cpp/test/compress/snappy_compressor_test.cc
b/cpp/test/compress/snappy_compressor_test.cc
index 249200cce..5ad3a569d 100644
--- a/cpp/test/compress/snappy_compressor_test.cc
+++ b/cpp/test/compress/snappy_compressor_test.cc
@@ -162,4 +162,24 @@ TEST_F(SnappyTest, AfterUncompressFreesParamNotMember) {
compressor.after_compress(compressed_a);
compressor.after_compress(compressed_b);
}
+
+TEST_F(SnappyTest, ReallocFailureReleasesTemporaryBuffer) {
+ storage::SnappyCompressor compressor;
+ std::string input(1024, 'S');
+ char* compressed = nullptr;
+ uint32_t compressed_len = 0;
+ int64_t memory_before =
+ common::ModStat::get_instance().get_stat(common::MOD_COMPRESSOR_OBJ);
+
+ common::TEST_fail_next_mem_realloc();
+ EXPECT_EQ(compressor.compress(&input[0], input.size(), compressed,
+ compressed_len),
+ common::E_OOM);
+ EXPECT_EQ(compressed, nullptr);
+ EXPECT_EQ(compressed_len, 0u);
+ EXPECT_EQ(
+ common::ModStat::get_instance().get_stat(common::MOD_COMPRESSOR_OBJ),
+ memory_before);
+ compressor.destroy();
+}
} // namespace
diff --git a/cpp/test/reader/chunk_reader_resource_test.cc
b/cpp/test/reader/chunk_reader_resource_test.cc
new file mode 100644
index 000000000..11d456a4d
--- /dev/null
+++ b/cpp/test/reader/chunk_reader_resource_test.cc
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <gtest/gtest.h>
+
+#include <cstdio>
+#include <fstream>
+#include <new>
+#include <string>
+#include <vector>
+
+#include "common/allocator/alloc_base.h"
+#include "reader/aligned_chunk_reader.h"
+
+namespace storage {
+namespace {
+
+class TempTsFile {
+ public:
+ explicit TempTsFile(const char* path) : path_(path) {
+ std::remove(path_.c_str());
+ std::ofstream out(path_, std::ios::binary | std::ios::trunc);
+ out.write(MAGIC_STRING_TSFILE, MAGIC_STRING_TSFILE_LEN);
+ out.put(VERSION_NUM_BYTE);
+ out.write(MAGIC_STRING_TSFILE, MAGIC_STRING_TSFILE_LEN);
+ }
+
+ ~TempTsFile() { std::remove(path_.c_str()); }
+
+ const std::string& path() const { return path_; }
+
+ private:
+ std::string path_;
+};
+
+AlignedChunkReader* allocate_reader(ReadFile* read_file) {
+ void* memory =
+ common::mem_alloc(sizeof(AlignedChunkReader),
common::MOD_CHUNK_READER);
+ if (memory == nullptr) {
+ return nullptr;
+ }
+ auto* reader = new (memory) AlignedChunkReader();
+ if (reader->init(read_file, common::String("s", 1), common::INT64,
+ nullptr) != common::E_OK) {
+ reader->destroy();
+ common::mem_free(reader);
+ return nullptr;
+ }
+ return reader;
+}
+
+void free_reader(AlignedChunkReader* reader) {
+ reader->destroy();
+ common::mem_free(reader);
+}
+
+TEST(ChunkReaderResourceTest, AlignedInitialShortReadReleasesBuffer) {
+ TempTsFile temp_file("aligned_initial_short_read.tsfile");
+ ReadFile read_file;
+ ASSERT_EQ(read_file.open(temp_file.path()), common::E_OK);
+ AlignedChunkReader* reader = allocate_reader(&read_file);
+ ASSERT_NE(reader, nullptr);
+ ChunkMeta time_meta;
+ time_meta.offset_of_chunk_header_ = read_file.file_size();
+ ChunkMeta value_meta;
+ int64_t memory_before =
+ common::ModStat::get_instance().get_stat(common::MOD_CHUNK_READER);
+
+ EXPECT_EQ(reader->load_by_aligned_meta(&time_meta, &value_meta),
+ common::E_TSFILE_CORRUPTED);
+ EXPECT_EQ(
+ common::ModStat::get_instance().get_stat(common::MOD_CHUNK_READER),
+ memory_before);
+
+ free_reader(reader);
+}
+
+TEST(ChunkReaderResourceTest, MultiAlignedInitialShortReadReleasesBuffer) {
+ TempTsFile temp_file("multi_aligned_initial_short_read.tsfile");
+ ReadFile read_file;
+ ASSERT_EQ(read_file.open(temp_file.path()), common::E_OK);
+ AlignedChunkReader* reader = allocate_reader(&read_file);
+ ASSERT_NE(reader, nullptr);
+ ChunkMeta time_meta;
+ time_meta.offset_of_chunk_header_ = read_file.file_size();
+ ChunkMeta value_meta;
+ std::vector<ChunkMeta*> value_metas{&value_meta};
+ int64_t memory_before =
+ common::ModStat::get_instance().get_stat(common::MOD_CHUNK_READER);
+
+ EXPECT_EQ(reader->load_by_aligned_meta_multi(&time_meta, value_metas),
+ common::E_TSFILE_CORRUPTED);
+ EXPECT_EQ(
+ common::ModStat::get_instance().get_stat(common::MOD_CHUNK_READER),
+ memory_before);
+
+ free_reader(reader);
+}
+
+} // namespace
+} // namespace storage
