diff options
37 files changed, 1462 insertions, 176 deletions
diff --git a/searchcore/src/vespa/searchcore/proton/matchengine/matchengine.cpp b/searchcore/src/vespa/searchcore/proton/matchengine/matchengine.cpp index f59bad95343..4157fa221ab 100644 --- a/searchcore/src/vespa/searchcore/proton/matchengine/matchengine.cpp +++ b/searchcore/src/vespa/searchcore/proton/matchengine/matchengine.cpp @@ -118,7 +118,7 @@ MatchEngine::performSearch(search::engine::SearchRequest::Source req, const search::engine::SearchRequest * searchRequest = req.get(); if (searchRequest) { // 3 is the minimum level required for backend tracing. - searchRequest->setTraceLevel(search::fef::indexproperties::trace::Level::lookup(searchRequest->propertiesMap.modelOverrides()), 3); + searchRequest->setTraceLevel(search::fef::indexproperties::trace::Level::lookup(searchRequest->propertiesMap.modelOverrides(), searchRequest->getTraceLevel()), 3); ISearchHandler::SP searchHandler; vespalib::SimpleThreadBundle::UP threadBundle = _threadBundlePool.obtain(); { // try to find the match handler corresponding to the specified search doc type diff --git a/searchcore/src/vespa/searchcore/proton/server/proton.cpp b/searchcore/src/vespa/searchcore/proton/server/proton.cpp index 978720f88ae..8aef8d51c5f 100644 --- a/searchcore/src/vespa/searchcore/proton/server/proton.cpp +++ b/searchcore/src/vespa/searchcore/proton/server/proton.cpp @@ -482,6 +482,23 @@ size_t Proton::getNumActiveDocs() const return numDocs; } +search::engine::SearchServer & +Proton::get_search_server() +{ + return *_matchEngine; +} + +search::engine::DocsumServer & +Proton::get_docsum_server() +{ + return *_summaryEngine; +} + +search::engine::MonitorServer & +Proton::get_monitor_server() +{ + return *this; +} vespalib::string Proton::getDelayedConfigs() const @@ -631,6 +648,7 @@ Proton::ping(MonitorRequest::UP request, MonitorClient & client) BootstrapConfig::SP configSnapshot = getActiveConfigSnapshot(); const ProtonConfig &protonConfig = configSnapshot->getProtonConfig(); ret.partid = protonConfig.partition; + ret.distribution_key = protonConfig.distributionkey; ret.timestamp = (_matchEngine->isOnline()) ? 42 : 0; ret.activeDocs = getNumActiveDocs(); ret.activeDocsRequested = request->reportActiveDocs; diff --git a/searchcore/src/vespa/searchcore/proton/server/proton.h b/searchcore/src/vespa/searchcore/proton/server/proton.h index 14ddcee3d5c..fe7e8fe2219 100644 --- a/searchcore/src/vespa/searchcore/proton/server/proton.h +++ b/searchcore/src/vespa/searchcore/proton/server/proton.h @@ -194,6 +194,10 @@ public: size_t getNumActiveDocs() const; DocsumBySlime & getDocsumBySlime() { return *_docsumBySlime; } + search::engine::SearchServer &get_search_server(); + search::engine::DocsumServer &get_docsum_server(); + search::engine::MonitorServer &get_monitor_server(); + vespalib::string getDelayedConfigs() const; StatusReport::List getStatusReports() const override; diff --git a/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.cpp b/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.cpp index 3e00bdeb370..2d63f192189 100644 --- a/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.cpp +++ b/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.cpp @@ -196,6 +196,10 @@ RPCHooksBase::RPCHooksBase(Params ¶ms) : _proton(params.proton), _docsumByRPC(new DocsumByRPC(_proton.getDocsumBySlime())), _orb(std::make_unique<FRT_Supervisor>()), + _proto_rpc_adapter(std::make_unique<ProtoRpcAdapter>( + _proton.get_search_server(), + _proton.get_docsum_server(), + _proton.get_monitor_server(), *_orb)), _regAPI(*_orb, params.slobrok_config), _stateLock(), _stateCond(), diff --git a/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.h b/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.h index 057706e3d97..ad0a69fcd55 100644 --- a/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.h +++ b/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.h @@ -8,6 +8,7 @@ #include <vespa/vespalib/stllike/string.h> #include <vespa/vespalib/util/threadstackexecutor.h> #include <vespa/searchlib/common/packets.h> +#include <vespa/searchlib/engine/proto_rpc_adapter.h> #include <mutex> #include <condition_variable> @@ -19,6 +20,8 @@ class DocsumByRPC; class RPCHooksBase : public FRT_Invokable { private: + using ProtoRpcAdapter = search::engine::ProtoRpcAdapter; + class Session { private: fastos::TimeStamp _createTime; @@ -62,6 +65,7 @@ private: Proton & _proton; std::unique_ptr<DocsumByRPC> _docsumByRPC; std::unique_ptr<FRT_Supervisor> _orb; + std::unique_ptr<ProtoRpcAdapter> _proto_rpc_adapter; slobrok::api::RegisterAPI _regAPI; std::mutex _stateLock; std::condition_variable _stateCond; diff --git a/searchlib/CMakeLists.txt b/searchlib/CMakeLists.txt index 11863bead42..f03a7cfd445 100644 --- a/searchlib/CMakeLists.txt +++ b/searchlib/CMakeLists.txt @@ -126,6 +126,8 @@ vespa_define_module( src/tests/docstore/store_by_bucket src/tests/engine/docsumapi src/tests/engine/monitorapi + src/tests/engine/proto_converter + src/tests/engine/proto_rpc_adapter src/tests/engine/searchapi src/tests/engine/transportserver src/tests/expression/attributenode diff --git a/searchlib/src/protobuf/search_protocol.proto b/searchlib/src/protobuf/search_protocol.proto index 1cdf15729eb..6a4d3cfb07a 100644 --- a/searchlib/src/protobuf/search_protocol.proto +++ b/searchlib/src/protobuf/search_protocol.proto @@ -50,6 +50,7 @@ message SearchReply { bool degraded_by_soft_timeout = 6; repeated Hit hits = 7; bytes grouping_blob = 8; // serialized opaquely like now, to be changed later + bytes slime_trace = 9; } message Hit { @@ -57,3 +58,34 @@ message Hit { double relevance = 2; bytes sort_data = 3; } + +message DocsumRequest { + int32 timeout = 1; // milliseconds + string session_key = 2; + string document_type = 3; + string summary_class = 4; + bool cache_query = 5; + bool dump_features = 6; + string rank_profile = 7; + repeated StringProperty feature_overrides = 8; + repeated TensorProperty tensor_feature_overrides = 9; + repeated StringProperty rank_properties = 10; + repeated TensorProperty tensor_rank_properties = 11; + repeated StringProperty highlight_terms = 12; + string geo_location = 13; // to be moved into query_tree + bytes query_tree_blob = 14; // serialized opaquely like now, to be changed later + repeated bytes global_ids = 15; +} + +message DocsumReply { + bytes slime_summaries = 1; // result array inside slime object +} + +message MonitorRequest { +} + +message MonitorReply { + bool online = 1; + int64 active_docs = 2; + int32 distribution_key = 3; +} diff --git a/searchlib/src/tests/engine/proto_converter/CMakeLists.txt b/searchlib/src/tests/engine/proto_converter/CMakeLists.txt new file mode 100644 index 00000000000..718e8120c2c --- /dev/null +++ b/searchlib/src/tests/engine/proto_converter/CMakeLists.txt @@ -0,0 +1,9 @@ +# Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +vespa_add_executable(searchlib_engine_proto_converter_test_app TEST + SOURCES + proto_converter_test.cpp + DEPENDS + searchlib + gtest +) +vespa_add_test(NAME searchlib_engine_proto_converter_test_app COMMAND searchlib_engine_proto_converter_test_app) diff --git a/searchlib/src/tests/engine/proto_converter/proto_converter_test.cpp b/searchlib/src/tests/engine/proto_converter/proto_converter_test.cpp new file mode 100644 index 00000000000..e38820b6e8b --- /dev/null +++ b/searchlib/src/tests/engine/proto_converter/proto_converter_test.cpp @@ -0,0 +1,543 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#include <vespa/vespalib/gtest/gtest.h> +#include <vespa/searchlib/engine/proto_converter.h> +#include <vespa/searchlib/common/transport.h> +#include <vespa/vespalib/data/slime/slime.h> +#include <vespa/vespalib/data/slime/binary_format.h> + +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Winline" + +using Converter = ::search::engine::ProtoConverter; + +using SearchRequest = ::search::engine::SearchRequest; +using SearchReply = ::search::engine::SearchReply; + +using DocsumRequest = ::search::engine::DocsumRequest; +using DocsumReply = ::search::engine::DocsumReply; + +using MonitorRequest = ::search::engine::MonitorRequest; +using MonitorReply = ::search::engine::MonitorReply; + +using vespalib::Slime; +using vespalib::Memory; +using vespalib::slime::BinaryFormat; + +//----------------------------------------------------------------------------- + +struct SearchRequestTest : ::testing::Test { + Converter::ProtoSearchRequest proto; + SearchRequest request; + void convert() { Converter::search_request_from_proto(proto, request); } +}; + +TEST_F(SearchRequestTest, require_that_offset_is_converted) { + proto.set_offset(123); + convert(); + EXPECT_EQ(request.offset, 123); +} + +TEST_F(SearchRequestTest, require_that_hits_is_converted) { + proto.set_hits(17); + convert(); + EXPECT_EQ(request.maxhits, 17); +} + +TEST_F(SearchRequestTest, require_that_timeout_is_converted) { + proto.set_timeout(500); + convert(); + EXPECT_EQ(request.getTimeout().ms(), 500); +} + +TEST_F(SearchRequestTest, require_that_trace_level_is_converted) { + proto.set_trace_level(9); + convert(); + EXPECT_EQ(request.getTraceLevel(), 9); +} + +TEST_F(SearchRequestTest, require_that_sorting_is_converted) { + auto *sort_field = proto.add_sorting(); + sort_field->set_ascending(true); + sort_field->set_field("foo"); + sort_field = proto.add_sorting(); + sort_field->set_ascending(false); + sort_field->set_field("bar"); + convert(); + EXPECT_EQ(request.sortSpec, "+foo -bar"); +} + +TEST_F(SearchRequestTest, require_that_session_key_is_converted) { + proto.set_session_key("my-session"); + convert(); + EXPECT_EQ(std::string(&request.sessionId[0], request.sessionId.size()), "my-session"); +} + +TEST_F(SearchRequestTest, require_that_document_type_is_converted) { + proto.set_document_type("music"); + convert(); + EXPECT_EQ(request.propertiesMap.matchProperties().lookup("documentdb", "searchdoctype").get(""), "music"); +} + +TEST_F(SearchRequestTest, require_that_cache_grouping_is_converted) { + proto.set_cache_grouping(true); + convert(); + EXPECT_TRUE(request.propertiesMap.cacheProperties().lookup("grouping").found()); + EXPECT_FALSE(request.propertiesMap.cacheProperties().lookup("query").found()); +} + +TEST_F(SearchRequestTest, require_that_cache_query_is_converted) { + proto.set_cache_query(true); + convert(); + EXPECT_FALSE(request.propertiesMap.cacheProperties().lookup("grouping").found()); + EXPECT_TRUE(request.propertiesMap.cacheProperties().lookup("query").found()); +} + +TEST_F(SearchRequestTest, require_that_rank_profile_is_converted) { + proto.set_rank_profile("mlr"); + convert(); + EXPECT_EQ(request.ranking, "mlr"); +} + +TEST_F(SearchRequestTest, require_that_feature_overrides_are_converted) { + auto *prop = proto.add_feature_overrides(); + prop->set_name("foo"); + prop->add_values("a"); + prop = proto.add_feature_overrides(); + prop->set_name("bar"); + prop->add_values("b"); + prop->add_values("c"); + auto *tprop = proto.add_tensor_feature_overrides(); + tprop->set_name("x1"); + tprop->set_value("[1,2,3]"); + tprop = proto.add_tensor_feature_overrides(); + tprop->set_name("y1"); + tprop->set_value("[4,5]"); + convert(); + auto foo = request.propertiesMap.featureOverrides().lookup("foo"); + EXPECT_EQ(foo.size(), 1u); + EXPECT_EQ(foo.get(), "a"); + auto bar = request.propertiesMap.featureOverrides().lookup("bar"); + EXPECT_EQ(bar.size(), 2u); + EXPECT_EQ(bar.get(), "b"); + EXPECT_EQ(bar.getAt(1), "c"); + auto x1 = request.propertiesMap.featureOverrides().lookup("x1"); + EXPECT_EQ(x1.size(), 1u); + EXPECT_EQ(x1.get(), "[1,2,3]"); + auto y1 = request.propertiesMap.featureOverrides().lookup("y1"); + EXPECT_EQ(y1.size(), 1u); + EXPECT_EQ(y1.get(), "[4,5]"); +} + +TEST_F(SearchRequestTest, require_that_rank_properties_are_converted) { + auto *prop = proto.add_rank_properties(); + prop->set_name("foo"); + prop->add_values("a"); + prop = proto.add_rank_properties(); + prop->set_name("bar"); + prop->add_values("b"); + prop->add_values("c"); + auto *tprop = proto.add_tensor_rank_properties(); + tprop->set_name("x1"); + tprop->set_value("[1,2,3]"); + tprop = proto.add_tensor_rank_properties(); + tprop->set_name("y1"); + tprop->set_value("[4,5]"); + convert(); + auto foo = request.propertiesMap.rankProperties().lookup("foo"); + EXPECT_EQ(foo.size(), 1u); + EXPECT_EQ(foo.get(), "a"); + auto bar = request.propertiesMap.rankProperties().lookup("bar"); + EXPECT_EQ(bar.size(), 2u); + EXPECT_EQ(bar.get(), "b"); + EXPECT_EQ(bar.getAt(1), "c"); + auto x1 = request.propertiesMap.rankProperties().lookup("x1"); + EXPECT_EQ(x1.size(), 1u); + EXPECT_EQ(x1.get(), "[1,2,3]"); + auto y1 = request.propertiesMap.rankProperties().lookup("y1"); + EXPECT_EQ(y1.size(), 1u); + EXPECT_EQ(y1.get(), "[4,5]"); +} + +TEST_F(SearchRequestTest, require_that_grouping_blob_is_converted) { + proto.set_grouping_blob("grouping-blob"); + convert(); + EXPECT_EQ(std::string(&request.groupSpec[0], request.groupSpec.size()), "grouping-blob"); +} + +TEST_F(SearchRequestTest, require_that_geo_location_is_converted) { + proto.set_geo_location("x,y"); + convert(); + EXPECT_EQ(request.location, "x,y"); +} + +TEST_F(SearchRequestTest, require_that_query_tree_blob_is_converted) { + proto.set_query_tree_blob("query-tree-blob"); + convert(); + EXPECT_EQ(std::string(&request.stackDump[0], request.stackDump.size()), "query-tree-blob"); +} + +//----------------------------------------------------------------------------- + +struct SearchReplyTest : ::testing::Test { + SearchReply reply; + Converter::ProtoSearchReply proto; + void convert() { Converter::search_reply_to_proto(reply, proto); } +}; + +TEST_F(SearchReplyTest, require_that_total_hit_count_is_converted) { + reply.totalHitCount = 9001; + convert(); + EXPECT_EQ(proto.total_hit_count(), 9001); +} + +TEST_F(SearchReplyTest, require_that_coverage_docs_is_converted) { + reply.coverage.setCovered(150000); + convert(); + EXPECT_EQ(proto.coverage_docs(), 150000); +} + +TEST_F(SearchReplyTest, require_that_active_docs_is_converted) { + reply.coverage.setActive(200000); + convert(); + EXPECT_EQ(proto.active_docs(), 200000); +} + +TEST_F(SearchReplyTest, require_that_soon_active_docs_is_converted) { + reply.coverage.setSoonActive(250000); + convert(); + EXPECT_EQ(proto.soon_active_docs(), 250000); +} + +TEST_F(SearchReplyTest, require_that_degraded_by_match_phase_is_converted) { + reply.coverage.degradeMatchPhase(); + convert(); + EXPECT_TRUE(proto.degraded_by_match_phase()); + EXPECT_FALSE(proto.degraded_by_soft_timeout()); +} + +TEST_F(SearchReplyTest, require_that_degraded_by_soft_timeout_is_converted) { + reply.coverage.degradeTimeout(); + convert(); + EXPECT_FALSE(proto.degraded_by_match_phase()); + EXPECT_TRUE(proto.degraded_by_soft_timeout()); +} + +TEST_F(SearchReplyTest, require_that_multiple_degraded_reasons_are_converted) { + reply.coverage.degradeMatchPhase(); + reply.coverage.degradeTimeout(); + convert(); + EXPECT_TRUE(proto.degraded_by_match_phase()); + EXPECT_TRUE(proto.degraded_by_soft_timeout()); +} + +TEST_F(SearchReplyTest, require_that_hits_are_converted) { + constexpr size_t len = document::GlobalId::LENGTH; + ASSERT_EQ(len, 12); + char id0[len] = { 1, 2, 3, 4, 5, 6, 7, 8, 9,10,11,12}; + char id1[len] = {11,12,13,14,15,16,17,18,19,20,21,22}; + char id2[len] = {21,22,23,24,25,26,27,28,29,30,31,32}; + reply.hits.resize(3); + reply.hits[0].gid = document::GlobalId(id0); + reply.hits[0].metric = 100.0; + reply.hits[1].gid = document::GlobalId(id1); + reply.hits[1].metric = 50.0; + reply.hits[2].gid = document::GlobalId(id2); + reply.hits[2].metric = 10.0; + convert(); + ASSERT_EQ(proto.hits_size(), 3); + EXPECT_EQ(proto.hits(0).global_id(), std::string(id0, len)); + EXPECT_EQ(proto.hits(0).relevance(), 100.0); + EXPECT_TRUE(proto.hits(0).sort_data().empty()); + EXPECT_EQ(proto.hits(1).global_id(), std::string(id1, len)); + EXPECT_EQ(proto.hits(1).relevance(), 50.0); + EXPECT_TRUE(proto.hits(1).sort_data().empty()); + EXPECT_EQ(proto.hits(2).global_id(), std::string(id2, len)); + EXPECT_EQ(proto.hits(2).relevance(), 10.0); + EXPECT_TRUE(proto.hits(2).sort_data().empty()); +} + +TEST_F(SearchReplyTest, require_that_hits_with_sort_data_are_converted) { + constexpr size_t len = document::GlobalId::LENGTH; + ASSERT_EQ(len, 12); + char id0[len] = { 1, 2, 3, 4, 5, 6, 7, 8, 9,10,11,12}; + char id1[len] = {11,12,13,14,15,16,17,18,19,20,21,22}; + char id2[len] = {21,22,23,24,25,26,27,28,29,30,31,32}; + reply.hits.resize(3); + reply.hits[0].gid = document::GlobalId(id0); + reply.hits[0].metric = 100.0; + reply.hits[1].gid = document::GlobalId(id1); + reply.hits[1].metric = 50.0; + reply.hits[2].gid = document::GlobalId(id2); + reply.hits[2].metric = 10.0; + vespalib::string sort_data("fooxybar"); + reply.sortData.assign(sort_data.begin(), sort_data.end()); + reply.sortIndex.push_back(0); + reply.sortIndex.push_back(3); // hit1: 'foo' + reply.sortIndex.push_back(5); // hit2: 'xy' + reply.sortIndex.push_back(8); // hit3: 'bar' + convert(); + ASSERT_EQ(proto.hits_size(), 3); + EXPECT_EQ(proto.hits(0).global_id(), std::string(id0, len)); + EXPECT_EQ(proto.hits(0).relevance(), 100.0); + EXPECT_EQ(proto.hits(0).sort_data(), "foo"); + EXPECT_EQ(proto.hits(1).global_id(), std::string(id1, len)); + EXPECT_EQ(proto.hits(1).relevance(), 50.0); + EXPECT_EQ(proto.hits(1).sort_data(), "xy"); + EXPECT_EQ(proto.hits(2).global_id(), std::string(id2, len)); + EXPECT_EQ(proto.hits(2).relevance(), 10.0); + EXPECT_EQ(proto.hits(2).sort_data(), "bar"); +} + +TEST_F(SearchReplyTest, require_that_grouping_blob_is_converted) { + vespalib::string tmp("grouping-result"); + reply.groupResult.assign(tmp.begin(), tmp.end()); + convert(); + EXPECT_EQ(proto.grouping_blob(), "grouping-result"); +} + +TEST_F(SearchReplyTest, require_that_slime_trace_is_converted) { + reply.propertiesMap.lookupCreate("trace").add("slime", "slime-trace"); + convert(); + EXPECT_EQ(proto.slime_trace(), "slime-trace"); +} + +//----------------------------------------------------------------------------- + +struct DocsumRequestTest : ::testing::Test { + Converter::ProtoDocsumRequest proto; + DocsumRequest request; + DocsumRequestTest() : proto(), request(true) {} // <- use root slime + void convert() { Converter::docsum_request_from_proto(proto, request); } +}; + +TEST_F(DocsumRequestTest, require_that_root_slime_is_used) { + EXPECT_TRUE(request.useRootSlime()); +} + +TEST_F(DocsumRequestTest, require_that_timeout_is_converted) { + proto.set_timeout(500); + convert(); + EXPECT_EQ(request.getTimeout().ms(), 500); +} + +TEST_F(DocsumRequestTest, require_that_session_key_is_converted) { + proto.set_session_key("my-session"); + convert(); + EXPECT_EQ(std::string(&request.sessionId[0], request.sessionId.size()), "my-session"); +} + +TEST_F(DocsumRequestTest, require_that_document_type_is_converted) { + proto.set_document_type("music"); + convert(); + EXPECT_EQ(request.propertiesMap.matchProperties().lookup("documentdb", "searchdoctype").get(""), "music"); +} + +TEST_F(DocsumRequestTest, require_that_summary_class_is_converted) { + proto.set_summary_class("prefetch"); + convert(); + EXPECT_EQ(request.resultClassName, "prefetch"); +} + +TEST_F(DocsumRequestTest, require_that_cache_query_is_converted) { + proto.set_cache_query(true); + convert(); + EXPECT_TRUE(request.propertiesMap.cacheProperties().lookup("query").found()); + EXPECT_FALSE((request.queryFlags & search::fs4transport::QFLAG_DUMP_FEATURES) != 0); +} + +TEST_F(DocsumRequestTest, require_that_dump_features_is_converted) { + proto.set_dump_features(true); + convert(); + EXPECT_FALSE(request.propertiesMap.cacheProperties().lookup("query").found()); + EXPECT_TRUE((request.queryFlags & search::fs4transport::QFLAG_DUMP_FEATURES) != 0); +} + +TEST_F(DocsumRequestTest, require_that_rank_profile_is_converted) { + proto.set_rank_profile("mlr"); + convert(); + EXPECT_EQ(request.ranking, "mlr"); +} + +TEST_F(DocsumRequestTest, require_that_feature_overrides_are_converted) { + auto *prop = proto.add_feature_overrides(); + prop->set_name("foo"); + prop->add_values("a"); + prop = proto.add_feature_overrides(); + prop->set_name("bar"); + prop->add_values("b"); + prop->add_values("c"); + auto *tprop = proto.add_tensor_feature_overrides(); + tprop->set_name("x1"); + tprop->set_value("[1,2,3]"); + tprop = proto.add_tensor_feature_overrides(); + tprop->set_name("y1"); + tprop->set_value("[4,5]"); + convert(); + auto foo = request.propertiesMap.featureOverrides().lookup("foo"); + EXPECT_EQ(foo.size(), 1u); + EXPECT_EQ(foo.get(), "a"); + auto bar = request.propertiesMap.featureOverrides().lookup("bar"); + EXPECT_EQ(bar.size(), 2u); + EXPECT_EQ(bar.get(), "b"); + EXPECT_EQ(bar.getAt(1), "c"); + auto x1 = request.propertiesMap.featureOverrides().lookup("x1"); + EXPECT_EQ(x1.size(), 1u); + EXPECT_EQ(x1.get(), "[1,2,3]"); + auto y1 = request.propertiesMap.featureOverrides().lookup("y1"); + EXPECT_EQ(y1.size(), 1u); + EXPECT_EQ(y1.get(), "[4,5]"); +} + +TEST_F(DocsumRequestTest, require_that_rank_properties_are_converted) { + auto *prop = proto.add_rank_properties(); + prop->set_name("foo"); + prop->add_values("a"); + prop = proto.add_rank_properties(); + prop->set_name("bar"); + prop->add_values("b"); + prop->add_values("c"); + auto *tprop = proto.add_tensor_rank_properties(); + tprop->set_name("x1"); + tprop->set_value("[1,2,3]"); + tprop = proto.add_tensor_rank_properties(); + tprop->set_name("y1"); + tprop->set_value("[4,5]"); + convert(); + auto foo = request.propertiesMap.rankProperties().lookup("foo"); + EXPECT_EQ(foo.size(), 1u); + EXPECT_EQ(foo.get(), "a"); + auto bar = request.propertiesMap.rankProperties().lookup("bar"); + EXPECT_EQ(bar.size(), 2u); + EXPECT_EQ(bar.get(), "b"); + EXPECT_EQ(bar.getAt(1), "c"); + auto x1 = request.propertiesMap.rankProperties().lookup("x1"); + EXPECT_EQ(x1.size(), 1u); + EXPECT_EQ(x1.get(), "[1,2,3]"); + auto y1 = request.propertiesMap.rankProperties().lookup("y1"); + EXPECT_EQ(y1.size(), 1u); + EXPECT_EQ(y1.get(), "[4,5]"); +} + +TEST_F(DocsumRequestTest, require_that_highlight_terms_are_converted) { + auto *prop = proto.add_highlight_terms(); + prop->set_name("foo"); + prop->add_values("a"); + prop = proto.add_highlight_terms(); + prop->set_name("bar"); + prop->add_values("b"); + prop->add_values("c"); + convert(); + auto foo = request.propertiesMap.highlightTerms().lookup("foo"); + EXPECT_EQ(foo.size(), 1u); + EXPECT_EQ(foo.get(), "a"); + auto bar = request.propertiesMap.highlightTerms().lookup("bar"); + EXPECT_EQ(bar.size(), 2u); + EXPECT_EQ(bar.get(), "b"); + EXPECT_EQ(bar.getAt(1), "c"); +} + +TEST_F(DocsumRequestTest, require_that_geo_location_is_converted) { + proto.set_geo_location("x,y"); + convert(); + EXPECT_EQ(request.location, "x,y"); +} + +TEST_F(DocsumRequestTest, require_that_query_tree_blob_is_converted) { + proto.set_query_tree_blob("query-tree-blob"); + convert(); + EXPECT_EQ(std::string(&request.stackDump[0], request.stackDump.size()), "query-tree-blob"); +} + +TEST_F(DocsumRequestTest, require_that_global_ids_are_converted) { + constexpr size_t len = document::GlobalId::LENGTH; + ASSERT_EQ(len, 12); + char id0[len] = { 1, 2, 3, 4, 5, 6, 7, 8, 9,10,11,12}; + char id1[len] = {11,12,13,14,15,16,17,18,19,20,21,22}; + char id2[len] = {21,22,23,24,25,26,27,28,29,30,31,32}; + proto.add_global_ids(id0, len); + proto.add_global_ids(id1, len); + proto.add_global_ids(id2, len); + convert(); + ASSERT_EQ(request.hits.size(), 3); + EXPECT_EQ(request.hits[0].gid, document::GlobalId(id0)); + EXPECT_EQ(request.hits[1].gid, document::GlobalId(id1)); + EXPECT_EQ(request.hits[2].gid, document::GlobalId(id2)); +} + +//----------------------------------------------------------------------------- + +struct DocsumReplyTest : ::testing::Test { + DocsumReply reply; + Converter::ProtoDocsumReply proto; + void convert() { Converter::docsum_reply_to_proto(reply, proto); } +}; + +TEST_F(DocsumReplyTest, require_that_slime_summaries_are_converted) { + reply._root = std::make_unique<Slime>(); + auto &list = reply._root->setArray(); + auto &doc0 = list.addObject(); + doc0.setLong("my_field", 42); + convert(); + const auto &mem = proto.slime_summaries(); + Slime slime; + EXPECT_EQ(BinaryFormat::decode(Memory(mem.data(), mem.size()), slime), mem.size()); + EXPECT_EQ(slime.get()[0]["my_field"].asLong(), 42); +} + +TEST_F(DocsumReplyTest, require_that_missing_root_slime_gives_empty_payload) { + reply._root.reset(); + convert(); + EXPECT_EQ(proto.slime_summaries().size(), 0); +} + +//----------------------------------------------------------------------------- + +struct MonitorRequestTest : ::testing::Test { + Converter::ProtoMonitorRequest proto; + MonitorRequest request; + void convert() { Converter::monitor_request_from_proto(proto, request); } +}; + +TEST_F(MonitorRequestTest, require_that_active_docs_are_always_requested) { + convert(); + EXPECT_TRUE(request.reportActiveDocs); +} + +//----------------------------------------------------------------------------- + +struct MonitorReplyTest : ::testing::Test { + MonitorReply reply; + Converter::ProtoMonitorReply proto; + void convert() { Converter::monitor_reply_to_proto(reply, proto); } +}; + +TEST_F(MonitorReplyTest, require_that_zero_timestamp_is_converted_to_online_false) { + reply.timestamp = 0; + convert(); + EXPECT_FALSE(proto.online()); +} + +TEST_F(MonitorReplyTest, require_that_nonzero_timestamp_is_converted_to_online_true) { + reply.timestamp = 42; + convert(); + EXPECT_TRUE(proto.online()); +} + +TEST_F(MonitorReplyTest, require_that_active_docs_is_converted) { + reply.activeDocs = 12345; + convert(); + EXPECT_EQ(proto.active_docs(), 12345); +} + +TEST_F(MonitorReplyTest, require_that_distribution_key_is_converted) { + reply.distribution_key = 7; + convert(); + EXPECT_EQ(proto.distribution_key(), 7); +} + +//----------------------------------------------------------------------------- + +GTEST_MAIN_RUN_ALL_TESTS() + +#pragma GCC diagnostic pop diff --git a/searchlib/src/tests/engine/proto_rpc_adapter/CMakeLists.txt b/searchlib/src/tests/engine/proto_rpc_adapter/CMakeLists.txt new file mode 100644 index 00000000000..3fa638f16e5 --- /dev/null +++ b/searchlib/src/tests/engine/proto_rpc_adapter/CMakeLists.txt @@ -0,0 +1,9 @@ +# Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. +vespa_add_executable(searchlib_engine_proto_rpc_adapter_test_app TEST + SOURCES + proto_rpc_adapter_test.cpp + DEPENDS + searchlib + gtest +) +vespa_add_test(NAME searchlib_engine_proto_rpc_adapter_test_app COMMAND searchlib_engine_proto_rpc_adapter_test_app) diff --git a/searchlib/src/tests/engine/proto_rpc_adapter/proto_rpc_adapter_test.cpp b/searchlib/src/tests/engine/proto_rpc_adapter/proto_rpc_adapter_test.cpp new file mode 100644 index 00000000000..60cf1af13a0 --- /dev/null +++ b/searchlib/src/tests/engine/proto_rpc_adapter/proto_rpc_adapter_test.cpp @@ -0,0 +1,145 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#include <vespa/vespalib/gtest/gtest.h> +#include <vespa/searchlib/engine/search_protocol_proto.h> +#include <vespa/searchlib/engine/proto_rpc_adapter.h> +#include <vespa/searchlib/engine/searchapi.h> +#include <vespa/searchlib/engine/docsumapi.h> +#include <vespa/searchlib/engine/monitorapi.h> +#include <vespa/fnet/frt/frt.h> +#include <vespa/vespalib/data/slime/slime.h> +#include <vespa/vespalib/data/slime/binary_format.h> + +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Winline" + +using namespace search::engine; + +using vespalib::Slime; +using vespalib::Memory; +using vespalib::slime::BinaryFormat; + +using ProtoSearchRequest = ProtoRpcAdapter::ProtoSearchRequest; +using ProtoSearchReply = ProtoRpcAdapter::ProtoSearchReply; +using ProtoDocsumRequest = ProtoRpcAdapter::ProtoDocsumRequest; +using ProtoDocsumReply = ProtoRpcAdapter::ProtoDocsumReply; +using ProtoMonitorRequest = ProtoRpcAdapter::ProtoMonitorRequest; +using ProtoMonitorReply = ProtoRpcAdapter::ProtoMonitorReply; + +struct MySearchServer : SearchServer { + SearchReply::UP search(SearchRequest::Source src, SearchClient &client) override { + auto req = src.release(); + assert(req); + auto reply = std::make_unique<SearchReply>(); + reply->totalHitCount = req->offset; // simplified search implementation + client.searchDone(std::move(reply)); // simplified async response + return std::unique_ptr<SearchReply>(); + } +}; + +struct MyDocsumServer : DocsumServer { + DocsumReply::UP getDocsums(DocsumRequest::Source src, DocsumClient &client) override { + auto req = src.release(); + assert(req); + auto reply = std::make_unique<DocsumReply>(); + reply->_root = std::make_unique<Slime>(); + auto &list = reply->_root->setArray(); + list.addObject().setBool("use_root_slime", req->useRootSlime()); + list.addObject().setString("ranking", req->ranking); + client.getDocsumsDone(std::move(reply)); // simplified async response + return std::unique_ptr<DocsumReply>(); + } +}; + +struct MyMonitorServer : MonitorServer { + MonitorReply::UP ping(MonitorRequest::UP req, MonitorClient &) override { + (void) req; + assert(req); + auto reply = std::make_unique<MonitorReply>(); + reply->activeDocs = 53; + return reply; // proton does sync response here + } +}; + +struct ProtoRpcAdapterTest : ::testing::Test { + FRT_Supervisor orb; + MySearchServer search; + MyDocsumServer docsum; + MyMonitorServer monitor; + ProtoRpcAdapter adapter; + ProtoRpcAdapterTest() + : orb(), adapter(search, docsum, monitor, orb) + { + orb.Listen(0); + orb.Start(); + } + FRT_Target *connect() { + return orb.GetTarget(orb.GetListenPort()); + } + ~ProtoRpcAdapterTest() { + orb.ShutDown(true); + } +}; + +//----------------------------------------------------------------------------- + +TEST_F(ProtoRpcAdapterTest, require_that_plain_rpc_ping_works) { + auto target = connect(); + auto *req = new FRT_RPCRequest(); + req->SetMethodName("frt.rpc.ping"); + target->InvokeSync(req, 60.0); + EXPECT_TRUE(req->CheckReturnTypes("")); + req->SubRef(); + target->SubRef(); +} + +TEST_F(ProtoRpcAdapterTest, require_that_proto_rpc_search_works) { + auto target = connect(); + auto *rpc = new FRT_RPCRequest(); + ProtoSearchRequest req; + req.set_offset(42); + ProtoRpcAdapter::encode_search_request(req, *rpc); + target->InvokeSync(rpc, 60.0); + ProtoSearchReply reply; + EXPECT_TRUE(ProtoRpcAdapter::decode_search_reply(*rpc, reply)); + EXPECT_EQ(reply.total_hit_count(), 42); + rpc->SubRef(); + target->SubRef(); +} + +TEST_F(ProtoRpcAdapterTest, require_that_proto_rpc_getDocsums_works) { + auto target = connect(); + auto *rpc = new FRT_RPCRequest(); + ProtoDocsumRequest req; + req.set_rank_profile("mlr"); + ProtoRpcAdapter::encode_docsum_request(req, *rpc); + target->InvokeSync(rpc, 60.0); + ProtoDocsumReply reply; + EXPECT_TRUE(ProtoRpcAdapter::decode_docsum_reply(*rpc, reply)); + const auto &mem = reply.slime_summaries(); + Slime slime; + EXPECT_EQ(BinaryFormat::decode(Memory(mem.data(), mem.size()), slime), mem.size()); + EXPECT_EQ(slime.get()[0]["use_root_slime"].asBool(), true); + EXPECT_EQ(slime.get()[1]["ranking"].asString().make_string(), "mlr"); + rpc->SubRef(); + target->SubRef(); +} + +TEST_F(ProtoRpcAdapterTest, require_that_proto_rpc_ping_works) { + auto target = connect(); + auto *rpc = new FRT_RPCRequest(); + ProtoMonitorRequest req; + ProtoRpcAdapter::encode_monitor_request(req, *rpc); + target->InvokeSync(rpc, 60.0); + ProtoMonitorReply reply; + EXPECT_TRUE(ProtoRpcAdapter::decode_monitor_reply(*rpc, reply)); + EXPECT_EQ(reply.active_docs(), 53); + rpc->SubRef(); + target->SubRef(); +} + +//----------------------------------------------------------------------------- + +GTEST_MAIN_RUN_ALL_TESTS() + +#pragma GCC diagnostic pop diff --git a/searchlib/src/tests/engine/searchapi/searchapi_test.cpp b/searchlib/src/tests/engine/searchapi/searchapi_test.cpp index ed103bf501c..99d1967b0df 100644 --- a/searchlib/src/tests/engine/searchapi/searchapi_test.cpp +++ b/searchlib/src/tests/engine/searchapi/searchapi_test.cpp @@ -37,6 +37,7 @@ TEST("propertyNames") { EXPECT_EQUAL(search::MapNames::HIGHLIGHTTERMS, "highlightterms"); EXPECT_EQUAL(search::MapNames::MATCH, "match"); EXPECT_EQUAL(search::MapNames::CACHES, "caches"); + EXPECT_EQUAL(search::MapNames::TRACE, "trace"); } TEST("convertToReques") { diff --git a/searchlib/src/vespa/searchlib/CMakeLists.txt b/searchlib/src/vespa/searchlib/CMakeLists.txt index df29e76f85e..2ec3cabaf4a 100644 --- a/searchlib/src/vespa/searchlib/CMakeLists.txt +++ b/searchlib/src/vespa/searchlib/CMakeLists.txt @@ -36,3 +36,5 @@ vespa_add_library(searchlib staging_vespalib atomic ) + +vespa_add_target_package_dependency(searchlib Protobuf) diff --git a/searchlib/src/vespa/searchlib/common/mapnames.cpp b/searchlib/src/vespa/searchlib/common/mapnames.cpp index 79d10d4b95d..8aadf6254f3 100644 --- a/searchlib/src/vespa/searchlib/common/mapnames.cpp +++ b/searchlib/src/vespa/searchlib/common/mapnames.cpp @@ -10,5 +10,6 @@ const vespalib::string MapNames::HIGHLIGHTTERMS("highlightterms"); const vespalib::string MapNames::MATCH("match"); const vespalib::string MapNames::CACHES("caches"); const vespalib::string MapNames::MODEL("model"); +const vespalib::string MapNames::TRACE("trace"); } // namespace search diff --git a/searchlib/src/vespa/searchlib/common/mapnames.h b/searchlib/src/vespa/searchlib/common/mapnames.h index d321c666f0d..a754762a659 100644 --- a/searchlib/src/vespa/searchlib/common/mapnames.h +++ b/searchlib/src/vespa/searchlib/common/mapnames.h @@ -28,6 +28,9 @@ struct MapNames /** name of model property collection **/ static const vespalib::string MODEL; + + /** name of trace property collection **/ + static const vespalib::string TRACE; }; } // namespace search diff --git a/searchlib/src/vespa/searchlib/engine/.gitignore b/searchlib/src/vespa/searchlib/engine/.gitignore index 583460ae288..c04e26fa1c3 100644 --- a/searchlib/src/vespa/searchlib/engine/.gitignore +++ b/searchlib/src/vespa/searchlib/engine/.gitignore @@ -1,3 +1,5 @@ *.So .depend Makefile +/*.pb.h +/*.pb.cc diff --git a/searchlib/src/vespa/searchlib/engine/CMakeLists.txt b/searchlib/src/vespa/searchlib/engine/CMakeLists.txt index e475d786a60..1957a882d9a 100644 --- a/searchlib/src/vespa/searchlib/engine/CMakeLists.txt +++ b/searchlib/src/vespa/searchlib/engine/CMakeLists.txt @@ -1,20 +1,31 @@ # Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +find_package(Protobuf REQUIRED) +protobuf_generate_cpp(searchlib_engine_PROTOBUF_SRCS searchlib_engine_PROTOBUF_HDRS ../../../../src/protobuf/search_protocol.proto) + +# protoc-generated files emit compiler warnings that we normally treat as errors. +set_source_files_properties(${searchlib_engine_PROTOBUF_SRCS} + PROPERTIES COMPILE_FLAGS "-Wno-suggest-override") + vespa_add_library(searchlib_engine OBJECT SOURCES docsumapi.cpp docsumreply.cpp docsumrequest.cpp errorcodes.cpp + lazy_source.cpp monitorreply.cpp monitorrequest.cpp packetconverter.cpp propertiesmap.cpp + proto_converter.cpp + proto_rpc_adapter.cpp request.cpp searchreply.cpp searchrequest.cpp - source_description.cpp trace.cpp transport_metrics.cpp transportserver.cpp + ${searchlib_engine_PROTOBUF_SRCS} DEPENDS ) diff --git a/searchlib/src/vespa/searchlib/engine/docsumrequest.cpp b/searchlib/src/vespa/searchlib/engine/docsumrequest.cpp index 1a4413ad09b..e2dffb54e56 100644 --- a/searchlib/src/vespa/searchlib/engine/docsumrequest.cpp +++ b/searchlib/src/vespa/searchlib/engine/docsumrequest.cpp @@ -25,35 +25,4 @@ DocsumRequest::DocsumRequest(RelativeTime relativeTime, bool useRootSlime_) DocsumRequest::~DocsumRequest() = default; -void DocsumRequest::Source::lazyDecode() const -{ - if ( !_request && (_fs4Packet != nullptr)) { - _request = std::make_unique<DocsumRequest>(std::move(*_relativeTime), false); - PacketConverter::toDocsumRequest(*_fs4Packet, *_request); - _fs4Packet->Free(); - _fs4Packet = nullptr; - } -} - -DocsumRequest::Source::Source(FS4Packet_GETDOCSUMSX *query, SourceDescription desc) - : _request(), - _fs4Packet(query), - _desc(desc), - _relativeTime(std::make_unique<RelativeTime>(std::make_unique<FastosClock>())) -{ } - -DocsumRequest::Source::Source(Source && rhs) noexcept - : _request(std::move(rhs._request)), - _fs4Packet(rhs._fs4Packet), - _desc(std::move(rhs._desc)), - _relativeTime(std::move(rhs._relativeTime)) -{ - rhs._fs4Packet = nullptr; -} -DocsumRequest::Source::~Source() { - if (_fs4Packet != nullptr) { - _fs4Packet->Free(); - } -} - } diff --git a/searchlib/src/vespa/searchlib/engine/docsumrequest.h b/searchlib/src/vespa/searchlib/engine/docsumrequest.h index ef9c46de04b..8aa8d036b73 100644 --- a/searchlib/src/vespa/searchlib/engine/docsumrequest.h +++ b/searchlib/src/vespa/searchlib/engine/docsumrequest.h @@ -4,7 +4,7 @@ #include "propertiesmap.h" #include "request.h" -#include "source_description.h" +#include "lazy_source.h" #include <vespa/document/base/globalid.h> #include <vespa/searchlib/common/hitrank.h> @@ -19,39 +19,7 @@ public: using UP = std::unique_ptr<DocsumRequest>; using SP = std::shared_ptr<DocsumRequest>; - - class Source { - private: - mutable DocsumRequest::UP _request; - mutable FS4Packet_GETDOCSUMSX *_fs4Packet; - void lazyDecode() const; - const SourceDescription _desc; - std::unique_ptr<RelativeTime> _relativeTime; - public: - - Source(DocsumRequest * request) : _request(request), _fs4Packet(nullptr), _desc(0), _relativeTime() {} - Source(DocsumRequest::UP request) : _request(std::move(request)), _fs4Packet(nullptr), _desc(0), _relativeTime() {} - Source(FS4Packet_GETDOCSUMSX *query, SourceDescription desc); - - Source(Source && rhs) noexcept; - ~Source(); - - const DocsumRequest * operator -> () const { return get(); } - - const DocsumRequest * get() const { - lazyDecode(); - return _request.get(); - } - - Source& operator= (Source && rhs) = delete; - Source & operator= (const Source &) = delete; - Source(const Source &) = delete; - - UP release() { - lazyDecode(); - return std::move(_request); - } - }; + using Source = LazySource<DocsumRequest>; class Hit { public: diff --git a/searchlib/src/vespa/searchlib/engine/lazy_source.cpp b/searchlib/src/vespa/searchlib/engine/lazy_source.cpp new file mode 100644 index 00000000000..96fcfaf4a7d --- /dev/null +++ b/searchlib/src/vespa/searchlib/engine/lazy_source.cpp @@ -0,0 +1,6 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#include "lazy_source.h" + +namespace search::engine { +} diff --git a/searchlib/src/vespa/searchlib/engine/lazy_source.h b/searchlib/src/vespa/searchlib/engine/lazy_source.h new file mode 100644 index 00000000000..ca31ada8d21 --- /dev/null +++ b/searchlib/src/vespa/searchlib/engine/lazy_source.h @@ -0,0 +1,52 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#pragma once + +#include <memory> + +namespace search::engine { + +/** + * A lazy source uses a decoder interface to delay decoding an + * object. Decoding is typically done in another thread as well to + * avoid slowing down the critical path (io event loop). + **/ +template <typename T> +class LazySource { +public: + struct Decoder { + virtual std::unique_ptr<T> decode() = 0; + virtual ~Decoder() = default; + }; +private: + mutable std::unique_ptr<T> _object; + mutable std::unique_ptr<Decoder> _decoder; + void lazy_decode() const { + if (_decoder && !_object) { + _object = _decoder->decode(); + _decoder.reset(); + } + } +public: + explicit LazySource(T *object) noexcept : _object(object), _decoder() {} + LazySource(std::unique_ptr<T> object) noexcept : _object(std::move(object)), _decoder() {} + LazySource(std::unique_ptr<Decoder> decoder) noexcept : _object(), _decoder(std::move(decoder)) {} + LazySource(LazySource &&) = default; + ~LazySource() {} + + LazySource &operator=(LazySource &&) = delete; + LazySource(const LazySource &) = delete; + LazySource &operator=(const LazySource &) = delete; + + const T *get() const { + lazy_decode(); + return _object.get(); + } + const T *operator->() const { return get(); } + std::unique_ptr<T> release() { + lazy_decode(); + return std::move(_object); + } +}; + +} diff --git a/searchlib/src/vespa/searchlib/engine/monitorreply.cpp b/searchlib/src/vespa/searchlib/engine/monitorreply.cpp index 46525244dc4..c0f7fa0d42a 100644 --- a/searchlib/src/vespa/searchlib/engine/monitorreply.cpp +++ b/searchlib/src/vespa/searchlib/engine/monitorreply.cpp @@ -8,6 +8,7 @@ MonitorReply::MonitorReply() : mld(), activeDocsRequested(false), partid(), + distribution_key(-1), timestamp(), totalNodes(), activeNodes(), diff --git a/searchlib/src/vespa/searchlib/engine/monitorreply.h b/searchlib/src/vespa/searchlib/engine/monitorreply.h index 600a855a4ec..7d1f0ce1cef 100644 --- a/searchlib/src/vespa/searchlib/engine/monitorreply.h +++ b/searchlib/src/vespa/searchlib/engine/monitorreply.h @@ -13,6 +13,7 @@ struct MonitorReply bool mld; bool activeDocsRequested; uint32_t partid; + int32_t distribution_key; uint32_t timestamp; uint32_t totalNodes; // mld uint32_t activeNodes; // mld diff --git a/searchlib/src/vespa/searchlib/engine/propertiesmap.h b/searchlib/src/vespa/searchlib/engine/propertiesmap.h index ba80e4a2f05..a1884e33200 100644 --- a/searchlib/src/vespa/searchlib/engine/propertiesmap.h +++ b/searchlib/src/vespa/searchlib/engine/propertiesmap.h @@ -123,6 +123,14 @@ public: return lookup(MapNames::MODEL); } + /** + * Obtain trace + * + * @return trace + **/ + const Props &trace() const { + return lookup(MapNames::TRACE); + } }; } diff --git a/searchlib/src/vespa/searchlib/engine/proto_converter.cpp b/searchlib/src/vespa/searchlib/engine/proto_converter.cpp new file mode 100644 index 00000000000..1736cf1f72a --- /dev/null +++ b/searchlib/src/vespa/searchlib/engine/proto_converter.cpp @@ -0,0 +1,180 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#include "proto_converter.h" +#include <vespa/searchlib/common/mapnames.h> +#include <vespa/vespalib/data/slime/slime.h> +#include <vespa/vespalib/data/slime/binary_format.h> +#include <vespa/vespalib/data/smart_buffer.h> +#include <vespa/searchlib/common/transport.h> + +namespace search::engine { + +namespace { + +template <typename T> +vespalib::string make_sort_spec(const T &sorting) { + vespalib::string spec; + for (const auto &field_spec: sorting) { + if (!spec.empty()) { + spec.push_back(' '); + } + if (field_spec.ascending()) { + spec.push_back('+'); + } else { + spec.push_back('-'); + } + spec.append(field_spec.field()); + } + return spec; +} + +template <typename T> +void add_single_props(fef::Properties &dst, const T &src) { + for (const auto &entry: src) { + dst.add(entry.name(), entry.value()); + } +} + +template <typename T> +void add_multi_props(fef::Properties &dst, const T &src) { + for (const auto &entry: src) { + for (int i = 0; i < entry.values_size(); ++i) { + dst.add(entry.name(), entry.values(i)); + } + } +} + +} + +//----------------------------------------------------------------------------- + +void +ProtoConverter::search_request_from_proto(const ProtoSearchRequest &proto, SearchRequest &request) +{ + request.offset = proto.offset(); + request.maxhits = proto.hits(); + request.setTimeout(fastos::TimeStamp::Seconds(0.001 * proto.timeout())); + request.setTraceLevel(proto.trace_level()); + request.sortSpec = make_sort_spec(proto.sorting()); + request.sessionId.assign(proto.session_key().begin(), proto.session_key().end()); + request.propertiesMap.lookupCreate(MapNames::MATCH).add("documentdb.searchdoctype", proto.document_type()); + if (proto.cache_grouping()) { + request.propertiesMap.lookupCreate(MapNames::CACHES).add("grouping", "true"); + } + if (proto.cache_query()) { + request.propertiesMap.lookupCreate(MapNames::CACHES).add("query", "true"); + } + request.ranking = proto.rank_profile(); + if ((proto.feature_overrides_size() + proto.tensor_feature_overrides_size()) > 0) { + auto &feature_overrides = request.propertiesMap.lookupCreate(MapNames::FEATURE); + add_multi_props(feature_overrides, proto.feature_overrides()); + add_single_props(feature_overrides, proto.tensor_feature_overrides()); + } + if ((proto.rank_properties_size() + proto.tensor_rank_properties_size()) > 0) { + auto &rank_props = request.propertiesMap.lookupCreate(MapNames::RANK); + add_multi_props(rank_props, proto.rank_properties()); + add_single_props(rank_props, proto.tensor_rank_properties()); + } + request.groupSpec.assign(proto.grouping_blob().begin(), proto.grouping_blob().end()); + request.location = proto.geo_location(); + request.stackDump.assign(proto.query_tree_blob().begin(), proto.query_tree_blob().end()); +} + +void +ProtoConverter::search_reply_to_proto(const SearchReply &reply, ProtoSearchReply &proto) +{ + proto.set_total_hit_count(reply.totalHitCount); + proto.set_coverage_docs(reply.coverage.getCovered()); + proto.set_active_docs(reply.coverage.getActive()); + proto.set_soon_active_docs(reply.coverage.getSoonActive()); + proto.set_degraded_by_match_phase(reply.coverage.wasDegradedByMatchPhase()); + proto.set_degraded_by_soft_timeout(reply.coverage.wasDegradedByTimeout()); + bool has_sort_data = (reply.sortIndex.size() > 0); + assert(!has_sort_data || (reply.sortIndex.size() == (reply.hits.size() + 1))); + for (size_t i = 0; i < reply.hits.size(); ++i) { + auto *hit = proto.add_hits(); + hit->set_global_id(reply.hits[i].gid.get(), document::GlobalId::LENGTH); + hit->set_relevance(reply.hits[i].metric); + if (has_sort_data) { + size_t sort_data_offset = reply.sortIndex[i]; + size_t sort_data_size = (reply.sortIndex[i + 1] - reply.sortIndex[i]); + assert((sort_data_offset + sort_data_size) <= reply.sortData.size()); + hit->set_sort_data(&reply.sortData[sort_data_offset], sort_data_size); + } + } + proto.set_grouping_blob(&reply.groupResult[0], reply.groupResult.size()); + const auto &slime_trace = reply.propertiesMap.trace().lookup("slime"); + proto.set_slime_trace(slime_trace.get().data(), slime_trace.get().size()); +} + +//----------------------------------------------------------------------------- + +void +ProtoConverter::docsum_request_from_proto(const ProtoDocsumRequest &proto, DocsumRequest &request) +{ + request.setTimeout(fastos::TimeStamp::Seconds(0.001 * proto.timeout())); + request.sessionId.assign(proto.session_key().begin(), proto.session_key().end()); + request.propertiesMap.lookupCreate(MapNames::MATCH).add("documentdb.searchdoctype", proto.document_type()); + request.resultClassName = proto.summary_class(); + if (proto.cache_query()) { + request.propertiesMap.lookupCreate(MapNames::CACHES).add("query", "true"); + } + if (proto.dump_features()) { + request.queryFlags |= fs4transport::QFLAG_DUMP_FEATURES; + } + request.ranking = proto.rank_profile(); + if ((proto.feature_overrides_size() + proto.tensor_feature_overrides_size()) > 0) { + auto &feature_overrides = request.propertiesMap.lookupCreate(MapNames::FEATURE); + add_multi_props(feature_overrides, proto.feature_overrides()); + add_single_props(feature_overrides, proto.tensor_feature_overrides()); + } + if ((proto.rank_properties_size() + proto.tensor_rank_properties_size()) > 0) { + auto &rank_props = request.propertiesMap.lookupCreate(MapNames::RANK); + add_multi_props(rank_props, proto.rank_properties()); + add_single_props(rank_props, proto.tensor_rank_properties()); + } + if(proto.highlight_terms_size() > 0) { + auto &highlight_terms = request.propertiesMap.lookupCreate(MapNames::HIGHLIGHTTERMS); + add_multi_props(highlight_terms, proto.highlight_terms()); + } + request.location = proto.geo_location(); + request.stackDump.assign(proto.query_tree_blob().begin(), proto.query_tree_blob().end()); + request.hits.resize(proto.global_ids_size()); + for (int i = 0; i < proto.global_ids_size(); ++i) { + const auto &gid = proto.global_ids(i); + if (gid.size() == document::GlobalId::LENGTH) { + request.hits[i].gid = document::GlobalId(gid.data()); + } + } +} + +void +ProtoConverter::docsum_reply_to_proto(const DocsumReply &reply, ProtoDocsumReply &proto) +{ + if (reply._root) { + vespalib::SmartBuffer buf(4096); + vespalib::slime::BinaryFormat::encode(*reply._root, buf); + proto.set_slime_summaries(buf.obtain().data, buf.obtain().size); + } +} + +//----------------------------------------------------------------------------- + +void +ProtoConverter::monitor_request_from_proto(const ProtoMonitorRequest &proto, MonitorRequest &request) +{ + (void) proto; + request.reportActiveDocs = true; +} + +void +ProtoConverter::monitor_reply_to_proto(const MonitorReply &reply, ProtoMonitorReply &proto) +{ + proto.set_online(reply.timestamp != 0); + proto.set_active_docs(reply.activeDocs); + proto.set_distribution_key(reply.distribution_key); +} + +//----------------------------------------------------------------------------- + +} diff --git a/searchlib/src/vespa/searchlib/engine/proto_converter.h b/searchlib/src/vespa/searchlib/engine/proto_converter.h new file mode 100644 index 00000000000..3cdb14e11b8 --- /dev/null +++ b/searchlib/src/vespa/searchlib/engine/proto_converter.h @@ -0,0 +1,35 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#pragma once + +#include "searchrequest.h" +#include "searchreply.h" +#include "docsumrequest.h" +#include "docsumreply.h" +#include "monitorrequest.h" +#include "monitorreply.h" +#include "search_protocol_proto.h" + +namespace search::engine { + +struct ProtoConverter { + using ProtoSearchRequest = ::searchlib::searchprotocol::protobuf::SearchRequest; + using ProtoSearchReply = ::searchlib::searchprotocol::protobuf::SearchReply; + + using ProtoDocsumRequest = ::searchlib::searchprotocol::protobuf::DocsumRequest; + using ProtoDocsumReply = ::searchlib::searchprotocol::protobuf::DocsumReply; + + using ProtoMonitorRequest = ::searchlib::searchprotocol::protobuf::MonitorRequest; + using ProtoMonitorReply = ::searchlib::searchprotocol::protobuf::MonitorReply; + + static void search_request_from_proto(const ProtoSearchRequest &proto, SearchRequest &request); + static void search_reply_to_proto(const SearchReply &reply, ProtoSearchReply &proto); + + static void docsum_request_from_proto(const ProtoDocsumRequest &proto, DocsumRequest &request); + static void docsum_reply_to_proto(const DocsumReply &reply, ProtoDocsumReply &proto); + + static void monitor_request_from_proto(const ProtoMonitorRequest &proto, MonitorRequest &request); + static void monitor_reply_to_proto(const MonitorReply &reply, ProtoMonitorReply &proto); +}; + +} diff --git a/searchlib/src/vespa/searchlib/engine/proto_rpc_adapter.cpp b/searchlib/src/vespa/searchlib/engine/proto_rpc_adapter.cpp new file mode 100644 index 00000000000..652ef2e0889 --- /dev/null +++ b/searchlib/src/vespa/searchlib/engine/proto_rpc_adapter.cpp @@ -0,0 +1,268 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#include "proto_rpc_adapter.h" +#include "searchapi.h" +#include "docsumapi.h" +#include "monitorapi.h" +#include <vespa/fnet/frt/rpcrequest.h> +#include <vespa/fnet/frt/supervisor.h> +#include <vespa/vespalib/util/compressor.h> +#include <vespa/searchlib/util/slime_output_raw_buf_adapter.h> +#include <vespa/vespalib/data/databuffer.h> +#include <vespa/searchlib/common/packets.h> + +#include <vespa/log/log.h> +LOG_SETUP(".engine.proto_rpc_adapter"); + +namespace search::engine { + +using vespalib::DataBuffer; +using vespalib::ConstBufferRef; +using vespalib::compression::CompressionConfig; +using ProtoSearchRequest = ProtoConverter::ProtoSearchRequest; +using ProtoSearchReply = ProtoConverter::ProtoSearchReply; +using ProtoDocsumRequest = ProtoConverter::ProtoDocsumRequest; +using ProtoDocsumReply = ProtoConverter::ProtoDocsumReply; +using ProtoMonitorRequest = ProtoConverter::ProtoMonitorRequest; +using ProtoMonitorReply = ProtoConverter::ProtoMonitorReply; + +namespace { + +CompressionConfig get_compression_config() { + using search::fs4transport::FS4PersistentPacketStreamer; + const FS4PersistentPacketStreamer & streamer = FS4PersistentPacketStreamer::Instance; + return CompressionConfig(streamer.getCompressionType(), streamer.getCompressionLevel(), 80, streamer.getCompressionLimit()); +} + +template <typename MSG> +void encode_message(const MSG &src, FRT_Values &dst) { + using vespalib::compression::compress; + auto output = src.SerializeAsString(); + ConstBufferRef buf(output.data(), output.size()); + DataBuffer compressed(output.data(), output.size()); + CompressionConfig::Type type = compress(get_compression_config(), buf, compressed, true); + dst.AddInt8(type); + dst.AddInt32(buf.size()); + dst.AddData(compressed.getData(), compressed.getDataLen()); +} + +template <typename MSG> +bool decode_message(const FRT_Values &src, MSG &dst) { + using vespalib::compression::decompress; + uint8_t encoding = src[0]._intval8; + uint32_t uncompressed_size = src[1]._intval32; + DataBuffer uncompressed(src[2]._data._buf, src[2]._data._len); + ConstBufferRef blob(src[2]._data._buf, src[2]._data._len); + decompress(CompressionConfig::toType(encoding), uncompressed_size, blob, uncompressed, true); + assert(uncompressed_size == uncompressed.getDataLen()); + return dst.ParseFromArray(uncompressed.getData(), uncompressed.getDataLen()); +} + +//----------------------------------------------------------------------------- + +struct SearchRequestDecoder : SearchRequest::Source::Decoder { + FRT_RPCRequest &rpc; // valid until Return is called + RelativeTime relative_time; + SearchRequestDecoder(FRT_RPCRequest &rpc_in) + : rpc(rpc_in), relative_time(std::make_unique<FastosClock>()) {} + std::unique_ptr<SearchRequest> decode() override { + ProtoSearchRequest msg; + if (!decode_message(*rpc.GetParams(), msg)) { + LOG(warning, "got bad protobuf search request over rpc (unable to decode)"); + return std::unique_ptr<SearchRequest>(nullptr); + } + auto req = std::make_unique<SearchRequest>(std::move(relative_time)); + ProtoConverter::search_request_from_proto(msg, *req); + return req; + } +}; + +std::unique_ptr<SearchRequest::Source::Decoder> search_request_decoder(FRT_RPCRequest &rpc) { + return std::make_unique<SearchRequestDecoder>(rpc); +} + +// allocated in the stash of the request it is completing; no self-delete needed +struct SearchCompletionHandler : SearchClient { + FRT_RPCRequest &req; + SearchCompletionHandler(FRT_RPCRequest &req_in) : req(req_in) {} + void searchDone(SearchReply::UP reply) override { + ProtoSearchReply msg; + ProtoConverter::search_reply_to_proto(*reply, msg); + encode_message(msg, *req.GetReturn()); + req.Return(); + } +}; + +//----------------------------------------------------------------------------- + +struct DocsumRequestDecoder : DocsumRequest::Source::Decoder { + FRT_RPCRequest &rpc; // valid until Return is called + RelativeTime relative_time; + DocsumRequestDecoder(FRT_RPCRequest &rpc_in) + : rpc(rpc_in), relative_time(std::make_unique<FastosClock>()) {} + std::unique_ptr<DocsumRequest> decode() override { + ProtoDocsumRequest msg; + if (!decode_message(*rpc.GetParams(), msg)) { + LOG(warning, "got bad protobuf docsum request over rpc (unable to decode)"); + return std::unique_ptr<DocsumRequest>(nullptr); + } + auto req = std::make_unique<DocsumRequest>(std::move(relative_time), true); + ProtoConverter::docsum_request_from_proto(msg, *req); + return req; + } +}; + +std::unique_ptr<DocsumRequest::Source::Decoder> docsum_request_decoder(FRT_RPCRequest &rpc) { + return std::make_unique<DocsumRequestDecoder>(rpc); +} + +// allocated in the stash of the request it is completing; no self-delete needed +struct GetDocsumsCompletionHandler : DocsumClient { + FRT_RPCRequest &req; + GetDocsumsCompletionHandler(FRT_RPCRequest &req_in) : req(req_in) {} + void getDocsumsDone(DocsumReply::UP reply) override { + ProtoDocsumReply msg; + ProtoConverter::docsum_reply_to_proto(*reply, msg); + encode_message(msg, *req.GetReturn()); + req.Return(); + } +}; + +//----------------------------------------------------------------------------- + +// allocated in the stash of the request it is completing; no self-delete needed +struct PingCompletionHandler : MonitorClient { + FRT_RPCRequest &req; + PingCompletionHandler(FRT_RPCRequest &req_in) : req(req_in) {} + void pingDone(MonitorReply::UP reply) override { + ProtoMonitorReply msg; + ProtoConverter::monitor_reply_to_proto(*reply, msg); + encode_message(msg, *req.GetReturn()); + req.Return(); + } +}; + +//----------------------------------------------------------------------------- + +void describe_bix_param_return(FRT_ReflectionBuilder &rb) { + rb.ParamDesc("encoding", "0=raw, 6=lz4, 7=zstd"); + rb.ParamDesc("uncompressed_size", "uncompressed size of serialized request"); + rb.ParamDesc("request", "possibly compressed serialized request"); + rb.ReturnDesc("encoding", "0=raw, 6=lz4, 7=zstd"); + rb.ReturnDesc("uncompressed_size", "uncompressed size of serialized reply"); + rb.ReturnDesc("reply", "possibly compressed serialized reply"); +} + +} + +ProtoRpcAdapter::ProtoRpcAdapter(SearchServer &search_server, + DocsumServer &docsum_server, + MonitorServer &monitor_server, + FRT_Supervisor &orb) + : _search_server(search_server), + _docsum_server(docsum_server), + _monitor_server(monitor_server) +{ + FRT_ReflectionBuilder rb(&orb); + //------------------------------------------------------------------------- + rb.DefineMethod("vespa.searchprotocol.search", "bix", "bix", + FRT_METHOD(ProtoRpcAdapter::rpc_search), this); + rb.MethodDesc("perform a search against this back-end"); + describe_bix_param_return(rb); + //------------------------------------------------------------------------- + rb.DefineMethod("vespa.searchprotocol.getDocsums", "bix", "bix", + FRT_METHOD(ProtoRpcAdapter::rpc_getDocsums), this); + rb.MethodDesc("fetch document summaries from this back-end"); + describe_bix_param_return(rb); + //------------------------------------------------------------------------- + rb.DefineMethod("vespa.searchprotocol.ping", "bix", "bix", + FRT_METHOD(ProtoRpcAdapter::rpc_ping), this); + rb.MethodDesc("ping this back-end"); + describe_bix_param_return(rb); + //------------------------------------------------------------------------- +} + +void +ProtoRpcAdapter::rpc_search(FRT_RPCRequest *req) +{ + req->Detach(); + auto &client = req->getStash().create<SearchCompletionHandler>(*req); + auto reply = _search_server.search(search_request_decoder(*req), client); + if (reply) { + client.searchDone(std::move(reply)); + } +} + +void +ProtoRpcAdapter::rpc_getDocsums(FRT_RPCRequest *req) +{ + req->Detach(); + auto &client = req->getStash().create<GetDocsumsCompletionHandler>(*req); + auto reply = _docsum_server.getDocsums(docsum_request_decoder(*req), client); + if (reply) { + client.getDocsumsDone(std::move(reply)); + } +} + +void +ProtoRpcAdapter::rpc_ping(FRT_RPCRequest *rpc) +{ + rpc->Detach(); + ProtoMonitorRequest msg; + if (decode_message(*rpc->GetParams(), msg)) { + auto req = std::make_unique<MonitorRequest>(); + ProtoConverter::monitor_request_from_proto(msg, *req); + auto &client = rpc->getStash().create<PingCompletionHandler>(*rpc); + auto reply = _monitor_server.ping(std::move(req), client); + if (reply) { + client.pingDone(std::move(reply)); + } + } else { + LOG(warning, "got bad protobuf monitor request over rpc (unable to decode)"); + rpc->SetError(FRTE_RPC_METHOD_FAILED, "malformed monitor request"); + rpc->Return(); + } +} + +//----------------------------------------------------------------------------- + +void +ProtoRpcAdapter::encode_search_request(const ProtoSearchRequest &src, FRT_RPCRequest &dst) +{ + dst.SetMethodName("vespa.searchprotocol.search"); + encode_message(src, *dst.GetParams()); +} + +bool +ProtoRpcAdapter::decode_search_reply(FRT_RPCRequest &src, ProtoSearchReply &dst) +{ + return (src.CheckReturnTypes("bix") && decode_message(*src.GetReturn(), dst)); +} + +void +ProtoRpcAdapter::encode_docsum_request(const ProtoDocsumRequest &src, FRT_RPCRequest &dst) +{ + dst.SetMethodName("vespa.searchprotocol.getDocsums"); + encode_message(src, *dst.GetParams()); +} + +bool +ProtoRpcAdapter::decode_docsum_reply(FRT_RPCRequest &src, ProtoDocsumReply &dst) +{ + return (src.CheckReturnTypes("bix") && decode_message(*src.GetReturn(), dst)); +} + +void +ProtoRpcAdapter::encode_monitor_request(const ProtoMonitorRequest &src, FRT_RPCRequest &dst) +{ + dst.SetMethodName("vespa.searchprotocol.ping"); + encode_message(src, *dst.GetParams()); +} + +bool +ProtoRpcAdapter::decode_monitor_reply(FRT_RPCRequest &src, ProtoMonitorReply &dst) +{ + return (src.CheckReturnTypes("bix") && decode_message(*src.GetReturn(), dst)); +} + +} diff --git a/searchlib/src/vespa/searchlib/engine/proto_rpc_adapter.h b/searchlib/src/vespa/searchlib/engine/proto_rpc_adapter.h new file mode 100644 index 00000000000..cd864f3d5bf --- /dev/null +++ b/searchlib/src/vespa/searchlib/engine/proto_rpc_adapter.h @@ -0,0 +1,56 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#pragma once + +#include <vespa/fnet/frt/invokable.h> +#include "proto_converter.h" + +class FRT_Supervisor; + +namespace search::engine { + +class SearchServer; +class DocsumServer; +class MonitorServer; + +/** + * Class adapting the internal search engine interfaces (SearchServer, + * DocsumServer, MonitorServer) to the external searchprotocol api + * (possibly compressed protobuf over frt rpc). + **/ +class ProtoRpcAdapter : FRT_Invokable +{ +public: + using ProtoSearchRequest = ProtoConverter::ProtoSearchRequest; + using ProtoSearchReply = ProtoConverter::ProtoSearchReply; + using ProtoDocsumRequest = ProtoConverter::ProtoDocsumRequest; + using ProtoDocsumReply = ProtoConverter::ProtoDocsumReply; + using ProtoMonitorRequest = ProtoConverter::ProtoMonitorRequest; + using ProtoMonitorReply = ProtoConverter::ProtoMonitorReply; +private: + SearchServer &_search_server; + DocsumServer &_docsum_server; + MonitorServer &_monitor_server; +public: + ProtoRpcAdapter(SearchServer &search_server, + DocsumServer &docsum_server, + MonitorServer &monitor_server, + FRT_Supervisor &orb); + + void rpc_search(FRT_RPCRequest *req); + void rpc_getDocsums(FRT_RPCRequest *req); + void rpc_ping(FRT_RPCRequest *req); + + // convenience functions used for testing + static void encode_search_request(const ProtoSearchRequest &src, FRT_RPCRequest &dst); + static bool decode_search_reply(FRT_RPCRequest &src, ProtoSearchReply &dst); + + static void encode_docsum_request(const ProtoDocsumRequest &src, FRT_RPCRequest &dst); + static bool decode_docsum_reply(FRT_RPCRequest &src, ProtoDocsumReply &dst); + + static void encode_monitor_request(const ProtoMonitorRequest &src, FRT_RPCRequest &dst); + static bool decode_monitor_reply(FRT_RPCRequest &src, ProtoMonitorReply &dst); + +}; + +} diff --git a/searchlib/src/vespa/searchlib/engine/request.h b/searchlib/src/vespa/searchlib/engine/request.h index ab46b5d40fe..f5f24b6743f 100644 --- a/searchlib/src/vespa/searchlib/engine/request.h +++ b/searchlib/src/vespa/searchlib/engine/request.h @@ -34,6 +34,8 @@ public: _trace.setLevel(level); _trace.start(minLevel); } + Request & setTraceLevel(uint32_t level) { _trace.setLevel(level); return *this; } + uint32_t getTraceLevel() const { return _trace.getLevel(); } Trace & trace() const { return _trace; } private: diff --git a/searchlib/src/vespa/searchlib/engine/search_protocol_proto.h b/searchlib/src/vespa/searchlib/engine/search_protocol_proto.h new file mode 100644 index 00000000000..b13c778fa6f --- /dev/null +++ b/searchlib/src/vespa/searchlib/engine/search_protocol_proto.h @@ -0,0 +1,10 @@ +// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. + +#pragma once + +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wsuggest-override" + +#include "search_protocol.pb.h" + +#pragma GCC diagnostic pop diff --git a/searchlib/src/vespa/searchlib/engine/searchreply.h b/searchlib/src/vespa/searchlib/engine/searchreply.h index 40dca8a1c46..d2645bbb2f6 100644 --- a/searchlib/src/vespa/searchlib/engine/searchreply.h +++ b/searchlib/src/vespa/searchlib/engine/searchreply.h @@ -42,6 +42,8 @@ public: uint32_t getDegradeReason() const { return _degradeReason; } uint16_t getNodesQueried() const { return _nodesQueried; } uint16_t getNodesReplied() const { return _nodesReplied; } + bool wasDegradedByMatchPhase() const { return ((_degradeReason & MATCH_PHASE) != 0); } + bool wasDegradedByTimeout() const { return ((_degradeReason & TIMEOUT) != 0); } Coverage & setCovered(uint64_t v) { _covered = v; return *this; } Coverage & setActive(uint64_t v) { _active = v; return *this; } diff --git a/searchlib/src/vespa/searchlib/engine/searchrequest.cpp b/searchlib/src/vespa/searchlib/engine/searchrequest.cpp index 0e0cbdae429..53467e5c56c 100644 --- a/searchlib/src/vespa/searchlib/engine/searchrequest.cpp +++ b/searchlib/src/vespa/searchlib/engine/searchrequest.cpp @@ -2,6 +2,8 @@ #include "searchrequest.h" #include "packetconverter.h" +#include "proto_converter.h" +#include <vespa/fnet/frt/rpcrequest.h> namespace search::engine { @@ -15,49 +17,9 @@ SearchRequest::SearchRequest(RelativeTime relativeTime) sortSpec(), groupSpec(), sessionId() -{ } - -SearchRequest::~SearchRequest() = default; - - -SearchRequest::Source::Source(SearchRequest * request) - : _request(request), - _fs4Packet(nullptr), - _desc(0), - _relativeTime() -{ } - -SearchRequest::Source::Source(FS4Packet_QUERYX *query, SourceDescription desc) - : _request(), - _fs4Packet(query), - _desc(desc), - _relativeTime(std::make_unique<RelativeTime>(std::make_unique<FastosClock>())) -{ } - -SearchRequest::Source::Source(Source && rhs) noexcept - : _request(std::move(rhs._request)), - _fs4Packet(rhs._fs4Packet), - _desc(std::move(rhs._desc)), - _relativeTime(std::move(rhs._relativeTime)) { - rhs._fs4Packet = nullptr; } -void SearchRequest::Source::lazyDecode() const -{ - if ( ! _request && (_fs4Packet != nullptr)) { - _request = std::make_unique<SearchRequest>(std::move(*_relativeTime)); - PacketConverter::toSearchRequest(*_fs4Packet, *_request); - _fs4Packet->Free(); - _fs4Packet = nullptr; - } -} - -SearchRequest::Source::~Source() { - if (_fs4Packet != nullptr) { - _fs4Packet->Free(); - } -} +SearchRequest::~SearchRequest() = default; } - diff --git a/searchlib/src/vespa/searchlib/engine/searchrequest.h b/searchlib/src/vespa/searchlib/engine/searchrequest.h index b36215a31f2..5af795a3a27 100644 --- a/searchlib/src/vespa/searchlib/engine/searchrequest.h +++ b/searchlib/src/vespa/searchlib/engine/searchrequest.h @@ -4,50 +4,16 @@ #include "propertiesmap.h" #include "request.h" -#include "source_description.h" - -namespace search::fs4transport { class FS4Packet_QUERYX; } +#include "lazy_source.h" namespace search::engine { class SearchRequest : public Request { public: - typedef std::unique_ptr<SearchRequest> UP; - typedef fs4transport::FS4Packet_QUERYX FS4Packet_QUERYX; - - class Source { - private: - mutable std::unique_ptr<SearchRequest> _request; - mutable FS4Packet_QUERYX *_fs4Packet; - void lazyDecode() const; - const SourceDescription _desc; - std::unique_ptr<RelativeTime> _relativeTime; - public: - - Source(SearchRequest * request); - Source(FS4Packet_QUERYX *query, SourceDescription desc); - - Source(Source && rhs) noexcept; - ~Source(); - - const SearchRequest * operator -> () const { return get(); } - - const SearchRequest * get() const { - lazyDecode(); - return _request.get(); - } - - Source& operator= (Source && rhs) = delete; - Source & operator= (const Source &) = delete; - Source(const Source &) = delete; - - UP release() { - lazyDecode(); - return std::move(_request); - } - }; - typedef std::shared_ptr<SearchRequest> SP; + using UP = std::unique_ptr<SearchRequest>; + using SP = std::shared_ptr<SearchRequest>; + using Source = LazySource<SearchRequest>; uint32_t offset; uint32_t maxhits; diff --git a/searchlib/src/vespa/searchlib/engine/source_description.cpp b/searchlib/src/vespa/searchlib/engine/source_description.cpp deleted file mode 100644 index bd285dcdee2..00000000000 --- a/searchlib/src/vespa/searchlib/engine/source_description.cpp +++ /dev/null @@ -1,10 +0,0 @@ -// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. - -#include "source_description.h" - -namespace search::engine { - -const vespalib::string SourceDescription::protocol("FS4"); - -} - diff --git a/searchlib/src/vespa/searchlib/engine/source_description.h b/searchlib/src/vespa/searchlib/engine/source_description.h deleted file mode 100644 index 21274d4b097..00000000000 --- a/searchlib/src/vespa/searchlib/engine/source_description.h +++ /dev/null @@ -1,15 +0,0 @@ -// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. - -#pragma once -#include <vespa/vespalib/stllike/string.h> - -namespace search::engine { - -struct SourceDescription { - int listenPort; - static const vespalib::string protocol; - SourceDescription(int port) : listenPort(port) {} -}; - -} - diff --git a/searchlib/src/vespa/searchlib/engine/transportserver.cpp b/searchlib/src/vespa/searchlib/engine/transportserver.cpp index a8f3bf0b51a..d14735b7770 100644 --- a/searchlib/src/vespa/searchlib/engine/transportserver.cpp +++ b/searchlib/src/vespa/searchlib/engine/transportserver.cpp @@ -14,6 +14,44 @@ LOG_SETUP(".engine.transportserver"); namespace search::engine { +namespace { + +struct SearchRequestDecoder : SearchRequest::Source::Decoder { + PacketConverter::QUERYX *packet; + RelativeTime relative_time; + SearchRequestDecoder(PacketConverter::QUERYX *qx) + : packet(qx), relative_time(std::make_unique<FastosClock>()) {} + std::unique_ptr<SearchRequest> decode() override { + auto req = std::make_unique<SearchRequest>(std::move(relative_time)); + PacketConverter::toSearchRequest(*packet, *req); + return req; + } + ~SearchRequestDecoder() override { packet->Free(); } +}; + +std::unique_ptr<SearchRequest::Source::Decoder> search_request_decoder(PacketConverter::QUERYX *qx) { + return std::make_unique<SearchRequestDecoder>(qx); +} + +struct DocsumRequestDecoder : DocsumRequest::Source::Decoder { + PacketConverter::GETDOCSUMSX *packet; + RelativeTime relative_time; + DocsumRequestDecoder(PacketConverter::GETDOCSUMSX *gdx) + : packet(gdx), relative_time(std::make_unique<FastosClock>()) {} + std::unique_ptr<DocsumRequest> decode() override { + auto req = std::make_unique<DocsumRequest>(std::move(relative_time), false); + PacketConverter::toDocsumRequest(*packet, *req); + return req; + } + ~DocsumRequestDecoder() override { packet->Free(); } +}; + +std::unique_ptr<DocsumRequest::Source::Decoder> docsum_request_decoder(PacketConverter::GETDOCSUMSX *gdx) { + return std::make_unique<DocsumRequestDecoder>(gdx); +} + +} + //----------------------------------------------------------------------------- typedef search::fs4transport::FS4PersistentPacketStreamer PacketStreamer; @@ -191,7 +229,7 @@ TransportServer::HandlePacket(FNET_Packet *packet, FNET_Context context) if (shouldLog(DEBUG_SEARCH)) { logPacket("incoming packet", packet, channel, 0); } - SearchRequest::Source req(qx, _sourceDesc); + SearchRequest::Source req(search_request_decoder(qx)); packet = nullptr; _pending.push(new SearchHandler(*this, std::move(req), channel, _clients.size())); rc = FNET_CLOSE_CHANNEL; @@ -200,7 +238,7 @@ TransportServer::HandlePacket(FNET_Packet *packet, FNET_Context context) if (shouldLog(DEBUG_DOCSUM)) { logPacket("incoming packet", packet, channel, 0); } - DocsumRequest::Source req(gdx, _sourceDesc); + DocsumRequest::Source req(docsum_request_decoder(gdx)); packet = nullptr; _pending.push(new DocsumHandler(*this, std::move(req), channel)); rc = FNET_CLOSE_CHANNEL; @@ -364,7 +402,6 @@ TransportServer::TransportServer(SearchServer &searchServer, _failed(false), _doListen(true), _threadPool(256 * 1024), - _sourceDesc(port), _listenSpec(), _listener(0), _clients(), diff --git a/searchlib/src/vespa/searchlib/engine/transportserver.h b/searchlib/src/vespa/searchlib/engine/transportserver.h index 286b6a3ce01..60bcc79a7fe 100644 --- a/searchlib/src/vespa/searchlib/engine/transportserver.h +++ b/searchlib/src/vespa/searchlib/engine/transportserver.h @@ -3,7 +3,6 @@ #pragma once #include "transport_metrics.h" -#include "source_description.h" #include "searchapi.h" #include "docsumapi.h" #include "monitorapi.h" @@ -67,7 +66,6 @@ private: bool _failed; // flag indicating a critical failure bool _doListen; // flag telling us to accept requests or not FastOS_ThreadPool _threadPool; // thread pool owning transport thread - SourceDescription _sourceDesc; // description of where requests are coming from vespalib::string _listenSpec; // where to listen; FNET connect spec FNET_Connector *_listener; // object accepting incoming connections std::set<FNET_Channel*> _clients; // the admin channel of all client connections |