Skip to content

Add ArcBuffer support to Orleans serialization #10066

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.

Sign up for GitHub

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

Merged
ReubenBond merged 1 commit into from
Apr 30, 2026
Merged

Add ArcBuffer support to Orleans serialization #10066

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
35 changes: 18 additions & 17 deletions src/Orleans.Serialization.TestKit/FieldCodecTester.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -971,27 +971,28 @@ protected T RoundTripThroughCodec<T>(T original)
protected object RoundTripThroughUntypedSerializer(object original, out string formattedBitStream)
{
object result;
using (var readerSession = SessionPool.GetSession())
using (var writeSession = SessionPool.GetSession())
using var readerSession = SessionPool.GetSession();
using var writeSession = SessionPool.GetSession();

using var bufferWriter = new ArcBufferWriter();
var writer = Writer.Create(bufferWriter, writeSession);
try
{
var writer = Writer.CreatePooled(writeSession);
try
{
var serializer = ServiceProvider.GetService<Serializer<object>>();
serializer.Serialize(original, ref writer);
var serializer = ServiceProvider.GetService<Serializer<object>>();
serializer.Serialize(original, ref writer);

using var analyzerSession = SessionPool.GetSession();
var output = writer.Output.Slice();
formattedBitStream = BitStreamFormatter.Format(output, analyzerSession);
using var analyzerSession = SessionPool.GetSession();
using var output = bufferWriter.ConsumeSlice(bufferWriter.Length);
var analyzerReader = Reader.Create(output, analyzerSession);
formattedBitStream = BitStreamFormatter.Format(ref analyzerReader);

var reader = Reader.Create(output, readerSession);
var reader = Reader.Create(output, readerSession);

result = serializer.Deserialize(ref reader);
}
finally
{
writer.Dispose();
}
result = serializer.Deserialize(ref reader);
}
finally
{
writer.Dispose();
}

return result;
Expand Down
60 changes: 60 additions & 0 deletions src/Orleans.Serialization/Buffers/Adaptors/BufferSliceReaderInput.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,3 +110,63 @@ internal ReadOnlySpan<byte> GetNext()
[DoesNotReturn]
private static void ThrowInsufficientData() => throw new InvalidOperationException("Insufficient data present in buffer.");
}

/// <summary>
/// Input type for <see cref="Reader{TInput}"/> to support <see cref="ArcBuffer"/> buffers.
/// </summary>
/// <remarks>
/// Initializes a new instance of the <see cref="ArcBufferReaderInput"/> type.
/// </remarks>
/// <param name="slice">The underlying buffer.</param>
public struct ArcBufferReaderInput(in ArcBuffer slice)

Copilot AI Apr 30, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ArcBufferReaderInput is a new public type, but it is not present in the committed API baseline (src/api/Orleans.Serialization/Orleans.Serialization.cs). Update the generated API surface file so this type is captured for API review.

This issue also appears on line 121 of the same file.

Suggested change
public struct ArcBufferReaderInput(in ArcBuffer slice)
internal struct ArcBufferReaderInput(in ArcBuffer slice)

Copilot uses AI. Check for mistakes.
{
private readonly ArcBuffer _slice = slice;
private ArcBufferPage _page = slice.First;
private int _position;

internal readonly ArcBufferPage First => _slice.First;
internal readonly int Position => _position;
internal readonly int Offset => _slice.Offset;
internal readonly int Length => _slice.Length;
internal long PreviousBuffersSize;

internal readonly ArcBufferReaderInput ForkFrom(int position)
{
// Rely on the outer buffer being pinned.
var sliced = _slice.UnsafeSlice(position, Length - position);
return new ArcBufferReaderInput(in sliced);
}

internal ReadOnlySpan<byte> GetNext()
{
Debug.Assert(_position <= Length);
if (_page is not null)
{
if (_page == First)
{
Debug.Assert(_position == 0);
var offset = Offset;
var length = Math.Min(Length, _page.Length - offset);
_position += length;
var result = _page.AsSpan(offset, length);
_page = _page.Next;
return result;
}

if (_position != Length)
{
var length = Math.Min(Length - _position, _page.Length);
_position += length;
var result = _page.AsSpan(0, length);
_page = _page.Next;
return result;
}
}

ThrowInsufficientData();
return default;
}

[DoesNotReturn]
private static void ThrowInsufficientData() => throw new InvalidOperationException("Insufficient data present in buffer.");
}
83 changes: 75 additions & 8 deletions src/Orleans.Serialization/Buffers/Reader.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,6 +194,15 @@ public static class Reader
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static Reader<BufferSliceReaderInput> Create(BufferSlice input, SerializerSession session) => new(new BufferSliceReaderInput(in input), session, 0);

/// <summary>
/// Creates a reader for the provided buffer.
/// </summary>
/// <param name="input">The input.</param>
/// <param name="session">The session.</param>
/// <returns>A new <see cref="Reader{TInput}"/>.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static Reader<ArcBufferReaderInput> Create(ArcBuffer input, SerializerSession session) => new(new ArcBufferReaderInput(in input), session, 0);
Comment on lines +197 to +204

