/// Runs the block publisher's background thread and returns a handle for joining the thread
///
/// The publisher's background thread is responsible for the following:
/// * Receiving batches from the batch submitter
/// * Adding new batches to the pending batches pool
/// * Notifying all batch observers about new batches
/// * Adding new batches to an in-progress block if there is one
/// * Notifyng the batch sender when the pending batches pool is full or unblocked
Runs the block publisher's background thread and returns a handle for joining the thread
The publisher's background thread is responsible for the following:
Receiving batches from the batch submitter
Adding new batches to the pending batches pool
Notifying all batch observers about new batches
Adding new batches to an in-progress block if there is one
Notifyng the batch sender when the pending batches pool is full or unblocked
Runs the block publisher ' s background thread and returns a handle for joining the thread The publisher ' s background thread is responsible for the following : Receiving batches from the batch submitter Adding new batches to the pending batches pool Notifying all batch observers about new batches Adding new batches to an in - progress block if there is one Notifyng the batch sender when the pending batches pool is full or unblocked
fn start_publisher_thread(
    backpressure_sender: Sender<BackpressureMessage>,
    batch_observers: Vec<Box<dyn BatchObserver>>,
    pending_batches: Arc<RwLock<PendingBatchesPool>>,
    candidate_block: Arc<Mutex<Option<CandidateBlock>>>,
    commit_store: CommitStore,
    state_view_factory: StateViewFactory,
    receiver: Receiver<BlockPublisherMessage>,
) -> std::io::Result<thread::JoinHandle<()>> {
    thread::Builder::new()
        .name("Block Publisher".into())
        .spawn(move || {
            let pool_full = false;
            loop {
                match receiver.recv() {
                    Ok(BlockPublisherMessage::NewBatch(batch)) => {
                        // Verify that the batch signer is permitted
                        let chain_head = match commit_store.get_chain_head() {
                            Ok(chain_head) => chain_head,
                            Err(err) => {
                                error!("Failed to get chain head from commit store: {}", err);
                                continue;
                            }
                        };
                        let identity_view = match state_view_factory
                            .create_view::<IdentityView>(chain_head.header().state_root_hash())
                        {
                            Ok(identity_view) => identity_view,
                            Err(err) => {
                                error!("Failed to get identity view: {}", err);
                                continue;
                            }
                        };
                        let permission_verifier = PermissionVerifier::new(Box::new(identity_view));
                        match permission_verifier.is_batch_signer_authorized(&batch) {
                            Ok(true) => {}
                            Ok(false) => {
                                debug!(
                                    "Ignoring batch because signer is not authorized: {}",
                                    batch.batch().header_signature()
                                );
                                continue;
                            }
                            Err(err) => {
                                error!("Failed to perform batch permission verification: {}", err);
                                continue;
                            }
                        }
                        // Check if the batch is already committed
                        match commit_store.contains_batch(batch.batch().header_signature()) {
                            Ok(true) => {
                                debug!(
                                    "Ignoring batch that is already committed: {}",
                                    batch.batch().header_signature()
                                );
                                continue;
                            }
                            Ok(false) => {}
                            Err(err) => {
                                error!("Failed to check if batch is already committed: {}", err);
                                continue;
                            }
                        }
                        // Add the batch to the pool
                        match pending_batches.write() {
                            Ok(mut pending_batches) => {
                                // If the batch is already in the pool, don't do anything further
                                // with the batch
                                if !pending_batches.append(batch.clone()) {
                                    continue;
                                }
                                // Notify batch observers
                                for observer in &batch_observers {
                                    observer.notify_batch_pending(batch.batch());
                                }
                                // Notify the batch submitter if the batches pool has filled up or
                                // is open again
                                if !pool_full
                                    && pending_batches.is_full()
                                    && backpressure_sender
                                        .send(BackpressureMessage::PoolFull)
                                        .is_err()
                                {
                                    warn!("Batch submitter disconnected");
                                } else if pool_full
                                    && !pending_batches.is_full()
                                    && backpressure_sender
                                        .send(BackpressureMessage::PoolUnblocked)
                                        .is_err()
                                {
                                    warn!("Batch submitter disconnected");
                                }
                            }
                            Err(_) => {
                                error!("Pending batches pool lock poisoned");
                                break;
                            }
                        }
                        // If currently building a block and it's not already full, schedule the
                        // batch
                        match candidate_block.lock() {
                            Ok(mut candidate_block) => {
                                if let Some(mut candidate_block) = candidate_block.as_mut() {
                                    if candidate_block.can_schedule_batch() {
                                        if let Err(err) = schedule_batch(
                                            batch,
                                            &commit_store,
                                            &mut candidate_block,
                                        ) {
                                            error!("Failed to schedule batch: {}", err);
                                        }
                                    }
                                }
                            }
                            Err(_) => {
                                error!("Candidate block lock poisoned");
                                break;
                            }
                        }
                    }
                    Ok(BlockPublisherMessage::Shutdown) => break,
                    Err(_) => {
                        warn!(
                            "All senders have disconnected; shutting down block publisher thread"
                        );
                        break;
                    }
                }
            }
        })
}
fn start_publisher_thread ( backpressure_sender : Sender < BackpressureMessage > , batch_observers : Vec < Box < dyn BatchObserver > > , pending_batches : Arc < RwLock < PendingBatchesPool > > , candidate_block : Arc < Mutex < Option < CandidateBlock > > > , commit_store : CommitStore , state_view_factory : StateViewFactory , receiver : Receiver < BlockPublisherMessage > , ) -> std :: io :: Result < thread :: JoinHandle < ( ) > > { thread :: Builder :: new ( ) . name ( " " . into ( ) ) . spawn ( move | | { let pool_full = false ; loop { match receiver . recv ( ) { Ok ( BlockPublisherMessage :: NewBatch ( batch ) ) => { // Verify that the batch signer is permitted let chain_head = match commit_store . get_chain_head ( ) { Ok ( chain_head ) => chain_head , Err ( err ) => { error ! ( " " err ) ; continue ; } } ; let identity_view = match state_view_factory . create_view :: < IdentityView > ( chain_head . header ( ) . state_root_hash ( ) ) { Ok ( identity_view ) => identity_view , Err ( err ) => { error ! ( " " err ) ; continue ; } } ; let permission_verifier = PermissionVerifier :: new ( Box :: new ( identity_view ) ) ; match permission_verifier . is_batch_signer_authorized ( & batch ) { Ok ( true ) => { } Ok ( false ) => { debug ! ( " " batch batch ( ) header_signature ( ) ) ; continue ; } Err ( err ) => { error ! ( " " err ) ; continue ; } } // Check if the batch is already committed match commit_store . contains_batch ( batch . batch ( ) . header_signature ( ) ) { Ok ( true ) => { debug ! ( " " batch batch ( ) header_signature ( ) ) ; continue ; } Ok ( false ) => { } Err ( err ) => { error ! ( " " err ) ; continue ; } } // Add the batch to the pool match pending_batches . write ( ) { Ok ( mut pending_batches ) => { // If the batch is already in the pool, don't do anything further // with the batch if ! pending_batches . append ( batch . clone ( ) ) { continue ; } // Notify batch observers for observer in & batch_observers { observer . notify_batch_pending ( batch . batch ( ) ) ; } // Notify the batch submitter if the batches pool has filled up or // is open again if ! pool_full && pending_batches . is_full ( ) && backpressure_sender . send ( BackpressureMessage :: PoolFull ) . is_err ( ) { warn ! ( " " ) ; } else if pool_full && ! pending_batches . is_full ( ) && backpressure_sender . send ( BackpressureMessage :: PoolUnblocked ) . is_err ( ) { warn ! ( " " ) ; } } Err ( _ ) => { error ! ( " " ) ; break ; } } // If currently building a block and it's not already full, schedule the // batch match candidate_block . lock ( ) { Ok ( mut candidate_block ) => { if let Some ( mut candidate_block ) = candidate_block . as_mut ( ) { if candidate_block . can_schedule_batch ( ) { if let Err ( err ) = schedule_batch ( batch , & commit_store , & mut candidate_block , ) { error ! ( " " err ) ; } } } } Err ( _ ) => { error ! ( " " ) ; break ; } } } Ok ( BlockPublisherMessage :: Shutdown ) => break , Err ( _ ) => { warn ! ( " " ) ; break ; } } } } ) }
Runs the block publisher's background thread and returns a handle for joining the thread
The publisher's background thread is responsible for the following:
 Receiving batches from the batch submitter
 Adding new batches to the pending batches pool
 Notifying all batch observers about new batches
 Adding new batches to an in-progress block if there is one
 Notifyng the batch sender when the pending batches pool is full or unblocked
Runs the block publisher ' s background thread and returns a handle for joining the thread The publisher ' s background thread is responsible for the following : Receiving batches from the batch submitter Adding new batches to the pending batches pool Notifying all batch observers about new batches Adding new batches to an in - progress block if there is one Notifyng the batch sender when the pending batches pool is full or unblocked
backpressure_sender
Sender<BackpressureMessage>
batch_observers
Vec<Box<dyn BatchObserver>>
pending_batches
Arc<RwLock<PendingBatchesPool>>
candidate_block
Arc<Mutex<Option<CandidateBlock>>>
commit_store
CommitStore
state_view_factory
StateViewFactory
receiver
Receiver<BlockPublisherMessage>