DreamineThread.cs

이 파일의 문서화 페이지로 가기

using Dreamine.Logging.Interfaces;
using Dreamine.Threading.Interfaces;
using Dreamine.Threading.Models;
 
namespace Dreamine.Threading.Services;

public sealed class DreamineThread : IDreamineThread
{
    private readonly object _syncRoot = new();
    private readonly List<IDreamineThreadJob> _jobs = new();
    private readonly IThreadCyclePolicy _cyclePolicy;
    private readonly IThreadAffinityService? _affinityService;
    private readonly ITimerResolutionService? _timerResolutionService;
    private readonly IDreamineLogger? _logger;
    private readonly ManualResetEventSlim _pauseEvent = new(true);

    private Thread? _thread;
    private CancellationTokenSource? _cancellationTokenSource;
    private long _cycleCount;
    private DateTimeOffset? _startedAt;
    private DateTimeOffset? _stoppedAt;
    private string? _lastErrorMessage;
    private bool _disposed;

    public string Name { get; }

    public DreamineThreadOptions Options { get; }

    public DreamineThreadCoreAssignment CoreAssignment { get; }

    private volatile DreamineThreadStatus _status = DreamineThreadStatus.Created;

    public DreamineThreadStatus Status
    {
        get => _status;
        private set => _status = value;
    }

    public int JobCount
    {
        get
        {
            lock (_syncRoot)
            {
                return _jobs.Count;
            }
        }
    }

    public DreamineThread(
        DreamineThreadOptions options,
        DreamineThreadCoreAssignment coreAssignment,
        IThreadCyclePolicy cyclePolicy,
        IThreadAffinityService? affinityService = null,
        ITimerResolutionService? timerResolutionService = null,
        IDreamineLogger? logger = null)
    {
        ArgumentNullException.ThrowIfNull(options);
        ArgumentNullException.ThrowIfNull(coreAssignment);
        ArgumentNullException.ThrowIfNull(cyclePolicy);
 
        Options = options.Normalize();
        Name = Options.Name;
        CoreAssignment = coreAssignment;
        _cyclePolicy = cyclePolicy;
        _affinityService = affinityService;
        _timerResolutionService = timerResolutionService;
        _logger = logger;
    }

    public void AddJob(IDreamineThreadJob job)
    {
        ArgumentNullException.ThrowIfNull(job);
 
        lock (_syncRoot)
        {
            ThrowIfDisposed();
            _jobs.Add(job);
        }
    }

    public void Start()
    {
        Thread? threadToStart = null;
 
        lock (_syncRoot)
        {
            ThrowIfDisposed();
 
            if (Status == DreamineThreadStatus.Running)
            {
                return;
            }
 
            _cancellationTokenSource = new CancellationTokenSource();
            _pauseEvent.Set();
 
            _thread = new Thread(() => Run(_cancellationTokenSource.Token))
            {
                IsBackground = true,
                Name = Name,
                Priority = MapPriority(Options.Priority)
            };
 
            Status = DreamineThreadStatus.Running;
            _startedAt = DateTimeOffset.Now;
            _stoppedAt = null;
            _lastErrorMessage = null;
 
            threadToStart = _thread;
        }
 
        // Start the thread outside the lock so that any logging or scheduling
        // callbacks triggered by the OS during startup cannot deadlock against
        // the lock held by callers like the manager.
        threadToStart?.Start();
 
        _logger?.Info($"Thread started. Name={Name}");
    }

    public void Pause()
    {
        bool transitioned;
 
        lock (_syncRoot)
        {
            if (Status != DreamineThreadStatus.Running)
            {
                transitioned = false;
            }
            else
            {
                _pauseEvent.Reset();
                Status = DreamineThreadStatus.Paused;
                transitioned = true;
            }
        }
 
        if (transitioned)
        {
            _logger?.Info($"Thread paused. Name={Name}");
        }
    }

    public void Resume()
    {
        bool transitioned;
 
        lock (_syncRoot)
        {
            if (Status != DreamineThreadStatus.Paused)
            {
                transitioned = false;
            }
            else
            {
                _pauseEvent.Set();
                Status = DreamineThreadStatus.Running;
                transitioned = true;
            }
        }
 
        if (transitioned)
        {
            _logger?.Info($"Thread resumed. Name={Name}");
        }
    }

    [Obsolete("Stop() blocks the calling thread and risks deadlock on a SynchronizationContext. Use StopAsync() instead. / 호출 스레드를 블로킹하며 SynchronizationContext 환경에서 데드락 위험이 있습니다. StopAsync()를 사용하세요.")]
    public void Stop()
    {
        StopAsync().AsTask().GetAwaiter().GetResult();
    }

