Потоки C#: задача, отмена и пул без доступа к Entity

using System;
using System.Threading;

public static class PrimeWork
{
    public const int MaxLimit = 2_000_000;

    public static int Count(int limit, CancellationToken token = default)
    {
        if (limit < 0 || limit > MaxLimit)
            throw new ArgumentOutOfRangeException(nameof(limit));
        token.ThrowIfCancellationRequested();
        int count = 0;
        for (int number = 2; number <= limit; number++)
        {
            if ((number & 127) == 0) token.ThrowIfCancellationRequested();
            bool prime = true;
            for (int divisor = 2; divisor <= number / divisor; divisor++)
                if (number % divisor == 0) { prime = false; break; }
            if (prime) count++;
        }
        token.ThrowIfCancellationRequested();
        return count;
    }
}

using System;
using System.Threading;
using System.Threading.Tasks;
using Reckfell;

public partial class BackgroundJob : Entity
{
    public enum Outcome { Idle, Running, Completed, Cancelled, Failed }
    public bool Busy => _task is not null;
    public Outcome State { get; private set; }
    public int Result { get; private set; }
    public string LastError { get; private set; } = "";
    public event Action? Settled;

    private Task<int>? _task;
    private CancellationTokenSource? _cancellation;

    public Error Start(Func<CancellationToken, int>? work)
    {
        if (!ReckfellThread.IsMainThread()) return Error.Unavailable;
        if (!IsInsideTree()) return Error.Unconfigured;
        if (work is null) return Error.InvalidParameter;
        if (Busy) return Error.Busy;
        var source = new CancellationTokenSource();
        CancellationToken token = source.Token;
        _cancellation = source;
        Result = 0;
        LastError = "";
        State = Outcome.Running;
        _task = Task.Run(() => work(token), token);
        SetProcess(true);
        return Error.Ok;
    }

    public void Cancel()
    {
        if (!ReckfellThread.IsMainThread())
            throw new InvalidOperationException("Cancel must run on the main thread");
        _cancellation?.Cancel();
    }

    public override void _Process(double delta)
    {
        if (_task is not { IsCompleted: true } task) return;
        var source = _cancellation!;
        _task = null;
        _cancellation = null;
        SetProcess(false);
        try
        {
            int result = task.GetAwaiter().GetResult();
            State = source.IsCancellationRequested ? Outcome.Cancelled : Outcome.Completed;
            if (State == Outcome.Completed) Result = result;
        }
        catch (OperationCanceledException) when (source.IsCancellationRequested)
        {
            State = Outcome.Cancelled;
        }
        catch (Exception error)
        {
            State = Outcome.Failed;
            LastError = error.GetType().Name + ": " + error.Message;
        }
        finally { source.Dispose(); }
        Settled?.Invoke();
    }

    public override void _ExitTree()
    {
        var task = _task;
        var source = _cancellation;
        _task = null;
        _cancellation = null;
        State = Outcome.Idle;
        SetProcess(false);
        if (task is null || source is null) return;
        source.Cancel();
        _ = task.ContinueWith(static (finished, state) =>
        {
            _ = finished.Exception;
            ((CancellationTokenSource)state!).Dispose();
        }, source, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously,
            TaskScheduler.Default);
    }
}

using System;
using System.Threading;
using Reckfell;

public partial class PoolPrimeBatch : Entity
{
    private sealed class Batch(int[] limits)
    {
        public readonly int[] Limits = limits;
        public readonly int[] Results = new int[limits.Length];
        public readonly Exception?[] Errors = new Exception?[limits.Length];
        public readonly CancellationTokenSource Cancellation = new();

        public void Compute(int index)
        {
            try { Results[index] = PrimeWork.Count(Limits[index], Cancellation.Token); }
            catch (Exception error) { Errors[index] = error; }
        }
    }

    public bool Busy => _group >= 0;
    public BackgroundJob.Outcome State { get; private set; }
    public int[] Results => (int[])_results.Clone();
    public string LastError { get; private set; } = "";
    public event Action? Settled;
    private long _group = -1;
    private Batch? _batch;
    private int[] _results = [];

    public Error Start(int[]? limits)
    {
        if (!ReckfellThread.IsMainThread()) return Error.Unavailable;
        if (!IsInsideTree()) return Error.Unconfigured;
        if (Busy) return Error.Busy;
        if (limits is null || limits.Length < 1 || limits.Length > 32)
            return Error.InvalidParameter;
        int[] copy = (int[])limits.Clone();
        foreach (int limit in copy)
            if (limit < 0 || limit > PrimeWork.MaxLimit) return Error.InvalidParameter;
        _batch = new Batch(copy);
        _results = [];
        LastError = "";
        State = BackgroundJob.Outcome.Running;
        _group = WorkerThreadPool.AddGroupTask(Callable.From<int>(_batch.Compute),
            copy.Length, description: "Prime counts");
        if (_group < 0)
        {
            _batch.Cancellation.Dispose();
            _batch = null;
            State = BackgroundJob.Outcome.Failed;
            LastError = "Could not schedule group";
            return Error.CantCreate;
        }
        SetProcess(true);
        return Error.Ok;
    }

