notify file announcement to sync layer only if shard config matches (#174)
Some checks are pending
abi-consistent-check / build-and-compare (push) Waiting to run
code-coverage / unittest-cov (push) Waiting to run
rust / check (push) Waiting to run
rust / test (push) Waiting to run
rust / lints (push) Waiting to run
functional-test / test (push) Waiting to run

* notify file announcement to sync layer only if shard config matches

* use FromStr trait
This commit is contained in:
Bo QIU 2024-08-29 18:35:35 +08:00 committed by GitHub
parent 75531a5878
commit 6685757a03
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 141 additions and 55 deletions

View File

@ -681,6 +681,12 @@ impl Libp2pEventHandler {
return MessageAcceptance::Reject; return MessageAcceptance::Reject;
} }
// verify announced shard config
let announced_shard_config = match ShardConfig::new(msg.shard_id, msg.num_shard) {
Ok(v) => v,
Err(_) => return MessageAcceptance::Reject,
};
// propagate gossip to peers // propagate gossip to peers
let d = duration_since( let d = duration_since(
msg.resend_timestamp, msg.resend_timestamp,
@ -692,13 +698,16 @@ impl Libp2pEventHandler {
return MessageAcceptance::Ignore; return MessageAcceptance::Ignore;
} }
// notify sync layer // notify sync layer if shard config matches
for tx_id in msg.tx_ids.iter() { let my_shard_config = self.store.get_store().flow().get_shard_config();
self.send_to_sync(SyncMessage::AnnounceFileGossip { if my_shard_config.intersect(&announced_shard_config) {
tx_id: *tx_id, for tx_id in msg.tx_ids.iter() {
peer_id: msg.peer_id.clone().into(), self.send_to_sync(SyncMessage::AnnounceFileGossip {
addr: addr.clone(), tx_id: *tx_id,
}); peer_id: msg.peer_id.clone().into(),
addr: addr.clone(),
});
}
} }
// insert message to cache // insert message to cache

View File

@ -253,6 +253,6 @@ impl ZgsConfig {
} }
fn shard_config(&self) -> Result<ShardConfig, String> { fn shard_config(&self) -> Result<ShardConfig, String> {
ShardConfig::new(&self.shard_position) self.shard_position.clone().try_into()
} }
} }

View File

