Struct iroh_bitswap::Block

pub struct Block {
    pub cid: Cid,
    pub data: Bytes,
}

A wrapper around bytes with their Cid.

Fields

cid: Cid

data: Bytes

Implementations

impl Block

pub fn new(data: Bytes, cid: Cid) -> Self

Examples found in repository

src/block.rs (line 61)

    pub fn create_block_v1<B: Into<Bytes>>(bytes: B) -> Block {
        let bytes = bytes.into();
        let digest = Code::Sha2_256.digest(&bytes);
        let cid = Cid::new_v1(RAW, digest);
        Block::new(bytes, cid)
    }

    pub fn create_block_v0<B: Into<Bytes>>(bytes: B) -> Block {
        let bytes = bytes.into();
        let digest = Code::Sha2_256.digest(&bytes);
        let cid = Cid::new_v0(digest).unwrap();
        Block::new(bytes, cid)
    }

src/message.rs (line 460)

    fn try_from(pbm: pb::Message) -> Result<Self, Self::Error> {
        let full = pbm.wantlist.as_ref().map(|w| w.full).unwrap_or_default();
        let mut message = BitswapMessage::new(full);

        if let Some(wantlist) = pbm.wantlist {
            for entry in wantlist.entries {
                let cid = Cid::try_from(entry.block)?;
                message.add_full_entry(
                    cid,
                    entry.priority,
                    entry.cancel,
                    entry.want_type.try_into()?,
                    entry.send_dont_have,
                );
            }
        }

        // deprecated
        for data in pbm.blocks {
            // CID v0, SHA26
            let block = Block::from_v0_data(data)?;
            message.add_block(block);
        }

        for block in pbm.payload {
            let prefix = Prefix::new(&block.prefix)?;
            let cid = prefix.to_cid(&block.data)?;
            let block = Block::new(block.data, cid);
            message.add_block(block);
        }

        for block_presence in pbm.block_presences {
            let cid = Cid::try_from(block_presence.cid)?;
            message.add_block_presence(cid, block_presence.r#type.try_into()?);
        }

        message.pending_bytes = pbm.pending_bytes;

        Ok(message)
    }

pub fn from_v0_data(data: Bytes) -> Result<Self>

Examples found in repository

src/message.rs (line 453)

    fn try_from(pbm: pb::Message) -> Result<Self, Self::Error> {
        let full = pbm.wantlist.as_ref().map(|w| w.full).unwrap_or_default();
        let mut message = BitswapMessage::new(full);

        if let Some(wantlist) = pbm.wantlist {
            for entry in wantlist.entries {
                let cid = Cid::try_from(entry.block)?;
                message.add_full_entry(
                    cid,
                    entry.priority,
                    entry.cancel,
                    entry.want_type.try_into()?,
                    entry.send_dont_have,
                );
            }
        }

        // deprecated
        for data in pbm.blocks {
            // CID v0, SHA26
            let block = Block::from_v0_data(data)?;
            message.add_block(block);
        }

        for block in pbm.payload {
            let prefix = Prefix::new(&block.prefix)?;
            let cid = prefix.to_cid(&block.data)?;
            let block = Block::new(block.data, cid);
            message.add_block(block);
        }

        for block_presence in pbm.block_presences {
            let cid = Cid::try_from(block_presence.cid)?;
            message.add_block_presence(cid, block_presence.r#type.try_into()?);
        }

        message.pending_bytes = pbm.pending_bytes;

        Ok(message)
    }

pub fn cid(&self) -> &Cid

Examples found in repository

src/message.rs (line 350)

    pub fn add_block(&mut self, block: Block) {
        self.block_presences.remove(block.cid());
        self.blocks.insert(*block.cid(), block);
    }

    pub fn add_block_presence(&mut self, cid: Cid, typ: BlockPresenceType) {
        if self.blocks.contains_key(&cid) {
            return;
        }
        self.block_presences.insert(cid, typ);
    }

    pub fn add_have(&mut self, cid: Cid) {
        self.add_block_presence(cid, BlockPresenceType::Have);
    }

    pub fn add_dont_have(&mut self, cid: Cid) {
        self.add_block_presence(cid, BlockPresenceType::DontHave);
    }

    pub fn encoded_len(&self) -> usize {
        let block_size: usize = self.blocks.values().map(|b| b.data.len()).sum();
        let block_presence_size: usize = self.block_presences().map(|bp| bp.encoded_len()).sum();

        let wantlist_size: usize = self.wantlist.values().map(|e| e.encoded_len()).sum();

        block_size + block_presence_size + wantlist_size
    }

    pub fn encode_as_proto_v0(&self) -> pb::Message {
        let mut message = pb::Message::default();

        // wantlist
        let mut wantlist = pb::message::Wantlist::default();
        for entry in self.wantlist.values() {
            wantlist.entries.push(entry.into());
        }
        wantlist.full = self.full;
        message.wantlist = Some(wantlist);

        // blocks
        for block in self.blocks.values() {
            message.blocks.push(block.data().clone());
        }

        message
    }

    pub fn encode_as_proto_v1(&self) -> pb::Message {
        let mut message = pb::Message::default();

        // wantlist
        let mut wantlist = pb::message::Wantlist::default();
        for entry in self.wantlist.values() {
            wantlist.entries.push(entry.into());
        }
        wantlist.full = self.full;
        message.wantlist = Some(wantlist);

        // blocks
        for block in self.blocks.values() {
            message.payload.push(pb::message::Block {
                prefix: Prefix::from(block.cid()).to_bytes(),
                data: block.data().clone(),
            });
        }

        // block presences
        for (cid, typ) in &self.block_presences {
            message.block_presences.push(pb::message::BlockPresence {
                cid: cid.to_bytes(),
                r#type: (*typ).into(),
            });
        }

        message.pending_bytes = self.pending_bytes();

        message
    }

src/server.rs (line 286)

    pub async fn notify_new_blocks(&self, blocks: &[Block]) -> Result<()> {
        //  send wanted blocks to the decision engine
        self.engine.notify_new_blocks(blocks).await;
        if self.inner.provide_enabled {
            for block in blocks {
                if let Err(err) = self.inner.new_blocks.send(*block.cid()).await {
                    warn!("failed to send new blocks: {:?}", err);
                }
            }
        }

        Ok(())
    }

src/server/decision.rs (line 467)

    pub async fn message_sent(&self, peer: &PeerId, message: &BitswapMessage) {
        let l = self.find_or_create(peer).await;
        let mut ledger = l.lock().await;

        // remove sent blocks from the want list for the peer
        for block in message.blocks() {
            self.score_ledger
                .add_to_sent_bytes(ledger.partner(), block.data().len())
                .await;
            ledger
                .wantlist_mut()
                .remove_type(block.cid(), WantType::Block);
        }

        // remove sent block presences from the wantlist for the peer
        for bp in message.block_presences() {
            // don't record sent data, we reserve that for data blocks
            if bp.typ == BlockPresenceType::Have {
                ledger.wantlist_mut().remove_type(&bp.cid, WantType::Have);
            }
        }
    }

    fn split_wants<'a>(
        &self,
        peer: &PeerId,
        entries: impl Iterator<Item = &'a Entry>,
    ) -> (Vec<&'a Entry>, Vec<&'a Entry>, Vec<&'a Entry>) {
        let mut wants = Vec::new();
        let mut cancels = Vec::new();
        let mut denials = Vec::new();

        for entry in entries {
            if entry.cancel {
                cancels.push(entry);
            } else if let Some(ref filter) = self.peer_block_request_filter {
                if (filter)(peer, &entry.cid) {
                    wants.push(entry);
                } else {
                    denials.push(entry);
                }
            } else {
                wants.push(entry);
            }
        }

        (wants, cancels, denials)
    }

    pub async fn received_blocks(&self, from: PeerId, blocks: Vec<Block>) {
        if blocks.is_empty() {
            return;
        }

        let l = self.find_or_create(&from).await;
        let ledger = l.lock().await;
        for block in blocks {
            self.score_ledger
                .add_to_recv_bytes(ledger.partner(), block.data().len())
                .await;
        }
    }

    pub async fn notify_new_blocks(&self, blocks: &[Block]) {
        if blocks.is_empty() {
            return;
        }

        // get the sizes of each block
        let block_sizes: AHashMap<_, _> = blocks
            .iter()
            .map(|block| (block.cid(), block.data().len()))
            .collect();

        let mut work = false;
        let mut missing_wants: AHashMap<PeerId, Vec<Cid>> = AHashMap::new();
        for block in blocks {
            let cid = block.cid();
            let peer_ledger = self.peer_ledger.lock().await;
            let peers = peer_ledger.peers(cid);
            if peers.is_none() {
                continue;
            }
            for peer in peers.unwrap() {
                let l = self.ledger_map.read().await.get(peer).cloned();
                if l.is_none() {
                    missing_wants.entry(*peer).or_default().push(*cid);
                    continue;
                }
                let l = l.unwrap();
                let ledger = l.lock().await;
                let entry = ledger.wantlist_get(cid);
                if entry.is_none() {
                    missing_wants.entry(*peer).or_default().push(*cid);
                    continue;
                }
                let entry = entry.unwrap();

                work = true;
                let block_size = block_sizes.get(cid).copied().unwrap_or_default();
                let is_want_block = self.send_as_block(entry.want_type, block_size);
                let entry_size = if is_want_block {
                    block_size
                } else {
                    BlockPresence::encoded_len_for_cid(*cid)
                };

                self.peer_task_queue
                    .push_task(
                        *peer,
                        Task {
                            topic: entry.cid,
                            priority: entry.priority as isize,
                            work: entry_size,
                            data: TaskData {
                                block_size,
                                have_block: true,
                                is_want_block,
                                send_dont_have: false,
                            },
                        },
                    )
                    .await;
                self.update_metrics().await;
            }
        }

        // If we found missing wants remove them from the list
        if !missing_wants.is_empty() {
            let ledger_map = self.ledger_map.read().await;
            let mut peer_ledger = self.peer_ledger.lock().await;
            for (peer, wants) in missing_wants.into_iter() {
                if let Some(l) = ledger_map.get(&peer) {
                    let ledger = l.lock().await;
                    for cid in wants {
                        if ledger.wantlist_get(&cid).is_some() {
                            continue;
                        }
                        peer_ledger.cancel_want(&peer, &cid);
                    }
                } else {
                    for cid in wants {
                        peer_ledger.cancel_want(&peer, &cid);
                    }
                }
            }
        }

        if work {
            self.signal_new_work();
        }
    }

src/client/session_interest_manager.rs (line 147)

    pub async fn split_wanted_unwanted<'a>(
        &self,
        blocks: &'a [Block],
    ) -> (Vec<&'a Block>, Vec<&'a Block>) {
        debug!("split_wanted_unwantedn",);

        let wants = &*self.wants.read().await;

        let mut wanted_keys = AHashSet::new();
        for block in blocks {
            let cid = block.cid();
            if let Some(wants) = wants.get(cid) {
                for wanted in wants.values() {
                    if *wanted {
                        wanted_keys.insert(*cid);
                    }
                }
            }
        }

        let mut wanted_blocks = Vec::new();
        let mut not_wanted_blocks = Vec::new();

        for block in blocks {
            if wanted_keys.contains(block.cid()) {
                wanted_blocks.push(block);
            } else {
                not_wanted_blocks.push(block);
            }
        }

        (wanted_blocks, not_wanted_blocks)
    }

