Consider using dotNext.Threading ValueTaskCompletionSource for creating a ValueTask pool

[arontsang]

Consider using ValueTaskCompletionSource to create a ValueTask pool.

This might be faster than creating TaskCompletionSource for each and every method call.

[arontsang]

See https://dotnet.github.io/dotNext/api/DotNext.Threading.Tasks.ValueTaskCompletionSource-1.html for usage

[arontsang]

Did some testing

Got this result

BenchmarkDotNet v0.14.0, Debian GNU/Linux 12 (bookworm) WSL
AMD Ryzen 5 3600, 1 CPU, 4 logical and 2 physical cores
.NET SDK 9.0.102
[Host] : .NET 9.0.1 (9.0.124.61010), X64 RyuJIT AVX2
DefaultJob : .NET 9.0.1 (9.0.124.61010), X64 RyuJIT AVX2

Method	Mean	Error	StdDev	Ratio	Allocated	Alloc Ratio
Baseline	276.2 ns	2.01 ns	1.57 ns	1.00	128 B	1.00
Fast	156.8 ns	1.52 ns	1.34 ns	0.57	-	0.00

Using the following code

public static class Helper
{
    
    [MethodImpl(MethodImplOptions.AggressiveOptimization)]
    public static ValueTask<ulong> ReturnAfterMsFast(this AsyncService service, ulong x, ulong ms)
    {
        return ReturnAfterMsHelper.ReturnAfterMsFast(service, x, ms);
    }

    private static class ReturnAfterMsHelper
    {
        private static int CurrentCallId = 0;
        private static IntPtr CallbackPointer;

        private static RecyclableTaskCompletionSource<ulong> Foo = new();

        static ReturnAfterMsHelper()
        {
            unsafe
            {
                delegate* unmanaged[Cdecl] <IntPtr, nint, void> callback = &Callback;
                CallbackPointer = (IntPtr)callback;
            }
        }
        
        
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        public static ValueTask<ulong> ReturnAfterMsFast(AsyncService service, ulong x, ulong ms)
        {
            var tcs = RecyclableTaskCompletionSource<ulong>.Get();
            var handle = GCHandle.Alloc(tcs, GCHandleType.Normal);
            var ac = new AsyncCallbackCommonNative {
                _ptr = CallbackPointer,
                _ts = GCHandle.ToIntPtr(handle),
            };

            Interop.async_service_return_after_ms(service.Context, x, ms, ac);
            
            return tcs.GetTask();
        }
        
        [UnmanagedCallersOnly(CallConvs = [typeof(CallConvCdecl)])]
        static void Callback(IntPtr data, IntPtr callbackId)
        {
            var handle = GCHandle.FromIntPtr(callbackId);
            try
            {
                if (handle.Target is not RecyclableTaskCompletionSource<ulong> tcs)
                    return;
                Span<ResultU64Error.Unmanaged> unmanaged;
                unsafe
                {
                    unmanaged = new Span<ResultU64Error.Unmanaged>((void*)data, 1);
                }
                var managed = unmanaged[0].ToManaged();
                if (managed.IsOk)
                {
                    tcs.SetResult(managed.AsOk());
                }
                else
                {
                    tcs.SetException(new InteropException());
                }
            }
            finally
            {
                handle.Free();
            }
        }
    }
    
    private class RecyclableTaskCompletionSourcePolicy<T> : IPooledObjectPolicy<RecyclableTaskCompletionSource<T>>
    {
        public static RecyclableTaskCompletionSourcePolicy<T> Instance { get; } = new();
        
        public RecyclableTaskCompletionSource<T> Create()
        {
            return new RecyclableTaskCompletionSource<T>();
        }

        public bool Return(RecyclableTaskCompletionSource<T> obj)
        {
            obj.Reset();
            return true;
        }
    }
    
    internal sealed class RecyclableTaskCompletionSource<T> : IValueTaskSource<T>
    {
        private static readonly ObjectPool<RecyclableTaskCompletionSource<T>> Pool =
            ObjectPool.Create(RecyclableTaskCompletionSourcePolicy<T>.Instance);
        
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        public static RecyclableTaskCompletionSource<T> Get() => Pool.Get();

        public RecyclableTaskCompletionSource()
        {
        }

        private ManualResetValueTaskSourceCore<T> _core = new();
        
