Azure Blob storage: pooled reads, streaming serializer, buffered writes #9879
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,7 @@ | ||
| using System; | ||
| using System.IO; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using Microsoft.Extensions.DependencyInjection; | ||
| using Microsoft.Extensions.Options; | ||
| using Orleans.Runtime; | ||
|
|
@@ -27,6 +30,32 @@ public interface IGrainStorageSerializer | |
| T Deserialize<T>(BinaryData input); | ||
| } | ||
|
|
||
| #nullable enable | ||
| /// <summary> | ||
| /// Optional stream-based serializer for grain state. | ||
| /// </summary> | ||
| public interface IGrainStorageStreamingSerializer | ||
|
Contributor
Author
There was a problem hiding this comment. @ReubenBond said on Discord he prefers something like this for the abstraction. /// <summary>
/// Optional stream-based serializer for grain state.
/// </summary>
public interface IGrainStorageStreamingSerializer
{
/// <summary>
/// Serializes the object input to a stream.
/// </summary>
/// <param name="input">The object to serialize.</param>
/// <param name="destination">The destination buffer writer.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <typeparam name="T">The input type.</typeparam>
ValueTask SerializeAsync<T>(T input, IBufferWriter<byte> destination, CancellationToken cancellationToken = default);
/// <summary>
/// Deserializes the provided data from a stream.
/// </summary>
/// <param name="input">The input byte sequence.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <typeparam name="T">The output type.</typeparam>
/// <returns>The deserialized object, or null.</returns>
ValueTask<T?> DeserializeAsync<T>(ReadOnlySequence<byte> input, CancellationToken cancellationToken = default);
}For the data providers where
Contributor
Author
There was a problem hiding this comment. For serialization, we can use ready build adapters in Orleans: public static class GrainStorageStreamingSerializerExtensions
{
/// <summary>
/// Serializes the object input to a stream.
/// </summary>
/// <param name="input">The object to serialize.</param>
/// <param name="destination">The destination stream.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <typeparam name="T">The input type.</typeparam>
public static ValueTask SerializeAsync<T>(this IGrainStorageStreamingSerializer serializer, T input, Stream destination, CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(serializer);
ArgumentNullException.ThrowIfNull(destination);
if (destination is MemoryStream memoryStream)
{
return serializer.SerializeAsync(input, new MemoryStreamBufferWriter(memoryStream), cancellationToken);
}
else
{
return serializer.SerializeAsync(input, new ArrayStreamBufferWriter(destination), cancellationToken);
}
}
}For There was a problem hiding this comment. The naming feels off, now you've switched from stream, maybe change to My gut is telling me, drop the ValueTask, by this point all IO should be done, and Task/Await is just overhead 99% of the time, unless I am missing why you would need...
Contributor
Author
There was a problem hiding this comment.
I hope IO is not done at this point, no, since that means data has been loaded into memory, which was the main problem I was trying to avoid. With large datasets, e.g., blobs, that leads to more GC churn. So there need to be support for streaming data from blob storage to the serializer and then to objects. Calling the interface There was a problem hiding this comment. I was thinking about the IO and buffering in the serializer implementations ( Note: overhead here is tiny — pure micro-optimization territory! The overall solution I was getting at earlier on Discord before I dropped off 😅 has been implemented at the calling level (in the provider) rather than inside the serializer. // In AzureBlobGrainStorage.UploadSerializedStateBufferedAsync<T>
var bufferStream = PooledBufferStream.Rent();
try
{
// Serialize: sync write to the pooled stream (no real await needed inside most impls)
await streamSerializer.SerializeAsync(value, bufferStream).ConfigureAwait(false);
bufferStream.Position = 0;
// Actual IO: upload from the pooled stream
return await blob.UploadAsync(bufferStream, options).ConfigureAwait(false);
}
finally
{
PooledBufferStream.Return(bufferStream);
}In JsonGrainStorageSerializer.SerializeAsync (similar for Orleans serializer): public ValueTask SerializeAsync<T>(T value, Stream destination, CancellationToken ct = default)
{
ct.ThrowIfCancellationRequested();
_orleansJsonSerializer.Serialize(value, typeof(T), destination); // sync write
return ValueTask.CompletedTask; // no suspension
}The await on serialize is still basically zero-cost (completed ValueTask), but unnecessary.
Contributor
Author
There was a problem hiding this comment. Ahh I see. If there is no need for asynchrony in the serializer implementations, then dropping ValueTask is fine. Looks like there are no asynchronous methods on ReadOnlySequence nor on IBufferWriter, so probably a good indicator ValueTask is not needed? |
||
| { | ||
| /// <summary> | ||
| /// Serializes the object input to a stream. | ||
| /// </summary> | ||
| /// <param name="input">The object to serialize.</param> | ||
| /// <param name="destination">The destination stream.</param> | ||
| /// <param name="cancellationToken">The cancellation token.</param> | ||
| /// <typeparam name="T">The input type.</typeparam> | ||
| ValueTask SerializeAsync<T>(T input, Stream destination, CancellationToken cancellationToken = default); | ||
|
|
||
| /// <summary> | ||
| /// Deserializes the provided data from a stream. | ||
| /// </summary> | ||
| /// <param name="input">The input stream.</param> | ||
| /// <param name="cancellationToken">The cancellation token.</param> | ||
| /// <typeparam name="T">The output type.</typeparam> | ||
| /// <returns>The deserialized object.</returns> | ||
| ValueTask<T?> DeserializeAsync<T>(Stream input, CancellationToken cancellationToken = default); | ||
| } | ||
| #nullable restore | ||
|
|
||
| /// <summary> | ||
| /// Extensions for <see cref="IGrainStorageSerializer"/>. | ||
| /// </summary> | ||
|
|
@@ -76,7 +105,7 @@ public void PostConfigure(string name, TOptions options) | |
| { | ||
| if (options.GrainStorageSerializer == default) | ||
| { | ||
| // First, try to get a IGrainStorageSerializer that was registered with | ||
| // the same name as the storage provider | ||
| // If none is found, fallback to system wide default | ||
| options.GrainStorageSerializer = _serviceProvider.GetKeyedService<IGrainStorageSerializer>(name) ?? _serviceProvider.GetRequiredService<IGrainStorageSerializer>(); | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.