MoShiro/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs

using System.Net;
using Fleet.Core;
using MoShiro.Protocol;

namespace MoShiro.Measurement;

/// <summary>积木的公共部分：环境、参数、把行切成列块交出去。</summary>
public abstract class BlockBase : IBlock
{
    protected BlockBase(BlockEnv env, BlockSpec spec)
    {
        Env = env;
        Spec = spec;
    }

    public virtual long Produced => 0;

    protected BlockEnv Env { get; }

    protected BlockSpec Spec { get; }

    public abstract Task RunAsync(CancellationToken ct);

    public virtual int Flush(FlushMode mode, long peerProduced = 0) => 0;

    public virtual RowStore? Records(string streamName) => null;

    protected IPEndPoint PeerEndpoint(int port) => new(Env.PeerAddress!, port);

    /// <summary>交出一个流的行；起点和时长取自时间列 <paramref name="timeColumn"/> 的首尾（本侧的单调时钟，微秒）。</summary>
    protected void Ship(RawStream stream, string timeColumn, IReadOnlyList<RawStream>? extra = null, IReadOnlyDictionary<string, System.Text.Json.JsonElement>? parameters = null)
    {
        RawColumn? time = stream.Find(timeColumn);
        long first = stream.Count > 0 && time is not null ? time.Values[0] : Env.Clock.NowUs();
        long last = stream.Count > 0 && time is not null ? time.Values[stream.Count - 1] : first;
        List<RawStream> streams = [stream];
        if (extra is not null)
        {
            streams.AddRange(extra);
        }

        Env.Fragments!.Ship(Spec.KindId, Spec.Scope, streams, parameters ?? Spec.Params.Values, Env.Clock.UtcMsAt(first), (int)((last - first) / 1000), Spec.Port);
    }

    /// <summary>
    /// 按自适应的节奏发包：下一个包的时刻按绝对时间排（不累积误差），每秒评估一次状态，状态变化记进日志（供审计）。
    /// 落后了就从现在重新排，不补发。
    /// </summary>
    protected async Task PaceAsync(AdaptiveRate rate, Func<ValueTask> sendOne, CancellationToken ct)
    {
        long due = Env.Clock.NowUs();
        long lastEvaluation = due;
        while (!ct.IsCancellationRequested)
        {
            due += (long)(rate.NextDelay().TotalMilliseconds * 1000);
            long wait = due - Env.Clock.NowUs();
            if (wait > 0)
            {
                await Task.Delay(TimeSpan.FromMicroseconds(wait), Env.Time, ct);
            }
            else
            {
                due = Env.Clock.NowUs();
            }

            await sendOne();
            long now = Env.Clock.NowUs();
            if (now - lastEvaluation >= 1_000_000 || rate.State == ProbeState.Dead)
            {
                lastEvaluation = now;
                if (rate.Evaluate(now) is { } change)
                {
                    LogTransition(change);
                }
            }
        }
    }

    /// <summary>读这个任务的自适应频率参数（<c>udp_echo</c> 和 <c>tcp_echo</c> 共用）。</summary>
    protected AdaptiveRate? CreateAdaptiveRate()
    {
        if (!Spec.Bool("adaptive"))
        {
            return null;
        }

        double max = Math.Max(Spec.Int("rate_max_hz"), Spec.Int("rate_min_hz"));
        return new AdaptiveRate(new AdaptiveRateOptions
        {
            MinHz = Spec.Int("rate_min_hz"),
            MaxHz = max,
            TauPct = Spec.Number("tau_pct"),
            RttTolPct = Spec.Number("rtt_tol_pct"),
            Prior = Env.LossPrior,
        });
    }

    private void LogTransition(ProbeTransition change)
    {
        var fields = new (string, object?)[]
        {
            ("stream", Spec.StreamId), ("kind", Spec.KindId), ("from", change.From.ToString().ToLowerInvariant()), ("to", change.To.ToString().ToLowerInvariant()), ("hz", Math.Round(change.Hz, 1)),
            ("reason", change.Reason), ("loss_pct", Math.Round(change.LossPct, 3)), ("sent", change.Sent), ("lost", change.Lost),
        };
        if (change.To == ProbeState.Dead)
        {
            Env.Log.Warn("probe rate changed", fields);
        }
        else
        {
            Env.Log.Info("probe rate changed", fields);
        }
    }

    /// <summary>什么也不做、一直等到取消（监听方里不需要自己运行逻辑的积木）。</summary>
    protected async Task WaitUntilCancelledAsync(CancellationToken ct)
    {
        try
        {
            await Task.Delay(Timeout.InfiniteTimeSpan, Env.Time, ct);
        }
        catch (OperationCanceledException)
        {
            // 被停止
        }
    }
}

/// <summary>
/// 拨号方作为行的主导方时记录的行：行按键（序号）追加，回复到达时按键回填列，到点把已经稳定的行从游标处切下来交出去。
/// 环形缓冲被覆盖的行（交出太慢）直接跳过。
/// </summary>
internal sealed class DriverLog
{
    private readonly object _gate = new();
    private long _cursor;

    public DriverLog(StreamSpec spec, Side driver, int capacity)
    {
        Store = new RowStore(spec, driver, capacity);
        Spec = spec;
    }

    public RowStore Store { get; }

    public StreamSpec Spec { get; }

    public long Count => Store.NextKey;

    /// <summary>取出游标之后已经稳定的行：<paramref name="settled"/> 为 null 或 <paramref name="all"/> 时取全部。</summary>
    public RawStream? Take(Func<long, bool>? settled, bool all, int maxRows = int.MaxValue)
    {
        long from;
        long to;
        lock (_gate)
        {
            from = _cursor;
            to = from;
            long end = Store.NextKey;
            while (to < end && to - from < maxRows && (all || settled is null || !Store.Contains(to) || settled(to)))
            {
                to++;
            }

            if (to == from)
            {
                return null;
            }

            _cursor = to;
        }

        RawStream stream = Store.Cut(from, to, includeKey: Spec.Key is not null);
        return stream.Count == 0 ? null : stream;
    }
}