Copilot AI Apr 30, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reader.Create(ArcBuffer, ...) is a new public API, but the committed API baseline file (src/api/Orleans.Serialization/Orleans.Serialization.cs) does not currently list this overload. Update the generated API surface file to include this new method.

Copilot uses AI. Check for mistakes.

/// <summary>
/// Creates a reader for the provided input stream.
/// </summary>
Expand Down Expand Up @@ -267,6 +276,7 @@ public ref struct Reader<TInput>
private readonly static bool IsReadOnlySequenceInput = typeof(TInput) == typeof(ReadOnlySequenceInput);
private readonly static bool IsReaderInput = typeof(ReaderInput).IsAssignableFrom(typeof(TInput));
private readonly static bool IsBufferSliceInput = typeof(TInput) == typeof(BufferSliceReaderInput);
private readonly static bool IsArcBufferInput = typeof(TInput) == typeof(ArcBufferReaderInput);

private ReadOnlySpan<byte> _currentSpan;
private int _bufferPos;
Expand Down Expand Up @@ -297,6 +307,15 @@ internal Reader(TInput input, SerializerSession session, long globalOffset)
_bufferSize = _currentSpan.Length;
_sequenceOffset = globalOffset;
}
else if (IsArcBufferInput)
{
_input = input;
ref var slice = ref Unsafe.As<TInput, ArcBufferReaderInput>(ref _input);
_currentSpan = slice.GetNext();
_bufferPos = 0;
_bufferSize = _currentSpan.Length;
_sequenceOffset = globalOffset;
}
else if (IsReaderInput)
{
_input = input;
Expand Down Expand Up @@ -357,6 +376,11 @@ public long Position
var previousBuffersSize = Unsafe.As<TInput, BufferSliceReaderInput>(ref _input).PreviousBuffersSize;
return _sequenceOffset + previousBuffersSize + _bufferPos;
}
else if (IsArcBufferInput)
{
var previousBuffersSize = Unsafe.As<TInput, ArcBufferReaderInput>(ref _input).PreviousBuffersSize;
return _sequenceOffset + previousBuffersSize + _bufferPos;
}
else if (IsSpanInput)
{
return _sequenceOffset + _bufferPos;
Expand Down Expand Up @@ -388,6 +412,10 @@ public long Length
{
return Unsafe.As<TInput, BufferSliceReaderInput>(ref _input).Length;
}
else if (IsArcBufferInput)
{
return Unsafe.As<TInput, ArcBufferReaderInput>(ref _input).Length;
}
else if (IsSpanInput)
{
return _currentSpan.Length;
Expand Down Expand Up @@ -441,6 +469,22 @@ public void Skip(long count)
}
}
}
else if (IsArcBufferInput)
{
var end = Position + count;
while (Position < end)
{
var previousBuffersSize = Unsafe.As<TInput, ArcBufferReaderInput>(ref _input).PreviousBuffersSize;
if (end - previousBuffersSize <= _bufferSize)
{
_bufferPos = (int)(end - previousBuffersSize);
}
else
{
MoveNext();
}
}
}
else if (IsSpanInput)
{
_bufferPos += (int)count;
Expand Down Expand Up @@ -496,6 +540,17 @@ public void ForkFrom(long position, out Reader<TInput> forked)
ThrowInvalidPosition(position, forked.Position);
}
}
else if (IsArcBufferInput)
{
ref var input = ref Unsafe.As<TInput, ArcBufferReaderInput>(ref _input);
var newInput = input.ForkFrom(checked((int)position));
forked = new Reader<TInput>(Unsafe.As<ArcBufferReaderInput, TInput>(ref newInput), Session, position);

if (forked.Position != position)
{
ThrowInvalidPosition(position, forked.Position);
}
}
else if (IsSpanInput)
{
forked = new Reader<TInput>(_currentSpan[(int)position..], Session, position);
Expand Down Expand Up @@ -541,6 +596,10 @@ public void ResumeFrom(long position)
{
// Nothing is required.
}
else if (IsArcBufferInput)
{
// Nothing is required.
}
else if (IsSpanInput)
{
// Nothing is required.
Expand Down Expand Up @@ -598,6 +657,14 @@ private void MoveNext()
_bufferPos = 0;
_bufferSize = _currentSpan.Length;
}
else if (IsArcBufferInput)
{
ref var slice = ref Unsafe.As<TInput, ArcBufferReaderInput>(ref _input);
slice.PreviousBuffersSize += _bufferSize;
_currentSpan = slice.GetNext();
_bufferPos = 0;
_bufferSize = _currentSpan.Length;
}
else if (IsSpanInput)
{
ThrowInsufficientData();
Expand All @@ -615,7 +682,7 @@ private void MoveNext()
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public byte ReadByte()
{
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput)
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput || IsArcBufferInput)
{
var pos = _bufferPos;
if ((uint)pos < (uint)_currentSpan.Length)
Expand Down Expand Up @@ -657,7 +724,7 @@ private static byte ReadByteSlow(ref Reader<TInput> reader)
/// <returns>The <see cref="uint"/> which was read.</returns>
public uint ReadUInt32()
{
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput)
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput || IsArcBufferInput)
{
const int width = 4;
if (_bufferPos + width > _bufferSize)
Expand Down Expand Up @@ -701,7 +768,7 @@ static uint ReadSlower(ref Reader<TInput> r)
/// <returns>The <see cref="ulong"/> which was read.</returns>
public ulong ReadUInt64()
{
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput)
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput || IsArcBufferInput)
{
const int width = 8;
if (_bufferPos + width > _bufferSize)
Expand Down Expand Up @@ -779,7 +846,7 @@ public byte[] ReadBytes(uint count)
}

var bytes = new byte[count];
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput)
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput || IsArcBufferInput)
{
var destination = new Span<byte>(bytes);
ReadBytes(destination);
Expand All @@ -798,7 +865,7 @@ public byte[] ReadBytes(uint count)
/// <param name="destination">The destination.</param>
public void ReadBytes(scoped Span<byte> destination)
{
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput)
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput || IsArcBufferInput)
{
if (_bufferPos + destination.Length <= _bufferSize)
{
Expand Down Expand Up @@ -846,7 +913,7 @@ private void ReadBytesMultiSegment(scoped Span<byte> dest)
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool TryReadBytes(int length, out ReadOnlySpan<byte> bytes)
{
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput)
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput || IsArcBufferInput)
{
if (_bufferPos + length <= _bufferSize)
{
Expand Down Expand Up @@ -879,7 +946,7 @@ public bool TryReadBytes(int length, out ReadOnlySpan<byte> bytes)
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public unsafe uint ReadVarUInt32()
{
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput)
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput || IsArcBufferInput)
{
var pos = _bufferPos;

Expand Down Expand Up @@ -937,7 +1004,7 @@ private uint ReadVarUInt32Slow()
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public ulong ReadVarUInt64()
{
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput)
if (IsReadOnlySequenceInput || IsSpanInput || IsBufferSliceInput || IsArcBufferInput)
{
var pos = _bufferPos;

Expand Down
28 changes: 28 additions & 0 deletions src/Orleans.Serialization/Buffers/Writer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,15 @@ public static class Writer
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static Writer<TBufferWriter> Create<TBufferWriter>(TBufferWriter destination, SerializerSession session) where TBufferWriter : IBufferWriter<byte> => new(destination, session);

/// <summary>
/// Creates a writer which writes to the specified destination.
/// </summary>
/// <param name="destination">The destination.</param>
/// <param name="session">The session.</param>
/// <returns>A new <see cref="Writer{TBufferWriter}"/>.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static Writer<ArcBufferWriterWrapper> Create(ArcBufferWriter destination, SerializerSession session) => new(new(destination), session);

Comment on lines +33 to +41

Copilot AI Apr 30, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Writer.Create(ArcBufferWriter, ...) adds new public API surface in a packable src/ project, but src/api/Orleans.Serialization/Orleans.Serialization.cs does not currently include this overload. Update the generated API surface file to reflect the new API.

Copilot uses AI. Check for mistakes.
/// <summary>
/// Creates a writer which writes to the specified destination.
/// </summary>
Expand Down Expand Up @@ -95,6 +104,25 @@ public static class Writer
public static Writer<PooledBuffer> CreatePooled(SerializerSession session) => new(new PooledBuffer(), session);
}

/// <summary>
/// Wraps an <see cref="ArcBufferWriter"/> for use as a serialization writer target.
/// </summary>
/// <remarks>
/// Disposing a <see cref="Writer{TBufferWriter}"/> over this wrapper does not dispose the underlying <see cref="ArcBufferWriter"/>.
/// </remarks>
/// <param name="bufferWriter">The wrapped buffer writer.</param>
public readonly struct ArcBufferWriterWrapper(ArcBufferWriter bufferWriter) : IBufferWriter<byte>
{
/// <inheritdoc/>
public void Advance(int count) => ((IBufferWriter<byte>)bufferWriter).Advance(count);

/// <inheritdoc/>
public Memory<byte> GetMemory(int sizeHint = 0) => bufferWriter.GetMemory(sizeHint);

/// <inheritdoc/>
public Span<byte> GetSpan(int sizeHint = 0) => bufferWriter.GetSpan(sizeHint);
}
Comment on lines +111 to +124

Copilot AI Apr 30, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ArcBufferWriterWrapper is a new public type, but the committed API baseline in src/api/Orleans.Serialization/Orleans.Serialization.cs does not include it. Update the generated API surface file so this new public type is represented.

Copilot uses AI. Check for mistakes.

/// <summary>
/// Provides functionality for writing to an output stream.
/// </summary>
Expand Down
Loading
Loading