Merge pull request #607 from yhmo/ongoing

- \#552 - Server down during building index_type: IVF_PQ using GPU-edition

- \#561 - Milvus server should report exception/error message or terminate on mysql metadata backend error

- \#579 - Build index hang in GPU version when gpu_resources disabled

- \#596 - Frequently insert operation cost too much disk space

- \#599 - Build index log is incorrect

- \#602 - Optimizer specify wrong gpu_id

- \#606 - No log generated during building index with CPU

    {

        std::lock_guard<std::mutex> lock(mutex_);

        cache_count = lru_.size();

#if 0

        for (auto it = lru_.begin(); it != lru_.end(); ++it) {

            SERVER_LOG_DEBUG << it->first;

        }

#endif

    }

    SERVER_LOG_DEBUG << "[Cache item count]: " << cache_count;

constexpr uint64_t COMPACT_ACTION_INTERVAL = 1;

constexpr uint64_t INDEX_ACTION_INTERVAL = 1;

constexpr uint64_t INDEX_FAILED_RETRY_TIME = 1;

static const Status SHUTDOWN_ERROR = Status(DB_ERROR, "Milsvus server is shutdown!");

void

    WaitMergeFileFinish();

    // step 4: wait and build index

    status = CleanFailedIndexFileOfTable(table_id);

    status = index_failed_checker_.CleanFailedIndexFileOfTable(table_id);

    status = BuildTableIndexRecursively(table_id, index);

    return status;

    TimeRecorder rc("");

    // step 1: get files to search

    // step 1: construct search job

    auto status = ongoing_files_checker_.MarkOngoingFiles(files);

    ENGINE_LOG_DEBUG << "Engine query begin, index file count: " << files.size();

    scheduler::SearchJobPtr job = std::make_shared<scheduler::SearchJob>(k, nq, nprobe, vectors);

    for (auto& file : files) {

        job->AddIndexFile(file_ptr);

    }

    // step 2: put search task to scheduler

    // step 2: put search job to scheduler and wait result

    scheduler::JobMgrInst::GetInstance()->Put(job);

    job->WaitResult();

    status = ongoing_files_checker_.UnmarkOngoingFiles(files);

    if (!job->GetStatus().ok()) {

        return job->GetStatus();

    }

        auto file_schema = file;

        file_schema.file_type_ = meta::TableFileSchema::TO_DELETE;

        updated.push_back(file_schema);

        ENGINE_LOG_DEBUG << "Merging file " << file_schema.file_id_;

        index_size = index->Size();

        if (index_size >= file_schema.index_file_size_) {

    // step 3: serialize to disk

    try {

        index->Serialize();

        status = index->Serialize();

        if (!status.ok()) {

            ENGINE_LOG_ERROR << status.message();

        }

    } catch (std::exception& ex) {

        // typical error: out of disk space or permition denied

        std::string msg = "Serialize merged index encounter exception: " + std::string(ex.what());

        ENGINE_LOG_ERROR << msg;

        status = Status(DB_ERROR, msg);

    }

    if (!status.ok()) {

        // if failed to serialize merge file to disk

        // typical error: out of disk space, out of memory or permition denied

        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 merged index file: " << table_file.location_

                  << ", possible out of disk space" << std::endl;

        ENGINE_LOG_ERROR << "Failed to persist merged file: " << table_file.location_

                         << ", possible out of disk space or memory";

        return Status(DB_ERROR, msg);

        return status;

    }

    // step 4: update table files state

    }

    for (auto& kv : raw_files) {

        auto files = kv.second;

        meta::TableFilesSchema& files = kv.second;

        if (files.size() < options_.merge_trigger_number_) {

            ENGINE_LOG_TRACE << "Files number not greater equal than merge trigger number, skip merge action";

            continue;

        }

        status = ongoing_files_checker_.MarkOngoingFiles(files);

        MergeFiles(table_id, kv.first, kv.second);

        status = ongoing_files_checker_.UnmarkOngoingFiles(files);

        if (shutting_down_.load(std::memory_order_acquire)) {

            ENGINE_LOG_DEBUG << "Server will shutdown, skip merge action for table: " << table_id;

    meta_ptr_->Archive();

    {

        uint64_t ttl = 10 * meta::SECOND;  // default: file data will be erase from cache after few seconds

        meta_ptr_->CleanUpCacheWithTTL(ttl);

    }

    {

        uint64_t ttl = 5 * meta::M_SEC;  // default: file will be deleted after few minutes

        uint64_t ttl = 10 * meta::SECOND;  // default: file will be hard-deleted few seconds after soft-deleted

        if (options_.mode_ == DBOptions::MODE::CLUSTER_WRITABLE) {

            ttl = meta::D_SEC;

            ttl = meta::H_SEC;

        }

        meta_ptr_->CleanUpFilesWithTTL(ttl);

        meta_ptr_->CleanUpFilesWithTTL(ttl, &ongoing_files_checker_);

    }

    // ENGINE_LOG_TRACE << " Background compaction thread exit";

void

DBImpl::BackgroundBuildIndex() {

    // ENGINE_LOG_TRACE << "Background build index thread start";

    std::unique_lock<std::mutex> lock(build_index_mutex_);

    meta::TableFilesSchema to_index_files;

    meta_ptr_->FilesToIndex(to_index_files);

    Status status = IgnoreFailedIndexFiles(to_index_files);

    Status status = index_failed_checker_.IgnoreFailedIndexFiles(to_index_files);

    if (!to_index_files.empty()) {

        ENGINE_LOG_DEBUG << "Background build index thread begin";

        status = ongoing_files_checker_.MarkOngoingFiles(to_index_files);

        // step 2: put build index task to scheduler

        std::vector<std::pair<scheduler::BuildIndexJobPtr, scheduler::TableFileSchemaPtr>> job2file_map;

        for (auto& file : to_index_files) {

            job2file_map.push_back(std::make_pair(job, file_ptr));

        }

        // step 3: wait build index finished and mark failed files

        for (auto iter = job2file_map.begin(); iter != job2file_map.end(); ++iter) {

            scheduler::BuildIndexJobPtr job = iter->first;

            meta::TableFileSchema& file_schema = *(iter->second.get());

                Status status = job->GetStatus();

                ENGINE_LOG_ERROR << "Building index job " << job->id() << " failed: " << status.ToString();

                MarkFailedIndexFile(file_schema);

                index_failed_checker_.MarkFailedIndexFile(file_schema);

            } else {

                MarkSucceedIndexFile(file_schema);

                ENGINE_LOG_DEBUG << "Building index job " << job->id() << " succeed.";

                index_failed_checker_.MarkSucceedIndexFile(file_schema);

            }

            status = ongoing_files_checker_.UnmarkOngoingFile(file_schema);

        }

        ENGINE_LOG_DEBUG << "Background build index thread finished";

    }

    // ENGINE_LOG_TRACE << "Background build index thread exit";

}

Status

Status

DBImpl::GetFilesToSearch(const std::string& table_id, const std::vector<size_t>& file_ids, const meta::DatesT& dates,

                         meta::TableFilesSchema& files) {

    ENGINE_LOG_DEBUG << "Collect files from table: " << table_id;

    meta::DatePartionedTableFilesSchema date_files;

    auto status = meta_ptr_->FilesToSearch(table_id, file_ids, dates, date_files);

    if (!status.ok()) {

    if (dates.empty()) {

        status = mem_mgr_->EraseMemVector(table_id);  // not allow insert

        status = meta_ptr_->DropTable(table_id);      // soft delete table

        CleanFailedIndexFileOfTable(table_id);

        index_failed_checker_.CleanFailedIndexFileOfTable(table_id);

        // scheduler will determine when to delete table files

        auto nres = scheduler::ResMgrInst::GetInstance()->GetNumOfComputeResource();

        GetFilesToBuildIndex(table_id, file_types, table_files);

        times++;

        IgnoreFailedIndexFiles(table_files);

        index_failed_checker_.IgnoreFailedIndexFiles(table_files);

    }

    // build index for partition

    // failed to build index for some files, return error

    std::vector<std::string> failed_files;

    GetFailedIndexFileOfTable(table_id, failed_files);

    index_failed_checker_.GetFailedIndexFileOfTable(table_id, failed_files);

    if (!failed_files.empty()) {

        std::string msg = "Failed to build index for " + std::to_string(failed_files.size()) +

                          ((failed_files.size() == 1) ? " file" : " files");

Status

DBImpl::DropTableIndexRecursively(const std::string& table_id) {

    ENGINE_LOG_DEBUG << "Drop index for table: " << table_id;

    CleanFailedIndexFileOfTable(table_id);

    index_failed_checker_.CleanFailedIndexFileOfTable(table_id);

    auto status = meta_ptr_->DropTableIndex(table_id);

    if (!status.ok()) {

        return status;

    return Status::OK();

}

Status

DBImpl::CleanFailedIndexFileOfTable(const std::string& table_id) {

    std::lock_guard<std::mutex> lck(index_failed_mutex_);

    index_failed_files_.erase(table_id);  // rebuild failed index files for this table

    return Status::OK();

}

Status

DBImpl::GetFailedIndexFileOfTable(const std::string& table_id, std::vector<std::string>& failed_files) {

    failed_files.clear();

    std::lock_guard<std::mutex> lck(index_failed_mutex_);

    auto iter = index_failed_files_.find(table_id);

    if (iter != index_failed_files_.end()) {

        FileID2FailedTimes& failed_map = iter->second;

        for (auto it_file = failed_map.begin(); it_file != failed_map.end(); ++it_file) {

            failed_files.push_back(it_file->first);

        }

    }

    return Status::OK();

}

Status

DBImpl::MarkFailedIndexFile(const meta::TableFileSchema& file) {

    std::lock_guard<std::mutex> lck(index_failed_mutex_);

    auto iter = index_failed_files_.find(file.table_id_);

    if (iter == index_failed_files_.end()) {

        FileID2FailedTimes failed_files;

        failed_files.insert(std::make_pair(file.file_id_, 1));

        index_failed_files_.insert(std::make_pair(file.table_id_, failed_files));

    } else {

        auto it_failed_files = iter->second.find(file.file_id_);

        if (it_failed_files != iter->second.end()) {

            it_failed_files->second++;

        } else {

            iter->second.insert(std::make_pair(file.file_id_, 1));

        }

    }

    return Status::OK();

}

Status

DBImpl::MarkSucceedIndexFile(const meta::TableFileSchema& file) {

    std::lock_guard<std::mutex> lck(index_failed_mutex_);

    auto iter = index_failed_files_.find(file.table_id_);

    if (iter != index_failed_files_.end()) {

        iter->second.erase(file.file_id_);

    }

    return Status::OK();

}

Status

DBImpl::IgnoreFailedIndexFiles(meta::TableFilesSchema& table_files) {

    std::lock_guard<std::mutex> lck(index_failed_mutex_);

    // there could be some failed files belong to different table.

    // some files may has failed for several times, no need to build index for these files.

    // thus we can avoid dead circle for build index operation

    for (auto it_file = table_files.begin(); it_file != table_files.end();) {

        auto it_failed_files = index_failed_files_.find((*it_file).table_id_);

        if (it_failed_files != index_failed_files_.end()) {

            auto it_failed_file = it_failed_files->second.find((*it_file).file_id_);

            if (it_failed_file != it_failed_files->second.end()) {

                if (it_failed_file->second >= INDEX_FAILED_RETRY_TIME) {

                    it_file = table_files.erase(it_file);

                    continue;

                }

            }

        }

        ++it_file;

    }

    return Status::OK();

}

}  // namespace engine

}  // namespace milvus

#pragma once

#include "DB.h"

#include "Types.h"

#include "src/db/insert/MemManager.h"

#include "db/IndexFailedChecker.h"

#include "db/OngoingFileChecker.h"

#include "db/Types.h"

#include "db/insert/MemManager.h"

#include "utils/ThreadPool.h"

#include <atomic>

    Status

    GetTableRowCountRecursively(const std::string& table_id, uint64_t& row_count);

    Status

    CleanFailedIndexFileOfTable(const std::string& table_id);

    Status

    GetFailedIndexFileOfTable(const std::string& table_id, std::vector<std::string>& failed_files);

    Status

    MarkFailedIndexFile(const meta::TableFileSchema& file);

    Status

    MarkSucceedIndexFile(const meta::TableFileSchema& file);

    Status

    IgnoreFailedIndexFiles(meta::TableFilesSchema& table_files);

 private:

    const DBOptions options_;

    std::list<std::future<void>> index_thread_results_;

    std::mutex build_index_mutex_;

    std::mutex index_failed_mutex_;

    using FileID2FailedTimes = std::map<std::string, uint64_t>;

    using Table2FailedFiles = std::map<std::string, FileID2FailedTimes>;

    Table2FailedFiles index_failed_files_;  // file id mapping to failed times

    IndexFailedChecker index_failed_checker_;

    OngoingFileChecker ongoing_files_checker_;

};  // DBImpl

}  // namespace engine

// 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 "db/IndexFailedChecker.h"

#include <utility>

namespace milvus {

namespace engine {

constexpr uint64_t INDEX_FAILED_RETRY_TIME = 1;

Status

IndexFailedChecker::CleanFailedIndexFileOfTable(const std::string& table_id) {

    std::lock_guard<std::mutex> lck(mutex_);

    index_failed_files_.erase(table_id);  // rebuild failed index files for this table

    return Status::OK();

}

Status

IndexFailedChecker::GetFailedIndexFileOfTable(const std::string& table_id, std::vector<std::string>& failed_files) {

    failed_files.clear();

    std::lock_guard<std::mutex> lck(mutex_);

    auto iter = index_failed_files_.find(table_id);

    if (iter != index_failed_files_.end()) {

        File2RefCount& failed_map = iter->second;

        for (auto it_file = failed_map.begin(); it_file != failed_map.end(); ++it_file) {

            failed_files.push_back(it_file->first);

        }

    }

    return Status::OK();

}

Status

IndexFailedChecker::MarkFailedIndexFile(const meta::TableFileSchema& file) {

    std::lock_guard<std::mutex> lck(mutex_);

    auto iter = index_failed_files_.find(file.table_id_);

    if (iter == index_failed_files_.end()) {

        File2RefCount failed_files;

        failed_files.insert(std::make_pair(file.file_id_, 1));

        index_failed_files_.insert(std::make_pair(file.table_id_, failed_files));

    } else {

        auto it_failed_files = iter->second.find(file.file_id_);

        if (it_failed_files != iter->second.end()) {

            it_failed_files->second++;

        } else {

            iter->second.insert(std::make_pair(file.file_id_, 1));

        }

    }

    return Status::OK();

}

Status

IndexFailedChecker::MarkSucceedIndexFile(const meta::TableFileSchema& file) {

    std::lock_guard<std::mutex> lck(mutex_);

    auto iter = index_failed_files_.find(file.table_id_);

    if (iter != index_failed_files_.end()) {

        iter->second.erase(file.file_id_);

        if (iter->second.empty()) {

            index_failed_files_.erase(file.table_id_);

        }

    }

    return Status::OK();

}

Status

IndexFailedChecker::IgnoreFailedIndexFiles(meta::TableFilesSchema& table_files) {

    std::lock_guard<std::mutex> lck(mutex_);

    // there could be some failed files belong to different table.

    // some files may has failed for several times, no need to build index for these files.

    // thus we can avoid dead circle for build index operation

    for (auto it_file = table_files.begin(); it_file != table_files.end();) {

        auto it_failed_files = index_failed_files_.find((*it_file).table_id_);

        if (it_failed_files != index_failed_files_.end()) {

            auto it_failed_file = it_failed_files->second.find((*it_file).file_id_);

            if (it_failed_file != it_failed_files->second.end()) {

                if (it_failed_file->second >= INDEX_FAILED_RETRY_TIME) {

                    it_file = table_files.erase(it_file);

                    continue;

                }

            }

        }

        ++it_file;

    }

    return Status::OK();

}

}  // namespace engine

}  // namespace milvus