Speed up port exclusions and avoid redundant port copies

.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 @@ def args(self) -> list[str]:
         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 @@ def expected(self, listeners: dict) -> set[str]:
             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 ports(ranges: tuple[tuple[int, int], ...]) -> str:
     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();