Murzaby/sniffnet
1
0
Fork 0
mirror of https://github.com/GyulyVGC/sniffnet.git synced 2026-10-21 22:51:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

use a tokio broadcast channel to block both parsing packet routines

Browse source
This commit is contained in:
GyulyVGC 2025-10-31 23:06:13 +01:00
parent 8afd77c94f
commit a20162d37c
2 changed files with 17 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

7
src/gui/sniffer.rs
View file

@ -124,7 +124,7 @@ pub struct Sniffer {
/// Flag reporting whether the packet capture is frozen
pub frozen: bool,
/// Sender to freeze the packet capture
freeze_tx: Option<std::sync::mpsc::Sender<()>>,
freeze_tx: Option<tokio::sync::broadcast::Sender<()>>,
}
impl Sniffer {
@ -772,7 +772,8 @@ impl Sniffer {
self.traffic_chart
.change_capture_source(matches!(capture_source, CaptureSource::Device(_)));
let (tx, rx) = async_channel::unbounded();
let (freeze_tx, freeze_rx) = std::sync::mpsc::channel();
let (freeze_tx, freeze_rx) = tokio::sync::broadcast::channel(65535);
let freeze_rx2 = freeze_tx.subscribe();
let filters = self.conf.filters.clone();
let _ = thread::Builder::new()
.name("thread_parse_packets".to_string())
@ -784,7 +785,7 @@ impl Sniffer {
capture_context,
filters,
&tx,
freeze_rx,
(freeze_rx, freeze_rx2),
);
})
.log_err(location!());

26
src/networking/parse_packets.rs
View file

@ -30,10 +30,10 @@ use etherparse::{EtherType, LaxPacketHeaders};
use pcap::{Address, Packet, PacketHeader};
use std::collections::HashMap;
use std::net::IpAddr;
use std::sync::mpsc::Receiver;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
use tokio::sync::broadcast::Receiver;
/// The calling thread enters a loop in which it waits for network packets
pub fn parse_packets(
@ -43,8 +43,10 @@ pub fn parse_packets(
capture_context: CaptureContext,
filters: Filters,
tx: &Sender<BackendTrafficMessage>,
freeze_rx: Receiver<()>,
freeze_rxs: (Receiver<()>, Receiver<()>),
) {
let (mut freeze_rx, mut freeze_rx2) = freeze_rxs;
let my_link_type = capture_context.my_link_type();
let (cap, mut savefile) = capture_context.consume();
@ -59,19 +61,17 @@ pub fn parse_packets(
let (pcap_tx, pcap_rx) = std::sync::mpsc::sync_channel(10_000);
let _ = thread::Builder::new()
.name("thread_packet_stream".to_string())
.spawn(move || packet_stream(cap, &pcap_tx, &freeze_rx, &filters))
.spawn(move || packet_stream(cap, &pcap_tx, &mut freeze_rx2, &filters))
.log_err(location!());
loop {
// // check if we need to freeze the parsing
// if freeze_rx.try_recv().is_ok() {
// // wait until unfreeze
// let _ = freeze_rx.recv();
// // discard old buffered packets
//
// // reset the first packet ticks
// first_packet_ticks = Some(Instant::now());
// }
if freeze_rx.try_recv().is_ok() {
// wait until unfreeze
let _ = freeze_rx.blocking_recv();
// reset the first packet ticks
first_packet_ticks = Some(Instant::now());
}
let (packet_res, cap_stats) = pcap_rx
.recv_timeout(Duration::from_millis(150))
@ -500,7 +500,7 @@ fn maybe_send_tick_run_offline(
fn packet_stream(
mut cap: CaptureType,
tx: &std::sync::mpsc::SyncSender<(Result<PacketOwned, pcap::Error>, Option<pcap::Stat>)>,
freeze_rx: &Receiver<()>,
freeze_rx: &mut Receiver<()>,
filters: &Filters,
) {
loop {
@ -509,7 +509,7 @@ fn packet_stream(
// pause the capture
cap.pause();
// wait until unfreeze
let _ = freeze_rx.recv();
let _ = freeze_rx.blocking_recv();
// resume the capture
cap.resume(filters);
}