Optimized intermediate buffers with stackalloc, spans and array pools

Author: pl752

src/FirebirdSql.Data.FirebirdClient/Client/Managed/DataProviderStreamWrapper.cs

@@ -15,8 +15,10 @@
 
 //$Authors = Jiri Cincura (jiri@cincura.net)
 
+using System;
 using System.IO;
 using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -36,12 +38,31 @@ public int Read(byte[] buffer, int offset, int count)
 	{
 		return _stream.Read(buffer, offset, count);
 	}
+
+	[MethodImpl(MethodImplOptions.AggressiveInlining)]
+	public int Read(Span<byte> buffer, int offset, int count)
+	{
+		return _stream.Read(buffer[offset..(offset+count)]);
+	}
+
 	[MethodImpl(MethodImplOptions.AggressiveInlining)]
 	public ValueTask<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken = default)
 	{
 		return new ValueTask<int>(_stream.ReadAsync(buffer, offset, count, cancellationToken));
 	}
 
+	[MethodImpl(MethodImplOptions.AggressiveInlining)]
+    public ValueTask<int> ReadAsync(Memory<byte> buffer, int offset, int count, CancellationToken cancellationToken = default)
+    {
+        return _stream.ReadAsync(buffer.Slice(offset, count), cancellationToken);
+    }
+
+	[MethodImpl(MethodImplOptions.AggressiveInlining)]
+	public void Write(ReadOnlySpan<byte> buffer)
+	{
+		_stream.Write(buffer);
+	}
+
 	[MethodImpl(MethodImplOptions.AggressiveInlining)]
 	public void Write(byte[] buffer, int offset, int count)
 	{
@@ -53,6 +74,12 @@ public ValueTask WriteAsync(byte[] buffer, int offset, int count, CancellationTo
 		return new ValueTask(_stream.WriteAsync(buffer, offset, count, cancellationToken));
 	}
 
+	[MethodImpl(MethodImplOptions.AggressiveInlining)]
+    public ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, int offset, int count, CancellationToken cancellationToken = default)
+    {
+        return _stream.WriteAsync(buffer.Slice(offset, count), cancellationToken);
+    }
+
 	[MethodImpl(MethodImplOptions.AggressiveInlining)]
 	public void Flush()
 	{

src/FirebirdSql.Data.FirebirdClient/Client/Managed/FirebirdNetworkHandlingWrapper.cs

@@ -87,6 +87,33 @@ public int Read(byte[] buffer, int offset, int count)
 		var dataLength = ReadFromInputBuffer(buffer, offset, count);
 		return dataLength;
 	}
+
+	public int Read(Span<byte> buffer, int offset, int count)
+	{
+		if (_inputBuffer.Count < count) {
+			var readBuffer = _readBuffer;
+			int read;
+			try {
+				read = _dataProvider.Read(readBuffer, 0, readBuffer.Length);
+			}
+			catch (IOException) {
+				IOFailed = true;
+				throw;
+			}
+			if (read != 0) {
+				if (_decryptor != null) {
+					_decryptor.ProcessBytes(readBuffer, 0, read, readBuffer, 0);
+				}
+				if (_decompressor != null) {
+					read = HandleDecompression(readBuffer, read);
+					readBuffer = _compressionBuffer;
+				}
+				WriteToInputBuffer(readBuffer, read);
+			}
+		}
+		var dataLength = ReadFromInputBuffer(buffer, offset, count);
+		return dataLength;
+	}
 	public async ValueTask<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken = default)
 	{
 		if (_inputBuffer.Count < count)
@@ -120,6 +147,24 @@ public async ValueTask<int> ReadAsync(byte[] buffer, int offset, int count, Canc
 		return dataLength;
 	}
 
+    public async ValueTask<int> ReadAsync(Memory<byte> buffer, int offset, int count, CancellationToken cancellationToken = default)
+    {
+        var rented = new byte[count];
+        try
+        {
+            var read = await ReadAsync(rented, 0, count, cancellationToken).ConfigureAwait(false);
+            rented.AsSpan(0, read).CopyTo(buffer.Span.Slice(offset, read));
+            return read;
+        }
+        finally { }
+    }
+
+	public void Write(ReadOnlySpan<byte> buffer)
+	{
+		foreach (var b in buffer)
+			_outputBuffer.Enqueue(b);
+	}
+
 	public void Write(byte[] buffer, int offset, int count)
 	{
 		for (var i = 0; i < count; i++)
@@ -132,6 +177,14 @@ public ValueTask WriteAsync(byte[] buffer, int offset, int count, CancellationTo
 		return ValueTask.CompletedTask;
 	}
 
+	public ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, int offset, int count, CancellationToken cancellationToken = default)
+	{
+		var span = buffer.Span.Slice(offset, count);
+		foreach (var b in span)
+			_outputBuffer.Enqueue(b);
+		return ValueTask2.CompletedTask;
+	}
+
 	public void Flush()
 	{
 		var buffer = _outputBuffer.ToArray();
@@ -206,6 +259,15 @@ int ReadFromInputBuffer(byte[] buffer, int offset, int count)
 		return read;
 	}
 
+	int ReadFromInputBuffer(Span<byte> buffer, int offset, int count)
+	{
+		var read = Math.Min(count, _inputBuffer.Count);
+		for (var i = 0; i < read; i++) {
+			buffer[offset+i] = _inputBuffer.Dequeue();
+		}
+		return read;
+	}
+
 	void WriteToInputBuffer(byte[] data, int count)
 	{
 		for (var i = 0; i < count; i++)

src/FirebirdSql.Data.FirebirdClient/Client/Managed/IDataProvider.cs

@@ -15,6 +15,7 @@
 
 //$Authors = Jiri Cincura (jiri@cincura.net)
 
+using System;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -23,10 +24,14 @@ namespace FirebirdSql.Data.Client.Managed;
 interface IDataProvider
 {
 	int Read(byte[] buffer, int offset, int count);
+	int Read(Span<byte> buffer, int offset, int count);
 	ValueTask<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken = default);
+	ValueTask<int> ReadAsync(Memory<byte> buffer, int offset, int count, CancellationToken cancellationToken = default);
 
+	void Write(ReadOnlySpan<byte> buffer);
 	void Write(byte[] buffer, int offset, int count);
 	ValueTask WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken = default);
+	ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, int offset, int count, CancellationToken cancellationToken = default);
 
 	void Flush();
 	ValueTask FlushAsync(CancellationToken cancellationToken = default);

