From 7bba02f3c5e7efcddb2c9f05dcfcfbc878ce42f6 Mon Sep 17 00:00:00 2001 From: Groovie | Mango <95291500+grooviegermanikus@users.noreply.github.com> Date: Thu, 22 Feb 2024 14:28:29 +0100 Subject: [PATCH] wrap producedblock struct and avoid cloning it (#336) wrap ProducedBlock in Arc --- .../block_stores/postgres/postgres_block.rs | 8 +-- .../postgres/postgres_block_store_writer.rs | 8 +-- .../tests/blockstore_integration_tests.rs | 10 +--- .../multiple_strategy_block_store_tests.rs | 8 +-- cluster-endpoints/src/grpc_subscription.rs | 7 ++- .../src/rpc_polling/poll_blocks.rs | 8 +-- core/src/structures/produced_block.rs | 59 +++++++++++++++---- services/src/data_caching_service.rs | 4 +- 8 files changed, 72 insertions(+), 40 deletions(-) diff --git a/blockstore/src/block_stores/postgres/postgres_block.rs b/blockstore/src/block_stores/postgres/postgres_block.rs index b51697eb..3d1b5812 100644 --- a/blockstore/src/block_stores/postgres/postgres_block.rs +++ b/blockstore/src/block_stores/postgres/postgres_block.rs @@ -2,7 +2,7 @@ use super::postgres_epoch::PostgresEpoch; use super::postgres_session::PostgresSession; use log::{debug, warn}; use solana_lite_rpc_core::structures::epoch::EpochRef; -use solana_lite_rpc_core::structures::produced_block::TransactionInfo; +use solana_lite_rpc_core::structures::produced_block::{ProducedBlockInner, TransactionInfo}; use solana_lite_rpc_core::{encoding::BASE64, structures::produced_block::ProducedBlock}; use solana_sdk::clock::Slot; use solana_sdk::commitment_config::CommitmentConfig; @@ -56,7 +56,7 @@ impl PostgresBlock { .map(|x| BASE64.deserialize::>(x).ok()) .unwrap_or(None); - ProducedBlock { + let inner = ProducedBlockInner { // TODO implement transactions: transaction_infos, leader_id: None, @@ -65,10 +65,10 @@ impl PostgresBlock { slot: self.slot as Slot, parent_slot: self.parent_slot as Slot, block_time: self.block_time as u64, - commitment_config, previous_blockhash: self.previous_blockhash.clone(), rewards: rewards_vec, - } + }; + ProducedBlock::new(inner, commitment_config) } } diff --git a/blockstore/src/block_stores/postgres/postgres_block_store_writer.rs b/blockstore/src/block_stores/postgres/postgres_block_store_writer.rs index abbad945..155dfb26 100644 --- a/blockstore/src/block_stores/postgres/postgres_block_store_writer.rs +++ b/blockstore/src/block_stores/postgres/postgres_block_store_writer.rs @@ -316,7 +316,7 @@ fn div_ceil(a: usize, b: usize) -> usize { #[cfg(test)] mod tests { use super::*; - use solana_lite_rpc_core::structures::produced_block::TransactionInfo; + use solana_lite_rpc_core::structures::produced_block::{ProducedBlockInner, TransactionInfo}; use solana_sdk::commitment_config::CommitmentConfig; use solana_sdk::signature::Signature; use std::str::FromStr; @@ -364,7 +364,7 @@ mod tests { let sig1 = Signature::from_str("5VBroA4MxsbZdZmaSEb618WRRwhWYW9weKhh3md1asGRx7nXDVFLua9c98voeiWdBE7A9isEoLL7buKyaVRSK1pV").unwrap(); let sig2 = Signature::from_str("3d9x3rkVQEoza37MLJqXyadeTbEJGUB6unywK4pjeRLJc16wPsgw3dxPryRWw3UaLcRyuxEp1AXKGECvroYxAEf2").unwrap(); - ProducedBlock { + let inner = ProducedBlockInner { block_height: 42, blockhash: "blockhash".to_string(), previous_blockhash: "previous_blockhash".to_string(), @@ -373,10 +373,10 @@ mod tests { transactions: vec![create_test_tx(sig1), create_test_tx(sig2)], // TODO double if this is unix millis or seconds block_time: 1699260872000, - commitment_config: CommitmentConfig::finalized(), leader_id: None, rewards: None, - } + }; + ProducedBlock::new(inner, CommitmentConfig::finalized()) } fn create_test_tx(signature: Signature) -> TransactionInfo { diff --git a/blockstore/tests/blockstore_integration_tests.rs b/blockstore/tests/blockstore_integration_tests.rs index e5094c4d..dadde559 100644 --- a/blockstore/tests/blockstore_integration_tests.rs +++ b/blockstore/tests/blockstore_integration_tests.rs @@ -296,15 +296,11 @@ fn spawn_client_to_blockstorage( // note: no startup deloy loop { match blocks_notifier.recv().await { - Ok(ProducedBlock { - slot, - commitment_config, - .. - }) => { - if commitment_config != CommitmentConfig::confirmed() { + Ok(produced_block) => { + if produced_block.commitment_config != CommitmentConfig::confirmed() { continue; } - let confirmed_slot = slot; + let confirmed_slot = produced_block.slot; // we cannot expect the most recent data let query_slot = confirmed_slot - 3; match block_storage_query.query_block(query_slot).await { diff --git a/blockstore/tests/multiple_strategy_block_store_tests.rs b/blockstore/tests/multiple_strategy_block_store_tests.rs index 422180fc..deecedef 100644 --- a/blockstore/tests/multiple_strategy_block_store_tests.rs +++ b/blockstore/tests/multiple_strategy_block_store_tests.rs @@ -4,21 +4,20 @@ use solana_lite_rpc_blockstore::block_stores::postgres::postgres_block_store_que use solana_lite_rpc_blockstore::block_stores::postgres::postgres_block_store_writer::PostgresBlockStore; use solana_lite_rpc_blockstore::block_stores::postgres::PostgresSessionConfig; use solana_lite_rpc_core::structures::epoch::EpochCache; -use solana_lite_rpc_core::structures::produced_block::ProducedBlock; +use solana_lite_rpc_core::structures::produced_block::{ProducedBlock, ProducedBlockInner}; use solana_sdk::pubkey::Pubkey; use solana_sdk::reward_type::RewardType; use solana_sdk::{commitment_config::CommitmentConfig, hash::Hash}; use solana_transaction_status::Reward; pub fn create_test_block(slot: u64, commitment_config: CommitmentConfig) -> ProducedBlock { - ProducedBlock { + let inner = ProducedBlockInner { block_height: slot, blockhash: Hash::new_unique().to_string(), previous_blockhash: Hash::new_unique().to_string(), parent_slot: slot - 1, transactions: vec![], block_time: 0, - commitment_config, leader_id: None, slot, rewards: Some(vec![Reward { @@ -28,7 +27,8 @@ pub fn create_test_block(slot: u64, commitment_config: CommitmentConfig) -> Prod reward_type: Some(RewardType::Voting), commission: None, }]), - } + }; + ProducedBlock::new(inner, commitment_config) } #[ignore = "need postgres database"] diff --git a/cluster-endpoints/src/grpc_subscription.rs b/cluster-endpoints/src/grpc_subscription.rs index ed56ebb2..d16422ba 100644 --- a/cluster-endpoints/src/grpc_subscription.rs +++ b/cluster-endpoints/src/grpc_subscription.rs @@ -35,6 +35,7 @@ use tracing::debug_span; use crate::rpc_polling::vote_accounts_and_cluster_info_polling::{ poll_cluster_info, poll_vote_accounts, }; +use solana_lite_rpc_core::structures::produced_block::ProducedBlockInner; use yellowstone_grpc_proto::prelude::SubscribeUpdateBlock; /// grpc version of ProducedBlock mapping @@ -254,7 +255,7 @@ pub fn from_grpc_block_update( None }; - ProducedBlock { + let inner = ProducedBlockInner { transactions: txs, block_height: block .block_height @@ -263,12 +264,12 @@ pub fn from_grpc_block_update( block_time: block.block_time.map(|time| time.timestamp).unwrap() as u64, blockhash: block.blockhash, previous_blockhash: block.parent_blockhash, - commitment_config, leader_id, parent_slot: block.parent_slot, slot: block.slot, rewards, - } + }; + ProducedBlock::new(inner, commitment_config) } pub fn create_grpc_subscription( diff --git a/cluster-endpoints/src/rpc_polling/poll_blocks.rs b/cluster-endpoints/src/rpc_polling/poll_blocks.rs index d68910b4..18ad918c 100644 --- a/cluster-endpoints/src/rpc_polling/poll_blocks.rs +++ b/cluster-endpoints/src/rpc_polling/poll_blocks.rs @@ -1,7 +1,7 @@ use anyhow::{bail, Context}; use solana_client::nonblocking::rpc_client::RpcClient; use solana_lite_rpc_core::encoding::BinaryEncoding; -use solana_lite_rpc_core::structures::produced_block::TransactionInfo; +use solana_lite_rpc_core::structures::produced_block::{ProducedBlockInner, TransactionInfo}; use solana_lite_rpc_core::{ structures::{ produced_block::ProducedBlock, @@ -319,7 +319,7 @@ pub fn from_ui_block( let block_time = block.block_time.unwrap_or(0) as u64; - ProducedBlock { + let inner = ProducedBlockInner { transactions: txs, block_height, leader_id, @@ -328,9 +328,9 @@ pub fn from_ui_block( parent_slot, block_time, slot, - commitment_config, rewards, - } + }; + ProducedBlock::new(inner, commitment_config) } #[inline] diff --git a/core/src/structures/produced_block.rs b/core/src/structures/produced_block.rs index 746f28e2..8f25b0b1 100644 --- a/core/src/structures/produced_block.rs +++ b/core/src/structures/produced_block.rs @@ -3,6 +3,9 @@ use solana_sdk::message::v0::MessageAddressTableLookup; use solana_sdk::pubkey::Pubkey; use solana_sdk::{slot_history::Slot, transaction::TransactionError}; use solana_transaction_status::Reward; +use std::fmt::Debug; +use std::ops::Deref; +use std::sync::Arc; #[derive(Debug, Clone)] pub struct TransactionInfo { @@ -19,9 +22,42 @@ pub struct TransactionInfo { pub address_lookup_tables: Vec, } -// TODO try to remove Clone -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct ProducedBlock { + // Arc is required for channels + inner: Arc, + pub commitment_config: CommitmentConfig, +} + +impl ProducedBlock { + pub fn new(inner: ProducedBlockInner, commitment_config: CommitmentConfig) -> Self { + ProducedBlock { + inner: Arc::new(inner), + commitment_config, + } + } +} + +/// # Example +/// ```text +/// ProducedBlock { slot: 254169151, commitment_config: processed, blockhash: BULfZwLswkDbHhTrHGDASUtmNAG8gk6TV2njnobjYLyd, transactions_count: 806 } +/// ``` +impl Debug for ProducedBlock { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "ProducedBlock {{ slot: {}, commitment_config: {}, blockhash: {}, transactions_count: {} }}", + self.slot, self.commitment_config.commitment, self.blockhash, self.transactions.len()) + } +} + +impl Deref for ProducedBlock { + type Target = ProducedBlockInner; + + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +pub struct ProducedBlockInner { pub transactions: Vec, pub leader_id: Option, pub blockhash: String, @@ -29,25 +65,24 @@ pub struct ProducedBlock { pub slot: Slot, pub parent_slot: Slot, pub block_time: u64, - pub commitment_config: CommitmentConfig, pub previous_blockhash: String, pub rewards: Option>, } impl ProducedBlock { - /// moving commitment level to finalized - pub fn to_finalized_block(&self) -> Self { - ProducedBlock { - commitment_config: CommitmentConfig::finalized(), - ..self.clone() - } - } - /// moving commitment level to confirmed pub fn to_confirmed_block(&self) -> Self { ProducedBlock { + inner: self.inner.clone(), commitment_config: CommitmentConfig::confirmed(), - ..self.clone() + } + } + + /// moving commitment level to finalized + pub fn to_finalized_block(&self) -> Self { + ProducedBlock { + inner: self.inner.clone(), + commitment_config: CommitmentConfig::finalized(), } } } diff --git a/services/src/data_caching_service.rs b/services/src/data_caching_service.rs index 17b68988..05315ea3 100644 --- a/services/src/data_caching_service.rs +++ b/services/src/data_caching_service.rs @@ -64,7 +64,7 @@ impl DataCachingService { _ => TransactionConfirmationStatus::Processed, }; - for tx in block.transactions { + for tx in &block.transactions { let block_info = data_cache .block_information_store .get_block_info(&tx.recent_blockhash); @@ -101,7 +101,7 @@ impl DataCachingService { // notify data_cache .tx_subs - .notify(block.slot, &tx, block.commitment_config) + .notify(block.slot, tx, block.commitment_config) .await; } }