commit 1080e5e415824804fab289e04a67a00081ed372d
parent d4780f5d5efbb95f7035e5bea6cb957ac131bd6d
Author: Matthias Jaros <jaros@mailbox.org>
Date: Wed, 1 Jul 2026 18:12:19 +0200
Added CompletionService
Use while loop to fill efficiently all available possible virtual threads before starting to check for results
Diffstat:
1 file changed, 56 insertions(+), 48 deletions(-)
diff --git a/src/main/java/com/it_jaros/jscanner/Scanner.java b/src/main/java/com/it_jaros/jscanner/Scanner.java
@@ -9,14 +9,12 @@ import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
-import java.util.ArrayList;
-import java.util.BitSet;
-import java.util.HashMap;
-import java.util.List;
+import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
public class Scanner {
@@ -59,7 +57,7 @@ public class Scanner {
}
public List<ScanResult> scanHosts(List<String> hostsArgv, String ports) {
- try (Stream<String> stream = hostsArgv.stream()){
+ try (Stream<String> stream = hostsArgv.stream()) {
return scanHosts(stream, ports);
} catch (IOException e) {
System.out.printf("Error while trying to read hosts: %s -> %s %n", e.getClass().getSimpleName(), e.getMessage());
@@ -72,28 +70,33 @@ public class Scanner {
List<ScanResult> results = new ArrayList<>();
progressBar.start();
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
- List<Future<ScanResult>> futures = new ArrayList<>();
- Semaphore maxHosts = new Semaphore(maxHostsLimit);
- hosts.forEach((final String host) -> {
- maxHosts.acquireUninterruptibly();
- futures.add(executor.submit(() -> {
- try {
- return scanHost(host, ports);
- } finally {
- maxHosts.release();
- }
- }));
- });
+ CompletionService<ScanResult> completionService = new ExecutorCompletionService<>(executor);
+ Queue<String> hostsInQueue = hosts.collect(Collectors.toCollection(ArrayDeque::new));
+
+ int hostsToProcess = hostsInQueue.size();
+ int activeHostWorkers = 0;
+ while (hostsToProcess > 0) {
+ // process queue
+ while (!hostsInQueue.isEmpty() && activeHostWorkers <= maxHostsLimit) {
+ activeHostWorkers++;
+ final String host = hostsInQueue.poll();
+ completionService.submit(() -> scanHost(host, ports));
+ }
- futures.forEach((Future<ScanResult> f) -> {
try {
- results.add(f.get());
+ Future<ScanResult> finishedHost = completionService.poll(10, TimeUnit.MILLISECONDS);
+ if (finishedHost == null) {
+ continue;
+ }
+ activeHostWorkers--;
+ hostsToProcess--;
+ results.add(finishedHost.get());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (ExecutionException e) {
System.out.printf("%s -> %s%n", e.getClass().getSimpleName(), e.getMessage());
}
- });
+ }
}
progressBar.stop();
@@ -108,49 +111,55 @@ public class Scanner {
private ScanResult scanHost(String host, String ports) {
PortRange portRange = new PortRange(ports);
AtomicLong portSlotFactory = new AtomicLong(System.nanoTime());
- Semaphore maxWorkers = new Semaphore(maxWorkersPerHost);
BitSet openPorts = new BitSet(PortRange.MAX_PORT);
BitSet filteredPorts = new BitSet(PortRange.MAX_PORT);
- HashMap<Integer, ServiceType> bannerRecognition = new HashMap<Integer, ServiceType>();
+ HashMap<Integer, ServiceType> bannerRecognition = new HashMap<>();
+ // Give progressbar the current progress object which is then updated in the sub virtual threads
Progress progress = new Progress(host, portRange.getTotal(), new AtomicInteger(), new AtomicInteger(), new AtomicInteger());
progressBar.submit(progress);
- if(!disableOnlineCheck && !checkHostOnline(host)) {
+ if (!disableOnlineCheck && !checkHostOnline(host)) {
// Visually show that this host is basically done
progress.done().set(progress.total());
return new ScanResult(host, openPorts, filteredPorts, bannerRecognition);
}
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
- List<Future<PortResult>> futures = new ArrayList<>();
- while (portRange.hasNext()) {
- maxWorkers.acquireUninterruptibly();
- final int currentPort = portRange.next();
- futures.add(executor.submit(() -> {
- try {
+ CompletionService<PortResult> completionService = new ExecutorCompletionService<>(executor);
+ List<Throwable> errors = new ArrayList<>();
+ int activePerHostWorkers = 0;
+ int portsToProcess = portRange.getTotal();
+ while (portsToProcess > 0) {
+ while (portRange.hasNext() && activePerHostWorkers <= maxWorkersPerHost) {
+ activePerHostWorkers++;
+ final int currentPort = portRange.next();
+ completionService.submit(() -> {
waitForSlot(portSlotFactory);
return checkPort(host, currentPort);
- } finally {
- maxWorkers.release();
- progress.done().incrementAndGet();
- }
- }));
- }
+ });
+ }
- List<Throwable> errors = new ArrayList<>();
- for (Future<PortResult> f : futures) {
try {
+ Future<PortResult> f = completionService.poll(10, TimeUnit.MILLISECONDS);
+ if (f == null) {
+ continue;
+ }
+ activePerHostWorkers--;
+ portsToProcess--;
+ progress.done().incrementAndGet();
+
// bitset is not thread-safe, so it is set
// outside the other virtual threads that update progress
PortResult portResult = f.get();
- if (portResult.getState().equals(PortState.OPEN)) {
- openPorts.set(portResult.getPort());
- bannerRecognition.put(portResult.getPort(), ServiceDetector.detect(portResult.getBanner()));
- } else if (portResult.getState().equals(PortState.FILTERED)) {
- filteredPorts.set(portResult.getPort());
+ switch (portResult.getState()) {
+ case OPEN -> {
+ openPorts.set(portResult.getPort());
+ bannerRecognition.put(portResult.getPort(), ServiceDetector.detect(portResult.getBanner()));
+ }
+ case FILTERED -> filteredPorts.set(portResult.getPort());
}
- } catch (InterruptedException e) {
+ } catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
} catch (ExecutionException e) {
errors.add(e);
@@ -194,15 +203,14 @@ public class Scanner {
socket.connect(new InetSocketAddress(host, port), timeoutInMillis);
result.setBanner(getBanner(socket));
result.setState(PortState.OPEN);
- } catch (NoRouteToHostException ignored) {
- // this can be safely ignored because the port is closed if a host is unreachable
} catch (SocketTimeoutException ignored) {
result.setState(PortState.FILTERED);
} catch (ConnectException ignored) {
result.setState(PortState.CLOSED);
} catch (IOException ignored) {
+ // NoRouteToHostException: this can be safely ignored because the port is closed if a host is unreachable
// Will happen a lot when scanning for open ports, so not needed
- } finally {
+ } finally {
counter.dec();
socketLimit.release();
}
@@ -213,13 +221,13 @@ public class Scanner {
private String getBanner(Socket socket) {
int READ_BUFFER_SIZE = 1024;
byte[] buffer = new byte[READ_BUFFER_SIZE];
- try(InputStream input = socket.getInputStream()) {
+ try (InputStream input = socket.getInputStream()) {
socket.setSoTimeout(timeoutInMillis);
int bytesRead = input.read(buffer);
if (bytesRead <= 0) {
return null;
}
- return new String(buffer,0, bytesRead, StandardCharsets.UTF_8).trim();
+ return new String(buffer, 0, bytesRead, StandardCharsets.UTF_8).trim();
} catch (IOException e) {
// we ignore this failure
}