src/FirebirdSql.Data.FirebirdClient/Client/Managed/IXdrReader.cs

@@ -27,12 +27,18 @@ namespace FirebirdSql.Data.Client.Managed;
 interface IXdrReader
 {
 	byte[] ReadBytes(byte[] buffer, int count);
+	void ReadBytes(Span<byte> buffer, int count);
+	ValueTask ReadBytesAsync(Memory<byte> buffer, int count, CancellationToken cancellationToken = default);
 	ValueTask<byte[]> ReadBytesAsync(byte[] buffer, int count, CancellationToken cancellationToken = default);
 
 	byte[] ReadOpaque(int length);
+	void ReadOpaque(Span<byte> buffer, int length);
+	ValueTask ReadOpaqueAsync(Memory<byte> buffer, int length, CancellationToken cancellationToken = default);
 	ValueTask<byte[]> ReadOpaqueAsync(int length, CancellationToken cancellationToken = default);
 
 	byte[] ReadBuffer();
+	void ReadBuffer(Span<byte> buffer);
+	ValueTask ReadBufferAsync(Memory<byte> buffer, CancellationToken cancellationToken = default);
 	ValueTask<byte[]> ReadBufferAsync(CancellationToken cancellationToken = default);
 
 	string ReadString();

src/FirebirdSql.Data.FirebirdClient/Client/Managed/IXdrWriter.cs

@@ -28,25 +28,36 @@ interface IXdrWriter
 	void Flush();
 	ValueTask FlushAsync(CancellationToken cancellationToken = default);
 
+	void WriteBytes(ReadOnlySpan<byte> buffer);
 	void WriteBytes(byte[] buffer, int count);
 	ValueTask WriteBytesAsync(byte[] buffer, int count, CancellationToken cancellationToken = default);
 
 	void WriteOpaque(byte[] buffer);
+	void WriteOpaque(ReadOnlySpan<byte> buffer);
+	ValueTask WriteOpaqueAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default);
 	ValueTask WriteOpaqueAsync(byte[] buffer, CancellationToken cancellationToken = default);
 
 	void WriteOpaque(byte[] buffer, int length);
+	void WriteOpaque(ReadOnlySpan<byte> buffer, int length);
 	ValueTask WriteOpaqueAsync(byte[] buffer, int length, CancellationToken cancellationToken = default);
+	ValueTask WriteOpaqueAsync(ReadOnlyMemory<byte> buffer, int length, CancellationToken cancellationToken = default);
 
 	void WriteBuffer(byte[] buffer);
+	void WriteBuffer(ReadOnlySpan<byte> buffer);
+	ValueTask WriteBufferAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default);
 	ValueTask WriteBufferAsync(byte[] buffer, CancellationToken cancellationToken = default);
 
 	void WriteBuffer(byte[] buffer, int length);
 	ValueTask WriteBufferAsync(byte[] buffer, int length, CancellationToken cancellationToken = default);
 
 	void WriteBlobBuffer(byte[] buffer);
+	void WriteBlobBuffer(ReadOnlySpan<byte> buffer);
+	ValueTask WriteBlobBufferAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default);
 	ValueTask WriteBlobBufferAsync(byte[] buffer, CancellationToken cancellationToken = default);
 
 	void WriteTyped(int type, byte[] buffer);
+	void WriteTyped(int type, ReadOnlySpan<byte> buffer);
+	ValueTask WriteTypedAsync(int type, ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default);
 	ValueTask WriteTypedAsync(int type, byte[] buffer, CancellationToken cancellationToken = default);
 
 	void Write(string value);