Check-in high-level skeleton of some things for the protobuf transport

src/io/crc_frame.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
+
+namespace memgraph::io {
+/// Protocol:
+///   crc32:        4 bytes
+///   len:          8 bytes
+///   buffer:       <len bytes>
+
+std::optional<std::pair<size_t, size_t>> make_frame(char *ptr, size_t len) { return std::nullopt; }
+}  // namespace memgraph::io

src/io/protobuf_transport/protobuf_transport.hpp

@@ -0,0 +1,61 @@
+// 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 "io/address.hpp"
+#include "io/message_conversion.hpp"
+#include "io/protobuf_transport/protobuf_transport_handle.hpp"
+#include "io/transport.hpp"
+
+namespace memgraph::io::protobuf_transport {
+
+class ProtobufTransport {
+  std::shared_ptr<ProtobufTransportHandle> protobuf_transport_handle_;
+  uint16_t listen_port_;
+
+ public:
+  explicit ProtobufTransport(uint16_t listen_port)
+      : protobuf_transport_handle_(std::make_unique<ProtobufTransportHandle>()), listen_port_(listen_port) {}
+
+  template <Message RequestT, Message ResponseT>
+  ResponseFuture<ResponseT> Request(Address to_address, Address from_address, RequestId request_id, RequestT request,
+                                    Duration timeout) {
+    auto [future, promise] = memgraph::io::FuturePromisePair<ResponseResult<ResponseT>>();
+
+    protobuf_transport_handle_->SubmitRequest(to_address, from_address, request_id, std::move(request), timeout,
+                                              std::move(promise));
+
+    return std::move(future);
+  }
+
+  template <Message... Ms>
+  requires(sizeof...(Ms) > 0) RequestResult<Ms...> Receive(Address receiver_address, Duration timeout) {
+    return protobuf_transport_handle_->template Receive<Ms...>(receiver_address, timeout);
+  }
+
+  template <Message M>
+  void Send(Address to_address, Address from_address, RequestId request_id, M &&message) {
+    return protobuf_transport_handle_->template Send<M>(to_address, from_address, request_id, std::forward<M>(message));
+  }
+
+  Time Now() const { return protobuf_transport_handle_->Now(); }
+
+  bool ShouldShutDown() const { return protobuf_transport_handle_->ShouldShutDown(); }
+
+  template <class D = std::poisson_distribution<>, class Return = uint64_t>
+  Return Rand(D distrib) {
+    std::random_device rng;
+    return distrib(rng);
+  }
+};
+
+};  // namespace memgraph::io::protobuf_transport

src/io/protobuf_transport/protobuf_transport_handle.hpp

