add serializationScope and refactoring for efficiency

pull/19792/head
Jan Tattermusch 5 years ago
parent 1c177fff1e
commit aaddd42c00
  1. 2
      src/csharp/Grpc.Core.Tests/ContextualMarshallerTest.cs
  2. 15
      src/csharp/Grpc.Core/Internal/AsyncCall.cs
  3. 18
      src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
  4. 5
      src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
  5. 38
      src/csharp/Grpc.Core/Internal/DefaultSerializationContext.cs

@ -52,6 +52,8 @@ namespace Grpc.Core.Tests
} }
if (str == "SERIALIZE_TO_NULL") if (str == "SERIALIZE_TO_NULL")
{ {
// TODO: test for not calling complete Complete() (that resulted in null payload before...)
// TODO: test for calling Complete(null byte array)
return; return;
} }
var bytes = System.Text.Encoding.UTF8.GetBytes(str); var bytes = System.Text.Encoding.UTF8.GetBytes(str);

@ -95,10 +95,10 @@ namespace Grpc.Core.Internal
readingDone = true; readingDone = true;
} }
var payload = UnsafeSerialize(msg); using (var serializationScope = DefaultSerializationContext.GetInitializedThreadLocalScope())
using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers))
{ {
var payload = UnsafeSerialize(msg, serializationScope.Context); // do before metadata array?
var ctx = details.Channel.Environment.BatchContextPool.Lease(); var ctx = details.Channel.Environment.BatchContextPool.Lease();
try try
{ {
@ -160,11 +160,14 @@ namespace Grpc.Core.Internal
halfcloseRequested = true; halfcloseRequested = true;
readingDone = true; readingDone = true;
var payload = UnsafeSerialize(msg); //var payload = UnsafeSerialize(msg);
unaryResponseTcs = new TaskCompletionSource<TResponse>(); unaryResponseTcs = new TaskCompletionSource<TResponse>();
using (var serializationScope = DefaultSerializationContext.GetInitializedThreadLocalScope())
using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers))
{ {
var payload = UnsafeSerialize(msg, serializationScope.Context); // do before metadata array?
call.StartUnary(UnaryResponseClientCallback, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); call.StartUnary(UnaryResponseClientCallback, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags);
callStartedOk = true; callStartedOk = true;
} }
@ -235,11 +238,15 @@ namespace Grpc.Core.Internal
halfcloseRequested = true; halfcloseRequested = true;
var payload = UnsafeSerialize(msg); //var payload = UnsafeSerialize(msg);
streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); streamingResponseCallFinishedTcs = new TaskCompletionSource<object>();
using (var serializationScope = DefaultSerializationContext.GetInitializedThreadLocalScope())
using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers))
{ {
var payload = UnsafeSerialize(msg, serializationScope.Context); // do before metadata array?
call.StartServerStreaming(ReceivedStatusOnClientCallback, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); call.StartServerStreaming(ReceivedStatusOnClientCallback, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags);
callStartedOk = true; callStartedOk = true;
} }

