Skip to content
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

parallel psw handling in actor #12656

Draft
wants to merge 5 commits into
base: master
Choose a base branch
from
+191 −83
Draft

parallel psw handling in actor #12656

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
80a6dff
spawn thread in handle_partial_encoded_state_witness
stedfn Dec 18, 2024
3613619
remove file
stedfn Dec 18, 2024
c12d23b
clippy
stedfn Dec 18, 2024
1de100c
modify handle_partial_encoded_state_witness_forward
stedfn Dec 19, 2024
3f63d03
.
stedfn Dec 20, 2024
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
132 changes: 109 additions & 23 deletions chain/client/src/stateless_validation/partial_witness/partial_witness_actor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ pub struct PartialWitnessActor {
epoch_manager: Arc<dyn EpochManagerAdapter>,
runtime: Arc<dyn RuntimeAdapter>,
/// Tracks the parts of the state witness sent from chunk producers to chunk validators.
partial_witness_tracker: PartialEncodedStateWitnessTracker,
partial_witness_tracker: Arc<PartialEncodedStateWitnessTracker>,
partial_deploys_tracker: PartialEncodedContractDeploysTracker,
/// Tracks a collection of state witnesses sent from chunk producers to chunk validators.
state_witness_tracker: ChunkStateWitnessTracker,
Expand All @@ -75,6 +75,7 @@ pub struct PartialWitnessActor {
/// Same as above for contract deploys.
contract_deploys_encoders: ReedSolomonEncoderCache,
compile_contracts_spawner: Arc<dyn AsyncComputationSpawner>,
partial_witness_spawner: Arc<dyn AsyncComputationSpawner>,
/// AccountId in the key corresponds to the requester (chunk validator).
processed_contract_code_requests: LruCache<(ChunkProductionKey, AccountId), ()>,
}
Expand Down Expand Up @@ -166,9 +167,10 @@ impl PartialWitnessActor {
epoch_manager: Arc<dyn EpochManagerAdapter>,
runtime: Arc<dyn RuntimeAdapter>,
compile_contracts_spawner: Arc<dyn AsyncComputationSpawner>,
partial_witness_spawner: Arc<dyn AsyncComputationSpawner>,
) -> Self {
let partial_witness_tracker =
PartialEncodedStateWitnessTracker::new(client_sender, epoch_manager.clone());
Arc::new(PartialEncodedStateWitnessTracker::new(client_sender, epoch_manager.clone()));
Self {
network_adapter,
my_signer,
Expand All @@ -182,6 +184,7 @@ impl PartialWitnessActor {
CONTRACT_DEPLOYS_RATIO_DATA_PARTS,
),
compile_contracts_spawner,
partial_witness_spawner,
processed_contract_code_requests: LruCache::new(
NonZeroUsize::new(PROCESSED_CONTRACT_CODE_REQUESTS_CACHE_SIZE).unwrap(),
),
Expand Down Expand Up @@ -368,7 +371,7 @@ impl PartialWitnessActor {
}

/// Sends the witness part to the chunk validators, except the chunk producer that generated the witness part.
fn forward_state_witness_part(
fn _forward_state_witness_part(
&self,
partial_witness: PartialEncodedStateWitness,
) -> Result<(), Error> {
Expand Down Expand Up @@ -403,17 +406,60 @@ impl PartialWitnessActor {
partial_witness: PartialEncodedStateWitness,
) -> Result<(), Error> {
tracing::debug!(target: "client", ?partial_witness, "Receive PartialEncodedStateWitnessMessage");

let signer = self.my_validator_signer()?;
// Validate the partial encoded state witness and forward the part to all the chunk validators.
if validate_partial_encoded_state_witness(
self.epoch_manager.as_ref(),
&partial_witness,
&signer,
self.runtime.store(),
)? {
self.forward_state_witness_part(partial_witness)?;
}
let epoch_manager = self.epoch_manager.clone();
let runtime_adapter = self.runtime.clone();

let ChunkProductionKey { shard_id, epoch_id, height_created } =
partial_witness.chunk_production_key();

let chunk_producer = self
.epoch_manager
.get_chunk_producer_info(&ChunkProductionKey { epoch_id, height_created, shard_id })?
.take_account_id();

// Forward witness part to chunk validators except the validator that produced the chunk and witness.
let target_chunk_validators = self
.epoch_manager
.get_chunk_validator_assignments(&epoch_id, shard_id, height_created)?
.ordered_chunk_validators()
.into_iter()
.filter(|validator| validator != &chunk_producer)
.collect();

let network_adapter = self.network_adapter.clone();

self.partial_witness_spawner.spawn("handle_partial_encoded_state_witness", move || {
// Validate the partial encoded state witness and forward the part to all the chunk validators.
let validation = validate_partial_encoded_state_witness(
epoch_manager.as_ref(),
&partial_witness,
&signer,
runtime_adapter.store(),
);
match validation {
Ok(true) => {
forward_state_witness_part_v2(
partial_witness,
target_chunk_validators,
network_adapter,
);
}
Ok(false) => {
tracing::warn!(
target: "client",
"Received partial encoded state witness that is not valid"
);
}
Err(err) => {
tracing::warn!(
target: "client",
"Encountered error during validation: {}",
err
);
}
}
});

Ok(())
}
Expand All @@ -426,15 +472,41 @@ impl PartialWitnessActor {
tracing::debug!(target: "client", ?partial_witness, "Receive PartialEncodedStateWitnessForwardMessage");

let signer = self.my_validator_signer()?;
// Validate the partial encoded state witness and store the partial encoded state witness.
if validate_partial_encoded_state_witness(
self.epoch_manager.as_ref(),
&partial_witness,
&signer,
self.runtime.store(),
)? {
self.partial_witness_tracker.store_partial_encoded_state_witness(partial_witness)?;
}
let partial_witness_tracker = self.partial_witness_tracker.clone();
let epoch_manager = self.epoch_manager.clone();
let runtime_adapter = self.runtime.clone();
self.partial_witness_spawner.spawn(
"handle_partial_encoded_state_witness_forward",
move || {
// Validate the partial encoded state witness and store the partial encoded state witness.
let validation = validate_partial_encoded_state_witness(
epoch_manager.as_ref(),
&partial_witness,
&signer,
runtime_adapter.store(),
);
match validation {
Ok(true) => {
partial_witness_tracker
.store_partial_encoded_state_witness(partial_witness)
.unwrap();
}
Ok(false) => {
tracing::warn!(
target: "client",
"Received partial encoded state witness that is not valid"
);
}
Err(err) => {
tracing::warn!(
target: "client",
"Encountered error during validation: {}",
err
);
}
}
},
);

Ok(())
}
Expand Down Expand Up @@ -598,7 +670,7 @@ impl PartialWitnessActor {

/// Sends the contract accesses to the same chunk validators
/// (except for the chunk producers that track the same shard),
/// which will receive the state witness for the new chunk.
/// which will receive the state witness for the new chunk.
fn send_contract_accesses_to_chunk_validators(
&self,
key: ChunkProductionKey,
Expand Down Expand Up @@ -799,3 +871,17 @@ fn contracts_cache_contains_contract(
let cache_key = get_contract_cache_key(contract_hash.0, &runtime_config.wasm_config);
cache.memory_cache().contains(cache_key) || cache.has(&cache_key).is_ok_and(|has| has)
}

/// Sends the witness part to the chunk validators, except the chunk producer that generated the witness part.
fn forward_state_witness_part_v2(
partial_witness: PartialEncodedStateWitness,
target_chunk_validators: Vec<AccountId>,
network_adapter: PeerManagerAdapter,
) {
network_adapter.send(PeerManagerMessageRequest::NetworkRequests(
NetworkRequests::PartialEncodedStateWitnessForward(
target_chunk_validators,
partial_witness,
),
));
}
Loading
Loading