@@ -0,0 +1,161 @@
+// 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>
+#include <condition_variable>
+#include <iostream>
+#include <map>
+#include <mutex>
+
+#include "google/protobuf/message.h"
+
+#include "io/address.hpp"
+#include "io/transport.hpp"
+//#include "protobuf/messages.pb.cc"
+#include "protobuf/messages.pb.h"
+
+namespace memgraph::io::protobuf_transport {
+
+using PbAddress = memgraph::protobuf::Address;
+using memgraph::protobuf::UberMessage;
+
+class ProtobufTransportHandle {
+  mutable std::mutex mu_{};
+  mutable std::condition_variable cv_;
+  bool should_shut_down_ = false;
+
+  // 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::vector<std::string> can_receive_;
+
+  // serialized outbound messages
+  std::vector<std::string> outbox_;
+
+ public:
+  ~ProtobufTransportHandle() {
+    for (auto &&[pk, promise] : promises_) {
+      std::move(promise.promise).TimeOut();
+    }
+    promises_.clear();
+  }
+
+  void ShutDown() {
+    std::unique_lock<std::mutex> lock(mu_);
+    should_shut_down_ = true;
+    cv_.notify_all();
+  }
+
+  bool ShouldShutDown() const {
+    std::unique_lock<std::mutex> lock(mu_);
+    return should_shut_down_;
+  }
+
+  static Time Now() {
+    auto nano_time = std::chrono::system_clock::now();
+    return std::chrono::time_point_cast<std::chrono::microseconds>(nano_time);
+  }
+
+  template <Message... Ms>
+  requires(sizeof...(Ms) > 0) RequestResult<Ms...> Receive(Address /* receiver_address */, Duration timeout) {
+    std::unique_lock lock(mu_);
+
+    Time before = Now();
+
+    spdlog::info("can_receive_ size: {}", can_receive_.size());
+
+    while (can_receive_.empty()) {
+      Time now = Now();
+
+      // protection against non-monotonic timesources
+      auto maxed_now = std::max(now, before);
+      auto elapsed = maxed_now - before;
+
+      if (timeout < elapsed) {
+        return TimedOut{};
+      }
+
+      Duration relative_timeout = timeout - elapsed;
+
+      std::cv_status cv_status_value = cv_.wait_for(lock, relative_timeout);
+
+      if (cv_status_value == std::cv_status::timeout) {
+        return TimedOut{};
+      }
+    }
+
+    auto current_message = std::move(can_receive_.back());
+    can_receive_.pop_back();
+
+    auto m_opt = std::move(current_message).Take<Ms...>();
+
+    return std::move(m_opt).value();
+  }
+
+  template <Message M>
+  void Send(Address to_address, Address from_address, RequestId request_id, M &&message) {
+    PromiseKey promise_key{.requester_address = to_address, .request_id = request_id, .replier_address = from_address};
+
+    {
+      std::unique_lock<std::mutex> lock(mu_);
+
+      if (promises_.contains(promise_key)) {
+        // hair-pin local message optimization
+        spdlog::info("using message to fill local promise");
+        DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key));
+        promises_.erase(promise_key);
+
+        std::any message_any(std::forward<M>(message));
+        OpaqueMessage opaque_message{.to_address = to_address,
+                                     .from_address = from_address,
+                                     .request_id = request_id,
+                                     .message = std::move(message_any)};
+
+        dop.promise.Fill(std::move(opaque_message));
+      } else {
+        spdlog::info("placing message in outbox");
+
+        // serialize protobuf message and place it in the outbox
+
+        std::string bytes;
+        bool success = message.SerializeToString(&bytes);
+        MG_ASSERT(success);
+
+        outbox_.emplace_back(std::move(bytes));
+      }
+    }  // lock dropped
+
+    cv_.notify_all();
+  }
+
+  template <Message RequestT, Message ResponseT>
+  void SubmitRequest(Address to_address, Address from_address, RequestId request_id, RequestT &&request,
+                     Duration timeout, ResponsePromise<ResponseT> promise) {
+    const Time deadline = Now() + timeout;
+
+    {
+      std::unique_lock<std::mutex> lock(mu_);
+
+      PromiseKey promise_key{
+          .requester_address = from_address, .request_id = request_id, .replier_address = to_address};
+      OpaquePromise opaque_promise(std::move(promise).ToUnique());
+      DeadlineAndOpaquePromise dop{.deadline = deadline, .promise = std::move(opaque_promise)};
+      promises_.emplace(std::move(promise_key), std::move(dop));
+    }  // lock dropped
+
+    Send(to_address, from_address, request_id, std::forward<RequestT>(request));
+  }
+};
+
+}  // namespace memgraph::io::protobuf_transport

src/io/protobuf_transport/server.hpp

@@ -0,0 +1,25 @@
+// 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>
+#include <condition_variable>
+#include <iostream>
+#include <map>
+#include <mutex>
+
+#include "google/protobuf/message.h"
+
+#include "io/address.hpp"
+#include "io/transport.hpp"
+
+namespace memgraph::io::protobuf_transport {}

src/protobuf/messages.proto

@@ -0,0 +1,24 @@
+syntax = "proto3";
+
+package memgraph.protobuf;
+
+message Address {
+  bytes unique_id = 1;
+  // this is either 4 bytes for ipv4 or 16 bytes for ipv6
+  bytes last_known_ip = 2;
+  uint32 last_known_port = 3;
+}
+
+message TestRequest {
+  uint64 request_id = 1;
+  string content = 2;
+}
+
+message UberMessage {
+  uint64 request_id = 1;
+  Address to_address = 2;
+  Address from_address = 3;
+  oneof specific_message {
+    TestRequest test_request = 4;
+  }
+}

tests/unit/CMakeLists.txt

