THRIFT-5133: Use ArrayPool when reading and writing strings to improve performance
Client: netstd
Patch: Mikel Blanchard
This closes #2057
diff --git a/lib/netstd/Thrift/Protocol/TCompactProtocol.cs b/lib/netstd/Thrift/Protocol/TCompactProtocol.cs
index a8a46f2..bb531f4 100644
--- a/lib/netstd/Thrift/Protocol/TCompactProtocol.cs
+++ b/lib/netstd/Thrift/Protocol/TCompactProtocol.cs
@@ -16,6 +16,7 @@
// under the License.
using System;
+using System.Buffers;
using System.Buffers.Binary;
using System.Collections.Generic;
using System.Diagnostics;
@@ -66,7 +67,7 @@
// minimize memory allocations by means of an preallocated bytes buffer
// The value of 128 is arbitrarily chosen, the required minimum size must be sizeof(long)
- private byte[] PreAllocatedBuffer = new byte[128];
+ private readonly byte[] PreAllocatedBuffer = new byte[128];
private struct VarInt
{
@@ -411,11 +412,19 @@
return;
}
- var bytes = Encoding.UTF8.GetBytes(str);
+ var buf = ArrayPool<byte>.Shared.Rent(Encoding.UTF8.GetByteCount(str));
+ try
+ {
+ var numberOfBytes = Encoding.UTF8.GetBytes(str, 0, str.Length, buf, 0);
- Int32ToVarInt((uint) bytes.Length, ref PreAllocatedVarInt);
- await Trans.WriteAsync(PreAllocatedVarInt.bytes, 0, PreAllocatedVarInt.count, cancellationToken);
- await Trans.WriteAsync(bytes, 0, bytes.Length, cancellationToken);
+ Int32ToVarInt((uint)numberOfBytes, ref PreAllocatedVarInt);
+ await Trans.WriteAsync(PreAllocatedVarInt.bytes, 0, PreAllocatedVarInt.count, cancellationToken);
+ await Trans.WriteAsync(buf, 0, numberOfBytes, cancellationToken);
+ }
+ finally
+ {
+ ArrayPool<byte>.Shared.Return(buf);
+ }
}
public override async Task WriteBinaryAsync(byte[] bytes, CancellationToken cancellationToken)
@@ -706,9 +715,17 @@
}
Transport.CheckReadBytesAvailable(length);
- var buf = new byte[length];
- await Trans.ReadAllAsync(buf, 0, length, cancellationToken);
- return Encoding.UTF8.GetString(buf, 0, length);
+
+ var buf = ArrayPool<byte>.Shared.Rent(length);
+ try
+ {
+ await Trans.ReadAllAsync(buf, 0, length, cancellationToken);
+ return Encoding.UTF8.GetString(buf, 0, length);
+ }
+ finally
+ {
+ ArrayPool<byte>.Shared.Return(buf);
+ }
}
public override async ValueTask<byte[]> ReadBinaryAsync(CancellationToken cancellationToken)
@@ -717,7 +734,7 @@
var length = (int) await ReadVarInt32Async(cancellationToken);
if (length == 0)
{
- return new byte[0];
+ return Array.Empty<byte>();
}
// read data
diff --git a/lib/netstd/Thrift/Thrift.csproj b/lib/netstd/Thrift/Thrift.csproj
index e40db00..b63a12a 100644
--- a/lib/netstd/Thrift/Thrift.csproj
+++ b/lib/netstd/Thrift/Thrift.csproj
@@ -19,7 +19,7 @@
-->
<PropertyGroup>
- <TargetFramework>netstandard2.0</TargetFramework>
+ <TargetFrameworks>netstandard2.1;netstandard2.0</TargetFrameworks>
<AssemblyName>Thrift</AssemblyName>
<PackageId>Thrift</PackageId>
<AutoGenerateBindingRedirects>true</AutoGenerateBindingRedirects>
diff --git a/lib/netstd/Thrift/Transport/Client/THttpTransport.cs b/lib/netstd/Thrift/Transport/Client/THttpTransport.cs
index bbd94fa..0790cc8 100644
--- a/lib/netstd/Thrift/Transport/Client/THttpTransport.cs
+++ b/lib/netstd/Thrift/Transport/Client/THttpTransport.cs
@@ -109,7 +109,11 @@
try
{
+#if NETSTANDARD2_1
+ var ret = await _inputStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
+#else
var ret = await _inputStream.ReadAsync(buffer, offset, length, cancellationToken);
+#endif
if (ret == -1)
{
throw new TTransportException(TTransportException.ExceptionType.EndOfFile, "No more data available");
diff --git a/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs b/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs
index f7f10b7..8dab6a0 100644
--- a/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs
+++ b/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs
@@ -72,7 +72,11 @@
}
CheckReadBytesAvailable(length);
+#if NETSTANDARD2_1
+ var numRead = await PipeStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
+#else
var numRead = await PipeStream.ReadAsync(buffer, offset, length, cancellationToken);
+#endif
CountConsumedMessageBytes(numRead);
return numRead;
}
diff --git a/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs b/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs
index e04b3b3..ccadad0 100644
--- a/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs
+++ b/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs
@@ -14,7 +14,7 @@
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
-
+using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
@@ -82,7 +82,11 @@
"Cannot read from null inputstream");
}
+#if NETSTANDARD2_1
+ return await InputStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
+#else
return await InputStream.ReadAsync(buffer, offset, length, cancellationToken);
+#endif
}
public override async Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken)
diff --git a/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs b/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs
index a8b64c4..4b82cbd 100644
--- a/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs
+++ b/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs
@@ -277,7 +277,11 @@
}
CheckReadBytesAvailable(length);
+#if NETSTANDARD2_1
+ var numBytes = await PipeStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
+#else
var numBytes = await PipeStream.ReadAsync(buffer, offset, length, cancellationToken);
+#endif
CountConsumedMessageBytes(numBytes);
return numBytes;
}