This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4821-lifo-warm-client-selection in repository https://gitbox.apache.org/repos/asf/tika.git
commit 5a1d4816e604f3a41d4e00fbb083e230ffbffe9a Author: tallison <[email protected]> AuthorDate: Fri Aug 14 12:36:46 2026 -0400 TIKA-4821: warm-first pipes client selection; size single-fork JVMs uniformly --- docs/modules/ROOT/pages/pipes/configuration.adoc | 2 +- docs/modules/ROOT/pages/pipes/cpu-sizing.adoc | 49 +++++-- .../tika/pipes/core/PerClientServerManager.java | 108 ++++++++++++++-- .../org/apache/tika/pipes/core/PipesConfig.java | 2 +- .../org/apache/tika/pipes/core/PipesParser.java | 22 +++- .../core/PerClientServerManagerSizingTest.java | 141 +++++++++++++++++++++ .../tika/pipes/core/TikaPipesConfigTest.java | 2 +- .../tika/pipes/core/PipesParserWarmClientTest.java | 65 ++++++++++ 8 files changed, 357 insertions(+), 34 deletions(-) diff --git a/docs/modules/ROOT/pages/pipes/configuration.adoc b/docs/modules/ROOT/pages/pipes/configuration.adoc index 9a75633aa7..ad3de231ff 100644 --- a/docs/modules/ROOT/pages/pipes/configuration.adoc +++ b/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` diff --git a/docs/modules/ROOT/pages/pipes/cpu-sizing.adoc b/docs/modules/ROOT/pages/pipes/cpu-sizing.adoc index 22cd9dabf4..d0084986d0 100644 --- a/docs/modules/ROOT/pages/pipes/cpu-sizing.adoc +++ b/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`. ==== diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java index 9655b30a5b..ae326d7218 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java @@ -80,6 +80,83 @@ public class PerClientServerManager implements ServerManager { 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 + private static long totalMemorySize() { + java.lang.management.OperatingSystemMXBean os = + java.lang.management.ManagementFactory.getOperatingSystemMXBean(); + if (os instanceof com.sun.management.OperatingSystemMXBean sunOs) { + return sunOs.getTotalMemorySize(); + } + return -1; + } + private final PipesConfig pipesConfig; private final Path tikaConfigPath; @@ -144,8 +221,6 @@ public class PerClientServerManager implements ServerManager { 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 +231,18 @@ public class PerClientServerManager implements ServerManager { 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 +392,7 @@ public class PerClientServerManager implements ServerManager { 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 +527,9 @@ public class PerClientServerManager implements ServerManager { } } - 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 +581,7 @@ public class PerClientServerManager implements ServerManager { // 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 +598,7 @@ public class PerClientServerManager implements ServerManager { // 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 +611,11 @@ public class PerClientServerManager implements ServerManager { } 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 +629,7 @@ public class PerClientServerManager implements ServerManager { commandLine.add(ProcessUtils.escapeCommandLine(javaPath)); if (!hasClassPath) { - Path argFile = writeArgFile(); + Path argFile = writeArgFile(tmpDir); commandLine.add("@" + argFile.toAbsolutePath()); } @@ -573,7 +655,7 @@ public class PerClientServerManager implements ServerManager { 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")); diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java index 8f2907fc0a..82bbb80ab8 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java @@ -280,7 +280,7 @@ public class PipesConfig { } 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); } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesParser.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesParser.java index 22fd3dd672..dd68dca002 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesParser.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesParser.java @@ -21,7 +21,7 @@ import java.io.IOException; 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 class PipesParser implements Closeable { 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 @@ public class PipesParser implements Closeable { 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 @@ public class PipesParser implements Closeable { serverManagers.add(serverManager); PipesClient client = new PipesClient(pipesConfig, serverManager); - clientQueue.offer(client); + clientQueue.offerLast(client); clients.add(client); } } @@ -120,7 +123,7 @@ public class PipesParser implements Closeable { 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; @@ -128,7 +131,7 @@ public class PipesParser implements Closeable { return client.process(t); } finally { if (client != null) { - clientQueue.offer(client); + clientQueue.offerFirst(client); } } } @@ -160,6 +163,11 @@ public class PipesParser implements Closeable { } } + // 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. * diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PerClientServerManagerSizingTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PerClientServerManagerSizingTest.java new file mode 100644 index 0000000000..28118d7db7 --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PerClientServerManagerSizingTest.java @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.tika.pipes.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Auto-injected fork sizing (-XX:MaxRAMPercentage, -XX:ActiveProcessorCount) + * applies uniformly, including numClients=1, unless the user set their own. + */ +public class PerClientServerManagerSizingTest { + + @TempDir + Path tmp; + + private List<String> commandLine(int numClients, String... forkedJvmArgs) throws Exception { + PipesConfig pipesConfig = new PipesConfig(); + pipesConfig.setNumClients(numClients); + pipesConfig.setForkedJvmArgs(new ArrayList<>(Arrays.asList(forkedJvmArgs))); + PerClientServerManager manager = + new PerClientServerManager(pipesConfig, tmp.resolve("tika-config.json"), 0); + return Arrays.asList(manager.getCommandline(tmp)); + } + + private static List<String> withPrefix(List<String> args, String prefix) { + return args.stream().filter(a -> a.startsWith(prefix)).toList(); + } + + @Test + public void heapInjectedForSingleClient() throws Exception { + assertEquals(List.of("-XX:MaxRAMPercentage=75"), + withPrefix(commandLine(1), "-XX:MaxRAMPercentage")); + } + + @Test + public void heapDividedAcrossClients() throws Exception { + assertEquals(List.of("-XX:MaxRAMPercentage=25"), + withPrefix(commandLine(3), "-XX:MaxRAMPercentage")); + } + + @Test + public void userHeapSuppressesInjection() throws Exception { + List<String> args = commandLine(1, "-Xmx512m"); + assertTrue(withPrefix(args, "-XX:MaxRAMPercentage").isEmpty(), + "user -Xmx must suppress auto MaxRAMPercentage: " + args); + assertTrue(args.contains("-Xmx512m")); + } + + @Test + public void cpuCapInjectedForSingleClient() throws Exception { + List<String> caps = withPrefix(commandLine(1), "-XX:ActiveProcessorCount="); + int slice = Math.max(1, Runtime.getRuntime().availableProcessors() - 2); + if (slice >= 2) { + assertEquals(List.of("-XX:ActiveProcessorCount=" + slice), caps); + } else { + // Host too small for the auto-cap; injection is skipped by design. + assertTrue(caps.isEmpty(), "expected no auto-cap on tiny host: " + caps); + } + } + + @Test + public void userCpuCapSuppressesInjection() throws Exception { + assertEquals(List.of("-XX:ActiveProcessorCount=3"), + withPrefix(commandLine(1, "-XX:ActiveProcessorCount=3"), + "-XX:ActiveProcessorCount=")); + } + + @Test + public void parseJvmMemArg() { + assertEquals(512L * 1024 * 1024, PerClientServerManager.parseJvmMemArg("512m")); + assertEquals(2L * 1024 * 1024 * 1024, PerClientServerManager.parseJvmMemArg("2g")); + assertEquals(1024L * 1024, PerClientServerManager.parseJvmMemArg("1024k")); + assertEquals(1L * 1024 * 1024 * 1024 * 1024, PerClientServerManager.parseJvmMemArg("1t")); + assertEquals(1000, PerClientServerManager.parseJvmMemArg("1000")); + assertEquals(-1, PerClientServerManager.parseJvmMemArg("abc")); + assertEquals(-1, PerClientServerManager.parseJvmMemArg("")); + assertEquals(-1, PerClientServerManager.parseJvmMemArg("-2g")); + assertEquals(-1, PerClientServerManager.parseJvmMemArg("999999999t")); + } + + private static final long GB = 1024L * 1024 * 1024; + + @Test + public void xmxOvercommitWarns() { + // 4 x 2g = 8g > 75% of 8g + assertNotNull(PerClientServerManager.heapOvercommitWarning( + List.of("-Xmx2g"), 4, 8 * GB)); + // 4 x 2g = 8g <= 75% of 16g + assertNull(PerClientServerManager.heapOvercommitWarning( + List.of("-Xmx2g"), 4, 16 * GB)); + // last -Xmx wins, same as the JVM + assertNull(PerClientServerManager.heapOvercommitWarning( + List.of("-Xmx8g", "-Xmx1g"), 4, 16 * GB)); + // unknown total memory: no basis to warn + assertNull(PerClientServerManager.heapOvercommitWarning( + List.of("-Xmx2g"), 4, -1)); + } + + @Test + public void maxRamPercentageOvercommitWarns() { + // 3 x 50% = 150% of memory + assertNotNull(PerClientServerManager.heapOvercommitWarning( + List.of("-XX:MaxRAMPercentage=50"), 3, -1)); + // 3 x 25% = 75% budget exactly: allowed + assertNull(PerClientServerManager.heapOvercommitWarning( + List.of("-XX:MaxRAMPercentage=25"), 3, -1)); + // a fitting -Xmx overrides an overcommitted percentage, same as the JVM + assertNull(PerClientServerManager.heapOvercommitWarning( + List.of("-XX:MaxRAMPercentage=50", "-Xmx1g"), 3, 16 * GB)); + } + + @Test + public void noExplicitHeapNoWarning() { + assertNull(PerClientServerManager.heapOvercommitWarning(List.of(), 8, 4 * GB)); + } +} diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java index ffa24c5879..c1a92265a8 100644 --- a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java @@ -57,7 +57,7 @@ public class TikaPipesConfigTest extends TikaTest { @Test void testNumClientsRejectsNonPositive() { - // ArrayBlockingQueue(0)'s message-less IAE pointed at nothing; this names the field. + // The client queue's message-less IAE pointed at nothing; this names the field. PipesConfig config = new PipesConfig(); assertThrows(IllegalArgumentException.class, () -> config.setNumClients(0)); assertThrows(IllegalArgumentException.class, () -> config.setNumClients(-1)); diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesParserWarmClientTest.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesParserWarmClientTest.java new file mode 100644 index 0000000000..486fcc43ea --- /dev/null +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesParserWarmClientTest.java @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.tika.pipes.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.nio.file.Path; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import org.apache.tika.config.loader.TikaJsonConfig; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.parser.ParseContext; +import org.apache.tika.pipes.api.FetchEmitTuple; +import org.apache.tika.pipes.api.PipesResult; +import org.apache.tika.pipes.api.emitter.EmitKey; +import org.apache.tika.pipes.api.fetcher.FetchKey; + +/** + * Client selection is LIFO: sequential (non-concurrent) parses reuse the warm + * client instead of round-robining every fork awake. FIFO selection would + * cold-start one fork per request here. + */ +public class PipesParserWarmClientTest { + + @Test + public void sequentialParsesReuseTheWarmClient(@TempDir Path tmp) throws Exception { + String testDoc = "testOverlappingText.pdf"; + Path tikaConfigPath = PluginsTestHelper.getFileSystemFetcherConfig( + tmp, tmp.resolve("input"), tmp.resolve("output")); + PluginsTestHelper.copyTestFilesToTmpInput(tmp, testDoc); + + TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(tikaConfigPath); + PipesConfig pipesConfig = PipesConfig.load(tikaJsonConfig); + pipesConfig.setNumClients(3); + + try (PipesParser parser = PipesParser.load(tikaJsonConfig, pipesConfig, tikaConfigPath)) { + for (int i = 0; i < 4; i++) { + PipesResult result = parser.parse( + new FetchEmitTuple(testDoc + "-" + i, new FetchKey("fsf", testDoc), + new EmitKey(), new Metadata(), new ParseContext(), + FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP)); + assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS, result.status(), + "parse " + i + ": " + result.message()); + } + assertEquals(1, parser.startedServerCount(), + "sequential parses should stay on one warm fork"); + } + } +}