@@ -433,3 +433,7 @@ target_link_libraries(${test_prefix}local_transport mg-io)
 # Test MachineManager with LocalTransport
 add_unit_test(machine_manager.cpp)
 target_link_libraries(${test_prefix}machine_manager mg-io mg-coordinator mg-storage-v3)
+
+# Test ProtobufTransport
+add_unit_test(protobuf_transport.cpp)
+target_link_libraries(${test_prefix}protobuf_transport mg-io protobuf)

tests/unit/machine_manager.cpp

@@ -15,18 +15,17 @@
 
 #include <gtest/gtest.h>
 
-#include <coordinator/coordinator.hpp>
-#include <coordinator/coordinator_client.hpp>
-#include <coordinator/hybrid_logical_clock.hpp>
-#include <coordinator/shard_map.hpp>
-#include <io/local_transport/local_system.hpp>
-#include <io/local_transport/local_transport.hpp>
-#include <io/rsm/rsm_client.hpp>
-#include <io/transport.hpp>
-#include <machine_manager/machine_config.hpp>
-#include <machine_manager/machine_manager.hpp>
-#include <query/v2/requests.hpp>
+#include "coordinator/coordinator.hpp"
+#include "coordinator/coordinator_client.hpp"
+#include "coordinator/hybrid_logical_clock.hpp"
+#include "coordinator/shard_map.hpp"
+#include "io/local_transport/local_system.hpp"
+#include "io/local_transport/local_transport.hpp"
 #include "io/rsm/rsm_client.hpp"
+#include "io/transport.hpp"
+#include "machine_manager/machine_config.hpp"
+#include "machine_manager/machine_manager.hpp"
+#include "query/v2/requests.hpp"
 #include "storage/v3/id_types.hpp"
 #include "storage/v3/schemas.hpp"
 

tests/unit/protobuf_transport.cpp

@@ -0,0 +1,85 @@
+// 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.
+
+#include <chrono>
+#include <limits>
+#include <string>
+#include <thread>
+
+#include <gtest/gtest.h>
+
+#include "io/address.hpp"
+#include "io/protobuf_transport/protobuf_transport.hpp"
+#include "io/transport.hpp"
+#include "protobuf/messages.pb.cc"
+#include "protobuf/messages.pb.h"
+
+namespace memgraph::io::tests {
+
+using memgraph::io::protobuf_transport::ProtobufTransport;
+using MgAddress = memgraph::io::Address;
+using PbAddress = memgraph::protobuf::Address;
+using memgraph::protobuf::TestRequest;
+using memgraph::protobuf::UberMessage;
+
+TEST(ProtobufTransport, Echo) {
+  uint16_t cli_port = 6000;
+  uint16_t srv_port = 7000;
+
+  MgAddress cli_addr = MgAddress::TestAddress(cli_port);
+  MgAddress srv_addr = MgAddress::TestAddress(srv_port);
+
+  ProtobufTransport cli_pt{cli_port};
+  ProtobufTransport srv_pt{srv_port};
+
+  Io<ProtobufTransport> cli_io{cli_pt, cli_addr};
+  Io<ProtobufTransport> srv_io{srv_pt, srv_addr};
+
+  auto response_result_future = cli_io.Request<UberMessage, UberMessage>(srv_addr, UberMessage{});
+
+  auto request_result = srv_io.Receive<UberMessage>();
+
+  auto request_envelope = request_result.GetValue();
+
+  UberMessage request = std::get<UberMessage>(request_envelope.message);
+
+  // send it back as an echo
+  srv_io.Send(request_envelope.from_address, request_envelope.request_id, request);
+
+  // client receives it
+  auto response_result = std::move(response_result_future).Wait();
+  auto response_envelope = response_result.GetValue();
+  UberMessage response = response_envelope.message;
+
+  PbAddress to_addr;
+  to_addr.set_last_known_port(1);
+
+  PbAddress from_addr;
+  to_addr.set_last_known_port(2);
+
+  auto req = new TestRequest{};
+  req->set_content("ping");
+
+  UberMessage um;
+  um.set_request_id(1);
+  um.set_allocated_test_request(req);
+
+  std::string out;
+
+  bool success = um.SerializeToString(&out);
+
+  MG_ASSERT(success);
+
+  TestRequest rt;
+  rt.ParseFromString(out);
+}
+
+}  // namespace memgraph::io::tests