Skip to content

perf: buffer WebGL stream writes until flush #71

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

Draft
kochounoyume wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

perf: buffer WebGL stream writes until flush #71

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
33 changes: 33 additions & 0 deletions src/LocalPrefs.Unity/Assets/Tests/IDBStreamTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
#if UNITY_WEBGL
#nullable enable

using System;
using System.Collections;
using NUnit.Framework;
using UnityEngine.TestTools;

namespace AndanteTribe.IO.Unity.Tests
{
public class IDBStreamTest
{
[UnityTest]
public IEnumerator MultipleWrites_ArePersistedOnDisposeAsync()
{
yield return new ToCoroutineEnumerator(async () =>
{
var path = $"idb-stream-buffered-write-{Guid.NewGuid():N}";
await using (var stream = new IDBStream(path))
{
await stream.WriteAsync(new byte[] { 1, 2 });
await stream.WriteAsync(new byte[] { 3, 4 });
}

var actual = await IDBUtils.ReadAllBytesAsync(path);
Assert.That(actual, Is.EqualTo(new byte[] { 1, 2, 3, 4 }));
await IDBUtils.DeleteAsync(path);
});
}
}
}

#endif
3 changes: 3 additions & 0 deletions src/LocalPrefs.Unity/Assets/Tests/IDBStreamTest.cs.meta
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

22 changes: 21 additions & 1 deletion src/LocalPrefs.Unity/Assets/Tests/LSPrefsTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
#if UNITY_WEBGL
#if UNITY_WEBGL
#nullable enable

using System;
Expand Down Expand Up @@ -200,6 +200,26 @@ public IEnumerator AddAndRemoveMultipleTimes([ValueSource(nameof(s_factories))]
await LocalPrefsTest.AddAndRemoveMultipleTimes(factory);
});
}

[Test]
public void LSStream_MultipleWrites_ArePersistedOnFlush()
{
const string path = "ls-stream-buffered-write";
try
{
using var stream = new LSStream(path);
stream.Write(new byte[] { 1, 2 }, 0, 2);
stream.Write(new byte[] { 3, 4 }, 0, 2);

Assert.That(LSUtils.ReadAllBytes(path), Is.Empty);
stream.Flush();
Assert.That(LSUtils.ReadAllBytes(path), Is.EqualTo(new byte[] { 1, 2, 3, 4 }));
}
finally
{
LSUtils.Delete(path);
}
}
}
}

Expand Down
91 changes: 84 additions & 7 deletions src/LocalPrefs.Unity/Packages/jp.andantetribe.localprefs/Runtime/IDBStream.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
#if UNITY_WEBGL
#if UNITY_WEBGL
#nullable enable

using System;
Expand All @@ -11,12 +11,18 @@ namespace AndanteTribe.IO.Unity
/// <summary>
/// Represents a stream for IndexedDB operations.
/// </summary>
/// <remarks>No multi-threading support because multi-threading is not allowed in the WebGL environment.</remarks>
/// <remarks>
/// No multi-threading support because multi-threading is not allowed in the WebGL environment.
/// Writes are buffered until <see cref="FlushAsync(CancellationToken)"/> or <see cref="DisposeAsync"/> is called.
/// </remarks>
public class IDBStream : Stream
{
private readonly string _path;
private byte[] _buffer = Array.Empty<byte>();
private int _written;
private int _writeVersion;
private bool _isDirty;
private bool _isDisposed;

/// <inheritdoc />
public override bool CanRead => true;
Expand Down Expand Up @@ -46,7 +52,59 @@ public override long Position
/// <inheritdoc />
public override void Flush()
{
// Flush is typically implemented as an empty method to ensure full compatibility with other Stream types.
ThrowIfDisposed();
if (_isDirty)
{
throw new NotSupportedException("Synchronous Flush is not supported in WebGL. Use FlushAsync instead.");
}
}

/// <inheritdoc />
public override async Task FlushAsync(CancellationToken cancellationToken)
{
ThrowIfDisposed();
cancellationToken.ThrowIfCancellationRequested();
if (!_isDirty)
{
return;
}

var flushedVersion = _writeVersion;
await IDBUtils.WriteAllBytesAsync(_path, new ReadOnlyMemory<byte>(_buffer, 0, _written), cancellationToken);
if (_writeVersion == flushedVersion)
{
_isDirty = false;
}
}

/// <inheritdoc />
public override async ValueTask DisposeAsync()
{
if (_isDisposed)
{
return;
}

await FlushAsync(CancellationToken.None);
Dispose();
GC.SuppressFinalize(this);
}

/// <inheritdoc />
protected override void Dispose(bool disposing)
{
if (disposing && !_isDisposed)
{
if (_isDirty)
{
throw new InvalidOperationException("The stream has buffered data. Use DisposeAsync to persist it to IndexedDB.");
}

_buffer = Array.Empty<byte>();
_isDisposed = true;
}

base.Dispose(disposing);
}

/// <inheritdoc />
Expand Down Expand Up @@ -91,26 +149,45 @@ public override Task WriteAsync(byte[] buffer, int offset, int count, Cancellati
{
cancellationToken.ThrowIfCancellationRequested();
WriteBuffer(new ReadOnlySpan<byte>(buffer, offset, count));
return IDBUtils.WriteAllBytesAsync(_path, new ReadOnlyMemory<byte>(_buffer, 0, _written), cancellationToken).AsTask();
return Task.CompletedTask;
}

/// <inheritdoc />
public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
{
cancellationToken.ThrowIfCancellationRequested();
WriteBuffer(buffer.Span);
return IDBUtils.WriteAllBytesAsync(_path, new ReadOnlyMemory<byte>(_buffer, 0, _written), cancellationToken);
return default;
}

private void WriteBuffer(in ReadOnlySpan<byte> value)
{
if (_buffer.Length < _written + value.Length)
ThrowIfDisposed();
if (value.IsEmpty)
{
Array.Resize(ref _buffer, _written + value.Length);
return;
}

var requiredLength = checked(_written + value.Length);
if (_buffer.Length < requiredLength)
{
var doubledLength = _buffer.Length > int.MaxValue / 2 ? int.MaxValue : _buffer.Length * 2;
var newLength = _buffer.Length == 0 ? requiredLength : Math.Max(requiredLength, doubledLength);
Array.Resize(ref _buffer, newLength);
}

value.CopyTo(_buffer.AsSpan()[_written..]);
_written += value.Length;
_writeVersion++;
_isDirty = true;
}

private void ThrowIfDisposed()
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(IDBStream));
}
}
}
}
Expand Down
64 changes: 49 additions & 15 deletions src/LocalPrefs.Unity/Packages/jp.andantetribe.localprefs/Runtime/LSStream.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
#if UNITY_WEBGL
#if UNITY_WEBGL
#nullable enable