src/client.rs (line 97)

    pub async fn new(
        network: Network,
        store: S,
        blocks_received_cb: Option<Box<BlocksReceivedCb>>,
        config: Config,
    ) -> Self {
        let self_id = *network.self_id();

        let (notify, mut default_receiver): (
            async_broadcast::Sender<Block>,
            async_broadcast::Receiver<Block>,
        ) = async_broadcast::broadcast(4096);
        // ensure no blocking is generated

        // TODO: track task
        tokio::task::spawn(async move {
            debug!("starting default receiver");
            loop {
                inc!(BitswapMetrics::ClientLoopTick);
                match default_receiver.recv().await {
                    Ok(block) => {
                        debug!("received block {}", block.cid());
                    }
                    Err(async_broadcast::RecvError::Closed) => {
                        debug!("shutting down default receiver");
                        break;
                    }
                    Err(e) => {
                        warn!("broadcast error: {:?}", e);
                    }
                }
            }
        });

        let session_manager = SessionManager::new(self_id, network.clone(), notify.clone()).await;

        Client {
            network,
            store,
            session_manager,
            provider_search_delay: config.provider_search_delay,
            rebroadcast_delay: config.rebroadcast_delay,
            simulate_dont_haves_on_timeout: config.simluate_donthaves_on_timeout,
            blocks_received_cb: blocks_received_cb.map(Arc::new),
            notify,
        }
    }

    pub async fn stop(self) -> Result<()> {
        self.session_manager.stop().await?;

        Ok(())
    }

    /// Attempts to retrieve a particular block from peers.
    pub async fn get_block(&self, key: &Cid) -> Result<Block> {
        let session = self.new_session().await;
        let block = session.get_block(key).await;
        session.stop().await?;
        block
    }

    pub async fn get_block_with_session_id(
        &self,
        session_id: u64,
        key: &Cid,
        providers: &[PeerId],
    ) -> Result<Block> {
        let session = self.get_or_create_session(session_id).await;
        for provider in providers {
            session.add_provider(key, *provider).await;
        }
        session.get_block(key).await
    }

    pub async fn get_blocks_with_session_id(
        &self,
        session_id: u64,
        keys: &[Cid],
    ) -> Result<BlockReceiver> {
        let session = self.get_or_create_session(session_id).await;
        session.get_blocks(keys).await
    }

    /// Returns a channel where the caller may receive blocks that correspond to the
    /// provided `keys`.
    pub async fn get_blocks(&self, keys: &[Cid]) -> Result<BlockReceiver> {
        let session = self.new_session().await;
        session.get_blocks(keys).await
    }

    /// Announces the existence of blocks to this bitswap service.
    /// Bitswap itself doesn't store new blocks. It's the caller responsibility to ensure
    /// that those blocks are available in the blockstore before calling this function.
    pub async fn notify_new_blocks(&self, blocks: &[Block]) -> Result<()> {
        let block_cids: Vec<Cid> = blocks.iter().map(|b| *b.cid()).collect();
        // Send all block keys (including duplicates) to any session that wants them.
        self.session_manager
            .receive_from(None, &block_cids, &[][..], &[][..])
            .await;

        // Publish the block to any Bitswap clients that had requested blocks.
        // (the sessions use this pubsub mechanism to inform clients of incoming blocks)

        for block in blocks {
            if let Err(err) = self.notify.broadcast(block.clone()).await {
                error!("failed to broadcast block {}: {:?}", block.cid(), err);
            }
        }

        Ok(())
    }

    /// Process blocks received from the network.
    async fn receive_blocks_from(
        &self,
        from: &PeerId,
        incoming: &BitswapMessage,
        haves: &[Cid],
        dont_haves: &[Cid],
    ) -> Result<()> {
        info!("recv_msg start");
        let all_keys: Vec<Cid> = incoming.blocks().map(|b| *b.cid()).collect();
        // Determine wanted and unwanted blocks
        let blocks = incoming.blocks().cloned().collect::<Vec<_>>();
        let (wanted, not_wanted) = self
            .session_manager
            .session_interest_manager()
            .split_wanted_unwanted(&blocks)
            .await;

        for block in not_wanted {
            debug!("recv block not in wantlist: {} from {}", block.cid(), from);
        }

        // Inform the PeerManager so that we can calculate per-peer latency.
        let mut combined = all_keys.clone();
        combined.extend_from_slice(haves);
        combined.extend_from_slice(dont_haves);

        info!("recv_msg peer_manager");
        self.peer_manager().response_received(from, &combined).await;

        info!("recv_msg session_manager");
        // Send all block keys (including duplicates to any sessions that want them for accounting purposes).
        self.session_manager
            .receive_from(Some(*from), &all_keys, haves, dont_haves)
            .await;

        info!("recv_msg broadcast");
        // Publish the blocks
        for block in &wanted {
            if let Err(err) = self.notify.broadcast((*block).clone()).await {
                error!("failed to broadcast block {}: {:?}", block.cid(), err);
            }
        }
        if let Some(ref cb) = self.blocks_received_cb {
            (cb)(*from, wanted.into_iter().cloned().collect()).await;
        }

        info!("recv_msg end");
        Ok(())
    }

    /// Called by the network interface when a new message is received.
    pub async fn receive_message(&self, peer: &PeerId, incoming: &BitswapMessage) {
        inc!(BitswapMetrics::MessagesProcessingClient);

        if incoming.blocks_len() > 0 {
            debug!("client::receive_message {} blocks", incoming.blocks_len());

            for block in incoming.blocks() {
                debug!("recv block; {} from {}", block.cid(), peer);
                inc!(BitswapMetrics::BlocksIn);
                record!(
                    BitswapMetrics::ReceivedBlockBytes,
                    block.data().len() as u64
                );
            }
        }

        // TODO: investigate if the allocations below can be avoided.

        let haves: Vec<Cid> = incoming.haves().copied().collect();
        let dont_haves: Vec<Cid> = incoming.dont_haves().copied().collect();

        if incoming.blocks_len() > 0 || !haves.is_empty() || !dont_haves.is_empty() {
            // Process blocks
            if let Err(err) = self
                .receive_blocks_from(peer, incoming, &haves, &dont_haves)
                .await
            {
                warn!("ReceiveMessage recvBlockFrom error: {:?}", err);
            }
        }
    }

