update

Author: luoluoyuyu

src/runtime/streaming/job/edge_manager.rs

@@ -12,10 +12,14 @@
 
 use std::collections::HashMap;
 
-use protocol::grpc::api::{FsEdge, FsNode};
+use anyhow::{Result, anyhow};
 use tokio::sync::mpsc;
+use tracing::{debug, info, warn};
 
 use crate::runtime::streaming::protocol::event::TrackedEvent;
+use protocol::grpc::api::{FsEdge, FsNode};
+
+const DEFAULT_CHANNEL_CAPACITY: usize = 2048;
 
 type TrackedEventEndpoints = (
     Vec<mpsc::Receiver<TrackedEvent>>,
@@ -28,34 +32,69 @@ pub struct EdgeManager {
 
 impl EdgeManager {
     pub fn build(nodes: &[FsNode], edges: &[FsEdge]) -> Self {
-        let mut tx_map: HashMap<u32, Vec<mpsc::Sender<TrackedEvent>>> = HashMap::new();
-        let mut rx_map: HashMap<u32, Vec<mpsc::Receiver<TrackedEvent>>> = HashMap::new();
+        Self::build_with_capacity(nodes, edges, DEFAULT_CHANNEL_CAPACITY)
+    }
+
+    pub fn build_with_capacity(nodes: &[FsNode], edges: &[FsEdge], capacity: usize) -> Self {
+        info!(
+            "Building EdgeManager for {} nodes and {} edges (channel capacity: {})",
+            nodes.len(),
+            edges.len(),
+            capacity
+        );
+
+        let mut tx_map: HashMap<u32, Vec<mpsc::Sender<TrackedEvent>>> =
+            HashMap::with_capacity(nodes.len());
+        let mut rx_map: HashMap<u32, Vec<mpsc::Receiver<TrackedEvent>>> =
+            HashMap::with_capacity(nodes.len());
 
         for edge in edges {
-            let (tx, rx) = mpsc::channel(2048);
-            tx_map.entry(edge.source as u32).or_default().push(tx);
-            rx_map.entry(edge.target as u32).or_default().push(rx);
+            let source_id = edge.source as u32;
+            let target_id = edge.target as u32;
+
+            let (tx, rx) = mpsc::channel(capacity);
+
+            tx_map.entry(source_id).or_default().push(tx);
+            rx_map.entry(target_id).or_default().push(rx);
+
+            debug!(
+                "Created physical edge channel: Node {} -> Node {}",
+                source_id, target_id
+            );
         }
 
-        let mut endpoints = HashMap::new();
+        let mut endpoints = HashMap::with_capacity(nodes.len());
         for node in nodes {
             let id = node.node_index as u32;
+
             let inboxes = rx_map.remove(&id).unwrap_or_default();
-            endpoints.insert(id, (inboxes, tx_map.remove(&id).unwrap_or_default()));
+            let outboxes = tx_map.remove(&id).unwrap_or_default();
+
+            endpoints.insert(id, (inboxes, outboxes));
+        }
+
+        for remaining_target in rx_map.keys() {
+            warn!(
+                "Topology Warning: Found incoming edges pointing to non-existent Node {}",
+                remaining_target
+            );
+        }
+        for remaining_source in tx_map.keys() {
+            warn!(
+                "Topology Warning: Found outgoing edges coming from non-existent Node {}",
+                remaining_source
+            );
         }
 
         Self { endpoints }
     }
 
-    pub fn take_endpoints(
-        &mut self,
-        id: u32,
-    ) -> (
-        Vec<mpsc::Receiver<TrackedEvent>>,
-        Vec<mpsc::Sender<TrackedEvent>>,
-    ) {
+    pub fn take_endpoints(&mut self, id: u32) -> Result<TrackedEventEndpoints> {
         self.endpoints
             .remove(&id)
-            .expect("Critical: Execution Graph Inconsistent")
+            .ok_or_else(|| anyhow!(
+                "Topology Error: Endpoints for Node {} not found or already taken. Execution Graph may be inconsistent.",
+                id
+            ))
     }
 }

src/runtime/streaming/job/job_manager.rs

@@ -13,7 +13,7 @@
 use std::collections::HashMap;
 use std::sync::{Arc, OnceLock, RwLock};
 
-use anyhow::{Context, Result, anyhow, bail};
+use anyhow::{Context, Result, anyhow, bail, ensure};
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::{error, info, warn};
@@ -330,40 +330,52 @@ impl JobManager {
         operators: &[ChainedOperator],
         edge_manager: &mut EdgeManager,
     ) -> Result<PhysicalPipeline> {
-        let (raw_inboxes, raw_outboxes) = edge_manager.take_endpoints(pipeline_id);
+        let (raw_inboxes, raw_outboxes) =
+            edge_manager.take_endpoints(pipeline_id).with_context(|| {
+                format!(
+                    "Failed to retrieve network endpoints for pipeline {}",
+                    pipeline_id
+                )
+            })?;
 
-        let physical_outboxes = raw_outboxes
+        let physical_outboxes: Vec<PhysicalSender> = raw_outboxes
             .into_iter()
             .map(PhysicalSender::Local)
             .collect();
+
         let physical_inboxes: Vec<BoxedEventStream> = raw_inboxes
             .into_iter()
             .map(|rx| Box::pin(ReceiverStream::new(rx)) as _)
             .collect();
 
-        let chain = self.build_operator_chain(operators)?;
-
-        if chain.source.is_none() && physical_inboxes.is_empty() {
-            bail!(
-                "Topology Error: pipeline '{}' contains no source and no upstream inputs.",
-                pipeline_id
-            );
-        }
-        if chain.source.is_some() && !physical_inboxes.is_empty() {
-            bail!(
-                "Topology Error: source pipeline '{}' should not have upstream inputs.",
+        let chain = self.build_operator_chain(operators).with_context(|| {
+            format!(
+                "Failed to build operator chain for pipeline {}",
                 pipeline_id
-            );
-        }
+            )
+        })?;
+
+        ensure!(
+            chain.source.is_some() || !physical_inboxes.is_empty(),
+            "Topology Error: Pipeline '{}' contains no source and has no upstream inputs (Dead end).",
+            pipeline_id
+        );
+        ensure!(
+            chain.source.is_none() || physical_inboxes.is_empty(),
+            "Topology Error: Source pipeline '{}' cannot have upstream inputs.",
+            pipeline_id
+        );
 
         let (control_tx, control_rx) = mpsc::channel(64);
         let status = Arc::new(RwLock::new(PipelineStatus::Initializing));
 
+        let subtask_index = 0;
+        let parallelism = 1;
         let ctx = TaskContext::new(
             job_id.clone(),
             pipeline_id,
-            0,
-            1,
+            subtask_index,
+            parallelism,
             physical_outboxes,
             Arc::clone(&self.memory_pool),
         );
@@ -373,12 +385,15 @@ impl JobManager {
             PipelineRunner::Source(SourceDriver::new(source, chain_head, ctx, control_rx))
         } else {
             PipelineRunner::Standard(
-                Pipeline::new(chain.operators, ctx, physical_inboxes, control_rx)
-                    .map_err(|e| anyhow!("Pipeline init failed: {e}"))?,
+                Pipeline::new(chain.operators, ctx, physical_inboxes, control_rx).with_context(
+                    || format!("Failed to initialize Standard Pipeline {}", pipeline_id),
+                )?,
             )
         };
 
-        let handle = self.spawn_worker_thread(job_id, pipeline_id, runner, Arc::clone(&status))?;
+        let handle = self
+            .spawn_worker_thread(job_id, pipeline_id, runner, Arc::clone(&status))
+            .with_context(|| format!("Failed to spawn OS thread for pipeline {}", pipeline_id))?;
 
         Ok(PhysicalPipeline {
             pipeline_id,