huginn_net_http/process/
parallel.rs

1//! Parallel processing support for HTTP analysis using worker pool architecture.
2//!
3//! This module provides multi-threaded packet processing with hash-based worker assignment
4//! to maintain HTTP flow consistency (request/response tracking). Unlike TCP which hashes
5//! only the source IP, HTTP hashes the complete flow (src_ip, dst_ip, src_port, dst_port)
6//! to ensure requests and responses from the same connection are processed by the same worker.
7
8use super::flow::{FlowKey, HttpProcessors, TcpFlow};
9use crate::error::HuginnNetHttpError;
10use crate::filter::raw as raw_filter;
11use crate::filter::FilterConfig;
12use crate::matcher_api::HttpMatcher;
13use crate::output::HttpAnalysisResult;
14use crate::parser::hash as packet_hash;
15use crossbeam_channel::{bounded, Sender};
16use std::fmt;
17use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
18use std::sync::{Arc, Mutex};
19use std::thread;
20use tracing::debug;
21use ttl_cache::TtlCache;
22
23/// Shared, thread-safe HTTP matcher used by [`WorkerPool`] and
24/// [`crate::HuginnNetHttp`]. The reference implementation is provided by
25/// `huginn-net-db` (`SharedHttpSignatureMatcher`).
26pub type SharedHttpMatcher = Arc<dyn HttpMatcher + Send + Sync>;
27
28/// Worker configuration parameters
29struct WorkerConfig {
30    batch_size: usize,
31    timeout_ms: u64,
32    max_connections: usize,
33}
34
35/// Worker pool for parallel HTTP packet processing.
36pub struct WorkerPool {
37    packet_senders: Arc<Vec<Sender<Vec<u8>>>>,
38    result_sender: Arc<Mutex<Option<std::sync::mpsc::Sender<HttpAnalysisResult>>>>,
39    shutdown_flag: Arc<AtomicBool>,
40    dispatched_count: AtomicU64,
41    dropped_count: AtomicU64,
42    worker_dropped: Vec<Arc<AtomicU64>>,
43    num_workers: std::num::NonZeroUsize,
44    pub batch_size: usize,
45    pub timeout_ms: u64,
46}
47
48/// Statistics for a single worker thread.
49#[derive(Debug, Clone)]
50pub struct WorkerStats {
51    pub id: usize,
52    pub queue_size: usize,
53    pub dropped: u64,
54}
55
56/// Pool-level statistics.
57#[derive(Debug, Clone)]
58pub struct PoolStats {
59    pub total_dispatched: u64,
60    pub total_dropped: u64,
61    pub workers: Vec<WorkerStats>,
62}
63
64/// Result of dispatching a packet to a worker.
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66pub enum DispatchResult {
67    /// Packet successfully queued for processing
68    Queued,
69    /// Worker queue full, packet dropped
70    Dropped,
71}
72
73impl WorkerPool {
74    /// Creates a new worker pool for HTTP analysis.
75    ///
76    /// # Parameters
77    /// - `num_workers`: Number of worker threads
78    /// - `queue_size`: Size of each worker's packet queue
79    /// - `batch_size`: Maximum packets to process in one batch
80    /// - `timeout_ms`: Worker receive timeout in milliseconds
81    /// - `result_sender`: Channel to send analysis results
82    /// - `matcher`: Optional shared signature matcher
83    /// - `max_connections`: Maximum HTTP flows to track per worker
84    /// - `filter_config`: Optional filter configuration for packet filtering
85    ///
86    /// # Returns
87    /// A new `WorkerPool` or an error if creation fails.
88    #[allow(clippy::too_many_arguments)]
89    pub fn new(
90        num_workers: usize,
91        queue_size: usize,
92        batch_size: usize,
93        timeout_ms: u64,
94        result_sender: std::sync::mpsc::Sender<HttpAnalysisResult>,
95        matcher: Option<SharedHttpMatcher>,
96        max_connections: usize,
97        filter_config: Option<FilterConfig>,
98    ) -> Result<Arc<Self>, HuginnNetHttpError> {
99        let num_workers_nz = std::num::NonZeroUsize::new(num_workers).ok_or_else(|| {
100            HuginnNetHttpError::Misconfiguration("Worker count must be at least 1".to_string())
101        })?;
102
103        let mut packet_senders = Vec::with_capacity(num_workers);
104        let worker_dropped: Vec<Arc<AtomicU64>> = (0..num_workers)
105            .map(|_| Arc::new(AtomicU64::new(0)))
106            .collect();
107        let shutdown_flag = Arc::new(AtomicBool::new(false));
108
109        for (worker_id, dropped_slot) in worker_dropped.iter().enumerate() {
110            let (tx, rx) = bounded::<Vec<u8>>(queue_size);
111            packet_senders.push(tx);
112
113            let result_sender_clone = result_sender.clone();
114            let matcher_clone = matcher.clone();
115            let shutdown_flag_clone = Arc::clone(&shutdown_flag);
116            let worker_filter = filter_config.clone();
117            let dropped_clone = Arc::clone(dropped_slot);
118
119            thread::Builder::new()
120                .name(format!("http-worker-{worker_id}"))
121                .spawn(move || {
122                    Self::worker_loop(
123                        worker_id,
124                        rx,
125                        result_sender_clone,
126                        matcher_clone,
127                        dropped_clone,
128                        shutdown_flag_clone,
129                        WorkerConfig { batch_size, timeout_ms, max_connections },
130                        worker_filter,
131                    )
132                })
133                .map_err(|e| {
134                    HuginnNetHttpError::Misconfiguration(format!(
135                        "Failed to spawn worker thread {worker_id}: {e}"
136                    ))
137                })?;
138        }
139
140        Ok(Arc::new(Self {
141            packet_senders: Arc::new(packet_senders),
142            result_sender: Arc::new(Mutex::new(Some(result_sender))),
143            shutdown_flag,
144            dispatched_count: AtomicU64::new(0),
145            dropped_count: AtomicU64::new(0),
146            worker_dropped,
147            num_workers: num_workers_nz,
148            batch_size,
149            timeout_ms,
150        }))
151    }
152
153    /// Worker thread main loop with batching support.
154    #[allow(clippy::too_many_arguments)]
155    fn worker_loop(
156        worker_id: usize,
157        rx: crossbeam_channel::Receiver<Vec<u8>>,
158        result_sender: std::sync::mpsc::Sender<HttpAnalysisResult>,
159        matcher: Option<SharedHttpMatcher>,
160        dropped: Arc<AtomicU64>,
161        shutdown_flag: Arc<AtomicBool>,
162        config: WorkerConfig,
163        filter_config: Option<FilterConfig>,
164    ) {
165        use crossbeam_channel::RecvTimeoutError;
166        use std::time::Duration;
167
168        debug!("HTTP worker {} started", worker_id);
169
170        let mut http_flows = TtlCache::new(config.max_connections);
171        let http_processors = HttpProcessors::new();
172        let timeout = Duration::from_millis(config.timeout_ms);
173        let mut batch = Vec::with_capacity(config.batch_size);
174
175        loop {
176            if shutdown_flag.load(Ordering::Relaxed) {
177                debug!("HTTP worker {} received shutdown signal", worker_id);
178                break;
179            }
180
181            match rx.recv_timeout(timeout) {
182                Ok(packet) => {
183                    batch.push(packet);
184
185                    while batch.len() < config.batch_size {
186                        match rx.try_recv() {
187                            Ok(packet) => batch.push(packet),
188                            Err(_) => break,
189                        }
190                    }
191
192                    let matcher_ref: Option<&dyn HttpMatcher> =
193                        matcher.as_deref().map(|m| m as &dyn HttpMatcher);
194                    for packet in batch.drain(..) {
195                        match Self::process_packet(
196                            &packet,
197                            &mut http_flows,
198                            &http_processors,
199                            matcher_ref,
200                            filter_config.as_ref(),
201                        ) {
202                            Ok(result) => {
203                                if result_sender.send(result).is_err() {
204                                    debug!("HTTP worker {} result channel closed", worker_id);
205                                    return;
206                                }
207                            }
208                            Err(_) => {
209                                // Packet processing error inside this worker. Feed it into the
210                                // per-worker drop counter so it is visible via `PoolStats`.
211                                // (See also `dispatch()` for queue-overflow drops.)
212                                dropped.fetch_add(1, Ordering::Relaxed);
213                            }
214                        }
215                    }
216                }
217                Err(RecvTimeoutError::Timeout) => {
218                    if shutdown_flag.load(Ordering::Relaxed) {
219                        debug!("HTTP worker {} received shutdown signal", worker_id);
220                        break;
221                    }
222                    continue;
223                }
224                Err(RecvTimeoutError::Disconnected) => {
225                    debug!("HTTP worker {} channel disconnected", worker_id);
226                    break;
227                }
228            }
229        }
230
231        debug!("HTTP worker {} stopped", worker_id);
232    }
233
234    fn process_packet(
235        packet: &[u8],
236        http_flows: &mut TtlCache<FlowKey, TcpFlow>,
237        http_processors: &HttpProcessors,
238        matcher: Option<&dyn HttpMatcher>,
239        filter: Option<&FilterConfig>,
240    ) -> Result<HttpAnalysisResult, HuginnNetHttpError> {
241        if let Some(filter_cfg) = filter {
242            if !raw_filter::apply(packet, filter_cfg) {
243                debug!("Filtered out packet before parsing");
244                return Ok(HttpAnalysisResult::empty());
245            }
246        }
247
248        use crate::packet_parser::{parse_packet, IpPacket};
249        use crate::process;
250
251        match parse_packet(packet) {
252            IpPacket::Ipv4(ipv4) => {
253                process::process_ipv4_packet(&ipv4, http_flows, http_processors, matcher)
254            }
255            IpPacket::Ipv6(ipv6) => {
256                process::process_ipv6_packet(&ipv6, http_flows, http_processors, matcher)
257            }
258            IpPacket::None => Ok(HttpAnalysisResult::empty()),
259        }
260    }
261
262    /// Deterministically computes which worker would process `packet` based
263    /// on its flow hash. Useful for tests that want to assert routing
264    /// decisions without going through the actual queue.
265    pub fn worker_index_for_packet(&self, packet: &[u8]) -> usize {
266        packet_hash::hash_flow(packet, self.num_workers.get())
267    }
268
269    pub fn dispatch(&self, packet: Vec<u8>) -> DispatchResult {
270        if self.shutdown_flag.load(Ordering::Relaxed) {
271            self.dropped_count.fetch_add(1, Ordering::Relaxed);
272            return DispatchResult::Dropped;
273        }
274
275        let worker_id = self.worker_index_for_packet(&packet);
276
277        self.dispatched_count.fetch_add(1, Ordering::Relaxed);
278
279        if let Some(sender) = self.packet_senders.get(worker_id) {
280            match sender.try_send(packet) {
281                Ok(()) => DispatchResult::Queued,
282                Err(_) => {
283                    self.dropped_count.fetch_add(1, Ordering::Relaxed);
284                    self.worker_dropped[worker_id].fetch_add(1, Ordering::Relaxed);
285                    DispatchResult::Dropped
286                }
287            }
288        } else {
289            self.dropped_count.fetch_add(1, Ordering::Relaxed);
290            DispatchResult::Dropped
291        }
292    }
293
294    pub fn stats(&self) -> PoolStats {
295        let workers = self
296            .packet_senders
297            .iter()
298            .enumerate()
299            .map(|(id, sender)| WorkerStats {
300                id,
301                queue_size: sender.len(),
302                dropped: self.worker_dropped[id].load(Ordering::Relaxed),
303            })
304            .collect();
305
306        PoolStats {
307            total_dispatched: self.dispatched_count.load(Ordering::Relaxed),
308            total_dropped: self.dropped_count.load(Ordering::Relaxed),
309            workers,
310        }
311    }
312
313    /// Initiates graceful shutdown of the worker pool.
314    pub fn shutdown(&self) {
315        self.shutdown_flag.store(true, Ordering::Relaxed);
316
317        if let Ok(mut sender) = self.result_sender.lock() {
318            *sender = None;
319        }
320    }
321}
322
323impl fmt::Display for PoolStats {
324    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
325        writeln!(f, "HTTP Worker Pool Statistics:")?;
326        writeln!(f, "  Total dispatched: {}", self.total_dispatched)?;
327        writeln!(f, "  Total dropped: {}", self.total_dropped)?;
328        writeln!(f, "  Workers: {}", self.workers.len())?;
329        for worker in &self.workers {
330            writeln!(f, "    {worker}")?;
331        }
332        Ok(())
333    }
334}
335
336impl fmt::Display for WorkerStats {
337    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
338        write!(
339            f,
340            "Worker {}: queue_size={}, dropped={}",
341            self.id, self.queue_size, self.dropped
342        )
343    }
344}
huginn_net_http/process/parallel.rs

huginn_net_http/process/
parallel.rs
