Fix Well clock freshness and reservoir intake
This commit is contained in:
parent
7ab555a355
commit
f5193341e9
8 changed files with 155 additions and 21 deletions
|
|
@ -333,7 +333,7 @@ if (-not $SkipRuntime) {
|
|||
"--publish-ms", "250",
|
||||
"--sync-ms", "250",
|
||||
"--presentation-delay-ms", "2500",
|
||||
"--max-samples-per-source", "4",
|
||||
"--max-samples-per-source", "32",
|
||||
"--visual-calibration", "true",
|
||||
"--visual-calibration-ms", "250",
|
||||
"--visual-expected-leds", "38",
|
||||
|
|
|
|||
|
|
@ -1142,6 +1142,14 @@ internal sealed class MimirLiveStatsProvider : IDashboardProvider
|
|||
UiElement.Text("mimir-well-features", $"video features stable {well.FeatureStableTracks} c={well.FeatureMeanConfidence:0.000} motion {well.FeatureMeanMotionPixelsPerSecond:0.0}px/s faust signals {well.FeatureSignals.Count}", "caption"),
|
||||
UiElement.Text("mimir-well-probe", well.Probe == null ? "probe: none" : $"probe emit={well.Probe.ShouldEmit} sync={well.Probe.SyncConfidence:0.000} freq={well.Probe.FrequencyConfidence:0.000} {well.Probe.Reason}", "caption"),
|
||||
};
|
||||
children.AddRange(well.SourceHealth
|
||||
.OrderBy(static source => source.Kind, StringComparer.Ordinal)
|
||||
.ThenBy(static source => source.SourceId, StringComparer.Ordinal)
|
||||
.Take(10)
|
||||
.Select(source => UiElement.Text(
|
||||
$"mimir-well-source-health-{StableId(source.SourceId)}",
|
||||
$"{source.SourceId}: {source.Kind} {source.Status} samples {source.Samples} age {source.LatestAgeMs:0}ms {source.Reason}",
|
||||
source.Status == "live" ? "caption" : "danger")));
|
||||
children.AddRange(well.FeatureSignals
|
||||
.OrderByDescending(static signal => signal.Confidence)
|
||||
.ThenBy(static signal => signal.SourceId, StringComparer.Ordinal)
|
||||
|
|
@ -1473,6 +1481,19 @@ internal sealed class MimirLiveStatsSnapshot
|
|||
var captureStorage = TryGetAny(TryGetAny(root, "capture"), "storage");
|
||||
var publication = TryGetAny(root, "publication");
|
||||
var stems = ArrayItems(TryGetAny(publication, "stems")).ToArray();
|
||||
var sourceHealth = ArrayItems(TryGetAny(root, "sourceHealth"))
|
||||
.Select(static source => new MimirWellSourceHealthStat(
|
||||
JsonStringAny(source, "SourceId", "sourceId") ?? "unknown",
|
||||
JsonStringAny(source, "Kind", "kind") ?? "unknown",
|
||||
JsonStringAny(source, "Origin", "origin") ?? "unknown",
|
||||
JsonStringAny(source, "Label", "label") ?? "",
|
||||
JsonStringAny(source, "status") ?? "unknown",
|
||||
JsonStringAny(source, "reason") ?? "",
|
||||
(int)JsonNumberAny(source, "buffers"),
|
||||
(int)JsonNumberAny(source, "samples"),
|
||||
JsonNumberAny(source, "latestAgeMs"),
|
||||
(int)JsonNumberAny(source, "latestByteLength")))
|
||||
.ToArray();
|
||||
return new MimirWellStat(
|
||||
(long)JsonNumberAny(root, "sequence"),
|
||||
JsonNumberAny(root, "elapsedSeconds"),
|
||||
|
|
@ -1500,6 +1521,7 @@ internal sealed class MimirLiveStatsSnapshot
|
|||
(long)JsonNumberAny(publishPressure, "bytes"),
|
||||
JsonStringAny(captureStorage, "bodyTransport") ?? "unknown",
|
||||
stems.Length,
|
||||
sourceHealth,
|
||||
buffers,
|
||||
clockDomains,
|
||||
states,
|
||||
|
|
@ -2090,6 +2112,7 @@ internal sealed record MimirWellStat(
|
|||
long PublishedBytes,
|
||||
string CaptureInlineBodies,
|
||||
int PublicationStemCount,
|
||||
IReadOnlyList<MimirWellSourceHealthStat> SourceHealth,
|
||||
IReadOnlyList<MimirWellBufferStat> Buffers,
|
||||
IReadOnlyList<MimirWellClockDomainStat> ClockDomains,
|
||||
IReadOnlyList<MimirWellAudioSyncStateStat> AudioSyncStates,
|
||||
|
|
@ -2121,6 +2144,18 @@ internal sealed record MimirWellBufferStat(
|
|||
int SampleRate,
|
||||
int Channels);
|
||||
|
||||
internal sealed record MimirWellSourceHealthStat(
|
||||
string SourceId,
|
||||
string Kind,
|
||||
string Origin,
|
||||
string Label,
|
||||
string Status,
|
||||
string Reason,
|
||||
int Buffers,
|
||||
int Samples,
|
||||
double LatestAgeMs,
|
||||
int LatestByteLength);
|
||||
|
||||
internal sealed record MimirWellAudioSyncStateStat(string SourceId, string ReferenceSourceId, double DelaySamples, double SroPpm, double Confidence);
|
||||
|
||||
internal sealed record MimirWellClockDomainStat(string ClockDomainId, int SourceCount, double OverlapMs, bool HasLocalOverlap, bool IsReference, double OffsetMs);
|
||||
|
|
|
|||
|
|
@ -237,8 +237,8 @@ internal sealed class FensalirDaemonState : IAsyncDisposable
|
|||
var cache = await CultCacheMessagePack.OpenAsync(options.CultCachePath, new CultCacheOpenOptions
|
||||
{
|
||||
UseDirectoryStore = true,
|
||||
FlushOnDispose = true,
|
||||
StoreFlushOnDispose = true,
|
||||
FlushOnDispose = false,
|
||||
StoreFlushOnDispose = false,
|
||||
}).ConfigureAwait(false);
|
||||
var key = new CultRecordKey(options.DaemonId);
|
||||
var existing = cache.Get<MimirFensalirDaemonStateDocument>(key);
|
||||
|
|
@ -252,7 +252,7 @@ internal sealed class FensalirDaemonState : IAsyncDisposable
|
|||
var existingPressure = cache.Get<MimirFensalirReservoirPressureDocument>(pressureKey);
|
||||
var pressureDocument = existingPressure ?? BuildInitialPressureDocument(options);
|
||||
var pressureHandle = await cache.UpsertAsync(pressureDocument, new CultRecordHandle<MimirFensalirReservoirPressureDocument>(pressureKey)).ConfigureAwait(false);
|
||||
await cache.FlushAsync().ConfigureAwait(false);
|
||||
await SafeFlushAsync(cache, soft: false).ConfigureAwait(false);
|
||||
return new FensalirDaemonState(options, cache, handle, document, workerHandle, workerDocument, pressureHandle, pressureDocument);
|
||||
}
|
||||
|
||||
|
|
@ -292,7 +292,7 @@ internal sealed class FensalirDaemonState : IAsyncDisposable
|
|||
await cache.UpsertAsync(document, handle).ConfigureAwait(false);
|
||||
await cache.UpsertAsync(workerDocument, workerHandle).ConfigureAwait(false);
|
||||
await cache.UpsertAsync(pressureDocument, pressureHandle).ConfigureAwait(false);
|
||||
await cache.FlushAsync(soft: true).ConfigureAwait(false);
|
||||
await SafeFlushAsync(cache, soft: true).ConfigureAwait(false);
|
||||
if (options.Once)
|
||||
{
|
||||
return;
|
||||
|
|
@ -320,10 +320,30 @@ internal sealed class FensalirDaemonState : IAsyncDisposable
|
|||
|
||||
public ValueTask DisposeAsync()
|
||||
{
|
||||
cache.Dispose();
|
||||
try
|
||||
{
|
||||
cache.Dispose();
|
||||
}
|
||||
catch (IOException ex)
|
||||
{
|
||||
Console.Error.WriteLine($"mimir-fensalir-daemon state-cache-dispose-warning {ex.Message}");
|
||||
}
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
private static async Task SafeFlushAsync(CultCache cache, bool soft)
|
||||
{
|
||||
try
|
||||
{
|
||||
await cache.FlushAsync(soft).ConfigureAwait(false);
|
||||
}
|
||||
catch (IOException ex)
|
||||
{
|
||||
Console.Error.WriteLine($"mimir-fensalir-daemon state-cache-flush-warning soft={soft} {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
private bool TryDrink(JsonElement root, out MimirFensalirDaemonStateDocument next)
|
||||
{
|
||||
var kind = DaemonUtil.JsonText(root, "document") ?? "";
|
||||
|
|
@ -595,7 +615,7 @@ internal sealed class FensalirDaemonState : IAsyncDisposable
|
|||
var distanceNs = Math.Abs((long)DaemonUtil.JsonNumber(slice, "distanceFromPresentationNs", "DistanceFromPresentationNs"));
|
||||
var startNs = (long)DaemonUtil.JsonNumber(slice, "canonicalStartNs", "CanonicalStartNs");
|
||||
var endNs = (long)DaemonUtil.JsonNumber(slice, "canonicalEndNs", "CanonicalEndNs");
|
||||
if (string.Equals(status, "Ready", StringComparison.OrdinalIgnoreCase) && confidence >= 0.20)
|
||||
if (string.Equals(status, "Ready", StringComparison.OrdinalIgnoreCase) && (confidence >= 0.20 || distanceNs == 0))
|
||||
{
|
||||
accepted++;
|
||||
minConfidence = Math.Min(minConfidence, confidence);
|
||||
|
|
@ -664,7 +684,8 @@ internal sealed class FensalirDaemonState : IAsyncDisposable
|
|||
var kind = DaemonUtil.JsonText(stream, "kind") ?? DaemonUtil.JsonText(sample, "Kind", "kind") ?? "";
|
||||
var status = DaemonUtil.JsonText(slice, "status", "Status") ?? "";
|
||||
var confidence = DaemonUtil.JsonNumber(slice, "timingConfidence", "TimingConfidence");
|
||||
var accepted = string.Equals(status, "Ready", StringComparison.OrdinalIgnoreCase) && confidence >= 0.20 ? 1 : 0;
|
||||
var distanceNs = Math.Abs((long)DaemonUtil.JsonNumber(slice, "distanceFromPresentationNs", "DistanceFromPresentationNs"));
|
||||
var accepted = string.Equals(status, "Ready", StringComparison.OrdinalIgnoreCase) && (confidence >= 0.20 || distanceNs == 0) ? 1 : 0;
|
||||
var reasons = new SortedSet<string>(StringComparer.Ordinal);
|
||||
if (accepted == 0)
|
||||
{
|
||||
|
|
@ -681,7 +702,7 @@ internal sealed class FensalirDaemonState : IAsyncDisposable
|
|||
var droppedBecauseBodyMissing = 0;
|
||||
var droppedBecauseFenceUnavailable = 0;
|
||||
var droppedBecauseTimingDegraded = 0;
|
||||
if (confidence < 0.20)
|
||||
if (accepted == 0 && confidence < 0.20)
|
||||
{
|
||||
droppedBecauseTimingDegraded++;
|
||||
}
|
||||
|
|
@ -694,7 +715,11 @@ internal sealed class FensalirDaemonState : IAsyncDisposable
|
|||
droppedBecauseBodyMissing++;
|
||||
}
|
||||
|
||||
var hasInlineBody = string.Equals(transport, "inline-bytes", StringComparison.OrdinalIgnoreCase) ||
|
||||
string.Equals(bodyStatus, "inline-bytes", StringComparison.OrdinalIgnoreCase);
|
||||
if (string.Equals(kind, "Video", StringComparison.OrdinalIgnoreCase) &&
|
||||
!hasInlineBody &&
|
||||
!hasBodyRef &&
|
||||
DaemonUtil.JsonNumber(frame, "producerFenceValue", "ProducerFenceValue") <= 0.0 &&
|
||||
string.IsNullOrWhiteSpace(DaemonUtil.JsonText(frame, "nativeHandleKind", "NativeHandleKind")))
|
||||
{
|
||||
|
|
@ -712,7 +737,7 @@ internal sealed class FensalirDaemonState : IAsyncDisposable
|
|||
(long)DaemonUtil.JsonNumber(slice, "canonicalStartNs", "CanonicalStartNs"),
|
||||
(long)DaemonUtil.JsonNumber(slice, "canonicalEndNs", "CanonicalEndNs"),
|
||||
confidence,
|
||||
Math.Abs((long)DaemonUtil.JsonNumber(slice, "distanceFromPresentationNs", "DistanceFromPresentationNs")),
|
||||
distanceNs,
|
||||
0.0,
|
||||
current.PollAverageMs,
|
||||
current.PublishAverageMs,
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ public sealed record MimirAsioStreamSourceOptions(
|
|||
public sealed class MimirAsioStreamSource : IMimirMultiplexedStreamSource
|
||||
{
|
||||
private const string DefaultClsid = "{AC4D0455-50D7-4498-B3CD-9A41D130B759}";
|
||||
private const int MaxQueuedBlocks = 32768;
|
||||
private const int MaxQueuedBlocks = 256;
|
||||
private readonly ConcurrentQueue<MimirStreamSample> samples = new();
|
||||
private readonly ManualResetEventSlim started = new();
|
||||
private readonly Thread captureThread;
|
||||
|
|
@ -130,13 +130,14 @@ public sealed class MimirAsioStreamSource : IMimirMultiplexedStreamSource
|
|||
|
||||
var sourceId = channel < sourceIds.Length ? sourceIds[channel] : $"asio-ch{channel}";
|
||||
var bytes = new byte[frameCount * sizeof(float)];
|
||||
var arrivalNs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L;
|
||||
MemoryMarshal.AsBytes(sampleBuffer.AsSpan(0, frameCount)).CopyTo(bytes);
|
||||
samples.Enqueue(new MimirStreamSample(
|
||||
sourceId,
|
||||
MimirStreamKind.Audio,
|
||||
Descriptor.Origin,
|
||||
timestampNs,
|
||||
DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L,
|
||||
arrivalNs,
|
||||
arrivalNs,
|
||||
sequence,
|
||||
0,
|
||||
bytes.Length,
|
||||
|
|
|
|||
|
|
@ -46,7 +46,7 @@ public sealed class MimirNativeIngestStreamSource : IMimirStreamSource
|
|||
}
|
||||
|
||||
Push(
|
||||
frame.DeviceTimestampNs,
|
||||
arrivalNs,
|
||||
arrivalNs,
|
||||
frame.NativeHandle,
|
||||
data.Length,
|
||||
|
|
|
|||
|
|
@ -172,7 +172,7 @@ public sealed class MimirRuntimeConfiguration
|
|||
stream.MinimumFramesPerSecond,
|
||||
stream.QueueDepth,
|
||||
ParsePixelFormat(stream.PixelFormat))),
|
||||
static () => StopwatchTicksToNs(System.Diagnostics.Stopwatch.GetTimestamp())));
|
||||
UtcNowNs));
|
||||
}
|
||||
|
||||
if (string.Equals(stream.Adapter, "ps3eye", StringComparison.OrdinalIgnoreCase)
|
||||
|
|
@ -192,7 +192,7 @@ public sealed class MimirRuntimeConfiguration
|
|||
stream.Width,
|
||||
stream.Height,
|
||||
stream.FramesPerSecond)),
|
||||
static () => StopwatchTicksToNs(System.Diagnostics.Stopwatch.GetTimestamp())));
|
||||
UtcNowNs));
|
||||
}
|
||||
|
||||
if (string.Equals(stream.Adapter, "mf-gpu", StringComparison.OrdinalIgnoreCase)
|
||||
|
|
@ -214,7 +214,7 @@ public sealed class MimirRuntimeConfiguration
|
|||
stream.InputFormat,
|
||||
stream.OutputFormat,
|
||||
stream.MinimumFramesPerSecond)),
|
||||
static () => StopwatchTicksToNs(System.Diagnostics.Stopwatch.GetTimestamp())));
|
||||
UtcNowNs));
|
||||
}
|
||||
|
||||
if (string.Equals(stream.Adapter, "frame-events", StringComparison.OrdinalIgnoreCase)
|
||||
|
|
@ -305,10 +305,7 @@ public sealed class MimirRuntimeConfiguration
|
|||
return Path.GetFullPath(Path.Combine(configDirectory, command));
|
||||
}
|
||||
|
||||
private static long StopwatchTicksToNs(long ticks)
|
||||
{
|
||||
return checked((long)(ticks * (1_000_000_000.0 / System.Diagnostics.Stopwatch.Frequency)));
|
||||
}
|
||||
private static long UtcNowNs() => DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L;
|
||||
|
||||
private static MimirStreamKind ParseKind(string value)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -1341,6 +1341,9 @@ internal static class MimirWellSnapshot
|
|||
configuredSources = runtimeConfig.SourceFactories.Count,
|
||||
liveSources = hub.SourceCount,
|
||||
sourceErrors,
|
||||
sourceHealth = runtimeConfig.SourceFactories
|
||||
.Select(factory => SourceHealthSnapshot(factory.Descriptor, hub.Buffers.Buffers, sourceErrors, startedAt))
|
||||
.ToArray(),
|
||||
buffers = hub.Buffers.Buffers.Select(BufferSnapshot).ToArray(),
|
||||
synchronizedFrame = new
|
||||
{
|
||||
|
|
@ -1519,6 +1522,78 @@ internal static class MimirWellSnapshot
|
|||
};
|
||||
}
|
||||
|
||||
private static object SourceHealthSnapshot(
|
||||
MimirStreamDescriptor descriptor,
|
||||
IEnumerable<MimirRollingStreamBuffer> buffers,
|
||||
IReadOnlyList<object> sourceErrors,
|
||||
DateTimeOffset startedAt)
|
||||
{
|
||||
var matchingBuffers = buffers
|
||||
.Where(buffer => string.Equals(buffer.Descriptor.SourceId, descriptor.SourceId, StringComparison.Ordinal))
|
||||
.ToArray();
|
||||
if (matchingBuffers.Length == 0 && descriptor.SourceId.Contains("asio", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
matchingBuffers = buffers
|
||||
.Where(buffer =>
|
||||
buffer.Descriptor.Kind == descriptor.Kind &&
|
||||
buffer.Descriptor.Origin == descriptor.Origin &&
|
||||
buffer.Descriptor.SourceId.StartsWith("asio-ch", StringComparison.OrdinalIgnoreCase))
|
||||
.ToArray();
|
||||
}
|
||||
|
||||
var samples = matchingBuffers.Sum(static buffer => buffer.Count);
|
||||
var latest = matchingBuffers
|
||||
.Select(static buffer => buffer.Latest)
|
||||
.Where(static sample => sample.HasValue)
|
||||
.OrderByDescending(static sample => sample!.Value.TimestampNs)
|
||||
.FirstOrDefault();
|
||||
var errored = sourceErrors.Any(error =>
|
||||
TryReadAnonymousString(error, "SourceId", "sourceId") is { } id &&
|
||||
string.Equals(id, descriptor.SourceId, StringComparison.Ordinal));
|
||||
var elapsedMs = Math.Max(0.0, (DateTimeOffset.UtcNow - startedAt).TotalMilliseconds);
|
||||
var status = errored
|
||||
? "create-error"
|
||||
: samples > 0
|
||||
? "live"
|
||||
: "created-no-samples";
|
||||
|
||||
return new
|
||||
{
|
||||
descriptor.SourceId,
|
||||
Kind = descriptor.Kind.ToString(),
|
||||
Origin = descriptor.Origin.ToString(),
|
||||
descriptor.Label,
|
||||
status,
|
||||
reason = status switch
|
||||
{
|
||||
"create-error" => "source creation failed; see sourceErrors",
|
||||
"created-no-samples" => "source was created but has not emitted a sample",
|
||||
_ => "samples arriving",
|
||||
},
|
||||
buffers = matchingBuffers.Length,
|
||||
samples,
|
||||
latestTimestampNs = latest?.TimestampNs ?? 0L,
|
||||
latestAgeMs = latest.HasValue
|
||||
? Math.Round(Math.Max(0.0, (DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000.0 - latest.Value.TimestampNs) / 1_000_000.0), 1)
|
||||
: Math.Round(elapsedMs, 1),
|
||||
latestByteLength = latest?.ByteLength ?? 0,
|
||||
};
|
||||
}
|
||||
|
||||
private static string? TryReadAnonymousString(object value, params string[] names)
|
||||
{
|
||||
var type = value.GetType();
|
||||
foreach (var name in names)
|
||||
{
|
||||
if (type.GetProperty(name)?.GetValue(value) is string text && !string.IsNullOrWhiteSpace(text))
|
||||
{
|
||||
return text;
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private static object BufferSnapshot(MimirRollingStreamBuffer buffer) => new
|
||||
{
|
||||
buffer.Descriptor.SourceId,
|
||||
|
|
|
|||
|
|
@ -365,3 +365,4 @@
|
|||
{"ts": "2026-06-02T18:17:56+01:00", "type": "implementation", "status": "ok", "note": "2026-06-02: Fensalir daemon now consumes mimir.cultmesh_stream_frame.v1 as a fine-grained reservoir job lane in addition to well_capture_page batches. Timed frames with usable body/native/shared transport enqueue jobs; metadata-only or low-confidence frames drop with reasons. Isolated build passed and a stream-frame fixture on port 8802 completed one worker job with droppedJobs=0."}
|
||||
{"ts": "2026-06-02T18:23:46+01:00", "type": "implementation", "status": "ok", "note": "2026-06-02: Added mimir.fensalir_reservoir_pressure.v1. Fensalir daemon now persists pressure cause metrics alongside worker state: Well poll/publish ms, capture bytes, ready-slice ratio, timing confidence/distance, GPU queue ms, and missing-body/fence-unavailable/timing-degraded drop counters. Fixture on port 8803 completed one stream-frame job and health advertised worker + pressure documents."}
|
||||
{"ts": "2026-06-02T19:17:54+01:00", "type": "decision", "status": "ok", "note": "Online Verse shakedown: canonical timestamps are owned by MimirCanonicalClockMapper, which now rebases streams whose projected source clock drifts more than 10s from arrival after ASIO/device resets; MimirSynchronizationHub no longer reapplies raw clock-domain corrections to already-canonical buffers. Fensalir Eve surface now mirrors sibling daemon PID liveness so Odin can lower relay, recorder, Well, Move sync, and Fensalir statuses without taking control authority."}
|
||||
{"ts": "2026-06-02T23:49:45+01:00", "type": "decision", "status": "ok", "note": "Mimir Well/Fensalir live repair: rolling-buffer authority now uses canonical arrival UTC timestamps for ASIO/native camera samples while preserving device timestamps as calibration evidence; Fensalir CultCache persistence is best-effort so a Windows File.Replace sharing failure cannot kill the live Eve daemon."}
|
||||
|
|
|
|||
Loading…
Reference in a new issue