refactor: Avoid having to build a giant vec from the store iterators

Filtering is now done one by one

protocols/kad/src/jobs.rs

@@ -62,6 +62,7 @@
 //! > out of the job to the consumer, where they can be dropped after being sent.
 
 use crate::record::{self, store::RecordStore, ProviderRecord, Record};
+use crate::store::{ProviderRecordsIter, RecordsIter};
 use futures::prelude::*;
 use futures_timer::Delay;
 use instant::Instant;
@@ -135,7 +136,7 @@ pub struct PutRecordJob {
     publish_interval: Option<Duration>,
     record_ttl: Option<Duration>,
     skipped: HashSet<record::Key>,
-    inner: PeriodicJob<vec::IntoIter<Record>>,
+    inner: PeriodicJob<RecordsIter>,
 }
 
 impl PutRecordJob {
@@ -197,32 +198,28 @@ impl PutRecordJob {
     {
         if self.inner.check_ready(cx, now) {
             let publish = self.next_publish.map_or(false, |t_pub| now >= t_pub);
-            let records = store
-                .records()
-                .filter_map(|r| {
-                    let is_publisher = r.publisher.as_ref() == Some(&self.local_id);
-                    if self.skipped.contains(&r.key) || (!publish && is_publisher) {
-                        None
-                    } else {
-                        let mut record = r;
-                        if publish && is_publisher {
-                            record.expires = record
-                                .expires
-                                .or_else(|| self.record_ttl.map(|ttl| now + ttl));
-                        }
-                        Some(record)
+            let mut skipped = Default::default();
+            std::mem::swap(&mut skipped, &mut self.skipped);
+            let record_ttl = self.record_ttl;
+            let local_id = self.local_id;
+            let records = Box::new(store.records().filter_map(move |r| {
+                let is_publisher = r.publisher.as_ref() == Some(&local_id);
+                if skipped.contains(&r.key) || (!publish && is_publisher) {
+                    None
+                } else {
+                    let mut record = r;
+                    if publish && is_publisher {
+                        record.expires = record.expires.or_else(|| record_ttl.map(|ttl| now + ttl));
                     }
-                })
-                .collect::<Vec<_>>()
-                .into_iter();
+                    Some(record)
+                }
+            }));
 
             // Schedule the next publishing run.
             if publish {
                 self.next_publish = self.publish_interval.map(|i| now + i);
             }
 
-            self.skipped.clear();
-
             self.inner.state = PeriodicJobState::Running(records);
         }
 
@@ -251,7 +248,7 @@ impl PutRecordJob {
 
 /// Periodic job for replicating provider records.
 pub struct AddProviderJob {
-    inner: PeriodicJob<vec::IntoIter<ProviderRecord>>,
+    inner: PeriodicJob<ProviderRecordsIter>,
 }
 
 impl AddProviderJob {
@@ -297,7 +294,7 @@ impl AddProviderJob {
         T: RecordStore,
     {
         if self.inner.check_ready(cx, now) {
-            let records = store.provided().collect::<Vec<_>>().into_iter();
+            let records = store.provided();
             self.inner.state = PeriodicJobState::Running(records);
         }
 

protocols/kad/src/record/store.rs

@@ -45,6 +45,9 @@ pub enum Error {
     ValueTooLarge,
 }
 
+pub type RecordsIter = Box<dyn Iterator<Item = Record> + Send + Sync + 'static>;
+pub type ProviderRecordsIter = Box<dyn Iterator<Item = ProviderRecord> + Send + Sync + 'static>;
+
 /// Trait for types implementing a record store.
 ///
 /// There are two types of records managed by a `RecordStore`:
@@ -64,9 +67,6 @@ pub enum Error {
 ///      to the closest nodes to the key.
 ///
 pub trait RecordStore {
-    type RecordsIter: Iterator<Item = Record>;
-    type ProvidedIter: Iterator<Item = ProviderRecord>;
-
     /// Gets a record from the store, given its key.
     fn get(&self, k: &Key) -> Option<Record>;
 
@@ -77,7 +77,7 @@ pub trait RecordStore {
     fn remove(&mut self, k: &Key);
 
     /// Gets an iterator over all (value-) records currently stored.
-    fn records(&self) -> Self::RecordsIter;
+    fn records(&self) -> RecordsIter;
 
     /// Adds a provider record to the store.
     ///
@@ -91,7 +91,7 @@ pub trait RecordStore {
 
     /// Gets an iterator over all stored provider records for which the
     /// node owning the store is itself the provider.
-    fn provided(&self) -> Self::ProvidedIter;
+    fn provided(&self) -> ProviderRecordsIter;
 
     /// Removes a provider record from the store.
     fn remove_provider(&mut self, k: &Key, p: &PeerId);

protocols/kad/src/record/store/memory.rs

@@ -23,8 +23,7 @@ use super::*;
 use crate::kbucket;
 use libp2p_core::PeerId;
 use smallvec::SmallVec;
-use std::collections::{hash_map, hash_set, HashMap, HashSet};
-use std::iter;
+use std::collections::{hash_map, HashMap, HashSet};
 
 /// In-memory implementation of a `RecordStore`.
 pub struct MemoryStore {
@@ -96,10 +95,6 @@ impl MemoryStore {
 }
 
 impl RecordStore for MemoryStore {
-    type RecordsIter = Box<dyn Iterator<Item = Record>>;
-
-    type ProvidedIter = Box<dyn Iterator<Item = ProviderRecord>>;
-
     fn get(&self, k: &Key) -> Option<Record> {
         self.records.get(k).cloned()
     }
@@ -130,7 +125,7 @@ impl RecordStore for MemoryStore {
         self.records.remove(k);
     }
 
-    fn records(&self) -> Self::RecordsIter {
+    fn records(&self) -> RecordsIter {
         Box::new(
             self.records
                 .values()
@@ -196,8 +191,14 @@ impl RecordStore for MemoryStore {
             .map_or_else(Vec::new, |ps| ps.clone().into_vec())
     }
 
-    fn provided(&self) -> Self::ProvidedIter {
-        Box::new(self.provided.iter().cloned())
+    fn provided(&self) -> ProviderRecordsIter {
+        Box::new(
+            self.provided
+                .iter()
+                .cloned()
+                .collect::<Vec<_>>()
+                .into_iter(),
+        )
     }
 
     fn remove_provider(&mut self, key: &Key, provider: &PeerId) {