Snapshot.h 8.0 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// Licensed 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.

#pragma once

#include <assert.h>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <iostream>
#include <limits>
#include <map>
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <string>
#include <thread>
X
XuPeng-SH 已提交
26
#include <tuple>
27 28
#include <utility>
#include <vector>
29
#include "db/snapshot/Store.h"
X
XuPeng-SH 已提交
30
#include "db/snapshot/Utils.h"
31
#include "db/snapshot/WrappedTypes.h"
32
#include "utils/Status.h"
33 34 35 36 37

namespace milvus {
namespace engine {
namespace snapshot {

X
XuPeng-SH 已提交
38 39 40 41 42
using ScopedResourcesT =
    std::tuple<CollectionCommit::ScopedMapT, Collection::ScopedMapT, SchemaCommit::ScopedMapT, FieldCommit::ScopedMapT,
               Field::ScopedMapT, FieldElement::ScopedMapT, PartitionCommit::ScopedMapT, Partition::ScopedMapT,
               SegmentCommit::ScopedMapT, Segment::ScopedMapT, SegmentFile::ScopedMapT>;

43 44 45 46 47 48
class Snapshot : public ReferenceProxy {
 public:
    using Ptr = std::shared_ptr<Snapshot>;
    explicit Snapshot(ID_TYPE id);

    ID_TYPE
X
XuPeng-SH 已提交
49 50
    GetID() {
        return GetCollectionCommit()->GetID();
51
    }
X
XuPeng-SH 已提交
52

53 54
    ID_TYPE
    GetCollectionId() const {
X
XuPeng-SH 已提交
55 56
        auto it = GetResources<Collection>().begin();
        return it->first;
57
    }
X
XuPeng-SH 已提交
58

59 60 61 62 63
    CollectionPtr
    GetCollection() {
        return GetResources<Collection>().begin()->second.Get();
    }

64 65 66 67 68 69
    SchemaCommitPtr
    GetSchemaCommit() {
        auto id = GetLatestSchemaCommitId();
        return GetResource<SchemaCommit>(id);
    }

70 71
    const std::string&
    GetName() const {
X
XuPeng-SH 已提交
72
        return GetResources<Collection>().begin()->second->GetName();
73
    }
X
XuPeng-SH 已提交
74

75 76 77 78 79
    size_t
    NumberOfPartitions() const {
        return GetResources<Partition>().size();
    }

X
XuPeng-SH 已提交
80 81 82 83 84
    const LSN_TYPE&
    GetMaxLsn() const {
        return max_lsn_;
    }

85 86 87 88 89 90 91 92 93 94
    Status
    GetPartitionId(const std::string& name, ID_TYPE& id) const {
        auto it = partition_names_map_.find(name);
        if (it == partition_names_map_.end()) {
            return Status(SS_NOT_FOUND_ERROR, "Specified partition name not found");
        }
        id = it->second;
        return Status::OK();
    }

95 96
    CollectionCommitPtr
    GetCollectionCommit() {
X
XuPeng-SH 已提交
97
        return GetResources<CollectionCommit>().begin()->second.Get();
98 99 100 101 102 103 104 105 106 107 108 109 110
    }

    ID_TYPE
    GetLatestSchemaCommitId() const {
        return latest_schema_commit_id_;
    }

    // PXU TODO: add const. Need to change Scopedxxxx::Get
    SegmentCommitPtr
    GetSegmentCommit(ID_TYPE segment_id) {
        auto it = seg_segc_map_.find(segment_id);
        if (it == seg_segc_map_.end())
            return nullptr;
X
XuPeng-SH 已提交
111
        return GetResource<SegmentCommit>(it->second);
112 113 114 115 116 117 118
    }

    PartitionCommitPtr
    GetPartitionCommitByPartitionId(ID_TYPE partition_id) {
        auto it = p_pc_map_.find(partition_id);
        if (it == p_pc_map_.end())
            return nullptr;
X
XuPeng-SH 已提交
119
        return GetResource<PartitionCommit>(it->second);
120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188
    }

    std::vector<std::string>
    GetFieldNames() const {
        std::vector<std::string> names;
        for (auto& kv : field_names_map_) {
            names.emplace_back(kv.first);
        }
        return std::move(names);
    }

    bool
    HasField(const std::string& name) const {
        auto it = field_names_map_.find(name);
        return it != field_names_map_.end();
    }

    bool
    HasFieldElement(const std::string& field_name, const std::string& field_element_name) const {
        auto id = GetFieldElementId(field_name, field_element_name);
        return id > 0;
    }