@ -1,6 +1,6 @@
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use ssz_derive::{Decode, Encode}; use ssz_derive::{Decode, Encode};
use std::{cell::RefCell, path::PathBuf, rc::Rc}; use std::{cell::RefCell, path::PathBuf, rc::Rc, str::FromStr};
pub const SHARD_CONFIG_KEY: &str = "shard_config"; pub const SHARD_CONFIG_KEY: &str = "shard_config";
@ -25,51 +25,11 @@ impl Default for ShardConfig {
} }
} }
impl ShardConfig { impl FromStr for ShardConfig {
pub fn new(shard_position: &Option<String>) -> Result<Self, String> { type Err = String;
let (id, num) = if let Some(position) = shard_position {
Self::parse_position(position)?
} else {
(0, 1)
};
if id >= num { fn from_str(s: &str) -> Result<Self, Self::Err> {
return Err(format!( let parts: Vec<&str> = s.trim().split('/').map(|s| s.trim()).collect();
"Incorrect shard_id: expected [0, {}), actual {}",
num, id
));
}
if !num.is_power_of_two() {
return Err(format!(
"Incorrect shard group bytes: {}, should be power of two",
num
));
}
Ok(ShardConfig {
shard_id: id,
num_shard: num,
})
}
pub fn miner_shard_mask(&self) -> u64 {
!(self.num_shard - 1) as u64
}
pub fn miner_shard_id(&self) -> u64 {
self.shard_id as u64
}
pub fn is_valid(&self) -> bool {
self.num_shard > 0 && self.num_shard.is_power_of_two() && self.shard_id < self.num_shard
}
pub fn in_range(&self, segment_index: u64) -> bool {
segment_index as usize % self.num_shard == self.shard_id
}
pub fn parse_position(input: &str) -> Result<(usize, usize), String> {
let parts: Vec<&str> = input.trim().split('/').map(|s| s.trim()).collect();
if parts.len() != 2 { if parts.len() != 2 {
return Err("Incorrect format, expected like: '0 / 8'".into()); return Err("Incorrect format, expected like: '0 / 8'".into());
@ -82,7 +42,66 @@ impl ShardConfig {
.parse::<usize>() .parse::<usize>()
.map_err(|e| format!("Cannot parse shard position {:?}", e))?; .map_err(|e| format!("Cannot parse shard position {:?}", e))?;
Ok((numerator, denominator)) Self::new(numerator, denominator)
}
}
impl TryFrom<Option<String>> for ShardConfig {
type Error = String;
fn try_from(value: Option<String>) -> Result<Self, Self::Error> {
if let Some(position) = value {
Self::from_str(&position)
} else {
Ok(Self::default())
}
}
}
impl ShardConfig {
pub fn new(id: usize, num: usize) -> Result<Self, String> {
let config = ShardConfig {
shard_id: id,
num_shard: num,
};
config.validate()?;
Ok(config)
}
pub fn miner_shard_mask(&self) -> u64 {
!(self.num_shard - 1) as u64
}
pub fn miner_shard_id(&self) -> u64 {
self.shard_id as u64
}
pub fn validate(&self) -> Result<(), String> {
if self.shard_id >= self.num_shard {
return Err(format!(
"Incorrect shard_id: expected [0, {}), actual {}",
self.num_shard, self.shard_id
));
}
if self.num_shard == 0 {
return Err("Shard num is 0".into());
}
if !self.num_shard.is_power_of_two() {
return Err(format!(
"Incorrect shard group bytes: {}, should be power of two",
self.num_shard
));
}
Ok(())
}
pub fn in_range(&self, segment_index: u64) -> bool {
segment_index as usize % self.num_shard == self.shard_id
} }
pub fn next_segment_index(&self, current: usize, start_index: usize) -> usize { pub fn next_segment_index(&self, current: usize, start_index: usize) -> usize {
@ -90,6 +109,30 @@ impl ShardConfig {
let shift = (start_index + current + self.num_shard - self.shard_id) % self.num_shard; let shift = (start_index + current + self.num_shard - self.shard_id) % self.num_shard;
current + self.num_shard - shift current + self.num_shard - shift
} }
/// Whether `self` intersect with the `other` shard config.
pub fn intersect(&self, other: &ShardConfig) -> bool {
let ShardConfig {
num_shard: mut left_num_shard,
shard_id: mut left_shard_id,
} = self;
let ShardConfig {
num_shard: mut right_num_shard,
shard_id: mut right_shard_id,
} = other;
while left_num_shard != right_num_shard {
if left_num_shard < right_num_shard {
right_num_shard /= 2;
right_shard_id /= 2;
} else {
left_num_shard /= 2;
left_shard_id /= 2;
}
}
left_shard_id == right_shard_id
}
} }
struct ShardSegmentTreeNode { struct ShardSegmentTreeNode {
@ -146,7 +189,7 @@ impl ShardSegmentTreeNode {
pub fn all_shards_available(shard_configs: Vec<ShardConfig>) -> bool { pub fn all_shards_available(shard_configs: Vec<ShardConfig>) -> bool {
let mut root = ShardSegmentTreeNode::new(1); let mut root = ShardSegmentTreeNode::new(1);
for shard_config in shard_configs.iter() { for shard_config in shard_configs.iter() {
if !shard_config.is_valid() { if shard_config.validate().is_err() {
continue; continue;
} }
root.insert(shard_config.num_shard, shard_config.shard_id); root.insert(shard_config.num_shard, shard_config.shard_id);
@ -163,6 +206,10 @@ mod tests {
use super::ShardConfig; use super::ShardConfig;
fn new_config(id: usize, num: usize) -> ShardConfig {
ShardConfig::new(id, num).unwrap()
}
#[test] #[test]
fn test_all_shards_available() { fn test_all_shards_available() {
assert!(all_shards_available(vec![ assert!(all_shards_available(vec![
@ -210,4 +257,34 @@ mod tests {
}, },
])); ]));
} }
#[test]
fn test_shard_intersect() {
// 1 shard
assert_eq!(new_config(0, 1).intersect(&new_config(0, 1)), true);
// either is 1 shard
assert_eq!(new_config(0, 1).intersect(&new_config(0, 2)), true);
assert_eq!(new_config(0, 1).intersect(&new_config(1, 2)), true);
assert_eq!(new_config(0, 2).intersect(&new_config(0, 1)), true);
assert_eq!(new_config(1, 2).intersect(&new_config(0, 1)), true);
// same shards
assert_eq!(new_config(1, 4).intersect(&new_config(0, 4)), false);
assert_eq!(new_config(1, 4).intersect(&new_config(1, 4)), true);
assert_eq!(new_config(1, 4).intersect(&new_config(2, 4)), false);
assert_eq!(new_config(1, 4).intersect(&new_config(3, 4)), false);
// left shards is less
assert_eq!(new_config(1, 2).intersect(&new_config(0, 4)), false);
assert_eq!(new_config(1, 2).intersect(&new_config(1, 4)), false);
assert_eq!(new_config(1, 2).intersect(&new_config(2, 4)), true);
assert_eq!(new_config(1, 2).intersect(&new_config(3, 4)), true);
// right shards is less
assert_eq!(new_config(1, 4).intersect(&new_config(0, 2)), true);
assert_eq!(new_config(1, 4).intersect(&new_config(1, 2)), false);
assert_eq!(new_config(2, 4).intersect(&new_config(0, 2)), false);
assert_eq!(new_config(2, 4).intersect(&new_config(1, 2)), true);
}
} }