commit 8049d220c48b9baabfb01094d3ed5f8d39d5953a
parent c85eacd226eb207a24bbd8d90b08d91d32fc0554
Author: Matthias Jaros <jaros@mailbox.org>
Date: Fri, 3 Jul 2026 12:06:01 +0200
Replaced the nested executors with one per scanner. Which is fine for a
virtualthread executor
Diffstat:
1 file changed, 86 insertions(+), 87 deletions(-)
diff --git 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<ScanResult> completionService = new ExecutorCompletionService<>(executor);
- Queue<String> 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<ScanResult> completionService = new ExecutorCompletionService<>(executor);
+ Queue<String> 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<ScanResult> 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<ScanResult> 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<PortResult> completionService = new ExecutorCompletionService<>(executor);
- List<Throwable> 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<PortResult> completionService = new ExecutorCompletionService<>(executor);
+ List<Throwable> 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<PortResult> portResultFuture = completionService.poll(10, TimeUnit.MILLISECONDS);
+ if (portResultFuture == null) {
+ continue;
}
- try {
- Future<PortResult> 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);
}
}
