Some changes

CMakeLists.txt

@@ -4,7 +4,6 @@ cmake_minimum_required(VERSION 3.8)
 
 # !! IMPORTANT !! run ./project_root/init.sh before cmake command
 # to download dependencies
-
 if(NOT UNIX)
   message(FATAL_ERROR "Unsupported operating system.")
 endif()
@@ -19,6 +18,7 @@ set_directory_properties(PROPERTIES CLEAN_NO_CUSTOM TRUE)
 find_program(CCACHE_FOUND ccache)
 option(USE_CCACHE "ccache:" ON)
 message(STATUS "CCache: ${USE_CCACHE}")
+
 if(CCACHE_FOUND AND USE_CCACHE)
   set_property(GLOBAL PROPERTY RULE_LAUNCH_COMPILE ccache)
   set_property(GLOBAL PROPERTY RULE_LAUNCH_LINK ccache)
@@ -28,7 +28,8 @@ endif(CCACHE_FOUND AND USE_CCACHE)
 # NOTE: must be choosen before use of project() or enable_language()
 find_program(CLANG_FOUND clang)
 find_program(CLANGXX_FOUND clang++)
-if (CLANG_FOUND AND CLANGXX_FOUND)
+
+if(CLANG_FOUND AND CLANGXX_FOUND)
   set(CMAKE_C_COMPILER ${CLANG_FOUND})
   set(CMAKE_CXX_COMPILER ${CLANGXX_FOUND})
 else()
@@ -36,12 +37,11 @@ else()
 endif()
 
 # -----------------------------------------------------------------------------
-
 project(memgraph)
 
 # Install licenses.
 install(DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/licenses/
-        DESTINATION share/doc/memgraph)
+  DESTINATION share/doc/memgraph)
 
 # For more information about how to release a new version of Memgraph, see
 # `release/README.md`.
@@ -61,61 +61,65 @@ set(MEMGRAPH_OVERRIDE_VERSION "")
 set(MEMGRAPH_OVERRIDE_VERSION_SUFFIX "")
 
 # Variables used to generate the versions.
-if (MG_ENTERPRISE)
+if(MG_ENTERPRISE)
   set(get_version_offering "")
 else()
   set(get_version_offering "--open-source")
 endif()
+
 set(get_version_script "${CMAKE_CURRENT_SOURCE_DIR}/release/get_version.py")
 
 # Get version that should be used in the binary.
 execute_process(
-    OUTPUT_VARIABLE MEMGRAPH_VERSION
-    RESULT_VARIABLE MEMGRAPH_VERSION_RESULT
-    COMMAND "${get_version_script}" ${get_version_offering}
-                "${MEMGRAPH_OVERRIDE_VERSION}"
-                "${MEMGRAPH_OVERRIDE_VERSION_SUFFIX}"
-                "--memgraph-root-dir"
-                "${CMAKE_CURRENT_SOURCE_DIR}"
+  OUTPUT_VARIABLE MEMGRAPH_VERSION
+  RESULT_VARIABLE MEMGRAPH_VERSION_RESULT
+  COMMAND "${get_version_script}" ${get_version_offering}
+  "${MEMGRAPH_OVERRIDE_VERSION}"
+  "${MEMGRAPH_OVERRIDE_VERSION_SUFFIX}"
+  "--memgraph-root-dir"
+  "${CMAKE_CURRENT_SOURCE_DIR}"
 )
+
 if(MEMGRAPH_VERSION_RESULT AND NOT MEMGRAPH_VERSION_RESULT EQUAL 0)
-    message(FATAL_ERROR "Unable to get Memgraph version.")
+  message(FATAL_ERROR "Unable to get Memgraph version.")
 else()
-    MESSAGE(STATUS "Memgraph version: ${MEMGRAPH_VERSION}")
+  MESSAGE(STATUS "Memgraph version: ${MEMGRAPH_VERSION}")
 endif()
 
 # Get version that should be used in the DEB package.
 execute_process(
-    OUTPUT_VARIABLE MEMGRAPH_VERSION_DEB
-    RESULT_VARIABLE MEMGRAPH_VERSION_DEB_RESULT
-    COMMAND "${get_version_script}" ${get_version_offering}
-                --variant deb
-                "${MEMGRAPH_OVERRIDE_VERSION}"
-                "${MEMGRAPH_OVERRIDE_VERSION_SUFFIX}"
-                "--memgraph-root-dir"
-                "${CMAKE_CURRENT_SOURCE_DIR}"
+  OUTPUT_VARIABLE MEMGRAPH_VERSION_DEB
+  RESULT_VARIABLE MEMGRAPH_VERSION_DEB_RESULT
+  COMMAND "${get_version_script}" ${get_version_offering}
+  --variant deb
+  "${MEMGRAPH_OVERRIDE_VERSION}"
+  "${MEMGRAPH_OVERRIDE_VERSION_SUFFIX}"
+  "--memgraph-root-dir"
+  "${CMAKE_CURRENT_SOURCE_DIR}"
 )
