Fix TCP send: server sends 12-byte status messages when receiving
All checks were successful
CI / test (push) Successful in 1m19s
All checks were successful
CI / test (push) Successful in 1m19s
pcap of MikroTik-as-server showed it sends periodic 12-byte status messages back to the client even in RX-only mode. The client needs these to display speed. Added tcp_status_sender that writes status messages containing rx_bytes on the TCP write half every 1 second. Reverted the "always bidirectional" change — TCP direction is conditional, but RX mode now uses the writer for status instead of keeping it idle. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
104
src/server.rs
104
src/server.rs
@@ -293,53 +293,83 @@ async fn handle_client(
|
|||||||
async fn run_tcp_test_server(stream: TcpStream, cmd: Command) -> Result<()> {
|
async fn run_tcp_test_server(stream: TcpStream, cmd: Command) -> Result<()> {
|
||||||
let state = BandwidthState::new();
|
let state = BandwidthState::new();
|
||||||
let tx_size = cmd.tx_size as usize;
|
let tx_size = cmd.tx_size as usize;
|
||||||
|
let server_should_tx = cmd.server_tx();
|
||||||
|
let server_should_rx = cmd.server_rx();
|
||||||
let tx_speed = cmd.remote_tx_speed;
|
let tx_speed = cmd.remote_tx_speed;
|
||||||
|
|
||||||
let (reader, writer) = stream.into_split();
|
let (reader, writer) = stream.into_split();
|
||||||
|
|
||||||
// TCP mode: ALWAYS start both TX and RX threads.
|
let mut _writer_keepalive = None;
|
||||||
// MikroTik expects bidirectional data flow on TCP regardless of direction.
|
let mut _reader_keepalive = None;
|
||||||
// The direction only controls what gets measured/reported.
|
|
||||||
let state_tx = state.clone();
|
let state_tx = state.clone();
|
||||||
let tx_handle = tokio::spawn(async move {
|
let tx_handle = if server_should_tx {
|
||||||
tcp_tx_loop(writer, tx_size, tx_speed, state_tx).await
|
Some(tokio::spawn(async move {
|
||||||
});
|
tcp_tx_loop(writer, tx_size, tx_speed, state_tx).await
|
||||||
|
}))
|
||||||
|
} else if server_should_rx {
|
||||||
|
// Server receives: use the writer to send periodic status messages
|
||||||
|
// back to the client (MikroTik needs this to show speed)
|
||||||
|
let st = state.clone();
|
||||||
|
Some(tokio::spawn(async move {
|
||||||
|
tcp_status_sender(writer, st).await
|
||||||
|
}))
|
||||||
|
} else {
|
||||||
|
_writer_keepalive = Some(writer);
|
||||||
|
None
|
||||||
|
};
|
||||||
|
|
||||||
let state_rx = state.clone();
|
let state_rx = state.clone();
|
||||||
let rx_handle = tokio::spawn(async move {
|
let rx_handle = if server_should_rx {
|
||||||
tcp_rx_loop(reader, state_rx).await
|
Some(tokio::spawn(async move {
|
||||||
});
|
tcp_rx_loop(reader, state_rx).await
|
||||||
|
}))
|
||||||
|
} else {
|
||||||
|
_reader_keepalive = Some(reader);
|
||||||
|
None
|
||||||
|
};
|
||||||
|
|
||||||
status_report_loop(&cmd, &state).await;
|
status_report_loop(&cmd, &state).await;
|
||||||
|
|
||||||
state.running.store(false, Ordering::SeqCst);
|
state.running.store(false, Ordering::SeqCst);
|
||||||
let _ = tx_handle.await;
|
if let Some(h) = tx_handle { let _ = h.await; }
|
||||||
let _ = rx_handle.await;
|
if let Some(h) = rx_handle { let _ = h.await; }
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// TCP multi-connection: always TX+RX on all streams.
|
/// TCP multi-connection.
|
||||||
async fn run_tcp_multiconn_server(streams: Vec<TcpStream>, cmd: Command) -> Result<()> {
|
async fn run_tcp_multiconn_server(streams: Vec<TcpStream>, cmd: Command) -> Result<()> {
|
||||||
let state = BandwidthState::new();
|
let state = BandwidthState::new();
|
||||||
let tx_size = cmd.tx_size as usize;
|
let tx_size = cmd.tx_size as usize;
|
||||||
|
let server_should_tx = cmd.server_tx();
|
||||||
|
let server_should_rx = cmd.server_rx();
|
||||||
let tx_speed = cmd.remote_tx_speed;
|
let tx_speed = cmd.remote_tx_speed;
|
||||||
|
|
||||||
let mut tx_handles = Vec::new();
|
let mut tx_handles = Vec::new();
|
||||||
let mut rx_handles = Vec::new();
|
let mut rx_handles = Vec::new();
|
||||||
|
let mut _writer_keepalives: Vec<tokio::net::tcp::OwnedWriteHalf> = Vec::new();
|
||||||
|
let mut _reader_keepalives: Vec<tokio::net::tcp::OwnedReadHalf> = Vec::new();
|
||||||
|
|
||||||
// TCP: always start both TX and RX on every stream
|
|
||||||
for tcp_stream in streams {
|
for tcp_stream in streams {
|
||||||
let (reader, writer) = tcp_stream.into_split();
|
let (reader, writer) = tcp_stream.into_split();
|
||||||
|
|
||||||
let st = state.clone();
|
if server_should_tx {
|
||||||
tx_handles.push(tokio::spawn(async move {
|
let st = state.clone();
|
||||||
tcp_tx_loop(writer, tx_size, tx_speed, st).await
|
tx_handles.push(tokio::spawn(async move {
|
||||||
}));
|
tcp_tx_loop(writer, tx_size, tx_speed, st).await
|
||||||
|
}));
|
||||||
|
} else {
|
||||||
|
_writer_keepalives.push(writer);
|
||||||
|
}
|
||||||
|
|
||||||
let st = state.clone();
|
if server_should_rx {
|
||||||
rx_handles.push(tokio::spawn(async move {
|
let st = state.clone();
|
||||||
tcp_rx_loop(reader, st).await
|
rx_handles.push(tokio::spawn(async move {
|
||||||
}));
|
tcp_rx_loop(reader, st).await
|
||||||
|
}));
|
||||||
|
} else {
|
||||||
|
_reader_keepalives.push(reader);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
@@ -413,6 +443,38 @@ async fn tcp_rx_loop(mut reader: tokio::net::tcp::OwnedReadHalf, state: Arc<Band
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Send periodic 12-byte status messages on the TCP connection.
|
||||||
|
/// Used when server is in RX mode — tells the client how many bytes we received.
|
||||||
|
async fn tcp_status_sender(
|
||||||
|
mut writer: tokio::net::tcp::OwnedWriteHalf,
|
||||||
|
state: Arc<BandwidthState>,
|
||||||
|
) {
|
||||||
|
let mut seq: u32 = 0;
|
||||||
|
let mut interval = tokio::time::interval(Duration::from_secs(1));
|
||||||
|
interval.tick().await; // consume first immediate tick
|
||||||
|
|
||||||
|
while state.running.load(Ordering::Relaxed) {
|
||||||
|
interval.tick().await;
|
||||||
|
if !state.running.load(Ordering::Relaxed) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
seq += 1;
|
||||||
|
let rx_bytes = state.rx_bytes.load(Ordering::Relaxed);
|
||||||
|
|
||||||
|
let status = StatusMessage {
|
||||||
|
seq,
|
||||||
|
bytes_received: rx_bytes as u32,
|
||||||
|
};
|
||||||
|
|
||||||
|
if writer.write_all(&status.serialize()).await.is_err() {
|
||||||
|
state.running.store(false, Ordering::SeqCst);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
let _ = writer.flush().await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// --- UDP Test Server ---
|
// --- UDP Test Server ---
|
||||||
|
|
||||||
async fn run_udp_test_server(
|
async fn run_udp_test_server(
|
||||||
|
|||||||
Reference in New Issue
Block a user