TIKA-4821: warm-first pipes client selection; size single-fork JVMs correctly

Author: tballison

docs/modules/ROOT/pages/pipes/configuration.adoc

@@ -71,7 +71,7 @@ See also xref:pipes/timeouts.adoc[Timeouts] for the full timeout model.
 
 |`socketTimeoutMillis`
 |`60000`
-|Maximum time (ms) to wait for data from a forked process. If no heartbeat or result is received within this window, the parse is considered hung.
+|Maximum time (ms) to wait for data from a forked process. If no heartbeat or result is received within this window, the parse is considered hung. Also serves as the fork's idle-shutdown timer: a fork that receives no work for this long exits and is restarted transparently on next use.
 
 |`heartbeatIntervalMillis`
 |`1000`

docs/modules/ROOT/pages/pipes/cpu-sizing.adoc

@@ -25,9 +25,10 @@ cores.
 
 To fix this, Tika Pipes auto-injects `-XX:ActiveProcessorCount` into each
 forked JVM's command line, sizing each fork's view of the CPU count to a fair
-slice of the host. This is on by default in per-client mode (`numClients > 1`)
-when the user has not already supplied `-XX:ActiveProcessorCount` in
-`forkedJvmArgs`.
+slice of the host. This is on by default in per-client mode — including
+`numClients=1`, where the slice is everything but the parent's reserved
+cores — whenever the user has not already supplied
+`-XX:ActiveProcessorCount` in `forkedJvmArgs`.
 
 == Mental model
 
@@ -73,10 +74,10 @@ For typical cloud-VM core counts:
 |===
 |hostCores |numClients |slice |Notes
 
-|2  |1 |n/a    |Tight; auto-cap not applied (single fork). Acceptable for low throughput.
-|4  |1 |n/a    |Comfortable single-fork deployment.
+|2  |1 |1 → skipped |Tight; auto-cap declines. Acceptable for low throughput.
+|4  |1 |2      |Comfortable single-fork deployment.
 |4  |2 |1 → skipped |Auto-cap declines; consider `numClients=1`.
-|8  |1 |n/a    |Lots of headroom; single-fork lifecycle isolation is fine.
+|8  |1 |6      |Lots of headroom; single-fork lifecycle isolation is fine.
 |8  |3 |2      |Sweet spot for medium pods.
 |16 |4 |3      |Sweet spot for 16-core hosts. Measured winner in benchmarks.
 |16 |6 |2      |Higher concurrency; tighter per-fork breathing room.
@@ -88,6 +89,26 @@ The general rule is: pick the largest `numClients` that satisfies
 `numClients × 2 + 2 ≤ hostCores`. Beyond that point, adding workers
 starts hurting throughput.
 
+== Lazy start, warm reuse, and idle shutdown
+
+`numClients` is a *ceiling* on concurrent forks, not a resident count:
+
+* **Lazy start.** No fork is started at construction; each starts on the
+  first request that needs it.
+* **Warm reuse (LIFO).** The client pool hands out the most-recently-used
+  client first. Sequential or lightly-concurrent traffic stays on the same
+  warm fork (or forks) instead of round-robining every fork awake; a cold
+  fork is only started when concurrency actually exceeds the number of warm
+  ones.
+* **Idle shutdown.** A fork that receives no work for `socketTimeoutMillis`
+  (default 60s) exits on its own and is restarted transparently on next use.
+
+Together these make the resident fork count track *recent concurrency*
+rather than `numClients`, so over-sizing `numClients` costs little at idle.
+Per-fork CPU and heap slices are still computed statically from
+`numClients` (a lone warm fork does not inherit its idle siblings'
+shares) — capacity planning above is unchanged.
+
 == Diagnostics
 
 Every `PipesParser` startup emits a one-shot summary line on its main
@@ -102,7 +123,6 @@ The `autoCap` field is one of:
 
 * `slice=N` — the auto-cap fired; each fork sees N CPUs.
 * `skipped (slice<2)` — over-provisioned; operator should reduce `numClients`.