+
 if(MEMGRAPH_VERSION_DEB_RESULT AND NOT MEMGRAPH_VERSION_DEB_RESULT EQUAL 0)
-    message(FATAL_ERROR "Unable to get Memgraph DEB version.")
+  message(FATAL_ERROR "Unable to get Memgraph DEB version.")
 else()
-    MESSAGE(STATUS "Memgraph DEB version: ${MEMGRAPH_VERSION_DEB}")
+  MESSAGE(STATUS "Memgraph DEB version: ${MEMGRAPH_VERSION_DEB}")
 endif()
 
 # Get version that should be used in the RPM package.
 execute_process(
-    OUTPUT_VARIABLE MEMGRAPH_VERSION_RPM
-    RESULT_VARIABLE MEMGRAPH_VERSION_RPM_RESULT
-    COMMAND "${get_version_script}" ${get_version_offering}
-                --variant rpm
-                "${MEMGRAPH_OVERRIDE_VERSION}"
-                "${MEMGRAPH_OVERRIDE_VERSION_SUFFIX}"
-                "--memgraph-root-dir"
-                "${CMAKE_CURRENT_SOURCE_DIR}"
+  OUTPUT_VARIABLE MEMGRAPH_VERSION_RPM
+  RESULT_VARIABLE MEMGRAPH_VERSION_RPM_RESULT
+  COMMAND "${get_version_script}" ${get_version_offering}
+  --variant rpm
+  "${MEMGRAPH_OVERRIDE_VERSION}"
+  "${MEMGRAPH_OVERRIDE_VERSION_SUFFIX}"
+  "--memgraph-root-dir"
+  "${CMAKE_CURRENT_SOURCE_DIR}"
 )
+
 if(MEMGRAPH_VERSION_RPM_RESULT AND NOT MEMGRAPH_VERSION_RPM_RESULT EQUAL 0)
-    message(FATAL_ERROR "Unable to get Memgraph RPM version.")
+  message(FATAL_ERROR "Unable to get Memgraph RPM version.")
 else()
-    MESSAGE(STATUS "Memgraph RPM version: ${MEMGRAPH_VERSION_RPM}")
+  MESSAGE(STATUS "Memgraph RPM version: ${MEMGRAPH_VERSION_RPM}")
 endif()
 
 # We want the above variables to be updated each time something is committed to
@@ -135,22 +139,24 @@ endif()
 # unnecessary recalculations of the release version. The release version only
 # changes on every `git commit` or `git checkout`. That is why we watch the
 # following files for changes:
-#   - `.git/HEAD` -> changes each time a `git checkout` is issued
-#   - `.git/refs/heads/...` -> the value in `.git/HEAD` is a branch name (when
-#     you are on a branch) and you have to monitor the file of the specific
-#     branch to detect when a `git commit` was issued
+# - `.git/HEAD` -> changes each time a `git checkout` is issued
+# - `.git/refs/heads/...` -> the value in `.git/HEAD` is a branch name (when
+# you are on a branch) and you have to monitor the file of the specific
+# branch to detect when a `git commit` was issued
 # More details about the contents of the `.git` directory and the specific
 # files used can be seen here:
 # https://git-scm.com/book/en/v2/Git-Internals-Git-References
 set(git_directory "${CMAKE_SOURCE_DIR}/.git")
-if (EXISTS "${git_directory}")
+
+if(EXISTS "${git_directory}")
   set_property(DIRECTORY APPEND PROPERTY
-      CMAKE_CONFIGURE_DEPENDS "${git_directory}/HEAD")
+    CMAKE_CONFIGURE_DEPENDS "${git_directory}/HEAD")
   file(STRINGS "${git_directory}/HEAD" git_head_data)
