Merge branch '1488_1317_qe_refactoring' into 1488_1317_fuzzing_step_2;

force-build-all

bindings/cproto/cproto.go

@@ -19,7 +19,8 @@ import (
 const (
 	defConnPoolSize        = 8
 	defConnPoolLBAlgorithm = bindings.LBRoundRobin
-	pingerTimeoutSec       = 60
+	pingerTimeoutSec       = uint32(60)
+	pingResponseTimeoutSec = uint32(20)
 	defAppName             = "Go-connector"
 
 	opRd = 0
@@ -586,7 +587,7 @@ func (binding *NetCProto) rpcCallNoResults(ctx context.Context, op int, cmd int,
 func (binding *NetCProto) pinger() {
 	timeout := time.Second
 	ticker := time.NewTicker(timeout)
-	var ticksCount uint16
+	var ticksCount uint32
 	for now := range ticker.C {
 		ticksCount++
 		select {
@@ -597,15 +598,32 @@ func (binding *NetCProto) pinger() {
 		if ticksCount == pingerTimeoutSec {
 			ticksCount = 0
 			conns := binding.getAllConns()
+			var wg sync.WaitGroup
+			cmpl := func(buf bindings.RawBuffer, err error) {
+				wg.Done()
+				if buf != nil {
+					buf.Free()
+				}
+			}
 			for _, conn := range conns {
 				if conn.hasError() {
 					continue
 				}
-				if conn.lastReadTime().Add(timeout).Before(now) {
-					buf, _ := conn.rpcCall(context.TODO(), cmdPing, uint32(binding.timeouts.RequestTimeout*time.Second))
-					buf.Free()
+				if !conn.lastReadTime().Add(timeout).Before(now) {
+					continue
 				}
+				if cap(conn.seqs)-len(conn.seqs) > 0 {
+					continue
+				}
+				timeout := pingResponseTimeoutSec
+				if uint32(binding.timeouts.RequestTimeout) > timeout {
+					timeout = uint32(binding.timeouts.RequestTimeout)
+				}
+
+				wg.Add(1)
+				conn.rpcCallAsync(context.TODO(), cmdPing, uint32(timeout), cmpl)
 			}
+			wg.Wait()
 		}
 	}
 }

cpp_src/CMakeLists.txt

@@ -695,15 +695,16 @@ if (NOT WIN32)
   SET(CMAKE_INSTALL_DEFAULT_COMPONENT_NAME "server")
   SET(DIST_INCLUDE_FILES
     "tools/errors.h" "tools/serializer.h" "tools/varint.h" "tools/stringstools.h" "tools/customhash.h" "tools/assertrx.h" "tools/jsonstring.h"
+    "tools/verifying_updater.h"
     "core/reindexer.h" "core/type_consts.h" "core/item.h" "core/payload/payloadvalue.h" "core/payload/payloadiface.h" "core/indexopts.h"
     "core/namespacedef.h" "core/keyvalue/variant.h" "core/keyvalue/geometry.h" "core/sortingprioritiestable.h"
     "core/rdxcontext.h" "core/activity_context.h" "core/type_consts_helpers.h" "core/payload/fieldsset.h" "core/payload/payloadtype.h"
     "core/cbinding/reindexer_c.h" "core/cbinding/reindexer_ctypes.h" "core/transaction.h" "core/payload/payloadfieldtype.h" "core/reindexerconfig.h"
     "core/query/query.h" "core/query/queryentry.h" "core/queryresults/queryresults.h" "core/indexdef.h" "core/queryresults/aggregationresult.h"
     "core/queryresults/itemref.h" "core/namespace/stringsholder.h" "core/keyvalue/key_string.h" "core/key_value_type.h" "core/keyvalue/uuid.h"
     "core/expressiontree.h" "core/lsn.h" "core/cjson/tagspath.h" "core/cjson/ctag.h"
-	"estl/cow.h" "estl/overloaded.h" "estl/one_of.h" "estl/h_vector.h" "estl/mutex.h" "estl/intrusive_ptr.h" "estl/trivial_reverse_iterator.h"
-	"estl/span.h" "estl/chunk.h" "estl/fast_hash_traits.h" "estl/debug_macros.h" "estl/defines.h"
+    "estl/cow.h" "estl/overloaded.h" "estl/one_of.h" "estl/h_vector.h" "estl/mutex.h" "estl/intrusive_ptr.h" "estl/trivial_reverse_iterator.h"
+    "estl/span.h" "estl/chunk.h" "estl/fast_hash_traits.h" "estl/debug_macros.h" "estl/defines.h"
     "client/reindexer.h" "client/item.h" "client/reindexerconfig.h" "client/queryresults.h" "client/resultserializer.h"
     "client/internalrdxcontext.h" "client/transaction.h"
     "client/cororeindexer.h" "client/coroqueryresults.h" "client/corotransaction.h"

cpp_src/client/cororeindexer.cc

@@ -33,7 +33,7 @@ CoroReindexer& CoroReindexer::operator=(CoroReindexer&& rdx) noexcept {
 Error CoroReindexer::Connect(const std::string& dsn, dynamic_loop& loop, const client::ConnectOpts& opts) {
 	return impl_->Connect(dsn, loop, opts);
 }
-Error CoroReindexer::Stop() { return impl_->Stop(); }
+void CoroReindexer::Stop() { impl_->Stop(); }
 Error CoroReindexer::AddNamespace(const NamespaceDef& nsDef) { return impl_->AddNamespace(nsDef, ctx_); }
 Error CoroReindexer::OpenNamespace(std::string_view nsName, const StorageOpts& storage) {
 	return impl_->OpenNamespace(nsName, ctx_, storage);

cpp_src/client/cororeindexer.h

@@ -46,7 +46,7 @@ class CoroReindexer {
 	/// @param opts - Connect options. May contaion any of <br>
 	Error Connect(const std::string &dsn, dynamic_loop &loop, const client::ConnectOpts &opts = client::ConnectOpts());
 	/// Stop - shutdown connector
-	Error Stop();
+	void Stop();
 	/// Open or create namespace
 	/// @param nsName - Name of namespace
 	/// @param opts - Storage options. Can be one of <br>

cpp_src/client/cororpcclient.cc

@@ -46,13 +46,12 @@ Error CoroRPCClient::Connect(const std::string& dsn, ev::dynamic_loop& loop, con
 	return errOK;
 }
 
-Error CoroRPCClient::Stop() {
+void CoroRPCClient::Stop() {
 	terminate_ = true;
 	conn_.Stop();
 	resubWg_.wait();
 	loop_ = nullptr;
 	terminate_ = false;
-	return errOK;
 }
 
 Error CoroRPCClient::AddNamespace(const NamespaceDef& nsDef, const InternalRdxContext& ctx) {
@@ -236,7 +235,7 @@ Error CoroRPCClient::Delete(const Query& query, CoroQueryResults& result, const
 	query.Serialize(ser);
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q._namespace)); });
+	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = CoroQueryResults(&conn_, std::move(nsArray), 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -257,7 +256,7 @@ Error CoroRPCClient::Update(const Query& query, CoroQueryResults& result, const
 	query.Serialize(ser);
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q._namespace)); });
+	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = CoroQueryResults(&conn_, std::move(nsArray), 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -322,7 +321,7 @@ Error CoroRPCClient::selectImpl(const Query& query, CoroQueryResults& result, se
 	}
 	NsArray nsArray;
 	query.Serialize(qser);
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q._namespace)); });
+	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 	h_vector<int32_t, 4> vers;
 	for (auto& ns : nsArray) {
 		vers.push_back(ns->tagsMatcher_.version() ^ ns->tagsMatcher_.stateToken());

cpp_src/client/cororpcclient.h

@@ -36,7 +36,7 @@ class CoroRPCClient {
 	~CoroRPCClient();
 
 	Error Connect(const std::string &dsn, ev::dynamic_loop &loop, const client::ConnectOpts &opts);
-	Error Stop();
+	void Stop();
 
 	Error OpenNamespace(std::string_view nsName, const InternalRdxContext &ctx,
 						const StorageOpts &opts = StorageOpts().Enabled().CreateIfMissing());

cpp_src/client/item.h

@@ -60,13 +60,13 @@ class Item {
 	/// Get status of item
 	/// @return data slice with JSON. Returned slice is allocated in temporary Item's buffer, and can be invalidated by any next operation
 	/// with Item
-	Error Status() { return status_; }
+	Error Status() const noexcept { return status_; }
 	/// Get internal ID of item
 	/// @return ID of item
 	int GetID() const noexcept { return id_; }
-	/// Get internal version of item
-	/// @return version of item
-	int NumFields();
+	/// Get count of indexed fields
+	/// @return count of indexed fields
+	int NumFields() const noexcept;
 	/// Set additional percepts for modify operation
 	/// @param precepts - strings in format "fieldName=Func()"
 	void SetPrecepts(const std::vector<std::string> &precepts);

cpp_src/client/queryresults.cc

@@ -114,8 +114,6 @@ void QueryResults::fetchNextResults() {
 	rawResult_.assign(rawResult.begin() + ser.Pos(), rawResult.end());
 }
 
-QueryResults::~QueryResults() {}
-
 h_vector<std::string_view, 1> QueryResults::GetNamespaces() const {
 	h_vector<std::string_view, 1> ret;
 	ret.reserve(nsArray_.size());
@@ -310,8 +308,5 @@ QueryResults::Iterator &QueryResults::Iterator::operator++() {
 	return *this;
 }
 
-bool QueryResults::Iterator::operator!=(const Iterator &other) const { return idx_ != other.idx_; }
-bool QueryResults::Iterator::operator==(const Iterator &other) const { return idx_ == other.idx_; }
-
 }  // namespace client
 }  // namespace reindexer

cpp_src/client/queryresults.h

@@ -26,7 +26,7 @@ class QueryResults {
 	QueryResults(int fetchFlags = 0);
 	QueryResults(const QueryResults&) = delete;
 	QueryResults(QueryResults&&) noexcept;
-	~QueryResults();
+	~QueryResults() = default;
 	QueryResults& operator=(const QueryResults&) = delete;
 	QueryResults& operator=(QueryResults&& obj) noexcept;
 
@@ -40,10 +40,10 @@ class QueryResults {
 		bool IsRaw();
 		std::string_view GetRaw();
 		Iterator& operator++();
-		Error Status() { return qr_->status_; }
-		bool operator!=(const Iterator&) const;
-		bool operator==(const Iterator&) const;
-		Iterator& operator*() { return *this; }
+		Error Status() const noexcept { return qr_->status_; }
+		bool operator==(const Iterator& other) const noexcept { return idx_ == other.idx_; }
+		bool operator!=(const Iterator& other) const noexcept { return !operator==(other); }
+		Iterator& operator*() noexcept { return *this; }
 		void readNext();
 		void getJSONFromCJSON(std::string_view cjson, WrSerializer& wrser, bool withHdrLen = true);
 
@@ -55,13 +55,15 @@ class QueryResults {
 	Iterator begin() const { return Iterator{this, 0, 0, 0, {}}; }
 	Iterator end() const { return Iterator{this, queryParams_.qcount, 0, 0, {}}; }
 
-	size_t Count() const { return queryParams_.qcount; }
-	int TotalCount() const { return queryParams_.totalcount; }
-	bool HaveRank() const { return queryParams_.flags & kResultsWithRank; }
-	bool NeedOutputRank() const { return queryParams_.flags & kResultsNeedOutputRank; }
-	const std::string& GetExplainResults() const { return queryParams_.explainResults; }
-	const std::vector<AggregationResult>& GetAggregationResults() const { return queryParams_.aggResults; }
-	Error Status() { return status_; }
+	size_t Count() const noexcept { return queryParams_.qcount; }
+	int TotalCount() const noexcept { return queryParams_.totalcount; }
+	bool HaveRank() const noexcept { return queryParams_.flags & kResultsWithRank; }
+	bool NeedOutputRank() const noexcept { return queryParams_.flags & kResultsNeedOutputRank; }
+	const std::string& GetExplainResults() const& noexcept { return queryParams_.explainResults; }
+	const std::string& GetExplainResults() const&& = delete;
+	const std::vector<AggregationResult>& GetAggregationResults() const& noexcept { return queryParams_.aggResults; }
+	const std::vector<AggregationResult>& GetAggregationResults() const&& = delete;
+	Error Status() const noexcept { return status_; }
 	h_vector<std::string_view, 1> GetNamespaces() const;
 	bool IsCacheEnabled() const { return queryParams_.flags & kResultsWithItemID; }
 

cpp_src/client/reindexer.cc

@@ -29,7 +29,7 @@ Error Reindexer::Connect(const std::string& dsn, const client::ConnectOpts& opts
 Error Reindexer::Connect(const std::vector<std::pair<std::string, client::ConnectOpts>>& connectData) {
 	return impl_->Connect(connectData);
 }
-Error Reindexer::Stop() { return impl_->Stop(); }
+void Reindexer::Stop() { impl_->Stop(); }
 Error Reindexer::AddNamespace(const NamespaceDef& nsDef) { return impl_->AddNamespace(nsDef, ctx_); }
 Error Reindexer::OpenNamespace(std::string_view nsName, const StorageOpts& storage) { return impl_->OpenNamespace(nsName, ctx_, storage); }
 Error Reindexer::DropNamespace(std::string_view nsName) { return impl_->DropNamespace(nsName, ctx_); }

cpp_src/client/reindexer.h

@@ -48,7 +48,7 @@ class Reindexer {
 	/// @param connectData - list of server dsn + it's ConnectOpts
 	Error Connect(const std::vector<std::pair<std::string, client::ConnectOpts>> &connectData);
 	/// Stop - shutdown connector
-	Error Stop();
+	void Stop();
 	/// Open or create namespace
 	/// @param nsName - Name of namespace
 	/// @param opts - Storage options. Can be one of <br>

cpp_src/client/rpcclient.cc

@@ -76,16 +76,15 @@ Error RPCClient::Connect(const std::vector<std::pair<std::string, client::Connec
 	return startWorkers();
 }
 
-Error RPCClient::Stop() {
-	if (!connections_.size()) return errOK;
+void RPCClient::Stop() {
+	if (!connections_.size()) return;
 	for (auto& worker : workers_) {
 		worker.stop_.send();
 		if (worker.thread_.joinable()) {
 			worker.thread_.join();
 		}
 	}
 	connections_.clear();
-	return errOK;
 }
 
 void RPCClient::run(size_t thIdx) {
@@ -383,7 +382,7 @@ Error RPCClient::Delete(const Query& query, QueryResults& result, const Internal
 	auto conn = getConn();
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q._namespace)); });
+	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = QueryResults(conn, std::move(nsArray), nullptr, 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -410,7 +409,7 @@ Error RPCClient::Update(const Query& query, QueryResults& result, const Internal
 	auto conn = getConn();
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q._namespace)); });
+	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = QueryResults(conn, std::move(nsArray), nullptr, 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -486,7 +485,7 @@ Error RPCClient::selectImpl(const Query& query, QueryResults& result, cproto::Cl
 	}
 	NsArray nsArray;
 	query.Serialize(qser);
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q._namespace)); });
+	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 	h_vector<int32_t, 4> vers;
 	for (auto& ns : nsArray) {
 		shared_lock<shared_timed_mutex> lck(ns->lck_);

cpp_src/client/rpcclient.h

@@ -41,7 +41,7 @@ class RPCClient {
 
 	Error Connect(const std::string &dsn, const client::ConnectOpts &opts);
 	Error Connect(const std::vector<std::pair<std::string, client::ConnectOpts>> &connectData);
-	Error Stop();
+	void Stop();
 
 	Error OpenNamespace(std::string_view nsName, const InternalRdxContext &ctx,
 						const StorageOpts &opts = StorageOpts().Enabled().CreateIfMissing());

cpp_src/client/rpcclientmock.cc

@@ -37,7 +37,7 @@ Error RPCClientMock::Delete(const Query& query, QueryResults& result, const Inte
 	auto conn = getConn();
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q._namespace)); });
+	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = QueryResults(conn, std::move(nsArray), nullptr, 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -69,7 +69,7 @@ Error RPCClientMock::Update(const Query& query, QueryResults& result, const Inte
 	auto conn = getConn();
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q._namespace)); });
+	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = QueryResults(conn, std::move(nsArray), nullptr, 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -308,7 +308,7 @@ Error RPCClientMock::selectImpl(const Query& query, QueryResults& result, cproto
 
 	NsArray nsArray;
 	query.Serialize(qser);
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q._namespace)); });
+	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 	h_vector<int32_t, 4> vers;
 	for (auto& ns : nsArray) {
 		shared_lock<shared_timed_mutex> lck(ns->lck_);

cpp_src/cmd/reindexer_server/main.cc

@@ -7,7 +7,12 @@
 int main(int argc, char* argv[]) {
 	reindexer::debug::backtrace_init();
 
-	reindexer::CheckRequiredSSESupport();
+	try {
+		reindexer::CheckRequiredSSESupport();
+	} catch (Error& err) {
+		std::cerr << err.what();
+		return EXIT_FAILURE;
+	}
 
 	reindexer_server::Server svc(reindexer_server::ServerMode::Standalone);
 	auto err = svc.InitFromCLI(argc, argv);