Skip to content

Commit 770af54

Browse files
tballisontballison
committed
Automatically add slices and/or log underprovisioned pipes configurations
1 parent c9da280 commit 770af54

1 file changed

Lines changed: 101 additions & 0 deletions

File tree

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

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,17 @@ public class PerClientServerManager implements ServerManager {
5151
private static final Logger LOG = LoggerFactory.getLogger(PerClientServerManager.class);
5252
private static final long WAIT_ON_DESTROY_MS = 10000;
5353
public static final int SOCKET_CONNECT_TIMEOUT_MS = 60000;
54+
/** Cores reserved for the parent JVM when auto-sizing forked JVMs'
55+
* -XX:ActiveProcessorCount. The parent has client-side serialization,
56+
* response deserialization, and heartbeat bookkeeping; if it's CPU-starved
57+
* small operations like socket flush show pathological tail latency. */
58+
private static final int PARENT_RESERVED_CORES = 2;
59+
/** Don't auto-cap below this many CPUs per fork. At cap=1 the fork's only
60+
* CPU is fully consumed by parsing, so its socket-reader thread can't run
61+
* and the parent's writes block on receiver-side back-pressure -- worse
62+
* than no cap at all. This guard matters for small k8s pods where the
63+
* formula could otherwise produce slice=1. */
64+
private static final int MIN_AUTO_CAP_SLICE = 2;
5465

5566
private final PipesConfig pipesConfig;
5667
private final Path tikaConfigPath;
@@ -67,6 +78,62 @@ public PerClientServerManager(PipesConfig pipesConfig, Path tikaConfigPath, int
6778
this.pipesConfig = pipesConfig;
6879
this.tikaConfigPath = tikaConfigPath;
6980
this.clientId = clientId;
81+
// Emit CPU-sizing diagnostics once per PipesParser (only on the first client).
82+
if (clientId == 0) {
83+
logCpuSizing();
84+
}
85+
}
86+
87+
/**
88+
* Emits a one-shot summary of how the auto-cap will behave for this PipesParser,
89+
* plus warnings for clearly-pathological provisioning. Grep for "pipes-cpu-sizing"
90+
* in logs to see the decision the JVM made.
91+
*/
92+
private void logCpuSizing() {
93+
int hostCores = Runtime.getRuntime().availableProcessors();
94+
int numClients = pipesConfig.getNumClients();
95+
boolean userSetCap = pipesConfig.getForkedJvmArgs().stream()
96+
.anyMatch(a -> a.startsWith("-XX:ActiveProcessorCount="));
97+
98+
// Hostile environment: fewer than 2 cores means the parser thread, GC, JIT,
99+
// and protocol heartbeat all share one CPU. Pipes will run but tail latency
100+
// will be poor regardless of numClients.
101+
if (hostCores < 2) {
102+
LOG.warn("pipes-cpu-sizing: hostCores={} is below the practical minimum. " +
103+
"Each fork JVM needs roughly 2 CPUs (1 for parsing, 1 for GC/JIT/" +
104+
"protocol heartbeat); on a single-CPU host these contend with each " +
105+
"other and performance will be poor.", hostCores);
106+
}
107+
108+
// Over-provisioned: numClients packed too tightly given the host's cores.
109+
// Triggers earlier than the slice<MIN guard so the user is warned even
110+
// when they explicitly set -XX:ActiveProcessorCount themselves.
111+
if (numClients > 1 && numClients * MIN_AUTO_CAP_SLICE + PARENT_RESERVED_CORES > hostCores) {
112+
int recommendedMax = Math.max(1,
113+
(hostCores - PARENT_RESERVED_CORES) / MIN_AUTO_CAP_SLICE);
114+
LOG.warn("pipes-cpu-sizing: numClients={} is over-provisioned for {}-core " +
115+
"host. Recommended max for this host: numClients={}. Forks need at " +
116+
"least {} CPUs each plus {} reserved for the parent JVM; otherwise " +
117+
"GC/JIT/protocol threads contend with parser threads across forks.",
118+
numClients, hostCores, recommendedMax,
119+
MIN_AUTO_CAP_SLICE, PARENT_RESERVED_CORES);
120+
}
121+
122+
// Always-on summary so ops can see what was decided. Grep for "pipes-cpu-sizing".
123+
String capDecision;
124+
if (userSetCap) {
125+
capDecision = "user-set in forkedJvmArgs";
126+
} else if (numClients <= 1) {
127+
capDecision = "n/a (single fork; not capped)";
128+
} else {
129+
int budget = Math.max(1, hostCores - PARENT_RESERVED_CORES);
130+
int slice = budget / numClients;
131+
capDecision = (slice >= MIN_AUTO_CAP_SLICE)
132+
? "slice=" + slice
133+
: "skipped (slice<" + MIN_AUTO_CAP_SLICE + ")";
134+
}
135+
LOG.info("pipes-cpu-sizing: hostCores={}, numClients={}, parentReserved={}, " +
136+
"autoCap={}", hostCores, numClients, PARENT_RESERVED_CORES, capDecision);
70137
}
71138

72139
@Override
@@ -305,6 +372,7 @@ private String[] getCommandline() throws IOException {
305372
boolean hasHeadless = false;
306373
boolean hasExitOnOOM = false;
307374
boolean hasLog4j = false;
375+
boolean hasActiveProcessorCount = false;
308376
String origGCString = null;
309377
String newGCLogString = null;
310378

@@ -321,12 +389,45 @@ private String[] getCommandline() throws IOException {
321389
if (arg.startsWith("-Dlog4j.configuration") || arg.startsWith("-Dlog4j2.configuration")) {
322390
hasLog4j = true;
323391
}
392+
if (arg.startsWith("-XX:ActiveProcessorCount=")) {
393+
hasActiveProcessorCount = true;
394+
}
324395
if (arg.startsWith("-Xloggc:")) {
325396
origGCString = arg;
326397
newGCLogString = arg.replace("${pipesClientId}", "id-" + clientId);
327398
}
328399
}
329400

401+
// If the user hasn't explicitly set -XX:ActiveProcessorCount, size each
402+
// forked JVM's view of CPUs to a fair slice of the host. Otherwise each
403+
// JVM defaults its GC, JIT, and common ForkJoinPool to "all cores", which
404+
// means N forked JVMs collectively spawn N x cores GC threads etc. and
405+
// fight each other. We also reserve PARENT_RESERVED_CORES so the parent
406+
// JVM (which serializes requests, deserializes responses, runs heartbeat
407+
// bookkeeping) isn't starved for CPU.
408+
// Skip the auto-cap when the computed slice would drop below
409+
// MIN_AUTO_CAP_SLICE -- below that, the fork can't keep its socket
410+
// reader responsive and back-pressures the parent.
411+
if (!hasActiveProcessorCount && pipesConfig.getNumClients() > 1) {
412+
int hostCores = Runtime.getRuntime().availableProcessors();
413+
int forkBudget = Math.max(1, hostCores - PARENT_RESERVED_CORES);
414+
int slice = forkBudget / pipesConfig.getNumClients();
415+
if (slice >= MIN_AUTO_CAP_SLICE) {
416+
configArgs.add("-XX:ActiveProcessorCount=" + slice);
417+
LOG.debug("clientId={}: auto-injected -XX:ActiveProcessorCount={} " +
418+
"(hostCores={}, parentReserved={}, numClients={})",
419+
clientId, slice, hostCores, PARENT_RESERVED_CORES,
420+
pipesConfig.getNumClients());
421+
} else {
422+
LOG.info("clientId={}: skipping -XX:ActiveProcessorCount auto-cap " +
423+
"(would yield slice={} < MIN_AUTO_CAP_SLICE={}; " +
424+
"hostCores={}, parentReserved={}, numClients={}). " +
425+
"Consider lowering numClients on this host.",
426+
clientId, slice, MIN_AUTO_CAP_SLICE, hostCores,
427+
PARENT_RESERVED_CORES, pipesConfig.getNumClients());
428+
}
429+
}
430+
330431
if (origGCString != null && newGCLogString != null) {
331432
configArgs.remove(origGCString);
332433
configArgs.add(newGCLogString);

0 commit comments

Comments
 (0)