feat: reduce cpu drag by making channel drop logging optional

Author: aaronlifton

README.md

@@ -6,8 +6,8 @@ core runtime.
 
 Current fastest speed, with default controls of `max-depth` 4, `max-links-per-page` 16, `politeness-ms` 250,
 `partition-strategy` 'wiki-prefix' (instead of 'hash'), `partition-buckets` 26, `remote-batch-size` 32, and
-`duration-secs` 4 (it crawls for 4 seconds, but any enqued link is still awaited, so it runs for approximately 26
-seconds.) is **71.95 pages/sec**.
+`duration-secs` 4 (it crawls for 4 seconds, but any enqued link is still awaited, so it ran for 26.61s) is **75.12
+pages/sec**.
 
 ## Metrics
 
@@ -89,6 +89,8 @@ cargo run --example wiki --features multi_thread -- --duration-secs 60
   the shards, and `--partition-namespace` to keep namespaces like `Talk:` or `Help:` on stable shards.
 - Tune `--remote-batch-size <n>` (default 8) to control how many cross-shard links get buffered before the router sends
   them; higher values reduce channel wakeups at the cost of slightly delayed enqueues on the destination shard.
+- Enable `--remote-channel-logs` only when debugging channel shutdowns; it reintroduces the verbose “remote shard …
+  closed” logs.
 
 ## Customizing Crawls
 

src/controls.rs

@@ -110,6 +110,11 @@ pub struct Cli {
     #[cfg(feature = "multi_thread")]
     #[arg(long, env = "FASTCRAWL_REMOTE_BATCH_SIZE", default_value_t = 0)]
     pub remote_batch_size: usize,
+
+    /// Emit per-shard channel-closed logs (helps debugging shutdown races).
+    #[cfg(feature = "multi_thread")]
+    #[arg(long, env = "FASTCRAWL_REMOTE_CHANNEL_LOGS", default_value_t = false)]
+    pub remote_channel_logs: bool,
 }
 
 impl Cli {
@@ -144,6 +149,7 @@ impl Cli {
             wiki_bucket_count: (self.partition_buckets > 0).then_some(self.partition_buckets),
             wiki_include_namespace: self.partition_namespace,
             remote_batch_size: (self.remote_batch_size > 0).then_some(self.remote_batch_size),
+            remote_channel_logs: self.remote_channel_logs,
         }
     }
 }
@@ -170,4 +176,6 @@ pub struct PartitionSettings {
     pub wiki_include_namespace: bool,
     /// Optional remote batch size override.
     pub remote_batch_size: Option<usize>,
+    /// Whether to log channel-closed warnings for cross-shard sends.
+    pub remote_channel_logs: bool,
 }

src/runtime.rs

@@ -302,6 +302,7 @@ fn run_multi_thread(cli: Cli, seeds: &[&str], filter: UrlFilter) -> Result<(), D
         .remote_batch_size
         .unwrap_or(DEFAULT_REMOTE_BATCH_SIZE)
         .max(1);
+    let remote_channel_logs = partition_settings.remote_channel_logs;
     let seeds_owned: Vec<String> = seeds.iter().map(|s| s.to_string()).collect();
     let start = Instant::now();
 
@@ -335,6 +336,7 @@ fn run_multi_thread(cli: Cli, seeds: &[&str], filter: UrlFilter) -> Result<(), D
                 senders_clone,
                 receiver,
                 remote_batch_size,
+                remote_channel_logs,
             )))
         })?);
     }
@@ -359,6 +361,7 @@ async fn run_shard_streaming(
     remotes: Arc<Vec<mpsc::Sender<LinkBatch>>>,
     inbox: mpsc::Receiver<LinkBatch>,
     remote_batch_size: usize,
+    remote_channel_logs: bool,
 ) -> Result<(), DynError> {
     let state = AppState::new_with_shared(filter, &shared)?;
     seed_partitioned_frontier(state.frontier.as_ref(), &seeds, &partition).await;
@@ -383,7 +386,12 @@ async fn run_shard_streaming(
         })
     };
 
-    let router = Arc::new(ShardRouter::new(partition, remotes, remote_batch_size));
+    let router = Arc::new(ShardRouter::new(
+        partition,
+        remotes,
+        remote_batch_size,
+        remote_channel_logs,
+    ));
     let mut workers = Vec::new();
     for id in 0..DEFAULT_AGENT_CAPACITY {
         workers.push(spawn_streaming_worker_sharded(
@@ -828,6 +836,7 @@ struct ShardRouter {
     remotes: Arc<Vec<mpsc::Sender<LinkBatch>>>,
     batch_size: usize,
     buffers: Arc<RemoteBuffers>,
+    log_channel_drop: bool,
 }
 
 #[cfg(feature = "multi_thread")]
@@ -836,12 +845,14 @@ impl ShardRouter {
         partition: ShardPartition,
         remotes: Arc<Vec<mpsc::Sender<LinkBatch>>>,
         batch_size: usize,
+        log_channel_drop: bool,
     ) -> Self {
         Self {
             partition,
             remotes,
             batch_size: batch_size.max(1),
             buffers: Arc::new(RemoteBuffers::default()),
+            log_channel_drop,
         }
     }
 
@@ -876,10 +887,12 @@ impl ShardRouter {
         if let Some(sender) = self.remotes.get(shard) {
             let count = links.len();
             if sender.send(LinkBatch { depth, links }).await.is_err() {
-                eprintln!(
-                    "shard {}: remote shard {shard} channel closed",
-                    self.partition.index()
-                );
+                if self.log_channel_drop {
+                    eprintln!(
+                        "shard {}: remote shard {shard} channel closed",
+                        self.partition.index()
+                    );
+                }
             } else {
                 metrics.record_remote_shard_links(count);
             }