-* `n/a (single fork; not capped)` — `numClients=1`; fork sees the whole host.
 * `user-set in forkedJvmArgs` — operator set `-XX:ActiveProcessorCount` themselves.
 
 Two `WARN`-level messages call out clearly-bad provisioning:
@@ -168,10 +188,11 @@ report `autoCap=user-set in forkedJvmArgs`.
 
 Heap is auto-sized the same way CPU is. Left to itself, every forked JVM takes its own
 default max heap — a fixed fraction of host or container memory — so `numClients` forks have
-a combined ceiling well above what the host actually has. When `numClients > 1` and you have
-not set `-Xmx` (or `-XX:MaxRAMPercentage`/`-XX:MaxRAMFraction`) yourself, Tika injects
-`-XX:MaxRAMPercentage=75/numClients`, leaving the remainder for the parent JVM and the OS.
-The `pipes-cpu-sizing` summary line reports the decision as `heap=...`.
+a combined ceiling well above what the host actually has. When you have not set `-Xmx`
+(or `-XX:MaxRAMPercentage`/`-XX:MaxRAMFraction`) yourself, Tika injects
+`-XX:MaxRAMPercentage=75/numClients` — 75 for a single fork — leaving the remainder for
+the parent JVM and the OS. The `pipes-cpu-sizing` summary line reports the decision as
+`heap=...`.
 
 [IMPORTANT]
 ====
@@ -187,6 +208,12 @@ its own heap at startup and logs a `WARN` if it came up under 256 MB — the poi
 ordinary documents, not just pathological ones, begin to fail. If you see that warning,
 lower `numClients`, raise the container memory limit, or set `-Xmx` explicitly.
 
+When you set fork heap explicitly, the parent checks the arithmetic for you at startup:
+if `numClients × -Xmx` exceeds 75% of host/container memory (or explicit
+`-XX:MaxRAMPercentage` values sum past 75%), it logs a `pipes-cpu-sizing` `WARN` naming
+the commitment. It warns rather than fails — a co-tenant box the operator has budgeted
+deliberately is indistinguishable from a mistake.
+
 Cross-check both constraints yourself when sizing: `numClients × 2 + 2 ≤ hostCores` **and**
 `numClients × per-worker-heap ≤ 75% of memory`.
 ====

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java

@@ -80,6 +80,87 @@ private static int forkHeapPercentage(int numClients) {
         return Math.max(1, FORK_HEAP_BUDGET_PERCENT / numClients);
     }
 