    ID_TYPE
    GetSegmentFileId(const std::string& field_name, const std::string& field_element_name, ID_TYPE segment_id) const {
        auto field_element_id = GetFieldElementId(field_name, field_element_name);
        auto it = element_segfiles_map_.find(field_element_id);
        if (it == element_segfiles_map_.end()) {
            return 0;
        }
        auto its = it->second.find(segment_id);
        if (its == it->second.end()) {
            return 0;
        }
        return its->second;
    }

    bool
    HasSegmentFile(const std::string& field_name, const std::string& field_element_name, ID_TYPE segment_id) const {
        auto id = GetSegmentFileId(field_name, field_element_name, segment_id);
        return id > 0;
    }

    ID_TYPE
    GetFieldElementId(const std::string& field_name, const std::string& field_element_name) const {
        auto itf = field_element_names_map_.find(field_name);
        if (itf == field_element_names_map_.end())
            return false;
        auto itfe = itf->second.find(field_element_name);
        if (itfe == itf->second.end()) {
            return 0;
        }

        return itfe->second;
    }

    NUM_TYPE
    GetMaxSegmentNumByPartition(ID_TYPE partition_id) {
        auto it = p_max_seg_num_.find(partition_id);
        if (it == p_max_seg_num_.end())
            return 0;
        return it->second;
    }

    void
    RefAll();
    void
    UnRefAll();

X
XuPeng-SH 已提交
189
    template <typename ResourceT>
190
    void
X
XuPeng-SH 已提交
191 192
    DumpResource(const std::string& tag = "") {
        auto& resources = GetResources<ResourceT>();
193 194
        std::cout << typeid(*this).name() << " Dump " << GetID() << " " << ResourceT::Name << " Start [" << tag
                  << "]:" << resources.size() << std::endl;
X
XuPeng-SH 已提交
195 196 197
        for (auto& kv : resources) {
            std::cout << "\t" << kv.second->ToString() << std::endl;
        }
198 199
        std::cout << typeid(*this).name() << " Dump " << GetID() << " " << ResourceT::Name << "  End [" << tag
                  << "]:" << resources.size() << std::endl;
X
XuPeng-SH 已提交
200 201 202
    }

    template <typename T>
203
    void
X
XuPeng-SH 已提交
204 205 206 207 208 209 210
    DoUnRef(T& resource_map) {
        for (auto& kv : resource_map) {
            kv.second->UnRef();
        }
    }

    template <typename T>
211
    void
X
XuPeng-SH 已提交
212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247
    DoRef(T& resource_map) {
        for (auto& kv : resource_map) {
            kv.second->Ref();
        }
    }

    template <typename ResourceT>
    typename ResourceT::ScopedMapT&
    GetResources() {
        return std::get<Index<typename ResourceT::ScopedMapT, ScopedResourcesT>::value>(resources_);
    }

    template <typename ResourceT>
    const typename ResourceT::ScopedMapT&
    GetResources() const {
        return std::get<Index<typename ResourceT::ScopedMapT, ScopedResourcesT>::value>(resources_);
    }

    template <typename ResourceT>
    typename ResourceT::Ptr
    GetResource(ID_TYPE id) {
        auto& resources = GetResources<ResourceT>();
        auto it = resources.find(id);
        if (it == resources.end()) {
            return nullptr;
        }

        return it->second.Get();
    }

    template <typename ResourceT>
    void
    AddResource(ScopedResource<ResourceT>& resource) {
        auto& resources = GetResources<ResourceT>();
        resources[resource->GetID()] = resource;
    }
248 249

 private:
250 251 252 253
    Snapshot(const Snapshot&) = delete;
    Snapshot&
    operator=(const Snapshot&) = delete;

254
    // PXU TODO: Re-org below data structures to reduce memory usage
X
XuPeng-SH 已提交
255
    ScopedResourcesT resources_;
256 257
    ID_TYPE current_schema_id_;
    std::map<std::string, ID_TYPE> field_names_map_;
258
    std::map<std::string, ID_TYPE> partition_names_map_;
259 260 261 262 263 264
    std::map<std::string, std::map<std::string, ID_TYPE>> field_element_names_map_;
    std::map<ID_TYPE, std::map<ID_TYPE, ID_TYPE>> element_segfiles_map_;
    std::map<ID_TYPE, ID_TYPE> seg_segc_map_;
    std::map<ID_TYPE, ID_TYPE> p_pc_map_;
    ID_TYPE latest_schema_commit_id_ = 0;
    std::map<ID_TYPE, NUM_TYPE> p_max_seg_num_;
X
XuPeng-SH 已提交
265
    LSN_TYPE max_lsn_;
266 267 268 269 270 271 272 273
};

using ScopedSnapshotT = ScopedResource<Snapshot>;
using GCHandler = std::function<void(Snapshot::Ptr)>;

}  // namespace snapshot
}  // namespace engine
}  // namespace milvus