    public async ValueTask StopAsync()
    {
        Thread? threadToJoin;
 
        lock (_syncRoot)
        {
            if (Status is DreamineThreadStatus.Stopped or DreamineThreadStatus.Disposed)
            {
                return;
            }
 
            Status = DreamineThreadStatus.Stopping;
            _pauseEvent.Set();
 
            _cancellationTokenSource?.Cancel();
            threadToJoin = _thread;
        }
 
        if (threadToJoin is not null && threadToJoin.IsAlive)
        {
            await Task.Run(
                () => threadToJoin.Join(Options.StopTimeout))
                .ConfigureAwait(false);
        }
 
        lock (_syncRoot)
        {
            _cancellationTokenSource?.Dispose();
            _cancellationTokenSource = null;
            _thread = null;
 
            if (Status != DreamineThreadStatus.Faulted)
            {
                Status = DreamineThreadStatus.Stopped;
            }
 
            _stoppedAt = DateTimeOffset.Now;
        }
 
        _logger?.Info($"Thread stopped. Name={Name}");
    }

    public DreamineThreadInfo GetInfo()
    {
        return new DreamineThreadInfo(
            Name,
            Status,
            Options.Priority,
            Options.IntervalMs,
            CoreAssignment.CoreIndex,
            CoreAssignment.UseAffinity,
            JobCount,
            Interlocked.Read(ref _cycleCount),
            _startedAt,
            _stoppedAt,
            _lastErrorMessage);
    }

    public void Dispose()
    {
        if (_disposed)
        {
            return;
        }
 
        StopAsync().AsTask().GetAwaiter().GetResult();
 
        _pauseEvent.Dispose();
        _disposed = true;
        Status = DreamineThreadStatus.Disposed;
    }

    private void Run(CancellationToken cancellationToken)
    {
        try
        {
            if (Options.UseHighPrecisionTimer)
            {
                _timerResolutionService?.Begin();
            }
 
            if (CoreAssignment.UseAffinity && CoreAssignment.CoreIndex is not null)
            {
                _affinityService?.ApplyToCurrentThread(CoreAssignment.CoreIndex.Value);
            }
 
            while (!cancellationToken.IsCancellationRequested)
            {
                _pauseEvent.Wait(cancellationToken);
 
                ExecuteDueJobs(cancellationToken);
 
                Interlocked.Increment(ref _cycleCount);
 
                var context = new DreamineThreadCycleContext(
                    Name,
                    Interlocked.Read(ref _cycleCount),
                    JobCount,
                    CoreAssignment.CoreIndex,
                    CoreAssignment.IsOverflowPolling,
                    DateTimeOffset.Now);
 
                var delayMs = _cyclePolicy.GetDelayMs(
                    Options,
                    CoreAssignment,
                    context);
 
                if (delayMs > 0)
                {
                    cancellationToken.WaitHandle.WaitOne(delayMs);
                }
                else if (Options.YieldWhenIntervalIsZero)
                {
                    Thread.Yield();
                }
            }
        }
        catch (OperationCanceledException)
        {
            // Normal stop path.
        }
        catch (Exception ex)
        {
            _lastErrorMessage = ex.Message;
            Status = DreamineThreadStatus.Faulted;
            _logger?.Error(ex, $"Thread faulted. Name={Name}");
        }
        finally
        {
            if (Options.UseHighPrecisionTimer)
            {
                _timerResolutionService?.End();
            }
 
            if (CoreAssignment.UseAffinity)
            {
                _affinityService?.ClearCurrentThreadAffinity();
            }
        }
    }

    private void ExecuteDueJobs(CancellationToken cancellationToken)
    {
        IDreamineThreadJob[] jobs;
 
        lock (_syncRoot)
        {
            jobs = _jobs.ToArray();
        }
 
        var now = DateTimeOffset.Now;
 
        foreach (var job in jobs)
        {
            if (!job.ShouldRun(now))
            {
                continue;
            }
 
            try
            {
                // Jobs run on a dedicated worker thread. Blocking here is
                // intentional so jobs preserve registration order and the cycle
                // policy observes completed work before calculating the next delay.
                job.ExecuteAsync(cancellationToken).GetAwaiter().GetResult();
            }
            catch (OperationCanceledException)
            {
                throw;
            }
            catch (Exception ex)
            {
                _lastErrorMessage = ex.Message;
                _logger?.Error(ex, $"Thread job failed. Thread={Name}, Job={job.Name}");
            }
        }
    }

    private static ThreadPriority MapPriority(DreamineThreadPriority priority)
    {
        return priority switch
        {
            DreamineThreadPriority.High => ThreadPriority.AboveNormal,
            DreamineThreadPriority.Low => ThreadPriority.BelowNormal,
            _ => ThreadPriority.Normal
        };
    }

    private void ThrowIfDisposed()
    {
        if (_disposed)
        {
            throw new ObjectDisposedException(nameof(DreamineThread));
        }
    }
}