Skip to content

Commit

Permalink
Remove unnecessary copies into OArray (#2777)
Browse files Browse the repository at this point in the history
  • Loading branch information
eddyashton authored Jul 7, 2021
1 parent eb3f966 commit d4c1509
Show file tree
Hide file tree
Showing 9 changed files with 39 additions and 52 deletions.
39 changes: 12 additions & 27 deletions src/consensus/aft/async_execution.h
Original file line number Diff line number Diff line change
Expand Up @@ -53,28 +53,23 @@ namespace aft
const ccf::NodeId& from_,
AppendEntries&& hdr_,
const uint8_t* data_,
size_t size_,
OArray&& oarray_) :
size_t size_) :
store(store_),
from(from_),
hdr(std::move(hdr_)),
data(data_),
size(size_),
oarray(std::move(oarray_))
body(data_, data_ + size_)
{}

void execute() override
{
store.recv_append_entries(from, hdr, data, size);
store.recv_append_entries(from, hdr, body.data(), body.size());
}

private:
AbstractConsensusCallback& store;
ccf::NodeId from;
AppendEntries hdr;
const uint8_t* data;
size_t size;
OArray oarray;
std::vector<uint8_t> body;
};

class AppendEntryResponseCallback : public AbstractMsgCallback
Expand Down Expand Up @@ -223,28 +218,23 @@ namespace aft
const ccf::NodeId& from_,
RequestViewChangeMsg&& hdr_,
const uint8_t* data_,
size_t size_,
OArray&& oarray_) :
size_t size_) :
store(store_),
from(from_),
hdr(std::move(hdr_)),
data(data_),
size(size_),
oarray(std::move(oarray_))
body(data_, data_ + size_)
{}

void execute() override
{
store.recv_view_change(from, hdr, data, size);
store.recv_view_change(from, hdr, body.data(), body.size());
}

private:
AbstractConsensusCallback& store;
ccf::NodeId from;
RequestViewChangeMsg hdr;
const uint8_t* data;
size_t size;
OArray oarray;
std::vector<uint8_t> body;
};

class SkipViewCallback : public AbstractMsgCallback
Expand Down Expand Up @@ -278,27 +268,22 @@ namespace aft
const ccf::NodeId& from_,
ViewChangeEvidenceMsg&& hdr_,
const uint8_t* data_,
size_t size_,
OArray&& oarray_) :
size_t size_) :
store(store_),
from(from_),
hdr(std::move(hdr_)),
data(data_),
size(size_),
oarray(std::move(oarray_))
body(data_, data_ + size_)
{}

void execute() override
{
store.recv_view_change_evidence(from, hdr, data, size);
store.recv_view_change_evidence(from, hdr, body.data(), body.size());
}

private:
AbstractConsensusCallback& store;
ccf::NodeId from;
ViewChangeEvidenceMsg hdr;
const uint8_t* data;
size_t size;
OArray oarray;
std::vector<uint8_t> body;
};
}
13 changes: 3 additions & 10 deletions src/consensus/aft/raft.h
Original file line number Diff line number Diff line change
Expand Up @@ -711,15 +711,8 @@ namespace aft
}

void recv_message(const ccf::NodeId& from, const uint8_t* data, size_t size)
{
recv_message(from, OArray({data, data + size}));
}

void recv_message(const ccf::NodeId& from, OArray&& d)
{
std::unique_ptr<AbstractMsgCallback> aee;
const uint8_t* data = d.data();
size_t size = d.size();
RaftMsgType type = serialized::peek<RaftMsgType>(data, size);

try
Expand All @@ -732,7 +725,7 @@ namespace aft
channels->template recv_authenticated<AppendEntries>(
from, data, size);
aee = std::make_unique<AppendEntryCallback>(
*this, from, std::move(r), data, size, std::move(d));
*this, from, std::move(r), data, size);
break;
}
case raft_append_entries_response:
Expand Down Expand Up @@ -800,7 +793,7 @@ namespace aft
->template recv_authenticated_with_load<RequestViewChangeMsg>(
from, data, size);
aee = std::make_unique<ViewChangeCallback>(
*this, from, std::move(r), data, size, std::move(d));
*this, from, std::move(r), data, size);
break;
}

Expand All @@ -821,7 +814,7 @@ namespace aft
from, data, size);

