This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4868-performance-improvements in repository https://gitbox.apache.org/repos/asf/tika.git
commit 5bdf823f0e291b14df2c9d07eca0b9f4e0d6e624 Author: tallison <[email protected]> AuthorDate: Tue Sep 1 17:24:40 2026 -0400 TIKA-4868: overlap the intermediate-result ACK with the parse; add per-stage timing logs --- CHANGES.txt | 5 ++ .../org/apache/tika/pipes/core/PipesClient.java | 69 ++++++++++++++++++++++ .../tika/pipes/core/server/ConnectionHandler.java | 61 ++++++++++++++++++- .../apache/tika/pipes/core/server/PipesServer.java | 68 ++++++++++++++++++++- .../apache/tika/pipes/core/server/PipesWorker.java | 48 ++++++++++++++- .../tika/pipes/core/server/ServerProtocolIO.java | 66 ++++++++++++++++++++- .../server/core/resource/PipesParsingHelper.java | 33 +++++++++++ 7 files changed, 343 insertions(+), 7 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 8c2b56af66..ecfea4793e 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,10 @@ Release 4.1.0 - unreleased + * Pipes workers no longer stall between pre-parse and parse waiting + for the client to acknowledge the intermediate-result frame; the + ACK round trip now overlaps the parse. Adds opt-in per-request + timing logs on org.apache.tika.pipes.timing.* (TIKA-4868). + * DefaultDetector honors CONTENT_TYPE_USER_OVERRIDE and CONTENT_TYPE_PARSER_OVERRIDE before running magic detection, matching CompositeDetector's contract. Removes the second full magic scan every 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 3988f9c90d..cb21f1deba 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 @@ -67,6 +67,10 @@ import org.apache.tika.utils.StringUtils; public class PipesClient implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(PipesClient.class); + /** Per-parse latency breakdown; one key=value line per parse. Route or silence + * independently of {@link #LOG}. Diagnostic only -- off unless INFO is enabled. */ + private static final Logger TIMING_LOG = + LoggerFactory.getLogger("org.apache.tika.pipes.timing.client"); 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; @@ -91,6 +95,17 @@ public class PipesClient implements Closeable { private volatile long connectionGeneration = Long.MAX_VALUE; private int filesProcessed = 0; + // Per-parse timing scratch. A PipesClient serves one parse at a time (the pool hands + // out an exclusive client), so plain fields are safe and avoid per-parse allocation. + private long tInitNanos; + private long tReqSerNanos; + private long tReqWriteNanos; + private long tFirstFrameNanos; + private long tFinishedNanos; + private long tRespDeserNanos; + private long tAckNanos; + private int tFrames; + /** * Creates a PipesClient with the given server manager. * <p> @@ -214,8 +229,12 @@ public class PipesClient implements Closeable { // Container object to hold latest intermediate result if the parser is doing that IntermediateResult intermediateResult = new IntermediateResult(); PipesResult result = null; + long callStart = System.nanoTime(); + resetTimings(); + long initStart = System.nanoTime(); try { maybeInit(); + tInitNanos = 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 @@ -277,9 +296,43 @@ public class PipesClient implements Closeable { closeConnection(); return buildFatalResult(t.getId(), t.getEmitKey(), UNSPECIFIED_CRASH, intermediateResult.get()); } + logTiming(t.getId(), result, System.nanoTime() - callStart); return result; } + private void resetTimings() { + tInitNanos = 0; + tReqSerNanos = 0; + tReqWriteNanos = 0; + tFirstFrameNanos = 0; + tFinishedNanos = 0; + tRespDeserNanos = 0; + tAckNanos = 0; + tFrames = 0; + } + + /** + * One line per parse on {@code org.apache.tika.pipes.timing.client}, microseconds. + * {@code finished_us} is measured from the end of the request write to the moment the + * FINISHED frame is fully read, so it contains the worker's own work plus everything + * the response leg costs; the worker logs its side under the same {@code id}. + */ + private void logTiming(String id, PipesResult result, long totalNanos) { + if (!TIMING_LOG.isInfoEnabled()) { + return; + } + TIMING_LOG.info("CLIENT_TIMING client={} id={} status={} init_us={} req_ser_us={}" + + " req_write_us={} first_frame_us={} finished_us={} resp_deser_us={}" + + " ack_us={} frames={} total_us={}", + pipesClientId, id, result == null ? "NULL" : result.status().name(), + us(tInitNanos), us(tReqSerNanos), us(tReqWriteNanos), us(tFirstFrameNanos), + us(tFinishedNanos), us(tRespDeserNanos), us(tAckNanos), tFrames, us(totalNanos)); + } + + private static long us(long nanos) { + return nanos < 0 ? nanos : nanos / 1000L; + } + private void maybeInit() throws InterruptedException, ServerInitializationException { boolean reconnect = false; @@ -375,7 +428,9 @@ public class PipesClient implements Closeable { throw new IOException("connection closed"); } LOG.debug("pipesClientId={}: sending NEW_REQUEST for id={}", pipesClientId, t.getId()); + long serStart = System.nanoTime(); byte[] bytes = JsonPipesIpc.toBytes(PipesRequest.of(t)); + tReqSerNanos = System.nanoTime() - serStart; // 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) { @@ -383,7 +438,9 @@ public class PipesClient implements Closeable { + " is " + bytes.length + " bytes, over maxIpcPayloadBytes=" + maxIpcPayloadBytes + "; raise maxIpcPayloadBytes or shrink the request"); } + long writeStart = System.nanoTime(); PipesMessage.newRequest(bytes).write(tuple.output); + tReqWriteNanos = System.nanoTime() - writeStart; } /** @@ -420,6 +477,9 @@ public class PipesClient implements Closeable { long clientBackstopMillis = clientBackstopMillis(limits); // nanoTime: the backstop must be immune to wall-clock steps long startNanos = System.nanoTime(); + // Separate anchor for the timing log: the backstop's is conceptually the deadline + // clock, and conflating them would silently break if either moves. + final long respAnchor = startNanos; while (true) { if (Thread.currentThread().isInterrupted()) { @@ -436,11 +496,17 @@ public class PipesClient implements Closeable { } try { PipesMessage msg = PipesMessage.read(tuple.input, maxIpcPayloadBytes); + long frameReadAt = System.nanoTime() - respAnchor; + if (++tFrames == 1) { + tFirstFrameNanos = frameReadAt; + } LOG.trace("clientId={}: received message type={} id={}", pipesClientId, msg.type(), t.getId()); // Send ACK only for messages that require it if (msg.type().requiresAck()) { + long ackStart = System.nanoTime(); PipesMessage.ack().write(tuple.output); + tAckNanos += System.nanoTime() - ackStart; } switch (msg.type()) { @@ -470,7 +536,10 @@ public class PipesClient implements Closeable { // is what keeps the blocking read below from hitting SO_TIMEOUT. break; case FINISHED: + tFinishedNanos = frameReadAt; + long deserStart = System.nanoTime(); PipesResult result = JsonPipesIpc.fromBytes(msg.payload(), PipesResult.class); + tRespDeserNanos = System.nanoTime() - deserStart; // Restore ParseContext from original FetchEmitTuple (not serialized back from server) if (result.emitData() instanceof EmitDataImpl emitDataImpl) { emitDataImpl.setParseContext(t.getParseContext()); diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java index a1755be2ab..a211829b15 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java @@ -82,6 +82,17 @@ public class ConnectionHandler implements Runnable, Closeable { private final long heartbeatIntervalMillis; private final ExecutorService executorService = Executors.newSingleThreadExecutor(); + /** Per-parse worker-side latency breakdown; joins the client line on {@code id}. */ + private static final Logger TIMING_LOG = + LoggerFactory.getLogger("org.apache.tika.pipes.timing.worker"); + + // Per-request timing scratch. One request at a time per connection. + private long tReqDeserNanos = -1; + private long tCtxMergeNanos = -1; + private long tSubmitAtNanos = -1; + private long tHandoffNanos = -1; + private PipesWorker tLastWorker; + private final ExecutorCompletionService<PipesResult> executorCompletionService = new ExecutorCompletionService<>(executorService); @@ -129,6 +140,42 @@ public class ConnectionHandler implements Runnable, Closeable { } } + private void resetTimings() { + tReqDeserNanos = -1; + tCtxMergeNanos = -1; + tSubmitAtNanos = -1; + tHandoffNanos = -1; + tLastWorker = null; + protocolIO.resetLastTimings(); + } + + /** + * One line per parse on {@code org.apache.tika.pipes.timing.worker}, microseconds. + * {@code handoff_us} is the executor scheduling plus completion-poll latency around the + * worker; {@code resp_*} is the FINISHED frame's serialize, socket write, and the + * client-ACK wait that follows it. + */ + private void logTiming(String id) { + if (!TIMING_LOG.isInfoEnabled()) { + return; + } + PipesWorker w = tLastWorker; + TIMING_LOG.info("WORKER_TIMING handler={} id={} req_deser_us={} ctx_merge_us={}" + + " handoff_us={} fetch_us={} parse_us={} emit_us={} worker_wall_us={}" + + " intermediate_us={} resp_ser_us={} resp_write_us={} resp_ack_us={}" + + " resp_bytes={}", + handlerId, id, us(tReqDeserNanos), us(tCtxMergeNanos), us(tHandoffNanos), + us(w == null ? -1 : w.getFetchNanos()), us(w == null ? -1 : w.getParseNanos()), + us(w == null ? -1 : w.getEmitNanos()), us(w == null ? -1 : w.getWallNanos()), + us(protocolIO.getLastIntermediateNanos()), us(protocolIO.getLastRespSerNanos()), + us(protocolIO.getLastRespWriteNanos()), us(protocolIO.getLastRespAckNanos()), + protocolIO.getLastRespBytes()); + } + + private static long us(long nanos) { + return nanos < 0 ? nanos : nanos / 1000L; + } + private void mainLoop() { ArrayBlockingQueue<Metadata> intermediateResult = new ArrayBlockingQueue<>(1); @@ -155,9 +202,12 @@ public class ConnectionHandler implements Runnable, Closeable { PipesRequest pipesRequest; FetchEmitTuple fetchEmitTuple; + resetTimings(); + long reqDeserStart = System.nanoTime(); try { pipesRequest = JsonPipesIpc.fromBytes(msg.payload(), PipesRequest.class); fetchEmitTuple = pipesRequest.getTuple(); + tReqDeserNanos = System.nanoTime() - reqDeserStart; } catch (IOException e) { LOG.error("handlerId={}: problem deserializing PipesRequest", handlerId, e); handleCrash(PipesMessageType.UNSPECIFIED_CRASH, "unknown", e); @@ -165,6 +215,7 @@ public class ConnectionHandler implements Runnable, Closeable { } ParseContext mergedContext = null; try { + long ctxStart = System.nanoTime(); mergedContext = resources.createMergedParseContext(fetchEmitTuple.getParseContext()); ParseContextUtils.resolveAll(mergedContext, getClass().getClassLoader()); ServerProtocolIO.validateParseContext(mergedContext); @@ -178,12 +229,16 @@ public class ConnectionHandler implements Runnable, Closeable { // ParseTimeout.getOrCreate(mergedContext) call (inside CompositeParser) // sees this instance rather than racing to install its own. ParseTimeout parseTimeout = ParseTimeout.getOrCreate(mergedContext); + tCtxMergeNanos = System.nanoTime() - ctxStart; PipesWorker pipesWorker = createPipesWorker(intermediateResult, fetchEmitTuple, mergedContext, countDownLatch); + tLastWorker = pipesWorker; + tSubmitAtNanos = System.nanoTime(); executorCompletionService.submit(pipesWorker); loopUntilDone(fetchEmitTuple, mergedContext, intermediateResult, countDownLatch, parseTimeout); + logTiming(fetchEmitTuple.getId()); } catch (TikaConfigException e) { LOG.error("handlerId={}: config error processing request", handlerId, e); handleCrash(PipesMessageType.UNSPECIFIED_CRASH, fetchEmitTuple.getId(), e); @@ -285,7 +340,9 @@ public class ConnectionHandler implements Runnable, Closeable { if (intermediate != null) { if (!clientGone) { try { - protocolIO.writeIntermediate(intermediate); + // Frame-flush releases the latch so the worker parses during the + // ACK round trip; the countDown below is the failure-path net. + protocolIO.writeIntermediate(intermediate, countDownLatch::countDown); } catch (IOException e) { clientGone = true; LOG.debug("handlerId={}: client gone (writing intermediate); keeping the " @@ -300,6 +357,8 @@ public class ConnectionHandler implements Runnable, Closeable { // Check for task completion Future<PipesResult> future = executorCompletionService.poll(100, TimeUnit.MILLISECONDS); if (future != null) { + tHandoffNanos = System.nanoTime() - tSubmitAtNanos + - (tLastWorker == null ? 0 : Math.max(0, tLastWorker.getWallNanos())); PipesResult pipesResult = null; try { pipesResult = future.get(); diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java index 173a7efc6e..dd545e38bb 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java @@ -191,6 +191,18 @@ public class PipesServer implements AutoCloseable { private final EmitStrategy emitStrategy; private final ServerProtocolIO protocolIO; + /** Per-parse worker-side latency breakdown; joins the client line on {@code id}. */ + private static final Logger TIMING_LOG = + LoggerFactory.getLogger("org.apache.tika.pipes.timing.worker"); + + // Per-request timing scratch. A per-client fork handles one request at a time. + private long tReqDeserNanos = -1; + private long tCtxMergeNanos = -1; + private long tSubmitAtNanos = -1; + private long tHandoffNanos = -1; + private long tIntermediateWaitNanos = -1; + private PipesWorker tLastWorker; + public static PipesServer load(int port, Path tikaConfigPath) throws Exception { String pipesClientId = System.getProperty("pipesClientId", "unknown"); LOG.debug("connecting to client on port={}", port); @@ -430,9 +442,12 @@ public class PipesServer implements AutoCloseable { PipesRequest pipesRequest; FetchEmitTuple fetchEmitTuple; + resetTimings(); + long reqDeserStart = System.nanoTime(); try { pipesRequest = JsonPipesIpc.fromBytes(msg.payload(), PipesRequest.class); fetchEmitTuple = pipesRequest.getTuple(); + tReqDeserNanos = System.nanoTime() - reqDeserStart; } catch (IOException e) { LOG.error("problem deserializing PipesRequest", e); handleCrash(PipesMessageType.UNSPECIFIED_CRASH, "unknown", e); @@ -440,6 +455,7 @@ public class PipesServer implements AutoCloseable { } ParseContext mergedContext; ParseTimeout parseTimeout; + long ctxStart = System.nanoTime(); try { mergedContext = createMergedParseContext(fetchEmitTuple.getParseContext()); ParseContextUtils.resolveAll(mergedContext, getClass().getClassLoader()); @@ -454,6 +470,7 @@ public class PipesServer implements AutoCloseable { // ParseTimeout.getOrCreate(mergedContext) call (inside CompositeParser) // sees this instance rather than racing to install its own. parseTimeout = ParseTimeout.getOrCreate(mergedContext); + tCtxMergeNanos = System.nanoTime() - ctxStart; } catch (Exception e) { // write the reason to the client instead of a bare exit code handleCrash(PipesMessageType.UNSPECIFIED_CRASH, fetchEmitTuple.getId(), e); @@ -461,9 +478,12 @@ public class PipesServer implements AutoCloseable { } PipesWorker pipesWorker = getPipesWorker(intermediateResult, fetchEmitTuple, mergedContext, countDownLatch); + tLastWorker = pipesWorker; + tSubmitAtNanos = System.nanoTime(); executorCompletionService.submit(pipesWorker); try { loopUntilDone(fetchEmitTuple, mergedContext, executorCompletionService, intermediateResult, countDownLatch, parseTimeout); + logTiming(fetchEmitTuple.getId()); } catch (Throwable t) { if (t instanceof Error) { // OOM or other JVM-level error: exit rather than continue in a @@ -522,6 +542,42 @@ public class PipesServer implements AutoCloseable { /** Steady-state slice once the intermediate result is out of the way. */ private static final long COMPLETION_POLL_MS = 100; + private void resetTimings() { + tReqDeserNanos = -1; + tCtxMergeNanos = -1; + tSubmitAtNanos = -1; + tHandoffNanos = -1; + tIntermediateWaitNanos = -1; + tLastWorker = null; + protocolIO.resetLastTimings(); + } + + /** + * One line per parse on {@code org.apache.tika.pipes.timing.worker}, microseconds. + * {@code handoff_us} is executor scheduling plus completion-poll latency around the worker; + * {@code resp_*} is the FINISHED frame's serialize, socket write, and the client-ACK wait. + */ + private void logTiming(String id) { + if (!TIMING_LOG.isInfoEnabled()) { + return; + } + PipesWorker w = tLastWorker; + TIMING_LOG.info("WORKER_TIMING id={} req_deser_us={} ctx_merge_us={} intermediate_wait_us={}" + + " handoff_us={} fetch_us={} parse_us={} emit_us={} worker_wall_us={}" + + " intermediate_us={} resp_ser_us={} resp_write_us={} resp_ack_us={}" + + " resp_bytes={}", + id, us(tReqDeserNanos), us(tCtxMergeNanos), us(tIntermediateWaitNanos), + us(tHandoffNanos), us(w == null ? -1 : w.getFetchNanos()), + us(w == null ? -1 : w.getParseNanos()), us(w == null ? -1 : w.getEmitNanos()), + us(w == null ? -1 : w.getWallNanos()), us(protocolIO.getLastIntermediateNanos()), + us(protocolIO.getLastRespSerNanos()), us(protocolIO.getLastRespWriteNanos()), + us(protocolIO.getLastRespAckNanos()), protocolIO.getLastRespBytes()); + } + + private static long us(long nanos) { + return nanos < 0 ? nanos : nanos / 1000L; + } + private void loopUntilDone(FetchEmitTuple fetchEmitTuple, ParseContext mergedContext, ExecutorCompletionService<PipesResult> executorCompletionService, ArrayBlockingQueue<Metadata> intermediateResult, CountDownLatch countDownLatch, @@ -539,7 +595,11 @@ public class PipesServer implements AutoCloseable { if (!wroteIntermediateResult) { Metadata intermediate = intermediateResult.poll(PRE_INTERMEDIATE_POLL_MS, TimeUnit.MILLISECONDS); if (intermediate != null) { - writeIntermediate(intermediate); + tIntermediateWaitNanos = System.nanoTime() - startNanos; + // The latch is released as soon as the frame is flushed, so the worker + // parses while the ACK is in flight; the extra countDown below is a + // no-op then, and the safety net when the write was skipped or failed. + writeIntermediate(intermediate, countDownLatch); countDownLatch.countDown(); wroteIntermediateResult = true; } @@ -551,6 +611,8 @@ public class PipesServer implements AutoCloseable { Future<PipesResult> future = executorCompletionService.poll( wroteIntermediateResult ? COMPLETION_POLL_MS : 0, TimeUnit.MILLISECONDS); if (future != null) { + tHandoffNanos = System.nanoTime() - tSubmitAtNanos + - (tLastWorker == null ? 0 : Math.max(0, tLastWorker.getWallNanos())); PipesResult pipesResult = null; try { pipesResult = future.get(); @@ -775,9 +837,9 @@ public class PipesServer implements AutoCloseable { } } - private void writeIntermediate(Metadata metadata) { + private void writeIntermediate(Metadata metadata, CountDownLatch latch) { try { - protocolIO.writeIntermediate(metadata); + protocolIO.writeIntermediate(metadata, latch::countDown); } catch (ShutDownReceivedException e) { handleShutDown(); } catch (IOException e) { 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 2f0b49eeec..8cc9962649 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 @@ -89,8 +89,40 @@ class PipesWorker implements Callable<PipesResult> { this.defaultParseMode = defaultParseMode; } + // Per-parse stage nanos, read by ConnectionHandler for the timing line after the + // worker completes. -1 means the stage never ran. + private volatile long fetchNanos = -1; + private volatile long parseNanos = -1; + private volatile long emitNanos = -1; + private volatile long wallNanos = -1; + + long getFetchNanos() { + return fetchNanos; + } + + long getParseNanos() { + return parseNanos; + } + + long getEmitNanos() { + return emitNanos; + } + + long getWallNanos() { + return wallNanos; + } + @Override public PipesResult call() throws Exception { + long wallStart = System.nanoTime(); + try { + return doCall(); + } finally { + wallNanos = System.nanoTime() - wallStart; + } + } + + private PipesResult doCall() throws Exception { MetadataListAndEmbeddedBytes parseData = null; TempFileUnpackHandler tempHandler = null; FrictionlessUnpackHandler frictionlessHandler = null; @@ -126,7 +158,12 @@ class PipesWorker implements Callable<PipesResult> { } } - return emitHandler.emitParseData(fetchEmitTuple, parseData, parseContext); + long emitStart = System.nanoTime(); + try { + return emitHandler.emitParseData(fetchEmitTuple, parseData, parseContext); + } finally { + emitNanos = System.nanoTime() - emitStart; + } } finally { // Clean up handlers if used if (frictionlessHandler != null) { @@ -490,13 +527,20 @@ 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(); + try { + return parseHandler.parseWithStream(fetchEmitTuple, tis, metadata, localContext); + } finally { + parseNanos = System.nanoTime() - parseStart; + } } catch (SecurityException e) { LOG.error("security exception id={}", fetchEmitTuple.getId(), e); throw e; diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ServerProtocolIO.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ServerProtocolIO.java index f8512bc6bf..1e8eb30ee6 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ServerProtocolIO.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ServerProtocolIO.java @@ -121,9 +121,12 @@ public class ServerProtocolIO { */ public void writeFinished(PipesResult pipesResult) throws IOException { BoundedOutputStream bos = new BoundedOutputStream(maxIpcPayloadBytes); + long serStart = System.nanoTime(); try { JsonPipesIpc.toStream(pipesResult, bos); + lastRespSerNanos = System.nanoTime() - serStart; } catch (IOException e) { + lastRespSerNanos = System.nanoTime() - serStart; if (!bos.overflowed()) { throw e; } @@ -153,8 +156,52 @@ public class ServerProtocolIO { doWritePayloadLimitExceeded(); return; } - PipesMessage.finished(bos.toByteArray()).write(output); + byte[] payload = bos.toByteArray(); + lastRespBytes = payload.length; + long writeStart = System.nanoTime(); + PipesMessage.finished(payload).write(output); + long ackStart = System.nanoTime(); + lastRespWriteNanos = ackStart - writeStart; awaitAck(); + lastRespAckNanos = System.nanoTime() - ackStart; + } + + // Last FINISHED-frame costs, read by ConnectionHandler for its per-parse timing line. + // One request at a time per connection, so plain fields suffice. + private long lastRespSerNanos = -1; + private long lastRespWriteNanos = -1; + private long lastRespAckNanos = -1; + private int lastRespBytes = -1; + + long getLastRespSerNanos() { + return lastRespSerNanos; + } + + long getLastRespWriteNanos() { + return lastRespWriteNanos; + } + + long getLastRespAckNanos() { + return lastRespAckNanos; + } + + int getLastRespBytes() { + return lastRespBytes; + } + + /** Write+ACK cost of the INTERMEDIATE_RESULT frame; -1 when none was sent. */ + private long lastIntermediateNanos = -1; + + long getLastIntermediateNanos() { + return lastIntermediateNanos; + } + + void resetLastTimings() { + lastRespSerNanos = -1; + lastRespWriteNanos = -1; + lastRespAckNanos = -1; + lastRespBytes = -1; + lastIntermediateNanos = -1; } /** @@ -184,6 +231,18 @@ public class ServerProtocolIO { * @throws IOException on serialization or I/O errors */ public void writeIntermediate(Metadata metadata) throws IOException { + writeIntermediate(metadata, null); + } + + /** + * Like {@link #writeIntermediate(Metadata)}, but runs {@code afterFrameWritten} once the + * frame is flushed to the socket, before waiting for the client's ACK. Lets the caller + * unblock the parse worker while the ACK is still in flight -- the ACK round trip would + * otherwise sit between pre-parse and parse on every request. Not invoked when the + * oversized intermediate is skipped or the write fails; callers must handle those paths + * themselves. + */ + public void writeIntermediate(Metadata metadata, Runnable afterFrameWritten) throws IOException { BoundedOutputStream bos = new BoundedOutputStream(maxIpcPayloadBytes); try { JsonPipesIpc.toStream(metadata, bos); @@ -195,7 +254,12 @@ public class ServerProtocolIO { } throw e; } + long interStart = System.nanoTime(); PipesMessage.intermediateResult(bos.toByteArray()).write(output); + if (afterFrameWritten != null) { + afterFrameWritten.run(); + } + lastIntermediateNanos = System.nanoTime() - interStart; awaitAck(); } diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java index 9c1c1765b8..b88cdc0bd9 100644 --- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java @@ -67,6 +67,9 @@ import org.apache.tika.server.core.TikaServerParseException; public class PipesParsingHelper { private static final Logger LOG = LoggerFactory.getLogger(PipesParsingHelper.class); + /** Per-request server-layer latency breakdown; joins the pipes lines on {@code id}. */ + private static final Logger TIMING_LOG = + LoggerFactory.getLogger("org.apache.tika.pipes.timing.server"); /** * The fetcher ID used for reading temp files. @@ -179,10 +182,15 @@ public class PipesParsingHelper { String requestId = UUID.randomUUID().toString(); PayloadRouter.Routed routed = null; String callerSuppliedName = metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY); + long entryNanos = System.nanoTime(); + long routeNanos = -1; + long pipesNanos = -1; + long postNanos = -1; try { routed = PayloadRouter.route(tis, maxInlineBytes, () -> Files.createTempFile(inputTempDirectory, "tika-", getSuffix(metadata))); + routeNanos = System.nanoTime() - entryNanos; String relativeName = null; FetchKey fetchKey; @@ -217,13 +225,19 @@ public class PipesParsingHelper { ); // Execute parse via pipes - results will be passed back through socket + long pipesStart = System.nanoTime(); PipesResult result = pipesParser.parse(tuple); + pipesNanos = System.nanoTime() - pipesStart; // Process result + long postStart = System.nanoTime(); List<Metadata> metadataList = processResult(result); if (relativeName != null) { stripSpoolIdentity(metadataList, relativeName, callerSuppliedName); } + postNanos = System.nanoTime() - postStart; + logTiming(requestId, routed.route().name(), routeNanos, pipesNanos, postNanos, + System.nanoTime() - entryNanos); return metadataList; } catch (InterruptedException e) { @@ -241,6 +255,25 @@ public class PipesParsingHelper { } } + /** + * One line per request on {@code org.apache.tika.pipes.timing.server}, microseconds. + * {@code route_us} covers reading the request body and deciding inline-vs-spool; + * {@code pipes_us} is the whole pipes round trip; {@code post_us} is result unpacking. + * The HTTP/JAX-RS layer outside this method is measured from the client. + */ + private static void logTiming(String id, String route, long routeNanos, long pipesNanos, + long postNanos, long totalNanos) { + if (!TIMING_LOG.isInfoEnabled()) { + return; + } + TIMING_LOG.info("SERVER_TIMING id={} route={} route_us={} pipes_us={} post_us={} total_us={}", + id, route, us(routeNanos), us(pipesNanos), us(postNanos), us(totalNanos)); + } + + private static long us(long nanos) { + return nanos < 0 ? nanos : nanos / 1000L; + } + /** Longest suffix carried over from a client filename; keeps well clear of NAME_MAX. */ private static final int MAX_SUFFIX_LENGTH = 20;
