Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions include/knowhere/index/index_node.h
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,11 @@
#ifndef INDEX_NODE_H
#define INDEX_NODE_H

#define KNOWHERE_SEARCH_CONFIG_CACHE_VERSION 1

#include <atomic>
#include <functional>
#include <memory>
#include <mutex>
#include <queue>
#include <utility>
Expand Down Expand Up @@ -618,6 +622,22 @@ class IndexNode : public Object {
SearchEmbList(const DataSetPtr dataset, std::unique_ptr<Config> cfg, const BitsetView& bitset,
milvus::OpContext* op_context = nullptr) const;

public:
virtual bool
SupportsSearchConfigCache() const {
return false;
}

virtual expected<DataSetPtr>
SearchWithPreparedConfig(const DataSetPtr, std::shared_ptr<const Config>, const BitsetView&,
milvus::OpContext* = nullptr) const {
return expected<DataSetPtr>::Err(Status::not_implemented, "prepared search config is not supported");
}

expected<std::shared_ptr<const Config>>
GetOrCreateSearchConfig(const Json& json) const;

protected:
static EmbListMetaHeader
ParseEmbListMetaHeader(const uint8_t* data, int64_t size);

Expand All @@ -637,6 +657,17 @@ class IndexNode : public Object {
std::shared_ptr<ThreadPool> pool, milvus::OpContext* op_context = nullptr) const;

Version version_;

private:
struct SearchConfigCacheEntry {
Json json;
std::shared_ptr<const Config> config;
};

mutable std::shared_ptr<const SearchConfigCacheEntry> search_config_cache_;
mutable std::mutex search_config_cache_mutex_;

protected:
std::shared_ptr<EmbListOffset> emb_list_offset_; // emb_list group offset structure (shared with strategy)
std::string el_metric_type_;
EmbListStrategyPtr emb_list_strategy_; // emb_list encoding strategy (tokenann/muvera)
Expand Down
29 changes: 22 additions & 7 deletions src/index/index.cc
Original file line number Diff line number Diff line change
Expand Up @@ -133,12 +133,23 @@ inline expected<DataSetPtr>
Index<T>::Search(const DataSetPtr dataset, const Json& json, const BitsetView& bitset_,
milvus::OpContext* op_context) const noexcept {
return GuardedCall([&]() -> expected<DataSetPtr> {
auto cfg = this->node->CreateConfig();
std::unique_ptr<BaseConfig> owned_cfg;
std::shared_ptr<const Config> prepared_cfg;
std::string msg;
const Status load_status = LoadConfig(cfg.get(), json, knowhere::SEARCH, "Search", &msg);
if (load_status != Status::success) {
return expected<DataSetPtr>::Err(load_status, msg);
if (this->node->SupportsSearchConfigCache()) {
auto result = this->node->GetOrCreateSearchConfig(json);
if (!result.has_value()) {
return expected<DataSetPtr>::Err(result.error(), result.what());
}
prepared_cfg = std::move(result.value());
} else {
owned_cfg = this->node->CreateConfig();
const Status load_status = LoadConfig(owned_cfg.get(), json, knowhere::SEARCH, "Search", &msg);
if (load_status != Status::success) {
return expected<DataSetPtr>::Err(load_status, msg);
}
}
const Config* cfg = prepared_cfg != nullptr ? prepared_cfg.get() : owned_cfg.get();
// when index is immutable, bitset size should always equal to data count in index
// when index is mutable, it could happen that data count larger than bitset size, see
// https://github.com/zilliztech/knowhere/issues/70
Expand Down Expand Up @@ -177,14 +188,18 @@ Index<T>::Search(const DataSetPtr dataset, const Json& json, const BitsetView& b
// LCOV_EXCL_STOP

TimeRecorder rc("Search");
auto k = cfg->k.value();
auto res = this->node->SearchEmbListIfNeed(dataset, std::move(cfg), bitset, op_context);
auto k = b_cfg.k.value();
auto res = prepared_cfg != nullptr
? this->node->SearchWithPreparedConfig(dataset, std::move(prepared_cfg), bitset, op_context)
: this->node->SearchEmbListIfNeed(dataset, std::move(owned_cfg), bitset, op_context);
auto time = rc.ElapseFromBegin("done");
time *= 0.001; // convert to ms
this->node->GetSearchLatencyMetric().Observe(time);
knowhere_search_topk.Observe(k);
#else
auto res = this->node->SearchEmbListIfNeed(dataset, std::move(cfg), bitset, op_context);
auto res = prepared_cfg != nullptr
? this->node->SearchWithPreparedConfig(dataset, std::move(prepared_cfg), bitset, op_context)
: this->node->SearchEmbListIfNeed(dataset, std::move(owned_cfg), bitset, op_context);
#endif
return res;
});
Expand Down
34 changes: 34 additions & 0 deletions src/index/index_node.cc
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,40 @@

namespace knowhere {

expected<std::shared_ptr<const Config>>
IndexNode::GetOrCreateSearchConfig(const Json& json) const {
auto cached = std::atomic_load_explicit(&search_config_cache_, std::memory_order_acquire);
if (cached != nullptr && cached->json == json) {
return cached->config;
}

std::scoped_lock lock(search_config_cache_mutex_);
cached = std::atomic_load_explicit(&search_config_cache_, std::memory_order_relaxed);
if (cached != nullptr && cached->json == json) {
return cached->config;
}

auto cfg = CreateConfig();
Json normalized_json(json);
std::string msg;
auto status = Config::FormatAndCheck(*cfg, normalized_json, &msg);
LOG_KNOWHERE_DEBUG_ << "Search config dump: " << normalized_json.dump();
if (status != Status::success) {
return expected<std::shared_ptr<const Config>>::Err(status, msg);
}
cfg->CaptureRawJson(normalized_json);
status = Config::Load(*cfg, normalized_json, knowhere::SEARCH, &msg);
if (status != Status::success) {
return expected<std::shared_ptr<const Config>>::Err(status, msg);
}

std::shared_ptr<const Config> prepared_config(std::move(cfg));
auto entry =
std::make_shared<const SearchConfigCacheEntry>(SearchConfigCacheEntry{.json = json, .config = prepared_config});
std::atomic_store_explicit(&search_config_cache_, std::move(entry), std::memory_order_release);
return prepared_config;
}

// NOLINTBEGIN(google-default-arguments)
expected<DataSetPtr>
IndexNode::RangeSearch(const DataSetPtr dataset, std::unique_ptr<Config> cfg, const BitsetView& bitset,
Expand Down
86 changes: 86 additions & 0 deletions tests/ut/test_index_node.cc
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
// 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 <atomic>
#include <future>
#include <unordered_set>

#include "catch2/catch_approx.hpp"
Expand Down Expand Up @@ -136,6 +138,49 @@ class BaseFlatIndexNode : public IndexNode {
}
};

template <typename DataType>
class CachedSearchConfigIndexNode : public BaseFlatIndexNode<DataType> {
public:
CachedSearchConfigIndexNode(const int32_t& version, const Object& object)
: BaseFlatIndexNode<DataType>(version, object) {
}

bool
SupportsSearchConfigCache() const override {
return true;
}

expected<DataSetPtr>
SearchWithPreparedConfig(const DataSetPtr, std::shared_ptr<const Config> cfg, const BitsetView&,
milvus::OpContext*) const override {
const Config* expected = nullptr;
first_config_.compare_exchange_strong(expected, cfg.get());
reused_same_config_.store(reused_same_config_.load() && first_config_.load() == cfg.get());
return std::make_shared<DataSet>();
}

std::unique_ptr<BaseConfig>
CreateConfig() const override {
create_config_calls_.fetch_add(1);
return std::make_unique<BaseConfig>();
}

int
CreateConfigCalls() const {
return create_config_calls_.load();
}

bool
ReusedSameConfig() const {
return reused_same_config_.load();
}

private:
mutable std::atomic<int> create_config_calls_{0};
mutable std::atomic<const Config*> first_config_{nullptr};
mutable std::atomic<bool> reused_same_config_{true};
};

TEST_CASE("Test index node") {
auto version = GenTestVersionList();
DataSetPtr ds = std::make_shared<DataSet>();
Expand Down Expand Up @@ -208,3 +253,44 @@ TEST_CASE("Test index node") {
}
#pragma GCC diagnostic pop
}

TEST_CASE("Search reuses an immutable prepared config", "[search_config_cache]") {
KNOWHERE_SIMPLE_REGISTER_GLOBAL(SEARCH_CONFIG_CACHE, CachedSearchConfigIndexNode, fp32, knowhere::feature::FLOAT32);
const auto version = GenTestVersionList();
auto dataset = std::make_shared<DataSet>();
const Json base_search_config = {{meta::METRIC_TYPE, metric::L2}, {meta::TOPK, 10}};

SECTION("same config reuses the prepared object") {
auto index = IndexFactory::Instance().Create<fp32>("SEARCH_CONFIG_CACHE", version).value();
auto* node = dynamic_cast<CachedSearchConfigIndexNode<fp32>*>(index.Node());
REQUIRE(node != nullptr);

REQUIRE(index.Search(dataset, base_search_config, nullptr).has_value());
REQUIRE(index.Search(dataset, base_search_config, nullptr).has_value());
REQUIRE(node->CreateConfigCalls() == 1);
REQUIRE(node->ReusedSameConfig());

auto changed_search_config = base_search_config;
changed_search_config[meta::TOPK] = 20;
REQUIRE(index.Search(dataset, changed_search_config, nullptr).has_value());
REQUIRE(node->CreateConfigCalls() == 2);
}

SECTION("concurrent searches prepare the config once") {
auto index = IndexFactory::Instance().Create<fp32>("SEARCH_CONFIG_CACHE", version).value();
auto* node = dynamic_cast<CachedSearchConfigIndexNode<fp32>*>(index.Node());
REQUIRE(node != nullptr);

std::vector<std::future<expected<DataSetPtr>>> searches;
for (int i = 0; i < 32; ++i) {
searches.emplace_back(
std::async(std::launch::async, [&] { return index.Search(dataset, base_search_config, nullptr); }));
}
for (auto& search : searches) {
REQUIRE(search.get().has_value());
}

REQUIRE(node->CreateConfigCalls() == 1);
REQUIRE(node->ReusedSameConfig());
}
}
Loading