Skip to content

fix: reduce hot-path allocation, cloning, and lock contention #524

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
varex83agent wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: reduce hot-path allocation, cloning, and lock contention #524

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion crates/app/src/monitoringapi/checker.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -307,7 +307,7 @@ pub fn quorum_peers_connected(p2p_context: &P2PContext) -> bool {
let connected = known_peers
.iter()
.filter(|peer_id| **peer_id != local_peer_id)
.filter(|peer_id| !peer_store.connections_to_peer(peer_id).is_empty())
.filter(|peer_id| peer_store.has_connection(peer_id))
.count();

connected >= required
Expand Down
32 changes: 20 additions & 12 deletions crates/consensus/src/qbft/component.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,16 +186,19 @@ pub enum Error {
}

/// Canonicalizes inbound `Any` values into the hash map used by QBFT messages.
pub(crate) fn values_by_hash(values: &[Any]) -> Result<ValueMap> {
///
/// Consumes `values` and moves each entry into the map instead of cloning,
/// since the caller owns the values and discards the source vector afterwards.
pub(crate) fn values_by_hash(values: Vec<Any>) -> Result<ValueMap> {
let mut out = ValueMap::new();

for value in values {
let decoded = decode_supported_any(value)?;
let decoded = decode_supported_any(&value)?;
let hash = match decoded {
DecodedValue::UnsignedDataSet(inner) => msg::hash_proto(&inner)?,
DecodedValue::PriorityResult(inner) => msg::hash_proto(&inner)?,
};
out.insert(hash, value.clone());
out.insert(hash, value);
}

Ok(out)
Expand Down Expand Up @@ -343,13 +346,14 @@ impl Consensus {
/// Validates, wraps, and queues an inbound QBFT consensus message.
pub async fn handle(
&self,
pb_msg: pbconsensus::QbftConsensusMsg,
mut pb_msg: pbconsensus::QbftConsensusMsg,
ct: &CancellationToken,
) -> Result<()> {
let msg = pb_msg.msg.as_ref().ok_or(Error::InvalidConsensusMessage)?;
// Verify the inner message and derive its duty (borrow only).
let inner = pb_msg.msg.as_ref().ok_or(Error::InvalidConsensusMessage)?;

self.verify_msg(msg)?;
let duty = duty_from_msg(msg)?;
self.verify_msg(inner)?;
let duty = duty_from_msg(inner)?;

if !(self.duty_gater)(&duty) {
return Err(Error::InvalidDuty);
Expand All @@ -366,8 +370,12 @@ impl Consensus {
}
}

let values = values_by_hash(&pb_msg.values)?;
let wrapped = msg::Msg::new(msg.clone(), pb_msg.justification.clone(), Arc::new(values))?;
// Move the validated parts out of `pb_msg` (it is dropped after this),
// avoiding a clone of the message, justifications, and values.
let msg = pb_msg.msg.take().ok_or(Error::InvalidConsensusMessage)?;
let justification = std::mem::take(&mut pb_msg.justification);
let values = values_by_hash(std::mem::take(&mut pb_msg.values))?;
let wrapped = msg::Msg::new(msg, justification, Arc::new(values))?;

if ct.is_cancelled() {
return Err(Error::ReceiveCancelledDuringVerification);
Expand Down Expand Up @@ -932,7 +940,7 @@ pub(crate) mod tests {

#[test]
fn values_by_hash_rejects_invalid_type_url() {
let err = values_by_hash(&[Any {
let err = values_by_hash(vec![Any {
type_url: "type.googleapis.com/unknown.Type".to_string(),
value: vec![],
}])
Expand All @@ -943,7 +951,7 @@ pub(crate) mod tests {

#[test]
fn values_by_hash_rejects_malformed_any_value() {
let err = values_by_hash(&[Any {
let err = values_by_hash(vec![Any {
type_url: pbcore::UnsignedDataSet::type_url(),
value: b"not-protobuf".to_vec(),
}])
Expand All @@ -955,7 +963,7 @@ pub(crate) mod tests {
#[test]
fn values_by_hash_hashes_decoded_inner_message() {
let any = unsigned_any("a", b"first");
let values = values_by_hash(std::slice::from_ref(&any)).unwrap();
let values = values_by_hash(vec![any.clone()]).unwrap();
let decoded = pbcore::UnsignedDataSet::decode(any.value.as_slice()).unwrap();
let hash = msg::hash_proto(&decoded).unwrap();

Expand Down
44 changes: 32 additions & 12 deletions crates/consensus/src/qbft/p2p.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,13 @@ impl Handle {
pub async fn broadcast(&self, msg: pbconsensus::QbftConsensusMsg) -> BroadcastResult {
let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed);
self.cmd_tx
.send(BroadcastCommand { request_id, msg })
.send(BroadcastCommand {
request_id,
// Wrap once so the fan-out shares a single allocation across all
// peers rather than deep-cloning the (potentially multi-MB)
// payload per target, matching Charon's shared-pointer broadcast.
msg: Arc::new(msg),
})
.map_err(|_| Box::new(Error::BehaviourClosed) as _)
}

Expand All @@ -200,15 +206,15 @@ impl Handle {
#[derive(Debug)]
struct BroadcastCommand {
request_id: u64,
msg: pbconsensus::QbftConsensusMsg,
msg: Arc<pbconsensus::QbftConsensusMsg>,
}

#[doc(hidden)]
#[derive(Debug)]
pub enum ToHandler {
Send {
request_id: u64,
msg: pbconsensus::QbftConsensusMsg,
msg: Arc<pbconsensus::QbftConsensusMsg>,
},
}

Expand All @@ -227,7 +233,7 @@ type ActiveFuture = futures::future::BoxFuture<'static, Option<FromHandler>>;
pub struct Handler {
consensus: Arc<Consensus>,
cancellation: CancellationToken,
pending_open: VecDeque<(u64, pbconsensus::QbftConsensusMsg)>,
pending_open: VecDeque<(u64, Arc<pbconsensus::QbftConsensusMsg>)>,
active_futures: futures::stream::FuturesUnordered<ActiveFuture>,
}

Expand Down Expand Up @@ -272,7 +278,7 @@ impl Handler {
&mut self,
mut stream: Stream,
request_id: u64,
msg: pbconsensus::QbftConsensusMsg,
msg: Arc<pbconsensus::QbftConsensusMsg>,
) {
stream.ignore_for_keep_alive();
self.active_futures.push(
Expand Down Expand Up @@ -308,7 +314,7 @@ impl ConnectionHandler for Handler {
type FromBehaviour = ToHandler;
type InboundOpenInfo = ();
type InboundProtocol = ReadyUpgrade<StreamProtocol>;
type OutboundOpenInfo = (u64, pbconsensus::QbftConsensusMsg);
type OutboundOpenInfo = (u64, Arc<pbconsensus::QbftConsensusMsg>);
type OutboundProtocol = ReadyUpgrade<StreamProtocol>;
type ToBehaviour = FromHandler;

Expand Down Expand Up @@ -455,7 +461,7 @@ where
#[derive(Debug)]
struct PendingSend {
request_id: u64,
msg: pbconsensus::QbftConsensusMsg,
msg: Arc<pbconsensus::QbftConsensusMsg>,
}

/// libp2p behaviour for QBFT consensus messages.
Expand Down Expand Up @@ -506,12 +512,10 @@ impl Behaviour {

/// Returns whether the peer store has any live connection for the peer.
fn is_connected(&self, peer_id: &PeerId) -> bool {
!self
.config
self.config
.p2p_context
.peer_store_lock()
.connections_to_peer(peer_id)
.is_empty()
.has_connection(peer_id)
}

/// Drains outbound broadcast commands queued through the public handle.
Expand Down Expand Up @@ -539,7 +543,8 @@ impl Behaviour {
peer_id,
PendingSend {
request_id: command.request_id,
msg: command.msg.clone(),
// Cheap refcount bump; the payload buffer is shared.
msg: Arc::clone(&command.msg),
},
);
}
Expand Down Expand Up @@ -945,6 +950,21 @@ mod tests {
assert!(targets.contains(&peer_ids[0]));
assert!(targets.contains(&peer_ids[2]));
assert!(!targets.contains(&local_peer_id));

// The fan-out must share one `Arc<QbftConsensusMsg>` across all targets
// rather than deep-cloning the payload per peer.
let payloads = events
.iter()
.filter_map(|event| match event {
ToSwarm::NotifyHandler {
event: Either::Left(ToHandler::Send { msg, .. }),
..
} => Some(msg.clone()),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(payloads.len(), 2);
assert!(Arc::ptr_eq(&payloads[0], &payloads[1]));
Ok(())
}

Expand Down
2 changes: 1 addition & 1 deletion crates/consensus/src/qbft/qbft_run_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -422,7 +422,7 @@ async fn replay_sniffed_instance_decides(instance: pbconsensus::SniffedConsensus
for sniffed in instance.msgs {
let outer = sniffed.msg.expect("sniffed entry has outer message");
let raw = outer.msg.expect("sniffed outer message has inner message");
let values = component::values_by_hash(&outer.values).expect("sniffed values decode");
let values = component::values_by_hash(outer.values).expect("sniffed values decode");
let wrapped = msg::Msg::new(raw, outer.justification, Arc::new(values))
.expect("sniffed message wraps");
let wrapped: qbft::Msg<ConsensusQbftTypes> = Arc::new(wrapped);
Expand Down
67 changes: 46 additions & 21 deletions crates/core/src/parsigdb/memory.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -258,7 +258,7 @@ impl MemDB {
let mut output: HashMap<PubKey, Vec<ParSignedData>> = HashMap::new();

for (pub_key, par_signed) in signed_data.inner().iter() {
let sigs = self
let outcome = self
.store(
Key {
duty: duty.clone(),
Expand All @@ -268,19 +268,16 @@ impl MemDB {
)
.await?;

let Some(sigs) = sigs else {
debug!("Ignoring duplicate partial signature");

continue;
};

let psigs = get_threshold_matching(&duty.duty_type, &sigs, self.threshold).await?;

let Some(psigs) = psigs else {
continue;
};

output.insert(*pub_key, psigs);
match outcome {
StoreOutcome::Duplicate => {
debug!("Ignoring duplicate partial signature");
continue;
}
StoreOutcome::Stored => continue,
StoreOutcome::Threshold(psigs) => {
output.insert(*pub_key, psigs);
}
}
}

if output.is_empty() {
Expand Down Expand Up @@ -330,16 +327,29 @@ impl MemDB {
}
}

async fn store(&self, k: Key, value: ParSignedData) -> Result<Option<Vec<ParSignedData>>> {
/// Stores `value` under key `k`, then evaluates the threshold under the
/// same lock against the borrowed accumulator so no full clone of the
/// growing set is produced per store.
///
/// Returns:
/// - [`StoreOutcome::Duplicate`] if `value` is a duplicate (same
/// `share_idx`, equal data) and nothing was stored.
/// - [`StoreOutcome::Stored`] if stored but the threshold is not yet met.
/// - [`StoreOutcome::Threshold`] carrying exactly the matched set forwarded
/// to threshold subscribers.
///
/// Returns `Err(ParsigDataMismatch)` on a conflicting duplicate and
/// propagates `MessageRoot` errors from the threshold check.
async fn store(&self, k: Key, value: ParSignedData) -> Result<StoreOutcome> {
let mut inner = self.inner.lock().await;

// Check if we already have an entry with this ShareIdx
if let Some(existing_entries) = inner.entries.get(&k) {
for s in existing_entries {
if s.share_idx == value.share_idx {
if s == &value {
// Duplicate, return None to indicate no new data
return Ok(None);
// Duplicate, nothing stored.
return Ok(StoreOutcome::Duplicate);
} else {
return Err(MemDBError::ParsigDataMismatch {
pubkey: k.pub_key,
Expand All @@ -354,7 +364,7 @@ impl MemDB {
.entries
.entry(k.clone())
.or_insert_with(Vec::new)
.push(value.clone());
.push(value);
inner
.keys_by_duty
.entry(k.duty.clone())
Expand All @@ -365,13 +375,28 @@ impl MemDB {
PARSIG_DB_METRICS.exit_total[&k.pub_key.to_string()].inc();
}

let result = inner.entries.get(&k).cloned().unwrap_or_default();
// Threshold check runs under the lock against the borrowed slice; only
// the matched subset (if any) is cloned out. No full-accumulator clone.
let sigs = inner.entries.get(&k).map(Vec::as_slice).unwrap_or_default();

Ok(Some(result))
match get_threshold_matching(&k.duty.duty_type, sigs, self.threshold)? {
Some(psigs) => Ok(StoreOutcome::Threshold(psigs)),
None => Ok(StoreOutcome::Stored),
}
}
}

async fn get_threshold_matching(
/// Result of [`MemDB::store`].
enum StoreOutcome {
/// `value` was a duplicate; nothing stored.
Duplicate,
/// Stored, but the threshold for the key is not yet reached.
Stored,
/// Stored and threshold reached; carries the matched set.
Threshold(Vec<ParSignedData>),
}

fn get_threshold_matching(
typ: &DutyType,
sigs: &[ParSignedData],
threshold: u64,
Expand Down
Loading
Loading