@ -115,8 +115,9 @@ namespace Grpc.Core.Internal
/// </summary> /// </summary>
protected Task SendMessageInternalAsync(TWrite msg, WriteFlags writeFlags) protected Task SendMessageInternalAsync(TWrite msg, WriteFlags writeFlags)
{ {
var payload = UnsafeSerialize(msg); using (var serializationScope = DefaultSerializationContext.GetInitializedThreadLocalScope())
{
var payload = UnsafeSerialize(msg, serializationScope.Context); // do before metadata array?
lock (myLock) lock (myLock)
{ {
GrpcPreconditions.CheckState(started); GrpcPreconditions.CheckState(started);
@ -134,6 +135,7 @@ namespace Grpc.Core.Internal
return streamingWriteTcs.Task; return streamingWriteTcs.Task;
} }
} }
}
/// <summary> /// <summary>
/// Initiates reading a message. Only one read operation can be active at a time. /// Initiates reading a message. Only one read operation can be active at a time.
@ -213,20 +215,12 @@ namespace Grpc.Core.Internal
/// </summary> /// </summary>
protected abstract Task CheckSendAllowedOrEarlyResult(); protected abstract Task CheckSendAllowedOrEarlyResult();
protected SliceBufferSafeHandle UnsafeSerialize(TWrite msg) // runs the serializer, propagating any exceptions being thrown without modifying them
{ protected SliceBufferSafeHandle UnsafeSerialize(TWrite msg, DefaultSerializationContext context)
DefaultSerializationContext context = null;
try
{ {
context = DefaultSerializationContext.GetInitializedThreadLocal();
serializer(msg, context); serializer(msg, context);
return context.GetPayload(); return context.GetPayload();
} }
finally
{
context?.Reset();
}
}
protected Exception TryDeserialize(IBufferReader reader, out TRead msg) protected Exception TryDeserialize(IBufferReader reader, out TRead msg)
{ {

@ -129,7 +129,9 @@ namespace Grpc.Core.Internal
/// </summary> /// </summary>
public Task SendStatusFromServerAsync(Status status, Metadata trailers, ResponseWithFlags? optionalWrite) public Task SendStatusFromServerAsync(Status status, Metadata trailers, ResponseWithFlags? optionalWrite)
{ {
var payload = optionalWrite.HasValue ? UnsafeSerialize(optionalWrite.Value.Response) : null; using (var serializationScope = DefaultSerializationContext.GetInitializedThreadLocalScope())
{
var payload = optionalWrite.HasValue ? UnsafeSerialize(optionalWrite.Value.Response, serializationScope.Context) : null;
var writeFlags = optionalWrite.HasValue ? optionalWrite.Value.WriteFlags : default(WriteFlags); var writeFlags = optionalWrite.HasValue ? optionalWrite.Value.WriteFlags : default(WriteFlags);
lock (myLock) lock (myLock)
@ -153,6 +155,7 @@ namespace Grpc.Core.Internal
return sendStatusFromServerTcs.Task; return sendStatusFromServerTcs.Task;
} }
} }
}
/// <summary> /// <summary>
/// Gets cancellation token that gets cancelled once close completion /// Gets cancellation token that gets cancelled once close completion

@ -29,8 +29,7 @@ namespace Grpc.Core.Internal
new ThreadLocal<DefaultSerializationContext>(() => new DefaultSerializationContext(), false); new ThreadLocal<DefaultSerializationContext>(() => new DefaultSerializationContext(), false);
bool isComplete; bool isComplete;
//byte[] payload; SliceBufferSafeHandle sliceBuffer = SliceBufferSafeHandle.Create();
SliceBufferSafeHandle sliceBuffer;
public DefaultSerializationContext() public DefaultSerializationContext()
{ {
@ -42,12 +41,10 @@ namespace Grpc.Core.Internal
GrpcPreconditions.CheckState(!isComplete); GrpcPreconditions.CheckState(!isComplete);
this.isComplete = true; this.isComplete = true;
GetBufferWriter();
var destSpan = sliceBuffer.GetSpan(payload.Length); var destSpan = sliceBuffer.GetSpan(payload.Length);
payload.AsSpan().CopyTo(destSpan); payload.AsSpan().CopyTo(destSpan);
sliceBuffer.Advance(payload.Length); sliceBuffer.Advance(payload.Length);
sliceBuffer.Complete(); sliceBuffer.Complete();
//this.payload = payload;
} }
/// <summary> /// <summary>
@ -55,11 +52,6 @@ namespace Grpc.Core.Internal
/// </summary> /// </summary>
public override IBufferWriter<byte> GetBufferWriter() public override IBufferWriter<byte> GetBufferWriter()
{ {
if (sliceBuffer == null)
{
// TODO: avoid allocation..
sliceBuffer = SliceBufferSafeHandle.Create();
}
return sliceBuffer; return sliceBuffer;
} }
@ -81,17 +73,35 @@ namespace Grpc.Core.Internal
public void Reset() public void Reset()
{ {
this.isComplete = false; this.isComplete = false;
//this.payload = null; this.sliceBuffer.Reset();
this.sliceBuffer = null; // reset instead...
} }
public static DefaultSerializationContext GetInitializedThreadLocal() // Get a cached thread local instance of deserialization context
// and wrap it in a disposable struct that allows easy resetting
// via "using" statement.
public static UsageScope GetInitializedThreadLocalScope()
{ {
var instance = threadLocalInstance.Value; var instance = threadLocalInstance.Value;
instance.Reset(); return new UsageScope(instance);
return instance;
} }
public struct UsageScope : IDisposable
{
readonly DefaultSerializationContext context;
public UsageScope(DefaultSerializationContext context)
{
this.context = context;
}
public DefaultSerializationContext Context => context;
// TODO: add Serialize method...
public void Dispose()
{
context.Reset();
}
}
} }
} }

Loading…
Cancel
Save