src/client/session.rs (line 319)

    pub async fn get_blocks(&self, keys: &[Cid]) -> Result<BlockReceiver> {
        ensure!(!keys.is_empty(), "missing keys");
        debug!("get blocks: {:?}", keys);

        let (s, r) = async_channel::bounded(8);
        let mut remaining: AHashSet<Cid> = keys.iter().copied().collect();
        let mut block_channel = self.inner.notify.new_receiver();
        let incoming = self.inner.incoming.clone();
        let (closer_s, mut closer_r) = oneshot::channel();
        let worker = tokio::task::spawn(async move {
            loop {
                inc!(BitswapMetrics::SessionGetBlockLoopTick);
                tokio::select! {
                    biased;
                    _ = &mut closer_r => {
                        // shutting down
                        break;
                    }
                    maybe_block = block_channel.recv() => {
                        match maybe_block {
                            Ok(block) => {
                                let cid = *block.cid();
                                if remaining.contains(&cid) {
                                    debug!("received wanted block {}", cid);
                                    match s.send(block).await {
                                        Ok(_) => {
                                            remaining.remove(&cid);
                                        }
                                        Err(_) => {
                                            // receiver dropped, shutdown
                                            break;
                                        }
                                    }
                                }

                                if remaining.is_empty() {
                                    debug!("found all requested blocks");
                                    break;
                                }
                            }
                            Err(async_broadcast::RecvError::Closed) => {
                                break;
                            }
                            Err(async_broadcast::RecvError::Overflowed(n)) => {
                                warn!("receiver is overflowing by {}", n);
                                continue;
                            }
                        }
                    }
                }
            }

            // cancel all remaining
            if !remaining.is_empty() {
                if let Err(err) = incoming
                    .send(Op::Cancel(remaining.into_iter().collect()))
                    .await
                {
                    warn!("failed to send cancel: {:?}", err);
                }
            }
        });

        self.inner.incoming.send(Op::Want(keys.to_vec())).await?;

        Ok(BlockReceiver {
            receiver: r,
            guard: BlockReceiverGuard {
                closer: Some(closer_s),
                _worker: worker,
            },
        })
    }