using System;
Expand All @@ -12,11 +12,14 @@ namespace AndanteTribe.IO.Unity
/// <summary>
/// Represents a stream that reads from and writes to Local Storage in WebGL builds.
/// </summary>
/// <remarks>Writes are buffered until <see cref="Flush"/> or <see cref="Dispose()"/> is called.</remarks>
public class LSStream : Stream
{
private readonly string _path;
private NativeArray<byte> _buffer;
private int _written;
private bool _isDirty;
private bool _isDisposed;

/// <inheritdoc />
public override bool CanRead => true;
Expand Down Expand Up @@ -46,17 +49,35 @@ public override long Position
/// <inheritdoc />
protected override void Dispose(bool disposing)
{
if (_buffer.IsCreated)
if (disposing && !_isDisposed)
{
_buffer.Dispose();
try
{
Flush();
}
finally
{
if (_buffer.IsCreated)
{
_buffer.Dispose();
}
_isDisposed = true;
}
}
base.Dispose(disposing);
}

/// <inheritdoc />
public override void Flush()
{
// Flush is typically implemented as an empty method to ensure full compatibility with other Stream types.
ThrowIfDisposed();
if (!_isDirty)
{
return;
}

LSUtils.WriteAllBytes(_path, _buffer.AsSpan()[.._written]);
_isDirty = false;
}

/// <inheritdoc />
Expand Down Expand Up @@ -103,7 +124,7 @@ public override int ReadByte()

/// <inheritdoc />
public override void Write(byte[] buffer, int offset, int count) =>
LSUtils.WriteAllBytes(_path, WriteBuffer(new ReadOnlySpan<byte>(buffer, offset, count)));
WriteBuffer(new ReadOnlySpan<byte>(buffer, offset, count));

/// <inheritdoc />
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
Expand All @@ -117,32 +138,45 @@ public override Task WriteAsync(byte[] buffer, int offset, int count, Cancellati
public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
{
cancellationToken.ThrowIfCancellationRequested();
LSUtils.WriteAllBytes(_path, WriteBuffer(buffer.Span));
WriteBuffer(buffer.Span);
return default;
}

private ReadOnlySpan<byte> WriteBuffer(in ReadOnlySpan<byte> value)
private void WriteBuffer(in ReadOnlySpan<byte> value)
{
if (!_buffer.IsCreated && _buffer.Length != 0)
ThrowIfDisposed();
if (value.IsEmpty)
{
throw new ObjectDisposedException(nameof(LSStream));
return;
}
if (_buffer.Length < _written + value.Length)

var requiredLength = checked(_written + value.Length);
if (_buffer.Length < requiredLength)
{
var newBuffer = new NativeArray<byte>(_written + value.Length, Allocator.Persistent);
if (_buffer.Length != 0)
var doubledLength = _buffer.Length > int.MaxValue / 2 ? int.MaxValue : _buffer.Length * 2;
var newLength = _buffer.Length == 0 ? requiredLength : Math.Max(requiredLength, doubledLength);
var newBuffer = new NativeArray<byte>(newLength, Allocator.Persistent);
if (_written != 0)
{
_buffer.CopyTo(newBuffer);
_buffer.AsSpan()[.._written].CopyTo(newBuffer.AsSpan());
_buffer.Dispose();
}
_buffer = newBuffer;
}

value.CopyTo(_buffer.AsSpan()[_written..]);
_written += value.Length;
return _buffer.AsSpan()[.._written];
_isDirty = true;
}

private void ThrowIfDisposed()
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(LSStream));
}
}
}
}

#endif
#endif