+    /**
+     * Explicit heap settings suppress auto-division, so nothing else checks that
+     * numClients forks at the user's size can actually coexist -- the mistake only
+     * surfaces later as fork OOMs or the host OOM-killer. Returns a warning when
+     * the combined explicit ceiling exceeds the fork budget, else null.
+     * Package-private for tests.
+     */
+    static String heapOvercommitWarning(List<String> forkedJvmArgs, int numClients,
+                                        long totalMemBytes) {
+        long xmxBytes = -1;
+        double maxRamPct = -1;
+        // last occurrence wins, same as the JVM
+        for (String arg : forkedJvmArgs) {
+            if (arg.startsWith("-Xmx")) {
+                xmxBytes = parseJvmMemArg(arg.substring("-Xmx".length()));
+            } else if (arg.startsWith("-XX:MaxRAMPercentage=")) {
+                try {
+                    maxRamPct = Double.parseDouble(
+                            arg.substring("-XX:MaxRAMPercentage=".length()));
+                } catch (NumberFormatException e) {
+                    maxRamPct = -1;
+                }
+            }
+        }
+        // -Xmx beats MaxRAMPercentage in the JVM, so only judge the percentage alone
+        if (xmxBytes <= 0 && maxRamPct > 0 && numClients * maxRamPct > FORK_HEAP_BUDGET_PERCENT) {
+            return String.format(java.util.Locale.ROOT,
+                    "numClients=%d x -XX:MaxRAMPercentage=%s commits %.0f%% of memory; " +
+                    "the parent JVM and OS need the remainder (fork budget: %d%%). " +
+                    "Lower numClients or the percentage.",
+                    numClients, maxRamPct, numClients * maxRamPct, FORK_HEAP_BUDGET_PERCENT);
+        }
+        if (xmxBytes > 0 && totalMemBytes > 0
+                && numClients * (double) xmxBytes > totalMemBytes * FORK_HEAP_BUDGET_PERCENT / 100.0) {
+            return String.format(java.util.Locale.ROOT,
+                    "numClients=%d x -Xmx (%,dMB each) commits %,dMB of %,dMB total memory, " +
+                    "over the %d%% fork budget; the parent JVM and OS need the remainder. " +
+                    "Lower numClients or -Xmx.",
+                    numClients, xmxBytes / MB, numClients * xmxBytes / MB, totalMemBytes / MB,
+                    FORK_HEAP_BUDGET_PERCENT);
+        }
+        return null;
+    }
+
+    private static final long MB = 1024 * 1024;
+
+    // "-Xmx" style value: digits with optional k/m/g/t suffix; -1 if unparseable
+    static long parseJvmMemArg(String value) {
+        if (value == null || value.isEmpty()) {
+            return -1;
+        }
+        long multiplier = switch (Character.toLowerCase(value.charAt(value.length() - 1))) {
+            case 'k' -> 1024L;
+            case 'm' -> 1024L * 1024;
+            case 'g' -> 1024L * 1024 * 1024;
+            case 't' -> 1024L * 1024 * 1024 * 1024;
+            default -> 1;
+        };
+        String digits = multiplier == 1 ? value : value.substring(0, value.length() - 1);
+        try {
+            long n = Long.parseLong(digits);
+            return n <= 0 ? -1 : Math.multiplyExact(n, multiplier);
+        } catch (NumberFormatException | ArithmeticException e) {
+            return -1;
+        }
+    }
+
+    // Container-aware on JDK 17 (honors cgroup memory limits); -1 if unavailable.
+    // Read via JMX attribute name: the typed accessor lives on a com.sun
+    // interface that forbiddenapis bans as non-portable.
+    private static long totalMemorySize() {
+        try {
+            Object value = java.lang.management.ManagementFactory.getPlatformMBeanServer()
+                    .getAttribute(new javax.management.ObjectName("java.lang:type=OperatingSystem"),
+                            "TotalMemorySize");
+            return value instanceof Number n ? n.longValue() : -1;
+        } catch (Exception e) {
+            return -1;
+        }
+    }
+
 
     private final PipesConfig pipesConfig;
     private final Path tikaConfigPath;
