GameCult/Mimir
Watch
9
0
Fork
You've already forked Mimir
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Expose Well sender pressure

Browse source
This commit is contained in:
Meta Vens 2026-06-03 05:25:50 +01:00
commit 4c75abcada
3 changed files with 74 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

51
src/Mimir.Well/Program.cs
View file

@ -141,7 +141,7 @@ while (!stopping.IsCancellationRequested)
featureSignalFrame,
exposureController.Statuses,
sourceErrors,
streamTelemetry.Snapshot(now, subscribersAreExternal: true),
streamTelemetry.Snapshot(now, subscribersAreExternal: true, publisher.SenderPressureSnapshot()),
++sequence,
startedAt);
streamTelemetry.ObservePublish(
@ -681,7 +681,7 @@ internal sealed class MimirWellStreamTelemetry
lastPublishMilliseconds = receipt.TotalMilliseconds;
}
public object Snapshot(DateTimeOffset now, bool subscribersAreExternal) => new
public object Snapshot(DateTimeOffset now, bool subscribersAreExternal, object senderPressure) => new
{
document = "mimir.well_stream_pressure.v1",
status = "transitional-websocket-json",
@ -709,6 +709,7 @@ internal sealed class MimirWellStreamTelemetry
lastMilliseconds = lastPublishMilliseconds,
subscribersAreExternal,
},
sender = senderPressure,
nextOrgan = new
{
controlLane = "CultMesh typed state and compact stream cursors",
@ -834,8 +835,16 @@ internal sealed class MimirWellPublisher(Uri url) : IAsyncDisposable
private readonly ConcurrentQueue<QueuedPublish> bodyQueue = new();
private readonly SemaphoreSlim signal = new(0);
private readonly CancellationTokenSource senderStopping = new();
private readonly object sendTelemetryLock = new();
private Task? senderTask;
private int queueDepth;
private long sentDocuments;
private long sentBytes;
private double totalSendMilliseconds;
private double maxSendMilliseconds;
private string lastSentDocument = "";
private int lastSentBytes;
private double lastSendMilliseconds;
private const int MaxQueuedDocuments = 256;
private const int StreamFrameDropThreshold = 128;
@ -909,7 +918,10 @@ internal sealed class MimirWellPublisher(Uri url) : IAsyncDisposable
Interlocked.Decrement(ref queueDepth);
try
{
var stopwatch = Stopwatch.StartNew();
await socket.SendAsync(item.Bytes, WebSocketMessageType.Text, endOfMessage: true, stopping).ConfigureAwait(false);
stopwatch.Stop();
ObserveSend(item, stopwatch.Elapsed);
}
catch (OperationCanceledException)
{
@ -927,6 +939,41 @@ internal sealed class MimirWellPublisher(Uri url) : IAsyncDisposable
private bool TryDequeueNext(out QueuedPublish item) =>
controlQueue.TryDequeue(out item!) || bodyQueue.TryDequeue(out item!);
public object SenderPressureSnapshot()
{
lock (sendTelemetryLock)
{
return new
{
document = "mimir.well_sender_pressure.v1",
authority = "Mimir.Well websocket sender loop observes wire send pressure after enqueue.",
sentDocuments,
sentBytes,
averageMilliseconds = sentDocuments == 0 ? 0.0 : totalSendMilliseconds / sentDocuments,
maxMilliseconds = maxSendMilliseconds,
lastDocument = lastSentDocument,
lastBytes = lastSentBytes,
lastMilliseconds = lastSendMilliseconds,
currentQueueDepth = Volatile.Read(ref queueDepth),
};
}
}
private void ObserveSend(QueuedPublish item, TimeSpan elapsed)
{
var milliseconds = elapsed.TotalMilliseconds;
lock (sendTelemetryLock)
{
sentDocuments++;
sentBytes += item.Bytes.Length;
totalSendMilliseconds += milliseconds;
maxSendMilliseconds = Math.Max(maxSendMilliseconds, milliseconds);
lastSentDocument = item.Document;
lastSentBytes = item.Bytes.Length;
lastSendMilliseconds = milliseconds;
}
}
public async ValueTask DisposeAsync()
{
senderStopping.Cancel();

1
state/evidence.jsonl
View file

@ -372,3 +372,4 @@
{"ts": "2026-06-03T05:10:12+01:00", "type": "observation", "status": "ok", "note": "Kiyo Pro source-health diagnosis: after stopping Well, direct native KS smoke through mimir_camera_capture.dll produced 0 frames for Kiyo Pro across 640x480 YUY2, 640x480 MJPG, and 1920x1080 MJPG, while Kiyo basic produced 107 frames in 3s through the same DLL. MF/GPU fallback probe hung under the live stack. Added source factory diagnostics to Well snapshots so Odin sees adapter/path/mode and starved KS-camera reason; live restart showed Kiyo Pro waiting-device-samples with reason 'KS camera opened but no frames...' for 640x480 YUY2, while Leap and Kiyo basic stayed live and Well held 6/6 Ready at 40 ms."}
{"ts": "2026-06-03T05:13:42+01:00", "type": "implementation", "status": "ok", "note": "Presentation availability cut: MimirPresentationControlState now keeps user Enabled/Solo intent separate from derived Available/SampleCount. IncludesVideo requires live samples, so configured-but-starved video sources no longer occupy program output. Live Well verification after restart showed Kiyo Pro, Raven display, and Eve camera available=false/included=false, while Leap and Kiyo basic were available=true/included=true and synchronized frame stayed 6/6 Ready at 40 ms."}
{"ts": "2026-06-03T05:20:09+01:00", "type": "implementation", "status": "ok", "note": "Mimir Well shakedown now defaults to the latest monotonic Well run instead of aggregating restarted ledgers; verified 20260602-233914 latest run at 1609 snapshots, 6/6 Ready after 120s and after 300s, recent presentation delay 40 ms. Remaining unavailable sources are source-health/device/network issues, not synchronized-frame collapse."}
{"ts": "2026-06-03T05:25:36+01:00", "type": "implementation", "status": "ok", "note": "Well stream pressure now includes Mimir.Well sender-loop telemetry at the websocket send authority. Live restart of 20260602-233914 reached after-120 readiness 1.0 with 6/6 Ready, recent presentation delay 40 ms, sender queue depth 0, sender max about 10 ms, and no drops; historical enqueue max is not active wire pressure."}

26
tools/mimir_well_shakedown.py
View file

@ -176,6 +176,7 @@ def summarize_well(path: Path, all_runs: bool = False) -> dict[str, Any]:
pressure = last.get("streamPressure") if isinstance(last.get("streamPressure"), dict) else {}
publish = pressure.get("publish") if isinstance(pressure.get("publish"), dict) else {}
poll = pressure.get("poll") if isinstance(pressure.get("poll"), dict) else {}
sender = pressure.get("sender") if isinstance(pressure.get("sender"), dict) else {}
visual = last.get("visualCalibration", {}).get("cameras", [])
return {
@ -224,6 +225,16 @@ def summarize_well(path: Path, all_runs: bool = False) -> dict[str, Any]:
"lastDocument": publish.get("lastDocument", ""),
"lastBytes": int(number(publish.get("lastBytes"))),
},
"senderPressure": {
"sentDocuments": int(number(sender.get("sentDocuments"))),
"sentBytes": int(number(sender.get("sentBytes"))),
"averageMilliseconds": number(sender.get("averageMilliseconds")),
"maxMilliseconds": number(sender.get("maxMilliseconds")),
"lastDocument": sender.get("lastDocument", ""),
"lastBytes": int(number(sender.get("lastBytes"))),
"lastMilliseconds": number(sender.get("lastMilliseconds")),
"currentQueueDepth": int(number(sender.get("currentQueueDepth"))),
},
"captureBodies": {
"medianSampleCount": statistics.median(capture_sample_counts) if capture_sample_counts else 0.0,
"bodyStatuses": dict(body_statuses),
@ -425,16 +436,24 @@ def hypotheses(summary: dict[str, Any]) -> list[str]:
readiness = summary["readiness"]
overlap = summary["overlapAfter120Ms"]
pressure = summary["streamPressure"]
sender = summary.get("senderPressure", {})
after300_full_frame_rate = readiness.get("after300FullFrameRate")
if isinstance(after300_full_frame_rate, (int, float)) and after300_full_frame_rate >= 0.95:
result.append(
"After warm-up, frame readiness is high; fixed 2500 ms delay should be reduced by an adaptive controller until misses reappear."
)
if overlap["p10MinOverlap"] <= 25.0:
if readiness.get("after120Samples", 0) > 0 and overlap["p10MinOverlap"] <= 25.0:
result.append(
"ASIO blocks expose a very thin per-domain overlap; audio should be aligned by Faust delay/resampling, not by asking the presentation planner for seconds of holdback."
)
if pressure["publishMaxMs"] > 50.0 or pressure["lastBytes"] > 1_000_000:
sender_queue_depth = sender.get("currentQueueDepth") if isinstance(sender, dict) else None
sender_max_ms = sender.get("maxMilliseconds") if isinstance(sender, dict) else None
if (
pressure["publishMaxMs"] > 50.0
and (not isinstance(sender_queue_depth, (int, float)) or sender_queue_depth > 0)
) or (
isinstance(sender_max_ms, (int, float)) and sender_max_ms > 50.0
) or pressure["lastBytes"] > 1_000_000:
result.append(
"Capture-page publication creates burst pressure; body paging/stream frames need their own compact lane before latency experiments can trust publish timing."
)
@ -483,6 +502,9 @@ def main() -> int:
print("Stream pressure:")
for key, value in summary["streamPressure"].items():
print(f" {key}: {value}")
print("Sender pressure:")
for key, value in summary["senderPressure"].items():
print(f" {key}: {value}")
if move:
print(f"Move score: bpm={move['bpm']:.1f} c={move['bpmConfidence']:.2f} key={move['key']} chord={move['chord']} voices={move['activeVoiceCount']}/{move['voiceCount']}")
print(f" tail score confidence median/max: {move['scoreConfidenceMedian']:.3f}/{move['scoreConfidenceMax']:.3f}")