-  if (git_head_data MATCHES "^ref: ")
+
+  if(git_head_data MATCHES "^ref: ")
     string(SUBSTRING "${git_head_data}" 5 -1 git_head_ref)
     set_property(DIRECTORY APPEND PROPERTY
-        CMAKE_CONFIGURE_DEPENDS "${git_directory}/${git_head_ref}")
+      CMAKE_CONFIGURE_DEPENDS "${git_directory}/${git_head_ref}")
   endif()
 endif()
 
@@ -159,16 +165,19 @@ endif()
 # setup CMake module path, defines path for include() and find_package()
 # https://cmake.org/cmake/help/latest/variable/CMAKE_MODULE_PATH.html
 set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} ${PROJECT_SOURCE_DIR}/cmake)
+
 # custom function definitions
 include(functions)
+
 # -----------------------------------------------------------------------------
 
 # We want out of source builds, so that cmake generated files don't get mixed
 # with source files. This allows for easier clean up.
 disallow_in_source_build()
 add_custom_target(clean_all
-                  COMMAND ${CMAKE_COMMAND} -P ${PROJECT_SOURCE_DIR}/cmake/clean_all.cmake
-                  COMMENT "Removing all files in ${CMAKE_BINARY_DIR}")
+  COMMAND ${CMAKE_COMMAND} -P ${PROJECT_SOURCE_DIR}/cmake/clean_all.cmake
+  COMMENT "Removing all files in ${CMAKE_BINARY_DIR}")
+
 # -----------------------------------------------------------------------------
 
 # build flags -----------------------------------------------------------------
@@ -179,17 +188,19 @@ set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 
 set(CMAKE_CXX_STANDARD 20)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
+
 # c99-designator is disabled because of required mixture of designated and
 # non-designated initializers in Python Query Module code (`py_module.cpp`).
 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall \
     -Werror=switch -Werror=switch-bool -Werror=return-type \
     -Werror=return-stack-address \
     -Wno-c99-designator \
+    -DBOOST_STACKTRACE_USE_ADDR2LINE \
     -DBOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT")
 
 # Don't omit frame pointer in RelWithDebInfo, for additional callchain debug.
 set(CMAKE_CXX_FLAGS_RELWITHDEBINFO
-    "${CMAKE_CXX_FLAGS_RELWITHDEBINFO} -fno-omit-frame-pointer")
+  "${CMAKE_CXX_FLAGS_RELWITHDEBINFO} -fno-omit-frame-pointer")
 
 # Statically link libgcc and libstdc++, the GCC allows this according to:
 # https://gcc.gnu.org/onlinedocs/gcc-10.2.0/libstdc++/manual/manual/license.html
@@ -207,42 +218,47 @@ set(CMAKE_CXX_FLAGS_RELEASE "-O2 -DNDEBUG")
 
 SET(CMAKE_CXX_LINK_FLAGS "${CMAKE_CXX_LINK_FLAGS} -pthread")
 
-#debug flags
+# debug flags
 set(PREFERRED_DEBUGGER "gdb" CACHE STRING
-    "Tunes the debug output for your preferred debugger (gdb or lldb).")
-if ("${PREFERRED_DEBUGGER}" STREQUAL "gdb" AND
-    "${CMAKE_CXX_COMPILER_ID}" MATCHES "Clang|GNU")
-    set(CMAKE_CXX_FLAGS_DEBUG "-ggdb")
-elseif ("${PREFERRED_DEBUGGER}" STREQUAL "lldb" AND
-        "${CMAKE_CXX_COMPILER_ID}" STREQUAL "Clang")
-    set(CMAKE_CXX_FLAGS_DEBUG "-glldb")
+  "Tunes the debug output for your preferred debugger (gdb or lldb).")
+
+if("${PREFERRED_DEBUGGER}" STREQUAL "gdb" AND
+  "${CMAKE_CXX_COMPILER_ID}" MATCHES "Clang|GNU")
+  set(CMAKE_CXX_FLAGS_DEBUG "-ggdb")
+elseif("${PREFERRED_DEBUGGER}" STREQUAL "lldb" AND
+  "${CMAKE_CXX_COMPILER_ID}" STREQUAL "Clang")
+  set(CMAKE_CXX_FLAGS_DEBUG "-glldb")
 else()
-    message(WARNING "Unable to tune for PREFERRED_DEBUGGER: "
-            "'${PREFERRED_DEBUGGER}' with compiler: '${CMAKE_CXX_COMPILER_ID}'")
-    set(CMAKE_CXX_FLAGS_DEBUG "-g")
+  message(WARNING "Unable to tune for PREFERRED_DEBUGGER: "
+    "'${PREFERRED_DEBUGGER}' with compiler: '${CMAKE_CXX_COMPILER_ID}'")
+  set(CMAKE_CXX_FLAGS_DEBUG "-g")
 endif()
 
 # -----------------------------------------------------------------------------
 
 # default build type is debug