@@ -144,8 +225,6 @@ private void logCpuSizing() {
         String capDecision;
         if (userSetCap) {
             capDecision = "user-set in forkedJvmArgs";
-        } else if (numClients <= 1) {
-            capDecision = "n/a (single fork; not capped)";
         } else {
             int budget = Math.max(1, hostCores - PARENT_RESERVED_CORES);
             int slice = budget / numClients;
@@ -156,14 +235,18 @@ private void logCpuSizing() {
         String heapDecision;
         if (userSetHeap(pipesConfig.getForkedJvmArgs())) {
             heapDecision = "user-set in forkedJvmArgs";
-        } else if (numClients <= 1) {
-            heapDecision = "n/a (single fork; JVM default)";
         } else {
             heapDecision = "MaxRAMPercentage=" + forkHeapPercentage(numClients);
         }
         LOG.info("pipes-cpu-sizing: hostCores={}, numClients={}, parentReserved={}, " +
                 "autoCap={}, heap={}", hostCores, numClients, PARENT_RESERVED_CORES,
                 capDecision, heapDecision);
+
+        String overcommit = heapOvercommitWarning(pipesConfig.getForkedJvmArgs(),
+                numClients, totalMemorySize());
+        if (overcommit != null) {
+            LOG.warn("pipes-cpu-sizing: {}", overcommit);
+        }
     }
 
     @Override
@@ -313,7 +396,7 @@ private synchronized void startServer() throws IOException, InterruptedException
         LOG.trace("clientId={}: starting server on port={}", clientId, port);
 
         tmpDir = Files.createTempDirectory("pipes-server-" + clientId + "-");
-        ProcessBuilder pb = new ProcessBuilder(getCommandline());
+        ProcessBuilder pb = new ProcessBuilder(getCommandline(tmpDir));
         // Tell the child our PID so it can watch ProcessHandle.onExit() and
         // self-terminate promptly if we die. Without this, an orphan child
         // can only notice via socket-read timeout (default 60s) and can't
@@ -448,7 +531,9 @@ private void deleteDir(Path dir) {
         }
     }
 
-    private String[] getCommandline() throws IOException {
+    // package-private and tmpDir passed in (rather than read from the field set by
+    // startServer) so tests can exercise arg injection without forking a JVM
+    String[] getCommandline(Path tmpDir) throws IOException {
         List<String> configArgs = new ArrayList<>(pipesConfig.getForkedJvmArgs());
         boolean hasClassPath = false;
         boolean hasHeadless = false;
@@ -500,7 +585,7 @@ private String[] getCommandline() throws IOException {
         // host has -- the same "each JVM thinks it owns the machine" problem the
         // ActiveProcessorCount cap solves. Give each fork a slice of a fixed budget
         // instead, leaving the remainder for the parent and the OS.
-        if (!userSetHeap(configArgs) && pipesConfig.getNumClients() > 1) {
+        if (!userSetHeap(configArgs)) {
             int pct = forkHeapPercentage(pipesConfig.getNumClients());
             configArgs.add("-XX:MaxRAMPercentage=" + pct);
             LOG.debug("clientId={}: auto-injected -XX:MaxRAMPercentage={} (numClients={})",
@@ -517,7 +602,7 @@ private String[] getCommandline() throws IOException {
         // Skip the auto-cap when the computed slice would drop below
         // MIN_AUTO_CAP_SLICE -- below that, the fork can't keep its socket
         // reader responsive and back-pressures the parent.
-        if (!hasActiveProcessorCount && pipesConfig.getNumClients() > 1) {
+        if (!hasActiveProcessorCount) {
             int hostCores = Runtime.getRuntime().availableProcessors();
             int forkBudget = Math.max(1, hostCores - PARENT_RESERVED_CORES);
             int slice = forkBudget / pipesConfig.getNumClients();
@@ -530,10 +615,11 @@ private String[] getCommandline() throws IOException {
             } else {
                 LOG.info("clientId={}: skipping -XX:ActiveProcessorCount auto-cap " +
                         "(would yield slice={} < MIN_AUTO_CAP_SLICE={}; " +
-                        "hostCores={}, parentReserved={}, numClients={}). " +
-                        "Consider lowering numClients on this host.",
+                        "hostCores={}, parentReserved={}, numClients={}).{}",
                         clientId, slice, MIN_AUTO_CAP_SLICE, hostCores,
-                        PARENT_RESERVED_CORES, pipesConfig.getNumClients());
+                        PARENT_RESERVED_CORES, pipesConfig.getNumClients(),
+                        pipesConfig.getNumClients() > 1
+                                ? " Consider lowering numClients on this host." : "");
             }
         }
 
@@ -547,7 +633,7 @@ private String[] getCommandline() throws IOException {
         commandLine.add(ProcessUtils.escapeCommandLine(javaPath));
 
         if (!hasClassPath) {
-            Path argFile = writeArgFile();
+            Path argFile = writeArgFile(tmpDir);
             commandLine.add("@" + argFile.toAbsolutePath());
         }
 
@@ -573,7 +659,7 @@ private String[] getCommandline() throws IOException {
         return commandLine.toArray(new String[0]);
     }
 
-    private Path writeArgFile() throws IOException {
+    private Path writeArgFile(Path tmpDir) throws IOException {
         Path argFile = tmpDir.resolve("jvm-args.txt");
         // forward any tika.extras.dir jars to the forked PipesServer
         String classpath = TikaExtras.appendJarsToClasspath(System.getProperty("java.class.path"));

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java

@@ -280,7 +280,7 @@ public int getNumClients() {
     }
 
     public void setNumClients(int numClients) {
-        // Without this, 0 surfaces at startup as ArrayBlockingQueue's message-less IAE.
+        // Without this, 0 surfaces at startup as the client queue's message-less IAE.
         if (numClients <= 0) {
             throw new IllegalArgumentException("numClients must be > 0, was " + numClients);
         }

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesParser.java

@@ -21,7 +21,7 @@
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
@@ -80,14 +80,17 @@ public static PipesParser load(TikaJsonConfig tikaJsonConfig, PipesConfig pipesC
     private final Path tikaConfigPath;
     private final List<PipesClient> clients = new ArrayList<>();
     private final List<ServerManager> serverManagers = new ArrayList<>();
-    private final ArrayBlockingQueue<PipesClient> clientQueue;
+    // LIFO: the most-recently-returned client is borrowed next, so light traffic
+    // concentrates on warm forks instead of round-robining every fork awake (and,
+    // at intervals over the idle shutdown, cold-starting on every request).
+    private final LinkedBlockingDeque<PipesClient> clientQueue;
     private final boolean isSharedMode;
 
     private PipesParser(PipesConfig pipesConfig, Path tikaConfigPath) {
         this.pipesConfig = pipesConfig;
         this.tikaConfigPath = tikaConfigPath;
         this.isSharedMode = pipesConfig.isUseSharedServer();
-        this.clientQueue = new ArrayBlockingQueue<>(pipesConfig.getNumClients());
+        this.clientQueue = new LinkedBlockingDeque<>(pipesConfig.getNumClients());
 
         if (isSharedMode) {
             // Shared mode: one ServerManager for all clients
@@ -98,7 +101,7 @@ private PipesParser(PipesConfig pipesConfig, Path tikaConfigPath) {
 
             for (int i = 0; i < pipesConfig.getNumClients(); i++) {
                 PipesClient client = new PipesClient(pipesConfig, sharedManager);
-                clientQueue.offer(client);
+                clientQueue.offerLast(client);
                 clients.add(client);
             }
         } else {
@@ -110,7 +113,7 @@ private PipesParser(PipesConfig pipesConfig, Path tikaConfigPath) {
                 serverManagers.add(serverManager);
 
                 PipesClient client = new PipesClient(pipesConfig, serverManager);
-                clientQueue.offer(client);
+                clientQueue.offerLast(client);
                 clients.add(client);
             }
         }
@@ -120,15 +123,15 @@ public PipesResult parse(FetchEmitTuple t) throws InterruptedException,
             PipesException, IOException {
         PipesClient client = null;
         try {
-            client = clientQueue.poll(pipesConfig.getMaxWaitForClientMillis(),
+            client = clientQueue.pollFirst(pipesConfig.getMaxWaitForClientMillis(),
                     TimeUnit.MILLISECONDS);
             if (client == null) {
                 return PipesResults.CLIENT_UNAVAILABLE_WITHIN_MS;
             }
             return client.process(t);
         } finally {
             if (client != null) {
-                clientQueue.offer(client);
+                clientQueue.offerFirst(client);
             }
         }
     }
@@ -160,6 +163,11 @@ public void close() throws IOException {
         }
     }
 
+    // package-private for tests: how many forked servers are actually live
+    long startedServerCount() {
+        return serverManagers.stream().filter(ServerManager::isRunning).count();
+    }
+
     /**
      * Returns whether this parser is using shared server mode.
      *