From 4ff0fb1cb5acfc6d2685b398f228e10467065332 Mon Sep 17 00:00:00 2001 From: Roger Johansson Date: Sun, 21 Jan 2024 16:50:05 +0100 Subject: [PATCH] add modulus --- src/Proto.Remote/Endpoints/Endpoint.cs | 24 +++++++++++++++--------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/src/Proto.Remote/Endpoints/Endpoint.cs b/src/Proto.Remote/Endpoints/Endpoint.cs index 1d73fde762..cbe25f00ff 100644 --- a/src/Proto.Remote/Endpoints/Endpoint.cs +++ b/src/Proto.Remote/Endpoints/Endpoint.cs @@ -386,21 +386,27 @@ private async Task RunAsync() var t1 = _remoteDelivers.Reader.WaitToReadAsync(CancellationToken).AsTask(); var t2 = _remotePriorityDelivers.Reader.WaitToReadAsync(CancellationToken).AsTask(); await Task.WhenAny(t1, t2); - + var i = 0; while (true) { var didWrite = false; - if (_remotePriorityDelivers.Reader.TryRead(out var remoteDeliver)) + RemoteDeliver? remoteDeliver; + + //we don´t need complete priority, we need "enough" important messages to get over + if (i++ % 10 == 0) { - messages.Add(remoteDeliver); - didWrite = true; + if (_remotePriorityDelivers.Reader.TryRead(out remoteDeliver)) + { + messages.Add(remoteDeliver); + didWrite = true; + } + + if (messages.Count >= RemoteConfig.EndpointWriterOptions.EndpointWriterBatchSize) + { + break; + } } - if (messages.Count >= RemoteConfig.EndpointWriterOptions.EndpointWriterBatchSize) - { - break; - } - if (_remoteDelivers.Reader.TryRead(out remoteDeliver)) { messages.Add(remoteDeliver);