Preserve timeout precision and avoid unnecessary timer allocation

This commit is contained in:
bee committed 2026-10-01 18:08:36 +01:00
1 parent 397526df8a
commit 62ed937768
6 files changed
+140 -28

No files matched your search

Generated
+7
View File
@@ -544,6 +544,12 @@ version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988"
[[package]]
name = "futures-timer"
version = "3.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "af43fadb8a98512d547e37b4e92e0ced13e205c061b87b4623eff01d918d6968"
[[package]]
name = "futures-util"
version = "0.3.31"
@@ -1455,6 +1461,7 @@ dependencies = [
"dirs",
"env_logger",
"futures",
"futures-timer",
"gcd",
"hickory-resolver",
"itertools 0.15.0",
+1
View File
@@ -21,6 +21,7 @@ clap = { version = "4.6.1", features = ["derive", "wrap_help"] }
colored = "3.1.1"
tokio = { version = "1.53.1", features = ["rt", "net", "time"] }
futures = "0.3"
futures-timer = "3.0.4"
rlimit = "0.11.0"
log = "0.4.32"
env_logger = "0.11.10"
+5 -2
View File
@@ -10,6 +10,10 @@ use std::io;
use std::net::IpAddr;
use std::time::Duration;
// Benchmark the production timeout path with simulated I/O operations.
#[path = "../src/scanner/timeout.rs"]
mod timeout;
fn bench_address() {
let _addrs = ["127.0.0.1".parse::<IpAddr>().unwrap()];
}
@@ -52,14 +56,13 @@ fn old_payload_for_port(udp_map: &'static BTreeMap<Vec<u16>, Vec<u8>>, port: u16
// Exercise the scanner's bounded scheduling pattern without opening sockets.
// Ready operations model immediate replies; pending operations expire instead.
async fn simulated_operation(expires: bool) -> io::Result<()> {
tokio::time::timeout(Duration::from_millis(10), async {
timeout::io_timeout(Duration::from_millis(10), async {
if expires {
std::future::pending::<()>().await;
}
Ok(())
})
.await
.map_err(|error| io::Error::new(io::ErrorKind::TimedOut, error))?
}
async fn simulated_batch(expires: bool) {
+14 -7
View File
@@ -6,8 +6,7 @@ and [#944](https://github.com/bee-san/RustScan/issues/944), based on master at
## CLI behavior
TCP connections, UDP sockets, connection/receive timeouts, and the delay between
ports now use Tokio. The CLI creates a runtime on the calling thread for the
TCP connections and UDP sockets now use Tokio. The CLI creates a runtime on the calling thread for the
scan. `FuturesUnordered` still polls at most `batch_size` socket attempts at once;
concurrent sockets do not require a thread per socket or a task per socket.
@@ -16,15 +15,19 @@ runtime. Script execution runs after the runtime has been dropped. Port ordering
exclusions, retries, UDP payloads, closed-port reporting and CLI options retain
their existing implementations.
Timeouts and port intervals use the runtime-independent `futures-timer` crate.
An expired timer maps to `std::io::ErrorKind::TimedOut`. Socket errors retain
their original kind, so TCP refusal reporting and UDP timeout retries continue
to distinguish replies from failures. Successful TCP streams are converted to
standard streams, shut down in both directions, and dropped before another
attempt is polled.
Tokio timers have millisecond granularity, with potentially coarser resolution
on Windows. Very short timeouts and intervals can therefore take longer than
with the previous runtime; they are minimum waits, not precision deadlines.
Using Tokio's timers for short waits added approximately one millisecond per
interval in the first benchmark. `futures-timer` uses a shared helper thread and
OS waits, avoiding that extra rounding while keeping the async runtime responsive.
Dropping a scan cancels its timers; runtime shutdown never has to wait for a
sleeping blocking task. Immediately ready I/O operations avoid timer allocation,
and the time spent polling a pending operation is charged to its original timeout.
## Library callers
@@ -68,10 +71,14 @@ owns its own runtime and performs blocking lookups.
The direct Tokio dependency enables only `rt`, `net`, and `time`; `test-util` is
enabled for deterministic tests with a paused clock. Tokio was already present
through Hickory. The existing `futures` dependency continues to provide
`FuturesUnordered` and `StreamExt`.
`FuturesUnordered`, `StreamExt` and future polling. The new `futures-timer` 3.0.4
dependency has no transitive dependencies on native targets; its purpose is to
preserve short timeout and interval performance. `async-std` is absent from the
resolved dependency tree.
Regression tests cover successful operations (including zero-byte responses),
preservation of I/O errors, timeout cancellation, and interval timing with
preservation of I/O errors, timeout cancellation, immediate results at a zero
deadline, wakeups during the first poll, scan cancellation, and interval timing with
excluded or empty port sets. They use simulated operations and scanners with no
target addresses, following the repository's rule against network tests.
+85 -19
View File
@@ -5,7 +5,9 @@ use crate::tui::println_safe;
use log::debug;
mod socket_iterator;
mod timeout;
use socket_iterator::SocketIterator;
use timeout::io_timeout;
use colored::Colorize;
use futures::{stream::FuturesUnordered, StreamExt};
@@ -19,21 +21,7 @@ use std::{
sync::Arc,
time::Duration,
};
use tokio::{
net::{TcpStream, UdpSocket},
time::{sleep, timeout},
};
/// Keep I/O errors intact while mapping an expired Tokio timer to `TimedOut`.
/// Dropping the timed-out future also cancels its pending socket operation.
async fn io_timeout<T>(
duration: Duration,
operation: impl Future<Output = io::Result<T>>,
) -> io::Result<T> {
timeout(duration, operation)
.await
.map_err(|error| io::Error::new(io::ErrorKind::TimedOut, error))?
}
use tokio::net::{TcpStream, UdpSocket};
/// UDP payload lookup: port -> payload bytes
///
@@ -174,6 +162,18 @@ impl Scanner {
///
/// Must be awaited inside a Tokio runtime with I/O and time enabled.
pub async fn run_with_status(&self) -> Vec<PortStatus> {
// Tokio's millisecond rounding adds about 1 ms to each short interval.
// This cancellable delay keeps the requested spacing without blocking
// the async runtime or accumulating that rounding on every port.
self.run_with_delay(futures_timer::Delay::new).await
}
/// Supplying the delay lets tests verify spacing with a virtual clock.
async fn run_with_delay<D, F>(&self, delay: D) -> Vec<PortStatus>
where
D: Fn(Duration) -> F,
F: Future<Output = ()>,
{
let ports: Vec<u16> = self
.port_strategy
.order()
@@ -209,7 +209,7 @@ impl Scanner {
// before moving on to the next port.
for (i, port) in ports.iter().enumerate() {
if i > 0 {
sleep(self.interval).await;
delay(self.interval).await;
}
let sockets = SocketIterator::new(&self.ips, std::slice::from_ref(port));
self.scan_sockets(sockets, &udp_payloads, &mut found_sockets, &mut errors)
@@ -504,6 +504,37 @@ mod tests {
});
}
#[test]
fn ready_operation_wins_at_zero_timeout() {
test_runtime().block_on(async {
assert_eq!(
io_timeout(Duration::ZERO, async { Ok(7) }).await.unwrap(),
7
);
});
}
#[test]
fn timeout_preserves_wakes_from_the_first_poll() {
test_runtime().block_on(async {
let mut polled = false;
let operation = std::future::poll_fn(|context| {
if polled {
std::task::Poll::Ready(Ok(7))
} else {
polled = true;
context.waker().wake_by_ref();
std::task::Poll::Pending
}
});
assert_eq!(
io_timeout(Duration::from_secs(1), operation).await.unwrap(),
7
);
});
}
#[test]
fn expired_timeout_cancels_the_operation() {
use std::cell::Cell;
@@ -518,8 +549,8 @@ mod tests {
test_runtime().block_on(async {
let dropped = Cell::new(false);
let started = tokio::time::Instant::now();
let duration = Duration::from_millis(100);
let started = std::time::Instant::now();
let duration = Duration::from_millis(5);
let operation = async {
let _guard = DropGuard(&dropped);
std::future::pending::<io::Result<()>>().await
@@ -543,11 +574,46 @@ mod tests {
scanner.exclude_ports = vec![443];
let started = tokio::time::Instant::now();
assert!(scanner.run_with_status().await.is_empty());
assert!(scanner.run_with_delay(tokio::time::sleep).await.is_empty());
assert_eq!(started.elapsed(), interval * 2);
});
}
#[test]
fn dropping_a_scan_cancels_a_pending_interval() {
use std::cell::Cell;
use std::task::Poll;
struct PendingDelay<'a>(&'a Cell<bool>);
impl Future for PendingDelay<'_> {
type Output = ();
fn poll(self: std::pin::Pin<&mut Self>, _: &mut std::task::Context<'_>) -> Poll<()> {
Poll::Pending
}
}
impl Drop for PendingDelay<'_> {
fn drop(&mut self) {
self.0.set(true);
}
}
test_runtime().block_on(async {
let dropped = Cell::new(false);
let mut scanner = test_scanner().with_interval(Duration::from_secs(3600));
scanner.ips.clear();
scanner.port_strategy = PortStrategy::Manual(vec![80, 443]);
let mut scan = Box::pin(scanner.run_with_delay(|_| PendingDelay(&dropped)));
assert!(futures::poll!(scan.as_mut()).is_pending());
assert!(!dropped.get());
drop(scan);
assert!(dropped.get());
});
}
#[test]
fn zero_interval_and_empty_port_sets_do_not_wait() {
test_runtime().block_on(async {
+28
View File
@@ -0,0 +1,28 @@
use futures::future::{select, Either};
use std::{
future::Future,
io,
task::Poll,
time::{Duration, Instant},
};
/// Preserve I/O errors and cancel pending operations when their deadline expires.
/// Poll once before allocating a timer so immediate replies pay no timer cost.
pub(super) async fn io_timeout<T>(
duration: Duration,
operation: impl Future<Output = io::Result<T>>,
) -> io::Result<T> {
let started = Instant::now();
futures::pin_mut!(operation);
if let Poll::Ready(result) = futures::poll!(operation.as_mut()) {
return result;
}
// Charge socket setup to the original deadline. The timer uses OS waits
// without rounding each short timeout up to Tokio's next millisecond tick.
let delay = futures_timer::Delay::new(duration.saturating_sub(started.elapsed()));
match select(operation, delay).await {
Either::Left((result, _)) => result,
Either::Right(_) => Err(io::Error::new(io::ErrorKind::TimedOut, "future timed out")),
}
}