acdream/src/AcDream.App/Streaming/LandblockStreamer.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using AcDream.Core.World;

namespace AcDream.App.Streaming;

/// <summary>
/// Services landblock load/unload requests by invoking a caller-supplied
/// load delegate (the production instance wraps
/// <see cref="LandblockLoader.Load"/>) and posting results to an outbox
/// the render thread drains once per OnUpdate.
///
/// <para>
/// <b>Currently runs synchronously on the calling thread.</b> The original
/// Phase A.1 design ran loads on a dedicated worker thread, but DatReaderWriter's
/// <c>DatCollection</c> is not thread-safe — concurrent reads from a worker
/// and the render thread (animation tick, live spawn handlers) corrupt
/// internal buffer state and produce half-populated <c>LandBlock.Height[]</c>
/// arrays which render as wildly distorted terrain. Until Phase A.3 introduces
/// a thread-safe dat wrapper, loads are synchronous: <see cref="EnqueueLoad"/>
/// invokes the load delegate inline and writes the result to the outbox in
/// a single call. This causes a frame hitch when crossing landblock
/// boundaries, but the rendering is correct.
/// </para>
///
/// <para>
/// The Channel-based outbox + <see cref="DrainCompletions"/> API is
/// preserved so the move back to async loading is a single-class change
/// when DatCollection thread safety lands.
/// </para>
///
/// <para>
/// Unloads pass through the outbox as <see cref="LandblockStreamResult.Unloaded"/>
/// records so the render thread can release GPU state on the next drain —
/// the streamer never touches GPU resources directly.
/// </para>
///
/// <remarks>
/// Threading: synchronous mode means all methods must be called from the
/// same thread (the render thread in production).
/// </remarks>
/// </summary>
public sealed class LandblockStreamer : IDisposable
{
    /// <summary>
    /// Default drain batch size. Tuned to cap GPU upload work the render
    /// thread does per frame while still draining a moderate backlog in a
    /// few frames. Callers can override on a per-call basis.
    /// </summary>
    public const int DefaultDrainBatchSize = 4;

    private readonly Func<uint, LoadedLandblock?> _loadLandblock;
    private readonly Channel<LandblockStreamJob> _inbox;
    private readonly Channel<LandblockStreamResult> _outbox;
    private readonly CancellationTokenSource _cancel = new();
#pragma warning disable CS0649  // _worker stays declared for the future async path; unused in synchronous mode.
    private Thread? _worker;
#pragma warning restore CS0649
    private int _disposed;

    public LandblockStreamer(Func<uint, LoadedLandblock?> loadLandblock)
    {
        _loadLandblock = loadLandblock;
        _inbox = Channel.CreateUnbounded<LandblockStreamJob>(
            new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });
        _outbox = Channel.CreateUnbounded<LandblockStreamResult>(
            new UnboundedChannelOptions { SingleReader = true, SingleWriter = true });
    }

    /// <summary>
    /// No-op in synchronous mode. Preserved on the API surface so callers
    /// don't need to change when async loading returns in Phase A.3.
    /// </summary>
    public void Start()
    {
        if (System.Threading.Volatile.Read(ref _disposed) != 0)
            throw new ObjectDisposedException(nameof(LandblockStreamer));
        // No worker thread in synchronous mode.
    }

    public void EnqueueLoad(uint landblockId)
    {
        if (System.Threading.Volatile.Read(ref _disposed) != 0)
            throw new ObjectDisposedException(nameof(LandblockStreamer));
        // Synchronous mode: invoke the load delegate inline. The result lands
        // in the outbox and DrainCompletions picks it up later in the same
        // (or next) frame.
        HandleJob(new LandblockStreamJob.Load(landblockId));
    }

    public void EnqueueUnload(uint landblockId)
    {
        if (System.Threading.Volatile.Read(ref _disposed) != 0)
            throw new ObjectDisposedException(nameof(LandblockStreamer));
        HandleJob(new LandblockStreamJob.Unload(landblockId));
    }

    /// <summary>
    /// Drain up to <paramref name="maxBatchSize"/> completed results.
    /// Non-blocking. Call from the render thread once per OnUpdate.
    /// </summary>
    /// <remarks>
    /// Must be called from a single consumer thread. The outbox channel is
    /// configured with SingleReader = true and will throw on concurrent reads.
    /// </remarks>
    public IReadOnlyList<LandblockStreamResult> DrainCompletions(int maxBatchSize = DefaultDrainBatchSize)
    {
        var batch = new List<LandblockStreamResult>(maxBatchSize);
        while (batch.Count < maxBatchSize && _outbox.Reader.TryRead(out var result))
            batch.Add(result);
        return batch;
    }

    private void WorkerLoop()
    {
        try
        {
            // Synchronous read loop via .WaitToReadAsync + ReadAllAsync
            // would be idiomatic but requires async; the blocking reader
            // is simpler and the thread is dedicated anyway.
            while (!_cancel.Token.IsCancellationRequested)
            {
                // Safe to block: this is a dedicated worker thread with no
                // SynchronizationContext, so .Result/.GetResult cannot deadlock
                // against any captured continuation. Using the sync pattern
                // here keeps the loop linear; an async-enumerable alternative
                // would force WorkerLoop to be async Task and lose the
                // simple thread-start shape.
                if (!_inbox.Reader.WaitToReadAsync(_cancel.Token).AsTask().GetAwaiter().GetResult())
                    break;

                while (_inbox.Reader.TryRead(out var job))
                {
                    if (_cancel.Token.IsCancellationRequested) return;
                    HandleJob(job);
                }
            }
        }
        catch (OperationCanceledException) { /* graceful shutdown */ }
        catch (Exception ex)
        {
            // Last-ditch: surface via outbox so the caller at least sees
            // something. We never retry a crashed worker.
            _outbox.Writer.TryWrite(new LandblockStreamResult.WorkerCrashed(ex.ToString()));
        }
        finally
        {
            _outbox.Writer.TryComplete();
        }
    }

    private void HandleJob(LandblockStreamJob job)
    {
        switch (job)
        {
            case LandblockStreamJob.Load load:
                try
                {
                    var lb = _loadLandblock(load.LandblockId);
                    if (lb is null)
                        _outbox.Writer.TryWrite(new LandblockStreamResult.Failed(
                            load.LandblockId, "LandblockLoader.Load returned null"));
                    else
                        _outbox.Writer.TryWrite(new LandblockStreamResult.Loaded(
                            load.LandblockId, lb));
                }
                catch (Exception ex)
                {
                    _outbox.Writer.TryWrite(new LandblockStreamResult.Failed(
                        load.LandblockId, ex.ToString()));
                }
                break;

            case LandblockStreamJob.Unload unload:
                _outbox.Writer.TryWrite(new LandblockStreamResult.Unloaded(unload.LandblockId));
                break;
        }
    }

    public void Dispose()
    {
        if (System.Threading.Interlocked.Exchange(ref _disposed, 1) != 0) return;
        _cancel.Cancel();
        _inbox.Writer.TryComplete();
        _worker?.Join(TimeSpan.FromSeconds(2));
        _cancel.Dispose();
    }
}