fn process_blob(
        &mut self,
        id: &Pubkey,
        blob: SharedBlob,
        pix: u64,
        consume_queue: &mut Vec<Entry>,
        consumed: &mut u64,
        tick_height: &mut u64,
        leader_unknown: bool,
        pending_retransmits: &mut bool,
    ) {
        let w = (pix % self.window_size()) as usize;

        let is_coding = blob.read().unwrap().is_coding();

        // insert a newly received blob into a window slot, clearing out and recycling any previous
        //  blob unless the incoming blob is a duplicate (based on idx)
        // returns whether the incoming is a duplicate blob
        fn insert_blob_is_dup(
            id: &Pubkey,
            blob: SharedBlob,
            pix: u64,
            window_slot: &mut Option<SharedBlob>,
            c_or_d: &str,
        ) -> bool {
            if let Some(old) = mem::replace(window_slot, Some(blob)) {
                let is_dup = old.read().unwrap().index().unwrap() == pix;
                trace!(
                    "{}: occupied {} window slot {:}, is_dup: {}",
                    id,
                    c_or_d,
                    pix,
                    is_dup
                );
                is_dup
            } else {
                trace!("{}: empty {} window slot {:}", id, c_or_d, pix);
                false
            }
        }

        // insert the new blob into the window, overwrite and recycle old (or duplicate) entry
        let is_duplicate = if is_coding {
            insert_blob_is_dup(id, blob, pix, &mut self[w].coding, "coding")
        } else {
            insert_blob_is_dup(id, blob, pix, &mut self[w].data, "data")
        };

        if is_duplicate {
            return;
        }

        self[w].leader_unknown = leader_unknown;
        *pending_retransmits = true;

        #[cfg(feature = "erasure")]
        {
            let window_size = self.window_size();
            if erasure::recover(id, self, *consumed, (*consumed % window_size) as usize).is_err() {
                trace!("{}: erasure::recover failed", id);
            }
        }

        // push all contiguous blobs into consumed queue, increment consumed
        loop {
            let k = (*consumed % self.window_size()) as usize;
            trace!("{}: k: {} consumed: {}", id, k, *consumed,);

            let k_data_blob;
            let k_data_slot = &mut self[k].data;
            if let Some(blob) = k_data_slot {
                if blob.read().unwrap().index().unwrap() < *consumed {
                    // window wrap-around, end of received
                    break;
                }
                k_data_blob = (*blob).clone();
            } else {
                // self[k].data is None, end of received
                break;
            }

            // Check that we can get the entries from this blob
            match reconstruct_entries_from_blobs(vec![k_data_blob]) {
                Ok(entries) => {
                    for entry in &entries {
                        *tick_height += entry.is_tick() as u64;
                    }
                    consume_queue.extend(entries);
                }
                Err(_) => {
                    // If the blob can't be deserialized, then remove it from the
                    // window and exit. *k_data_slot cannot be None at this point,
                    // so it's safe to unwrap.
                    k_data_slot.take();
                    break;
                }
            }

            *consumed += 1;
        }
    }
/// process a blob: Add blob to the window. If a continuous set of blobs
///      starting from consumed is thereby formed, add that continuous
///      range of blobs to a queue to be sent on to the next stage.
///
/// * `self` - the window we're operating on
/// * `id` - this node's id
/// * `blob` -  the blob to be processed into the window and rebroadcast
/// * `pix` -  the index of the blob, corresponds to
///            the entry height of this blob
/// * `consume_queue` - output, blobs to be rebroadcast are placed here
/// * `consumed` - input/output, the entry-height to which this
///                 node has populated and rebroadcast entries
process a blob: Add blob to the window. If a continuous set of blobs
starting from consumed is thereby formed, add that continuous
range of blobs to a queue to be sent on to the next stage.


