Merge pull request paritytech#261 from subspace/client-archiving-tweaks

Just peek into root blocks to not avoid order of elements in LRU cache, tiny simplification

Author: nazar-pc

crates/sc-consensus-subspace/src/archiver.rs

@@ -75,7 +75,7 @@ where
 pub fn start_subspace_archiver<Block: BlockT, Client>(
     subspace_link: &SubspaceLink<Block>,
     client: Arc<Client>,
-    spawner: &impl sp_core::traits::SpawnNamed,
+    spawner: &impl sp_core::traits::SpawnEssentialNamed,
 ) where
     Client: ProvideRuntimeApi<Block>
         + BlockBackend<Block>
@@ -222,7 +222,7 @@ pub fn start_subspace_archiver<Block: BlockT, Client>(
         }
     }
 
-    spawner.spawn_blocking(
+    spawner.spawn_essential_blocking(
         "subspace-archiver",
         None,
         Box::pin({

crates/sc-consensus-subspace/src/lib.rs

@@ -568,7 +568,7 @@ impl<Block: BlockT> SubspaceLink<Block> {
     pub fn root_blocks_for_block(&self, block_number: NumberFor<Block>) -> Vec<RootBlock> {
         self.root_blocks
             .lock()
-            .get(&block_number)
+            .peek(&block_number)
             .cloned()
             .unwrap_or_default()
     }
@@ -1177,22 +1177,16 @@ where
         }
 
         let import_result = self.inner.import_block(block, new_cache).await?;
-        let (root_block_sender, mut root_block_receiver) = mpsc::channel(0);
+        let (root_block_sender, root_block_receiver) = mpsc::channel(0);
 
         self.imported_block_notification_sender
             .notify(move || (block_number, root_block_sender));
 
-        let next_block_number = block_number + One::one();
-        while let Some(root_block) = root_block_receiver.next().await {
-            {
-                let mut root_blocks = self.root_blocks.lock();
-                if let Some(list) = root_blocks.get_mut(&next_block_number) {
-                    list.push(root_block);
-                } else {
-                    root_blocks.put(next_block_number, vec![root_block]);
-                }
-            }
-        }
+        let root_blocks = root_block_receiver.collect().await;
+
+        self.root_blocks
+            .lock()
+            .put(block_number + One::one(), root_blocks);
 
         Ok(import_result)
     }

crates/sc-consensus-subspace/src/tests.rs

@@ -485,7 +485,11 @@ fn run_one_test(mutator: impl Fn(&mut TestHeader, Stage) + Send + Sync + 'static
 
         let task_manager = TaskManager::new(tokio_runtime.handle().clone(), None).unwrap();
 
-        super::start_subspace_archiver(&data.link, client.clone(), &task_manager.spawn_handle());
+        super::start_subspace_archiver(
+            &data.link,
+            client.clone(),
+            &task_manager.spawn_essential_handle(),
+        );
 
         let (archived_pieces_sender, archived_pieces_receiver) = oneshot::channel();
 
@@ -838,7 +842,7 @@ fn verify_slots_are_strictly_increasing() {
 //     let tokio_runtime = sc_cli::build_runtime().unwrap();
 //     let task_manager = TaskManager::new(tokio_runtime.handle().clone(), None).unwrap();
 //
-//     super::start_subspace_archiver(&data.link, client.clone(), &task_manager.spawn_handle());
+//     super::start_subspace_archiver(&data.link, client.clone(), &task_manager.spawn_essential_handle());
 //
 //     let mut archived_segment_notification_stream =
 //         data.link.archived_segment_notification_stream.subscribe();

crates/subspace-node/src/service.rs

@@ -167,7 +167,7 @@ pub fn new_partial(
     sc_consensus_subspace::start_subspace_archiver(
         &subspace_link,
         client.clone(),
-        &task_manager.spawn_handle(),
+        &task_manager.spawn_essential_handle(),
     );
 
     let slot_duration = subspace_link.config().slot_duration();