Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -187,17 +187,24 @@ private async Task ProcessReceives()
{
// Resolve `input` PipeWriter via the IDuplexPipe interface prior to loop start for performance.
var input = Input;

Memory<byte> buffer = default;
if (!_waitForData)
{
// Ensure we have some reasonable amount of buffer space to start.
buffer = input.GetMemory(MinAllocBufferSize);
}

while (true)
{
if (_waitForData)
{
// Wait for data before allocating a buffer.
await _receiver.WaitForDataAsync();
// Data ready, allocate some memory to receive it.
buffer = input.GetMemory(MinAllocBufferSize);
}

// Ensure we have some reasonable amount of buffer space
var buffer = input.GetMemory(MinAllocBufferSize);

var bytesReceived = await _receiver.ReceiveAsync(buffer);

if (bytesReceived == 0)
Expand All @@ -209,6 +216,13 @@ private async Task ProcessReceives()

input.Advance(bytesReceived);

if (!_waitForData)
{
// If not waiting for data allocate another buffer before flushing,
// to reduce reader/writer contention on MemoryPool.
buffer = input.GetMemory(MinAllocBufferSize);
}

var flushTask = input.FlushAsync();

var paused = !flushTask.IsCompleted;
Expand Down