    public void Cancel()
    {
        if (!ReckfellThread.IsMainThread())
            throw new InvalidOperationException("Cancel must run on the main thread");
        _batch?.Cancellation.Cancel();
    }

    public override void _Process(double delta)
    {
        if (!Busy || !WorkerThreadPool.IsGroupTaskCompleted(_group)) return;
        WorkerThreadPool.WaitForGroupTaskCompletion(_group);
        _group = -1;
        Batch batch = _batch!;
        _batch = null;
        SetProcess(false);
        State = batch.Cancellation.IsCancellationRequested
            ? BackgroundJob.Outcome.Cancelled : BackgroundJob.Outcome.Completed;
        foreach (Exception? error in batch.Errors)
        {
            if (error is null || error is OperationCanceledException &&
                batch.Cancellation.IsCancellationRequested) continue;
            State = BackgroundJob.Outcome.Failed;
            LastError = error.GetType().Name + ": " + error.Message;
            break;
        }
        if (State == BackgroundJob.Outcome.Completed) _results = batch.Results;
        batch.Cancellation.Dispose();
        Settled?.Invoke();
    }

    public override void _ExitTree()
    {
        SetProcess(false);
        if (!Busy) return;
        _batch!.Cancellation.Cancel();
        WorkerThreadPool.WaitForGroupTaskCompletion(_group);
        _batch.Cancellation.Dispose();
        _batch = null;
        _group = -1;
        State = BackgroundJob.Outcome.Idle;
    }
}

using System;
using Reckfell;

public partial class JobTelemetry : Entity
{
    public string SecondsId { get; private set; } = "";
    public string CountId { get; private set; } = "";
    public double Seconds { get; private set; }
    public int Count { get; private set; }
    private bool _registered;

    public override void _EnterTree()
    {
        string category = "docs_perf_" + GetInstanceId();
        SecondsId = category + "/prime_seconds";
        CountId = category + "/prime_count";
        Performance.AddCustomMonitor(SecondsId, Callable.From(() => Seconds),
            type: Performance.MonitorType.Time);
        Performance.AddCustomMonitor(CountId, Callable.From(() => Count),
            type: Performance.MonitorType.Quantity);
        _registered = true;
    }

    public void Publish(double seconds, int count)
    {
        if (!ReckfellThread.IsMainThread())
            throw new InvalidOperationException("Publish must run on the main thread");
        if (!double.IsFinite(seconds) || seconds < 0 || count < 0)
            throw new ArgumentOutOfRangeException(nameof(seconds));
        Seconds = seconds;
        Count = count;
    }

    public override void _ExitTree()
    {
        if (!_registered) return;
        Performance.RemoveCustomMonitor(SecondsId);
        Performance.RemoveCustomMonitor(CountId);
        _registered = false;
    }
}

using System.Diagnostics;
using Reckfell;

public partial class PrimeJobsDemo : Entity
{
    private readonly Stopwatch _clock = new();
    private BackgroundJob _job = null!;
    private PoolPrimeBatch _pool = null!;
    public JobTelemetry Telemetry { get; private set; } = null!;

    public override void _Ready()
    {
        Telemetry = new JobTelemetry();
        _job = new BackgroundJob();
        _pool = new PoolPrimeBatch();
        AddChild(Telemetry);
        AddChild(_job);
        AddChild(_pool);
        _job.Settled += OnJobSettled;
        _pool.Settled += OnPoolSettled;
        _clock.Restart();
        Error error = _job.Start(static token => PrimeWork.Count(100_000, token));
        if (error != Error.Ok) ReckfellApi.PrintErr("JOB start: " + error);
        error = _pool.Start([10_000, 100_000]);
        if (error != Error.Ok) ReckfellApi.PrintErr("POOL start: " + error);
    }

    private void OnJobSettled()
    {
        _clock.Stop();
        if (_job.State == BackgroundJob.Outcome.Completed)
        {
            Telemetry.Publish(_clock.Elapsed.TotalSeconds, _job.Result);
            ReckfellApi.Print($"JOB primes={_job.Result} main_thread={ReckfellThread.IsMainThread()}");
        }
        else ReckfellApi.PrintErr("JOB: " + _job.State + " " + _job.LastError);
    }

    private void OnPoolSettled()
    {
        if (_pool.State == BackgroundJob.Outcome.Completed)
            ReckfellApi.Print("POOL primes=" + string.Join(",", _pool.Results));
        else ReckfellApi.PrintErr("POOL: " + _pool.State + " " + _pool.LastError);
    }

    public override void _ExitTree()
    {
        _job.Settled -= OnJobSettled;
        _pool.Settled -= OnPoolSettled;
    }
}