DBImpl.cpp 24.1 KB
Newer Older
X
Xu Peng 已提交
1 2 3 4 5
/*******************************************************************************
 * Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
 * Unauthorized copying of this file, via any medium is strictly prohibited.
 * Proprietary and confidential.
 ******************************************************************************/
6 7
#include "DBImpl.h"
#include "DBMetaImpl.h"
G
groot 已提交
8
#include "Log.h"
G
groot 已提交
9
#include "EngineFactory.h"
Z
update  
zhiru 已提交
10
#include "Factories.h"
G
groot 已提交
11
#include "metrics/Metrics.h"
G
groot 已提交
12
#include "scheduler/TaskScheduler.h"
J
jinhai 已提交
13

G
groot 已提交
14
#include "scheduler/context/DeleteContext.h"
G
groot 已提交
15
#include "utils/TimeRecorder.h"
Z
update  
zhiru 已提交
16
#include "MetaConsts.h"
X
Xu Peng 已提交
17

X
Xu Peng 已提交
18
#include <assert.h>
X
Xu Peng 已提交
19
#include <chrono>
X
Xu Peng 已提交
20
#include <thread>
21
#include <iostream>
X
xj.lin 已提交
22
#include <cstring>
X
Xu Peng 已提交
23
#include <cache/CpuCacheMgr.h>
G
groot 已提交
24
#include <boost/filesystem.hpp>
X
Xu Peng 已提交
25