        internal void Reset() => _core.Reset();
        
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        internal ValueTask<T> GetTask() => new(this, _core.Version);
        
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        T IValueTaskSource<T>.GetResult(short token)
        {
            try
            {
                return _core.GetResult(token);
            }
            finally
            {
                Pool.Return(this);
            }
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        ValueTaskSourceStatus IValueTaskSource<T>.GetStatus(short token)
        {
            return _core.GetStatus(token);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        void IValueTaskSource<T>.OnCompleted(Action<object?> continuation, object? state, short token, ValueTaskSourceOnCompletedFlags flags)
        {
            _core.OnCompleted(continuation, state, token, flags);
        }
        
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void SetResult(T result) => _core.SetResult(result);
        
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void SetException(Exception result) => _core.SetException(result);
    }
}

[arontsang]

Seems a large amount of time is spent in getting the correct bucket in the Dictionary. By switching to using GcHandle instead, we swap a GetHashCode + mod divide, with a pin.

[ralfbiedert]

Thanks, this is definitely interesting. Main question is probably how to handle the external dependency.

[arontsang]

@ralfbiedert Did a bit more work. I swapped out my toy example code for an instant return function, along with a "runtime" based on futures::executor::block_on (completely unusable in production, but will obviously run with almost zero overhead).

Based on my testing, we are close to the limits of what I can optimize for on the C# side, wrt to synchronous complete cases (stripping out my help, and returning a ValueTask.FromResult(0) still yields ~100ns).

#[ffi_service]
impl AsyncService {
    pub fn new() -> ffi::Result<Self, Error> {
        result_to_ffi(|| {
            Ok(Self {  })
        })
    }

    pub async fn return_after_ms(_this: This, x: u64, ms: u64) -> ffi::Result<u64, Error> {
        result_to_ffi_async(async || {
            Ok(x)
        }).await
    }
}

impl AsyncRuntime for AsyncService {
    fn spawn<Fn, F>(&self, f: Fn)
    where
        Fn: FnOnce(()) -> F,
        F: Future<Output=()> + Send + 'static
    {
        let future = f(());
        futures::executor::block_on(async move { future.await })
    }
}

Method	Mean	Error	StdDev	Ratio	RatioSD	Allocated	Alloc Ratio
Baseline	161.17 ns	3.247 ns	3.037 ns	1.00	0.03	128 B	1.00
Fast	73.42 ns	1.019 ns	0.953 ns	0.46	0.01	-	0.00

With the following code

public static class Helper
{
    [MethodImpl(MethodImplOptions.AggressiveOptimization)]
    public static ValueTask<ulong> ReturnAfterMsFast(this AsyncService service, ulong x, ulong ms)
    {
        return ReturnAfterMsHelper.ReturnAfterMsFast(service, x, ms);
    }

    private static class ReturnAfterMsHelper
    {
        private static readonly IntPtr CallbackPointer;


        static ReturnAfterMsHelper()
        {
            unsafe
            {
                delegate* unmanaged[Cdecl] <IntPtr, nint, void> callback = &Callback;
                CallbackPointer = (IntPtr)callback;
            }
        }
        
        
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        public static ValueTask<ulong> ReturnAfterMsFast(AsyncService service, ulong x, ulong ms)
        {
            var tcs = RecyclableTaskCompletionSource<ulong>.Get();
            var ac = new AsyncCallbackCommonNative {
                _ptr = CallbackPointer,
                _ts = tcs.Id,
            };

            Interop.async_service_return_after_ms(service.Context, x, ms, ac);
            
            

            return tcs.GetTask();
        }
        
        [UnmanagedCallersOnly(CallConvs = [typeof(CallConvCdecl)])]
        static void Callback(IntPtr data, IntPtr callbackId)
        {
            
            var tcs = RecyclableTaskCompletionSource<ulong>.GetById(unchecked((int)callbackId));
            Span<ResultU64Error.Unmanaged> unmanaged;
            unsafe
            {
                unmanaged = new Span<ResultU64Error.Unmanaged>((void*)data, 1);
            }
            var managed = unmanaged[0].ToManaged();
            if (managed.IsOk)
            {
                tcs.SetResult(managed.AsOk());
            }
            else
            {
                tcs.SetException(new InteropException());
            }
        }
    }
    
    private sealed class RecyclableTaskCompletionSourcePolicy<T> : IPooledObjectPolicy<RecyclableTaskCompletionSource<T>>
    {
        public static RecyclableTaskCompletionSourcePolicy<T> Instance { get; } = new();
        
        
        public RecyclableTaskCompletionSource<T> Create()
        {
            return new RecyclableTaskCompletionSource<T>();
        }

        public bool Return(RecyclableTaskCompletionSource<T> obj)
        {
            obj.Reset();
            return true;
        }
    }


    
    internal sealed class RecyclableTaskCompletionSource<T> : IValueTaskSource<T>
    {
        private static readonly ConcurrentDictionary<int, RecyclableTaskCompletionSource<T>> Registered = new();
        private static int NextId = 0;
        
        internal static readonly ObjectPool<RecyclableTaskCompletionSource<T>> Pool =
            ObjectPool.Create(RecyclableTaskCompletionSourcePolicy<T>.Instance);

        public static RecyclableTaskCompletionSource<T> GetById(int id) => Registered[id];
        