aee = std::make_unique<ViewChangeEvidenceCallback>(
*this, from, std::move(r), data, size, std::move(d));
*this, from, std::move(r), data, size);
break;
}

Expand Down
5 changes: 3 additions & 2 deletions src/consensus/aft/raft_consensus.h
Original file line number Diff line number Diff line change
Expand Up @@ -121,9 +121,10 @@ namespace aft
return aft->active_nodes();
}

void recv_message(const ccf::NodeId& from, OArray&& data) override
void recv_message(
const ccf::NodeId& from, const uint8_t* data, size_t size) override
{
return aft->recv_message(from, std::move(data));
return aft->recv_message(from, data, size);
}

void add_configuration(
Expand Down
4 changes: 3 additions & 1 deletion src/consensus/aft/test/logging_stub.h
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,9 @@ namespace aft
return true;
}

void recv_message(const ccf::NodeId& from, OArray&& oa) override {}
void recv_message(
const ccf::NodeId& from, const uint8_t* data, size_t size) override
{}

void initialize(
const ccf::NodeId& self_id,
Expand Down
3 changes: 2 additions & 1 deletion src/kv/kv_types.h
Original file line number Diff line number Diff line change
Expand Up @@ -409,7 +409,8 @@ namespace kv
virtual bool view_change_in_progress() = 0;
virtual std::set<NodeId> active_nodes() = 0;

virtual void recv_message(const NodeId& from, OArray&& oa) = 0;
virtual void recv_message(
const NodeId& from, const uint8_t* data, size_t size) = 0;

virtual bool on_request(const TxHistory::RequestCallbackArgs&)
{
Expand Down
4 changes: 3 additions & 1 deletion src/kv/test/stub_consensus.h
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,9 @@ namespace kv::test
view_history.initialise(view_history_);
}

void recv_message(const NodeId& from, OArray&& oa) override {}
void recv_message(
const NodeId& from, const uint8_t* data, size_t size) override
{}

void add_configuration(
ccf::SeqNo seqno,
Expand Down
12 changes: 7 additions & 5 deletions src/node/node_state.h
Original file line number Diff line number Diff line change
Expand Up @@ -1449,21 +1449,23 @@ namespace ccf
return;
}

OArray oa(std::move(data));
const uint8_t* payload_data = data.data();
size_t payload_size = data.size();

NodeMsgType msg_type =
serialized::overlay<NodeMsgType>(oa.data(), oa.size());
NodeId from = serialized::read<NodeId::Value>(oa.data(), oa.size());
serialized::overlay<NodeMsgType>(payload_data, payload_size);
NodeId from = serialized::read<NodeId::Value>(payload_data, payload_size);

switch (msg_type)
{
case channel_msg:
{
n2n_channels->recv_message(from, std::move(oa));
n2n_channels->recv_message(from, payload_data, payload_size);
break;
}
case consensus_msg:
{
consensus->recv_message(from, std::move(oa));
consensus->recv_message(from, payload_data, payload_size);
break;
}

Expand Down
8 changes: 4 additions & 4 deletions src/node/node_to_node.h
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,8 @@ namespace ccf
virtual bool recv_authenticated(
const NodeId& from, CBuffer cb, const uint8_t*& data, size_t& size) = 0;

virtual void recv_message(const NodeId& from, OArray&& oa) = 0;
virtual void recv_message(
const NodeId& from, const uint8_t* data, size_t size) = 0;

virtual void initialize(
const NodeId& self_id,
Expand Down Expand Up @@ -294,12 +295,11 @@ namespace ccf
}
}

void recv_message(const NodeId& from, OArray&& oa) override
void recv_message(
const NodeId& from, const uint8_t* data, size_t size) override
{
try
{
const uint8_t* data = oa.data();
size_t size = oa.size();
auto chmsg = serialized::read<ChannelMsg>(data, size);
switch (chmsg)
{
Expand Down
3 changes: 2 additions & 1 deletion src/node/test/channels.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -739,7 +739,8 @@ TEST_CASE("Full NodeToNode test")
{
case NodeMsgType::channel_msg:
{
n2n.recv_message(msg.from, msg.data());
const auto msg_body = msg.data();
n2n.recv_message(msg.from, msg_body.data(), msg_body.size());

auto d = msg.data();
const uint8_t* data = d.data();
Expand Down

0 comments on commit d4c1509

Please sign in to comment.