pub fn data(&self) -> &Bytes

Examples found in repository

src/message.rs (line 391)

    pub fn encode_as_proto_v0(&self) -> pb::Message {
        let mut message = pb::Message::default();

        // wantlist
        let mut wantlist = pb::message::Wantlist::default();
        for entry in self.wantlist.values() {
            wantlist.entries.push(entry.into());
        }
        wantlist.full = self.full;
        message.wantlist = Some(wantlist);

        // blocks
        for block in self.blocks.values() {
            message.blocks.push(block.data().clone());
        }

        message
    }

    pub fn encode_as_proto_v1(&self) -> pb::Message {
        let mut message = pb::Message::default();

        // wantlist
        let mut wantlist = pb::message::Wantlist::default();
        for entry in self.wantlist.values() {
            wantlist.entries.push(entry.into());
        }
        wantlist.full = self.full;
        message.wantlist = Some(wantlist);

        // blocks
        for block in self.blocks.values() {
            message.payload.push(pb::message::Block {
                prefix: Prefix::from(block.cid()).to_bytes(),
                data: block.data().clone(),
            });
        }

        // block presences
        for (cid, typ) in &self.block_presences {
            message.block_presences.push(pb::message::BlockPresence {
                cid: cid.to_bytes(),
                r#type: (*typ).into(),
            });
        }

        message.pending_bytes = self.pending_bytes();

        message
    }

