diff --git a/cpp/CHANGELOG.md b/cpp/CHANGELOG.md index 1ebe05eb21fb91317f8ad389d3e88d6d85753262..0520eaf331d8d66ed6b94228a1f83ee2160bf8b5 100644 --- a/cpp/CHANGELOG.md +++ b/cpp/CHANGELOG.md @@ -11,6 +11,7 @@ Please mark all change in change log and use the ticket from JIRA. - MS-89 - Fix compile failed, libgpufaiss.a link missing - MS-90 - Fix arch match incorrect on ARM - MS-99 - Fix compilation bug +- MS-110 - Avoid huge file size ## Improvement - MS-82 - Update server startup welcome message diff --git a/cpp/src/db/DBImpl.cpp b/cpp/src/db/DBImpl.cpp index e66471d5c31ed604010583181af927e6c6d330b3..a0abb5e1ad3ff4f10b5e0fd4ba7d3ed6f04cdd72 100644 --- a/cpp/src/db/DBImpl.cpp +++ b/cpp/src/db/DBImpl.cpp @@ -472,7 +472,7 @@ void DBImpl::StartCompactionTask() { } //serialize memory data - std::vector temp_table_ids; + std::set temp_table_ids; mem_mgr_->Serialize(temp_table_ids); for(auto& id : temp_table_ids) { compact_table_ids_.insert(id); @@ -543,7 +543,8 @@ Status DBImpl::MergeFiles(const std::string& table_id, const meta::DateT& date, ENGINE_LOG_DEBUG << "New merged file " << table_file.file_id_ << " of size=" << index->PhysicalSize()/(1024*1024) << " M"; - index->Cache(); + //current disable this line to avoid memory + //index->Cache(); return status; } @@ -659,7 +660,8 @@ Status DBImpl::BuildIndex(const meta::TableFileSchema& file) { << index->PhysicalSize()/(1024*1024) << " M" << " from file " << to_remove.file_id_; - index->Cache(); + //current disable this line to avoid memory + //index->Cache(); } catch (std::exception& ex) { return Status::Error("Build index encounter exception", ex.what()); @@ -698,7 +700,7 @@ Status DBImpl::Size(uint64_t& result) { DBImpl::~DBImpl() { shutting_down_.store(true, std::memory_order_release); bg_timer_thread_.join(); - std::vector ids; + std::set ids; mem_mgr_->Serialize(ids); } diff --git a/cpp/src/db/MemManager.cpp b/cpp/src/db/MemManager.cpp index d7a2087ee04edd96766e0c666206df41f0174663..fa7f3c54b083abf031876b458b75669ad6607444 100644 --- a/cpp/src/db/MemManager.cpp +++ b/cpp/src/db/MemManager.cpp @@ -20,36 +20,54 @@ namespace engine { MemVectors::MemVectors(const std::shared_ptr& meta_ptr, const meta::TableFileSchema& schema, const Options& options) - : pMeta_(meta_ptr), + : meta_(meta_ptr), options_(options), schema_(schema), - pIdGenerator_(new SimpleIDGenerator()), - pEE_(EngineFactory::Build(schema_.dimension_, schema_.location_, (EngineType)schema_.engine_type_)) { + id_generator_(new SimpleIDGenerator()), + active_engine_(EngineFactory::Build(schema_.dimension_, schema_.location_, (EngineType)schema_.engine_type_)) { } -void MemVectors::Add(size_t n_, const float* vectors_, IDNumbers& vector_ids_) { +Status MemVectors::Add(size_t n_, const float* vectors_, IDNumbers& vector_ids_) { + if(active_engine_ == nullptr) { + return Status::Error("index engine is null"); + } + auto start_time = METRICS_NOW_TIME; - pIdGenerator_->GetNextIDNumbers(n_, vector_ids_); - pEE_->AddWithIds(n_, vectors_, vector_ids_.data()); + id_generator_->GetNextIDNumbers(n_, vector_ids_); + Status status = active_engine_->AddWithIds(n_, vectors_, vector_ids_.data()); auto end_time = METRICS_NOW_TIME; auto total_time = METRICS_MICROSECONDS(start_time, end_time); server::Metrics::GetInstance().AddVectorsPerSecondGaugeSet(static_cast(n_), static_cast(schema_.dimension_), total_time); + + return status; } -size_t MemVectors::Total() const { - return pEE_->Count(); +size_t MemVectors::RowCount() const { + if(active_engine_ == nullptr) { + return 0; + } + + return active_engine_->Count(); } -size_t MemVectors::ApproximateSize() const { - return pEE_->Size(); +size_t MemVectors::Size() const { + if(active_engine_ == nullptr) { + return 0; + } + + return active_engine_->Size(); } Status MemVectors::Serialize(std::string& table_id) { + if(active_engine_ == nullptr) { + return Status::Error("index engine is null"); + } + table_id = schema_.table_id_; - auto size = ApproximateSize(); + auto size = Size(); auto start_time = METRICS_NOW_TIME; - pEE_->Serialize(); + active_engine_->Serialize(); auto end_time = METRICS_NOW_TIME; auto total_time = METRICS_MICROSECONDS(start_time, end_time); schema_.size_ = size; @@ -59,20 +77,20 @@ Status MemVectors::Serialize(std::string& table_id) { schema_.file_type_ = (size >= options_.index_trigger_size) ? meta::TableFileSchema::TO_INDEX : meta::TableFileSchema::RAW; - auto status = pMeta_->UpdateTableFile(schema_); + auto status = meta_->UpdateTableFile(schema_); LOG(DEBUG) << "New " << ((schema_.file_type_ == meta::TableFileSchema::RAW) ? "raw" : "to_index") - << " file " << schema_.file_id_ << " of size " << (double)(pEE_->Size()) / (double)meta::M << " M"; + << " file " << schema_.file_id_ << " of size " << (double)(active_engine_->Size()) / (double)meta::M << " M"; - pEE_->Cache(); + active_engine_->Cache(); return status; } MemVectors::~MemVectors() { - if (pIdGenerator_ != nullptr) { - delete pIdGenerator_; - pIdGenerator_ = nullptr; + if (id_generator_ != nullptr) { + delete id_generator_; + id_generator_ = nullptr; } } @@ -81,20 +99,20 @@ MemVectors::~MemVectors() { */ MemManager::MemVectorsPtr MemManager::GetMemByTable( const std::string& table_id) { - auto memIt = memMap_.find(table_id); - if (memIt != memMap_.end()) { + auto memIt = mem_id_map_.find(table_id); + if (memIt != mem_id_map_.end()) { return memIt->second; } meta::TableFileSchema table_file; table_file.table_id_ = table_id; - auto status = pMeta_->CreateTableFile(table_file); + auto status = meta_->CreateTableFile(table_file); if (!status.ok()) { return nullptr; } - memMap_[table_id] = MemVectorsPtr(new MemVectors(pMeta_, table_file, options_)); - return memMap_[table_id]; + mem_id_map_[table_id] = MemVectorsPtr(new MemVectors(meta_, table_file, options_)); + return mem_id_map_[table_id]; } Status MemManager::InsertVectors(const std::string& table_id_, @@ -114,37 +132,44 @@ Status MemManager::InsertVectorsNoLock(const std::string& table_id, if (mem == nullptr) { return Status::NotFound("Group " + table_id + " not found!"); } - mem->Add(n, vectors, vector_ids); - return Status::OK(); + //makesure each file size less than index_trigger_size + if(mem->Size() > options_.index_trigger_size) { + std::unique_lock lock(serialization_mtx_); + immu_mem_list_.push_back(mem); + mem_id_map_.erase(table_id); + return InsertVectorsNoLock(table_id, n, vectors, vector_ids); + } else { + return mem->Add(n, vectors, vector_ids); + } } Status MemManager::ToImmutable() { std::unique_lock lock(mutex_); - for (auto& kv: memMap_) { - immMems_.push_back(kv.second); + for (auto& kv: mem_id_map_) { + immu_mem_list_.push_back(kv.second); } - memMap_.clear(); + mem_id_map_.clear(); return Status::OK(); } -Status MemManager::Serialize(std::vector& table_ids) { +Status MemManager::Serialize(std::set& table_ids) { ToImmutable(); std::unique_lock lock(serialization_mtx_); std::string table_id; table_ids.clear(); - for (auto& mem : immMems_) { + for (auto& mem : immu_mem_list_) { mem->Serialize(table_id); - table_ids.push_back(table_id); + table_ids.insert(table_id); } - immMems_.clear(); + immu_mem_list_.clear(); return Status::OK(); } Status MemManager::EraseMemVector(const std::string& table_id) { std::unique_lock lock(mutex_); - memMap_.erase(table_id); + mem_id_map_.erase(table_id); return Status::OK(); } diff --git a/cpp/src/db/MemManager.h b/cpp/src/db/MemManager.h index 1b329f971b07261382bac062d69c4b6bd43d4ab1..2aa01838984d54852f4c8223180c3e7946603fdd 100644 --- a/cpp/src/db/MemManager.h +++ b/cpp/src/db/MemManager.h @@ -15,6 +15,7 @@ #include #include #include +#include namespace zilliz { namespace milvus { @@ -32,11 +33,11 @@ public: explicit MemVectors(const std::shared_ptr&, const meta::TableFileSchema&, const Options&); - void Add(size_t n_, const float* vectors_, IDNumbers& vector_ids_); + Status Add(size_t n_, const float* vectors_, IDNumbers& vector_ids_); - size_t Total() const; + size_t RowCount() const; - size_t ApproximateSize() const; + size_t Size() const; Status Serialize(std::string& table_id); @@ -49,11 +50,11 @@ private: MemVectors(const MemVectors&) = delete; MemVectors& operator=(const MemVectors&) = delete; - MetaPtr pMeta_; + MetaPtr meta_; Options options_; meta::TableFileSchema schema_; - IDGenerator* pIdGenerator_; - ExecutionEnginePtr pEE_; + IDGenerator* id_generator_; + ExecutionEnginePtr active_engine_; }; // MemVectors @@ -66,14 +67,14 @@ public: using Ptr = std::shared_ptr; MemManager(const std::shared_ptr& meta, const Options& options) - : pMeta_(meta), options_(options) {} + : meta_(meta), options_(options) {} MemVectorsPtr GetMemByTable(const std::string& table_id); Status InsertVectors(const std::string& table_id, size_t n, const float* vectors, IDNumbers& vector_ids); - Status Serialize(std::vector& table_ids); + Status Serialize(std::set& table_ids); Status EraseMemVector(const std::string& table_id); @@ -82,11 +83,11 @@ private: size_t n, const float* vectors, IDNumbers& vector_ids); Status ToImmutable(); - using MemMap = std::map; - using ImmMemPool = std::vector; - MemMap memMap_; - ImmMemPool immMems_; - MetaPtr pMeta_; + using MemIdMap = std::map; + using MemList = std::vector; + MemIdMap mem_id_map_; + MemList immu_mem_list_; + MetaPtr meta_; Options options_; std::mutex mutex_; std::mutex serialization_mtx_;