process a blob : Add blob to the window . If a continuous set of blobs starting from consumed is thereby formed add that continuous range of blobs to a queue to be sent on to the next stage .
fn process_blob(
        &mut self,
        id: &Pubkey,
        blob: SharedBlob,
        pix: u64,
        consume_queue: &mut Vec<Entry>,
        consumed: &mut u64,
        tick_height: &mut u64,
        leader_unknown: bool,
        pending_retransmits: &mut bool,
    ) {
        let w = (pix % self.window_size()) as usize;
        let is_coding = blob.read().unwrap().is_coding();
        // insert a newly received blob into a window slot, clearing out and recycling any previous
        //  blob unless the incoming blob is a duplicate (based on idx)
        // returns whether the incoming is a duplicate blob
        fn insert_blob_is_dup(
            id: &Pubkey,
            blob: SharedBlob,
            pix: u64,
            window_slot: &mut Option<SharedBlob>,
            c_or_d: &str,
        ) -> bool {
            if let Some(old) = mem::replace(window_slot, Some(blob)) {
                let is_dup = old.read().unwrap().index().unwrap() == pix;
                trace!(
                    "{}: occupied {} window slot {:}, is_dup: {}",
                    id,
                    c_or_d,
                    pix,
                    is_dup
                );
                is_dup
            } else {
                trace!("{}: empty {} window slot {:}", id, c_or_d, pix);
                false
            }
        }
        // insert the new blob into the window, overwrite and recycle old (or duplicate) entry
        let is_duplicate = if is_coding {
            insert_blob_is_dup(id, blob, pix, &mut self[w].coding, "coding")
        } else {
            insert_blob_is_dup(id, blob, pix, &mut self[w].data, "data")
        };
        if is_duplicate {
            return;
        }
        self[w].leader_unknown = leader_unknown;
        *pending_retransmits = true;
        #[cfg(feature = "erasure")]
        {
            let window_size = self.window_size();
            if erasure::recover(id, self, *consumed, (*consumed % window_size) as usize).is_err() {
                trace!("{}: erasure::recover failed", id);
            }
        }
        // push all contiguous blobs into consumed queue, increment consumed
        loop {
            let k = (*consumed % self.window_size()) as usize;
            trace!("{}: k: {} consumed: {}", id, k, *consumed,);
            let k_data_blob;
            let k_data_slot = &mut self[k].data;
            if let Some(blob) = k_data_slot {
                if blob.read().unwrap().index().unwrap() < *consumed {
                    // window wrap-around, end of received
                    break;
                }
                k_data_blob = (*blob).clone();
            } else {
                // self[k].data is None, end of received
                break;
            }
            // Check that we can get the entries from this blob
            match reconstruct_entries_from_blobs(vec![k_data_blob]) {
                Ok(entries) => {
                    for entry in &entries {
                        *tick_height += entry.is_tick() as u64;
                    }
                    consume_queue.extend(entries);
                }
                Err(_) => {
                    // If the blob can't be deserialized, then remove it from the
                    // window and exit. *k_data_slot cannot be None at this point,
                    // so it's safe to unwrap.
                    k_data_slot.take();
                    break;
                }
            }
            *consumed += 1;
        }
    }
fn process_blob ( & mut self , id : & Pubkey , blob : SharedBlob , pix : u64 , consume_queue : & mut Vec < Entry > , consumed : & mut u64 , tick_height : & mut u64 , leader_unknown : bool , pending_retransmits : & mut bool , ) { let w = ( pix % self . window_size ( ) ) as usize ; let is_coding = blob . read ( ) . unwrap ( ) . is_coding ( ) ; // insert a newly received blob into a window slot, clearing out and recycling any previous //  blob unless the incoming blob is a duplicate (based on idx) // returns whether the incoming is a duplicate blob fn insert_blob_is_dup ( id : & Pubkey , blob : SharedBlob , pix : u64 , window_slot : & mut Option < SharedBlob > , c_or_d : & str , ) -> bool { if let Some ( old ) = mem :: replace ( window_slot , Some ( blob ) ) { let is_dup = old . read ( ) . unwrap ( ) . index ( ) . unwrap ( ) == pix ; trace ! ( " " id c_or_d pix is_dup ) ; is_dup } else { trace ! ( " " id c_or_d pix ) ; false } } // insert the new blob into the window, overwrite and recycle old (or duplicate) entry let is_duplicate = if is_coding { insert_blob_is_dup ( id , blob , pix , & mut self [ w ] . coding , " " ) } else { insert_blob_is_dup ( id , blob , pix , & mut self [ w ] . data , " " ) } ; if is_duplicate { return ; } self [ w ] . leader_unknown = leader_unknown ; * pending_retransmits = true ; # [ cfg ( feature " " ) ] { let window_size = self . window_size ( ) ; if erasure :: recover ( id , self , * consumed , ( * consumed % window_size ) as usize ) . is_err ( ) { trace ! ( " " id ) ; } } // push all contiguous blobs into consumed queue, increment consumed loop { let k = ( * consumed % self . window_size ( ) ) as usize ; trace ! ( " " id k consumed ) ; let k_data_blob ; let k_data_slot = & mut self [ k ] . data ; if let Some ( blob ) = k_data_slot { if blob . read ( ) . unwrap ( ) . index ( ) . unwrap ( ) < * consumed { // window wrap-around, end of received break ; } k_data_blob = ( * blob ) . clone ( ) ; } else { // self[k].data is None, end of received break ; } // Check that we can get the entries from this blob match reconstruct_entries_from_blobs ( vec ! [ k_data_blob ] ) { Ok ( entries ) => { for entry in & entries { * tick_height += entry . is_tick ( ) as u64 ; } consume_queue . extend ( entries ) ; } Err ( _ ) => { // If the blob can't be deserialized, then remove it from the // window and exit. *k_data_slot cannot be None at this point, // so it's safe to unwrap. k_data_slot . take ( ) ; break ; } } * consumed += 1 ; } }
process a blob: Add blob to the window.
process a blob : Add blob to the window .
&mut self
id
&Pubkey
blob
SharedBlob
pix
u64
consume_queue
&mut Vec<Entry>
consumed
&mut u64
tick_height
&mut u64
leader_unknown
bool
pending_retransmits
&mut bool
&mut self
id
&Pubkey
blob
SharedBlob
pix
u64
consume_queue
&mut Vec<Entry>
consumed
&mut u64
tick_height
&mut u64
leader_unknown
bool
pending_retransmits
&mut bool