src/server/decision.rs (line 463)

    pub async fn message_sent(&self, peer: &PeerId, message: &BitswapMessage) {
        let l = self.find_or_create(peer).await;
        let mut ledger = l.lock().await;

        // remove sent blocks from the want list for the peer
        for block in message.blocks() {
            self.score_ledger
                .add_to_sent_bytes(ledger.partner(), block.data().len())
                .await;
            ledger
                .wantlist_mut()
                .remove_type(block.cid(), WantType::Block);
        }

        // remove sent block presences from the wantlist for the peer
        for bp in message.block_presences() {
            // don't record sent data, we reserve that for data blocks
            if bp.typ == BlockPresenceType::Have {
                ledger.wantlist_mut().remove_type(&bp.cid, WantType::Have);
            }
        }
    }

    fn split_wants<'a>(
        &self,
        peer: &PeerId,
        entries: impl Iterator<Item = &'a Entry>,
    ) -> (Vec<&'a Entry>, Vec<&'a Entry>, Vec<&'a Entry>) {
        let mut wants = Vec::new();
        let mut cancels = Vec::new();
        let mut denials = Vec::new();

        for entry in entries {
            if entry.cancel {
                cancels.push(entry);
            } else if let Some(ref filter) = self.peer_block_request_filter {
                if (filter)(peer, &entry.cid) {
                    wants.push(entry);
                } else {
                    denials.push(entry);
                }
            } else {
                wants.push(entry);
            }
        }

        (wants, cancels, denials)
    }

    pub async fn received_blocks(&self, from: PeerId, blocks: Vec<Block>) {
        if blocks.is_empty() {
            return;
        }

        let l = self.find_or_create(&from).await;
        let ledger = l.lock().await;
        for block in blocks {
            self.score_ledger
                .add_to_recv_bytes(ledger.partner(), block.data().len())
                .await;
        }
    }

    pub async fn notify_new_blocks(&self, blocks: &[Block]) {
        if blocks.is_empty() {
            return;
        }

        // get the sizes of each block
        let block_sizes: AHashMap<_, _> = blocks
            .iter()
            .map(|block| (block.cid(), block.data().len()))
            .collect();

        let mut work = false;
        let mut missing_wants: AHashMap<PeerId, Vec<Cid>> = AHashMap::new();
        for block in blocks {
            let cid = block.cid();
            let peer_ledger = self.peer_ledger.lock().await;
            let peers = peer_ledger.peers(cid);
            if peers.is_none() {
                continue;
            }
            for peer in peers.unwrap() {
                let l = self.ledger_map.read().await.get(peer).cloned();
                if l.is_none() {
                    missing_wants.entry(*peer).or_default().push(*cid);
                    continue;
                }
                let l = l.unwrap();
                let ledger = l.lock().await;
                let entry = ledger.wantlist_get(cid);
                if entry.is_none() {
                    missing_wants.entry(*peer).or_default().push(*cid);
                    continue;
                }
                let entry = entry.unwrap();

                work = true;
                let block_size = block_sizes.get(cid).copied().unwrap_or_default();
                let is_want_block = self.send_as_block(entry.want_type, block_size);
                let entry_size = if is_want_block {
                    block_size
                } else {
                    BlockPresence::encoded_len_for_cid(*cid)
                };

                self.peer_task_queue
                    .push_task(
                        *peer,
                        Task {
                            topic: entry.cid,
                            priority: entry.priority as isize,
                            work: entry_size,
                            data: TaskData {
                                block_size,
                                have_block: true,
                                is_want_block,
                                send_dont_have: false,
                            },
                        },
                    )
                    .await;
                self.update_metrics().await;
            }
        }

        // If we found missing wants remove them from the list
        if !missing_wants.is_empty() {
            let ledger_map = self.ledger_map.read().await;
            let mut peer_ledger = self.peer_ledger.lock().await;
            for (peer, wants) in missing_wants.into_iter() {
                if let Some(l) = ledger_map.get(&peer) {
                    let ledger = l.lock().await;
                    for cid in wants {
                        if ledger.wantlist_get(&cid).is_some() {
                            continue;
                        }
                        peer_ledger.cancel_want(&peer, &cid);
                    }
                } else {
                    for cid in wants {
                        peer_ledger.cancel_want(&peer, &cid);
                    }
                }
            }
        }

        if work {
            self.signal_new_work();
        }
    }