-if (NOT CMAKE_BUILD_TYPE)
-    set(CMAKE_BUILD_TYPE "Debug")
+if(NOT CMAKE_BUILD_TYPE)
+  set(CMAKE_BUILD_TYPE "Debug")
 endif()
-message(STATUS "CMake build type: ${CMAKE_BUILD_TYPE}")
-# -----------------------------------------------------------------------------
 
+message(STATUS "CMake build type: ${CMAKE_BUILD_TYPE}")
+
+# -----------------------------------------------------------------------------
 set(MG_ARCH "x86_64" CACHE STRING "Host architecture to build Memgraph on. Supported values are x86_64 (default), ARM64.")
 
 # setup external dependencies -------------------------------------------------
 
 # threading
 find_package(Threads REQUIRED)
+
 # optional readline
 option(USE_READLINE "Use GNU Readline library if available (default ON). \
 Set this to OFF to prevent linking with Readline even if it is available." ON)
-if (USE_READLINE)
+
+if(USE_READLINE)
   find_package(Readline)
-  if (READLINE_FOUND)
+
+  if(READLINE_FOUND)
     add_definitions(-DHAS_READLINE)
   endif()
 endif()
@@ -258,24 +274,27 @@ option(ASAN "Build with Address Sanitizer. To get a reasonable performance optio
 option(TSAN "Build with Thread Sanitizer. To get a reasonable performance option should be used only in Release or RelWithDebInfo build " OFF)
 option(UBSAN "Build with Undefined Behaviour Sanitizer" OFF)
 
-if (TEST_COVERAGE)
+if(TEST_COVERAGE)
   string(TOLOWER ${CMAKE_BUILD_TYPE} lower_build_type)
-  if (NOT lower_build_type STREQUAL "debug")
+
+  if(NOT lower_build_type STREQUAL "debug")
     message(FATAL_ERROR "Generating test coverage unsupported in non Debug builds. Current build type is '${CMAKE_BUILD_TYPE}'")
   endif()
+
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fprofile-instr-generate -fcoverage-mapping")
   set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fprofile-instr-generate -fcoverage-mapping")
 endif()
 
-if (MG_ENTERPRISE)
+if(MG_ENTERPRISE)
   add_definitions(-DMG_ENTERPRISE)
 endif()
 
 set(ENABLE_JEMALLOC ON)
 
-if (ASAN)
+if(ASAN)
   message(WARNING "Disabling jemalloc as it doesn't work well with ASAN")
   set(ENABLE_JEMALLOC OFF)
+
   # Enable Addres sanitizer and get nicer stack traces in error messages.
   # NOTE: AddressSanitizer uses llvm-symbolizer binary from the Clang
   # distribution to symbolize the stack traces (note that ideally the
@@ -284,19 +303,20 @@ if (ASAN)
   # provide it in separate ASAN_SYMBOLIZER_PATH environment variable.
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=address -fno-omit-frame-pointer")
   set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fsanitize=address")
+
   # To detect Stack-use-after-return bugs set run-time flag:
-  #   ASAN_OPTIONS=detect_stack_use_after_return=1
+  # ASAN_OPTIONS=detect_stack_use_after_return=1
   # To check initialization order bugs set run-time flag:
-  #   ASAN_OPTIONS=check_initialization_order=true
-  #     This mode reports an error if initializer for a global variable accesses
-  #     dynamically initialized global from another translation unit, which is
-  #     not yet initialized
-  #   ASAN_OPTIONS=strict_init_order=true
-  #     This mode reports an error if initializer for a global variable accesses
-  #     any dynamically initialized global from another translation unit.
+  # ASAN_OPTIONS=check_initialization_order=true
+  # This mode reports an error if initializer for a global variable accesses
+  # dynamically initialized global from another translation unit, which is
+  # not yet initialized
+  # ASAN_OPTIONS=strict_init_order=true
+  # This mode reports an error if initializer for a global variable accesses
+  # any dynamically initialized global from another translation unit.
 endif()
 
-if (TSAN)
+if(TSAN)
   # ThreadSanitizer generally requires all code to be compiled with -fsanitize=thread.
   # If some code (e.g. dynamic libraries) is not compiled with the flag, it can
   # lead to false positive race reports, false negative race reports and/or
@@ -309,19 +329,21 @@ if (TSAN)
   # (unwinding stack on each memory access is too expensive).
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=thread")
   set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fsanitize=thread")
+
   # By default ThreadSanitizer uses addr2line utility to symbolize reports.
   # llvm-symbolizer is faster, consumes less memory and produces much better
   # reports. To use it set runtime flag:
-  #   TSAN_OPTIONS="extern-symbolizer-path=~/llvm-symbolizer"
+  # TSAN_OPTIONS="extern-symbolizer-path=~/llvm-symbolizer"
   # For more runtime flags see: https://github.com/google/sanitizers/wiki/ThreadSanitizerFlags
 endif()
 
-if (UBSAN)
+if(UBSAN)
   # Compile with UBSAN but disable vptr check. This is disabled because it
   # requires linking with clang++ to make sure C++ specific parts of the
   # runtime library and c++ standard libraries are present.
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=undefined -fno-omit-frame-pointer -fno-sanitize=vptr")
   set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fsanitize=undefined -fno-sanitize=vptr")
+
   # Run program with environment variable UBSAN_OPTIONS=print_stacktrace=1.
   # Make sure llvm-symbolizer binary is in path.
   # To make the program abort on undefined behavior, use UBSAN_OPTIONS=halt_on_error=1.
@@ -340,7 +362,7 @@ add_subdirectory(release)
 option(MG_ENABLE_TESTING "Set this to OFF to disable building test binaries" ON)
 message(STATUS "MG_ENABLE_TESTING: ${MG_ENABLE_TESTING}")
 
-if (MG_ENABLE_TESTING)
+if(MG_ENABLE_TESTING)
   enable_testing()
   add_subdirectory(tests)
 endif()

src/io/rsm/raft.hpp

@@ -312,12 +312,19 @@ class Raft {
   }
 
   void Run() {
+    // debug - gvolfing
+    uint64_t tick_count = 0;
+
     while (!io_.ShouldShutDown()) {
       const auto now = io_.Now();
       if (now >= next_cron_) {
         next_cron_ = Cron();
       }
 
+      // debug - gvolfing
+      spdlog::info("Raft::Run() awakened on thread {}, Cron tick: {}", std::this_thread::get_id(), tick_count);
+      tick_count++;
+
       const Duration receive_timeout = RandomTimeout(kMinimumReceiveTimeout, kMaximumReceiveTimeout);
 
       auto request_result =

src/io/simulator/simulator_handle.cpp

@@ -58,6 +58,20 @@ bool SimulatorHandle::MaybeTickSimulator() {
 
   const size_t blocked_servers = blocked_on_receive_.size();
 
+  // Don't judge, this is a workaround for poc. gvolfing
+  server_count_++;
+  if (server_count_ == 3) {
+    is_quiescent_state_achieved_ = true;
+  }
+
+  // if (!is_quiescent_state_achieved_) {
+  if (!is_quiescent_state_achieved_) {
+    // return SimulatorProgression::DidSetUpOneServer;
+    cv_.notify_all();  // -> deadlock
+    // return false;// -> deadlock
+    return true;  // -> deadlock
+  }
+  // if (blocked_servers < server_addresses_.size() && is_quiescent_state_achieved_) {
   if (blocked_servers < server_addresses_.size()) {
     // we only need to advance the simulator when all
     // servers have reached a quiescent state, blocked
@@ -71,13 +85,24 @@ bool SimulatorHandle::MaybeTickSimulator() {
 
   TimeoutPromisesPastDeadline();
 
+  auto sort_based_on_message_id = [](const std::pair<Address, OpaqueMessage> &lhs,
+                                     const std::pair<Address, OpaqueMessage> &rhs) {
+    // return a.second.request_id < b.second.request_id;
+    if (lhs.second.from_address != rhs.second.from_address) {
+      return lhs.second.from_address < rhs.second.from_address;
+    }
+
+    return lhs.second.request_id < rhs.second.request_id;
+  };
+  std::sort(in_flight_.begin(), in_flight_.end(), sort_based_on_message_id);
+
   if (in_flight_.empty()) {
     // return early here because there are no messages to schedule
 
     // We tick the clock forward when all servers are blocked but
     // there are no in-flight messages to schedule delivery of.
-    const Duration clock_advance = std::chrono::microseconds{time_distrib_(rng_)};
-    cluster_wide_time_microseconds_ += clock_advance;
+    // const Duration clock_advance = std::chrono::microseconds{Rand(time_distrib_)};
+    // cluster_wide_time_microseconds_ += clock_advance;
 
     if (cluster_wide_time_microseconds_ >= config_.abort_time) {
       if (should_shut_down_) {
@@ -88,20 +113,25 @@ bool SimulatorHandle::MaybeTickSimulator() {
           "in an expected amount of time.");
       throw utils::BasicException{"Cluster has executed beyond its configured abort_time"};
     }
+    // spdlog::info(
+    //     "Time increased with {} to {}", clock_advance.count(),
+    //     std::chrono::duration_cast<std::chrono::milliseconds>(cluster_wide_time_microseconds_.time_since_epoch())
+    //         .count());
     return true;
   }
 
   if (config_.scramble_messages) {
     // scramble messages
     std::uniform_int_distribution<size_t> swap_distrib(0, in_flight_.size() - 1);
-    const size_t swap_index = swap_distrib(rng_);
+    const size_t swap_index = RandLocked(swap_distrib);
     std::swap(in_flight_[swap_index], in_flight_.back());
   }
 
   auto [to_address, opaque_message] = std::move(in_flight_.back());
   in_flight_.pop_back();
+  event_log_.AddAndLog(opaque_message, "Handle ");
 
-  const int drop_threshold = drop_distrib_(rng_);
+  const int drop_threshold = RandLocked(drop_distrib_);
   const bool should_drop = drop_threshold < config_.drop_percent;
 
   if (should_drop) {
@@ -111,6 +141,7 @@ bool SimulatorHandle::MaybeTickSimulator() {
   PromiseKey promise_key{.requester_address = to_address, .request_id = opaque_message.request_id};
 
   if (promises_.contains(promise_key)) {
+    event_log_.AddAndLog(opaque_message, "Promise contains");
     // complete waiting promise if it's there
     DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key));
     promises_.erase(promise_key);
@@ -128,9 +159,16 @@ bool SimulatorHandle::MaybeTickSimulator() {
       histograms_.Measure(type_info, response_latency);
     }
   } else if (should_drop) {
+    event_log_.AddAndLog(opaque_message, "Dropped");
     // don't add it anywhere, let it drop
   } else {
+    event_log_.AddAndLog(opaque_message, "To can_receive");
     // add to can_receive_ if not
+    // This might be needed I am not sure, it can cause deadlock if you uncomment. Didn't find the issue yet.
+    // MG_ASSERT(server_addresses_.contains(to_address));
+    // MG_ASSERT(blocked_on_receive_.contains(to_address));
+    // blocked_on_receive_.erase(to_address);
+
     const auto &[om_vec, inserted] =
         can_receive_.try_emplace(to_address.ToPartialAddress(), std::vector<OpaqueMessage>());
     om_vec->second.emplace_back(std::move(opaque_message));
@@ -144,6 +182,11 @@ Time SimulatorHandle::Now() const {
   return cluster_wide_time_microseconds_;
 }
 
+Time SimulatorHandle::NowLocked() const {
+  // std::unique_lock<std::mutex> lock(mu_);
+  return cluster_wide_time_microseconds_;
+}
+
 SimulatorStats SimulatorHandle::Stats() {
   std::unique_lock<std::mutex> lock(mu_);
   return stats_;

src/io/simulator/simulator_handle.hpp

@@ -17,11 +17,20 @@
 #include <map>
 #include <memory>
 #include <optional>
+#include <queue>
 #include <set>
+#include <string>
 #include <utility>
 #include <variant>
 #include <vector>
 
+#include <sstream>
+
+#include <boost/lexical_cast.hpp>
+#include <boost/stacktrace.hpp>
+#include <boost/uuid/uuid_io.hpp>
+
+#include "fmt/format.h"
 #include "io/address.hpp"
 #include "io/errors.hpp"
 #include "io/message_conversion.hpp"
@@ -30,9 +39,50 @@
 #include "io/simulator/simulator_stats.hpp"
 #include "io/time.hpp"
 #include "io/transport.hpp"
+#include "spdlog/spdlog.h"
 
 namespace memgraph::io::simulator {
 
+struct EventDescriptor {
+  boost::asio::ip::address from_address_ip;
+  uint16_t from_address_port;
+  boost::asio::ip::address to_address_ip;
+  uint16_t to_address_port;
+  uint64_t request_id;
+  std::string_view caller;
+  utils::TypeInfoRef type_info;
+  uint64_t event_id;
+  Time time;
+};
+
+class SimulatorHandle;
+class EventLog {
+  std::vector<EventDescriptor> event_log_;
+  static constexpr const char *eventlog_signal_string_ = "EVENTLOG_UPDATE";
+
+  void LogEvent(const EventDescriptor &event) const {
+    spdlog::info(
+        fmt::format("{} ({})-> caller: {}, from_address: {}:{}, to_address: {}:{}, request_id: {}, type: {}, event_id: "
+                    "{}, thread_id: {}",
+                    eventlog_signal_string_,
+                    std::chrono::duration_cast<std::chrono::milliseconds>(event.time.time_since_epoch()).count(),
+                    event.caller, event.from_address_ip.to_string(), event.from_address_port,
+                    event.to_address_ip.to_string(), event.to_address_port, event.request_id,
+                    event.type_info.get().name(), event.event_id, std::this_thread::get_id()));
+  }
+
+ public:
+  SimulatorHandle *handle{nullptr};
+
+  void AddAndLog(const OpaqueMessage &message, std::string_view caller);
+
+  void LogAllEvents() const {
+    for (const auto &event : event_log_) {
+      LogEvent(event);
+    }
+  }
+};
+
 class SimulatorHandle {
   mutable std::mutex mu_{};
   mutable std::condition_variable cv_;
@@ -46,6 +96,10 @@ class SimulatorHandle {
   // messages that are sent to servers that may later receive them
   std::map<PartialAddress, std::vector<OpaqueMessage>> can_receive_;
 
+  // maybe a boolean here
+  bool is_quiescent_state_achieved_ = false;
+  std::atomic<int> server_count_;
+
   Time cluster_wide_time_microseconds_;
   bool should_shut_down_ = false;
   SimulatorStats stats_;
@@ -57,13 +111,17 @@ class SimulatorHandle {
   SimulatorConfig config_;
   MessageHistogramCollector histograms_;
   RequestId request_id_counter_{0};
+  EventLog event_log_;
 
   void TimeoutPromisesPastDeadline() {
     const Time now = cluster_wide_time_microseconds_;
     for (auto it = promises_.begin(); it != promises_.end();) {
       auto &[promise_key, dop] = *it;
-      if (dop.deadline < now && config_.perform_timeouts) {
-        spdlog::info("timing out request from requester {}.", promise_key.requester_address.ToString());
+      const bool timed_out = dop.deadline < now;
+      if (timed_out && config_.perform_timeouts) {
+        // spdlog::info("timing out request from requester {}.", promise_key.requester_address.ToString());
+        spdlog::info("timing out request from requester {}. bool timed_out: {}",
+                     promise_key.requester_address.ToString(), timed_out);
         std::move(dop).promise.TimeOut();
         it = promises_.erase(it);
 
@@ -76,7 +134,10 @@ class SimulatorHandle {
 
  public:
   explicit SimulatorHandle(SimulatorConfig config)
-      : cluster_wide_time_microseconds_(config.start_time), rng_(config.rng_seed), config_(config) {}
+      : server_count_(0), cluster_wide_time_microseconds_(config.start_time), rng_(config.rng_seed), config_(config) {
+    spdlog::info("SimulatorHandle constructed.");
+    event_log_.handle = this;
+  }
 
   LatencyHistogramSummaries ResponseLatencies();
 
@@ -120,6 +181,7 @@ class SimulatorHandle {
                      .request_id = request_id,
                      .message = std::move(message),
                      .type_info = type_info};
+    event_log_.AddAndLog(om, "SubmitRequest");
     in_flight_.emplace_back(std::make_pair(to_address, std::move(om)));
 
     PromiseKey promise_key{.requester_address = from_address, .request_id = request_id};
@@ -159,6 +221,8 @@ class SimulatorHandle {
           OpaqueMessage message = std::move(can_rx.back());
           can_rx.pop_back();
 
+          event_log_.AddAndLog(message, "Receivexxxxxx");
+
           // 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 = std::move(message).Take<Ms...>();
@@ -171,9 +235,9 @@ class SimulatorHandle {
       }
 
       lock.unlock();
-      bool made_progress = MaybeTickSimulator();
+      auto simulator_progress = MaybeTickSimulator();
       lock.lock();
-      if (!should_shut_down_ && !made_progress) {
+      if (!should_shut_down_ && !simulator_progress) {
         cv_.wait(lock);
       }
     }
@@ -193,6 +257,7 @@ class SimulatorHandle {
                      .request_id = request_id,
                      .message = std::move(message_any),
                      .type_info = type_info};
+    event_log_.AddAndLog(om, "Sendxxxxxxxxx");
     in_flight_.emplace_back(std::make_pair(std::move(to_address), std::move(om)));
 
     stats_.total_messages++;
@@ -202,12 +267,37 @@ class SimulatorHandle {
 
   Time Now() const;
 
+  Time NowLocked() const;
+
   template <class D = std::poisson_distribution<>, class Return = uint64_t>
   Return Rand(D distrib) {
     std::unique_lock<std::mutex> lock(mu_);
-    return distrib(rng_);
+    return RandLocked<D, Return>(std::forward<D>(distrib));
+  }
+
+  template <class D = std::poisson_distribution<>, class Return = uint64_t>
+  Return RandLocked(D distrib) {
+    Return res = distrib(rng_);
+    spdlog::info("Getting random value from thread {}: {}. Stacktrace:\n{}", std::this_thread::get_id(), res,
+                 boost::stacktrace::stacktrace());
+    return res;
   }
 
   SimulatorStats Stats();
 };
+
+inline void EventLog::AddAndLog(const OpaqueMessage &message, std::string_view caller) {
+  EventDescriptor event{.from_address_ip = message.from_address.last_known_ip,
+                        .from_address_port = message.from_address.last_known_port,
+                        .to_address_ip = message.to_address.last_known_ip,
+                        .to_address_port = message.to_address.last_known_port,
+                        .request_id = message.request_id,
+                        .caller = caller,
+                        .type_info = message.type_info,
+                        .event_id = event_log_.size(),
+                        .time = handle->NowLocked()};
+
+  event_log_.push_back(event);
+  LogEvent(event);
+}
 };  // namespace memgraph::io::simulator

src/io/simulator/simulator_transport.hpp

@@ -54,9 +54,11 @@ class SimulatorTransport {
 
   bool ShouldShutDown() const { return simulator_handle_->ShouldShutDown(); }
 
-  template <class D = std::poisson_distribution<>, class Return = uint64_t>
+  template <class D = std::poisson_distribution<uint64_t>, class Return = uint64_t>
   Return Rand(D distrib) {
-    return distrib(rng_);
+    // debug - gvolfing
+    auto ret = distrib(rng_);
+    return ret;
   }
 
   LatencyHistogramSummaries ResponseLatencies() { return simulator_handle_->ResponseLatencies(); }

tests/simulation/raft.cpp

@@ -23,6 +23,7 @@
 #include "io/rsm/rsm_client.hpp"
 #include "io/simulator/simulator.hpp"
 #include "io/simulator/simulator_transport.hpp"
+#include "utils/thread.hpp"
 
 using memgraph::io::Address;
 using memgraph::io::Duration;
@@ -120,6 +121,8 @@ class TestState {
 
 template <typename IoImpl>
 void RunRaft(Raft<IoImpl, TestState, CasRequest, CasResponse, GetRequest, GetResponse> server) {
+  memgraph::utils::ThreadSetName(fmt::format("Port {}", server.GetAddress().last_known_port));
+  spdlog::info("something_random init raft servers... thread_id: {}", std::this_thread::get_id());
   server.Run();
 }
 
@@ -163,6 +166,8 @@ void RunSimulation() {
   auto srv_thread_3 = std::jthread(RunRaft<SimulatorTransport>, std::move(srv_3));
   simulator.IncrementServerCountAndWaitForQuiescentState(srv_addr_3);
 
+  // wait for all the servers.
+
   spdlog::info("beginning test after servers have become quiescent");
 
   std::mt19937 cli_rng_{0};
@@ -187,11 +192,13 @@ void RunSimulation() {
 
     cas_req.new_value = i;
 
+    spdlog::info("Sending WriteRequest...");
     auto write_cas_response_result = client.SendWriteRequest(cas_req);
     if (write_cas_response_result.HasError()) {
       // timed out
       continue;
     }
+    spdlog::info("Wainting on the response of WriteRequest...");
     CasResponse cas_response = write_cas_response_result.GetValue();
 
     bool cas_succeeded = cas_response.cas_success;
@@ -249,7 +256,7 @@ void RunSimulation() {
 }
 
 int main() {
-  int n_tests = 50;
+  int n_tests = 1;
 
   for (int i = 0; i < n_tests; i++) {
     spdlog::info("========================== NEW SIMULATION {} ==========================", i);