stgsharp/src/StgSharp.Common/Timing/TimeSpanProvider.Awaiting.cs

// -----------------------------------------------------------------------------
// file="TimeSpanProvider.Awaiting"
// Project: StgSharp
// Copyright (c) Nitload.
// SPDX-License-Identifier: MIT
// -----------------------------------------------------------------------------

using StgSharp.HighPerformance.Memory;

using System;
using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;

namespace StgSharp.Timing
{
    public partial class TimeSpanProvider
    {

        private static readonly ConcurrentBufferStack<int> s_unusedID = new ConcurrentBufferStack<int>(8);

        private static int s_maxID;

        private EventHandler? _refreshHandler, _missHandler;
        private int _size, _id;
        private long _pendingSpans; // Number of unconsumed span ticks since last Wait.
        private readonly object _participantObject = new();

        public event EventHandler MissTimeSpanRefreshed
        {
            add => _missHandler += value;
            remove => _missHandler -= value;
        }

        public event EventHandler TimeSpanRefreshed
        {
            add => _refreshHandler += value;
            remove => _refreshHandler -= value;
        }

        /// <summary>
        ///   True if there is at least one unconsumed span tick (PendingSpanCount &gt; 0).
        /// </summary>
        public bool HasPendingSpan => PendingSpanCount > 0;

        public int TokenID
        {
            get => _id;
            private set => _id = value;
        }

        /// <summary>
        ///   Number of span ticks not yet consumed by Wait ( &gt; 0 means last tick not waited).
        /// </summary>
        public long PendingSpanCount => Volatile.Read(ref _pendingSpans);

        internal ManualResetEventSlim ResetEvent { get; } = new();

        public void Dispose()
        {
#pragma warning disable CA1816 
            // mark provider as ended so external users can observe terminal state
            Volatile.Write(ref _endedFlag, 1);
            s_unusedID.Push(TokenID);
            ResetEvent.Set();
            GC.SuppressFinalize(this);
#pragma warning restore CA1816 
        }

        public override int GetHashCode()
        {
            return _id;
        }

        public void Participant()
        {
            lock (_participantObject)
            {
                // Increase the maximum concurrent consumers for the next tick.
                _size++;
            }
        }

        public void WaitNextSpan()
        {
            ResetEvent.Wait();

            // Consume one span slot (clamp at 0 so multiple subscribers won't make it negative).
            while (true)
            {
                long cur = Volatile.Read(ref _pendingSpans);
                if (cur == 0) {
                    return;
                }

                if (Interlocked.CompareExchange(ref _pendingSpans, cur - 1, cur) == cur)
                {
                    if (cur - 1 == 0)
                    {
                        // Last consumer resets the gate for the next tick.
                        ResetEvent.Reset();
                    }
                    return;
                }
            }
        }

        public async Task WaitNextSpanAsync()
        {
            await Task.Run(ResetEvent.Wait).ConfigureAwait(false);
            while (true)
            {
                long cur = Volatile.Read(ref _pendingSpans);
                if (cur == 0)
                {
                    break;
                }

                if (Interlocked.CompareExchange(ref _pendingSpans, cur - 1, cur) == cur)
                {
                    if (cur - 1 == 0) {
                        ResetEvent.Reset();
                    }

                    break;
                }
            }
        }

        /// <summary>
        ///   Wait for the next timespan tick, but honor the provided cancellation token. Throws
        ///   OperationCanceledException when token is cancelled. If the provider is disposed during
        ///   wait an OperationCanceledException is thrown to indicate end.
        /// </summary>
        public Task WaitNextSpanAsync(
                    CancellationToken ct
        )
        {
            // Block the current thread until the event is signaled or cancellation requested.
            ResetEvent.Wait(ct);

            // consume one pending span (same logic as non-cancellable variant)
            while (true)
            {
                long cur = Volatile.Read(ref _pendingSpans);
                if (cur == 0)
                {
                    break;
                }

                if (Interlocked.CompareExchange(ref _pendingSpans, cur - 1, cur) == cur)
                {
                    if (cur - 1 == 0) {
                        ResetEvent.Reset();
                    }

                    break;
                }
            }

            return Task.CompletedTask;
        }

        internal void MissRefresh()
        {
            _missHandler?.Invoke(this, EventArgs.Empty);
        } // Legacy compatibility.

        /// <summary>
        ///
        /// </summary>
        /// <param name="currentTick">
        ///   Tick passed since binded time source.
        /// </param>
        /// <returns>
        ///   True if the span provider is working. False if time is over (the sequence ended).
        /// </returns>
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        internal bool OnSpanTick(
                      long currentTick
        )
        {
            if (_sequence.EndsAt(currentTick))
            {
                // mark ended due to sequence timeout, then dispose
                Volatile.Write(ref _endedFlag, 1);

                // Console.WriteLine("Time span ended.");
                Dispose();
                return false;
            }
            if (!_sequence.IsNextFrameReady(currentTick)) {
                return true;
            }

            // Console.WriteLine("Frame hit!");

            // Fast path: if no pending consumers, initialize the counter to current capacity (_size).
            if (Volatile.Read(ref _pendingSpans) == 0)
            {
                int toIssue = Math.Max(1, Volatile.Read(ref _size));
                if (Interlocked.CompareExchange(ref _pendingSpans, toIssue, 0) == 0)
                {
                    // Open the gate for all current waiters (up to `_size`).
                    ResetEvent.Set();
                    _refreshHandler?.Invoke(this, EventArgs.Empty);
                }
            } else
            {
                // A tick is already pending (or was just set by another thread): mark as miss.
                _missHandler?.Invoke(this, EventArgs.Empty);
            }
            return true;
        }

        ~TimeSpanProvider()
        {
            Dispose();
        }

    }
}