From bfca3c7aae27a22e040bde0c4e62a769af8b84f4 Mon Sep 17 00:00:00 2001 From: Matthias Jaros Date: Fri, 3 Jul 2026 12:06:01 +0200 Subject: Replaced the nested executors with one per scanner. Which is fine for a virtualthread executor --- src/main/java/com/it_jaros/jscanner/Scanner.java | 173 +++++++++++------------ 1 file changed, 86 insertions(+), 87 deletions(-) (limited to 'src/main/java/com/it_jaros/jscanner/Scanner.java') diff --git a/src/main/java/com/it_jaros/jscanner/Scanner.java b/src/main/java/com/it_jaros/jscanner/Scanner.java index 0421e9b..26d3b14 100644 --- a/src/main/java/com/it_jaros/jscanner/Scanner.java +++ b/src/main/java/com/it_jaros/jscanner/Scanner.java @@ -17,6 +17,7 @@ public class Scanner implements AutoCloseable { private volatile boolean cancelled = false; + private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor(); private final Semaphore socketLimit; private final boolean bannerRecognition; private final boolean disableOnlineCheck; @@ -24,7 +25,6 @@ public class Scanner implements AutoCloseable { private final int maxWorkersPerHost; private final int timeoutInMillis; private final long delayInNanos; - private final Phaser runningScans = new Phaser(0); public Scanner( int socketLimit, @@ -55,41 +55,36 @@ public class Scanner implements AutoCloseable { } scan.start(); - runningScans.register(); - try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) { - CompletionService completionService = new ExecutorCompletionService<>(executor); - Queue hostsInQueue = new ArrayDeque<>(scan.getHosts()); - - int hostsToProcess = hostsInQueue.size(); - int activeHostWorkers = 0; - while ((hostsToProcess > 0 && !cancelled) || activeHostWorkers > 0) { - // process queue - while ( - !hostsInQueue.isEmpty() // continue as long as we have hosts to process - && activeHostWorkers < maxHostsLimit // limit is not reached - && !cancelled // and scanner not closed - ) { - activeHostWorkers++; - final String host = hostsInQueue.poll(); - completionService.submit(() -> scanHostPorts(host, scan.getPorts(), scan)); - } + CompletionService completionService = new ExecutorCompletionService<>(executor); + Queue hostsInQueue = new ArrayDeque<>(scan.getHosts()); + + int hostsToProcess = hostsInQueue.size(); + int activeHostWorkers = 0; + while ((hostsToProcess > 0 && !cancelled) || activeHostWorkers > 0) { + // process queue + while ( + !hostsInQueue.isEmpty() // continue as long as we have hosts to process + && activeHostWorkers < maxHostsLimit // limit is not reached + && !cancelled // and scanner not closed + ) { + activeHostWorkers++; + final String host = hostsInQueue.poll(); + completionService.submit(() -> scanHostPorts(host, scan.getPorts(), scan)); + } - try { - Future finishedHost = completionService.poll(10, TimeUnit.MILLISECONDS); - if (finishedHost == null) { - continue; - } - activeHostWorkers--; - hostsToProcess--; - scan.addScanResult(finishedHost.get()); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } catch (ExecutionException e) { - System.out.printf("%s -> %s%n", e.getClass().getSimpleName(), e.getMessage()); + try { + Future finishedHost = completionService.poll(10, TimeUnit.MILLISECONDS); + if (finishedHost == null) { + continue; } + activeHostWorkers--; + hostsToProcess--; + scan.addScanResult(finishedHost.get()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (ExecutionException e) { + System.out.printf("%s -> %s%n", e.getClass().getSimpleName(), e.getMessage()); } - } finally { - runningScans.arriveAndDeregister(); } scan.stop(); @@ -112,60 +107,58 @@ public class Scanner implements AutoCloseable { return new ScanResult(host, openPorts, filteredPorts, serviceTypes); } - try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) { - CompletionService completionService = new ExecutorCompletionService<>(executor); - List errors = new ArrayList<>(); - AtomicLong portSlotFactory = new AtomicLong(System.nanoTime()); - int activePerHostWorkers = 0; - int portsToProcess = portRange.getTotal(); - while ((portsToProcess > 0 && !cancelled) || activePerHostWorkers > 0) { - while ( - portRange.hasNext() // continue as long as we have ports - && activePerHostWorkers < maxWorkersPerHost // and we have not reached our limit - && !cancelled // and the scanner is not about to close - ) { - activePerHostWorkers++; - final int currentPort = portRange.next(); - completionService.submit(() -> { - waitForSlot(portSlotFactory); - return checkPort(host, currentPort, scan); - }); + CompletionService completionService = new ExecutorCompletionService<>(executor); + List errors = new ArrayList<>(); + AtomicLong portSlotFactory = new AtomicLong(System.nanoTime()); + int activePerHostWorkers = 0; + int portsToProcess = portRange.getTotal(); + while ((portsToProcess > 0 && !cancelled) || activePerHostWorkers > 0) { + while ( + portRange.hasNext() // continue as long as we have ports + && activePerHostWorkers < maxWorkersPerHost // and we have not reached our limit + && !cancelled // and the scanner is not about to close + ) { + activePerHostWorkers++; + final int currentPort = portRange.next(); + completionService.submit(() -> { + waitForSlot(portSlotFactory); + return checkPort(host, currentPort, scan); + }); + } + + try { + Future portResultFuture = completionService.poll(10, TimeUnit.MILLISECONDS); + if (portResultFuture == null) { + continue; } - try { - Future portResultFuture = completionService.poll(10, TimeUnit.MILLISECONDS); - if (portResultFuture == null) { - continue; + PortResult portResult = portResultFuture.get(); + + activePerHostWorkers--; + portsToProcess--; + progress.done().incrementAndGet(); + switch (portResult.getState()) { + case OPEN -> { + // bitset is not thread-safe, so it is set + // outside the other virtual threads that update progress + openPorts.set(portResult.getPort()); + progress.open().incrementAndGet(); + serviceTypes.put(portResult.getPort(), ServiceDetector.detect(portResult.getBanner())); } - - PortResult portResult = portResultFuture.get(); - - activePerHostWorkers--; - portsToProcess--; - progress.done().incrementAndGet(); - switch (portResult.getState()) { - case OPEN -> { - // bitset is not thread-safe, so it is set - // outside the other virtual threads that update progress - openPorts.set(portResult.getPort()); - progress.open().incrementAndGet(); - serviceTypes.put(portResult.getPort(), ServiceDetector.detect(portResult.getBanner())); - } - case FILTERED -> { - filteredPorts.set(portResult.getPort()); - progress.filtered().incrementAndGet(); - } + case FILTERED -> { + filteredPorts.set(portResult.getPort()); + progress.filtered().incrementAndGet(); } - } catch (InterruptedException ignored) { - Thread.currentThread().interrupt(); - } catch (ExecutionException e) { - errors.add(e); } + } catch (InterruptedException ignored) { + Thread.currentThread().interrupt(); + } catch (ExecutionException e) { + errors.add(e); } + } - if (!errors.isEmpty()) { - System.out.printf("Errors happened during scan of host %s%nErrors:%s -> %s", host, errors.size(), errors); - } + if (!errors.isEmpty()) { + System.out.printf("Errors happened during scan of host %s%nErrors:%s -> %s", host, errors.size(), errors); } return new ScanResult(host, openPorts, filteredPorts, serviceTypes); @@ -237,19 +230,25 @@ public class Scanner implements AutoCloseable { } public void cancel() { - cancelled = true; + if (!cancelled) { + cancelled = true; + } + executor.shutdown(); + } + + public void cancelNow() { + if (!cancelled) { + cancelled = true; + } + executor.shutdownNow(); } @Override public void close() throws Exception { cancel(); - awaitTermination(Duration.ofMillis(60000)); } - public void awaitTermination(Duration duration) throws InterruptedException, TimeoutException { - runningScans.awaitAdvanceInterruptibly( - runningScans.getPhase(), - duration.toMillis(), - TimeUnit.MILLISECONDS); + public boolean awaitTermination(Duration duration) throws InterruptedException { + return executor.awaitTermination(duration.toMillis(), TimeUnit.MILLISECONDS); } } -- cgit v1.3.1