X
Xu Peng 已提交
26
namespace zilliz {
J
jinhai 已提交
27
namespace milvus {
X
Xu Peng 已提交
28
namespace engine {
X
Xu Peng 已提交
29

G
groot 已提交
30 31
namespace {

J
jinhai 已提交
32 33 34
constexpr uint64_t METRIC_ACTION_INTERVAL = 1;
constexpr uint64_t COMPACT_ACTION_INTERVAL = 1;
constexpr uint64_t INDEX_ACTION_INTERVAL = 1;
G
groot 已提交
35

G
groot 已提交
36 37 38 39 40
void CollectInsertMetrics(double total_time, size_t n, bool succeed) {
    double avg_time = total_time / n;
    for (int i = 0; i < n; ++i) {
        server::Metrics::GetInstance().AddVectorsDurationHistogramOberve(avg_time);
    }
Y
yu yunfeng 已提交
41

G
groot 已提交
42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61
//    server::Metrics::GetInstance().add_vector_duration_seconds_quantiles().Observe((average_time));
    if (succeed) {
        server::Metrics::GetInstance().AddVectorsSuccessTotalIncrement(n);
        server::Metrics::GetInstance().AddVectorsSuccessGaugeSet(n);
    }
    else {
        server::Metrics::GetInstance().AddVectorsFailTotalIncrement(n);
        server::Metrics::GetInstance().AddVectorsFailGaugeSet(n);
    }
}

void CollectQueryMetrics(double total_time, size_t nq) {
    for (int i = 0; i < nq; ++i) {
        server::Metrics::GetInstance().QueryResponseSummaryObserve(total_time);
    }
    auto average_time = total_time / nq;
    server::Metrics::GetInstance().QueryVectorResponseSummaryObserve(average_time, nq);
    server::Metrics::GetInstance().QueryVectorResponsePerSecondGaugeSet(double (nq) / total_time);
}

G
groot 已提交
62
void CollectFileMetrics(int file_type, size_t file_size, double total_time) {
G
groot 已提交
63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81
    switch(file_type) {
        case meta::TableFileSchema::RAW:
        case meta::TableFileSchema::TO_INDEX: {
            server::Metrics::GetInstance().SearchRawDataDurationSecondsHistogramObserve(total_time);
            server::Metrics::GetInstance().RawFileSizeHistogramObserve(file_size);
            server::Metrics::GetInstance().RawFileSizeTotalIncrement(file_size);
            server::Metrics::GetInstance().RawFileSizeGaugeSet(file_size);
            break;
        }
        default: {
            server::Metrics::GetInstance().SearchIndexDataDurationSecondsHistogramObserve(total_time);
            server::Metrics::GetInstance().IndexFileSizeHistogramObserve(file_size);
            server::Metrics::GetInstance().IndexFileSizeTotalIncrement(file_size);
            server::Metrics::GetInstance().IndexFileSizeGaugeSet(file_size);
            break;
        }
    }
}
}
Y
yu yunfeng 已提交
82

G
groot 已提交
83 84

DBImpl::DBImpl(const Options& options)
G
groot 已提交
85
    : options_(options),
X
Xu Peng 已提交
86
      shutting_down_(false),
G
groot 已提交
87 88
      compact_thread_pool_(1, 1),
      index_thread_pool_(1, 1) {
Z
update  
zhiru 已提交
89
    meta_ptr_ = DBMetaImplFactory::Build(options.meta, options.mode);
Z
zhiru 已提交
90
    mem_mgr_ = MemManagerFactory::Build(meta_ptr_, options_);
Z
update  
zhiru 已提交
91
    if (options.mode != Options::MODE::READ_ONLY) {
92
        ENGINE_LOG_TRACE << "StartTimerTasks";
Z
update  
zhiru 已提交
93 94
        StartTimerTasks();
    }
S
starlord 已提交
95 96


X
Xu Peng 已提交
97 98
}

G
groot 已提交
99
Status DBImpl::CreateTable(meta::TableSchema& table_schema) {
G
groot 已提交
100
    return meta_ptr_->CreateTable(table_schema);
101 102
}

G
groot 已提交
103
Status DBImpl::DeleteTable(const std::string& table_id, const meta::DatesT& dates) {
G
groot 已提交
104
    //dates partly delete files of the table but currently we don't support
S
starlord 已提交
105
    ENGINE_LOG_DEBUG << "Prepare to delete table " << table_id;
G
groot 已提交
106 107 108 109 110 111 112 113

    mem_mgr_->EraseMemVector(table_id); //not allow insert
    meta_ptr_->DeleteTable(table_id); //soft delete table

    //scheduler will determine when to delete table files
    TaskScheduler& scheduler = TaskScheduler::GetInstance();
    DeleteContextPtr context = std::make_shared<DeleteContext>(table_id, meta_ptr_);
    scheduler.Schedule(context);
G
groot 已提交
114 115 116 117

    return Status::OK();
}

G
groot 已提交
118
Status DBImpl::DescribeTable(meta::TableSchema& table_schema) {
G
groot 已提交
119
    return meta_ptr_->DescribeTable(table_schema);
120 121
}

G
groot 已提交
122
Status DBImpl::HasTable(const std::string& table_id, bool& has_or_not) {
G
groot 已提交
123
    return meta_ptr_->HasTable(table_id, has_or_not);
124 125
}

G
groot 已提交
126
Status DBImpl::AllTables(std::vector<meta::TableSchema>& table_schema_array) {
G
groot 已提交
127
    return meta_ptr_->AllTables(table_schema_array);
G
groot 已提交
128 129 130
}

Status DBImpl::GetTableRowCount(const std::string& table_id, uint64_t& row_count) {
G
groot 已提交
131
    return meta_ptr_->Count(table_id, row_count);
G
groot 已提交
132 133
}

G
groot 已提交
134
Status DBImpl::InsertVectors(const std::string& table_id_,
G
groot 已提交
135
        uint64_t n, const float* vectors, IDNumbers& vector_ids_) {
S
starlord 已提交
136
    ENGINE_LOG_DEBUG << "Insert " << n << " vectors to cache";
Y
yu yunfeng 已提交
137 138

    auto start_time = METRICS_NOW_TIME;
G
groot 已提交
139
    Status status = mem_mgr_->InsertVectors(table_id_, n, vectors, vector_ids_);
Y
yu yunfeng 已提交
140
    auto end_time = METRICS_NOW_TIME;
G
groot 已提交
141
    double total_time = METRICS_MICROSECONDS(start_time,end_time);
Y
yu yunfeng 已提交
142 143 144
//    std::chrono::microseconds time_span = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);
//    double average_time = double(time_span.count()) / n;

S
starlord 已提交
145 146
    ENGINE_LOG_DEBUG << "Insert vectors to cache finished";

G
groot 已提交
147 148
    CollectInsertMetrics(total_time, n, status.ok());
    return status;
Y
yu yunfeng 已提交
149

X
Xu Peng 已提交
150 151
}

G
groot 已提交
152
Status DBImpl::Query(const std::string &table_id, uint64_t k, uint64_t nq,
X
xj.lin 已提交
153
                      const float *vectors, QueryResults &results) {
Y
yu yunfeng 已提交
154
    auto start_time = METRICS_NOW_TIME;
X
Xu Peng 已提交
155
    meta::DatesT dates = {meta::Meta::GetDate()};
Y
yu yunfeng 已提交
156 157 158
    Status result = Query(table_id, k, nq, vectors, dates, results);
    auto end_time = METRICS_NOW_TIME;
    auto total_time = METRICS_MICROSECONDS(start_time,end_time);
G
groot 已提交
159 160

    CollectQueryMetrics(total_time, nq);
Y
yu yunfeng 已提交
161

Y
yu yunfeng 已提交
162
    return result;
X
Xu Peng 已提交
163 164
}

G
groot 已提交
165
Status DBImpl::Query(const std::string& table_id, uint64_t k, uint64_t nq,
X
Xu Peng 已提交
166
        const float* vectors, const meta::DatesT& dates, QueryResults& results) {
S
starlord 已提交
167 168
    ENGINE_LOG_DEBUG << "Query by vectors";

169 170
    //get all table files from table
    meta::DatePartionedTableFilesSchema files;
G
groot 已提交
171
    auto status = meta_ptr_->FilesToSearch(table_id, dates, files);
172 173 174 175 176 177 178 179 180
    if (!status.ok()) { return status; }

    meta::TableFilesSchema file_id_array;
    for (auto &day_files : files) {
        for (auto &file : day_files.second) {
            file_id_array.push_back(file);
        }
    }

S
starlord 已提交
181 182 183 184
    cache::CpuCacheMgr::GetInstance()->PrintInfo(); //print cache info before query
    status = QueryAsync(table_id, file_id_array, k, nq, vectors, dates, results);
    cache::CpuCacheMgr::GetInstance()->PrintInfo(); //print cache info after query
    return status;
G
groot 已提交
185
}
X
Xu Peng 已提交
186

187 188 189
Status DBImpl::Query(const std::string& table_id, const std::vector<std::string>& file_ids,
        uint64_t k, uint64_t nq, const float* vectors,
        const meta::DatesT& dates, QueryResults& results) {
S
starlord 已提交
190 191
    ENGINE_LOG_DEBUG << "Query by file ids";

192
    //get specified files
193
    std::vector<size_t> ids;
194 195
    for (auto &id : file_ids) {
        meta::TableFileSchema table_file;
196 197
        table_file.table_id_ = table_id;
        std::string::size_type sz;
J
jinhai 已提交
198
        ids.push_back(std::stoul(id, &sz));
199 200 201 202 203 204
    }

    meta::TableFilesSchema files_array;
    auto status = meta_ptr_->GetTableFiles(table_id, ids, files_array);
    if (!status.ok()) {
        return status;
205 206
    }

G
groot 已提交
207 208 209 210
    if(files_array.empty()) {
        return Status::Error("Invalid file id");
    }

S
starlord 已提交
211 212 213 214
    cache::CpuCacheMgr::GetInstance()->PrintInfo(); //print cache info before query
    status = QueryAsync(table_id, files_array, k, nq, vectors, dates, results);
    cache::CpuCacheMgr::GetInstance()->PrintInfo(); //print cache info after query
    return status;
215 216 217 218 219
}

Status DBImpl::QueryAsync(const std::string& table_id, const meta::TableFilesSchema& files,
                          uint64_t k, uint64_t nq, const float* vectors,
                          const meta::DatesT& dates, QueryResults& results) {
K
kun yu 已提交
220
    auto start_time = METRICS_NOW_TIME;
S
starlord 已提交
221
    server::TimeRecorder rc("");
G
groot 已提交
222 223

    //step 1: get files to search
S
starlord 已提交
224
    ENGINE_LOG_DEBUG << "Engine query begin, index file count:" << files.size() << " date range count:" << dates.size();
G
groot 已提交
225
    SearchContextPtr context = std::make_shared<SearchContext>(k, nq, vectors);
226 227 228
    for (auto &file : files) {
        TableFileSchemaPtr file_ptr = std::make_shared<meta::TableFileSchema>(file);
        context->AddIndexFile(file_ptr);
G
groot 已提交
229 230
    }

G
groot 已提交
231
    //step 2: put search task to scheduler
G
groot 已提交
232 233
    TaskScheduler& scheduler = TaskScheduler::GetInstance();
    scheduler.Schedule(context);
G
groot 已提交
234 235

    context->WaitResult();
G
groot 已提交
236

S
starlord 已提交
237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259
    //step 3: print time cost information
    double load_cost = context->LoadCost();
    double search_cost = context->SearchCost();
    double reduce_cost = context->ReduceCost();
    std::string load_info = server::TimeRecorder::GetTimeSpanStr(load_cost);
    std::string search_info = server::TimeRecorder::GetTimeSpanStr(search_cost);
    std::string reduce_info = server::TimeRecorder::GetTimeSpanStr(reduce_cost);
    if(search_cost > 0.0 || reduce_cost > 0.0) {
        double total_cost = load_cost + search_cost + reduce_cost;
        double load_percent = load_cost/total_cost;
        double search_percent = search_cost/total_cost;
        double reduce_percent = reduce_cost/total_cost;

        ENGINE_LOG_DEBUG << "Engine load index totally cost:" << load_info << " percent: " << load_percent*100 << "%";
        ENGINE_LOG_DEBUG << "Engine search index totally cost:" << search_info << " percent: " << search_percent*100 << "%";
        ENGINE_LOG_DEBUG << "Engine reduce topk totally cost:" << reduce_info << " percent: " << reduce_percent*100 << "%";
    } else {
        ENGINE_LOG_DEBUG << "Engine load cost:" << load_info
            << " search cost: " << search_info
            << " reduce cost: " << reduce_info;
    }

    //step 4: construct results
J
jinhai 已提交
260
    results = context->GetResult();
S
starlord 已提交
261
    rc.ElapseFromBegin("Engine query totally cost");
G
groot 已提交
262

K
kun yu 已提交
263 264 265 266 267
    auto end_time = METRICS_NOW_TIME;
    auto total_time = METRICS_MICROSECONDS(start_time,end_time);

    CollectQueryMetrics(total_time, nq);

G
groot 已提交
268 269 270
    return Status::OK();
}

G
groot 已提交
271 272
void DBImpl::StartTimerTasks() {
    bg_timer_thread_ = std::thread(&DBImpl::BackgroundTimerTask, this);
X
Xu Peng 已提交
273 274
}

G
groot 已提交
275
void DBImpl::BackgroundTimerTask() {
X
Xu Peng 已提交
276
    Status status;
Y
yu yunfeng 已提交
277
    server::SystemInfo::GetInstance().Init();
X
Xu Peng 已提交
278
    while (true) {
G
groot 已提交
279 280 281 282 283 284 285
        if (shutting_down_.load(std::memory_order_acquire)){
            for(auto& iter : compact_thread_results_) {
                iter.wait();
            }
            for(auto& iter : index_thread_results_) {
                iter.wait();
            }
S
starlord 已提交
286 287

            ENGINE_LOG_DEBUG << "DB background thread exit";
G
groot 已提交
288 289
            break;
        }
X
Xu Peng 已提交
290

G
groot 已提交
291
        std::this_thread::sleep_for(std::chrono::seconds(1));
X
Xu Peng 已提交
292

G
groot 已提交
293
        StartMetricTask();
G
groot 已提交
294 295 296
        StartCompactionTask();
        StartBuildIndexTask();
    }
X
Xu Peng 已提交
297 298
}

G
groot 已提交
299 300 301 302 303 304 305
void DBImpl::StartMetricTask() {
    static uint64_t metric_clock_tick = 0;
    metric_clock_tick++;
    if(metric_clock_tick%METRIC_ACTION_INTERVAL != 0) {
        return;
    }

306
    ENGINE_LOG_TRACE << "Start metric task";
S
starlord 已提交
307

G
groot 已提交
308 309 310 311 312 313 314 315 316 317 318 319
    server::Metrics::GetInstance().KeepingAliveCounterIncrement(METRIC_ACTION_INTERVAL);
    int64_t cache_usage = cache::CpuCacheMgr::GetInstance()->CacheUsage();
    int64_t cache_total = cache::CpuCacheMgr::GetInstance()->CacheCapacity();
    server::Metrics::GetInstance().CacheUsageGaugeSet(cache_usage*100/cache_total);
    uint64_t size;
    Size(size);
    server::Metrics::GetInstance().DataFileSizeGaugeSet(size);
    server::Metrics::GetInstance().CPUUsagePercentSet();
    server::Metrics::GetInstance().RAMUsagePercentSet();
    server::Metrics::GetInstance().GPUPercentGaugeSet();
    server::Metrics::GetInstance().GPUMemoryUsageGaugeSet();
    server::Metrics::GetInstance().OctetsSet();
S
starlord 已提交
320

K
kun yu 已提交
321
    server::Metrics::GetInstance().CPUCoreUsagePercentSet();
K
kun yu 已提交
322 323
    server::Metrics::GetInstance().GPUTemperature();
    server::Metrics::GetInstance().CPUTemperature();
K
kun yu 已提交
324

325
    ENGINE_LOG_TRACE << "Metric task finished";
G
groot 已提交
326 327
}

G
groot 已提交
328
void DBImpl::StartCompactionTask() {
G
groot 已提交
329 330 331 332 333 334
    static uint64_t compact_clock_tick = 0;
    compact_clock_tick++;
    if(compact_clock_tick%COMPACT_ACTION_INTERVAL != 0) {
        return;
    }

G
groot 已提交
335
    //serialize memory data
G
groot 已提交
336
    std::set<std::string> temp_table_ids;
G
groot 已提交
337
    mem_mgr_->Serialize(temp_table_ids);
G
groot 已提交
338 339 340
    for(auto& id : temp_table_ids) {
        compact_table_ids_.insert(id);
    }
X
Xu Peng 已提交
341

342 343 344
    if(!temp_table_ids.empty()) {
        SERVER_LOG_DEBUG << "Insert cache serialized";
    }
S
starlord 已提交
345

G
groot 已提交
346 347 348 349 350 351 352
    //compactiong has been finished?
    if(!compact_thread_results_.empty()) {
        std::chrono::milliseconds span(10);
        if (compact_thread_results_.back().wait_for(span) == std::future_status::ready) {
            compact_thread_results_.pop_back();
        }
    }
X
Xu Peng 已提交
353

G
groot 已提交
354 355 356 357 358 359
    //add new compaction task
    if(compact_thread_results_.empty()) {
        compact_thread_results_.push_back(
                compact_thread_pool_.enqueue(&DBImpl::BackgroundCompaction, this, compact_table_ids_));
        compact_table_ids_.clear();
    }
X
Xu Peng 已提交
360 361
}

G
groot 已提交
362
Status DBImpl::MergeFiles(const std::string& table_id, const meta::DateT& date,
363
        const meta::TableFilesSchema& files) {
S
starlord 已提交
364
    ENGINE_LOG_DEBUG << "Merge files for table " << table_id;
S
starlord 已提交
365

X
Xu Peng 已提交
366
    meta::TableFileSchema table_file;
G
groot 已提交
367 368
    table_file.table_id_ = table_id;
    table_file.date_ = date;
369
    table_file.file_type_ = meta::TableFileSchema::NEW_MERGE;
G
groot 已提交
370
    Status status = meta_ptr_->CreateTableFile(table_file);
X
Xu Peng 已提交
371

372
    if (!status.ok()) {
S
starlord 已提交
373
        ENGINE_LOG_ERROR << "Failed to create table: " << status.ToString();
374 375 376
        return status;
    }

G
groot 已提交
377 378
    ExecutionEnginePtr index =
            EngineFactory::Build(table_file.dimension_, table_file.location_, (EngineType)table_file.engine_type_);
379

380
    meta::TableFilesSchema updated;
X
Xu Peng 已提交
381
    long  index_size = 0;
382 383

    for (auto& file : files) {
Y
yu yunfeng 已提交
384 385

        auto start_time = METRICS_NOW_TIME;
G
groot 已提交
386
        index->Merge(file.location_);
387
        auto file_schema = file;
Y
yu yunfeng 已提交
388 389
        auto end_time = METRICS_NOW_TIME;
        auto total_time = METRICS_MICROSECONDS(start_time,end_time);
Y
yu yunfeng 已提交
390
        server::Metrics::GetInstance().MemTableMergeDurationSecondsHistogramObserve(total_time);
Y
yu yunfeng 已提交
391

G
groot 已提交
392
        file_schema.file_type_ = meta::TableFileSchema::TO_DELETE;
393
        updated.push_back(file_schema);
G
groot 已提交
394
        ENGINE_LOG_DEBUG << "Merging file " << file_schema.file_id_;
G
groot 已提交
395
        index_size = index->Size();
X
Xu Peng 已提交
396

X
Xu Peng 已提交
397
        if (index_size >= options_.index_trigger_size) break;
398 399
    }

Y
yu yunfeng 已提交
400

G
groot 已提交
401
    index->Serialize();
X
Xu Peng 已提交
402

X
Xu Peng 已提交
403
    if (index_size >= options_.index_trigger_size) {
G
groot 已提交
404
        table_file.file_type_ = meta::TableFileSchema::TO_INDEX;
X
Xu Peng 已提交
405
    } else {
G
groot 已提交
406
        table_file.file_type_ = meta::TableFileSchema::RAW;
X
Xu Peng 已提交
407
    }
G
groot 已提交
408
    table_file.size_ = index_size;
X
Xu Peng 已提交
409
    updated.push_back(table_file);
G
groot 已提交
410 411
    status = meta_ptr_->UpdateTableFiles(updated);
    ENGINE_LOG_DEBUG << "New merged file " << table_file.file_id_ <<
S
starlord 已提交
412
        " of size " << index->PhysicalSize() << " bytes";
413

S
starlord 已提交
414 415 416
    if(options_.insert_cache_immediately_) {
        index->Cache();
    }
X
Xu Peng 已提交
417

418 419 420
    return status;
}

G
groot 已提交
421
Status DBImpl::BackgroundMergeFiles(const std::string& table_id) {
422
    meta::DatePartionedTableFilesSchema raw_files;
G
groot 已提交
423
    auto status = meta_ptr_->FilesToMerge(table_id, raw_files);
X
Xu Peng 已提交
424
    if (!status.ok()) {
S
starlord 已提交
425
        ENGINE_LOG_ERROR << "Failed to get merge files for table: " << table_id;
X
Xu Peng 已提交
426 427
        return status;
    }
428

X
Xu Peng 已提交
429
    bool has_merge = false;
430
    for (auto& kv : raw_files) {
X
Xu Peng 已提交
431
        auto files = kv.second;
S
starlord 已提交
432 433
        if (files.size() < options_.merge_trigger_number) {
            ENGINE_LOG_DEBUG << "Files number not greater equal than merge trigger number, skip merge action";
X
Xu Peng 已提交
434 435
            continue;
        }
X
Xu Peng 已提交
436
        has_merge = true;
X
Xu Peng 已提交
437
        MergeFiles(table_id, kv.first, kv.second);
G
groot 已提交
438 439

        if (shutting_down_.load(std::memory_order_acquire)){
S
starlord 已提交
440
            ENGINE_LOG_DEBUG << "Server will shutdown, skip merge action for table " << table_id;
G
groot 已提交
441 442
            break;
        }
443
    }
X
Xu Peng 已提交
444

G
groot 已提交
445 446
    return Status::OK();
}
447

G
groot 已提交
448
void DBImpl::BackgroundCompaction(std::set<std::string> table_ids) {
449
    ENGINE_LOG_TRACE << " Background compaction thread start";
S
starlord 已提交
450

G
groot 已提交
451
    Status status;
J
jinhai 已提交
452
    for (auto& table_id : table_ids) {
G
groot 已提交
453 454
        status = BackgroundMergeFiles(table_id);
        if (!status.ok()) {
S
starlord 已提交
455
            ENGINE_LOG_ERROR << "Merge files for table " << table_id << " failed: " << status.ToString();
S
starlord 已提交
456
            continue;//let other table get chance to merge
G
groot 已提交
457
        }
S
starlord 已提交
458 459 460 461 462

        if (shutting_down_.load(std::memory_order_acquire)){
            ENGINE_LOG_DEBUG << "Server will shutdown, skip merge action";
            break;
        }
G
groot 已提交
463
    }
X
Xu Peng 已提交
464

G
groot 已提交
465
    meta_ptr_->Archive();
Z
update  
zhiru 已提交
466 467

    int ttl = 1;
Z
update  
zhiru 已提交
468
    if (options_.mode == Options::MODE::CLUSTER) {
Z
update  
zhiru 已提交
469 470 471
        ttl = meta::D_SEC;
    }
    meta_ptr_->CleanUpFilesWithTTL(ttl);
S
starlord 已提交
472

473
    ENGINE_LOG_TRACE << " Background compaction thread exit";
G
groot 已提交
474
}
X
Xu Peng 已提交
475

P
peng.xu 已提交
476
void DBImpl::StartBuildIndexTask(bool force) {
G
groot 已提交
477 478
    static uint64_t index_clock_tick = 0;
    index_clock_tick++;
P
peng.xu 已提交
479
    if(!force && (index_clock_tick%INDEX_ACTION_INTERVAL != 0)) {
G
groot 已提交
480 481 482
        return;
    }

G
groot 已提交
483 484 485 486 487 488 489 490 491 492 493 494 495
    //build index has been finished?
    if(!index_thread_results_.empty()) {
        std::chrono::milliseconds span(10);
        if (index_thread_results_.back().wait_for(span) == std::future_status::ready) {
            index_thread_results_.pop_back();
        }
    }

    //add new build index task
    if(index_thread_results_.empty()) {
        index_thread_results_.push_back(
                index_thread_pool_.enqueue(&DBImpl::BackgroundBuildIndex, this));
    }
X
Xu Peng 已提交
496 497
}

P
peng.xu 已提交
498
Status DBImpl::BuildIndex(const std::string& table_id) {
P
peng.xu 已提交
499 500 501 502 503 504 505
    bool has = false;
    meta_ptr_->HasNonIndexFiles(table_id, has);
    int times = 1;

    while (has) {
        ENGINE_LOG_DEBUG << "Non index files detected! Will build index " << times;
        meta_ptr_->UpdateTableFilesToIndex(table_id);
506
        /* StartBuildIndexTask(true); */
P
peng.xu 已提交
507 508 509 510 511 512
        std::this_thread::sleep_for(std::chrono::milliseconds(std::min(10*1000, times*100)));
        meta_ptr_->HasNonIndexFiles(table_id, has);
        times++;
    }
    return Status::OK();
    /* return BuildIndexByTable(table_id); */
P
peng.xu 已提交
513 514
}

G
groot 已提交
515
Status DBImpl::BuildIndex(const meta::TableFileSchema& file) {
G
groot 已提交
516
    ExecutionEnginePtr to_index = EngineFactory::Build(file.dimension_, file.location_, (EngineType)file.engine_type_);
G
groot 已提交
517
    if(to_index == nullptr) {
S
starlord 已提交
518
        ENGINE_LOG_ERROR << "Invalid engine type";
G
groot 已提交
519 520
        return Status::Error("Invalid engine type");
    }
521

G
groot 已提交
522
    try {
G
groot 已提交
523
        //step 1: load index
S
starlord 已提交
524
        to_index->Load(options_.insert_cache_immediately_);
G
groot 已提交
525 526 527 528 529

        //step 2: create table file
        meta::TableFileSchema table_file;
        table_file.table_id_ = file.table_id_;
        table_file.date_ = file.date_;
530
        table_file.file_type_ = meta::TableFileSchema::NEW_INDEX; //for multi-db-path, distribute index file averagely to each path
G
groot 已提交
531 532
        Status status = meta_ptr_->CreateTableFile(table_file);
        if (!status.ok()) {
S
starlord 已提交
533
            ENGINE_LOG_ERROR << "Failed to create table: " << status.ToString();
G
groot 已提交
534 535 536 537
            return status;
        }

        //step 3: build index
538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558
        std::shared_ptr<ExecutionEngine> index;

        try {
            auto start_time = METRICS_NOW_TIME;
            index = to_index->BuildIndex(table_file.location_);
            auto end_time = METRICS_NOW_TIME;
            auto total_time = METRICS_MICROSECONDS(start_time, end_time);
            server::Metrics::GetInstance().BuildIndexDurationSecondsHistogramObserve(total_time);
        } catch (std::exception& ex) {
            //typical error: out of gpu memory
            std::string msg = "BuildIndex encounter exception" + std::string(ex.what());
            ENGINE_LOG_ERROR << msg;

            table_file.file_type_ = meta::TableFileSchema::TO_DELETE;
            status = meta_ptr_->UpdateTableFile(table_file);
            ENGINE_LOG_DEBUG << "Failed to update file to index, mark file: " << table_file.file_id_ << " to to_delete";

            std::cout << "ERROR: failed to build index, index file is too large or gpu memory is not enough" << std::endl;

            return Status::Error(msg);
        }
559

G
groot 已提交
560 561 562 563 564 565 566 567 568
        //step 4: if table has been deleted, dont save index file
        bool has_table = false;
        meta_ptr_->HasTable(file.table_id_, has_table);
        if(!has_table) {
            meta_ptr_->DeleteTableFiles(file.table_id_);
            return Status::OK();
        }

        //step 5: save index file
569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584
        try {
            index->Serialize();
        } catch (std::exception& ex) {
            //typical error: out of disk space or permition denied
            std::string msg = "Serialize index encounter exception" + std::string(ex.what());
            ENGINE_LOG_ERROR << msg;

            table_file.file_type_ = meta::TableFileSchema::TO_DELETE;
            status = meta_ptr_->UpdateTableFile(table_file);
            ENGINE_LOG_DEBUG << "Failed to update file to index, mark file: " << table_file.file_id_ << " to to_delete";

            std::cout << "ERROR: failed to persist index file: " << table_file.location_
                << ", possible out of disk space" << std::endl;

            return Status::Error(msg);
        }
G
groot 已提交
585 586

        //step 6: update meta
G
groot 已提交
587
        table_file.file_type_ = meta::TableFileSchema::INDEX;
S
starlord 已提交
588
        table_file.size_ = index->Size();
X
Xu Peng 已提交
589

G
groot 已提交
590 591
        auto to_remove = file;
        to_remove.file_type_ = meta::TableFileSchema::TO_DELETE;
X
Xu Peng 已提交
592

593 594 595 596 597 598
        meta::TableFilesSchema update_files = {table_file, to_remove};
        status = meta_ptr_->UpdateTableFiles(update_files);
        if(status.ok()) {
            ENGINE_LOG_DEBUG << "New index file " << table_file.file_id_ << " of size "
                             << index->PhysicalSize() << " bytes"
                             << " from file " << to_remove.file_id_;
X
Xu Peng 已提交
599

600 601 602 603 604 605 606 607 608 609 610 611
            if(options_.insert_cache_immediately_) {
                index->Cache();
            }
        } else {
            //failed to update meta, mark the new file as to_delete, don't delete old file
            to_remove.file_type_ = meta::TableFileSchema::TO_INDEX;
            status = meta_ptr_->UpdateTableFile(to_remove);
            ENGINE_LOG_DEBUG << "Failed to update file to index, mark file: " << to_remove.file_id_ << " to to_index";

            table_file.file_type_ = meta::TableFileSchema::TO_DELETE;
            status = meta_ptr_->UpdateTableFile(table_file);
            ENGINE_LOG_DEBUG << "Failed to update file to index, mark file: " << table_file.file_id_ << " to to_delete";
S
starlord 已提交
612
        }
G
groot 已提交
613 614

    } catch (std::exception& ex) {
S
starlord 已提交
615 616 617
        std::string msg = "Build index encounter exception" + std::string(ex.what());
        ENGINE_LOG_ERROR << msg;
        return Status::Error(msg);
G
groot 已提交
618
    }
X
Xu Peng 已提交
619

X
Xu Peng 已提交
620 621 622
    return Status::OK();
}

P
peng.xu 已提交
623
Status DBImpl::BuildIndexByTable(const std::string& table_id) {
P
peng.xu 已提交
624
    std::unique_lock<std::mutex> lock(build_index_mutex_);
P
peng.xu 已提交
625 626 627 628 629 630 631 632 633 634 635 636
    meta::TableFilesSchema to_index_files;
    meta_ptr_->FilesToIndex(to_index_files);

    Status status;

    for (auto& file : to_index_files) {
        status = BuildIndex(file);
        if (!status.ok()) {
            ENGINE_LOG_ERROR << "Building index for " << file.id_ << " failed: " << status.ToString();
            return status;
        }
        ENGINE_LOG_DEBUG << "Sync building index for " << file.id_ << " passed";
S
starlord 已提交
637 638 639 640 641

        if (shutting_down_.load(std::memory_order_acquire)){
            ENGINE_LOG_DEBUG << "Server will shutdown, skip build index action for table " << table_id;
            break;
        }
P
peng.xu 已提交
642 643 644 645 646
    }

    return status;
}

G
groot 已提交
647
void DBImpl::BackgroundBuildIndex() {
648
    ENGINE_LOG_TRACE << " Background build index thread start";
S
starlord 已提交
649

P
peng.xu 已提交
650
    std::unique_lock<std::mutex> lock(build_index_mutex_);
651
    meta::TableFilesSchema to_index_files;
G
groot 已提交
652
    meta_ptr_->FilesToIndex(to_index_files);
X
Xu Peng 已提交
653 654
    Status status;
    for (auto& file : to_index_files) {
X
Xu Peng 已提交
655
        status = BuildIndex(file);
X
Xu Peng 已提交
656
        if (!status.ok()) {
S
starlord 已提交
657
            ENGINE_LOG_ERROR << "Building index for " << file.id_ << " failed: " << status.ToString();
X
Xu Peng 已提交
658
            return;
X
Xu Peng 已提交
659
        }
660

G
groot 已提交
661
        if (shutting_down_.load(std::memory_order_acquire)){
S
starlord 已提交
662
            ENGINE_LOG_DEBUG << "Server will shutdown, skip build index action";
G
groot 已提交
663
            break;
X
Xu Peng 已提交
664
        }
665
    }
S
starlord 已提交
666

667
    ENGINE_LOG_TRACE << " Background build index thread exit";
X
Xu Peng 已提交
668 669
}

G
groot 已提交
670
Status DBImpl::DropAll() {
G
groot 已提交
671
    return meta_ptr_->DropAll();
X
Xu Peng 已提交
672 673
}

G
groot 已提交
674
Status DBImpl::Size(uint64_t& result) {
G
groot 已提交
675
    return  meta_ptr_->Size(result);
X
Xu Peng 已提交
676 677
}

G
groot 已提交
678
DBImpl::~DBImpl() {
G
groot 已提交
679
    shutting_down_.store(true, std::memory_order_release);
X
Xu Peng 已提交
680
    bg_timer_thread_.join();
G
groot 已提交
681
    std::set<std::string> ids;
G
groot 已提交
682
    mem_mgr_->Serialize(ids);
X
Xu Peng 已提交
683 684
}

X
Xu Peng 已提交
685
} // namespace engine
J
jinhai 已提交
686
} // namespace milvus
X
Xu Peng 已提交
687
} // namespace zilliz