Compare commits

...

1 Commits

Author SHA1 Message Date
Tyler Neely
fe756958b0 Check-in local shard manager sender code 2023-01-30 11:35:55 +00:00
2 changed files with 33 additions and 2 deletions

View File

@@ -49,10 +49,10 @@ template <Message M>
using ResponseResult = BasicResult<TimedOut, ResponseEnvelope<M>>;
template <Message M>
using ResponseFuture = memgraph::io::Future<ResponseResult<M>>;
using ResponseFuture = Future<ResponseResult<M>>;
template <Message M>
using ResponsePromise = memgraph::io::Promise<ResponseResult<M>>;
using ResponsePromise = Promise<ResponseResult<M>>;
template <Message... Ms>
struct RequestEnvelope {
@@ -65,6 +65,19 @@ struct RequestEnvelope {
template <Message... Ms>
using RequestResult = BasicResult<TimedOut, RequestEnvelope<Ms...>>;
/// This is a concrete type that allows one message type to be
/// sent to a single address. Initially intended to be used by the
/// Shard to send messages to the local ShardManager.
template <Message M>
class Sender {
std::function<void(M)> sender_;
public:
Sender(std::function<void(M)> sender) : sender_(sender) {}
void Send(M message) { sender_(message); }
};
template <typename I>
class Io {
I implementation_;
@@ -173,5 +186,16 @@ class Io {
}
LatencyHistogramSummaries ResponseLatencies() { return implementation_.ResponseLatencies(); }
template <Message M>
Sender<M> GetSender(Address address) {
Io<I> io_copy = Io(implementation_, address_);
std::function<void(M)> sender = [address, io_copy](M message) mutable {
io_copy.template Send<M>(address, 0, message);
};
return Sender{sender};
}
};
}; // namespace memgraph::io

View File

@@ -196,6 +196,13 @@ class ShardWorker {
// TODO(tyler) get peers from Coordinator in HeartbeatResponse
std::vector<Address> rsm_peers = {};
Address local_shard_manager_address = io_.GetAddress().ForkLocalShardManager();
io::Sender<io::messages::ShardManagerMessages> local_shard_manager_sender =
io_.template GetSender<io::messages::ShardManagerMessages>(local_shard_manager_address);
// TODO(tyler) pass this local_shard_manager_sender to the Shard so that it can communicate back to the local
// manager from split code
std::unique_ptr<Shard> shard = std::make_unique<Shard>(to_init.label_id, to_init.min_key, to_init.max_key,
to_init.schema, to_init.config, to_init.id_to_names);