src/client.rs (line 252)

    pub async fn receive_message(&self, peer: &PeerId, incoming: &BitswapMessage) {
        inc!(BitswapMetrics::MessagesProcessingClient);

        if incoming.blocks_len() > 0 {
            debug!("client::receive_message {} blocks", incoming.blocks_len());

            for block in incoming.blocks() {
                debug!("recv block; {} from {}", block.cid(), peer);
                inc!(BitswapMetrics::BlocksIn);
                record!(
                    BitswapMetrics::ReceivedBlockBytes,
                    block.data().len() as u64
                );
            }
        }

        // TODO: investigate if the allocations below can be avoided.

        let haves: Vec<Cid> = incoming.haves().copied().collect();
        let dont_haves: Vec<Cid> = incoming.dont_haves().copied().collect();

        if incoming.blocks_len() > 0 || !haves.is_empty() || !dont_haves.is_empty() {
            // Process blocks
            if let Err(err) = self
                .receive_blocks_from(peer, incoming, &haves, &dont_haves)
                .await
            {
                warn!("ReceiveMessage recvBlockFrom error: {:?}", err);
            }
        }
    }

Trait Implementations

impl Clone for Block

fn clone(&self) -> Block

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Block

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Ord for Block

fn cmp(&self, other: &Block) -> Ordering

This method returns an Ordering between self and other.

fn max(self, other: Self) -> Selfwhere
    Self: Sized,

Compares and returns the maximum of two values.

fn min(self, other: Self) -> Selfwhere
    Self: Sized,

Compares and returns the minimum of two values.

fn clamp(self, min: Self, max: Self) -> Selfwhere
    Self: Sized + PartialOrd<Self>,

Restrict a value to a certain interval.

impl PartialEq<Block> for Block

fn eq(&self, other: &Block) -> bool

This method tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

This method tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl PartialOrd<Block> for Block

fn partial_cmp(&self, other: &Block) -> Option<Ordering>

This method returns an ordering between self and other values if one exists.

fn lt(&self, other: &Rhs) -> bool

This method tests less than (for self and other) and is used by the < operator.

fn le(&self, other: &Rhs) -> bool

This method tests less than or equal to (for self and other) and is used by the <= operator.

fn gt(&self, other: &Rhs) -> bool

This method tests greater than (for self and other) and is used by the > operator.

fn ge(&self, other: &Rhs) -> bool

This method tests greater than or equal to (for self and other) and is used by the >= operator.

impl Eq for Block

impl StructuralEq for Block

impl StructuralPartialEq for Block