Speed up port exclusions and avoid redundant port copies (#954)

* test(scanner): preserve port filtering order and boundaries

* perf(scanner): reuse port lists and index large exclusions

* perf(scanner): use bitmap for dense exclusions

* bench(scanner): document production port preparation speedups

* ci(bench): diagnose repeated macOS listener misses

* ci(bench): capture socket errors immediately after listener misses

* perf(scanner): keep bitmap storage out of repeated async polls

* ci(bench): provision macOS TCP memory for repeated scans

.github/scan-benchmark/macos_tcp_memory.py

@@ -0,0 +1,57 @@
+"""Provision TCP memory for repeated scans on a disposable macOS CI runner."""
+
+import ctypes
+import json
+import os
+import subprocess
+
+
+class Statistics(ctypes.Structure):
+    # XNU's bsd/sys/mem_acct_private.h, struct memacct_statistics.
+    _fields_ = [("peak", ctypes.c_uint64), ("allocated", ctypes.c_int64),
+                ("softlimit", ctypes.c_uint64), ("hardlimit", ctypes.c_uint64),
+                ("name", ctypes.c_char * 16)]
+
+
+def tcp_memory(new_limit: int | None = None) -> dict:
+    """Read the TCP accounting object, optionally raising its hard limit."""
+    libc = ctypes.CDLL(None, use_errno=True)
+    libc.sysctlnametomib.argtypes = [ctypes.c_char_p, ctypes.POINTER(ctypes.c_int),
+                                   ctypes.POINTER(ctypes.c_size_t)]
+    libc.sysctl.argtypes = [ctypes.POINTER(ctypes.c_int), ctypes.c_uint,
+                           ctypes.c_void_p, ctypes.POINTER(ctypes.c_size_t),
+                           ctypes.c_void_p, ctypes.c_size_t]
+    mib = (ctypes.c_int * 24)()
+    count = ctypes.c_size_t(22)
+    if libc.sysctlnametomib(b"kern.memacct", mib, ctypes.byref(count)):
+        raise OSError(ctypes.get_errno(), os.strerror(ctypes.get_errno()))
+
+    def query(operation: int, index: int, output, replacement=None):
+        mib[count.value], mib[count.value + 1] = operation, index
+        length = ctypes.c_size_t(ctypes.sizeof(output))
+        if libc.sysctl(mib, count.value + 2, ctypes.byref(output), ctypes.byref(length),
+                       ctypes.byref(replacement) if replacement is not None else None,
+                       ctypes.sizeof(replacement) if replacement is not None else 0):
+            raise OSError(ctypes.get_errno(), os.strerror(ctypes.get_errno()))
+        return length.value
+
+    names = ctypes.create_string_buffer(8 * 16)
+    length = query(5, 0, names)  # MEM_ACCT_SUBSYSTEMS
+    index = next(i for i in range(length // 16)
+                 if names.raw[i * 16:(i + 1) * 16].rstrip(b"\0") == b"TCP")
+    stats = Statistics()
+    query(6, index, stats)  # MEM_ACCT_ALL_SUBSYSTEM_STATISTICS
+    previous_limit = stats.hardlimit
+    if new_limit is not None and 0 < previous_limit < new_limit:
+        query(3, index, ctypes.c_uint64(), ctypes.c_uint64(new_limit))
+        query(6, index, stats)
+    return {"previous_hardlimit": previous_limit,
+            **{name: getattr(stats, name) for name in ("peak", "allocated", "softlimit", "hardlimit")}}
+
+
+if __name__ == "__main__":
+    # The default TCP budget is 1/32 of physical memory. Repeated 10,000-socket
+    # batches hit ENOBUFS on Darwin 25.6 even on master. Give both builds 1/8
+    # before timing; retain every listener check and performance threshold.
+    memory = int(subprocess.check_output(["sysctl", "-n", "hw.memsize"], text=True))
+    print(json.dumps(tcp_memory(memory // 8)))

.github/scan-benchmark/scan_bench.py

@@ -142,11 +142,13 @@ class Scenario:
     timeout_ms: int = 1500
     udp: bool = False
     repeat: int = 1  # extra samples for very short scenarios
+    excluded_ports: tuple[int, ...] = ()
 
     @property
     def sockets(self) -> int:
         span = sum(end - start + 1 for start, end in self.ranges)
-        return len(self.addresses) * (len(self.ports) + span)
+        excluded = sum(self.covers(port) for port in set(self.excluded_ports))
+        return len(self.addresses) * (len(self.ports) + span - excluded)
 
     def covers(self, port: int) -> bool:
         return port in self.ports or any(s <= port <= e for s, e in self.ranges)
@@ -160,6 +162,8 @@ class Scenario:
         args += ["-b", str(self.batch), "-t", str(self.timeout_ms)]
         if self.udp:
             args.append("--udp")
+        if self.excluded_ports:
+            args += ["--exclude-ports", ",".join(map(str, self.excluded_ports))]
         return args
 
     def expected(self, listeners: dict) -> set[str]:
@@ -168,7 +172,7 @@ class Scenario:
             f"{address}:{port}"
             for address in self.addresses
             for port in by_address.get(address, [])
-            if self.covers(port)
+            if self.covers(port) and port not in self.excluded_ports
         }
 
     def describe(self) -> str:
@@ -193,6 +197,13 @@ def build_scenarios(
     scenarios = [
         Scenario("tcp-1-port", "TCP, 1 open port", lo, ports=(TCP_PORTS[0],), repeat=3),
         Scenario("tcp-sweep", f"TCP, {ports(sweep)}, default batch", lo, sweep),
+        Scenario(
+            "tcp-sweep-excluded",
+            f"TCP, {ports(sweep)}, 1,024 excluded ports",
+            lo,
+            sweep,
+            excluded_ports=tuple(range(1024, 2048)),
+        ),
         Scenario(
             "tcp-sweep-b500", f"TCP, {ports(small_sweep)}, small batch", lo, small_sweep, batch=500
         ),
@@ -546,6 +557,41 @@ def ephemeral_range() -> tuple[int, int] | None:
     return (49152, 65535) if system == "Windows" else None
 
 
+def diagnose_tcp(binary: str, scenario: Scenario, listeners: dict) -> dict:
+    """Untimed failure capture on a disposable macOS runner, using debug logs."""
+    from macos_tcp_memory import tcp_memory
+
+    def states() -> dict[str, int]:
+        snapshot = subprocess.run(
+            ["netstat", "-an", "-p", "tcp"], capture_output=True, text=True, check=True
+        )
+        counts: dict[str, int] = {}
+        for line in snapshot.stdout.splitlines():
+            if line.startswith("tcp"):
+                state = line.split()[-1]
+                counts[state] = counts.get(state, 0) + 1
+        return counts
+
+    before = states()
+    memory_before = tcp_memory()
+    probe = subprocess.run(
+        [binary, *scenario.args(), "--scripts", "none", "--accessible", "--no-banner", "--no-config"],
+        env=dict(os.environ, RUST_LOG="rustscan=info,rustscan::scanner=debug"),
+        capture_output=True, text=True, check=True,
+    )
+    opened = {
+        match.group(1) for line in probe.stdout.splitlines()
+        if (match := OPEN_LINE.match(line))
+    }
+    return {
+        "before": before, "after": states(),
+        "memory_before": memory_before, "memory_after": tcp_memory(),
+        "missing": sorted(scenario.expected(listeners) - opened),
+        "errors": [line for line in probe.stderr.splitlines()
+                   if "Typical socket connection errors" in line],
+    }
+
+
 def environment() -> dict[str, object]:
     system = platform.system()
     info: dict[str, object] = {
@@ -818,6 +864,7 @@ def run(args: argparse.Namespace) -> int:
 
     names = list(builds)
     results: Results = {sc.name: {name: [] for name in names} for sc in scenarios}
+    diagnostics = []
     try:
         warmup = next((sc for sc in scenarios if sc.name == "tcp-sweep"), scenarios[0])
         for name in names:
@@ -842,6 +889,16 @@ def run(args: argparse.Namespace) -> int:
                     )
                     if not res.ok:
                         print(f"    {res.error}", flush=True)
+                if platform.system() == "Darwin" and not diagnostics and any(
+                    sc.expected(listeners) - set(results[sc.name][name][-1].open) for name in names
+                ):
+                    # A failed pair cannot support a performance claim. Capture
+                    # its OS errors immediately, outside both measured scans.
+                    for name in names:
+                        record = {"build": name, "scenario": sc.name,
+                                  **diagnose_tcp(builds[name], sc, listeners)}
+                        diagnostics.append(record)
+                        print("TCP diagnosis: " + json.dumps(record), flush=True)
     finally:
         server.kill()
         server.wait()
@@ -861,6 +918,7 @@ def run(args: argparse.Namespace) -> int:
         "failures": failures,
         "mismatches": mismatches,
         "notes": notes,
+        "diagnostics": diagnostics,
     }
     markdown = render_markdown(report, results, scenarios)
     if args.json:

.github/workflows/scan-benchmark.yml

@@ -184,6 +184,14 @@ jobs:
           head -c 200 "$RUNNER_TEMP/far.json"; echo
           echo "DELAYED_OK=1" >> "$GITHUB_ENV"
 
+      - name: Provision macOS TCP memory for repeated socket batches
+        if: runner.os == 'macOS'
+        run: |
+          # XNU's TCP accounting hard limit otherwise makes both builds miss
+          # listeners with ENOBUFS during repeated sweeps. Apply one budget to
+          # both builds before timing; keep all input sizes and gates intact.
+          sudo "$(command -v python)" .github/scan-benchmark/macos_tcp_memory.py
+
       - name: Describe the runner
         continue-on-error: true
         run: |
@@ -195,7 +203,8 @@ jobs:
             ss -Hltnu | sort -k5 | head -n 50 || true
           elif [ "$RUNNER_OS" = macOS ]; then
             for oid in hw.ncpu machdep.cpu.brand_string kern.maxfilesperproc \
-              net.inet.ip.portrange.first net.inet.ip.portrange.last; do
+              net.inet.ip.portrange.first net.inet.ip.portrange.last \
+              kern.ipc.maxsockbuf kern.ipc.nmbclusters; do
               sysctl "$oid" || true
             done
           fi

benches/benchmark_helpers.rs

@@ -2,7 +2,7 @@ use criterion::{criterion_group, criterion_main, Criterion};
 use rustscan::generated::get_parsed_data;
 use rustscan::input::{Opts, PortRanges, ScanOrder};
 use rustscan::port_strategy::PortStrategy;
-use rustscan::scanner::build_udp_payload_lookup;
+use rustscan::scanner::{build_udp_payload_lookup, Scanner};
 use std::collections::BTreeMap;
 use std::hint::black_box;
 use std::net::IpAddr;
@@ -60,6 +60,50 @@ fn criterion_benchmark(c: &mut Criterion) {
     });
     address_group.finish();
 
+    // Exercise production port preparation without opening a socket. The
+    // scanner has no target addresses, and runtime construction is not timed.
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let mut preparation = c.benchmark_group("port preparation");
+    preparation.sample_size(20);
+    preparation.warm_up_time(Duration::from_millis(500));
+    preparation.measurement_time(Duration::from_secs(2));
+    for (port_count, excluded_count) in [
+        (1, 1),
+        (16, 4),
+        (64, 1),
+        (64, 4096),
+        (4096, 0),
+        (4096, 64),
+        (4096, 4096),
+        (65535, 0),
+        (65535, 4),
+        (65535, 64),
+        (65535, 1024),
+        (65535, 4096),
+    ] {
+        let scanner = Scanner::new(
+            &[],
+            500,
+            Duration::from_millis(100),
+            1,
+            true,
+            PortStrategy::Manual((1..=port_count).collect()),
+            true,
+            (0..excluded_count)
+                .map(|i| (i * 13 % 65536) as u16)
+                .collect(),
+            false,
+        );
+        preparation.bench_function(
+            format!("{port_count} ports, {excluded_count} exclusions"),
+            |b| b.iter(|| black_box(runtime.block_on(black_box(&scanner).run_with_status()))),
+        );
+    }
+    preparation.finish();
+
     // UDP payload lookup micro-benchmark: compares the old linear scan of the
     // payload map with the precomputed port -> payload lookup. No sockets.
     let udp_map = get_parsed_data();

docs/port-exclusion-benchmark.md

@@ -0,0 +1,114 @@
+# Port exclusion preparation benchmark
+
+Compared Tokio master `b7119f7f3eb1601e81126ef3572699ed7b3b7c0b` with
+implementation `270561d553688acbd99f61c460cdfa7c76c1423e` on 2026-10-02.
+
+## Method
+
+Release builds with LTO, Rust 1.98.1, Intel Core Ultra 7 165U, Linux
+7.2.6-1-cachyos. Both versions use the same lockfile and identical
+`benches/benchmark_helpers.rs`; copy the final benchmark harness into the
+baseline worktree before building it. The benchmark runs the production
+`Scanner::run_with_status` with **zero target addresses**, so it opens no
+sockets and does no DNS or scripting. It includes generating/cloning the
+port list, filtering and empty scan setup. Scanner construction and the
+current-thread Tokio runtime construction are outside the timer.
+
+Manual ports are `1..=port_count`; exclusions are
+`(i * 13 % 65536) as u16` for `i` in `0..excluded_count`. Each case has two
+independent twenty-sample Criterion runs per build, a 500 ms warmup and a
+2 s target measurement. Cases run in A/B then B/A order on CPU 0. No builds
+or other local checks ran during timing. CPU frequency was not fixed on
+this shared laptop, so the table retains both repetitions rather than
+selecting one result. Values are Criterion's mean point estimates.
+
+Build command for each worktree:
+
+```sh
+cargo bench --locked --bench benchmark_helpers --no-run
+```
+
+The saved benchmark executables are then run serially, for each case:
+
+```sh
+CARGO_TARGET_DIR=/tmp/preparation-results taskset -c 0 /path/to/benchmark_helpers \
+  --bench 'port preparation/65535 ports, 4096 exclusions' --save-baseline base-a
+```
+
+Repeat with the candidate binary as `head-a`, then the candidate as
+`head-b` and the baseline as `base-b`. To run the group normally without
+comparing builds:
+
+```sh
+cargo bench --locked --bench benchmark_helpers -- 'port preparation'
+```
+
+## Results
+
+Times are microseconds; A and B are separate matched repetitions.
+
+| Ports | Exclusions | Master µs A / B | Optimized µs A / B | Time saved A / B |
+| ---: | ---: | ---: | ---: | ---: |
+| 1 | 1 | 0.289 / 0.286 | 0.261 / 0.271 | 9.7% / 5.2% |
+| 16 | 4 | 0.260 / 0.268 | 0.232 / 0.237 | 10.6% / 11.6% |
+| 64 | 1 | 0.433 / 0.437 | 0.237 / 0.239 | 45.2% / 45.4% |
+| 64 | 4,096 | 6.231 / 10.093 | 2.484 / 3.038 | 60.1% / 69.9% |
+| 4,096 | 0 | 10.889 / 10.037 | 0.335 / 0.397 | 96.9% / 96.0% |
+| 4,096 | 64 | 20.680 / 20.363 | 2.930 / 3.064 | 85.8% / 85.0% |
+| 4,096 | 4,096 | 524.223 / 519.670 | 6.805 / 6.710 | 98.7% / 98.7% |
+| 65,535 | 0 | 166.113 / 115.686 | 3.655 / 2.333 | 97.8% / 98.0% |
+| 65,535 | 4 | 175.717 / 181.343 | 29.769 / 30.230 | 83.1% / 83.3% |
+| 65,535 | 64 | 207.617 / 206.910 | 31.595 / 31.462 | 84.8% / 84.8% |
+| 65,535 | 1,024 | 2018.463 / 3271.564 | 33.403 / 33.520 | 98.3% / 99.0% |
+| 65,535 | 4,096 | 8811.234 / 8469.706 | 59.853 / 63.211 | 99.3% / 99.3% |
+
+All twelve cases improved in both repetitions. This measures **preparation,
+not end-to-end network scan speed**: many scans are dominated by socket I/O,
+timeouts and reporting. In particular, the default no-exclusion saving is
+only a fraction of a millisecond.
+
+## Implementation and correctness
+
+With no exclusions, keep the already generated vector. For fewer than 64
+ports, filter in place with linear membership checks; this avoids bitmap
+setup and a second allocation. Larger lists use an 8 KiB bitmap covering
+all 65,536 values representable by `u16`, then retain ports in place. This
+changes the large-list filtering cost from ports × exclusions to ports +
+exclusions, while preserving order, duplicate ports and boundary values.
+The helper is kept out of line: release assembly showed an inlined bitmap
+reserved 8,424 bytes in every async poll, whereas the final poll frame uses
+344 bytes and the helper reserves its bitmap only during preparation.
+
+Exploration found that in-place linear membership slows large cases. A
+first production candidate also slowed 64 ports with 4,096 exclusions;
+using the bitmap for that case corrected the regression before the final
+matched runs. The cutoff is a performance choice, not a limit on input.
+
+One focused network-free regression test compares the implementation with
+an independent membership oracle for empty inputs, endpoints 0/65535,
+duplicate ports/exclusions, all ports, and a frozen random order.
+`cargo test --locked` passes 74 tests with one pre-existing ignored doctest.
+All-target Clippy, formatting, documentation and diff checks pass. The
+CI-only scan harness adds an excluded-port sweep using the existing
+listeners, checking expected results and using the actual attempted socket
+count for throughput.
+
+[Raw Criterion samples, confidence intervals, revisions and binary hashes](port-exclusion-benchmark-results.json)
+are included. Hosted correctness/performance results are linked in the PR.
+
+## macOS runner setup
+
+Repeated hosted sweeps on Darwin 25.6 returned `ENOBUFS` (error 55) on
+both master and the candidate, missing fixture listeners. Such runs are
+invalid for performance claims. The workflow provisions the TCP memory
+budget once before either build is measured, raising it from the default
+1/32 to 1/8 of physical memory when necessary, and prints the actual limit
+and allocation counters. It retains every scenario, repeat, listener
+comparison and performance threshold. An immediate untimed diagnostic
+scan records errors and kernel counters if a measured pair misses a
+listener; that failed dataset is not used to claim a speedup.
+
+This uses XNU's [TCP budget initialization](https://github.com/apple-oss-distributions/xnu/blob/main/bsd/netinet/tcp_subr.c)
+and [memory accounting interface](https://github.com/apple-oss-distributions/xnu/blob/main/bsd/sys/mem_acct_private.h).
+It changes the disposable benchmark environment, not RustScan's behavior
+or the operating system configuration on users' machines.

src/scanner/mod.rs

@@ -32,6 +32,29 @@ use tokio::time::{sleep, timeout};
 /// polls to about a millisecond on Linux.
 const WORK_PER_TURN: usize = 128;
 
+/// Keeps the generated order, including duplicate ports, while removing exclusions.
+// Keep the bitmap out of the async polling frame, which runs on every wake.
+#[inline(never)]
+fn filter_excluded_ports(mut ports: Vec<u16>, excluded: &[u16]) -> Vec<u16> {
+    if excluded.is_empty() {
+        return ports;
+    }
+
+    // Bitmap setup costs more than membership checks on short port lists.
+    if ports.len() < 64 {
+        ports.retain(|port| !excluded.contains(port));
+        return ports;
+    }
+
+    // The complete u16 port space fits in an 8 KiB bitmap.
+    let mut excluded_bits = [0_u64; 1024];
+    for &port in excluded {
+        excluded_bits[usize::from(port) / 64] |= 1 << (port % 64);
+    }
+    ports.retain(|&port| excluded_bits[usize::from(port) / 64] & (1 << (port % 64)) == 0);
+    ports
+}
+
 /// UDP payload lookup: port -> payload bytes
 ///
 /// `get_parsed_data()` returns a `&'static BTreeMap<...>`, so we can store
@@ -190,13 +213,7 @@ impl Scanner {
     }
 
     async fn scan(&self) -> Vec<PortStatus> {
-        let ports: Vec<u16> = self
-            .port_strategy
-            .order()
-            .iter()
-            .filter(|&port| !self.exclude_ports.contains(port))
-            .copied()
-            .collect();
+        let ports = filter_excluded_ports(self.port_strategy.order(), &self.exclude_ports);
         let mut found_sockets: Vec<PortStatus> = Vec::new();
         let mut errors: HashSet<String> = HashSet::new();
 
@@ -610,6 +627,35 @@ mod tests {
         )
     }
 
+    #[test]
+    fn port_exclusions_preserve_order_duplicates_and_boundaries() {
+        let inputs = [
+            Vec::new(),
+            vec![65535, 0, 80, 443, 80, 65535, 1],
+            (0..=65535).collect(),
+            PortStrategy::pick(&Some(PortRanges(vec![(0, 1023)])), None, ScanOrder::Random).order(),
+        ];
+        let exclusions = [
+            Vec::new(),
+            vec![0],
+            vec![0, 65535, 0, 443],
+            (0..1024).collect(),
+            (0..=65535).collect(),
+        ];
+
+        for ports in inputs {
+            for excluded in &exclusions {
+                let excluded_set: HashSet<_> = excluded.iter().copied().collect();
+                let expected: Vec<u16> = ports
+                    .iter()
+                    .filter(|&port| !excluded_set.contains(port))
+                    .copied()
+                    .collect();
+                assert_eq!(filter_excluded_ports(ports.clone(), excluded), expected);
+            }
+        }
+    }
+
     #[test]
     fn library_scanner_is_quiet_by_default() {
         let scanner = test_scanner();