IValueTaskSource and ManualResetValueTaskSourceCore for CPU-work (just as PoC)

IValueTaskSource and ManualResetValueTaskSourceCore for CPU-work.cs

using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Sources;
using BenchmarkDotNet.Attributes;

#if !DEBUG
using BenchmarkDotNet.Running;
#endif

await TestRunner.RunAsync(iterations: 2);

#if !DEBUG
Console.WriteLine("Starting benchmark");
BenchmarkRunner.Run<Bench>();
#endif

[MemoryDiagnoser]
public class Bench
{
    private int _a;
    private int _b;

    [GlobalSetup]
    public void GlobalSetup()
    {
        _a++;
        _b++;
    }

    [Benchmark(Baseline = true)]
    public async Task<int> TaskRun()
    {
        return await Task.Run(() => _a + _b);
    }

    [Benchmark]
    public async ValueTask<int> TaskRun_ValueTask()
    {
        return await Task.Run(() => _a + _b);
    }

    [Benchmark]
    public async Task<int> TaskFactory_with_State()
    {
        return await Task.Factory.StartNew(state =>
        {
            Bench? b = state as Bench;
            Debug.Assert(b is not null);

            return b._a + b._b;
        }, this);
    }

    private ValueTaskSourceWorker? _valueTaskSourceWorker;

    [Benchmark]
    public async ValueTask<int> ValueTaskSource()
    {
        _valueTaskSourceWorker ??= new ValueTaskSourceWorker();
        return await _valueTaskSourceWorker.RunAsync(_a, _b);
    }

    private TaskWorker? _taskWorker;

    [Benchmark]
    public async Task<int> AsyncTaskMethodBuilder()
    {
        _taskWorker ??= new TaskWorker();
        return await _taskWorker.RunAsync(_a, _b);
    }

    private ValueTaskWorker? _valueTaskWorker;

    [Benchmark]
    public async ValueTask<int> AsyncValueTaskMethodBuilder()
    {
        _valueTaskWorker ??= new ValueTaskWorker();
        return await _valueTaskWorker.RunAsync(_a, _b);
    }
}

public class ValueTaskSourceWorker : IThreadPoolWorkItem, IValueTaskSource<int>
{
    private ManualResetValueTaskSourceCore<int> _mrvtsc = new ManualResetValueTaskSourceCore<int>();
    private int _a;
    private int _b;

    public ValueTask<int> RunAsync(int a, int b)
    {
        _a = a;
        _b = b;

        ThreadPool.UnsafeQueueUserWorkItem(this, preferLocal: false);
        return new ValueTask<int>(this, _mrvtsc.Version);
    }

    void IThreadPoolWorkItem.Execute()
    {
        int result = _a + _b;
        _mrvtsc.SetResult(result);
    }

    int IValueTaskSource<int>.GetResult(short token)
    {
        int result = _mrvtsc.GetResult(token);
        _mrvtsc.Reset();
        return result;
    }

    ValueTaskSourceStatus IValueTaskSource<int>.GetStatus(short token) => _mrvtsc.GetStatus(token);
    void IValueTaskSource<int>.OnCompleted(Action<object?> continuation, object? state, short token, ValueTaskSourceOnCompletedFlags flags)
        => _mrvtsc.OnCompleted(continuation, state, token, flags);
}

public class TaskWorker : IThreadPoolWorkItem
{
    private AsyncTaskMethodBuilder<int> _atmb;
    private int _a;
    private int _b;

    public Task<int> RunAsync(int a, int b)
    {
        _a = a;
        _b = b;
        _atmb = AsyncTaskMethodBuilder<int>.Create();
        Task<int> task = _atmb.Task;    // must be initialized, otherwise it may not complete
        ThreadPool.UnsafeQueueUserWorkItem(this, preferLocal: false);
        return task;
    }

    void IThreadPoolWorkItem.Execute()
    {
        int result = _a + _b;
        _atmb.SetResult(result);
    }
}

public class ValueTaskWorker : IThreadPoolWorkItem
{
    private AsyncValueTaskMethodBuilder<int> _avtmb;
    private int _a;
    private int _b;

    public ValueTask<int> RunAsync(int a, int b)
    {
        _a = a;
        _b = b;
        _avtmb = AsyncValueTaskMethodBuilder<int>.Create();
        ValueTask<int> task = _avtmb.Task;  // must be initialized, otherwise it may not complete
        ThreadPool.UnsafeQueueUserWorkItem(this, preferLocal: false);
        return task;
    }

    void IThreadPoolWorkItem.Execute()
    {
        int result = _a + _b;
        _avtmb.SetResult(result);
    }
}

internal static class TestRunner
{
    public static async Task RunAsync(int iterations)
    {
        var bench = new Bench();
        for (int i = 0; i < iterations; ++i)
        {
            bench.GlobalSetup();

            int res = await bench.TaskRun();
            Console.WriteLine(res);

            res = await bench.TaskRun_ValueTask();
            Console.WriteLine(res);

            res = await bench.TaskFactory_with_State();
            Console.WriteLine(res);

            res = await bench.ValueTaskSource();
            Console.WriteLine(res);

            res = await bench.AsyncTaskMethodBuilder();
            Console.WriteLine(res);

            res = await bench.AsyncValueTaskMethodBuilder();
            Console.WriteLine(res);

            Console.WriteLine();
        }
    }
}

Allocations

Each run was recorded for about 3s (via using CancellationTokenSource cts = new CancellationTokenSource(3_000);) -- so absolute numbers have to be read with a grain of salt (especially the more performant an implementation, the more iterations will be done, hence more allocations of the state machine box).

TaskRun_ValueTask

TaskRun

TaskFactory_with_State

AsyncValueTaskMethodBuilder

AsyncTaskMethodBuilder

ValueTaskSource