gnodet-bot commented on code in PR #26548:
URL: https://github.com/apache/camel/pull/26548#discussion_r4036174181
##########
dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/LlmClient.java:
##########
@@ -146,7 +153,9 @@ public TokenUsage add(TokenUsage other) {
totalTokens + other.totalTokens,
cachedTokens + other.cachedTokens,
prefillMillis + other.prefillMillis,
- generationMillis + other.generationMillis);
+ generationMillis + other.generationMillis,
+ loadMillis + other.loadMillis,
+ totalMillis + other.totalMillis);
Review Comment:
📝 **Minor Javadoc nit** — the `@code` tag for `totalMillis` in the record
Javadoc says "the whole request" but would be clearer with a note that it
includes load + prefill + decode (i.e. it maps to Ollama's `total_duration`).
The preceding phase descriptions all say what they map to; `totalMillis` is the
only one that doesn't name its Ollama field. Suggest:
```suggestion
* ({@code loadMillis}, near zero for a warm model), and the whole
request wall time
* ({@code totalMillis}, Ollama's {@code total_duration}). Zero means
not reported.
```
##########
dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/OllamaMonitor.java:
##########
@@ -0,0 +1,1248 @@
+/*
+ * 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.camel.dsl.jbang.core.commands.tui;
+
+import java.io.InputStream;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.component.ai.observability.GenAiAttributes;
+import org.apache.camel.dsl.jbang.core.commands.LlmClient;
+import org.apache.camel.util.json.JsonArray;
+import org.apache.camel.util.json.JsonObject;
+import org.apache.camel.util.json.Jsoner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Collects what the Ollama tab shows: the Ollama server and its loaded models
over the REST API, the live state of the
+ * llama-server runner Ollama spawns on this machine, the load on the host,
and a log of requests with the timings
+ * Ollama reports per request.
+ * <p/>
+ * {@link #poll()} is called from the TUI's background refresh while the tab
is showing and throttles each source on its
+ * own interval. Requests arrive from two sides: the TUI's own AI panel calls
{@link #recordRequest} with the usage of
+ * each Ollama reply, and calls made by Camel routes come in through {@link
#ingestSpans} from the GenAI observability
+ * spans of the selected integration. The runner and host probes only run when
the Ollama host is this machine; a remote
+ * or containerised Ollama still gets the API and per-request data.
+ * <p/>
+ * All I/O happens outside the lock; the {@code update*} methods apply results
under it and are also what the tests
+ * feed.
+ */
+final class OllamaMonitor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(OllamaMonitor.class);
+
+ static final int MAX_REQUESTS = 200;
+ static final int HISTORY_POINTS = 120;
+
+ private static final long DETECT_INTERVAL_MS = 10_000;
+ private static final long VERSION_INTERVAL_MS = 30_000;
+ private static final long RECONNECT_INTERVAL_MS = 5_000;
+ private static final long PS_INTERVAL_MS = 1_000;
+ private static final long TAGS_INTERVAL_MS = 15_000;
+ private static final long RUNNER_SCAN_INTERVAL_MS = 5_000;
+ private static final long HOST_INTERVAL_MS = 1_000;
+ private static final long RATE_WINDOW_MS = 1_500;
+ private static final int MAX_SEEN_SPANS = 4_000;
+ private static final long SPAN_INTERVAL_MS = 5_000;
+
+ // ---- data ----
+
+ record ServerInfo(String baseUrl, String version, boolean local) {
+ }
+
+ record ModelShape(String architecture, int layers, int experts, int
expertsUsed, long maxContext,
+ int embeddingLength, long parameters, List<String> capabilities) {
+ }
+
+ record LoadedModel(String name, String family, String parameterSize,
String quantization, long sizeBytes,
+ long sizeVram, long contextLength, Instant expiresAt, ModelShape
shape) {
+
+ LoadedModel withShape(ModelShape newShape) {
+ return new LoadedModel(
+ name, family, parameterSize, quantization, sizeBytes,
sizeVram, contextLength,
+ expiresAt, newShape);
+ }
+
+ /** Share of the model held in GPU memory; 100 means fully offloaded.
*/
+ int gpuPercent() {
+ if (sizeBytes <= 0) {
+ return 0;
+ }
+ return (int) Math.min(100, sizeVram * 100 / sizeBytes);
+ }
+ }
+
+ /** Folded state of the runner's slots; {@code decoded} counts tokens of
the current or last request. */
+ record SlotState(boolean processing, long promptTokens, long
promptProcessed, long cacheTokens, long decoded,
+ long contextSize, String speculative, int slots, Instant
sampledAt) {
+
+ long contextUsed() {
+ return Math.max(promptTokens, cacheTokens) + decoded;
+ }
+
+ int cacheHitPercent() {
+ if (promptTokens <= 0) {
+ return 0;
+ }
+ return (int) Math.min(100, cacheTokens * 100 / promptTokens);
+ }
+ }
+
+ record RunnerInfo(long pid, int port, long contextSize, int parallel,
String modelPath, String executable) {
+ }
+
+ record GpuStats(String name, int utilizationPercent, long memoryUsedBytes,
long memoryTotalBytes, int count) {
+ }
+
+ record ProcessStats(long pid, String label, double cpuPercent, long
rssBytes) {
+ }
+
+ record HostStats(GpuStats gpu, ProcessStats server, ProcessStats runner,
Instant sampledAt) {
+ }
+
+ enum RequestSource {
+ TUI,
+ ROUTE
+ }
+
+ /**
+ * One request as Ollama reported it (TUI) or as the GenAI span of a route
recorded it (no phase timings).
+ * {@code contextSize} is the context window of the runner that served it,
0 when unknown.
+ */
+ record RequestEntry(Instant timestamp, RequestSource source, String
routeId, String model, int inputTokens,
+ int outputTokens, int cachedTokens, long prefillMs, long decodeMs,
long loadMs, long totalMs,
+ String doneReason, long contextSize, int question, String
questionText) {
+
+ RequestEntry(Instant timestamp, RequestSource source, String routeId,
String model, int inputTokens,
+ int outputTokens, int cachedTokens, long prefillMs, long
decodeMs, long loadMs, long totalMs,
+ String doneReason, long contextSize) {
+ this(timestamp, source, routeId, model, inputTokens, outputTokens,
cachedTokens, prefillMs, decodeMs,
+ loadMs, totalMs, doneReason, contextSize, 0, null);
+ }
+
+ /** Requests of the same AI panel question (its tool-call steps) share
this key; a route call stands alone. */
+ String groupKey() {
+ if (source == RequestSource.ROUTE) {
+ return "route:" + timestamp.toEpochMilli() + ":" + routeId;
+ }
+ return "q" + question + ":" + (questionText != null ? questionText
: "");
+ }
+
+ /**
+ * Everything the model had in front of it. Ollama's {@code
prompt_eval_count} is the whole prompt; the cached
+ * count is the part of it served from the KV cache, not an addition.
+ */
+ long promptTokens() {
+ return inputTokens;
+ }
+
+ /** Prompt tokens the runner actually had to evaluate this time. */
+ long evaluatedTokens() {
+ return Math.max(0, (long) inputTokens - cachedTokens);
+ }
+
+ /** Share of the prompt Ollama served from its cache. */
+ int cacheHitPercent() {
+ if (inputTokens <= 0) {
+ return 0;
+ }
+ return (int) Math.min(100, (long) cachedTokens * 100 /
inputTokens);
+ }
+
+ /** Share of the context window the prompt filled, or -1 when the
window is unknown. */
+ int contextPercent() {
+ if (contextSize <= 0) {
+ return -1;
+ }
+ return (int) Math.min(100, promptTokens() * 100 / contextSize);
+ }
+
+ /** Prefill speed over the tokens that were not served from cache. */
+ double prefillTokensPerSecond() {
+ return prefillMs > 0 && evaluatedTokens() > 0 ? evaluatedTokens()
* 1000.0 / prefillMs : 0;
+ }
+
+ double decodeTokensPerSecond() {
+ return decodeMs > 0 ? outputTokens * 1000.0 / decodeMs : 0;
+ }
+
+ /** Time to first token: loading the model plus processing the prompt.
Zero when timings are unknown. */
+ long ttftMs() {
+ return hasTimings() ? loadMs + prefillMs : 0;
+ }
+
+ boolean hasTimings() {
+ return prefillMs > 0 || decodeMs > 0;
+ }
+
+ /** A load of a second or more means the model was not in memory when
the request arrived. */
+ boolean coldStart() {
+ return loadMs >= 1000;
+ }
+ }
+
+ /**
+ * One AI panel question with all the requests it took (the model's tool
calls each cost a request), oldest step
+ * first, or a single route call. Aggregates are what the user experienced
for the whole question.
+ */
+ record QuestionGroup(String key, RequestSource source, int question,
String questionText, String routeId,
+ List<RequestEntry> steps) {
+
+ RequestEntry first() {
+ return steps.get(0);
+ }
+
+ RequestEntry last() {
+ return steps.get(steps.size() - 1);
+ }
+
+ /** The largest prompt of the question: how far the context was
pushed. */
+ long promptTokens() {
+ long max = 0;
+ for (RequestEntry e : steps) {
+ max = Math.max(max, e.promptTokens());
+ }
+ return max;
+ }
+
+ long outputTokens() {
+ long sum = 0;
+ for (RequestEntry e : steps) {
+ sum += e.outputTokens();
+ }
+ return sum;
+ }
+
+ int cacheHitPercent() {
+ long prompt = 0;
+ long cached = 0;
+ for (RequestEntry e : steps) {
+ prompt += e.inputTokens();
+ cached += e.cachedTokens();
+ }
+ return prompt > 0 ? (int) Math.min(100, cached * 100 / prompt) : 0;
+ }
+
+ int contextPercent() {
+ int max = -1;
+ for (RequestEntry e : steps) {
+ max = Math.max(max, e.contextPercent());
+ }
+ return max;
+ }
+
+ double prefillTokensPerSecond() {
+ long evaluated = 0;
+ long ms = 0;
+ for (RequestEntry e : steps) {
+ evaluated += e.evaluatedTokens();
+ ms += e.prefillMs();
+ }
+ return ms > 0 && evaluated > 0 ? evaluated * 1000.0 / ms : 0;
+ }
+
+ double decodeTokensPerSecond() {
+ long out = 0;
+ long ms = 0;
+ for (RequestEntry e : steps) {
+ out += e.outputTokens();
+ ms += e.decodeMs();
+ }
+ return ms > 0 && out > 0 ? out * 1000.0 / ms : 0;
+ }
+
+ /** Time to the first token of the first step: the wait before
anything happened. */
+ long ttftMs() {
+ return first().ttftMs();
+ }
+
+ boolean hasTimings() {
+ return first().hasTimings();
+ }
+
+ boolean coldStart() {
+ for (RequestEntry e : steps) {
+ if (e.coldStart()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** From the first request starting to the last one finishing. */
+ long wallMs() {
+ long start = first().timestamp().toEpochMilli();
+ long end = last().timestamp().toEpochMilli() + last().totalMs();
+ return Math.max(end - start, last().totalMs());
+ }
+
+ String doneReason() {
+ return last().doneReason();
+ }
+ }
+
+ /**
+ * Groups requests (newest first) into questions: consecutive AI panel
requests with the same question form one
+ * group with their steps oldest first; every route call is its own group.
Groups come back newest first.
+ */
+ static List<QuestionGroup> groupByQuestion(List<RequestEntry> newestFirst)
{
+ List<QuestionGroup> groups = new ArrayList<>();
+ int i = 0;
+ while (i < newestFirst.size()) {
+ RequestEntry head = newestFirst.get(i);
+ String key = head.groupKey();
+ List<RequestEntry> steps = new ArrayList<>();
+ int j = i;
+ while (j < newestFirst.size() &&
newestFirst.get(j).groupKey().equals(key)
+ && (head.source() == RequestSource.TUI || j == i)) {
+ steps.add(0, newestFirst.get(j));
+ j++;
+ }
+ groups.add(new QuestionGroup(
+ key, head.source(), head.question(), head.questionText(),
head.routeId(),
+ List.copyOf(steps)));
+ i = j;
+ }
+ return groups;
+ }
+
+ record SessionTotals(int requests, long inputTokens, long outputTokens,
long cachedTokens, long prefillMs,
+ long decodeMs, long loadMs, int coldStarts, int
peakContextPercent, int compactions) {
+
+ static final SessionTotals EMPTY = new SessionTotals(0, 0, 0, 0, 0, 0,
0, 0, 0, 0);
+
+ SessionTotals plus(RequestEntry e, boolean compaction) {
+ return new SessionTotals(
+ requests + 1, inputTokens + e.inputTokens(), outputTokens
+ e.outputTokens(),
+ cachedTokens + e.cachedTokens(), prefillMs +
e.prefillMs(), decodeMs + e.decodeMs(),
+ loadMs + e.loadMs(), coldStarts + (e.coldStart() ? 1 : 0),
+ Math.max(peakContextPercent, e.contextPercent()),
compactions + (compaction ? 1 : 0));
+ }
+
+ double avgDecodeTokensPerSecond() {
+ return decodeMs > 0 ? outputTokens * 1000.0 / decodeMs : 0;
+ }
+
+ double avgPrefillTokensPerSecond() {
+ long evaluated = Math.max(0, inputTokens - cachedTokens);
+ return prefillMs > 0 && evaluated > 0 ? evaluated * 1000.0 /
prefillMs : 0;
+ }
+ }
+
+ /** Immutable view for rendering and for the MCP tool. */
+ record Snapshot(ServerInfo server, List<LoadedModel> models, List<String>
installed, SlotState slot,
+ RunnerInfo runner, HostStats host, List<RequestEntry> requests,
double liveDecodeRate,
+ double livePrefillRate, long[] decodeHistory, SessionTotals
totals, String lastError, Instant lastPoll,
+ String probedUrl) {
+
+ boolean connected() {
+ return server != null;
+ }
+
+ boolean local() {
+ return server != null && server.local();
+ }
+
+ RequestEntry lastRequest() {
+ return requests.isEmpty() ? null : requests.get(0);
+ }
+ }
+
+ // ---- state ----
+
+ private final Object lock = new Object();
+ private final AtomicBoolean polling = new AtomicBoolean();
+ private final HttpClient http =
HttpClient.newBuilder().connectTimeout(Duration.ofMillis(1500)).build();
+
+ private volatile String baseUrl;
+ private ServerInfo server;
+ private List<LoadedModel> models = List.of();
+ private List<String> installed = List.of();
+ private final Map<String, ModelShape> shapes = new HashMap<>();
+ private final Set<String> shapeAttempted = new HashSet<>();
+ private SlotState slot;
+ private RunnerInfo runner;
+ private long serverPid;
+ private HostStats host;
+ private final Deque<RequestEntry> requests = new ArrayDeque<>();
+ private final LinkedHashSet<String> seenSpanIds = new LinkedHashSet<>();
+ private SessionTotals totals = SessionTotals.EMPTY;
+ private long lastTuiPromptTokens;
+ private final TokenRateWindow decodeWindow = new
TokenRateWindow(RATE_WINDOW_MS);
+ private final TokenRateWindow prefillWindow = new
TokenRateWindow(RATE_WINDOW_MS);
+ private final long[] decodeHistory = new long[HISTORY_POINTS];
+ private String lastError;
+ private Instant lastPoll;
+
+ private long lastProbe;
+ private long lastVersion;
+ private long lastPs;
+ private long lastTags;
+ private long lastRunnerScan;
+ private long lastHost;
+ private int psFailures;
+ private long lastSpanIngest;
+ private Boolean nvidiaSmiAvailable;
+ private final Map<Long, long[]> cpuSamples = new HashMap<>();
+
+ // ---- input from the rest of the TUI ----
+
+ /** Whether an Ollama server currently answers; the More menu lists the
tab only then. */
+ boolean isAvailable() {
+ synchronized (lock) {
+ return server != null;
+ }
+ }
+
+ /**
+ * Cheap background check while the tab is not showing, so the More menu
can list it as soon as Ollama comes up and
+ * drop it when Ollama goes away: endpoint detection and one version
request every ten seconds. Full polling happens
+ * in {@link #poll()} while the tab is active.
+ */
+ void probe() {
+ if (!polling.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ long now = System.currentTimeMillis();
+ if (now - lastProbe >= DETECT_INTERVAL_MS) {
+ lastProbe = now;
+ lastVersion = now;
+ checkServer();
+ }
+ } catch (Exception e) {
+ LOG.debug("Ollama probe failed", e);
+ } finally {
+ polling.set(false);
+ }
+ }
+
+ /** Uses the Ollama endpoint another part of the TUI already resolved (the
AI panel's client, an explicit URL). */
+ void adoptEndpoint(String url) {
+ if (url == null || url.isBlank()) {
+ return;
+ }
+ String normalized = url.endsWith("/") ? url.substring(0, url.length()
- 1) : url;
+ if (!normalized.equals(baseUrl)) {
+ baseUrl = normalized;
+ synchronized (lock) {
+ server = null;
+ lastVersion = 0;
+ runner = null;
+ slot = null;
+ }
+ }
+ }
+
+ /** As {@link #recordRequest(String, LlmClient.TokenUsage, long, String,
int, String)} without a question. */
+ void recordRequest(String model, LlmClient.TokenUsage usage, long
latencyMs, String doneReason) {
+ recordRequest(model, usage, latencyMs, doneReason, 0, null);
+ }
+
+ /**
+ * Records a request the TUI itself made to Ollama, with the usage and
timings Ollama returned. {@code question} and
+ * {@code questionText} identify the AI panel turn so the tool-call steps
of one question group together.
+ */
+ void recordRequest(
+ String model, LlmClient.TokenUsage usage, long latencyMs, String
doneReason, int question,
+ String questionText) {
+ if (usage == null) {
+ return;
+ }
+ long total = usage.totalMillis() > 0 ? usage.totalMillis() :
Math.max(0, latencyMs);
+ RequestEntry entry = new RequestEntry(
+ Instant.now(), RequestSource.TUI, null, model != null ? model
: "unknown",
+ usage.inputTokens(), usage.outputTokens(),
usage.cachedTokens(),
+ usage.prefillMillis(), usage.generationMillis(),
usage.loadMillis(), total, doneReason,
+ contextSizeForNewRequest(), question, questionText != null ?
questionText.strip() : null);
+ synchronized (lock) {
+ addRequest(entry);
+ }
+ }
+
+ /**
+ * The context window that served a request that just finished: the
runner's slot size when known, else the
+ * allocated context of the loaded model, fetched on the spot when the tab
has not polled yet. A request with a
+ * different {@code num_ctx} makes Ollama reload before answering, so what
is loaded after the reply is what served
+ * it.
+ */
+ private long contextSizeForNewRequest() {
+ long known = knownContextSize();
+ if (known > 0) {
+ return known;
+ }
+ String base = baseUrl;
+ if (base != null) {
+ JsonObject ps = getJsonObject(base + "/api/ps");
+ if (ps != null) {
+ List<LoadedModel> loaded = OllamaParsers.parsePs(ps);
+ if (!loaded.isEmpty()) {
+ synchronized (lock) {
+ List<LoadedModel> withShapes = new ArrayList<>();
+ for (LoadedModel m : loaded) {
+ withShapes.add(m.withShape(shapes.get(m.name())));
+ }
+ models = List.copyOf(withShapes);
+ }
+ return loaded.get(0).contextLength();
+ }
+ }
+ }
+ return 0;
+ }
+
+ private long knownContextSize() {
+ synchronized (lock) {
+ if (slot != null && slot.contextSize() > 0) {
+ return slot.contextSize();
+ }
+ return models.isEmpty() ? 0 : models.get(0).contextLength();
+ }
+ }
+
+ /**
+ * Adds the Ollama calls Camel routes made, from the GenAI observability
spans of the selected integration. Spans
+ * are deduplicated by id, so the same list can be handed over on every
refresh.
+ */
+ void ingestSpans(List<SpanEntry> spans) {
+ if (spans == null || spans.isEmpty()) {
+ return;
+ }
+ List<RequestEntry> fresh = new ArrayList<>();
+ synchronized (lock) {
+ for (SpanEntry span : spans) {
+ if (!GenAiSpanUsageExtractor.isGenAiSpan(span) ||
span.spanId() == null) {
+ continue;
+ }
+ Object system = span.attributes().get(GenAiAttributes.SYSTEM);
+ if (system == null ||
!system.toString().toLowerCase(Locale.ROOT).contains("ollama")) {
+ continue;
+ }
+ if (!seenSpanIds.add(span.spanId())) {
+ continue;
+ }
+ Map<String, Object> attrs = span.attributes();
+ String model = OllamaParsers.str(attrs,
GenAiAttributes.RESPONSE_MODEL);
+ if (model == null) {
+ model = OllamaParsers.str(attrs,
GenAiAttributes.REQUEST_MODEL);
+ }
+ if (model == null) {
+ model = span.name() != null ? span.name() : "unknown";
+ }
+ Instant ts = span.startEpochNanos() > 0 ?
Instant.ofEpochSecond(0, span.startEpochNanos()) : Instant.now();
+ fresh.add(new RequestEntry(
+ ts, RequestSource.ROUTE, span.routeId(), model,
+ (int) OllamaParsers.num(attrs,
GenAiAttributes.INPUT_TOKENS),
+ (int) OllamaParsers.num(attrs,
GenAiAttributes.OUTPUT_TOKENS),
+ 0, 0, 0, 0, Math.max(0, span.durationMs()),
+ OllamaParsers.str(attrs,
GenAiAttributes.FINISH_REASONS), knownContextSizeLocked()));
+ }
+ // spans arrive oldest first; keep the log newest first
+ fresh.sort((a, b) -> a.timestamp().compareTo(b.timestamp()));
+ for (RequestEntry e : fresh) {
+ addRequest(e);
+ }
+ while (seenSpanIds.size() > MAX_SEEN_SPANS) {
+ seenSpanIds.remove(seenSpanIds.iterator().next());
+ }
+ }
+ }
+
+ /** True once every few seconds: the GenAI spans of a route are worth
re-reading. */
+ boolean wantsSpans() {
+ long now = System.currentTimeMillis();
+ if (now - lastSpanIngest >= SPAN_INTERVAL_MS) {
+ lastSpanIngest = now;
+ return true;
+ }
+ return false;
+ }
+
+ /** Clears the request log, the session totals and the rate history. */
+ void reset() {
+ synchronized (lock) {
+ requests.clear();
+ totals = SessionTotals.EMPTY;
+ lastTuiPromptTokens = 0;
+ Arrays.fill(decodeHistory, 0);
+ decodeWindow.clear();
+ prefillWindow.clear();
+ }
+ }
+
+ /** As {@link #knownContextSize()} for callers already holding the lock. */
+ private long knownContextSizeLocked() {
+ if (slot != null && slot.contextSize() > 0) {
+ return slot.contextSize();
+ }
+ return models.isEmpty() ? 0 : models.get(0).contextLength();
+ }
+
+ private void addRequest(RequestEntry entry) {
+ // an AI panel prompt that shrinks by a fifth or more against the
previous turn means the history was
+ // compacted (or a new conversation started); either way the context
was freed
+ boolean compaction = false;
+ if (entry.source() == RequestSource.TUI) {
+ if (lastTuiPromptTokens > 0 && entry.promptTokens() <
lastTuiPromptTokens * 0.8) {
+ compaction = true;
+ }
+ lastTuiPromptTokens = entry.promptTokens();
+ }
+ requests.addFirst(entry);
+ while (requests.size() > MAX_REQUESTS) {
+ requests.removeLast();
+ }
+ totals = totals.plus(entry, compaction);
+ }
+
+ // ---- state updates (also the test seam) ----
+
+ void updateServer(ServerInfo info) {
+ synchronized (lock) {
+ server = info;
+ if (info != null) {
+ baseUrl = info.baseUrl();
+ lastError = null;
+ }
+ }
+ }
+
+ void updateModels(List<LoadedModel> loaded) {
+ synchronized (lock) {
+ models = loaded != null ? List.copyOf(loaded) : List.of();
+ }
+ }
+
+ void updateInstalled(List<String> names) {
+ synchronized (lock) {
+ installed = names != null ? List.copyOf(names) : List.of();
+ }
+ }
+
+ void updateRunner(RunnerInfo info) {
+ synchronized (lock) {
+ runner = info;
+ if (info == null) {
+ slot = null;
+ }
+ }
+ }
+
+ /** Applies a runner slot sample; the decode and prefill rate windows
advance from its counters. */
+ void updateSlot(SlotState state) {
+ synchronized (lock) {
+ slot = state;
+ if (state != null) {
+ long now = state.sampledAt() != null ?
state.sampledAt().toEpochMilli() : System.currentTimeMillis();
+ decodeWindow.sample(now, state.decoded());
+ prefillWindow.sample(now, state.promptProcessed());
+ }
+ }
+ }
+
+ void updateHost(HostStats stats) {
+ synchronized (lock) {
+ host = stats;
+ }
+ }
+
+ /** Ends one poll: appends the current live decode rate to the history and
stamps the poll time. */
+ void tick(long nowMillis) {
+ synchronized (lock) {
+ long rate = Math.round(decodeWindow.ratePerSecond(nowMillis));
+ System.arraycopy(decodeHistory, 1, decodeHistory, 0,
decodeHistory.length - 1);
+ decodeHistory[decodeHistory.length - 1] = rate;
+ lastPoll = Instant.ofEpochMilli(nowMillis);
+ }
+ }
+
+ Snapshot snapshot() {
+ synchronized (lock) {
+ long now = System.currentTimeMillis();
+ return new Snapshot(
+ server, models, installed, slot, runner, host,
List.copyOf(requests),
+ decodeWindow.ratePerSecond(now),
prefillWindow.ratePerSecond(now),
+ decodeHistory.clone(), totals, lastError, lastPoll,
baseUrl);
+ }
+ }
+
+ // ---- polling ----
+
+ /** One refresh cycle; safe to call every few hundred milliseconds, each
source keeps its own interval. */
+ void poll() {
+ if (!polling.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ doPoll(System.currentTimeMillis());
+ } catch (Exception e) {
+ LOG.debug("Ollama poll failed", e);
+ } finally {
+ polling.set(false);
+ }
+ }
+
+ private void doPoll(long now) {
+ boolean connected;
+ synchronized (lock) {
+ connected = server != null;
+ }
+ long versionInterval = connected ? VERSION_INTERVAL_MS :
RECONNECT_INTERVAL_MS;
+ if (now - lastVersion >= versionInterval) {
+ lastVersion = now;
+ lastProbe = now;
+ connected = checkServer();
+ }
+ if (!connected) {
+ tick(now);
+ return;
+ }
+ String base = baseUrl;
+
+ if (now - lastPs >= PS_INTERVAL_MS) {
+ lastPs = now;
+ JsonObject ps = getJsonObject(base + "/api/ps");
+ if (ps != null) {
+ psFailures = 0;
+ List<LoadedModel> loaded = new ArrayList<>();
+ for (LoadedModel m : OllamaParsers.parsePs(ps)) {
+ loaded.add(m.withShape(shapeFor(base, m.name())));
+ }
+ updateModels(loaded);
+ } else if (++psFailures >= 3) {
+ synchronized (lock) {
+ server = null;
+ lastError = "Ollama stopped answering at " +
OllamaParsers.displayHost(base);
+ }
+ lastVersion = 0;
+ tick(now);
+ return;
+ }
+ }
+
+ if (now - lastTags >= TAGS_INTERVAL_MS) {
+ lastTags = now;
+ JsonObject tags = getJsonObject(base + "/api/tags");
+ if (tags != null) {
+ List<String> names = new ArrayList<>();
+ Collection<?> list = tags.getCollection("models");
+ if (list != null) {
+ for (Object o : list) {
+ if (o instanceof Map<?, ?> m) {
+ String n = OllamaParsers.str(m, "name");
+ if (n != null) {
+ names.add(n);
+ }
+ }
+ }
+ }
+ updateInstalled(names);
+ }
+ }
+
+ if (OllamaParsers.isLoopbackUrl(base)) {
+ pollLocal(now);
+ }
+ tick(now);
+ }
+
+ private void pollLocal(long now) {
+ RunnerInfo current;
+ boolean haveModels;
+ synchronized (lock) {
+ current = runner;
+ haveModels = !models.isEmpty();
+ }
+ // scan often while a model is loaded but no runner is known yet,
rarely otherwise
+ long scanInterval = current == null && haveModels ? PS_INTERVAL_MS :
RUNNER_SCAN_INTERVAL_MS;
+ if (now - lastRunnerScan >= scanInterval) {
+ lastRunnerScan = now;
+ RunnerInfo found = findRunner();
+ if (found == null || current == null || found.pid() !=
current.pid() || found.port() != current.port()) {
+ updateRunner(found);
+ current = found;
+ }
+ }
+ if (current != null) {
+ JsonArray slots = getJsonArray("http://127.0.0.1:"; +
current.port() + "/slots");
+ if (slots != null) {
+ updateSlot(OllamaParsers.parseSlots(slots,
Instant.ofEpochMilli(now)));
+ } else {
+ updateRunner(null);
+ current = null;
+ }
+ }
+ if (now - lastHost >= HOST_INTERVAL_MS) {
+ lastHost = now;
+ updateHost(probeHost(now, current));
+ }
+ }
+
+ private ModelShape shapeFor(String base, String name) {
+ synchronized (lock) {
+ ModelShape cached = shapes.get(name);
+ if (cached != null || shapeAttempted.contains(name)) {
+ return cached;
+ }
+ shapeAttempted.add(name);
+ }
+ JsonObject body = new JsonObject();
+ body.put("model", name);
+ JsonObject show = postJsonObject(base + "/api/show", body.toJson());
+ ModelShape shape = OllamaParsers.parseShow(show);
+ if (shape != null) {
+ synchronized (lock) {
+ shapes.put(name, shape);
+ }
+ }
+ return shape;
+ }
+
+ /** Detects the endpoint when none is known yet and confirms the server
answers; true when connected. */
+ private boolean checkServer() {
+ String base = baseUrl;
+ if (base == null) {
+ base = detectEndpoint();
+ if (base == null) {
+ return false;
+ }
+ adoptEndpoint(base);
+ base = baseUrl;
+ }
+ JsonObject version = getJsonObject(base + "/api/version");
+ if (version == null) {
+ synchronized (lock) {
+ server = null;
+ models = List.of();
+ slot = null;
+ lastError = "Ollama not reachable at " +
OllamaParsers.displayHost(base);
+ }
+ return false;
+ }
+ updateServer(new ServerInfo(base, OllamaParsers.str(version,
"version"), OllamaParsers.isLoopbackUrl(base)));
+ return true;
+ }
+
+ private String detectEndpoint() {
+ try {
+ LlmClient client =
LlmClient.create().withApiType(LlmClient.ApiType.ollama);
+ if (client.detectEndpoint()) {
+ return client.endpointUrl();
+ }
+ } catch (Exception e) {
+ LOG.debug("Ollama endpoint detection failed", e);
+ }
+ return null;
+ }
+
+ // ---- runner and host probes ----
+
+ private RunnerInfo findRunner() {
+ try {
+ RunnerInfo[] found = new RunnerInfo[1];
+ long[] parent = new long[1];
+ ProcessHandle.allProcesses().forEach(ph -> {
+ if (found[0] != null) {
+ return;
+ }
Review Comment:
⚠️ **Partial update outside lock — transient inconsistency window**
`baseUrl` is written as a `volatile` field *before* acquiring the lock that
resets `server`, `runner`, and `slot`. Between the volatile write and the lock
acquisition, another thread calling `doPoll()` can read the **new** `baseUrl`
(line `String base = baseUrl`) while `synchronized(lock)` still sees `server !=
null` — i.e. the old server is flagged as connected but requests are now
directed to the new URL. The next `checkServer()` call corrects this, so the
window is transient, but it means one poll cycle can make HTTP calls to the new
URL while internally believing the old server is alive (e.g. skipping the
version check that would trigger reconnect).
Fix — do both writes under the same lock:
```suggestion
String normalized = url.endsWith("/") ? url.substring(0,
url.length() - 1) : url;
synchronized (lock) {
if (!normalized.equals(baseUrl)) {
baseUrl = normalized;
server = null;
lastVersion = 0;
runner = null;
slot = null;
}
}
```
With this change `baseUrl` no longer needs to be `volatile` for the
endpoint-switch path (it is still read unsynchronized from
`contextSizeForNewRequest` and `doPoll`, but those reads are already racy with
the present code and are benign for the same reason — `checkServer` corrects
any stale view within one poll).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