        public int Id { get; }
        
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        public static RecyclableTaskCompletionSource<T> Get() => Pool.Get();

        public RecyclableTaskCompletionSource()
        {
            Id = Interlocked.Increment(ref NextId) - 1;
            Registered[Id] = this;
        }

        private ManualResetValueTaskSourceCore<T> _core = new();
        
        internal void Reset()
        {
            _core.Reset();
        }
        
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        internal ValueTask<T> GetTask() => new(this, _core.Version);
        
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        T IValueTaskSource<T>.GetResult(short token)
        {
            try
            {
                return _core.GetResult(token);
            }
            finally
            {
                Pool.Return(this);
            }
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        ValueTaskSourceStatus IValueTaskSource<T>.GetStatus(short token)
        {
            return _core.GetStatus(token);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        void IValueTaskSource<T>.OnCompleted(Action<object?> continuation, object? state, short token, ValueTaskSourceOnCompletedFlags flags)
        {
            _core.OnCompleted(continuation, state, token, flags);
        }
        
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void SetResult(T result)
        {
            _core.SetResult(result);
        }
        
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void SetException(Exception result) => _core.SetException(result);
    }
}

[arontsang]

@ralfbiedert As to the external dependency. The only external dependency in this code (non BCL code) is in Microsoft.Extensions.ObjectPool.

We have two options

1. Suck it up as a dependency, it practically BCL.
2. Implement a simple ObjectPool based on ConcurrentQueue<RecyclableTaskCompletionSource<T>>

I personally vote for 2, since its not particularly complex.

[arontsang]

BCL only implementation using ConcurrentQueue

also found that GcHandle was taking ~20-30ns to build...

Method	Mean	Error	StdDev	Ratio	Allocated	Alloc Ratio
Baseline	146.32 ns	1.471 ns	1.229 ns	1.00	128 B	1.00
Fast	77.89 ns	0.576 ns	0.511 ns	0.53	-	0.00
NativeMethodOverhead	41.55 ns	0.183 ns	0.162 ns	0.28	-	0.00

public static class Overhead
{
    private static readonly IntPtr CallbackPointer;

    static Overhead()
    {
        
        unsafe
        {
            delegate* unmanaged[Cdecl] <IntPtr , IntPtr, void> callback = &Callback;
            CallbackPointer = (IntPtr)callback;
        }
    }
    
    
    [MethodImpl(MethodImplOptions.AggressiveOptimization)]
    public static void TestAsyncOverhead(this AsyncService service, ulong x, ulong ms)
    {
        var ac = new AsyncCallbackCommonNative {
            _ptr = CallbackPointer,
            _ts = 0,
        };
        Interop.async_service_return_after_ms(service.Context, x, ms, ac);
        // Do nothing to test overhead
    }
    
    [UnmanagedCallersOnly(CallConvs = [typeof(CallConvCdecl)])]
    static void Callback(IntPtr data, IntPtr callbackId)
    {
         // Do nothing to test overhead
    }
}

public static class Fast
{
    [MethodImpl(MethodImplOptions.AggressiveOptimization)]
    public static ValueTask<ulong> ReturnAfterMsFast(this AsyncService service, ulong x, ulong ms)
    {
        return ReturnAfterMsHelper.ReturnAfterMsFast(service, x, ms);
    }

    private static class ReturnAfterMsHelper
    {
        private static readonly IntPtr CallbackPointer;


        static ReturnAfterMsHelper()
        {
            unsafe
            {
                delegate* unmanaged[Cdecl] <IntPtr, nint, void> callback = &Callback;
                CallbackPointer = (IntPtr)callback;
            }
        }
        
        
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        public static ValueTask<ulong> ReturnAfterMsFast(AsyncService service, ulong x, ulong ms)
        {
            var tcs = RecyclableTaskCompletionSource<ulong>.Get();
            var ac = new AsyncCallbackCommonNative {
                _ptr = CallbackPointer,
                _ts = tcs.Id,
            };

            Interop.async_service_return_after_ms(service.Context, x, ms, ac);
            
            

            return tcs.GetTask();
        }
        
        [UnmanagedCallersOnly(CallConvs = [typeof(CallConvCdecl)])]
        static void Callback(IntPtr data, IntPtr callbackId)
        {
            
            var tcs = RecyclableTaskCompletionSource<ulong>.GetById(unchecked((int)callbackId));
            Span<ResultU64Error.Unmanaged> unmanaged;
            unsafe
            {
                unmanaged = new Span<ResultU64Error.Unmanaged>((void*)data, 1);
            }
            var managed = unmanaged[0].ToManaged();
            if (managed.IsOk)
            {
                tcs.SetResult(managed.AsOk());
            }
            else
            {
                tcs.SetException(new InteropException());
            }
        }
    }
    
