Merge branch 'feature/http-timeouts-v3' into 'develop'

Add timeouts support for http requests #1545 See merge request itv-backend/reindexer!1408
Restream · Sep 23, 2023 · 426be29 · 426be29
1 parent b0303b6
commit 426be29
Show file tree

Hide file tree

Showing 155 changed files with 4,977 additions and 2,988 deletions.
diff --git a/bindings/builtinserver/config/config.go b/bindings/builtinserver/config/config.go
@@ -14,10 +14,12 @@ type StorageConf struct {
 }
 
 type NetConf struct {
-	HTTPAddr string `yaml:"httpaddr"`
-	RPCAddr  string `yaml:"rpcaddr"`
-	WebRoot  string `yaml:"webroot"`
-	Security bool   `yaml:"security"`
+	HTTPAddr            string `yaml:"httpaddr"`
+	RPCAddr             string `yaml:"rpcaddr"`
+	WebRoot             string `yaml:"webroot"`
+	Security            bool   `yaml:"security"`
+	HttpReadTimeoutSec  int    `yaml:"http_read_timeout,omitempty"`
+	HttpWriteTimeoutSec int    `yaml:"http_write_timeout,omitempty"`
 }
 
 type LoggerConf struct {

diff --git a/cjson/decoder.go b/cjson/decoder.go
@@ -711,7 +711,7 @@ func (dec *Decoder) Decode(cjson []byte, dest interface{}) (err error) {
 		}
 	}()
 
-	fieldsoutcnt := make([]int, 64, 64)
+	fieldsoutcnt := make([]int, MaxIndexes)
 	ctagsPath := make([]int, 0, 8)
 
 	dec.decodeValue(nil, ser, reflect.ValueOf(dest), fieldsoutcnt, ctagsPath)

diff --git a/cpp_src/CMakeLists.txt b/cpp_src/CMakeLists.txt
@@ -89,6 +89,10 @@ endif ()
 
 set (EXTRA_FLAGS "")
 
+if (WITH_ASAN AND WITH_TSAN)
+  message(FATAL_ERROR "You cannot use the ASAN and TSAN options at the same time, CMake will exit.")
+endif()
+
 if (WITH_ASAN)
   set (EXTRA_FLAGS "-fsanitize=address")
   add_definitions(-DREINDEX_WITH_ASAN)
@@ -695,15 +699,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"

diff --git a/cpp_src/client/cororeindexer.cc b/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);

diff --git a/cpp_src/client/cororeindexer.h b/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>

diff --git a/cpp_src/client/cororpcclient.cc b/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());

diff --git a/cpp_src/client/cororpcclient.h b/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());

diff --git a/cpp_src/client/item.h b/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);

diff --git a/cpp_src/client/queryresults.cc b/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
diff --git a/cpp_src/client/queryresults.h b/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; }
 

diff --git a/cpp_src/client/reindexer.cc b/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_); }

diff --git a/cpp_src/client/reindexer.h b/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>

diff --git a/cpp_src/client/rpcclient.cc b/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_);

diff --git a/cpp_src/client/rpcclient.h b/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());

diff --git a/cpp_src/client/rpcclientmock.cc b/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_);