diff options
Diffstat (limited to 'src/csharp/Grpc.Core')
22 files changed, 627 insertions, 342 deletions
diff --git a/src/csharp/Grpc.Core/Channel.cs b/src/csharp/Grpc.Core/Channel.cs index b47d810672..d6bfbb7bc4 100644 --- a/src/csharp/Grpc.Core/Channel.cs +++ b/src/csharp/Grpc.Core/Channel.cs @@ -29,6 +29,7 @@ // OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. #endregion using System; +using System.Collections.Generic; using System.Runtime.InteropServices; using System.Threading; using System.Threading.Tasks; @@ -45,33 +46,41 @@ namespace Grpc.Core readonly string target; /// <summary> - /// Creates a channel. + /// Creates a channel that connects to a specific host. + /// Port will default to 80 for an unsecure channel and to 443 a secure channel. /// </summary> - public Channel(string target, Credentials credentials = null, ChannelArgs channelArgs = null) + /// <param name="host">The DNS name of IP address of the host.</param> + /// <param name="credentials">Optional credentials to create a secure channel.</param> + /// <param name="options">Channel options.</param> + public Channel(string host, Credentials credentials = null, IEnumerable<ChannelOption> options = null) { - using (ChannelArgsSafeHandle nativeChannelArgs = CreateNativeChannelArgs(channelArgs)) + using (ChannelArgsSafeHandle nativeChannelArgs = ChannelOptions.CreateChannelArgs(options)) { if (credentials != null) { using (CredentialsSafeHandle nativeCredentials = credentials.ToNativeCredentials()) { - this.handle = ChannelSafeHandle.CreateSecure(nativeCredentials, target, nativeChannelArgs); + this.handle = ChannelSafeHandle.CreateSecure(nativeCredentials, host, nativeChannelArgs); } } else { - this.handle = ChannelSafeHandle.Create(target, nativeChannelArgs); + this.handle = ChannelSafeHandle.Create(host, nativeChannelArgs); } } - this.target = GetOverridenTarget(target, channelArgs); + this.target = GetOverridenTarget(host, options); } - public string Target + /// <summary> + /// Creates a channel that connects to a specific host and port. + /// </summary> + /// <param name="host">DNS name or IP address</param> + /// <param name="port">the port</param> + /// <param name="credentials">Optional credentials to create a secure channel.</param> + /// <param name="options">Channel options.</param> + public Channel(string host, int port, Credentials credentials = null, IEnumerable<ChannelOption> options = null) : + this(string.Format("{0}:{1}", host, port), credentials, options) { - get - { - return this.target; - } } public void Dispose() @@ -80,6 +89,14 @@ namespace Grpc.Core GC.SuppressFinalize(this); } + internal string Target + { + get + { + return target; + } + } + internal ChannelSafeHandle Handle { get @@ -96,22 +113,25 @@ namespace Grpc.Core } } - private static string GetOverridenTarget(string target, ChannelArgs args) + /// <summary> + /// Look for SslTargetNameOverride option and return its value instead of originalTarget + /// if found. + /// </summary> + private static string GetOverridenTarget(string originalTarget, IEnumerable<ChannelOption> options) { - if (args != null && !string.IsNullOrEmpty(args.GetSslTargetNameOverride())) + if (options == null) { - return args.GetSslTargetNameOverride(); + return originalTarget; } - return target; - } - - private static ChannelArgsSafeHandle CreateNativeChannelArgs(ChannelArgs args) - { - if (args == null) + foreach (var option in options) { - return ChannelArgsSafeHandle.CreateNull(); + if (option.Type == ChannelOption.OptionType.String + && option.Name == ChannelOptions.SslTargetNameOverride) + { + return option.StringValue; + } } - return args.ToNativeChannelArgs(); + return originalTarget; } } } diff --git a/src/csharp/Grpc.Core/ChannelArgs.cs b/src/csharp/Grpc.Core/ChannelArgs.cs deleted file mode 100644 index 74ab310e44..0000000000 --- a/src/csharp/Grpc.Core/ChannelArgs.cs +++ /dev/null @@ -1,115 +0,0 @@ -#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.Collections.Generic; -using System.Collections.Immutable; -using System.Runtime.InteropServices; -using System.Threading; -using System.Threading.Tasks; -using Grpc.Core.Internal; - -namespace Grpc.Core -{ - /// <summary> - /// gRPC channel options. - /// </summary> - public class ChannelArgs - { - public const string SslTargetNameOverrideKey = "grpc.ssl_target_name_override"; - - readonly ImmutableDictionary<string, string> stringArgs; - - private ChannelArgs(ImmutableDictionary<string, string> stringArgs) - { - this.stringArgs = stringArgs; - } - - public string GetSslTargetNameOverride() - { - string result; - if (stringArgs.TryGetValue(SslTargetNameOverrideKey, out result)) - { - return result; - } - return null; - } - - public static Builder CreateBuilder() - { - return new Builder(); - } - - public class Builder - { - readonly Dictionary<string, string> stringArgs = new Dictionary<string, string>(); - - // TODO: AddInteger not supported yet. - public Builder AddString(string key, string value) - { - stringArgs.Add(key, value); - return this; - } - - public ChannelArgs Build() - { - return new ChannelArgs(stringArgs.ToImmutableDictionary()); - } - } - - /// <summary> - /// Creates native object for the channel arguments. - /// </summary> - /// <returns>The native channel arguments.</returns> - internal ChannelArgsSafeHandle ToNativeChannelArgs() - { - ChannelArgsSafeHandle nativeArgs = null; - try - { - nativeArgs = ChannelArgsSafeHandle.Create(stringArgs.Count); - int i = 0; - foreach (var entry in stringArgs) - { - nativeArgs.SetString(i, entry.Key, entry.Value); - i++; - } - return nativeArgs; - } - catch (Exception) - { - if (nativeArgs != null) - { - nativeArgs.Dispose(); - } - throw; - } - } - } -} diff --git a/src/csharp/Grpc.Core/ChannelOptions.cs b/src/csharp/Grpc.Core/ChannelOptions.cs new file mode 100644 index 0000000000..bc23bb59b1 --- /dev/null +++ b/src/csharp/Grpc.Core/ChannelOptions.cs @@ -0,0 +1,178 @@ +#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.Collections.Generic; +using System.Collections.Immutable; +using System.Runtime.InteropServices; +using System.Threading; +using System.Threading.Tasks; +using Grpc.Core.Internal; +using Grpc.Core.Utils; + +namespace Grpc.Core +{ + /// <summary> + /// Channel option specified when creating a channel. + /// Corresponds to grpc_channel_args from grpc/grpc.h. + /// </summary> + public sealed class ChannelOption + { + public enum OptionType + { + Integer, + String + } + + private readonly OptionType type; + private readonly string name; + private readonly int intValue; + private readonly string stringValue; + + /// <summary> + /// Creates a channel option with a string value. + /// </summary> + /// <param name="name">Name.</param> + /// <param name="stringValue">String value.</param> + public ChannelOption(string name, string stringValue) + { + this.type = OptionType.String; + this.name = Preconditions.CheckNotNull(name); + this.stringValue = Preconditions.CheckNotNull(stringValue); + } + + /// <summary> + /// Creates a channel option with an integer value. + /// </summary> + /// <param name="name">Name.</param> + /// <param name="stringValue">String value.</param> + public ChannelOption(string name, int intValue) + { + this.type = OptionType.Integer; + this.name = Preconditions.CheckNotNull(name); + this.intValue = intValue; + } + + public OptionType Type + { + get + { + return type; + } + } + + public string Name + { + get + { + return name; + } + } + + public int IntValue + { + get + { + Preconditions.CheckState(type == OptionType.Integer); + return intValue; + } + } + + public string StringValue + { + get + { + Preconditions.CheckState(type == OptionType.String); + return stringValue; + } + } + } + + public static class ChannelOptions + { + // Override SSL target check. Only to be used for testing. + public const string SslTargetNameOverride = "grpc.ssl_target_name_override"; + + // Enable census for tracing and stats collection + public const string Census = "grpc.census"; + + // Maximum number of concurrent incoming streams to allow on a http2 connection + public const string MaxConcurrentStreams = "grpc.max_concurrent_streams"; + + // Maximum message length that the channel can receive + public const string MaxMessageLength = "grpc.max_message_length"; + + // Initial sequence number for http2 transports + public const string Http2InitialSequenceNumber = "grpc.http2.initial_sequence_number"; + + /// <summary> + /// Creates native object for a collection of channel options. + /// </summary> + /// <returns>The native channel arguments.</returns> + internal static ChannelArgsSafeHandle CreateChannelArgs(IEnumerable<ChannelOption> options) + { + if (options == null) + { + return ChannelArgsSafeHandle.CreateNull(); + } + var optionList = new List<ChannelOption>(options); // It's better to do defensive copy + ChannelArgsSafeHandle nativeArgs = null; + try + { + nativeArgs = ChannelArgsSafeHandle.Create(optionList.Count); + for (int i = 0; i < optionList.Count; i++) + { + var option = optionList[i]; + if (option.Type == ChannelOption.OptionType.Integer) + { + nativeArgs.SetInteger(i, option.Name, option.IntValue); + } + else if (option.Type == ChannelOption.OptionType.String) + { + nativeArgs.SetString(i, option.Name, option.StringValue); + } + else + { + throw new InvalidOperationException("Unknown option type"); + } + } + return nativeArgs; + } + catch (Exception) + { + if (nativeArgs != null) + { + nativeArgs.Dispose(); + } + throw; + } + } + } +} diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj index fe2d446a35..a36a6a5acc 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.csproj +++ b/src/csharp/Grpc.Core/Grpc.Core.csproj @@ -5,8 +5,6 @@ <PropertyGroup> <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration> <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform> - <ProductVersion>8.0.30703</ProductVersion> - <SchemaVersion>2.0</SchemaVersion> <ProjectGuid>{CCC4440E-49F7-4790-B0AF-FEABB0837AE7}</ProjectGuid> <OutputType>Library</OutputType> <RootNamespace>Grpc.Core</RootNamespace> @@ -73,13 +71,11 @@ <Compile Include="Marshaller.cs" /> <Compile Include="ServerServiceDefinition.cs" /> <Compile Include="Utils\AsyncStreamExtensions.cs" /> - <Compile Include="Internal\BatchContextSafeHandleNotOwned.cs" /> <Compile Include="Utils\BenchmarkUtil.cs" /> <Compile Include="Utils\ExceptionHelper.cs" /> <Compile Include="Internal\CredentialsSafeHandle.cs" /> <Compile Include="Credentials.cs" /> <Compile Include="Internal\ChannelArgsSafeHandle.cs" /> - <Compile Include="ChannelArgs.cs" /> <Compile Include="Internal\AsyncCompletion.cs" /> <Compile Include="Internal\AsyncCallBase.cs" /> <Compile Include="Internal\AsyncCallServer.cs" /> @@ -101,6 +97,10 @@ <Compile Include="Internal\AtomicCounter.cs" /> <Compile Include="Internal\DebugStats.cs" /> <Compile Include="ServerCallContext.cs" /> + <Compile Include="Internal\CompletionQueueEvent.cs" /> + <Compile Include="Internal\CompletionRegistry.cs" /> + <Compile Include="Internal\BatchContextSafeHandle.cs" /> + <Compile Include="ChannelOptions.cs" /> </ItemGroup> <ItemGroup> <None Include="packages.config" /> @@ -130,4 +130,4 @@ </Target> <Import Project="..\packages\grpc.dependencies.openssl.redist.1.0.2.2\build\portable-net45\grpc.dependencies.openssl.redist.targets" Condition="Exists('..\packages\grpc.dependencies.openssl.redist.1.0.2.2\build\portable-net45\grpc.dependencies.openssl.redist.targets')" /> <Import Project="..\packages\grpc.dependencies.zlib.redist.1.2.8.9\build\portable-net45\grpc.dependencies.zlib.redist.targets" Condition="Exists('..\packages\grpc.dependencies.zlib.redist.1.2.8.9\build\portable-net45\grpc.dependencies.zlib.redist.targets')" /> -</Project>
\ No newline at end of file +</Project> diff --git a/src/csharp/Grpc.Core/Grpc.Core.nuspec b/src/csharp/Grpc.Core/Grpc.Core.nuspec index 42eb90c9a3..629b978fdf 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.nuspec +++ b/src/csharp/Grpc.Core/Grpc.Core.nuspec @@ -5,19 +5,19 @@ <title>gRPC C# Core</title> <summary>Core C# implementation of gRPC - an RPC library and framework</summary> <description>Core C# implementation of gRPC - an RPC library and framework. See project site for more info.</description> - <version>0.5.0</version> + <version>0.5.1</version> <authors>Google Inc.</authors> <owners>grpc-packages</owners> <licenseUrl>https://github.com/grpc/grpc/blob/master/LICENSE</licenseUrl> <projectUrl>https://github.com/grpc/grpc</projectUrl> <requireLicenseAcceptance>false</requireLicenseAcceptance> - <releaseNotes>Release 0.5.0 of gRPC C#</releaseNotes> + <releaseNotes>Release 0.5.1 of gRPC C#</releaseNotes> <copyright>Copyright 2015, Google Inc.</copyright> <tags>gRPC RPC Protocol HTTP/2</tags> <dependencies> <dependency id="Microsoft.Bcl.Immutable" version="1.0.34" /> <dependency id="Ix-Async" version="1.2.3" /> - <dependency id="grpc.native.csharp_ext" version="0.9.0.0" /> + <dependency id="grpc.native.csharp_ext" version="0.9.1" /> </dependencies> </metadata> <files> diff --git a/src/csharp/Grpc.Core/GrpcEnvironment.cs b/src/csharp/Grpc.Core/GrpcEnvironment.cs index 2e9e5a2ef6..30ff289714 100644 --- a/src/csharp/Grpc.Core/GrpcEnvironment.cs +++ b/src/csharp/Grpc.Core/GrpcEnvironment.cs @@ -54,6 +54,7 @@ namespace Grpc.Core static volatile GrpcEnvironment instance; readonly GrpcThreadPool threadPool; + readonly CompletionRegistry completionRegistry; bool isClosed; /// <summary> @@ -105,6 +106,19 @@ namespace Grpc.Core } } + internal static CompletionRegistry CompletionRegistry + { + get + { + var inst = instance; + if (inst == null) + { + throw new InvalidOperationException("GRPC environment not initialized"); + } + return inst.completionRegistry; + } + } + /// <summary> /// Creates gRPC environment. /// </summary> @@ -112,6 +126,7 @@ namespace Grpc.Core { GrpcLog.RedirectNativeLogs(Console.Error); grpcsharp_init(); + completionRegistry = new CompletionRegistry(); threadPool = new GrpcThreadPool(THREAD_POOL_SIZE); threadPool.Start(); // TODO: use proper logging here @@ -139,14 +154,24 @@ namespace Grpc.Core { var remainingClientCalls = DebugStats.ActiveClientCalls.Count; if (remainingClientCalls != 0) - { - Console.WriteLine("Warning: Detected {0} client calls that weren't disposed properly.", remainingClientCalls); + { + DebugWarning(string.Format("Detected {0} client calls that weren't disposed properly.", remainingClientCalls)); } var remainingServerCalls = DebugStats.ActiveServerCalls.Count; if (remainingServerCalls != 0) { - Console.WriteLine("Warning: Detected {0} server calls that weren't disposed properly.", remainingServerCalls); + DebugWarning(string.Format("Detected {0} server calls that weren't disposed properly.", remainingServerCalls)); + } + var pendingBatchCompletions = DebugStats.PendingBatchCompletions.Count; + if (pendingBatchCompletions != 0) + { + DebugWarning(string.Format("Detected {0} pending batch completions.", pendingBatchCompletions)); } } + + private static void DebugWarning(string message) + { + throw new Exception("Shutdown check: " + message); + } } } diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index 9bb918d53d..d350f45da6 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -47,9 +47,6 @@ namespace Grpc.Core.Internal /// </summary> internal class AsyncCall<TRequest, TResponse> : AsyncCallBase<TRequest, TResponse> { - readonly CompletionCallbackDelegate unaryResponseHandler; - readonly CompletionCallbackDelegate finishedHandler; - // Completion of a pending unary response if not null. TaskCompletionSource<TResponse> unaryResponseTcs; @@ -60,8 +57,6 @@ namespace Grpc.Core.Internal public AsyncCall(Func<TRequest, byte[]> serializer, Func<byte[], TResponse> deserializer) : base(serializer, deserializer) { - this.unaryResponseHandler = CreateBatchCompletionCallback(HandleUnaryResponse); - this.finishedHandler = CreateBatchCompletionCallback(HandleFinished); } public void Initialize(Channel channel, CompletionQueueSafeHandle cq, string methodName) @@ -96,7 +91,21 @@ namespace Grpc.Core.Internal using (var metadataArray = MetadataArraySafeHandle.Create(headers)) { - call.BlockingUnary(cq, payload, unaryResponseHandler, metadataArray); + using (var ctx = BatchContextSafeHandle.Create()) + { + call.StartUnary(payload, ctx, metadataArray); + var ev = cq.Pluck(ctx.Handle); + + bool success = (ev.success != 0); + try + { + HandleUnaryResponse(success, ctx); + } + catch (Exception e) + { + Console.WriteLine("Exception occured while invoking completion delegate: " + e); + } + } } try @@ -129,7 +138,7 @@ namespace Grpc.Core.Internal unaryResponseTcs = new TaskCompletionSource<TResponse>(); using (var metadataArray = MetadataArraySafeHandle.Create(headers)) { - call.StartUnary(payload, unaryResponseHandler, metadataArray); + call.StartUnary(payload, HandleUnaryResponse, metadataArray); } return unaryResponseTcs.Task; } @@ -151,7 +160,7 @@ namespace Grpc.Core.Internal unaryResponseTcs = new TaskCompletionSource<TResponse>(); using (var metadataArray = MetadataArraySafeHandle.Create(headers)) { - call.StartClientStreaming(unaryResponseHandler, metadataArray); + call.StartClientStreaming(HandleUnaryResponse, metadataArray); } return unaryResponseTcs.Task; @@ -175,7 +184,7 @@ namespace Grpc.Core.Internal using (var metadataArray = MetadataArraySafeHandle.Create(headers)) { - call.StartServerStreaming(payload, finishedHandler, metadataArray); + call.StartServerStreaming(payload, HandleFinished, metadataArray); } } } @@ -194,7 +203,7 @@ namespace Grpc.Core.Internal using (var metadataArray = MetadataArraySafeHandle.Create(headers)) { - call.StartDuplexStreaming(finishedHandler, metadataArray); + call.StartDuplexStreaming(HandleFinished, metadataArray); } } } @@ -229,7 +238,7 @@ namespace Grpc.Core.Internal Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null"); CheckSendingAllowed(); - call.StartSendCloseFromClient(halfclosedHandler); + call.StartSendCloseFromClient(HandleHalfclosed); halfcloseRequested = true; sendCompletionDelegate = completionDelegate; @@ -274,7 +283,7 @@ namespace Grpc.Core.Internal /// <summary> /// Handler for unary response completion. /// </summary> - private void HandleUnaryResponse(bool success, BatchContextSafeHandleNotOwned ctx) + private void HandleUnaryResponse(bool success, BatchContextSafeHandle ctx) { lock (myLock) { @@ -307,7 +316,7 @@ namespace Grpc.Core.Internal /// <summary> /// Handles receive status completion for calls with streaming response. /// </summary> - private void HandleFinished(bool success, BatchContextSafeHandleNotOwned ctx) + private void HandleFinished(bool success, BatchContextSafeHandle ctx) { var status = ctx.GetReceivedStatus(); diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index b4f4edb17a..64713c8c52 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -51,13 +51,8 @@ namespace Grpc.Core.Internal readonly Func<TWrite, byte[]> serializer; readonly Func<byte[], TRead> deserializer; - protected readonly CompletionCallbackDelegate sendFinishedHandler; - protected readonly CompletionCallbackDelegate readFinishedHandler; - protected readonly CompletionCallbackDelegate halfclosedHandler; - protected readonly object myLock = new object(); - protected GCHandle gchandle; protected CallSafeHandle call; protected bool disposed; @@ -77,10 +72,6 @@ namespace Grpc.Core.Internal { this.serializer = Preconditions.CheckNotNull(serializer); this.deserializer = Preconditions.CheckNotNull(deserializer); - - this.sendFinishedHandler = CreateBatchCompletionCallback(HandleSendFinished); - this.readFinishedHandler = CreateBatchCompletionCallback(HandleReadFinished); - this.halfclosedHandler = CreateBatchCompletionCallback(HandleHalfclosed); } /// <summary> @@ -121,9 +112,6 @@ namespace Grpc.Core.Internal { lock (myLock) { - // Make sure this object and the delegated held by it will not be garbage collected - // before we release this handle. - gchandle = GCHandle.Alloc(this); this.call = call; } } @@ -141,7 +129,7 @@ namespace Grpc.Core.Internal Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null"); CheckSendingAllowed(); - call.StartSendMessage(payload, sendFinishedHandler); + call.StartSendMessage(payload, HandleSendFinished); sendCompletionDelegate = completionDelegate; } } @@ -157,7 +145,7 @@ namespace Grpc.Core.Internal Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null"); CheckReadingAllowed(); - call.StartReceiveMessage(readFinishedHandler); + call.StartReceiveMessage(HandleReadFinished); readCompletionDelegate = completionDelegate; } } @@ -197,7 +185,6 @@ namespace Grpc.Core.Internal { call.Dispose(); } - gchandle.Free(); disposed = true; } @@ -282,29 +269,9 @@ namespace Grpc.Core.Internal } /// <summary> - /// Creates completion callback delegate that wraps the batch completion handler in a try catch block to - /// prevent propagating exceptions accross managed/unmanaged boundary. - /// </summary> - protected CompletionCallbackDelegate CreateBatchCompletionCallback(Action<bool, BatchContextSafeHandleNotOwned> handler) - { - return new CompletionCallbackDelegate((success, batchContextPtr) => - { - try - { - var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr); - handler(success, ctx); - } - catch (Exception e) - { - Console.WriteLine("Caught exception in a native handler: " + e); - } - }); - } - - /// <summary> /// Handles send completion. /// </summary> - private void HandleSendFinished(bool success, BatchContextSafeHandleNotOwned ctx) + protected void HandleSendFinished(bool success, BatchContextSafeHandle ctx) { AsyncCompletionDelegate<object> origCompletionDelegate = null; lock (myLock) @@ -328,7 +295,7 @@ namespace Grpc.Core.Internal /// <summary> /// Handles halfclose completion. /// </summary> - private void HandleHalfclosed(bool success, BatchContextSafeHandleNotOwned ctx) + protected void HandleHalfclosed(bool success, BatchContextSafeHandle ctx) { AsyncCompletionDelegate<object> origCompletionDelegate = null; lock (myLock) @@ -353,7 +320,7 @@ namespace Grpc.Core.Internal /// <summary> /// Handles streaming read completion. /// </summary> - private void HandleReadFinished(bool success, BatchContextSafeHandleNotOwned ctx) + protected void HandleReadFinished(bool success, BatchContextSafeHandle ctx) { var payload = ctx.GetReceivedMessage(); diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 1f0335e4e6..4f510ba40a 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -47,12 +47,10 @@ namespace Grpc.Core.Internal /// </summary> internal class AsyncCallServer<TRequest, TResponse> : AsyncCallBase<TResponse, TRequest> { - readonly CompletionCallbackDelegate finishedServersideHandler; readonly TaskCompletionSource<object> finishedServersideTcs = new TaskCompletionSource<object>(); public AsyncCallServer(Func<TResponse, byte[]> serializer, Func<byte[], TRequest> deserializer) : base(serializer, deserializer) { - this.finishedServersideHandler = CreateBatchCompletionCallback(HandleFinishedServerside); } public void Initialize(CallSafeHandle call) @@ -72,7 +70,7 @@ namespace Grpc.Core.Internal started = true; - call.StartServerSide(finishedServersideHandler); + call.StartServerSide(HandleFinishedServerside); return finishedServersideTcs.Task; } } @@ -107,8 +105,9 @@ namespace Grpc.Core.Internal Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null"); CheckSendingAllowed(); - call.StartSendStatusFromServer(status, halfclosedHandler); + call.StartSendStatusFromServer(status, HandleHalfclosed); halfcloseRequested = true; + readingDone = true; sendCompletionDelegate = completionDelegate; } } @@ -121,7 +120,7 @@ namespace Grpc.Core.Internal /// <summary> /// Handles the server side close completion. /// </summary> - private void HandleFinishedServerside(bool success, BatchContextSafeHandleNotOwned ctx) + private void HandleFinishedServerside(bool success, BatchContextSafeHandle ctx) { bool cancelled = ctx.GetReceivedCloseOnServerCancelled(); diff --git a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandleNotOwned.cs b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs index b562abaa7a..861cbbe4c6 100644 --- a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandleNotOwned.cs +++ b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs @@ -41,32 +41,50 @@ namespace Grpc.Core.Internal /// Not owned version of /// grpcsharp_batch_context /// </summary> - internal class BatchContextSafeHandleNotOwned : SafeHandleZeroIsInvalid + internal class BatchContextSafeHandle : SafeHandleZeroIsInvalid { [DllImport("grpc_csharp_ext.dll")] - static extern IntPtr grpcsharp_batch_context_recv_message_length(BatchContextSafeHandleNotOwned ctx); + static extern BatchContextSafeHandle grpcsharp_batch_context_create(); [DllImport("grpc_csharp_ext.dll")] - static extern void grpcsharp_batch_context_recv_message_to_buffer(BatchContextSafeHandleNotOwned ctx, byte[] buffer, UIntPtr bufferLen); + static extern IntPtr grpcsharp_batch_context_recv_message_length(BatchContextSafeHandle ctx); [DllImport("grpc_csharp_ext.dll")] - static extern StatusCode grpcsharp_batch_context_recv_status_on_client_status(BatchContextSafeHandleNotOwned ctx); + static extern void grpcsharp_batch_context_recv_message_to_buffer(BatchContextSafeHandle ctx, byte[] buffer, UIntPtr bufferLen); [DllImport("grpc_csharp_ext.dll")] - static extern IntPtr grpcsharp_batch_context_recv_status_on_client_details(BatchContextSafeHandleNotOwned ctx); // returns const char* + static extern StatusCode grpcsharp_batch_context_recv_status_on_client_status(BatchContextSafeHandle ctx); [DllImport("grpc_csharp_ext.dll")] - static extern CallSafeHandle grpcsharp_batch_context_server_rpc_new_call(BatchContextSafeHandleNotOwned ctx); + static extern IntPtr grpcsharp_batch_context_recv_status_on_client_details(BatchContextSafeHandle ctx); // returns const char* [DllImport("grpc_csharp_ext.dll")] - static extern IntPtr grpcsharp_batch_context_server_rpc_new_method(BatchContextSafeHandleNotOwned ctx); // returns const char* + static extern CallSafeHandle grpcsharp_batch_context_server_rpc_new_call(BatchContextSafeHandle ctx); [DllImport("grpc_csharp_ext.dll")] - static extern int grpcsharp_batch_context_recv_close_on_server_cancelled(BatchContextSafeHandleNotOwned ctx); + static extern IntPtr grpcsharp_batch_context_server_rpc_new_method(BatchContextSafeHandle ctx); // returns const char* - public BatchContextSafeHandleNotOwned(IntPtr handle) : base(false) + [DllImport("grpc_csharp_ext.dll")] + static extern int grpcsharp_batch_context_recv_close_on_server_cancelled(BatchContextSafeHandle ctx); + + [DllImport("grpc_csharp_ext.dll")] + static extern void grpcsharp_batch_context_destroy(IntPtr ctx); + + private BatchContextSafeHandle() + { + } + + public static BatchContextSafeHandle Create() { - SetHandle(handle); + return grpcsharp_batch_context_create(); + } + + public IntPtr Handle + { + get + { + return handle; + } } public Status GetReceivedStatus() @@ -102,5 +120,11 @@ namespace Grpc.Core.Internal { return grpcsharp_batch_context_recv_close_on_server_cancelled(this) != 0; } + + protected override bool ReleaseHandle() + { + grpcsharp_batch_context_destroy(handle); + return true; + } } }
\ No newline at end of file diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs index 491b8414ec..ef92b44402 100644 --- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs @@ -37,8 +37,6 @@ using Grpc.Core.Utils; namespace Grpc.Core.Internal { - internal delegate void CompletionCallbackDelegate(bool success, IntPtr batchContextPtr); - /// <summary> /// grpc_call from <grpc/grpc.h> /// </summary> @@ -57,49 +55,40 @@ namespace Grpc.Core.Internal [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_start_unary(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len, MetadataArraySafeHandle metadataArray); - - [DllImport("grpc_csharp_ext.dll")] - static extern void grpcsharp_call_blocking_unary(CallSafeHandle call, CompletionQueueSafeHandle dedicatedCq, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len, MetadataArraySafeHandle metadataArray); + BatchContextSafeHandle ctx, byte[] send_buffer, UIntPtr send_buffer_len, MetadataArraySafeHandle metadataArray); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_start_client_streaming(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - MetadataArraySafeHandle metadataArray); + BatchContextSafeHandle ctx, MetadataArraySafeHandle metadataArray); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_start_server_streaming(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len, - MetadataArraySafeHandle metadataArray); + BatchContextSafeHandle ctx, byte[] send_buffer, UIntPtr send_buffer_len, + MetadataArraySafeHandle metadataArray); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_start_duplex_streaming(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - MetadataArraySafeHandle metadataArray); + BatchContextSafeHandle ctx, MetadataArraySafeHandle metadataArray); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_send_message(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len); + BatchContextSafeHandle ctx, byte[] send_buffer, UIntPtr send_buffer_len); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_send_close_from_client(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); + BatchContextSafeHandle ctx); [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_send_status_from_server(CallSafeHandle call, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, StatusCode statusCode, string statusMessage); + static extern GRPCCallError grpcsharp_call_send_status_from_server(CallSafeHandle call, + BatchContextSafeHandle ctx, StatusCode statusCode, string statusMessage); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_recv_message(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); + BatchContextSafeHandle ctx); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_start_serverside(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); + BatchContextSafeHandle ctx); [DllImport("grpc_csharp_ext.dll")] static extern void grpcsharp_call_destroy(IntPtr call); @@ -113,64 +102,84 @@ namespace Grpc.Core.Internal return grpcsharp_channel_create_call(channel, cq, method, host, deadline); } - public void StartUnary(byte[] payload, CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) + public void StartUnary(byte[] payload, BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray) { - AssertCallOk(grpcsharp_call_start_unary(this, callback, payload, new UIntPtr((ulong)payload.Length), metadataArray)); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_call_start_unary(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray) + .CheckOk(); } - public void BlockingUnary(CompletionQueueSafeHandle dedicatedCq, byte[] payload, CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) + public void StartUnary(byte[] payload, BatchContextSafeHandle ctx, MetadataArraySafeHandle metadataArray) { - grpcsharp_call_blocking_unary(this, dedicatedCq, callback, payload, new UIntPtr((ulong)payload.Length), metadataArray); + grpcsharp_call_start_unary(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray) + .CheckOk(); } - public void StartClientStreaming(CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) + public void StartClientStreaming(BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray) { - AssertCallOk(grpcsharp_call_start_client_streaming(this, callback, metadataArray)); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_call_start_client_streaming(this, ctx, metadataArray).CheckOk(); } - public void StartServerStreaming(byte[] payload, CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) + public void StartServerStreaming(byte[] payload, BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray) { - AssertCallOk(grpcsharp_call_start_server_streaming(this, callback, payload, new UIntPtr((ulong)payload.Length), metadataArray)); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_call_start_server_streaming(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray).CheckOk(); } - public void StartDuplexStreaming(CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) + public void StartDuplexStreaming(BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray) { - AssertCallOk(grpcsharp_call_start_duplex_streaming(this, callback, metadataArray)); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_call_start_duplex_streaming(this, ctx, metadataArray).CheckOk(); } - public void StartSendMessage(byte[] payload, CompletionCallbackDelegate callback) + public void StartSendMessage(byte[] payload, BatchCompletionDelegate callback) { - AssertCallOk(grpcsharp_call_send_message(this, callback, payload, new UIntPtr((ulong)payload.Length))); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_call_send_message(this, ctx, payload, new UIntPtr((ulong)payload.Length)).CheckOk(); } - public void StartSendCloseFromClient(CompletionCallbackDelegate callback) + public void StartSendCloseFromClient(BatchCompletionDelegate callback) { - AssertCallOk(grpcsharp_call_send_close_from_client(this, callback)); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_call_send_close_from_client(this, ctx).CheckOk(); } - public void StartSendStatusFromServer(Status status, CompletionCallbackDelegate callback) + public void StartSendStatusFromServer(Status status, BatchCompletionDelegate callback) { - AssertCallOk(grpcsharp_call_send_status_from_server(this, callback, status.StatusCode, status.Detail)); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, status.Detail).CheckOk(); } - public void StartReceiveMessage(CompletionCallbackDelegate callback) + public void StartReceiveMessage(BatchCompletionDelegate callback) { - AssertCallOk(grpcsharp_call_recv_message(this, callback)); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_call_recv_message(this, ctx).CheckOk(); } - public void StartServerSide(CompletionCallbackDelegate callback) + public void StartServerSide(BatchCompletionDelegate callback) { - AssertCallOk(grpcsharp_call_start_serverside(this, callback)); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_call_start_serverside(this, ctx).CheckOk(); } public void Cancel() { - AssertCallOk(grpcsharp_call_cancel(this)); + grpcsharp_call_cancel(this).CheckOk(); } public void CancelWithStatus(Status status) { - AssertCallOk(grpcsharp_call_cancel_with_status(this, status.StatusCode, status.Detail)); + grpcsharp_call_cancel_with_status(this, status.StatusCode, status.Detail).CheckOk(); } protected override bool ReleaseHandle() @@ -179,11 +188,6 @@ namespace Grpc.Core.Internal return true; } - private static void AssertCallOk(GRPCCallError callError) - { - Preconditions.CheckState(callError == GRPCCallError.GRPC_CALL_OK, "Status not GRPC_CALL_OK"); - } - private static uint GetFlags(bool buffered) { return buffered ? 0 : GRPC_WRITE_BUFFER_HINT; diff --git a/src/csharp/Grpc.Core/Internal/ChannelArgsSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ChannelArgsSafeHandle.cs index c69f1a0d02..c12aec5a3a 100644 --- a/src/csharp/Grpc.Core/Internal/ChannelArgsSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/ChannelArgsSafeHandle.cs @@ -45,6 +45,9 @@ namespace Grpc.Core.Internal [DllImport("grpc_csharp_ext.dll", CharSet = CharSet.Ansi)] static extern void grpcsharp_channel_args_set_string(ChannelArgsSafeHandle args, UIntPtr index, string key, string value); + [DllImport("grpc_csharp_ext.dll", CharSet = CharSet.Ansi)] + static extern void grpcsharp_channel_args_set_integer(ChannelArgsSafeHandle args, UIntPtr index, string key, int value); + [DllImport("grpc_csharp_ext.dll")] static extern void grpcsharp_channel_args_destroy(IntPtr args); @@ -67,6 +70,11 @@ namespace Grpc.Core.Internal grpcsharp_channel_args_set_string(this, new UIntPtr((uint)index), key, value); } + public void SetInteger(int index, string key, int value) + { + grpcsharp_channel_args_set_integer(this, new UIntPtr((uint)index), key, value); + } + protected override bool ReleaseHandle() { grpcsharp_channel_args_destroy(handle); diff --git a/src/csharp/Grpc.Core/Internal/CompletionQueueEvent.cs b/src/csharp/Grpc.Core/Internal/CompletionQueueEvent.cs new file mode 100644 index 0000000000..3f517514a3 --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/CompletionQueueEvent.cs @@ -0,0 +1,60 @@ +#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; + +namespace Grpc.Core.Internal +{ + /// <summary> + /// grpc_event from grpc/grpc.h + /// </summary> + [StructLayout(LayoutKind.Sequential)] + internal struct CompletionQueueEvent + { + [DllImport("grpc_csharp_ext.dll")] + static extern int grpcsharp_sizeof_grpc_event(); + + public GRPCCompletionType type; + public int success; + public IntPtr tag; + + internal static int NativeSize + { + get + { + return grpcsharp_sizeof_grpc_event(); + } + } + } +} diff --git a/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs index 600d1fc87c..f64f3d4175 100644 --- a/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs @@ -46,7 +46,10 @@ namespace Grpc.Core.Internal static extern void grpcsharp_completion_queue_shutdown(CompletionQueueSafeHandle cq); [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCompletionType grpcsharp_completion_queue_next_with_callback(CompletionQueueSafeHandle cq); + static extern CompletionQueueEvent grpcsharp_completion_queue_next(CompletionQueueSafeHandle cq); + + [DllImport("grpc_csharp_ext.dll")] + static extern CompletionQueueEvent grpcsharp_completion_queue_pluck(CompletionQueueSafeHandle cq, IntPtr tag); [DllImport("grpc_csharp_ext.dll")] static extern void grpcsharp_completion_queue_destroy(IntPtr cq); @@ -60,9 +63,14 @@ namespace Grpc.Core.Internal return grpcsharp_completion_queue_create(); } - public GRPCCompletionType NextWithCallback() + public CompletionQueueEvent Next() + { + return grpcsharp_completion_queue_next(this); + } + + public CompletionQueueEvent Pluck(IntPtr tag) { - return grpcsharp_completion_queue_next_with_callback(this); + return grpcsharp_completion_queue_pluck(this, tag); } public void Shutdown() diff --git a/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs b/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs new file mode 100644 index 0000000000..80f006ae50 --- /dev/null +++ b/src/csharp/Grpc.Core/Internal/CompletionRegistry.cs @@ -0,0 +1,89 @@ +#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.Collections.Concurrent; +using System.Collections.Generic; +using System.Runtime.InteropServices; +using Grpc.Core.Utils; + +namespace Grpc.Core.Internal +{ + internal delegate void OpCompletionDelegate(bool success); + + internal delegate void BatchCompletionDelegate(bool success, BatchContextSafeHandle ctx); + + internal class CompletionRegistry + { + readonly ConcurrentDictionary<IntPtr, OpCompletionDelegate> dict = new ConcurrentDictionary<IntPtr, OpCompletionDelegate>(); + + public void Register(IntPtr key, OpCompletionDelegate callback) + { + DebugStats.PendingBatchCompletions.Increment(); + Preconditions.CheckState(dict.TryAdd(key, callback)); + } + + public void RegisterBatchCompletion(BatchContextSafeHandle ctx, BatchCompletionDelegate callback) + { + OpCompletionDelegate opCallback = ((success) => HandleBatchCompletion(success, ctx, callback)); + Register(ctx.Handle, opCallback); + } + + public OpCompletionDelegate Extract(IntPtr key) + { + OpCompletionDelegate value; + Preconditions.CheckState(dict.TryRemove(key, out value)); + DebugStats.PendingBatchCompletions.Decrement(); + return value; + } + + private static void HandleBatchCompletion(bool success, BatchContextSafeHandle ctx, BatchCompletionDelegate callback) + { + try + { + callback(success, ctx); + } + catch (Exception e) + { + Console.WriteLine("Exception occured while invoking completion delegate: " + e); + } + finally + { + if (ctx != null) + { + ctx.Dispose(); + } + } + } + } +} diff --git a/src/csharp/Grpc.Core/Internal/DebugStats.cs b/src/csharp/Grpc.Core/Internal/DebugStats.cs index 476914f751..ef9d9afe11 100644 --- a/src/csharp/Grpc.Core/Internal/DebugStats.cs +++ b/src/csharp/Grpc.Core/Internal/DebugStats.cs @@ -41,5 +41,7 @@ namespace Grpc.Core.Internal public static readonly AtomicCounter ActiveClientCalls = new AtomicCounter(); public static readonly AtomicCounter ActiveServerCalls = new AtomicCounter(); + + public static readonly AtomicCounter PendingBatchCompletions = new AtomicCounter(); } } diff --git a/src/csharp/Grpc.Core/Internal/Enums.cs b/src/csharp/Grpc.Core/Internal/Enums.cs index 2b4f6cae0c..af11b5b9f3 100644 --- a/src/csharp/Grpc.Core/Internal/Enums.cs +++ b/src/csharp/Grpc.Core/Internal/Enums.cs @@ -33,35 +33,47 @@ using System; using System.Runtime.InteropServices; +using Grpc.Core.Utils; namespace Grpc.Core.Internal { /// <summary> - /// from grpc/grpc.h + /// grpc_call_error from grpc/grpc.h /// </summary> internal enum GRPCCallError { /* everything went ok */ - GRPC_CALL_OK = 0, + OK = 0, /* something failed, we don't know what */ - GRPC_CALL_ERROR, + Error, /* this method is not available on the server */ - GRPC_CALL_ERROR_NOT_ON_SERVER, + NotOnServer, /* this method is not available on the client */ - GRPC_CALL_ERROR_NOT_ON_CLIENT, + NotOnClient, /* this method must be called before server_accept */ - GRPC_CALL_ERROR_ALREADY_ACCEPTED, + AlreadyAccepted, /* this method must be called before invoke */ - GRPC_CALL_ERROR_ALREADY_INVOKED, + AlreadyInvoked, /* this method must be called after invoke */ - GRPC_CALL_ERROR_NOT_INVOKED, + NotInvoked, /* this call is already finished (writes_done or write_status has already been called) */ - GRPC_CALL_ERROR_ALREADY_FINISHED, + AlreadyFinished, /* there is already an outstanding read/write operation on the call */ - GRPC_CALL_ERROR_TOO_MANY_OPERATIONS, + TooManyOperations, /* the flags value was illegal for this call */ - GRPC_CALL_ERROR_INVALID_FLAGS + InvalidFlags + } + + internal static class CallErrorExtensions + { + /// <summary> + /// Checks the call API invocation's result is OK. + /// </summary> + public static void CheckOk(this GRPCCallError callError) + { + Preconditions.CheckState(callError == GRPCCallError.OK, "Call error: " + callError); + } } /// <summary> @@ -70,12 +82,12 @@ namespace Grpc.Core.Internal internal enum GRPCCompletionType { /* Shutting down */ - GRPC_QUEUE_SHUTDOWN, + Shutdown, /* No event before timeout */ - GRPC_QUEUE_TIMEOUT, + Timeout, /* operation completion */ - GRPC_OP_COMPLETE + OpComplete } } diff --git a/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs b/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs index f4224668f1..89b44a4e2b 100644 --- a/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs +++ b/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs @@ -112,12 +112,26 @@ namespace Grpc.Core.Internal /// </summary> private void RunHandlerLoop() { - GRPCCompletionType completionType; + CompletionQueueEvent ev; do { - completionType = cq.NextWithCallback(); + ev = cq.Next(); + if (ev.type == GRPCCompletionType.OpComplete) + { + bool success = (ev.success != 0); + IntPtr tag = ev.tag; + try + { + var callback = GrpcEnvironment.CompletionRegistry.Extract(tag); + callback(success); + } + catch (Exception e) + { + Console.WriteLine("Exception occured while invoking completion delegate: " + e); + } + } } - while (completionType != GRPCCompletionType.GRPC_QUEUE_SHUTDOWN); + while (ev.type != GRPCCompletionType.Shutdown); Console.WriteLine("Completion queue has shutdown successfully, thread " + Thread.CurrentThread.Name + " exiting."); } } diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs index f494d9e0ff..c0e5bae13f 100644 --- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs +++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs @@ -267,8 +267,6 @@ namespace Grpc.Core.Internal var responseStream = new ServerResponseStream<byte[], byte[]>(asyncCall); await responseStream.WriteStatusAsync(new Status(StatusCode.Unimplemented, "No such method.")); - // TODO(jtattermusch): if we don't read what client has sent, the server call never gets disposed. - await requestStream.ToList(); await finishedTask; } } diff --git a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs index c44ee87bad..83dbb910aa 100644 --- a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs @@ -45,10 +45,7 @@ namespace Grpc.Core.Internal internal sealed class ServerSafeHandle : SafeHandleZeroIsInvalid { [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_server_request_call(ServerSafeHandle server, CompletionQueueSafeHandle cq, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); - - [DllImport("grpc_csharp_ext.dll")] - static extern ServerSafeHandle grpcsharp_server_create(CompletionQueueSafeHandle cq, IntPtr args); + static extern ServerSafeHandle grpcsharp_server_create(CompletionQueueSafeHandle cq, ChannelArgsSafeHandle args); [DllImport("grpc_csharp_ext.dll")] static extern int grpcsharp_server_add_http2_port(ServerSafeHandle server, string addr); @@ -60,19 +57,22 @@ namespace Grpc.Core.Internal static extern void grpcsharp_server_start(ServerSafeHandle server); [DllImport("grpc_csharp_ext.dll")] - static extern void grpcsharp_server_shutdown_and_notify_callback(ServerSafeHandle server, CompletionQueueSafeHandle cq, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); + static extern GRPCCallError grpcsharp_server_request_call(ServerSafeHandle server, CompletionQueueSafeHandle cq, BatchContextSafeHandle ctx); [DllImport("grpc_csharp_ext.dll")] static extern void grpcsharp_server_cancel_all_calls(ServerSafeHandle server); [DllImport("grpc_csharp_ext.dll")] + static extern void grpcsharp_server_shutdown_and_notify_callback(ServerSafeHandle server, CompletionQueueSafeHandle cq, BatchContextSafeHandle ctx); + + [DllImport("grpc_csharp_ext.dll")] static extern void grpcsharp_server_destroy(IntPtr server); private ServerSafeHandle() { } - public static ServerSafeHandle NewServer(CompletionQueueSafeHandle cq, IntPtr args) + public static ServerSafeHandle NewServer(CompletionQueueSafeHandle cq, ChannelArgsSafeHandle args) { return grpcsharp_server_create(cq, args); } @@ -91,15 +91,19 @@ namespace Grpc.Core.Internal { grpcsharp_server_start(this); } - - public void ShutdownAndNotify(CompletionQueueSafeHandle cq, CompletionCallbackDelegate callback) + + public void ShutdownAndNotify(CompletionQueueSafeHandle cq, BatchCompletionDelegate callback) { - grpcsharp_server_shutdown_and_notify_callback(this, cq, callback); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_server_shutdown_and_notify_callback(this, cq, ctx); } - public void RequestCall(CompletionQueueSafeHandle cq, CompletionCallbackDelegate callback) + public void RequestCall(CompletionQueueSafeHandle cq, BatchCompletionDelegate callback) { - AssertCallOk(grpcsharp_server_request_call(this, cq, callback)); + var ctx = BatchContextSafeHandle.Create(); + GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback); + grpcsharp_server_request_call(this, cq, ctx).CheckOk(); } protected override bool ReleaseHandle() @@ -113,10 +117,5 @@ namespace Grpc.Core.Internal { grpcsharp_server_cancel_all_calls(this); } - - private static void AssertCallOk(GRPCCallError callError) - { - Preconditions.CheckState(callError == GRPCCallError.GRPC_CALL_OK, "Status not GRPC_CALL_OK"); - } } } diff --git a/src/csharp/Grpc.Core/Internal/Timespec.cs b/src/csharp/Grpc.Core/Internal/Timespec.cs index 94d48c2c49..775af27db9 100644 --- a/src/csharp/Grpc.Core/Internal/Timespec.cs +++ b/src/csharp/Grpc.Core/Internal/Timespec.cs @@ -51,7 +51,7 @@ namespace Grpc.Core.Internal [DllImport("grpc_csharp_ext.dll")] static extern int gprsharp_sizeof_timespec(); - // TODO: revisit this. + // NOTE: on linux 64bit sizeof(gpr_timespec) = 16, on windows 32bit sizeof(gpr_timespec) = 8 // so IntPtr seems to have the right size to work on both. public System.IntPtr tv_sec; diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs index 0f4d77eaf3..8e818885d1 100644 --- a/src/csharp/Grpc.Core/Server.cs +++ b/src/csharp/Grpc.Core/Server.cs @@ -52,11 +52,6 @@ namespace Grpc.Core /// </summary> public const int PickUnusedPort = 0; - // TODO(jtattermusch) : make sure the delegate doesn't get garbage collected while - // native callbacks are in the completion queue. - readonly CompletionCallbackDelegate serverShutdownHandler; - readonly CompletionCallbackDelegate newServerRpcHandler; - readonly ServerSafeHandle handle; readonly object myLock = new object(); @@ -66,11 +61,16 @@ namespace Grpc.Core bool startRequested; bool shutdownRequested; - public Server() + /// <summary> + /// Create a new server. + /// </summary> + /// <param name="options">Channel options.</param> + public Server(IEnumerable<ChannelOption> options = null) { - this.handle = ServerSafeHandle.NewServer(GetCompletionQueue(), IntPtr.Zero); - this.newServerRpcHandler = HandleNewServerRpc; - this.serverShutdownHandler = HandleServerShutdown; + using (var channelArgs = ChannelOptions.CreateChannelArgs(options)) + { + this.handle = ServerSafeHandle.NewServer(GetCompletionQueue(), channelArgs); + } } /// <summary> @@ -108,7 +108,7 @@ namespace Grpc.Core /// </summary> /// <returns>The port on which server will be listening.</returns> /// <param name="host">the host</param> - /// <param name="port">the port. If zero, , an unused port is chosen automatically.</param> + /// <param name="port">the port. If zero, an unused port is chosen automatically.</param> public int AddListeningPort(string host, int port, ServerCredentials credentials) { Preconditions.CheckNotNull(credentials); @@ -144,7 +144,7 @@ namespace Grpc.Core shutdownRequested = true; } - handle.ShutdownAndNotify(GetCompletionQueue(), serverShutdownHandler); + handle.ShutdownAndNotify(GetCompletionQueue(), HandleServerShutdown); await shutdownTcs.Task; handle.Dispose(); } @@ -173,7 +173,7 @@ namespace Grpc.Core shutdownRequested = true; } - handle.ShutdownAndNotify(GetCompletionQueue(), serverShutdownHandler); + handle.ShutdownAndNotify(GetCompletionQueue(), HandleServerShutdown); handle.CancelAllCalls(); await shutdownTcs.Task; handle.Dispose(); @@ -208,7 +208,7 @@ namespace Grpc.Core { if (!shutdownRequested) { - handle.RequestCall(GetCompletionQueue(), newServerRpcHandler); + handle.RequestCall(GetCompletionQueue(), HandleNewServerRpc); } } } @@ -236,44 +236,28 @@ namespace Grpc.Core /// <summary> /// Handles the native callback. /// </summary> - private void HandleNewServerRpc(bool success, IntPtr batchContextPtr) + private void HandleNewServerRpc(bool success, BatchContextSafeHandle ctx) { - try - { - var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr); - - // TODO: handle error + // TODO: handle error - CallSafeHandle call = ctx.GetServerRpcNewCall(); - string method = ctx.GetServerRpcNewMethod(); - - // after server shutdown, the callback returns with null call - if (!call.IsInvalid) - { - Task.Run(async () => await InvokeCallHandler(call, method)); - } + CallSafeHandle call = ctx.GetServerRpcNewCall(); + string method = ctx.GetServerRpcNewMethod(); - AllowOneRpc(); - } - catch (Exception e) + // after server shutdown, the callback returns with null call + if (!call.IsInvalid) { - Console.WriteLine("Caught exception in a native handler: " + e); + Task.Run(async () => await InvokeCallHandler(call, method)); } + + AllowOneRpc(); } /// <summary> /// Handles native callback. /// </summary> - private void HandleServerShutdown(bool success, IntPtr batchContextPtr) + private void HandleServerShutdown(bool success, BatchContextSafeHandle ctx) { - try - { - shutdownTcs.SetResult(null); - } - catch (Exception e) - { - Console.WriteLine("Caught exception in a native handler: " + e); - } + shutdownTcs.SetResult(null); } private static CompletionQueueSafeHandle GetCompletionQueue() |