This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch spill-scoreboard in repository https://gitbox.apache.org/repos/asf/tika.git
commit eaecd350acd25b62860caa93d32b50ef7fd0dad2 Author: tallison <[email protected]> AuthorDate: Mon Aug 24 20:32:31 2026 -0400 TIKA-4835 -- record per-stage pipes timings (StageTimings, PIPES_TIMING log) and add a property-gated temp-file spill scoreboard (SpillStats) --- CHANGES.txt | 10 + .../main/java/org/apache/tika/io/SpillStats.java | 126 ++++++++ .../org/apache/tika/io/TemporaryResources.java | 4 + .../org/apache/tika/pipes/api/PipesResult.java | 21 +- .../org/apache/tika/pipes/api/StageTimings.java | 36 +++ .../org/apache/tika/pipes/core/PipesClient.java | 94 +++++- .../org/apache/tika/pipes/core/PipesParser.java | 4 +- .../serialization/PipesResultDeserializer.java | 10 +- .../core/serialization/PipesResultSerializer.java | 4 + .../apache/tika/pipes/core/server/PipesWorker.java | 29 +- .../org/apache/tika/pipes/bench/PdfParseProbe.java | 162 ++++++++++ .../apache/tika/pipes/bench/PipesBenchmark.java | 341 +++++++++++++++++++++ .../tika/pipes/bench/PipesTimingAnalyzer.java | 136 ++++++++ .../test/resources/configs/tika-config-bench.json | 56 ++++ 14 files changed, 1020 insertions(+), 13 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index eea5ad329b..4e06deaba1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,15 @@ Release 4.1.0 - unreleased + * Pipes now records per-stage timings. Each PipesResult carries an optional + StageTimings (fetch/parse/emit/server-wall nanos) stamped by the forked + worker, and PipesClient emits one TSV-friendly line per parse on logger + org.apache.tika.pipes.timing (INFO; off unless enabled in the parent + process's log config) covering client wait, init, request serialize/write, + server wait, the server stages, and client total. Overhead is negligible + when the logger is disabled. Note for binary compatibility: the canonical + PipesResult constructor gained the serverTimings component; the previous + three-argument constructor remains (TIKA-4835). + * The Kafka pipes iterator no longer stops at the first empty poll. A newly subscribed consumer spends its first poll(s) joining the group and returns empty even when the topic has a backlog, so the iterator could enqueue zero diff --git a/tika-core/src/main/java/org/apache/tika/io/SpillStats.java b/tika-core/src/main/java/org/apache/tika/io/SpillStats.java new file mode 100644 index 0000000000..f92e31a6dd --- /dev/null +++ b/tika-core/src/main/java/org/apache/tika/io/SpillStats.java @@ -0,0 +1,126 @@ +/* + * 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.io; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; + +/** + * Diagnostic scoreboard of temp-file spills by calling site. Enabled only when the system + * property {@code tika.debug.spillStats} names an output file; otherwise every call is a + * single volatile read. The forked pipes worker may be hard-killed, so the summary is + * rewritten every {@link #DUMP_EVERY} spills as well as at shutdown. + */ +public final class SpillStats { + + public static final String PROP = "tika.debug.spillStats"; + private static final int DUMP_EVERY = 100; + private static final int FRAMES = 4; + + private static final Path OUT; + private static final Map<String, long[]> SITES = new ConcurrentHashMap<>(); + private static final AtomicLong TOTAL_FILES = new AtomicLong(); + private static final AtomicLong TOTAL_BYTES = new AtomicLong(); + + static { + String p = System.getProperty(PROP); + OUT = p == null || p.isBlank() ? null : Paths.get(p); + if (OUT != null) { + Runtime.getRuntime().addShutdownHook(new Thread(SpillStats::dump, "spill-stats-dump")); + } + } + + private SpillStats() { + } + + public static boolean enabled() { + return OUT != null; + } + + /** Captures the calling site; returns the key to pass to {@link #recordDelete}. */ + public static String recordCreate() { + // key = the io-layer spill site, then the first FRAMES frames outside org.apache.tika.io + // (the parser/detector that forced the spill); io-internal plumbing frames are skipped. + List<StackTraceElement> frames = StackWalker.getInstance().walk(s -> s + .map(StackWalker.StackFrame::toStackTraceElement) + .filter(f -> f.getClassName().startsWith("org.apache.tika.")) + .filter(f -> !f.getClassName().endsWith("TemporaryResources") + && !f.getClassName().endsWith("SpillStats")) + .collect(Collectors.toList())); + StringBuilder sb = new StringBuilder(); + int outside = 0; + for (int i = 0; i < frames.size() && outside < FRAMES; i++) { + StackTraceElement f = frames.get(i); + boolean io = f.getClassName().startsWith("org.apache.tika.io."); + if (i == 0 || !io) { + if (sb.length() > 0) { + sb.append('<'); + } + sb.append(shortName(f.getClassName())).append('.').append(f.getMethodName()); + if (!io) { + outside++; + } + } + } + String site = sb.toString(); + SITES.computeIfAbsent(site, k -> new long[2])[0]++; + return site; + } + + public static void recordDelete(String site, Path path) { + long size; + try { + size = Files.size(path); + } catch (IOException e) { + return; + } + SITES.computeIfAbsent(site, k -> new long[2])[1] += size; + TOTAL_BYTES.addAndGet(size); + if (TOTAL_FILES.incrementAndGet() % DUMP_EVERY == 0) { + dump(); + } + } + + private static String shortName(String cls) { + return cls.substring(cls.lastIndexOf('.') + 1); + } + + static synchronized void dump() { + List<String> lines = new ArrayList<>(); + lines.add("# spill stats pid=" + ProcessHandle.current().pid() + " files=" + TOTAL_FILES.get() + + " bytes=" + TOTAL_BYTES.get()); + lines.add("bytes\tcount\tsite"); + SITES.entrySet().stream() + .sorted((a, b) -> Long.compare(b.getValue()[1], a.getValue()[1])) + .forEach(e -> lines.add(e.getValue()[1] + "\t" + e.getValue()[0] + "\t" + e.getKey())); + try { + Path out = OUT.resolveSibling(OUT.getFileName() + "." + ProcessHandle.current().pid()); + Files.write(out, lines, StandardCharsets.UTF_8); + } catch (IOException e) { + // diagnostics only; never fail the parse + } + } +} diff --git a/tika-core/src/main/java/org/apache/tika/io/TemporaryResources.java b/tika-core/src/main/java/org/apache/tika/io/TemporaryResources.java index c1565ab86d..9213e997f1 100644 --- a/tika-core/src/main/java/org/apache/tika/io/TemporaryResources.java +++ b/tika-core/src/main/java/org/apache/tika/io/TemporaryResources.java @@ -88,8 +88,12 @@ public class TemporaryResources implements Closeable { final Path path = tempFileDir == null ? Files.createTempFile("apache-tika-", actualSuffix) : Files.createTempFile(tempFileDir, "apache-tika-", actualSuffix); + final String site = SpillStats.enabled() ? SpillStats.recordCreate() : null; addResource(() -> { try { + if (site != null) { + SpillStats.recordDelete(site, path); + } Files.delete(path); } catch (IOException e) { // delete when exit if current delete fail diff --git a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java index fabbcd25ca..bfe4881866 100644 --- a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java +++ b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java @@ -20,7 +20,8 @@ import java.io.Serializable; import org.apache.tika.pipes.api.emitter.EmitData; -public record PipesResult(RESULT_STATUS status, EmitData emitData, String message) implements Serializable { +public record PipesResult(RESULT_STATUS status, EmitData emitData, String message, + StageTimings serverTimings) implements Serializable { /** * High-level categorization of result statuses. @@ -187,15 +188,27 @@ public record PipesResult(RESULT_STATUS status, EmitData emitData, String messag } public PipesResult(RESULT_STATUS status) { - this(status, null, null); + this(status, null, null, null); } public PipesResult(RESULT_STATUS status, EmitData emitData) { - this(status, emitData, null); + this(status, emitData, null, null); } public PipesResult(RESULT_STATUS status, String message) { - this(status, null, message); + this(status, null, message, null); + } + + public PipesResult(RESULT_STATUS status, EmitData emitData, String message) { + this(status, emitData, message, null); + } + + /** + * Returns a copy of this result with the given server timings attached. + * Used on the server side to stamp timings onto a result before sending FINISHED. + */ + public PipesResult withServerTimings(StageTimings timings) { + return new PipesResult(status, emitData, message, timings); } /** diff --git a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/StageTimings.java b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/StageTimings.java new file mode 100644 index 0000000000..3a5bb1898b --- /dev/null +++ b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/StageTimings.java @@ -0,0 +1,36 @@ +/* + * 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.api; + +import java.io.Serializable; + +/** + * Server-side per-stage timings attached to a {@link PipesResult} for the + * structured timing log. + * <p> + * Values are nanoseconds; -1 indicates the stage did not run (e.g., emit was + * skipped for a passback result, or fetch failed before parse started). + * <p> + * Failure paths (OOM, TIMEOUT, UNSPECIFIED_CRASH) generally do not produce a + * normal FINISHED message and therefore carry no server timings — only + * client-side timings will be available for those parses. + */ +public record StageTimings(long fetchNanos, long parseNanos, long emitNanos, + long serverWallNanos) implements Serializable { + + public static final long NOT_RUN = -1L; +} diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java index 7b2b34a974..eb18670c65 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java @@ -43,6 +43,7 @@ import org.apache.tika.metadata.Metadata; import org.apache.tika.metadata.TikaCoreProperties; import org.apache.tika.pipes.api.FetchEmitTuple; import org.apache.tika.pipes.api.PipesResult; +import org.apache.tika.pipes.api.StageTimings; import org.apache.tika.pipes.api.emitter.EmitKey; import org.apache.tika.pipes.core.emitter.EmitDataImpl; import org.apache.tika.pipes.core.protocol.PayloadLimitExceededException; @@ -67,6 +68,9 @@ import org.apache.tika.utils.StringUtils; public class PipesClient implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(PipesClient.class); + /** Dedicated logger for structured per-parse timing output. One TSV-friendly + * line per parse at INFO; can be silenced or routed independently of LOG. */ + private static final Logger TIMING_LOG = LoggerFactory.getLogger("org.apache.tika.pipes.timing"); private static final AtomicInteger CLIENT_COUNTER = new AtomicInteger(0); public static final int SOCKET_CONNECT_TIMEOUT_MS = 60000; public static final int SOCKET_TIMEOUT_MILLIS = 60000; @@ -204,11 +208,27 @@ public class PipesClient implements Closeable { } public PipesResult process(FetchEmitTuple t) throws IOException, InterruptedException { + return process(t, 0L); + } + + /** + * Like {@link #process(FetchEmitTuple)} but accepts the time the caller spent + * waiting for a free client (e.g., {@code PipesParser}'s queue-poll time) so it + * can be reported in the structured timing log. + * + * @param t the parse request + * @param clientWaitNanos nanoseconds spent acquiring this client; 0 when called directly + */ + public PipesResult process(FetchEmitTuple t, long clientWaitNanos) throws IOException, InterruptedException { + long callStart = System.nanoTime(); // Container object to hold latest intermediate result if the parser is doing that IntermediateResult intermediateResult = new IntermediateResult(); PipesResult result = null; + long initStart = System.nanoTime(); + long initNanos; try { maybeInit(); + initNanos = System.nanoTime() - initStart; } catch (InterruptedException e) { // Same invariant as the in-flight path below: an abandoned connection, // here possibly half-established, must not be re-queued, and an @@ -217,20 +237,34 @@ public class PipesClient implements Closeable { closeConnection(); throw e; } catch (ServerInitializationException e) { + initNanos = System.nanoTime() - initStart; LOG.error("server initialization failed: {} ", t.getId(), e); closeConnection(); - return buildFatalResult(t.getId(), t.getEmitKey(), PipesResult.RESULT_STATUS.FAILED_TO_INITIALIZE, - intermediateResult.get(), e.getMessage()); + PipesResult fatal = buildFatalResult(t.getId(), t.getEmitKey(), + PipesResult.RESULT_STATUS.FAILED_TO_INITIALIZE, intermediateResult.get(), e.getMessage()); + logTiming(t.getId(), fatal, clientWaitNanos, initNanos, 0L, 0L, + System.nanoTime() - callStart); + return fatal; } catch (SecurityException e) { + initNanos = System.nanoTime() - initStart; LOG.error("security exception during initialization: {} ", t.getId()); closeConnection(); - return buildFatalResult(t.getId(), t.getEmitKey(), PipesResult.RESULT_STATUS.FAILED_TO_INITIALIZE, - intermediateResult.get()); + PipesResult fatal = buildFatalResult(t.getId(), t.getEmitKey(), + PipesResult.RESULT_STATUS.FAILED_TO_INITIALIZE, intermediateResult.get()); + logTiming(t.getId(), fatal, clientWaitNanos, initNanos, 0L, 0L, + System.nanoTime() - callStart); + return fatal; } + long reqWriteNanos = 0L; + long serverWaitNanos = 0L; try { + long reqWriteStart = System.nanoTime(); writeTask(t); + reqWriteNanos = System.nanoTime() - reqWriteStart; + long serverWaitStart = System.nanoTime(); result = waitForServer(t, intermediateResult); + serverWaitNanos = System.nanoTime() - serverWaitStart; filesProcessed++; // Update server manager's file counter for maxFilesProcessedPerProcess tracking serverManager.incrementFilesProcessed(pipesConfig.getMaxFilesProcessedPerProcess()); @@ -252,11 +286,52 @@ public class PipesClient implements Closeable { } catch (Exception e) { LOG.error("exception waiting for server to complete task: {} ", t.getId(), e); closeConnection(); - return buildFatalResult(t.getId(), t.getEmitKey(), UNSPECIFIED_CRASH, intermediateResult.get()); + PipesResult crash = buildFatalResult(t.getId(), t.getEmitKey(), UNSPECIFIED_CRASH, + intermediateResult.get()); + logTiming(t.getId(), crash, clientWaitNanos, initNanos, reqWriteNanos, serverWaitNanos, + System.nanoTime() - callStart); + return crash; } + logTiming(t.getId(), result, clientWaitNanos, initNanos, reqWriteNanos, serverWaitNanos, + System.nanoTime() - callStart); return result; } + /** + * Emits a single TSV-friendly line on logger {@code org.apache.tika.pipes.timing} + * with all client + server timings in microseconds. Server timings are -1 when + * unavailable (e.g., crash paths). + */ + private void logTiming(String id, PipesResult result, long clientWaitNanos, long initNanos, + long reqWriteNanos, long serverWaitNanos, long clientTotalNanos) { + if (!TIMING_LOG.isInfoEnabled()) { + return; + } + StageTimings st = result == null ? null : result.serverTimings(); + long serverFetchUs = (st == null) ? -1L : nanosToMicros(st.fetchNanos()); + long serverParseUs = (st == null) ? -1L : nanosToMicros(st.parseNanos()); + long serverEmitUs = (st == null) ? -1L : nanosToMicros(st.emitNanos()); + long serverWallUs = (st == null) ? -1L : nanosToMicros(st.serverWallNanos()); + String status = result == null ? "NULL" : result.status().name(); + TIMING_LOG.info("PIPES_TIMING client={} id={} status={} client_wait_us={} init_us={}" + + " req_serialize_us={} req_socket_us={}" + + " req_write_us={} server_wait_us={} server_fetch_us={} server_parse_us={}" + + " server_emit_us={} server_wall_us={} client_total_us={}", + pipesClientId, id, status, + nanosToMicros(clientWaitNanos), nanosToMicros(initNanos), + nanosToMicros(lastSerializeNanos), nanosToMicros(lastSocketWriteNanos), + nanosToMicros(reqWriteNanos), nanosToMicros(serverWaitNanos), + serverFetchUs, serverParseUs, serverEmitUs, serverWallUs, + nanosToMicros(clientTotalNanos)); + } + + private static long nanosToMicros(long nanos) { + if (nanos < 0) { + return nanos; + } + return nanos / 1000L; + } + private void maybeInit() throws InterruptedException, ServerInitializationException { boolean reconnect = false; @@ -343,13 +418,21 @@ public class PipesClient implements Closeable { socket.setSoTimeout((int) pipesConfig.getSocketTimeoutMillis()); } + /** Last serialize-only nanos (set by writeTask, read by process for the timing log). */ + private long lastSerializeNanos; + /** Last write+flush nanos (set by writeTask, read by process for the timing log). */ + private long lastSocketWriteNanos; + private void writeTask(FetchEmitTuple t) throws IOException { ConnectionTuple tuple = connectionTuple; if (tuple == null) { throw new IOException("connection closed"); } LOG.debug("pipesClientId={}: sending NEW_REQUEST for id={}", pipesClientId, t.getId()); + long sStart = System.nanoTime(); byte[] bytes = JsonPipesIpc.toBytes(PipesRequest.of(t)); + long sEnd = System.nanoTime(); + lastSerializeNanos = sEnd - sStart; // Fail fast before sending: the server would refuse the frame anyway, but only by // dying or dropping the connection, misreported as a crash. if (bytes.length > maxIpcPayloadBytes) { @@ -358,6 +441,7 @@ public class PipesClient implements Closeable { + maxIpcPayloadBytes + "; raise maxIpcPayloadBytes or shrink the request"); } PipesMessage.newRequest(bytes).write(tuple.output); + lastSocketWriteNanos = System.nanoTime() - sEnd; } /** 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 dd68dca002..5c1015bc3d 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 @@ -122,13 +122,15 @@ public class PipesParser implements Closeable { public PipesResult parse(FetchEmitTuple t) throws InterruptedException, PipesException, IOException { PipesClient client = null; + long pollStart = System.nanoTime(); try { client = clientQueue.pollFirst(pipesConfig.getMaxWaitForClientMillis(), TimeUnit.MILLISECONDS); + long clientWaitNanos = System.nanoTime() - pollStart; if (client == null) { return PipesResults.CLIENT_UNAVAILABLE_WITHIN_MS; } - return client.process(t); + return client.process(t, clientWaitNanos); } finally { if (client != null) { clientQueue.offerFirst(client); diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesResultDeserializer.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesResultDeserializer.java index 72e27ddc61..d2b515b7b9 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesResultDeserializer.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesResultDeserializer.java @@ -18,6 +18,7 @@ package org.apache.tika.pipes.core.serialization; import static org.apache.tika.pipes.core.serialization.PipesResultSerializer.EMIT_DATA; import static org.apache.tika.pipes.core.serialization.PipesResultSerializer.MESSAGE; +import static org.apache.tika.pipes.core.serialization.PipesResultSerializer.SERVER_TIMINGS; import static org.apache.tika.pipes.core.serialization.PipesResultSerializer.STATUS; import java.io.IOException; @@ -29,6 +30,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.tika.pipes.api.PipesResult; +import org.apache.tika.pipes.api.StageTimings; import org.apache.tika.pipes.core.emitter.EmitDataImpl; public class PipesResultDeserializer extends JsonDeserializer<PipesResult> { @@ -49,7 +51,13 @@ public class PipesResultDeserializer extends JsonDeserializer<PipesResult> { String message = readString(MESSAGE, root, null, false); - return new PipesResult(status, emitData, message); + StageTimings serverTimings = null; + JsonNode timingsNode = root.get(SERVER_TIMINGS); + if (timingsNode != null && !timingsNode.isNull()) { + serverTimings = mapper.treeToValue(timingsNode, StageTimings.class); + } + + return new PipesResult(status, emitData, message, serverTimings); } private static String readString(String key, JsonNode root, String defaultVal, boolean required) throws IOException { diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesResultSerializer.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesResultSerializer.java index ebf01a5a47..ddfff19e5f 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesResultSerializer.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesResultSerializer.java @@ -30,6 +30,7 @@ public class PipesResultSerializer extends JsonSerializer<PipesResult> { public static final String STATUS = "status"; public static final String EMIT_DATA = "emitData"; public static final String MESSAGE = "message"; + public static final String SERVER_TIMINGS = "serverTimings"; @Override public void serialize(PipesResult pipesResult, JsonGenerator jsonGenerator, SerializerProvider serializerProvider) throws IOException { @@ -41,6 +42,9 @@ public class PipesResultSerializer extends JsonSerializer<PipesResult> { if (!StringUtils.isBlank(pipesResult.message())) { jsonGenerator.writeStringField(MESSAGE, pipesResult.message()); } + if (pipesResult.serverTimings() != null) { + jsonGenerator.writeObjectField(SERVER_TIMINGS, pipesResult.serverTimings()); + } jsonGenerator.writeEndObject(); } } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesWorker.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesWorker.java index f08f3fed3c..0268ac3ad5 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesWorker.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesWorker.java @@ -45,6 +45,7 @@ import org.apache.tika.parser.ParseContext; import org.apache.tika.pipes.api.FetchEmitTuple; import org.apache.tika.pipes.api.ParseMode; import org.apache.tika.pipes.api.PipesResult; +import org.apache.tika.pipes.api.StageTimings; import org.apache.tika.pipes.api.emitter.EmitKey; import org.apache.tika.pipes.api.emitter.Emitter; import org.apache.tika.pipes.api.emitter.StreamEmitter; @@ -74,6 +75,12 @@ class PipesWorker implements Callable<PipesResult> { private final MetadataWriteLimiterFactory defaultMetadataWriteLimiterFactory; private final ParseMode defaultParseMode; + // Per-stage server-side timings, set as the worker proceeds. Stamped onto + // the PipesResult returned from call(). NOT_RUN means the stage was skipped. + private long fetchNanos = StageTimings.NOT_RUN; + private long parseNanos = StageTimings.NOT_RUN; + private long emitNanos = StageTimings.NOT_RUN; + public PipesWorker(FetchEmitTuple fetchEmitTuple, ParseContext parseContext, AutoDetectParser autoDetectParser, EmitterManager emitterManager, FetchHandler fetchHandler, ParseHandler parseHandler, EmitHandler emitHandler, MetadataWriteLimiterFactory defaultMetadataWriteLimiterFactory, @@ -91,6 +98,14 @@ class PipesWorker implements Callable<PipesResult> { @Override public PipesResult call() throws Exception { + long serverStart = System.nanoTime(); + PipesResult result = runWork(); + long serverWall = System.nanoTime() - serverStart; + return result.withServerTimings( + new StageTimings(fetchNanos, parseNanos, emitNanos, serverWall)); + } + + private PipesResult runWork() throws Exception { MetadataListAndEmbeddedBytes parseData = null; TempFileUnpackHandler tempHandler = null; FrictionlessUnpackHandler frictionlessHandler = null; @@ -110,11 +125,13 @@ class PipesWorker implements Callable<PipesResult> { // Check if we need to zip and emit embedded files UnpackHandler handler = parseContext.get(UnpackHandler.class); + long emitStart = System.nanoTime(); if (handler instanceof FrictionlessUnpackHandler) { frictionlessHandler = (FrictionlessUnpackHandler) handler; PipesResult frictionlessResult = emitFrictionlessOutput(frictionlessHandler, parseData); if (frictionlessResult != null) { // Frictionless emit failed - return the error + emitNanos = System.nanoTime() - emitStart; return frictionlessResult; } } else if (handler instanceof TempFileUnpackHandler) { @@ -122,11 +139,14 @@ class PipesWorker implements Callable<PipesResult> { PipesResult zipResult = zipAndEmitEmbeddedFiles(tempHandler); if (zipResult != null) { // Zipping/emitting failed - return the error + emitNanos = System.nanoTime() - emitStart; return zipResult; } } - return emitHandler.emitParseData(fetchEmitTuple, parseData, parseContext); + PipesResult emitted = emitHandler.emitParseData(fetchEmitTuple, parseData, parseContext); + emitNanos = System.nanoTime() - emitStart; + return emitted; } finally { // Clean up handlers if used if (frictionlessHandler != null) { @@ -490,13 +510,18 @@ class PipesWorker implements Callable<PipesResult> { Metadata metadata = localContext.newMetadata(); // Carry the caller's resource name and Content-Type detection hints (see javadoc). carryCallerHints(fetchEmitTuple.getMetadata(), metadata); + long fetchStart = System.nanoTime(); FetchHandler.TisOrResult tisOrResult = fetchHandler.fetch(fetchEmitTuple, metadata, localContext); + fetchNanos = System.nanoTime() - fetchStart; if (tisOrResult.pipesResult() != null) { return new ParseDataOrPipesResult(null, tisOrResult.pipesResult()); } try (TikaInputStream tis = tisOrResult.tis()) { - return parseHandler.parseWithStream(fetchEmitTuple, tis, metadata, localContext); + long parseStart = System.nanoTime(); + ParseDataOrPipesResult result = parseHandler.parseWithStream(fetchEmitTuple, tis, metadata, localContext); + parseNanos = System.nanoTime() - parseStart; + return result; } catch (SecurityException e) { LOG.error("security exception id={}", fetchEmitTuple.getId(), e); throw e; diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/bench/PdfParseProbe.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/bench/PdfParseProbe.java new file mode 100644 index 0000000000..d40dbb398f --- /dev/null +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/bench/PdfParseProbe.java @@ -0,0 +1,162 @@ +/* + * 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.bench; + +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.HashMap; +import java.util.Map; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; +import org.xml.sax.helpers.DefaultHandler; + +import org.apache.tika.io.TikaInputStream; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.parser.AutoDetectParser; +import org.apache.tika.parser.ParseContext; +import org.apache.tika.parser.pdf.PDFParserConfig; + +/** + * Self-sampling profile of a single Tika parse. Runs warmup, then parses the + * target file while a sampler thread polls the parse thread's stack at 100Hz + * and tallies the topmost {@code org.apache.pdfbox} or {@code org.apache.tika} + * frame seen on each sample. Prints the top 25 frames sorted by sample count. + * <p> + * Run: + * <pre> + * ./mvnw test -pl tika-pipes/tika-pipes-integration-tests \ + * -Dtest=PdfParseProbe -Dpipes.probe.run=true \ + * -Dpipes.probe.file=<path-to-pdf> \ + * [-Dpipes.probe.iterations=5] + * </pre> + */ +public class PdfParseProbe { + + @Test + @EnabledIfSystemProperty(named = "pipes.probe.run", matches = "true") + public void probe() throws Exception { + Path file = Paths.get(System.getProperty("pipes.probe.file")); + int iterations = Integer.getInteger("pipes.probe.iterations", 5); + + AutoDetectParser parser = new AutoDetectParser(); + PDFParserConfig pdfConfig = buildPdfConfig(); + ParseContext ctxTemplate = new ParseContext(); + ctxTemplate.set(PDFParserConfig.class, pdfConfig); + + System.out.println("PDFParserConfig overrides:" + + " extractMarkedContent=" + pdfConfig.isExtractMarkedContent() + + " extractAcroFormContent=" + pdfConfig.isExtractAcroFormContent() + + " extractAnnotationText=" + pdfConfig.isExtractAnnotationText() + + " extractBookmarksText=" + pdfConfig.isExtractBookmarksText() + + " extractActions=" + pdfConfig.isExtractActions() + + " extractInlineImages=" + pdfConfig.isExtractInlineImages()); + + // warmup + for (int i = 0; i < 2; i++) { + parseOnce(parser, file, ctxTemplate); + } + + Thread parseThread = Thread.currentThread(); + Map<String, Integer> counts = new HashMap<>(); + int[] totalSamples = {0}; + Object stopFlag = new Object(); + boolean[] running = {true}; + + Thread sampler = new Thread(() -> { + while (true) { + synchronized (stopFlag) { + if (!running[0]) { + return; + } + } + StackTraceElement[] st = parseThread.getStackTrace(); + totalSamples[0]++; + for (StackTraceElement frame : st) { + String cn = frame.getClassName(); + if (cn.startsWith("org.apache.pdfbox") || cn.startsWith("org.apache.tika") + || cn.startsWith("org.apache.fontbox")) { + String key = cn + "." + frame.getMethodName(); + counts.merge(key, 1, Integer::sum); + break; + } + } + try { + Thread.sleep(10); + } catch (InterruptedException e) { + return; + } + } + }, "stack-sampler"); + sampler.setDaemon(true); + sampler.start(); + + long t = System.nanoTime(); + for (int i = 0; i < iterations; i++) { + parseOnce(parser, file, ctxTemplate); + } + long elapsedMs = (System.nanoTime() - t) / 1_000_000L; + + synchronized (stopFlag) { + running[0] = false; + } + sampler.join(1000); + + System.out.println(); + System.out.println("=== PdfParseProbe ==="); + System.out.println("file: " + file); + System.out.println("iterations: " + iterations); + System.out.println("wall time: " + elapsedMs + "ms (" + (elapsedMs / iterations) + "ms/parse)"); + System.out.println("samples: " + totalSamples[0]); + System.out.println(); + System.out.println("top 25 hot frames (by sample count):"); + counts.entrySet().stream() + .sorted(Map.Entry.<String, Integer>comparingByValue().reversed()) + .limit(25) + .forEach(e -> System.out.printf(" %5d %s%n", e.getValue(), e.getKey())); + } + + private static void parseOnce(AutoDetectParser parser, Path file, ParseContext ctxTemplate) + throws Exception { + ParseContext ctx = new ParseContext(); + PDFParserConfig pc = ctxTemplate.get(PDFParserConfig.class); + if (pc != null) { + ctx.set(PDFParserConfig.class, pc); + } + try (TikaInputStream is = TikaInputStream.get(file)) { + parser.parse(is, new DefaultHandler(), new Metadata(), ctx); + } + } + + private static PDFParserConfig buildPdfConfig() { + PDFParserConfig c = new PDFParserConfig(); + applyBool("pipes.probe.pdf.extractMarkedContent", c::setExtractMarkedContent); + applyBool("pipes.probe.pdf.extractAcroFormContent", c::setExtractAcroFormContent); + applyBool("pipes.probe.pdf.extractAnnotationText", c::setExtractAnnotationText); + applyBool("pipes.probe.pdf.extractBookmarksText", c::setExtractBookmarksText); + applyBool("pipes.probe.pdf.extractActions", c::setExtractActions); + applyBool("pipes.probe.pdf.extractInlineImages", c::setExtractInlineImages); + return c; + } + + private static void applyBool(String prop, java.util.function.Consumer<Boolean> setter) { + String v = System.getProperty(prop); + if (v != null) { + setter.accept(Boolean.parseBoolean(v)); + } + } +} diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/bench/PipesBenchmark.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/bench/PipesBenchmark.java new file mode 100644 index 0000000000..c5e23b0d51 --- /dev/null +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/bench/PipesBenchmark.java @@ -0,0 +1,341 @@ +/* + * 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.bench; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.attribute.BasicFileAttributes; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LoggerContext; +import org.apache.logging.log4j.core.appender.AsyncAppender; +import org.apache.logging.log4j.core.appender.FileAppender; +import org.apache.logging.log4j.core.config.AppenderRef; +import org.apache.logging.log4j.core.config.Configuration; +import org.apache.logging.log4j.core.config.LoggerConfig; +import org.apache.logging.log4j.core.layout.PatternLayout; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; +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; +import org.apache.tika.pipes.core.PipesConfig; +import org.apache.tika.pipes.core.PipesParser; +import org.apache.tika.pipes.core.PluginsTestHelper; + +/** + * Benchmark harness that drives a corpus through the pipes pipeline and writes + * one PIPES_TIMING TSV-friendly line per parse to a file for offline analysis. + * <p> + * <b>Run:</b> + * <pre> + * ./mvnw test -pl tika-pipes/tika-pipes-integration-tests \ + * -Dtest=PipesBenchmark -Dpipes.bench.run=true \ + * -Dpipes.bench.corpus=<corpus dir> \ + * [-Dpipes.bench.mock.ok=50] [-Dpipes.bench.mock.oom=5] [-Dpipes.bench.mock.timeout=2] \ + * [-Dpipes.bench.timing.out=<path>] [-Dpipes.bench.threads=8] + * </pre> + * <p> + * The harness configures the parent JVM's log4j2 to route the + * {@code org.apache.tika.pipes.timing} logger to the timing file. The forked + * PipesServer JVMs do not emit timing logs themselves — they stamp per-stage + * timings onto the PipesResult and the parent's PipesClient logs the line. + */ +public class PipesBenchmark { + + private static final String MOCK_OK_XML = "<?xml version=\"1.0\" encoding=\"UTF-8\" ?>" + + "<mock>" + + "<metadata action=\"add\" name=\"dc:creator\">Bench OK Author</metadata>" + + "<write element=\"p\">Bench OK content</write>" + + "</mock>"; + + private static final String FETCHER_NAME = "fsf"; + private static final String EMITTER_NAME = "fse"; + + @Test + @EnabledIfSystemProperty(named = "pipes.bench.run", matches = "true") + public void run(@TempDir Path tmp) throws Exception { + Path corpusSource = readCorpusDir(); + int mockOk = Integer.getInteger("pipes.bench.mock.ok", 50); + int threads = Integer.getInteger("pipes.bench.threads", 8); + Path timingOut = Paths.get(System.getProperty("pipes.bench.timing.out", + tmp.resolve("pipes-timing.tsv").toString())).toAbsolutePath(); + + int warmupPasses = Integer.getInteger("pipes.bench.warmup-passes", 0); + + Path inputDir = tmp.resolve("input"); + Path outputDir = tmp.resolve("output"); + Files.createDirectories(inputDir); + Files.createDirectories(outputDir); + + // Stage corpus + mock files + copyCorpus(corpusSource, inputDir); + writeMocks(inputDir, "bench-ok-", mockOk, MOCK_OK_XML); + + long fileCount = countFiles(inputDir); + System.out.println("PipesBenchmark: corpus=" + inputDir + " files=" + fileCount); + System.out.println("PipesBenchmark: warmup-passes=" + warmupPasses); + System.out.println("PipesBenchmark: timing log -> " + timingOut); + + Path tikaConfigPath = PluginsTestHelper.getFileSystemFetcherConfig( + "tika-config-bench.json", tmp, inputDir, outputDir, false); + TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(tikaConfigPath); + PipesConfig pipesConfig = PipesConfig.load(tikaJsonConfig); + Integer numClientsOverride = Integer.getInteger("pipes.bench.num-clients"); + if (numClientsOverride != null) { + pipesConfig.setNumClients(numClientsOverride); + } + if (Boolean.getBoolean("pipes.bench.shared-server")) { + pipesConfig.setUseSharedServer(true); + } + if (Boolean.getBoolean("pipes.bench.cap-cpu")) { + // Per-client mode runs N forked JVMs. Each one defaults its GC, JIT, + // and common ForkJoinPool sizes to Runtime.availableProcessors(), so + // 4 JVMs on 16 cores spawn ~64 GC threads + ~60 FJP threads + 16 JIT + // threads -- way more than the 4 actually-active parse threads. + // We size each JVM to a fair slice of the *non-parent* CPU budget so + // the parent isn't starved (which causes pathological req_socket_us + // tail latency from preemption between clock reads). + int cores = Runtime.getRuntime().availableProcessors(); + int parentReserved = Integer.getInteger("pipes.bench.parent-cores", 2); + int n = Math.max(1, pipesConfig.getNumClients()); + int forkBudget = Math.max(1, cores - parentReserved); + int slice = Math.max(1, forkBudget / n); + pipesConfig.getForkedJvmArgs().add("-XX:ActiveProcessorCount=" + slice); + System.out.println("PipesBenchmark: capping forked JVMs to " + slice + + " active CPUs (host=" + cores + ", parentReserved=" + parentReserved + + ", numClients=" + n + ")"); + } + System.out.println("PipesBenchmark: numClients=" + pipesConfig.getNumClients() + + " sharedServer=" + pipesConfig.isUseSharedServer() + + " forkedJvmArgs=" + pipesConfig.getForkedJvmArgs()); + + long progressInterval = Long.getLong("pipes.bench.progress.interval", 1000); + ExecutorService executor = Executors.newFixedThreadPool(threads); + long benchStart = System.nanoTime(); + + BenchCounters measured; + try (PipesParser pipesParser = PipesParser.load(tikaJsonConfig, pipesConfig, tikaConfigPath)) { + // Warmup passes - run BEFORE attaching the timing appender so + // these parses don't pollute the measurement TSV. + for (int pass = 0; pass < warmupPasses; pass++) { + long warmStart = System.nanoTime(); + runOnePass(pipesParser, executor, inputDir, progressInterval, "warmup-" + (pass + 1)); + long warmMs = (System.nanoTime() - warmStart) / 1_000_000L; + System.out.println("PipesBenchmark: warmup pass " + (pass + 1) + " done in " + + warmMs + "ms"); + } + + // Attach the timing appender now that JVMs are warm + attachTimingAppender(timingOut); + + long measureStart = System.nanoTime(); + measured = runOnePass(pipesParser, executor, inputDir, progressInterval, "measured"); + long measureMs = (System.nanoTime() - measureStart) / 1_000_000L; + System.out.println("PipesBenchmark: measured pass done in " + measureMs + "ms"); + } finally { + executor.shutdown(); + executor.awaitTermination(60, TimeUnit.SECONDS); + } + long benchWallMs = (System.nanoTime() - benchStart) / 1_000_000L; + + // Flush log4j so the timing file is fully written + LogManager.shutdown(); + + System.out.println("PipesBenchmark: done in " + benchWallMs + "ms"); + System.out.println(" files=" + measured.total.get() + + " success=" + measured.success.get() + + " other=" + measured.other.get()); + System.out.println(" timing log: " + timingOut); + } + + /** Per-pass tally - counters only, no result accumulation. */ + private static final class BenchCounters { + final AtomicInteger total = new AtomicInteger(); + final AtomicInteger success = new AtomicInteger(); + final AtomicInteger other = new AtomicInteger(); + } + + /** + * Submits all corpus files via an ExecutorCompletionService, drains + * results as they complete (so {@link PipesResult} instances aren't + * retained), and prints a heartbeat every {@code progressInterval} + * completions. Tally is kept in counters only -- safe for million-file + * runs. + */ + private static BenchCounters runOnePass(PipesParser pipesParser, ExecutorService executor, + Path inputDir, long progressInterval, + String passName) throws Exception { + ExecutorCompletionService<PipesResult> ecs = new ExecutorCompletionService<>(executor); + long submitted = 0; + try (var stream = Files.walk(inputDir)) { + for (var iter = stream.filter(Files::isRegularFile).iterator(); iter.hasNext(); ) { + Path p = iter.next(); + String key = inputDir.relativize(p).toString(); + ecs.submit(() -> pipesParser.parse(new FetchEmitTuple( + key, + new FetchKey(FETCHER_NAME, key), + new EmitKey(EMITTER_NAME, ""), + new Metadata(), + new ParseContext(), + FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP))); + submitted++; + } + } + + BenchCounters counters = new BenchCounters(); + long passStart = System.nanoTime(); + long lastReportNanos = passStart; + int lastReportedTotal = 0; + for (long i = 0; i < submitted; i++) { + PipesResult r = ecs.take().get(); + // Discard r reference asap so the heap doesn't accumulate metadata payloads. + int total = counters.total.incrementAndGet(); + if (r.isSuccess()) { + counters.success.incrementAndGet(); + } else { + counters.other.incrementAndGet(); + } + r = null; + if (progressInterval > 0 && total % progressInterval == 0) { + long now = System.nanoTime(); + long elapsedMs = (now - passStart) / 1_000_000L; + long sinceLastMs = Math.max(1, (now - lastReportNanos) / 1_000_000L); + int sinceLast = total - lastReportedTotal; + double overallRate = total * 1000.0 / Math.max(1, elapsedMs); + double recentRate = sinceLast * 1000.0 / sinceLastMs; + System.out.printf( + "PipesBenchmark[%s]: %,d/%,d done elapsed=%,dms overall=%.0f f/s recent=%.0f f/s success=%,d other=%,d%n", + passName, total, submitted, elapsedMs, + overallRate, recentRate, + counters.success.get(), counters.other.get()); + lastReportNanos = now; + lastReportedTotal = total; + } + } + return counters; + } + + private static Path readCorpusDir() { + String s = System.getProperty("pipes.bench.corpus"); + if (s == null || s.isBlank()) { + throw new IllegalArgumentException( + "set -Dpipes.bench.corpus=<path-to-corpus-dir>"); + } + Path p = Paths.get(s); + if (!Files.isDirectory(p)) { + throw new IllegalArgumentException("corpus dir does not exist: " + p); + } + return p; + } + + private static void copyCorpus(Path src, Path dst) throws IOException { + Files.walkFileTree(src, new SimpleFileVisitor<>() { + @Override + public java.nio.file.FileVisitResult visitFile(Path file, BasicFileAttributes attrs) + throws IOException { + Path rel = src.relativize(file); + Path target = dst.resolve(rel.toString()); + if (target.getParent() != null) { + Files.createDirectories(target.getParent()); + } + Files.copy(file, target); + return java.nio.file.FileVisitResult.CONTINUE; + } + }); + } + + private static void writeMocks(Path dir, String prefix, int count, String xml) throws IOException { + for (int i = 0; i < count; i++) { + Files.writeString(dir.resolve(prefix + i + ".xml"), xml, StandardCharsets.UTF_8); + } + } + + private static long countFiles(Path dir) throws IOException { + try (var stream = Files.walk(dir)) { + return stream.filter(Files::isRegularFile).count(); + } + } + + /** + * Programmatically attaches a file appender to the {@code org.apache.tika.pipes.timing} + * logger so each PIPES_TIMING line lands in the requested TSV file. + * <p> + * The {@link FileAppender} is wrapped in an {@link AsyncAppender} so concurrent + * PipesClient threads don't contend on the synchronous file write at + * high throughput (matters at million-file scale). + */ + private static void attachTimingAppender(Path timingOut) throws IOException { + if (timingOut.getParent() != null) { + Files.createDirectories(timingOut.getParent()); + } + LoggerContext ctx = (LoggerContext) LogManager.getContext(false); + Configuration cfg = ctx.getConfiguration(); + + PatternLayout layout = PatternLayout.newBuilder() + .withPattern("%m%n") + .withConfiguration(cfg) + .build(); + + FileAppender fileAppender = FileAppender.newBuilder() + .setName("PipesBenchTimingFile") + .withFileName(timingOut.toString()) + .withAppend(false) + .setLayout(layout) + .setConfiguration(cfg) + .build(); + fileAppender.start(); + cfg.addAppender(fileAppender); + + AppenderRef fileRef = AppenderRef.createAppenderRef("PipesBenchTimingFile", Level.INFO, null); + AsyncAppender asyncAppender = AsyncAppender.newBuilder() + .setName("PipesBenchTimingAsync") + .setConfiguration(cfg) + .setAppenderRefs(new AppenderRef[]{fileRef}) + .setBlocking(true) + .setBufferSize(8192) + .build(); + asyncAppender.start(); + cfg.addAppender(asyncAppender); + + AppenderRef asyncRef = AppenderRef.createAppenderRef( + "PipesBenchTimingAsync", Level.INFO, null); + AppenderRef[] refs = new AppenderRef[]{asyncRef}; + LoggerConfig loggerConfig = LoggerConfig.createLogger(false, Level.INFO, + "org.apache.tika.pipes.timing", "true", refs, null, cfg, null); + loggerConfig.addAppender(asyncAppender, Level.INFO, null); + cfg.addLogger("org.apache.tika.pipes.timing", loggerConfig); + + ctx.updateLoggers(); + } +} diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/bench/PipesTimingAnalyzer.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/bench/PipesTimingAnalyzer.java new file mode 100644 index 0000000000..20d7b717ec --- /dev/null +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/bench/PipesTimingAnalyzer.java @@ -0,0 +1,136 @@ +/* + * 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.bench; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.stream.Stream; + +/** + * Reads a PIPES_TIMING TSV produced by {@link PipesBenchmark} and prints + * per-stage p50/p95/p99/max plus a status count summary. + * <p> + * Run: {@code java org.apache.tika.pipes.bench.PipesTimingAnalyzer <tsv-path>} + * <p> + * Negative values (-1) mean the stage was not measured for that row (e.g., + * server-side timings on a crash path); these rows are excluded from per-stage + * percentiles but still counted under "status". + */ +public final class PipesTimingAnalyzer { + + private static final String[] STAGES = { + "client_wait_us", "init_us", "req_serialize_us", "req_socket_us", "req_write_us", + "server_wait_us", "server_fetch_us", "server_parse_us", "server_emit_us", + "server_wall_us", "client_total_us" + }; + + private PipesTimingAnalyzer() { + } + + public static void main(String[] args) throws Exception { + if (args.length != 1) { + System.err.println("usage: PipesTimingAnalyzer <pipes-timing.tsv>"); + System.exit(1); + } + Path tsv = Paths.get(args[0]); + if (!Files.isRegularFile(tsv)) { + System.err.println("not a file: " + tsv); + System.exit(1); + } + + Map<String, List<Long>> byStage = new LinkedHashMap<>(); + for (String s : STAGES) { + byStage.put(s, new ArrayList<>()); + } + Map<String, Integer> statusCounts = new TreeMap<>(); + int totalRows = 0; + + try (Stream<String> lines = Files.lines(tsv)) { + for (String line : (Iterable<String>) lines::iterator) { + if (!line.contains("PIPES_TIMING")) { + continue; + } + totalRows++; + Map<String, String> kv = parseLine(line); + String status = kv.getOrDefault("status", "UNKNOWN"); + statusCounts.merge(status, 1, Integer::sum); + for (String s : STAGES) { + String v = kv.get(s); + if (v == null) { + continue; + } + long n = Long.parseLong(v); + if (n >= 0) { + byStage.get(s).add(n); + } + } + } + } + + System.out.println("rows: " + totalRows); + System.out.println("status counts:"); + for (Map.Entry<String, Integer> e : statusCounts.entrySet()) { + System.out.printf(" %-30s %d%n", e.getKey(), e.getValue()); + } + System.out.println(); + System.out.printf("%-20s %10s %10s %10s %10s %10s%n", + "stage", "n", "p50_us", "p95_us", "p99_us", "max_us"); + for (String s : STAGES) { + List<Long> values = byStage.get(s); + if (values.isEmpty()) { + System.out.printf("%-20s %10d %10s %10s %10s %10s%n", + s, 0, "-", "-", "-", "-"); + continue; + } + Collections.sort(values); + System.out.printf("%-20s %10d %10d %10d %10d %10d%n", + s, values.size(), + pct(values, 0.50), pct(values, 0.95), pct(values, 0.99), + values.get(values.size() - 1)); + } + } + + private static long pct(List<Long> sorted, double p) { + if (sorted.isEmpty()) { + return 0; + } + int idx = (int) Math.min(sorted.size() - 1L, Math.ceil(p * sorted.size()) - 1); + if (idx < 0) { + idx = 0; + } + return sorted.get(idx); + } + + private static Map<String, String> parseLine(String line) { + Map<String, String> out = new LinkedHashMap<>(); + for (String tok : line.split("\\s+")) { + int eq = tok.indexOf('='); + if (eq <= 0) { + continue; + } + out.put(tok.substring(0, eq), tok.substring(eq + 1)); + } + return out; + } +} diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-bench.json b/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-bench.json new file mode 100644 index 0000000000..367850677b --- /dev/null +++ b/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-bench.json @@ -0,0 +1,56 @@ +{ + "content-handler-factory": { + "basic-content-handler-factory": { + "type": "TEXT", + "writeLimit": -1, + "throwOnWriteLimitReached": true + } + }, + "fetchers": { + "fsf": { + "file-system-fetcher": { + "basePath": "FETCHER_BASE_PATH", + "extractFileSystemMetadata": false + } + } + }, + "emitters": { + "fse": { + "file-system-emitter": { + "basePath": "EMITTER_BASE_PATH", + "fileExtension": "json", + "onExists": "OVERWRITE" + } + } + }, + "pipes-iterator": { + "file-system-pipes-iterator": { + "basePath": "FETCHER_BASE_PATH", + "countTotal": true, + "fetcherId": "fsf", + "emitterId": "fse" + } + }, + "pipes": { + "parseMode": "RMETA", + "onParseException": "EMIT", + "numClients": 2, + "useSharedServer": false, + "emitIntermediateResults": "EMIT_INTERMEDIATE_RESULTS", + "forkedJvmArgs": ["-Xmx512m"], + "emitStrategy": { + "type": "DYNAMIC", + "thresholdBytes": 1000000 + } + }, + "auto-detect-parser": { + "throwOnZeroBytes": false + }, + "parse-context": { + "mock-digester-factory": {}, + "timeout-limits": { + "progressTimeoutMillis": 5000 + } + }, + "plugin-roots": "PLUGINS_PATHS" +}
