feat: 2110 memory_leak_bugfix

src/Proto.Remote/Endpoints/Endpoint.cs

@@ -375,17 +375,19 @@ private void RejectRemoteDeliver(RemoteDeliver env)
 
     private async Task RunAsync()
     {
+        var waiter = new MultiTaskReuseWaiter<bool>(
+            () => _remoteDelivers.Reader.WaitToReadAsync(CancellationToken),
+            () => _remotePriorityDelivers.Reader.WaitToReadAsync(CancellationToken));
+
         while (!CancellationToken.IsCancellationRequested)
         {
             try
             {
                 var messages = new List<RemoteDeliver>(RemoteConfig.EndpointWriterOptions.EndpointWriterBatchSize);
-
                 while (true)
                 {
-                    var t1 = _remoteDelivers.Reader.WaitToReadAsync(CancellationToken).AsTask();
-                    var t2 = _remotePriorityDelivers.Reader.WaitToReadAsync(CancellationToken).AsTask();
-                    await Task.WhenAny(t1, t2);
+                    await waiter.WaitAnyAsync();
+
                     var i = 0;
                     while (true)
                     {
@@ -571,4 +573,38 @@ private MessageBatch CreateBatch(IReadOnlyCollection<RemoteDeliver> m)
         //     Logger.LogTrace("[{SystemAddress}] Sending {Count} envelopes for {Address}", System.Address, envelopes.Count, Address);
         return batch;
     }
+
+    /// <summary>
+    /// Preserves non completed Tasks between <see cref="WaitAnyAsync"/> calls.
+    /// This is necessary to prevent memory leak https://github.com/asynkron/protoactor-dotnet/issues/2110 
+    /// </summary>
+    class MultiTaskReuseWaiter<T>
+    {
+        private readonly Func<ValueTask<T>>[] _taskFactories;
+        private readonly Task<T>?[] _tasks;
+
+        public MultiTaskReuseWaiter(params Func<ValueTask<T>>[] taskFactories)
+        {
+            _taskFactories = taskFactories;
+            _tasks = new Task<T>?[taskFactories.Length];
+        }
+
+        public async ValueTask<T> WaitAnyAsync()
+        {
+            for (var i = 0; i < _taskFactories.Length; i++)
+            {
+                if (_tasks[i]?.IsCompleted == false)
+                    continue;
+
+                var vt = _taskFactories[i].Invoke();
+                if (vt.IsCompleted)
+                    return await vt;
+
+                _tasks[i] = vt.AsTask();
+            }
+
+            return await await Task.WhenAny(_tasks!);
+        }
+
+    }
 }