Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Improve StreamingHub performance #830

Open
wants to merge 20 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
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
@@ -1,6 +1,8 @@
using System;
using System.Buffers;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;

namespace MagicOnion.Internal.Buffers
{
Expand All @@ -11,11 +13,7 @@ internal sealed class ArrayPoolBufferWriter : IBufferWriter<byte>, IDisposable

public static ArrayPoolBufferWriter RentThreadStaticWriter()
{
if (staticInstance == null)
{
staticInstance = new ArrayPoolBufferWriter();
}
staticInstance.Prepare();
(staticInstance ??= new ArrayPoolBufferWriter()).Prepare();

#if DEBUG
var currentInstance = staticInstance;
Expand All @@ -26,17 +24,17 @@ public static ArrayPoolBufferWriter RentThreadStaticWriter()
#endif
}

const int MinimumBufferSize = 32767; // use 32k buffer.
const int PreAllocatedBufferSize = 8192; // use 8k buffer.
const int MinimumBufferSize = PreAllocatedBufferSize / 2;

readonly byte[] preAllocatedBuffer = new byte[PreAllocatedBufferSize];
byte[]? buffer;
int index;

[MethodImpl(MethodImplOptions.AggressiveInlining)]
void Prepare()
{
if (buffer == null)
{
buffer = ArrayPool<byte>.Shared.Rent(MinimumBufferSize);
}
buffer = preAllocatedBuffer;
index = 0;
}

Expand All @@ -49,24 +47,28 @@ void Prepare()

public int FreeCapacity => Capacity - index;

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Advance(int count)
{
if (count < 0) throw new ArgumentException(nameof(count));
index += count;
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public Memory<byte> GetMemory(int sizeHint = 0)
{
CheckAndResizeBuffer(sizeHint);
return buffer.AsMemory(index);
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public Span<byte> GetSpan(int sizeHint = 0)
{
CheckAndResizeBuffer(sizeHint);
return buffer.AsSpan(index);
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
void CheckAndResizeBuffer(int sizeHint)
{
if (buffer == null) throw new ObjectDisposedException(nameof(ArrayPoolBufferWriter));
Expand All @@ -82,32 +84,34 @@ void CheckAndResizeBuffer(int sizeHint)
if (sizeHint > availableSpace)
{
int growBy = Math.Max(sizeHint, buffer.Length);

int newSize = checked(buffer.Length + growBy);

byte[] oldBuffer = buffer;

buffer = ArrayPool<byte>.Shared.Rent(newSize);
oldBuffer.AsSpan(0, index).CopyTo(buffer);

Span<byte> previousBuffer = oldBuffer.AsSpan(0, index);
previousBuffer.CopyTo(buffer);
ArrayPool<byte>.Shared.Return(oldBuffer);
if (oldBuffer != preAllocatedBuffer)
{
ArrayPool<byte>.Shared.Return(oldBuffer);
}
}
}

public void Dispose()
{
if (buffer == null)
if (buffer != preAllocatedBuffer && buffer != null)
{
return;
ArrayPool<byte>.Shared.Return(buffer);
}

ArrayPool<byte>.Shared.Return(buffer);
buffer = null;

#if DEBUG
Debug.Assert(staticInstance is null);
staticInstance = this;
#if NETSTANDARD2_1 || NET6_0_OR_GREATER
Array.Fill<byte>(preAllocatedBuffer, 0xff);
#endif
#endif
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -75,23 +75,23 @@ internal class StreamingHubPayload : StreamingHubPayloadCore
internal class StreamingHubPayloadCore
{
byte[]? buffer;
ReadOnlyMemory<byte>? memory;
int length = -1;

#if DEBUG
public short Version { get; private set; }
#endif

public int Length => memory!.Value.Length;
public ReadOnlySpan<byte> Span => memory!.Value.Span;
public ReadOnlyMemory<byte> Memory => memory!.Value;
public int Length => length;
public ReadOnlySpan<byte> Span => buffer!.AsSpan(0, length);
public ReadOnlyMemory<byte> Memory => buffer!.AsMemory(0, length);

public void Initialize(ReadOnlySpan<byte> data)
{
ThrowIfUsing();

buffer = ArrayPool<byte>.Shared.Rent(data.Length);
length = data.Length;
data.CopyTo(buffer);
memory = buffer.AsMemory(0, (int)data.Length);
}

public void Initialize(ReadOnlySequence<byte> data)
Expand All @@ -100,25 +100,8 @@ public void Initialize(ReadOnlySequence<byte> data)
if (data.Length > int.MaxValue) throw new InvalidOperationException("A body size of StreamingHubPayload must be less than int.MaxValue");

buffer = ArrayPool<byte>.Shared.Rent((int)data.Length);
length = (int)data.Length;
data.CopyTo(buffer);
memory = buffer.AsMemory(0, (int)data.Length);
}

public void Initialize(ReadOnlyMemory<byte> data, bool holdReference)
{
ThrowIfUsing();

if (holdReference)
{
buffer = null;
memory = data;
}
else
{
buffer = ArrayPool<byte>.Shared.Rent((int)data.Length);
data.CopyTo(buffer);
memory = buffer.AsMemory(0, (int)data.Length);
}
}

public void Uninitialize()
Expand All @@ -133,21 +116,18 @@ public void Uninitialize()
ArrayPool<byte>.Shared.Return(buffer);
}

memory = null;
length = -1;
buffer = null;

#if DEBUG
Version++;
#endif
}

#if !UNITY_2021_1_OR_NEWER && !NETSTANDARD2_0 && !NETSTANDARD2_1
[MemberNotNull(nameof(memory))]
#endif
void ThrowIfUninitialized()
{
//Debug.Assert(memory is not null);
if (memory is null)
if (length == -1)
{
throw new InvalidOperationException("A StreamingHubPayload has been already uninitialized.");
}
Expand All @@ -156,7 +136,7 @@ void ThrowIfUninitialized()
void ThrowIfUsing()
{
//Debug.Assert(memory is null);
if (memory is not null)
if (length != -1)
{
throw new InvalidOperationException("A StreamingHubPayload is currently used by other caller.");
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -102,17 +102,6 @@ public StreamingHubPayload RentOrCreate(ReadOnlySpan<byte> data)
return new StreamingHubPayload(payload);
#else
return (StreamingHubPayload)payload;
#endif
}

public StreamingHubPayload RentOrCreate(ReadOnlyMemory<byte> data, bool holdReference)
{
var payload = pool.RentOrCreateCore();
payload.Initialize(data, holdReference);
#if DEBUG
return new StreamingHubPayload(payload);
#else
return (StreamingHubPayload)payload;
#endif
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ namespace MagicOnion.Internal;

internal class StreamingHubPayloadPool
{
const int MaximumRetained = 2 << 7;
const int MaximumRetained = 2 << 10;

readonly ObjectPool<StreamingHubPayloadCore> pool = new DefaultObjectPool<StreamingHubPayloadCore>(new Policy(), MaximumRetained);

Expand Down Expand Up @@ -34,17 +34,6 @@ public StreamingHubPayload RentOrCreate(ReadOnlySpan<byte> data)
#endif
}

public StreamingHubPayload RentOrCreate(ReadOnlyMemory<byte> data, bool holdReference)
{
var payload = pool.Get();
payload.Initialize(data, holdReference);
#if DEBUG
return new StreamingHubPayload(payload);
#else
return (StreamingHubPayload)payload;
#endif
}

public void Return(StreamingHubPayload payload)
{
#if DEBUG
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,7 @@
using System;
using System.Data;
using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
using MessagePack;

namespace MagicOnion.Internal
Expand All @@ -14,6 +17,7 @@ public StreamingHubServerMessageReader(ReadOnlyMemory<byte> data)
this.reader = new MessagePackReader(data);
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public StreamingHubMessageType ReadMessageType()
{
var arrayLength = this.reader.ReadArrayHeader();
Expand All @@ -27,12 +31,20 @@ public StreamingHubMessageType ReadMessageType()
0x01 => StreamingHubMessageType.ClientResultResponseWithError,
0x7e => StreamingHubMessageType.ClientHeartbeat,
0x7f => StreamingHubMessageType.ServerHeartbeatResponse,
var subType => throw new InvalidOperationException($"Unknown client response message: {subType}"),
var subType => ThrowUnknownMessageSubType(subType),
},
_ => throw new InvalidOperationException($"Unknown message format: ArrayLength = {arrayLength}"),
_ => ThrowUnknownMessageFormat(arrayLength),
};

[DoesNotReturn]
static StreamingHubMessageType ThrowUnknownMessageSubType(byte subType)
=> throw new InvalidOperationException($"Unknown client response message: {subType}");
[DoesNotReturn]
static StreamingHubMessageType ThrowUnknownMessageFormat(int arrayLength)
=> throw new InvalidOperationException($"Unknown message format: ArrayLength = {arrayLength}");
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public (int MethodId, ReadOnlyMemory<byte> Body) ReadRequestFireAndForget()
{
// void: [methodId, [argument]]
Expand All @@ -42,6 +54,7 @@ public StreamingHubMessageType ReadMessageType()
return (methodId, data.Slice(consumed));
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public (int MessageId, int MethodId, ReadOnlyMemory<byte> Body) ReadRequest()
{
// T: [messageId, methodId, [argument]]
Expand All @@ -52,6 +65,7 @@ public StreamingHubMessageType ReadMessageType()
return (messageId, methodId, data.Slice(consumed));
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public (Guid ClientResultMessageId, int ClientMethodId, ReadOnlyMemory<byte> Body) ReadClientResultResponse()
{
// T: [0, clientResultMessageId, methodId, result]
Expand All @@ -62,6 +76,7 @@ public StreamingHubMessageType ReadMessageType()
return (clientResultMessageId, clientMethodId, data.Slice(consumed));
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public (Guid ClientResultMessageId, int ClientMethodId, int StatusCode, string Detail, string Message) ReadClientResultResponseForError()
{
// T: [1, clientResultMessageId, methodId, [statusCode, detail, message]]
Expand All @@ -77,6 +92,7 @@ public StreamingHubMessageType ReadMessageType()
return (clientResultMessageId, clientMethodId, statusCode, detail, message);
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public (short Sequence, long ClientSentAt, ReadOnlyMemory<byte> Extra) ReadClientHeartbeat()
{
// [Sequence(int16), ClientSentAt(long), <Extra>]
Expand All @@ -87,6 +103,7 @@ public StreamingHubMessageType ReadMessageType()
return (sequence, clientSentAt, data.Slice((int)reader.Consumed));
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public short ReadServerHeartbeatResponse()
{
// [Sequence(int16), Nil, Nil]
Expand Down
Loading
Loading