    internal sealed class RecyclableTaskCompletionSource<T> : IValueTaskSource<T>
    {
        private static readonly ConcurrentDictionary<int, RecyclableTaskCompletionSource<T>> Registered = new();
        private static int NextId = 0;
        
        internal static readonly ConcurrentQueue<RecyclableTaskCompletionSource<T>> Pool = new();

        public static RecyclableTaskCompletionSource<T> GetById(int id) => Registered[id];
        
        public int Id { get; }
        
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        public static RecyclableTaskCompletionSource<T> Get()
        {
            if (Pool.TryDequeue(out var recyclableTaskCompletionSource))
                return recyclableTaskCompletionSource;
            return new RecyclableTaskCompletionSource<T>();
        }

        public RecyclableTaskCompletionSource()
        {
            // By keeping Id sequential, we limit the amount
            // of mod division we need to do.
            // Speeding up the lookup of TCS.
            Id = Interlocked.Increment(ref NextId) - 1;
            Registered[Id] = this;
        }

        private ManualResetValueTaskSourceCore<T> _core = new();
        
        internal void Reset()
        {
            _core.Reset();
        }
        
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        internal ValueTask<T> GetTask() => new(this, _core.Version);
        
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        T IValueTaskSource<T>.GetResult(short token)
        {
            try
            {
                return _core.GetResult(token);
            }
            finally
            {
                _core.Reset();
                Pool.Enqueue(this);
            }
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        ValueTaskSourceStatus IValueTaskSource<T>.GetStatus(short token)
        {
            return _core.GetStatus(token);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        void IValueTaskSource<T>.OnCompleted(Action<object?> continuation, object? state, short token, ValueTaskSourceOnCompletedFlags flags)
        {
            _core.OnCompleted(continuation, state, token, flags);
        }
        
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void SetResult(T result)
        {
            _core.SetResult(result);
        }
        
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void SetException(Exception result) => _core.SetException(result);
    }
}

[arontsang]

After extensive testing, I feel that I've reached the very limits of how fast I can actually make this.

With the AsyncService This handling ~10ns
The P/Invoke followed by the reverse P/Invoke costing ~28ns
futures::executor::block_on costing 7ns
Dequeue TCS / reset / return to pool costing ~15ns

Method	Mean	Error	StdDev	Ratio	Allocated	Alloc Ratio
Baseline	150.393 ns	1.1664 ns	1.0340 ns	1.00	128 B	1.00
Fast	77.795 ns	1.5057 ns	1.5462 ns	0.52	-	0.00
AsyncOverSync	23.505 ns	0.2023 ns	0.1892 ns	0.16	-	0.00
SyncOverSync	16.400 ns	0.1312 ns	0.1024 ns	0.11	-	0.00
SyncOverSyncManual	17.503 ns	0.2325 ns	0.1942 ns	0.12	-	0.00
SyncOverSyncManualWithoutService	3.569 ns	0.0952 ns	0.0890 ns	0.02	-	0.00
SyncCallback	29.570 ns	0.4679 ns	0.4147 ns	0.20	-	0.00

I don't have a huge amount of budget to optimize. I count ~10ns where I could conceivably improve performance.

I'll try to turn my results into a PR.

Regards

PS. See https://github.com/arontsang/benchmark-interoptopus for my benchmark tests.

[ralfbiedert]

Wow, thanks a lot for the thorough investigation! Let me have a look at this during next week and get back to you!

One immediate thing, could you add a sentence describing what each option (Fast, AsyncOverSync, ...) does relative to each other, and what has been measured / not measured.

[ralfbiedert]

Sorry for the delay, things have been a bit hectic.

We had some initial conversations around this, the main question is what the maintenance and UX impact of this would be. The C# backend would have to know about emitting this, and C# users would have to depend on that library (this won't become standard anytime soon, right?)

In that case this should be an optional feature, which probably means some more infra would be needed to nicely encapsulate it.

We definitely want faster async calls, but implementing this might be a bit further down the road. If .NET got a ValueTaskCompletionSource out of the box though this could be made the default.

[arontsang]

Hey @ralfbiedert
I've added a PR #174
This just adds the code to emit the trampoline required for the fast path.

I'll leave how we want to integrate that code up to you.

Possibly emitting a separate method

public class Service
{
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        public Task<ulong> ReturnAfterMs(ulong x, ulong ms)
        {
            
        }

        /// Some comment to describe the gotchas etc with using ValueAsync
        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        public Task<ulong> ReturnAfterMsValueAsync(ulong x, ulong ms)
        {
             throw new NotImplementedException();
        }
}

[arontsang]

@ralfbiedert Also, we are unlikely to be able to use a generic ValueTask pool, since the pooling behaviour is going to have to be matched to our use case. The lifetime of the IValueTaskSource specifically.

That is why microsoft ships the ManualResetValueTaskSourceCore class instead of a generic ValueTaskSource pool, at least in my opinion.