Merge branch 'T0879-MG-transport-prototype' of github.com:memgraph/memgraph into T0912-MG-in-memory-shard-map

src/io/address.hpp

@@ -36,14 +36,20 @@ struct Address {
   }
 
   bool operator==(const Address &other) const {
-    return (last_known_ip == other.last_known_ip) && (last_known_port == other.last_known_port);
+    return ((unique_id == other.unique_id) && last_known_ip == other.last_known_ip) &&
+           (last_known_port == other.last_known_port);
   }
 
+  /// unique_id is most dominant for ordering, then last_known_ip, then last_known_port
   bool operator<(const Address &other) const {
-    if (last_known_ip == other.last_known_ip) {
-      return last_known_port < other.last_known_port;
+    if (unique_id == other.unique_id) {
+      if (last_known_ip == other.last_known_ip) {
+        return last_known_port < other.last_known_port;
+      } else {
+        return last_known_ip < other.last_known_ip;
+      }
     } else {
-      return last_known_ip < other.last_known_ip;
+      return unique_id < other.unique_id;
     }
   }
 };

src/io/future.hpp

@@ -18,9 +18,8 @@
 #include <thread>
 #include <utility>
 
-#include "utils/logging.hpp"
-
 #include "io/errors.hpp"
+#include "utils/logging.hpp"
 
 namespace memgraph::io {
 
@@ -36,7 +35,7 @@ class Shared {
   std::optional<T> item_;
   bool consumed_ = false;
   bool waiting_ = false;
-  std::optional<std::function<bool()>> simulator_notifier_;
+  std::function<bool()> simulator_notifier_ = nullptr;
 
  public:
   explicit Shared(std::function<bool()> simulator_notifier) : simulator_notifier_(simulator_notifier) {}
@@ -47,13 +46,27 @@ class Shared {
   Shared &operator=(const Shared &) = delete;
   ~Shared() = default;
 
+  /// Takes the item out of our optional item_ and returns it.
+  /// Requires caller holds mutex, proving it by passing reference.
+  T Take(std::unique_lock<std::mutex> &) {
+    MG_ASSERT(item_, "Take called without item_ being present");
+    MG_ASSERT(!consumed_, "Take called on already-consumed Future");
+
+    T ret = std::move(item_).value();
+    item_.reset();
+
+    consumed_ = true;
+
+    return ret;
+  }
+
   T Wait() {
     std::unique_lock<std::mutex> lock(mu_);
     waiting_ = true;
 
     while (!item_) {
       bool simulator_progressed = false;
-      if (simulator_notifier_) {
+      if (simulator_notifier_) [[unlikely]] {
         // We can't hold our own lock while notifying
         // the simulator because notifying the simulator
         // involves acquiring the simulator's mutex
@@ -65,7 +78,7 @@ class Shared {
         // so we have to get out of its way to avoid
         // a cyclical deadlock.
         lock.unlock();
-        simulator_progressed = (*simulator_notifier_)();
+        simulator_progressed = (simulator_notifier_)();
         lock.lock();
         if (item_) {
           // item may have been filled while we
@@ -80,13 +93,9 @@ class Shared {
       MG_ASSERT(!consumed_, "Future consumed twice!");
     }
 
-    T ret = std::move(item_).value();
-    item_.reset();
-
     waiting_ = false;
-    consumed_ = true;
 
-    return ret;
+    return Take(lock);
   }
 
   bool IsReady() {
@@ -98,13 +107,7 @@ class Shared {
     std::unique_lock<std::mutex> lock(mu_);
 
     if (item_) {
-      T ret = std::move(item_).value();
-      item_.reset();
-
-      waiting_ = false;
-      consumed_ = true;
-
-      return ret;
+      return Take(lock);
     } else {
       return std::nullopt;
     }
@@ -140,16 +143,18 @@ class Future {
 
   Future() = delete;
   Future(Future &&old) {
+    MG_ASSERT(!old.consumed_or_moved_, "Future moved from after already being moved from or consumed.");
     shared_ = std::move(old.shared_);
     consumed_or_moved_ = old.consumed_or_moved_;
-    MG_ASSERT(!old.consumed_or_moved_, "Future moved from after already being moved from or consumed.");
     old.consumed_or_moved_ = true;
   }
+
   Future &operator=(Future &&old) {
-    shared_ = std::move(old.shared_);
     MG_ASSERT(!old.consumed_or_moved_, "Future moved from after already being moved from or consumed.");
+    shared_ = std::move(old.shared_);
     old.consumed_or_moved_ = true;
   }
+
   Future(const Future &) = delete;
   Future &operator=(const Future &) = delete;
   ~Future() = default;
@@ -177,7 +182,7 @@ class Future {
 
   /// Block on the corresponding promise to be filled,
   /// returning the inner item when ready.
-  T Wait() {
+  T Wait() && {
     MG_ASSERT(!consumed_or_moved_, "Future should only be consumed with Wait once!");
     T ret = shared_->Wait();
     consumed_or_moved_ = true;
@@ -201,13 +206,14 @@ class Promise {
 
   Promise() = delete;
   Promise(Promise &&old) {
-    shared_ = std::move(old.shared_);
     MG_ASSERT(!old.filled_or_moved_, "Promise moved from after already being moved from or filled.");
+    shared_ = std::move(old.shared_);
     old.filled_or_moved_ = true;
   }
+
   Promise &operator=(Promise &&old) {
-    shared_ = std::move(old.shared_);
     MG_ASSERT(!old.filled_or_moved_, "Promise moved from after already being moved from or filled.");
+    shared_ = std::move(old.shared_);
     old.filled_or_moved_ = true;
   }
   Promise(const Promise &) = delete;

src/io/simulator/simulator.hpp

@@ -12,6 +12,7 @@
 #pragma once
 
 #include <memory>
+#include <random>
 
 #include "io/address.hpp"
 #include "io/simulator/simulator_config.hpp"
@@ -20,7 +21,7 @@
 
 namespace memgraph::io::simulator {
 class Simulator {
-  std::mt19937 rng_{};
+  std::mt19937 rng_;
   std::shared_ptr<SimulatorHandle> simulator_handle_;
 
  public:
@@ -30,7 +31,7 @@ class Simulator {
   void ShutDown() { simulator_handle_->ShutDown(); }
 
   Io<SimulatorTransport> Register(Address address) {
-    std::uniform_int_distribution<uint64_t> seed_distrib{};
+    std::uniform_int_distribution<uint64_t> seed_distrib;
     uint64_t seed = seed_distrib(rng_);
     return Io(SimulatorTransport(simulator_handle_, address, seed), address);
   }

src/io/simulator/simulator_config.hpp

@@ -11,13 +11,20 @@
 
 #pragma once
 
+#include <chrono>
+
+#include "io/time.hpp"
+
 namespace memgraph::io::simulator {
+
+using memgraph::io::Time;
+
 struct SimulatorConfig {
-  int drop_percent = 0;
+  uint8_t drop_percent = 0;
   bool perform_timeouts = false;
   bool scramble_messages = true;
   uint64_t rng_seed = 0;
-  uint64_t start_time = 0;
-  uint64_t abort_time = ULLONG_MAX;
+  Time start_time = Time::min();
+  Time abort_time = Time::max();
 };
 };  // namespace memgraph::io::simulator

src/io/simulator/simulator_handle.hpp

@@ -27,9 +27,14 @@
 #include "io/errors.hpp"
 #include "io/simulator/simulator_config.hpp"
 #include "io/simulator/simulator_stats.hpp"
+#include "io/time.hpp"
 #include "io/transport.hpp"
 
 namespace memgraph::io::simulator {
+
+using memgraph::io::Duration;
+using memgraph::io::Time;
+
 struct OpaqueMessage {
   Address from_address;
   uint64_t request_id;
@@ -195,7 +200,7 @@ class OpaquePromise {
 };
 
 struct DeadlineAndOpaquePromise {
-  uint64_t deadline;
+  Time deadline;
   OpaquePromise promise;
 };
 
@@ -206,13 +211,13 @@ class SimulatorHandle {
   // messages that have not yet been scheduled or dropped
   std::vector<std::pair<Address, OpaqueMessage>> in_flight_;
 
-  // the responsese to requests that are being waited on
+  // the responses to requests that are being waited on
   std::map<PromiseKey, DeadlineAndOpaquePromise> promises_;
 
   // messages that are sent to servers that may later receive them
   std::map<Address, std::vector<OpaqueMessage>> can_receive_;
 
-  uint64_t cluster_wide_time_microseconds_;
+  Time cluster_wide_time_microseconds_;
   bool should_shut_down_ = false;
   SimulatorStats stats_;
   size_t blocked_on_receive_ = 0;
@@ -250,7 +255,7 @@ class SimulatorHandle {
   }
 
   void TimeoutPromisesPastDeadline() {
-    uint64_t now = cluster_wide_time_microseconds_;
+    const Time now = cluster_wide_time_microseconds_;
 
     for (auto &[promise_key, dop] : promises_) {
       // TODO(tyler) queue this up and drop it after its deadline
@@ -298,7 +303,7 @@ class SimulatorHandle {
       // We tick the clock forward when all servers are blocked but
       // there are no in-flight messages to schedule delivery of.
       std::poisson_distribution<> time_distrib(50);
-      uint64_t clock_advance = time_distrib(rng_);
+      Duration clock_advance = std::chrono::microseconds{time_distrib(rng_)};
       cluster_wide_time_microseconds_ += clock_advance;
 
       MG_ASSERT(cluster_wide_time_microseconds_ < config_.abort_time,
@@ -334,7 +339,7 @@ class SimulatorHandle {
       DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key));
       promises_.erase(promise_key);
 
-      bool normal_timeout = config_.perform_timeouts && (dop.deadline < cluster_wide_time_microseconds_);
+      const bool normal_timeout = config_.perform_timeouts && (dop.deadline < cluster_wide_time_microseconds_);
 
       if (should_drop || normal_timeout) {
         stats_.timed_out_requests++;
@@ -366,11 +371,11 @@ class SimulatorHandle {
   }
 
   template <Message Request, Message Response>
-  void SubmitRequest(Address to_address, Address from_address, uint64_t request_id, Request &&request,
-                     uint64_t timeout_microseconds, ResponsePromise<Response> &&promise) {
+  void SubmitRequest(Address to_address, Address from_address, uint64_t request_id, Request &&request, Duration timeout,
+                     ResponsePromise<Response> &&promise) {
     std::unique_lock<std::mutex> lock(mu_);
 
-    uint64_t deadline = cluster_wide_time_microseconds_ + timeout_microseconds;
+    const Time deadline = cluster_wide_time_microseconds_ + timeout;
 
     std::any message(std::move(request));
     OpaqueMessage om{.from_address = from_address, .request_id = request_id, .message = std::move(message)};
@@ -390,10 +395,12 @@ class SimulatorHandle {
   }
 
   template <Message... Ms>
-  requires(sizeof...(Ms) > 0) RequestResult<Ms...> Receive(const Address &receiver, uint64_t timeout_microseconds) {
+  requires(sizeof...(Ms) > 0) RequestResult<Ms...> Receive(const Address &receiver, Duration timeout) {
     std::unique_lock<std::mutex> lock(mu_);
 
-    uint64_t deadline = cluster_wide_time_microseconds_ + timeout_microseconds;
+    blocked_on_receive_ += 1;
+
+    const Time deadline = cluster_wide_time_microseconds_ + timeout;
 
     while (!should_shut_down_ && (cluster_wide_time_microseconds_ < deadline)) {
       if (can_receive_.contains(receiver)) {
@@ -405,20 +412,23 @@ class SimulatorHandle {
           // TODO(tyler) search for item in can_receive_ that matches the desired types, rather
           // than asserting that the last item in can_rx matches.
           auto m_opt = message.Take<Ms...>();
+
+          blocked_on_receive_ -= 1;
+
           return std::move(m_opt).value();
         }
       }
 
-      blocked_on_receive_ += 1;
       lock.unlock();
       bool made_progress = MaybeTickSimulator();
       lock.lock();
       if (!should_shut_down_ && !made_progress) {
         cv_.wait(lock);
       }
-      blocked_on_receive_ -= 1;
     }
 
+    blocked_on_receive_ -= 1;
+
     return TimedOut{};
   }
 
@@ -434,7 +444,7 @@ class SimulatorHandle {
     cv_.notify_all();
   }
 
-  uint64_t Now() {
+  Time Now() {
     std::unique_lock<std::mutex> lock(mu_);
     return cluster_wide_time_microseconds_;
   }

src/io/simulator/simulator_transport.hpp

@@ -16,8 +16,13 @@
 
 #include "io/address.hpp"
 #include "io/simulator/simulator_handle.hpp"
+#include "io/time.hpp"
 
 namespace memgraph::io::simulator {
+
+using memgraph::io::Duration;
+using memgraph::io::Time;
+
 class SimulatorTransport {
   std::shared_ptr<SimulatorHandle> simulator_handle_;
   Address address_;
@@ -28,21 +33,19 @@ class SimulatorTransport {
       : simulator_handle_(simulator_handle), address_(address), rng_(std::mt19937{seed}) {}
 
   template <Message Request, Message Response>
-  ResponseFuture<Response> Request(Address address, uint64_t request_id, Request request,
-                                   uint64_t timeout_microseconds) {
-    std::function<bool()> maybe_tick_simulator = [=] { return simulator_handle_->MaybeTickSimulator(); };
+  ResponseFuture<Response> Request(Address address, uint64_t request_id, Request request, Duration timeout) {
+    std::function<bool()> maybe_tick_simulator = [this] { return simulator_handle_->MaybeTickSimulator(); };
     auto [future, promise] =
         memgraph::io::FuturePromisePairWithNotifier<ResponseResult<Response>>(maybe_tick_simulator);
 
-    simulator_handle_->SubmitRequest(address, address_, request_id, std::move(request), timeout_microseconds,
-                                     std::move(promise));
+    simulator_handle_->SubmitRequest(address, address_, request_id, std::move(request), timeout, std::move(promise));
 
     return std::move(future);
   }
 
   template <Message... Ms>
-  requires(sizeof...(Ms) > 0) RequestResult<Ms...> Receive(uint64_t timeout_microseconds) {
-    return simulator_handle_->template Receive<Ms...>(address_, timeout_microseconds);
+  requires(sizeof...(Ms) > 0) RequestResult<Ms...> Receive(Duration timeout) {
+    return simulator_handle_->template Receive<Ms...>(address_, timeout);
   }
 
   template <Message M>
@@ -50,7 +53,7 @@ class SimulatorTransport {
     return simulator_handle_->template Send<M>(address, address_, request_id, message);
   }
 
-  uint64_t Now() { return simulator_handle_->Now(); }
+  Time Now() { return simulator_handle_->Now(); }
 
   bool ShouldShutDown() { return simulator_handle_->ShouldShutDown(); }
 

src/io/time.hpp

@@ -0,0 +1,21 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include <chrono>
+
+namespace memgraph::io {
+
+using Duration = std::chrono::duration<int64_t, std::ratio<1, 1000000>>;
+using Time = std::chrono::time_point<std::chrono::local_t, Duration>;
+
+}  // namespace memgraph::io

src/io/transport.hpp

@@ -11,15 +11,16 @@
 
 #pragma once
 
+#include <chrono>
 #include <concepts>
 #include <random>
 #include <variant>
 
-#include "utils/result.hpp"
-
 #include "io/address.hpp"
 #include "io/errors.hpp"
 #include "io/future.hpp"
+#include "io/time.hpp"
+#include "utils/result.hpp"
 
 using memgraph::utils::BasicResult;
 
@@ -61,46 +62,43 @@ class Io {
   I implementation_;
   Address address_;
   uint64_t request_id_counter_ = 0;
-  uint64_t default_timeout_microseconds_ = 50 * 1000;
+  Duration default_timeout_ = std::chrono::microseconds{50000};
 
  public:
   Io(I io, Address address) : implementation_(io), address_(address) {}
 
   /// Set the default timeout for all requests that are issued
   /// without an explicit timeout set.
-  void SetDefaultTimeoutMicroseconds(uint64_t timeout_microseconds) {
-    default_timeout_microseconds_ = timeout_microseconds;
-  }
+  void SetDefaultTimeout(Duration timeout) { default_timeout_ = timeout; }
 
   /// Issue a request with an explicit timeout in microseconds provided.
   template <Message Request, Message Response>
-  ResponseFuture<Response> RequestWithTimeout(Address address, Request request, uint64_t timeout_microseconds) {
+  ResponseFuture<Response> RequestWithTimeout(Address address, Request request, Duration timeout) {
     uint64_t request_id = ++request_id_counter_;
-    return implementation_.template Request<Request, Response>(address, request_id, request, timeout_microseconds);
+    return implementation_.template Request<Request, Response>(address, request_id, request, timeout);
   }
 
   /// Issue a request that times out after the default timeout.
   template <Message Request, Message Response>
   ResponseFuture<Response> Request(Address address, Request request) {
     uint64_t request_id = ++request_id_counter_;
-    uint64_t timeout_microseconds = default_timeout_microseconds_;
-    return implementation_.template Request<Request, Response>(address, request_id, std::move(request),
-                                                               timeout_microseconds);
+    Duration timeout = default_timeout_;
+    return implementation_.template Request<Request, Response>(address, request_id, std::move(request), timeout);
   }
 
   /// Wait for an explicit number of microseconds for a request of one of the
   /// provided types to arrive.
   template <Message... Ms>
-  RequestResult<Ms...> ReceiveWithTimeout(uint64_t timeout_microseconds) {
-    return implementation_.template Receive<Ms...>(timeout_microseconds);
+  RequestResult<Ms...> ReceiveWithTimeout(Duration timeout) {
+    return implementation_.template Receive<Ms...>(timeout);
   }
 
   /// Wait the default number of microseconds for a request of one of the
   /// provided types to arrive.
   template <Message... Ms>
   requires(sizeof...(Ms) > 0) RequestResult<Ms...> Receive() {
-    uint64_t timeout_microseconds = default_timeout_microseconds_;
-    return implementation_.template Receive<Ms...>(timeout_microseconds);
+    Duration timeout = default_timeout_;
+    return implementation_.template Receive<Ms...>(timeout);
   }
 
   /// Send a message in a best-effort fashion. If you need reliable delivery,
@@ -114,7 +112,7 @@ class Io {
   /// This time source should be preferred over any other, because it
   /// lets us deterministically control clocks from tests for making
   /// things like timeouts deterministic.
-  uint64_t Now() { return implementation_.Now(); }
+  Time Now() { return implementation_.Now(); }
 
   /// Returns true of the system should shut-down.
   bool ShouldShutDown() { return implementation_.ShouldShutDown(); }

src/query/v2/CMakeLists.txt

@@ -84,7 +84,7 @@ add_custom_command(
     OUTPUT ${antlr_opencypher_generated_src} ${antlr_opencypher_generated_include}
     COMMAND ${CMAKE_COMMAND} -E make_directory ${opencypher_generated}
     COMMAND
-    java -jar ${CMAKE_SOURCE_DIR}/libs/antlr-4.9.2-complete.jar
+    java -jar ${CMAKE_SOURCE_DIR}/libs/antlr-4.10.1-complete.jar
     -Dlanguage=Cpp -visitor -package antlropencypher
     -o ${opencypher_generated}
     ${opencypher_lexer_grammar} ${opencypher_parser_grammar}

src/query/v2/cypher_query_interpreter.cpp

@@ -21,8 +21,7 @@ namespace memgraph::query::v2 {
 CachedPlan::CachedPlan(std::unique_ptr<LogicalPlan> plan) : plan_(std::move(plan)) {}
 
 ParsedQuery ParseQuery(const std::string &query_string, const std::map<std::string, storage::v3::PropertyValue> &params,
-                       utils::SkipList<QueryCacheEntry> *cache, utils::SpinLock *antlr_lock,
-                       const InterpreterConfig::Query &query_config) {
+                       utils::SkipList<QueryCacheEntry> *cache, const InterpreterConfig::Query &query_config) {
   // Strip the query for caching purposes. The process of stripping a query
   // "normalizes" it by replacing any literals with new parameters. This
   // results in just the *structure* of the query being taken into account for
@@ -63,20 +62,16 @@ ParsedQuery ParseQuery(const std::string &query_string, const std::map<std::stri
   };
 
   if (it == accessor.end()) {
-    {
-      std::unique_lock<utils::SpinLock> guard(*antlr_lock);
+    try {
+      parser = std::make_unique<frontend::opencypher::Parser>(stripped_query.query());
+    } catch (const SyntaxException &e) {
+      // There is a syntax exception in the stripped query. Re-run the parser
+      // on the original query to get an appropriate error messsage.
+      parser = std::make_unique<frontend::opencypher::Parser>(query_string);
 
-      try {
-        parser = std::make_unique<frontend::opencypher::Parser>(stripped_query.query());
-      } catch (const SyntaxException &e) {
-        // There is a syntax exception in the stripped query. Re-run the parser
-        // on the original query to get an appropriate error messsage.
-        parser = std::make_unique<frontend::opencypher::Parser>(query_string);
-
-        // If an exception was not thrown here, the stripper messed something
-        // up.
-        LOG_FATAL("The stripped query can't be parsed, but the original can.");
-      }
+      // If an exception was not thrown here, the stripper messed something
+      // up.
+      LOG_FATAL("The stripped query can't be parsed, but the original can.");
     }
 
     // Convert the ANTLR4 parse tree into an AST.

src/query/v2/cypher_query_interpreter.hpp

@@ -111,8 +111,7 @@ struct ParsedQuery {
 };
 
 ParsedQuery ParseQuery(const std::string &query_string, const std::map<std::string, storage::v3::PropertyValue> &params,
-                       utils::SkipList<QueryCacheEntry> *cache, utils::SpinLock *antlr_lock,
-                       const InterpreterConfig::Query &query_config);
+                       utils::SkipList<QueryCacheEntry> *cache, const InterpreterConfig::Query &query_config);
 
 class SingleNodeLogicalPlan final : public LogicalPlan {
  public:

src/query/v2/frontend/ast/cypher_main_visitor.hpp

@@ -115,7 +115,7 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor {
     auto operators = ExtractOperators(all_children, allowed_operators);
 
     for (auto *expression : _expressions) {
-      expressions.push_back(expression->accept(this));
+      expressions.push_back(std::any_cast<Expression *>(expression->accept(this)));
     }
 
     Expression *first_operand = expressions[0];
@@ -131,7 +131,7 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor {
     DMG_ASSERT(_expression, "can't happen");
     auto operators = ExtractOperators(all_children, allowed_operators);
 
-    Expression *expression = _expression->accept(this);
+    Expression *expression = std::any_cast<Expression *>(_expression->accept(this));
     for (int i = (int)operators.size() - 1; i >= 0; --i) {
       expression = CreateUnaryOperatorByToken(operators[i], expression);
     }

src/query/v2/interpreter.cpp

@@ -1186,7 +1186,7 @@ PreparedQuery PrepareExplainQuery(ParsedQuery parsed_query, std::map<std::string
   // full query string) when given just the inner query to execute.
   ParsedQuery parsed_inner_query =
       ParseQuery(parsed_query.query_string.substr(kExplainQueryStart.size()), parsed_query.user_parameters,
-                 &interpreter_context->ast_cache, &interpreter_context->antlr_lock, interpreter_context->config.query);
+                 &interpreter_context->ast_cache, interpreter_context->config.query);
 
   auto *cypher_query = utils::Downcast<CypherQuery>(parsed_inner_query.query);
   MG_ASSERT(cypher_query, "Cypher grammar should not allow other queries in EXPLAIN");
@@ -1253,7 +1253,7 @@ PreparedQuery PrepareProfileQuery(ParsedQuery parsed_query, bool in_explicit_tra
   // full query string) when given just the inner query to execute.
   ParsedQuery parsed_inner_query =
       ParseQuery(parsed_query.query_string.substr(kProfileQueryStart.size()), parsed_query.user_parameters,
-                 &interpreter_context->ast_cache, &interpreter_context->antlr_lock, interpreter_context->config.query);
+                 &interpreter_context->ast_cache, interpreter_context->config.query);
 
   auto *cypher_query = utils::Downcast<CypherQuery>(parsed_inner_query.query);
   MG_ASSERT(cypher_query, "Cypher grammar should not allow other queries in PROFILE");
@@ -1571,8 +1571,7 @@ Callback CreateTrigger(TriggerQuery *trigger_query,
         interpreter_context->trigger_store.AddTrigger(
             std::move(trigger_name), trigger_statement, user_parameters, ToTriggerEventType(event_type),
             before_commit ? TriggerPhase::BEFORE_COMMIT : TriggerPhase::AFTER_COMMIT, &interpreter_context->ast_cache,
-            dba, &interpreter_context->antlr_lock, interpreter_context->config.query, std::move(owner),
-            interpreter_context->auth_checker);
+            dba, interpreter_context->config.query, std::move(owner), interpreter_context->auth_checker);
         return {};
       }};
 }
@@ -2129,8 +2128,8 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
     query_execution->summary["cost_estimate"] = 0.0;
 
     utils::Timer parsing_timer;
-    ParsedQuery parsed_query = ParseQuery(query_string, params, &interpreter_context_->ast_cache,
-                                          &interpreter_context_->antlr_lock, interpreter_context_->config.query);
+    ParsedQuery parsed_query =
+        ParseQuery(query_string, params, &interpreter_context_->ast_cache, interpreter_context_->config.query);
     query_execution->summary["parsing_time"] = parsing_timer.Elapsed().count();
 
     // Some queries require an active transaction in order to be prepared.

src/query/v2/interpreter.hpp

@@ -171,13 +171,6 @@ struct InterpreterContext {
 
   storage::v3::Storage *db;
 
-  // ANTLR has singleton instance that is shared between threads. It is
-  // protected by locks inside of ANTLR. Unfortunately, they are not protected
-  // in a very good way. Once we have ANTLR version without race conditions we
-  // can remove this lock. This will probably never happen since ANTLR
-  // developers introduce more bugs in each version. Fortunately, we have
-  // cache so this lock probably won't impact performance much...
-  utils::SpinLock antlr_lock;
   std::optional<double> tsc_frequency{utils::GetTSCFrequency()};
   std::atomic<bool> is_shutting_down{false};
 

src/query/v2/trigger.cpp

@@ -153,10 +153,10 @@ std::vector<std::pair<Identifier, TriggerIdentifierTag>> GetPredefinedIdentifier
 Trigger::Trigger(std::string name, const std::string &query,
                  const std::map<std::string, storage::v3::PropertyValue> &user_parameters,
                  const TriggerEventType event_type, utils::SkipList<QueryCacheEntry> *query_cache,
-                 DbAccessor *db_accessor, utils::SpinLock *antlr_lock, const InterpreterConfig::Query &query_config,
+                 DbAccessor *db_accessor, const InterpreterConfig::Query &query_config,
                  std::optional<std::string> owner, const query::v2::AuthChecker *auth_checker)
     : name_{std::move(name)},
-      parsed_statements_{ParseQuery(query, user_parameters, query_cache, antlr_lock, query_config)},
+      parsed_statements_{ParseQuery(query, user_parameters, query_cache, query_config)},
       event_type_{event_type},
       owner_{std::move(owner)} {
   // We check immediately if the query is valid by trying to create a plan.
@@ -257,7 +257,7 @@ inline constexpr uint64_t kVersion{2};
 TriggerStore::TriggerStore(std::filesystem::path directory) : storage_{std::move(directory)} {}
 
 void TriggerStore::RestoreTriggers(utils::SkipList<QueryCacheEntry> *query_cache, DbAccessor *db_accessor,
-                                   utils::SpinLock *antlr_lock, const InterpreterConfig::Query &query_config,
+                                   const InterpreterConfig::Query &query_config,
                                    const query::v2::AuthChecker *auth_checker) {
   MG_ASSERT(before_commit_triggers_.size() == 0 && after_commit_triggers_.size() == 0,
             "Cannot restore trigger when some triggers already exist!");
@@ -317,8 +317,8 @@ void TriggerStore::RestoreTriggers(utils::SkipList<QueryCacheEntry> *query_cache
 
     std::optional<Trigger> trigger;
     try {
-      trigger.emplace(trigger_name, statement, user_parameters, event_type, query_cache, db_accessor, antlr_lock,
-                      query_config, std::move(owner), auth_checker);
+      trigger.emplace(trigger_name, statement, user_parameters, event_type, query_cache, db_accessor, query_config,
+                      std::move(owner), auth_checker);
     } catch (const utils::BasicException &e) {
       spdlog::warn("Failed to create trigger '{}' because: {}", trigger_name, e.what());
       continue;
@@ -336,8 +336,8 @@ void TriggerStore::AddTrigger(std::string name, const std::string &query,
                               const std::map<std::string, storage::v3::PropertyValue> &user_parameters,
                               TriggerEventType event_type, TriggerPhase phase,
                               utils::SkipList<QueryCacheEntry> *query_cache, DbAccessor *db_accessor,
-                              utils::SpinLock *antlr_lock, const InterpreterConfig::Query &query_config,
-                              std::optional<std::string> owner, const query::v2::AuthChecker *auth_checker) {
+                              const InterpreterConfig::Query &query_config, std::optional<std::string> owner,
+                              const query::v2::AuthChecker *auth_checker) {
   std::unique_lock store_guard{store_lock_};
   if (storage_.Get(name)) {
     throw utils::BasicException("Trigger with the same name already exists.");
@@ -345,8 +345,8 @@ void TriggerStore::AddTrigger(std::string name, const std::string &query,
 
   std::optional<Trigger> trigger;
   try {
-    trigger.emplace(std::move(name), query, user_parameters, event_type, query_cache, db_accessor, antlr_lock,
-                    query_config, std::move(owner), auth_checker);
+    trigger.emplace(std::move(name), query, user_parameters, event_type, query_cache, db_accessor, query_config,
+                    std::move(owner), auth_checker);
   } catch (const utils::BasicException &e) {
     const auto identifiers = GetPredefinedIdentifiers(event_type);
     std::stringstream identifier_names_stream;

src/query/v2/trigger.hpp

@@ -35,8 +35,8 @@ struct Trigger {
   explicit Trigger(std::string name, const std::string &query,
                    const std::map<std::string, storage::v3::PropertyValue> &user_parameters,
                    TriggerEventType event_type, utils::SkipList<QueryCacheEntry> *query_cache, DbAccessor *db_accessor,
-                   utils::SpinLock *antlr_lock, const InterpreterConfig::Query &query_config,
-                   std::optional<std::string> owner, const AuthChecker *auth_checker);
+                   const InterpreterConfig::Query &query_config, std::optional<std::string> owner,
+                   const query::v2::AuthChecker *auth_checker);
 
   void Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution_memory, double max_execution_time_sec,
                std::atomic<bool> *is_shutting_down, const TriggerContext &context,
@@ -81,14 +81,13 @@ struct TriggerStore {
   explicit TriggerStore(std::filesystem::path directory);
 
   void RestoreTriggers(utils::SkipList<QueryCacheEntry> *query_cache, DbAccessor *db_accessor,
-                       utils::SpinLock *antlr_lock, const InterpreterConfig::Query &query_config,
-                       const query::v2::AuthChecker *auth_checker);
+                       const InterpreterConfig::Query &query_config, const query::v2::AuthChecker *auth_checker);
 
   void AddTrigger(std::string name, const std::string &query,
                   const std::map<std::string, storage::v3::PropertyValue> &user_parameters, TriggerEventType event_type,
                   TriggerPhase phase, utils::SkipList<QueryCacheEntry> *query_cache, DbAccessor *db_accessor,
-                  utils::SpinLock *antlr_lock, const InterpreterConfig::Query &query_config,
-                  std::optional<std::string> owner, const query::v2::AuthChecker *auth_checker);
+                  const InterpreterConfig::Query &query_config, std::optional<std::string> owner,
+                  const query::v2::AuthChecker *auth_checker);
 
   void DropTrigger(const std::string &name);
 

tests/simulation/CMakeLists.txt

@@ -29,6 +29,4 @@ add_simulation_test(future.cpp thread)
 
 add_simulation_test(basic_request.cpp address)
 
-add_simulation_test(raft.cpp address)
-
 add_simulation_test(trial_query_storage/query_storage_test.cpp address)

tests/simulation/basic_request.cpp

@@ -34,7 +34,7 @@ void run_server(Io<SimulatorTransport> io) {
 
   while (!io.ShouldShutDown()) {
     std::cout << "[SERVER] Is receiving..." << std::endl;
-    auto request_result = io.ReceiveWithTimeout<CounterRequest>(100000);
+    auto request_result = io.Receive<CounterRequest>();
     if (request_result.HasError()) {
       std::cout << "[SERVER] Error, continue" << std::endl;
       continue;
@@ -71,8 +71,8 @@ int main() {
     // send request
     CounterRequest cli_req;
     cli_req.proposal = i;
-    auto res_f = cli_io.RequestWithTimeout<CounterRequest, CounterResponse>(srv_addr, cli_req, 1000);
-    auto res_rez = res_f.Wait();
+    auto res_f = cli_io.Request<CounterRequest, CounterResponse>(srv_addr, cli_req);
+    auto res_rez = std::move(res_f).Wait();
     if (!res_rez.HasError()) {
       std::cout << "[CLIENT] Got a valid response" << std::endl;
       auto env = res_rez.GetValue();

tests/simulation/future.cpp

@@ -20,7 +20,7 @@ using namespace memgraph::io;
 void Fill(Promise<std::string> promise_1) { promise_1.Fill("success"); }
 
 void Wait(Future<std::string> future_1, Promise<std::string> promise_2) {
-  std::string result_1 = future_1.Wait();
+  std::string result_1 = std::move(future_1).Wait();
   MG_ASSERT(result_1 == "success");
   promise_2.Fill("it worked");
 }
@@ -49,7 +49,7 @@ int main() {
   t1.join();
   t2.join();
 
-  std::string result_2 = future_2.Wait();
+  std::string result_2 = std::move(future_2).Wait();
   MG_ASSERT(result_2 == "it worked");
 
   return 0;

tests/simulation/trial_query_storage/query_storage_test.cpp

@@ -26,7 +26,7 @@ using memgraph::io::simulator::SimulatorTransport;
 void run_server(Io<SimulatorTransport> io) {
   while (!io.ShouldShutDown()) {
     std::cout << "[STORAGE] Is receiving..." << std::endl;
-    auto request_result = io.ReceiveWithTimeout<ScanVerticesRequest>(100000);
+    auto request_result = io.Receive<ScanVerticesRequest>();
     if (request_result.HasError()) {
       std::cout << "[STORAGE] Error, continue" << std::endl;
       continue;
@@ -78,8 +78,8 @@ int main() {
 
   auto req = ScanVerticesRequest{2, std::nullopt};
 
-  auto res_f = cli_io.RequestWithTimeout<ScanVerticesRequest, VerticesResponse>(srv_addr, req, 1000);
-  auto res_rez = res_f.Wait();
+  auto res_f = cli_io.Request<ScanVerticesRequest, VerticesResponse>(srv_addr, req);
+  auto res_rez = std::move(res_f).Wait();
   // MG_ASSERT(res_rez.HasError());
   simulator.ShutDown();
   return 0;