From ba578b6b509a246b6e9a67fa5a8d4a8b1eeccd28 Mon Sep 17 00:00:00 2001 From: Myoungho Shin Date: Tue, 23 Jun 2026 18:40:47 -0700 Subject: [PATCH] fix: agent exits promptly after a short trace drains --- .../java/com/gpuflight/agent/GpuflAgent.java | 20 ++++---- .../java/com/gpuflight/agent/LogTailer.java | 16 ++----- .../agent/service/SessionWatcher.java | 7 +-- .../agent/service/TailerManager.java | 19 ++++---- .../java/com/gpuflight/agent/util/Delays.java | 48 +++++++++++++++++++ 5 files changed, 75 insertions(+), 35 deletions(-) create mode 100644 src/main/java/com/gpuflight/agent/util/Delays.java diff --git a/src/main/java/com/gpuflight/agent/GpuflAgent.java b/src/main/java/com/gpuflight/agent/GpuflAgent.java index 4869e51..f1af78a 100644 --- a/src/main/java/com/gpuflight/agent/GpuflAgent.java +++ b/src/main/java/com/gpuflight/agent/GpuflAgent.java @@ -12,6 +12,7 @@ import com.gpuflight.agent.publisher.PublisherFactory; import com.gpuflight.agent.service.SessionWatcher; import com.gpuflight.agent.service.TailerManager; +import com.gpuflight.agent.util.Delays; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -25,7 +26,6 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.atomic.AtomicInteger; public class GpuflAgent { @@ -114,7 +114,7 @@ public void start() throws Exception { if (exitWhenDrained) { log.info("exit-when-drained mode enabled"); - awaitDrainThenExit(watchedFolders.keySet(), tailerManager.getActiveTailers()); + awaitDrainThenExit(watchedFolders.keySet(), tailerManager); log.info("all sessions drained - exiting"); shutdown(); return; @@ -154,16 +154,16 @@ private void shutdown() { try { if (publisher != null) publisher.close(); } catch (Exception ignored) {} } - void awaitDrainThenExit(Collection folders, AtomicInteger activeTailers) - throws InterruptedException { - boolean sawActive = false; + void awaitDrainThenExit(Collection folders, TailerManager tailers) { int clean = 0; while (true) { - Thread.sleep(1000); - boolean activeNow = anyActiveSession(folders); - if (activeNow) sawActive = true; - boolean drained = sawActive && activeTailers.get() == 0 && !activeNow; - clean = drained ? clean + 1 : 0; + if (!Delays.sleep(Delays.DRAIN_CHECK_POLL)) break; + // Gate on "a session was discovered" (cumulative), not on observing the live + // .tmp/ marker: a short trace finalizes that marker between our 1s polls, which + // left the old sawActive gate spinning forever even after the upload drained. + boolean started = tailers.hasStartedAnySession(); + boolean idle = tailers.getActiveTailers().get() == 0 && !anyActiveSession(folders); + clean = (started && idle) ? clean + 1 : 0; if (clean >= 2) return; } } diff --git a/src/main/java/com/gpuflight/agent/LogTailer.java b/src/main/java/com/gpuflight/agent/LogTailer.java index d5d0007..2d1431a 100644 --- a/src/main/java/com/gpuflight/agent/LogTailer.java +++ b/src/main/java/com/gpuflight/agent/LogTailer.java @@ -6,7 +6,7 @@ import com.gpuflight.agent.filter.DeviceMetricDeduplicator; import com.gpuflight.agent.model.LogWrapper; import com.gpuflight.agent.publisher.Publisher; - +import com.gpuflight.agent.util.Delays; import java.io.BufferedInputStream; import java.io.ByteArrayOutputStream; import java.io.File; @@ -236,21 +236,21 @@ public void tail(Publisher publisher) { } else { // Publish failed mid-window; the cursor already holds the resume offset. offset = resume; - sleep(5000); + if (!Delays.sleep(Delays.LOG_TAILER_RETRY)) break; } continue; } // Window not published yet. Is the session still writing? if (sessionTmpDir().exists()) { - sleep(2000); + if (!Delays.sleep(Delays.LOG_TAILER_POLL)) break; continue; } // .tmp/ gone -> the client closed every channel. Each channel's last // window is published BEFORE .tmp/ is removed, so wait a moment then do // one final check for a straggler before finishing. - sleep(4500); + if (!Delays.sleep(Delays.SESSION_END_GRACE_PERIOD)) break; if (resolveRotated(idx) != null) continue; System.out.println("[" + logType + "] Session finished (.tmp gone, no window " + idx + "; last sent " + (idx - 1) + ")."); @@ -479,12 +479,4 @@ private static long skipFully(InputStream in, long n) throws IOException { } return total; } - - private static void sleep(long ms) { - try { - Thread.sleep(ms); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } } diff --git a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java index ac75e9e..f308d11 100644 --- a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java +++ b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java @@ -2,7 +2,7 @@ import com.gpuflight.agent.config.ConfigLoader; import com.gpuflight.agent.model.DiscoveredSession; - +import com.gpuflight.agent.util.Delays; import java.io.File; import java.util.ArrayList; import java.util.Arrays; @@ -43,10 +43,7 @@ public void start(ExecutorService executor) { for (DiscoveredSession s : discoverSources(folder, logTypes)) { onSessionDiscovered.accept(s); } - Thread.sleep(2_000); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - break; + if (!Delays.sleep(Delays.SESSION_WATCHER_POLL)) break; } catch (Exception e) { System.err.println("[watcher] error scanning " + folder + ": " + e.getMessage()); diff --git a/src/main/java/com/gpuflight/agent/service/TailerManager.java b/src/main/java/com/gpuflight/agent/service/TailerManager.java index 251fd2a..d279f37 100644 --- a/src/main/java/com/gpuflight/agent/service/TailerManager.java +++ b/src/main/java/com/gpuflight/agent/service/TailerManager.java @@ -6,10 +6,12 @@ import com.gpuflight.agent.filter.DeviceMetricDeduplicator; import com.gpuflight.agent.model.DiscoveredSession; import com.gpuflight.agent.publisher.Publisher; +import com.gpuflight.agent.util.Delays; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.nio.file.Path; +import java.time.Duration; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; @@ -77,19 +79,15 @@ public void spawnSessionTailers(DiscoveredSession session) { } private void signalSessionComplete(Publisher publisher, String sessionId) { + long delayMs = Delays.SESSION_COMPLETE_RETRY.toMillis(); for (int attempt = 1; attempt <= 3; attempt++) { try { if (publisher.publishSessionComplete(sessionId)) return; } catch (Exception e) { - System.err.println("[agent] session-complete signal error for " - + sessionId + ": " + e.getMessage()); - } - try { - Thread.sleep(2000L); - } catch (InterruptedException ie) { - Thread.currentThread().interrupt(); - return; + log.error("session-complete signal error for {}: {}", sessionId, e.getMessage()); } + if (!Delays.sleep(Duration.ofMillis(delayMs))) return; + delayMs *= 2; // Exponential backoff } log.warn("session-complete signal gave up for {} - backend grace finalize will apply", sessionId); } @@ -97,4 +95,9 @@ private void signalSessionComplete(Publisher publisher, String sessionId) { public AtomicInteger getActiveTailers() { return activeTailers; } + + /** True once any session has been discovered and tailed. Cumulative - never cleared. */ + public boolean hasStartedAnySession() { + return !startedSessions.isEmpty(); + } } diff --git a/src/main/java/com/gpuflight/agent/util/Delays.java b/src/main/java/com/gpuflight/agent/util/Delays.java new file mode 100644 index 0000000..5856651 --- /dev/null +++ b/src/main/java/com/gpuflight/agent/util/Delays.java @@ -0,0 +1,48 @@ +package com.gpuflight.agent.util; + +import java.time.Duration; + +/** + * Centralized operational delays and sleep utilities. + * Using named constants improves readability and makes the agent's + * timing behavior easier to tune. + */ +public final class Delays { + + private Delays() {} + + /** Polling interval for discovering new sessions in watched folders. */ + public static final Duration SESSION_WATCHER_POLL = Duration.ofSeconds(2); + + /** Polling interval for checking if all sessions have drained before exiting. */ + public static final Duration DRAIN_CHECK_POLL = Duration.ofSeconds(1); + + /** Initial delay between retries when signaling session completion to the backend. */ + public static final Duration SESSION_COMPLETE_RETRY = Duration.ofSeconds(2); + + /** Delay before retrying a failed window upload. */ + public static final Duration LOG_TAILER_RETRY = Duration.ofSeconds(5); + + /** Polling interval when waiting for a new window to be published in an active session. */ + public static final Duration LOG_TAILER_POLL = Duration.ofSeconds(2); + + /** + * Grace period after a session's .tmp directory is gone before the final check for + * straggler windows. This ensures any late flushes are noticed. + */ + public static final Duration SESSION_END_GRACE_PERIOD = Duration.ofMillis(4500); + + /** + * Sleep for the specified duration. + * @return true if the sleep finished normally, false if it was interrupted. + */ + public static boolean sleep(Duration duration) { + try { + Thread.sleep(duration.toMillis()); + return true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } +}