diff options
Diffstat (limited to 'src/csharp/Grpc.Core')
19 files changed, 371 insertions, 96 deletions
diff --git a/src/csharp/Grpc.Core/DefaultCallInvoker.cs b/src/csharp/Grpc.Core/DefaultCallInvoker.cs index 1a99e41153..b15aaefcd6 100644 --- a/src/csharp/Grpc.Core/DefaultCallInvoker.cs +++ b/src/csharp/Grpc.Core/DefaultCallInvoker.cs @@ -102,6 +102,7 @@ namespace Grpc.Core return Calls.AsyncDuplexStreamingCall(call); } + /// <summary>Creates call invocation details for given method.</summary> protected virtual CallInvocationDetails<TRequest, TResponse> CreateCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options) where TRequest : class where TResponse : class diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj index 622813fb38..d315e6d667 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.csproj +++ b/src/csharp/Grpc.Core/Grpc.Core.csproj @@ -138,6 +138,8 @@ <Compile Include="Internal\CallError.cs" /> <Compile Include="Logging\LogLevel.cs" /> <Compile Include="Logging\LogLevelFilterLogger.cs" /> + <Compile Include="Internal\RequestCallContextSafeHandle.cs" /> + <Compile Include="Utils\TaskUtils.cs" /> </ItemGroup> <ItemGroup> <None Include="Grpc.Core.nuspec" /> diff --git a/src/csharp/Grpc.Core/GrpcEnvironment.cs b/src/csharp/Grpc.Core/GrpcEnvironment.cs index 3ed2df203d..c57c904eb9 100644 --- a/src/csharp/Grpc.Core/GrpcEnvironment.cs +++ b/src/csharp/Grpc.Core/GrpcEnvironment.cs @@ -59,7 +59,6 @@ namespace Grpc.Core static ILogger logger = new NullLogger(); - readonly object myLock = new object(); readonly GrpcThreadPool threadPool; readonly DebugStats debugStats = new DebugStats(); readonly AtomicCounter cqPickerCounter = new AtomicCounter(); diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index 9abaf1120f..5e61e9ec12 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -52,9 +52,8 @@ namespace Grpc.Core.Internal // Completion of a pending unary response if not null. TaskCompletionSource<TResponse> unaryResponseTcs; - // TODO(jtattermusch): this field doesn't need to be initialized for unary response calls. - // Indicates that response streaming call has finished. - TaskCompletionSource<object> streamingCallFinishedTcs = new TaskCompletionSource<object>(); + // Completion of a streaming response call if not null. + TaskCompletionSource<object> streamingResponseCallFinishedTcs; // TODO(jtattermusch): this field could be lazy-initialized (only if someone requests the response headers). // Response headers set here once received. @@ -198,6 +197,7 @@ namespace Grpc.Core.Internal byte[] payload = UnsafeSerialize(msg); + streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) { call.StartServerStreaming(HandleFinished, payload, metadataArray, GetWriteFlagsForCall()); @@ -219,6 +219,7 @@ namespace Grpc.Core.Internal Initialize(details.Channel.CompletionQueue); + streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) { call.StartDuplexStreaming(HandleFinished, metadataArray); @@ -265,7 +266,7 @@ namespace Grpc.Core.Internal // the halfclose has already been done implicitly, so just return // completed task here. halfcloseRequested = true; - return Task.FromResult<object>(null); + return TaskUtils.CompletedTask; } call.StartSendCloseFromClient(HandleSendFinished); @@ -276,13 +277,13 @@ namespace Grpc.Core.Internal } /// <summary> - /// Get the task that completes once if streaming call finishes with ok status and throws RpcException with given status otherwise. + /// Get the task that completes once if streaming response call finishes with ok status and throws RpcException with given status otherwise. /// </summary> - public Task StreamingCallFinishedTask + public Task StreamingResponseCallFinishedTask { get { - return streamingCallFinishedTcs.Task; + return streamingResponseCallFinishedTcs.Task; } } @@ -529,11 +530,11 @@ namespace Grpc.Core.Internal var status = receivedStatus.Status; if (status.StatusCode != StatusCode.OK) { - streamingCallFinishedTcs.SetException(new RpcException(status)); + streamingResponseCallFinishedTcs.SetException(new RpcException(status)); return; } - streamingCallFinishedTcs.SetResult(null); + streamingResponseCallFinishedTcs.SetResult(null); } } } diff --git a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs index c28a6f64d3..26449ee539 100644 --- a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs @@ -93,21 +93,6 @@ namespace Grpc.Core.Internal return data; } - // Gets data of server_rpc_new completion. - public ServerRpcNew GetServerRpcNew(Server server) - { - var call = Native.grpcsharp_batch_context_server_rpc_new_call(this); - - var method = Marshal.PtrToStringAnsi(Native.grpcsharp_batch_context_server_rpc_new_method(this)); - var host = Marshal.PtrToStringAnsi(Native.grpcsharp_batch_context_server_rpc_new_host(this)); - var deadline = Native.grpcsharp_batch_context_server_rpc_new_deadline(this); - - IntPtr metadataArrayPtr = Native.grpcsharp_batch_context_server_rpc_new_request_metadata(this); - var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr); - - return new ServerRpcNew(server, call, method, host, deadline, metadata); - } - // Gets data of receive_close_on_server completion. public bool GetReceivedCloseOnServerCancelled() { diff --git a/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs b/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs index ad9423ff58..65bf60269a 100644 --- a/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs +++ b/src/csharp/Grpc.Core/Internal/ClientResponseStream.cs @@ -73,7 +73,7 @@ namespace Grpc.Core.Internal if (result == null) { - await call.StreamingCallFinishedTask.ConfigureAwait(false); + await call.StreamingResponseCallFinishedTask.ConfigureAwait(false); return false; } return true; diff --git a/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs b/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs index 628844f242..7e2f0e9c6c 100644 --- a/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs +++ b/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs @@ -44,6 +44,8 @@ namespace Grpc.Core.Internal internal delegate void BatchCompletionDelegate(bool success, BatchContextSafeHandle ctx); + internal delegate void RequestCallCompletionDelegate(bool success, RequestCallContextSafeHandle ctx); + internal class CompletionRegistry { static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<CompletionRegistry>(); @@ -68,6 +70,12 @@ namespace Grpc.Core.Internal Register(ctx.Handle, opCallback); } + public void RegisterRequestCallCompletion(RequestCallContextSafeHandle ctx, RequestCallCompletionDelegate callback) + { + OpCompletionDelegate opCallback = ((success) => HandleRequestCallCompletion(success, ctx, callback)); + Register(ctx.Handle, opCallback); + } + public OpCompletionDelegate Extract(IntPtr key) { OpCompletionDelegate value; @@ -84,7 +92,26 @@ namespace Grpc.Core.Internal } catch (Exception e) { - Logger.Error(e, "Exception occured while invoking completion delegate."); + Logger.Error(e, "Exception occured while invoking batch completion delegate."); + } + finally + { + if (ctx != null) + { + ctx.Dispose(); + } + } + } + + private static void HandleRequestCallCompletion(bool success, RequestCallContextSafeHandle ctx, RequestCallCompletionDelegate callback) + { + try + { + callback(success, ctx); + } + catch (Exception e) + { + Logger.Error(e, "Exception occured while invoking request call completion delegate."); } finally { diff --git a/src/csharp/Grpc.Core/Internal/InterceptingCallInvoker.cs b/src/csharp/Grpc.Core/Internal/InterceptingCallInvoker.cs index ef48dc7121..0c63e2092a 100644 --- a/src/csharp/Grpc.Core/Internal/InterceptingCallInvoker.cs +++ b/src/csharp/Grpc.Core/Internal/InterceptingCallInvoker.cs @@ -48,7 +48,7 @@ namespace Grpc.Core.Internal readonly Func<CallOptions, CallOptions> callOptionsInterceptor; /// <summary> - /// Initializes a new instance of the <see cref="Grpc.Core.InterceptingCallInvoker"/> class. + /// Initializes a new instance of the <see cref="Grpc.Core.Internal.InterceptingCallInvoker"/> class. /// </summary> public InterceptingCallInvoker(CallInvoker callInvoker, Func<string, string> hostInterceptor = null, diff --git a/src/csharp/Grpc.Core/Internal/NativeMetadataCredentialsPlugin.cs b/src/csharp/Grpc.Core/Internal/NativeMetadataCredentialsPlugin.cs index 26af6311d5..b3714481eb 100644 --- a/src/csharp/Grpc.Core/Internal/NativeMetadataCredentialsPlugin.cs +++ b/src/csharp/Grpc.Core/Internal/NativeMetadataCredentialsPlugin.cs @@ -78,7 +78,10 @@ namespace Grpc.Core.Internal { var context = new AuthInterceptorContext(Marshal.PtrToStringAnsi(serviceUrlPtr), Marshal.PtrToStringAnsi(methodNamePtr)); - StartGetMetadata(context, callbackPtr, userDataPtr); + // Don't await, we are in a native callback and need to return. + #pragma warning disable 4014 + GetMetadataAsync(context, callbackPtr, userDataPtr); + #pragma warning restore 4014 } catch (Exception e) { @@ -87,7 +90,7 @@ namespace Grpc.Core.Internal } } - private async Task StartGetMetadata(AuthInterceptorContext context, IntPtr callbackPtr, IntPtr userDataPtr) + private async Task GetMetadataAsync(AuthInterceptorContext context, IntPtr callbackPtr, IntPtr userDataPtr) { try { diff --git a/src/csharp/Grpc.Core/Internal/NativeMethods.cs b/src/csharp/Grpc.Core/Internal/NativeMethods.cs index f457c9dbf1..40ba7e30cb 100644 --- a/src/csharp/Grpc.Core/Internal/NativeMethods.cs +++ b/src/csharp/Grpc.Core/Internal/NativeMethods.cs @@ -64,14 +64,17 @@ namespace Grpc.Core.Internal public readonly Delegates.grpcsharp_batch_context_recv_status_on_client_status_delegate grpcsharp_batch_context_recv_status_on_client_status; public readonly Delegates.grpcsharp_batch_context_recv_status_on_client_details_delegate grpcsharp_batch_context_recv_status_on_client_details; public readonly Delegates.grpcsharp_batch_context_recv_status_on_client_trailing_metadata_delegate grpcsharp_batch_context_recv_status_on_client_trailing_metadata; - public readonly Delegates.grpcsharp_batch_context_server_rpc_new_call_delegate grpcsharp_batch_context_server_rpc_new_call; - public readonly Delegates.grpcsharp_batch_context_server_rpc_new_method_delegate grpcsharp_batch_context_server_rpc_new_method; - public readonly Delegates.grpcsharp_batch_context_server_rpc_new_host_delegate grpcsharp_batch_context_server_rpc_new_host; - public readonly Delegates.grpcsharp_batch_context_server_rpc_new_deadline_delegate grpcsharp_batch_context_server_rpc_new_deadline; - public readonly Delegates.grpcsharp_batch_context_server_rpc_new_request_metadata_delegate grpcsharp_batch_context_server_rpc_new_request_metadata; public readonly Delegates.grpcsharp_batch_context_recv_close_on_server_cancelled_delegate grpcsharp_batch_context_recv_close_on_server_cancelled; public readonly Delegates.grpcsharp_batch_context_destroy_delegate grpcsharp_batch_context_destroy; + public readonly Delegates.grpcsharp_request_call_context_create_delegate grpcsharp_request_call_context_create; + public readonly Delegates.grpcsharp_request_call_context_call_delegate grpcsharp_request_call_context_call; + public readonly Delegates.grpcsharp_request_call_context_method_delegate grpcsharp_request_call_context_method; + public readonly Delegates.grpcsharp_request_call_context_host_delegate grpcsharp_request_call_context_host; + public readonly Delegates.grpcsharp_request_call_context_deadline_delegate grpcsharp_request_call_context_deadline; + public readonly Delegates.grpcsharp_request_call_context_request_metadata_delegate grpcsharp_request_call_context_request_metadata; + public readonly Delegates.grpcsharp_request_call_context_destroy_delegate grpcsharp_request_call_context_destroy; + public readonly Delegates.grpcsharp_composite_call_credentials_create_delegate grpcsharp_composite_call_credentials_create; public readonly Delegates.grpcsharp_call_credentials_release_delegate grpcsharp_call_credentials_release; @@ -170,14 +173,17 @@ namespace Grpc.Core.Internal this.grpcsharp_batch_context_recv_status_on_client_status = GetMethodDelegate<Delegates.grpcsharp_batch_context_recv_status_on_client_status_delegate>(library); this.grpcsharp_batch_context_recv_status_on_client_details = GetMethodDelegate<Delegates.grpcsharp_batch_context_recv_status_on_client_details_delegate>(library); this.grpcsharp_batch_context_recv_status_on_client_trailing_metadata = GetMethodDelegate<Delegates.grpcsharp_batch_context_recv_status_on_client_trailing_metadata_delegate>(library); - this.grpcsharp_batch_context_server_rpc_new_call = GetMethodDelegate<Delegates.grpcsharp_batch_context_server_rpc_new_call_delegate>(library); - this.grpcsharp_batch_context_server_rpc_new_method = GetMethodDelegate<Delegates.grpcsharp_batch_context_server_rpc_new_method_delegate>(library); - this.grpcsharp_batch_context_server_rpc_new_host = GetMethodDelegate<Delegates.grpcsharp_batch_context_server_rpc_new_host_delegate>(library); - this.grpcsharp_batch_context_server_rpc_new_deadline = GetMethodDelegate<Delegates.grpcsharp_batch_context_server_rpc_new_deadline_delegate>(library); - this.grpcsharp_batch_context_server_rpc_new_request_metadata = GetMethodDelegate<Delegates.grpcsharp_batch_context_server_rpc_new_request_metadata_delegate>(library); this.grpcsharp_batch_context_recv_close_on_server_cancelled = GetMethodDelegate<Delegates.grpcsharp_batch_context_recv_close_on_server_cancelled_delegate>(library); this.grpcsharp_batch_context_destroy = GetMethodDelegate<Delegates.grpcsharp_batch_context_destroy_delegate>(library); + this.grpcsharp_request_call_context_create = GetMethodDelegate<Delegates.grpcsharp_request_call_context_create_delegate>(library); + this.grpcsharp_request_call_context_call = GetMethodDelegate<Delegates.grpcsharp_request_call_context_call_delegate>(library); + this.grpcsharp_request_call_context_method = GetMethodDelegate<Delegates.grpcsharp_request_call_context_method_delegate>(library); + this.grpcsharp_request_call_context_host = GetMethodDelegate<Delegates.grpcsharp_request_call_context_host_delegate>(library); + this.grpcsharp_request_call_context_deadline = GetMethodDelegate<Delegates.grpcsharp_request_call_context_deadline_delegate>(library); + this.grpcsharp_request_call_context_request_metadata = GetMethodDelegate<Delegates.grpcsharp_request_call_context_request_metadata_delegate>(library); + this.grpcsharp_request_call_context_destroy = GetMethodDelegate<Delegates.grpcsharp_request_call_context_destroy_delegate>(library); + this.grpcsharp_composite_call_credentials_create = GetMethodDelegate<Delegates.grpcsharp_composite_call_credentials_create_delegate>(library); this.grpcsharp_call_credentials_release = GetMethodDelegate<Delegates.grpcsharp_call_credentials_release_delegate>(library); @@ -302,14 +308,17 @@ namespace Grpc.Core.Internal public delegate StatusCode grpcsharp_batch_context_recv_status_on_client_status_delegate(BatchContextSafeHandle ctx); public delegate IntPtr grpcsharp_batch_context_recv_status_on_client_details_delegate(BatchContextSafeHandle ctx); // returns const char* public delegate IntPtr grpcsharp_batch_context_recv_status_on_client_trailing_metadata_delegate(BatchContextSafeHandle ctx); - public delegate CallSafeHandle grpcsharp_batch_context_server_rpc_new_call_delegate(BatchContextSafeHandle ctx); - public delegate IntPtr grpcsharp_batch_context_server_rpc_new_method_delegate(BatchContextSafeHandle ctx); // returns const char* - public delegate IntPtr grpcsharp_batch_context_server_rpc_new_host_delegate(BatchContextSafeHandle ctx); // returns const char* - public delegate Timespec grpcsharp_batch_context_server_rpc_new_deadline_delegate(BatchContextSafeHandle ctx); - public delegate IntPtr grpcsharp_batch_context_server_rpc_new_request_metadata_delegate(BatchContextSafeHandle ctx); public delegate int grpcsharp_batch_context_recv_close_on_server_cancelled_delegate(BatchContextSafeHandle ctx); public delegate void grpcsharp_batch_context_destroy_delegate(IntPtr ctx); + public delegate RequestCallContextSafeHandle grpcsharp_request_call_context_create_delegate(); + public delegate CallSafeHandle grpcsharp_request_call_context_call_delegate(RequestCallContextSafeHandle ctx); + public delegate IntPtr grpcsharp_request_call_context_method_delegate(RequestCallContextSafeHandle ctx); // returns const char* + public delegate IntPtr grpcsharp_request_call_context_host_delegate(RequestCallContextSafeHandle ctx); // returns const char* + public delegate Timespec grpcsharp_request_call_context_deadline_delegate(RequestCallContextSafeHandle ctx); + public delegate IntPtr grpcsharp_request_call_context_request_metadata_delegate(RequestCallContextSafeHandle ctx); + public delegate void grpcsharp_request_call_context_destroy_delegate(IntPtr ctx); + public delegate CallCredentialsSafeHandle grpcsharp_composite_call_credentials_create_delegate(CallCredentialsSafeHandle creds1, CallCredentialsSafeHandle creds2); public delegate void grpcsharp_call_credentials_release_delegate(IntPtr credentials); @@ -393,7 +402,7 @@ namespace Grpc.Core.Internal public delegate int grpcsharp_server_add_insecure_http2_port_delegate(ServerSafeHandle server, string addr); public delegate int grpcsharp_server_add_secure_http2_port_delegate(ServerSafeHandle server, string addr, ServerCredentialsSafeHandle creds); public delegate void grpcsharp_server_start_delegate(ServerSafeHandle server); - public delegate CallError grpcsharp_server_request_call_delegate(ServerSafeHandle server, CompletionQueueSafeHandle cq, BatchContextSafeHandle ctx); + public delegate CallError grpcsharp_server_request_call_delegate(ServerSafeHandle server, CompletionQueueSafeHandle cq, RequestCallContextSafeHandle ctx); public delegate void grpcsharp_server_cancel_all_calls_delegate(ServerSafeHandle server); public delegate void grpcsharp_server_shutdown_and_notify_callback_delegate(ServerSafeHandle server, CompletionQueueSafeHandle cq, BatchContextSafeHandle ctx); public delegate void grpcsharp_server_destroy_delegate(IntPtr server); diff --git a/src/csharp/Grpc.Core/Internal/RequestCallContextSafeHandle.cs b/src/csharp/Grpc.Core/Internal/RequestCallContextSafeHandle.cs new file mode 100644 index 0000000000..ea7819d7b1 --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/RequestCallContextSafeHandle.cs @@ -0,0 +1,85 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Runtime.InteropServices; +using Grpc.Core; + +namespace Grpc.Core.Internal +{ + /// <summary> + /// grpcsharp_request_call_context + /// </summary> + internal class RequestCallContextSafeHandle : SafeHandleZeroIsInvalid + { + static readonly NativeMethods Native = NativeMethods.Get(); + + private RequestCallContextSafeHandle() + { + } + + public static RequestCallContextSafeHandle Create() + { + return Native.grpcsharp_request_call_context_create(); + } + + public IntPtr Handle + { + get + { + return handle; + } + } + + // Gets data of server_rpc_new completion. + public ServerRpcNew GetServerRpcNew(Server server) + { + var call = Native.grpcsharp_request_call_context_call(this); + + var method = Marshal.PtrToStringAnsi(Native.grpcsharp_request_call_context_method(this)); + var host = Marshal.PtrToStringAnsi(Native.grpcsharp_request_call_context_host(this)); + var deadline = Native.grpcsharp_request_call_context_deadline(this); + + IntPtr metadataArrayPtr = Native.grpcsharp_request_call_context_request_metadata(this); + var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr); + + return new ServerRpcNew(server, call, method, host, deadline, metadata); + } + + protected override bool ReleaseHandle() + { + Native.grpcsharp_request_call_context_destroy(handle); + return true; + } + } +} diff --git a/src/csharp/Grpc.Core/Internal/SafeHandleZeroIsInvalid.cs b/src/csharp/Grpc.Core/Internal/SafeHandleZeroIsInvalid.cs index 230faacff6..a637a54f58 100644 --- a/src/csharp/Grpc.Core/Internal/SafeHandleZeroIsInvalid.cs +++ b/src/csharp/Grpc.Core/Internal/SafeHandleZeroIsInvalid.cs @@ -45,10 +45,6 @@ namespace Grpc.Core.Internal { } - public SafeHandleZeroIsInvalid(bool ownsHandle) : base(IntPtr.Zero, ownsHandle) - { - } - public override bool IsInvalid { get @@ -56,11 +52,5 @@ namespace Grpc.Core.Internal return handle == IntPtr.Zero; } } - - protected override bool ReleaseHandle() - { - // handle is not owned. - return true; - } } } diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs index 6a2f520163..600811ddd6 100644 --- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs +++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs @@ -76,7 +76,7 @@ namespace Grpc.Core.Internal Status status; Tuple<TResponse,WriteFlags> responseTuple = null; - var context = HandlerUtils.NewContext(newRpc, asyncCall.Peer, responseStream, asyncCall.CancellationToken); + var context = HandlerUtils.NewContext(newRpc, responseStream, asyncCall.CancellationToken); try { GrpcPreconditions.CheckArgument(await requestStream.MoveNext().ConfigureAwait(false)); @@ -134,7 +134,7 @@ namespace Grpc.Core.Internal var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall); Status status; - var context = HandlerUtils.NewContext(newRpc, asyncCall.Peer, responseStream, asyncCall.CancellationToken); + var context = HandlerUtils.NewContext(newRpc, responseStream, asyncCall.CancellationToken); try { GrpcPreconditions.CheckArgument(await requestStream.MoveNext().ConfigureAwait(false)); @@ -193,7 +193,7 @@ namespace Grpc.Core.Internal Status status; Tuple<TResponse,WriteFlags> responseTuple = null; - var context = HandlerUtils.NewContext(newRpc, asyncCall.Peer, responseStream, asyncCall.CancellationToken); + var context = HandlerUtils.NewContext(newRpc, responseStream, asyncCall.CancellationToken); try { var response = await handler(requestStream, context).ConfigureAwait(false); @@ -250,7 +250,7 @@ namespace Grpc.Core.Internal var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall); Status status; - var context = HandlerUtils.NewContext(newRpc, asyncCall.Peer, responseStream, asyncCall.CancellationToken); + var context = HandlerUtils.NewContext(newRpc, responseStream, asyncCall.CancellationToken); try { await handler(requestStream, responseStream, context).ConfigureAwait(false); @@ -277,20 +277,31 @@ namespace Grpc.Core.Internal } } - internal class NoSuchMethodCallHandler : IServerCallHandler + internal class UnimplementedMethodCallHandler : IServerCallHandler { - public static readonly NoSuchMethodCallHandler Instance = new NoSuchMethodCallHandler(); + public static readonly UnimplementedMethodCallHandler Instance = new UnimplementedMethodCallHandler(); - public async Task HandleCall(ServerRpcNew newRpc, CompletionQueueSafeHandle cq) + DuplexStreamingServerCallHandler<byte[], byte[]> callHandlerImpl; + + public UnimplementedMethodCallHandler() { - // We don't care about the payload type here. - var asyncCall = new AsyncCallServer<byte[], byte[]>( - (payload) => payload, (payload) => payload, newRpc.Server); - - asyncCall.Initialize(newRpc.Call, cq); - var finishedTask = asyncCall.ServerSideCallAsync(); - await asyncCall.SendStatusFromServerAsync(new Status(StatusCode.Unimplemented, ""), Metadata.Empty, null).ConfigureAwait(false); - await finishedTask.ConfigureAwait(false); + var marshaller = new Marshaller<byte[]>((payload) => payload, (payload) => payload); + var method = new Method<byte[], byte[]>(MethodType.DuplexStreaming, "", "", marshaller, marshaller); + this.callHandlerImpl = new DuplexStreamingServerCallHandler<byte[], byte[]>(method, new DuplexStreamingServerMethod<byte[], byte[]>(UnimplementedMethod)); + } + + /// <summary> + /// Handler used for unimplemented method. + /// </summary> + private Task UnimplementedMethod(IAsyncStreamReader<byte[]> requestStream, IServerStreamWriter<byte[]> responseStream, ServerCallContext ctx) + { + ctx.Status = new Status(StatusCode.Unimplemented, ""); + return TaskUtils.CompletedTask; + } + + public Task HandleCall(ServerRpcNew newRpc, CompletionQueueSafeHandle cq) + { + return callHandlerImpl.HandleCall(newRpc, cq); } } @@ -313,13 +324,13 @@ namespace Grpc.Core.Internal return writeOptions != null ? writeOptions.Flags : default(WriteFlags); } - public static ServerCallContext NewContext<TRequest, TResponse>(ServerRpcNew newRpc, string peer, ServerResponseStream<TRequest, TResponse> serverResponseStream, CancellationToken cancellationToken) + public static ServerCallContext NewContext<TRequest, TResponse>(ServerRpcNew newRpc, ServerResponseStream<TRequest, TResponse> serverResponseStream, CancellationToken cancellationToken) where TRequest : class where TResponse : class { DateTime realtimeDeadline = newRpc.Deadline.ToClockType(ClockType.Realtime).ToDateTime(); - return new ServerCallContext(newRpc.Call, newRpc.Method, newRpc.Host, peer, realtimeDeadline, + return new ServerCallContext(newRpc.Call, newRpc.Method, newRpc.Host, realtimeDeadline, newRpc.RequestMetadata, cancellationToken, serverResponseStream.WriteResponseHeadersAsync, serverResponseStream); } } diff --git a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs index 014a8db78f..7d7b838616 100644 --- a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs @@ -85,12 +85,12 @@ namespace Grpc.Core.Internal } } - public void RequestCall(BatchCompletionDelegate callback, CompletionQueueSafeHandle completionQueue) + public void RequestCall(RequestCallCompletionDelegate callback, CompletionQueueSafeHandle completionQueue) { using (completionQueue.NewScope()) { - var ctx = BatchContextSafeHandle.Create(); - completionQueue.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + var ctx = RequestCallContextSafeHandle.Create(); + completionQueue.CompletionRegistry.RegisterRequestCallCompletion(ctx, callback); Native.grpcsharp_server_request_call(this, completionQueue, ctx).CheckOk(); } } diff --git a/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs b/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs index 31e1402849..1553bdd687 100644 --- a/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs +++ b/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs @@ -134,7 +134,11 @@ namespace Grpc.Core.Internal { throw new MissingMethodException(string.Format("The native method \"{0}\" does not exist", methodName)); } - return Marshal.GetDelegateForFunctionPointer(ptr, typeof(T)) as T; +#if NETSTANDARD1_5 + return Marshal.GetDelegateForFunctionPointer<T>(ptr); // non-generic version is obsolete +#else + return Marshal.GetDelegateForFunctionPointer(ptr, typeof(T)) as T; // generic version not available in .NET45 +#endif } /// <summary> diff --git a/src/csharp/Grpc.Core/Metadata.cs b/src/csharp/Grpc.Core/Metadata.cs index 915bec146c..6fc715d6ee 100644 --- a/src/csharp/Grpc.Core/Metadata.cs +++ b/src/csharp/Grpc.Core/Metadata.cs @@ -95,11 +95,18 @@ namespace Grpc.Core #region IList members + + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public int IndexOf(Metadata.Entry item) { return entries.IndexOf(item); } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public void Insert(int index, Metadata.Entry item) { GrpcPreconditions.CheckNotNull(item); @@ -107,12 +114,18 @@ namespace Grpc.Core entries.Insert(index, item); } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public void RemoveAt(int index) { CheckWriteable(); entries.RemoveAt(index); } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public Metadata.Entry this[int index] { get @@ -128,6 +141,9 @@ namespace Grpc.Core } } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public void Add(Metadata.Entry item) { GrpcPreconditions.CheckNotNull(item); @@ -135,48 +151,75 @@ namespace Grpc.Core entries.Add(item); } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public void Add(string key, string value) { Add(new Entry(key, value)); } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public void Add(string key, byte[] valueBytes) { Add(new Entry(key, valueBytes)); } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public void Clear() { CheckWriteable(); entries.Clear(); } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public bool Contains(Metadata.Entry item) { return entries.Contains(item); } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public void CopyTo(Metadata.Entry[] array, int arrayIndex) { entries.CopyTo(array, arrayIndex); } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public int Count { get { return entries.Count; } } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public bool IsReadOnly { get { return readOnly; } } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public bool Remove(Metadata.Entry item) { CheckWriteable(); return entries.Remove(item); } + /// <summary> + /// <see cref="T:IList`1"/> + /// </summary> public IEnumerator<Metadata.Entry> GetEnumerator() { return entries.GetEnumerator(); @@ -221,7 +264,7 @@ namespace Grpc.Core public Entry(string key, byte[] valueBytes) { this.key = NormalizeKey(key); - GrpcPreconditions.CheckArgument(this.key.EndsWith(BinaryHeaderSuffix), + GrpcPreconditions.CheckArgument(HasBinaryHeaderSuffix(this.key), "Key for binary valued metadata entry needs to have suffix indicating binary value."); this.value = null; GrpcPreconditions.CheckNotNull(valueBytes, "valueBytes"); @@ -237,7 +280,7 @@ namespace Grpc.Core public Entry(string key, string value) { this.key = NormalizeKey(key); - GrpcPreconditions.CheckArgument(!this.key.EndsWith(BinaryHeaderSuffix), + GrpcPreconditions.CheckArgument(!HasBinaryHeaderSuffix(this.key), "Key for ASCII valued metadata entry cannot have suffix indicating binary value."); this.value = GrpcPreconditions.CheckNotNull(value, "value"); this.valueBytes = null; @@ -324,7 +367,7 @@ namespace Grpc.Core /// </summary> internal static Entry CreateUnsafe(string key, byte[] valueBytes) { - if (key.EndsWith(BinaryHeaderSuffix)) + if (HasBinaryHeaderSuffix(key)) { return new Entry(key, null, valueBytes); } @@ -338,6 +381,27 @@ namespace Grpc.Core "Metadata entry key not valid. Keys can only contain lowercase alphanumeric characters, underscores and hyphens."); return normalized; } + + /// <summary> + /// Returns <c>true</c> if the key has "-bin" binary header suffix. + /// </summary> + private static bool HasBinaryHeaderSuffix(string key) + { + // We don't use just string.EndsWith because its implementation is extremely slow + // on CoreCLR and we've seen significant differences in gRPC benchmarks caused by it. + // See https://github.com/dotnet/coreclr/issues/5612 + + int len = key.Length; + if (len >= 4 && + key[len - 4] == '-' && + key[len - 3] == 'b' && + key[len - 2] == 'i' && + key[len - 1] == 'n') + { + return true; + } + return false; + } } } } diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs index 3b554e5e87..63c1d9cd00 100644 --- a/src/csharp/Grpc.Core/Server.cs +++ b/src/csharp/Grpc.Core/Server.cs @@ -47,7 +47,7 @@ namespace Grpc.Core /// </summary> public class Server { - const int InitialAllowRpcTokenCountPerCq = 10; + const int DefaultRequestCallTokensPerCq = 2000; static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<Server>(); readonly AtomicCounter activeCallCounter = new AtomicCounter(); @@ -66,7 +66,7 @@ namespace Grpc.Core bool startRequested; volatile bool shutdownRequested; - + int requestCallTokensPerCq = DefaultRequestCallTokensPerCq; /// <summary> /// Creates a new server. @@ -133,6 +133,27 @@ namespace Grpc.Core } /// <summary> + /// Experimental API. Might anytime change without prior notice. + /// Number or calls requested via grpc_server_request_call at any given time for each completion queue. + /// </summary> + public int RequestCallTokensPerCompletionQueue + { + get + { + return requestCallTokensPerCq; + } + set + { + lock (myLock) + { + GrpcPreconditions.CheckState(!startRequested); + GrpcPreconditions.CheckArgument(value > 0); + requestCallTokensPerCq = value; + } + } + } + + /// <summary> /// Starts the server. /// </summary> public void Start() @@ -145,9 +166,7 @@ namespace Grpc.Core handle.Start(); - // Starting with more than one AllowOneRpc tokens can significantly increase - // unary RPC throughput. - for (int i = 0; i < InitialAllowRpcTokenCountPerCq; i++) + for (int i = 0; i < requestCallTokensPerCq; i++) { foreach (var cq in environment.CompletionQueues) { @@ -310,14 +329,14 @@ namespace Grpc.Core /// <summary> /// Selects corresponding handler for given call and handles the call. /// </summary> - private async Task HandleCallAsync(ServerRpcNew newRpc, CompletionQueueSafeHandle cq) + private async Task HandleCallAsync(ServerRpcNew newRpc, CompletionQueueSafeHandle cq, Action continuation) { try { IServerCallHandler callHandler; if (!callHandlers.TryGetValue(newRpc.Method, out callHandler)) { - callHandler = NoSuchMethodCallHandler.Instance; + callHandler = UnimplementedMethodCallHandler.Instance; } await callHandler.HandleCall(newRpc, cq).ConfigureAwait(false); } @@ -325,25 +344,40 @@ namespace Grpc.Core { Logger.Warning(e, "Exception while handling RPC."); } + finally + { + continuation(); + } } /// <summary> /// Handles the native callback. /// </summary> - private void HandleNewServerRpc(bool success, BatchContextSafeHandle ctx, CompletionQueueSafeHandle cq) + private void HandleNewServerRpc(bool success, RequestCallContextSafeHandle ctx, CompletionQueueSafeHandle cq) { - Task.Run(() => AllowOneRpc(cq)); - + bool nextRpcRequested = false; if (success) { - ServerRpcNew newRpc = ctx.GetServerRpcNew(this); + var newRpc = ctx.GetServerRpcNew(this); // after server shutdown, the callback returns with null call if (!newRpc.Call.IsInvalid) { - HandleCallAsync(newRpc, cq); // we don't need to await. + nextRpcRequested = true; + + // Start asynchronous handler for the call. + // Don't await, the continuations will run on gRPC thread pool once triggered + // by cq.Next(). + #pragma warning disable 4014 + HandleCallAsync(newRpc, cq, () => AllowOneRpc(cq)); + #pragma warning restore 4014 } } + + if (!nextRpcRequested) + { + AllowOneRpc(cq); + } } /// <summary> diff --git a/src/csharp/Grpc.Core/ServerCallContext.cs b/src/csharp/Grpc.Core/ServerCallContext.cs index 09a6b882a6..8f28fbc045 100644 --- a/src/csharp/Grpc.Core/ServerCallContext.cs +++ b/src/csharp/Grpc.Core/ServerCallContext.cs @@ -48,7 +48,6 @@ namespace Grpc.Core private readonly CallSafeHandle callHandle; private readonly string method; private readonly string host; - private readonly string peer; private readonly DateTime deadline; private readonly Metadata requestHeaders; private readonly CancellationToken cancellationToken; @@ -58,13 +57,12 @@ namespace Grpc.Core private Func<Metadata, Task> writeHeadersFunc; private IHasWriteOptions writeOptionsHolder; - internal ServerCallContext(CallSafeHandle callHandle, string method, string host, string peer, DateTime deadline, Metadata requestHeaders, CancellationToken cancellationToken, + internal ServerCallContext(CallSafeHandle callHandle, string method, string host, DateTime deadline, Metadata requestHeaders, CancellationToken cancellationToken, Func<Metadata, Task> writeHeadersFunc, IHasWriteOptions writeOptionsHolder) { this.callHandle = callHandle; this.method = method; this.host = host; - this.peer = peer; this.deadline = deadline; this.requestHeaders = requestHeaders; this.cancellationToken = cancellationToken; @@ -115,7 +113,10 @@ namespace Grpc.Core { get { - return this.peer; + // Getting the peer lazily is fine as the native call is guaranteed + // not to be disposed before user-supplied server side handler returns. + // Most users won't need to read this field anyway. + return this.callHandle.GetPeer(); } } diff --git a/src/csharp/Grpc.Core/Utils/TaskUtils.cs b/src/csharp/Grpc.Core/Utils/TaskUtils.cs new file mode 100644 index 0000000000..2cf1144143 --- /dev/null +++ b/src/csharp/Grpc.Core/Utils/TaskUtils.cs @@ -0,0 +1,59 @@ +#region Copyright notice and license + +// Copyright 2015, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#endregion + +using System; +using System.Threading.Tasks; + +namespace Grpc.Core.Utils +{ + /// <summary> + /// Utility methods for task parallel library. + /// </summary> + public static class TaskUtils + { + /// <summary> + /// Framework independent equivalent of <c>Task.CompletedTask</c>. + /// </summary> + public static Task CompletedTask + { + get + { +#if NETSTANDARD1_5 + return Task.CompletedTask; +#else + return Task.FromResult<object>(null); // for .NET45, emulate the functionality +#endif + } + } + } +} |