Struct narwhal_primary::BlockWaiter

pub struct BlockWaiter<SynchronizerHandler: Handler + Send + Sync + 'static> { /* private fields */ }

BlockWaiter is responsible for fetching the block data from the downstream worker nodes. A block is basically the aggregate of batches of transactions for a given certificate.

Example

Basic setup of the BlockWaiter

This example shows the basic setup of the BlockWaiter module. It showcases the necessary components that have to be used (e.x channels, datastore etc) and how a request (command) should be issued to get a block and receive the result of it.

struct BlockSynchronizerHandler;

#[async_trait]
impl Handler for BlockSynchronizerHandler {

    async fn get_and_synchronize_block_headers(&self, block_ids: Vec<CertificateDigest>) -> Vec<Result<Certificate, Error>> {
        vec![]
    }

    async fn get_block_headers(&self, block_ids: Vec<CertificateDigest>) -> Vec<BlockSynchronizeResult<BlockHeader>> {
        vec![]
    }

    async fn synchronize_block_payloads(&self, certificates: Vec<Certificate>) -> Vec<Result<Certificate, Error>> {
        vec![]
    }

}

#[tokio::main(flavor = "current_thread")]
    let (tx_commands, rx_commands) = test_utils::test_channel!(1);
    let (tx_batches, rx_batches) = test_utils::test_channel!(1);
    let (tx_get_block, mut rx_get_block) = oneshot::channel();

    let name = PublicKey::default();
    let committee = Committee{ epoch: 0, authorities: BTreeMap::new() };
    let worker_cache: SharedWorkerCache = WorkerCache{ epoch: 0, workers: BTreeMap::new() }.into();
    let (_tx_reconfigure, rx_reconfigure) = watch::channel(ReconfigureNotification::NewEpoch(committee.clone()));

    // A dummy certificate
    let certificate = Certificate::default();

    // Dummy - we expect the BlockSynchronizer to actually respond, here
    // we are using a mock
    BlockWaiter::spawn(
        name,
        committee,
        worker_cache,
        rx_reconfigure,
        rx_commands,
        rx_batches,
        Arc::new(BlockSynchronizerHandler{}),
        network::P2pNetwork::new(test_utils::random_network()),
    );

    // Send a command to receive a block
    tx_commands
        .send(BlockCommand::GetBlock {
            id: certificate.digest(),
            sender: tx_get_block,
        })
        .await;

    // Dummy - we expect to receive the requested batches via another component
    // and get fed via the tx_batches channel.
    tx_batches.send(Ok(BatchMessage{ id: BatchDigest::default(), transactions: Batch(vec![]) })).await;

    // Wait to receive the block output to the provided sender channel
    match rx_get_block.await.unwrap() {
        Ok(result) => {
            println!("Successfully received a block response");
        }
        Err(err) => {
            println!("Received an error {}", err);
        }
    }

Implementations

impl<SynchronizerHandler: Handler + Send + Sync + 'static> BlockWaiter<SynchronizerHandler>

pub fn spawn(
    name: PublicKey,
    committee: Committee,
    worker_cache: SharedWorkerCache,
    rx_reconfigure: Receiver<ReconfigureNotification>,
    rx_commands: Receiver<BlockCommand>,
    batch_receiver: Receiver<Result<BatchMessage, BatchMessageError>>,
    block_synchronizer_handler: Arc<SynchronizerHandler>,
    worker_network: P2pNetwork
) -> JoinHandle<()>