diff options
author | 2015-04-24 23:24:44 +0200 | |
---|---|---|
committer | 2015-04-24 23:24:44 +0200 | |
commit | fd2bf675f750f440d92bffeb88771e6d9253b498 (patch) | |
tree | 340f8a6b8a5b181c215a90e038b17475a3206855 /src/csharp/Grpc.Core/Server.cs | |
parent | 76e49fd80557c33595508686f3450982e63f0fce (diff) | |
parent | 4107ba289747be6fba879e6060d40969f1aee89c (diff) |
Merge branch 'master' of github.com:grpc/grpc into the-purge-2
Conflicts:
include/grpc++/completion_queue.h
Diffstat (limited to 'src/csharp/Grpc.Core/Server.cs')
-rw-r--r-- | src/csharp/Grpc.Core/Server.cs | 176 |
1 files changed, 93 insertions, 83 deletions
diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs index f086fa8beb..e686cdddef 100644 --- a/src/csharp/Grpc.Core/Server.cs +++ b/src/csharp/Grpc.Core/Server.cs @@ -38,27 +38,29 @@ using System.Diagnostics; using System.Runtime.InteropServices; using System.Threading.Tasks; using Grpc.Core.Internal; +using Grpc.Core.Utils; namespace Grpc.Core { /// <summary> - /// Server is implemented only to be able to do - /// in-process testing. + /// A gRPC server. /// </summary> public class Server { - // TODO: make sure the delegate doesn't get garbage collected while + // TODO(jtattermusch) : make sure the delegate doesn't get garbage collected while // native callbacks are in the completion queue. readonly ServerShutdownCallbackDelegate serverShutdownHandler; readonly CompletionCallbackDelegate newServerRpcHandler; - readonly BlockingCollection<NewRpcInfo> newRpcQueue = new BlockingCollection<NewRpcInfo>(); readonly ServerSafeHandle handle; + readonly object myLock = new object(); readonly Dictionary<string, IServerCallHandler> callHandlers = new Dictionary<string, IServerCallHandler>(); - readonly TaskCompletionSource<object> shutdownTcs = new TaskCompletionSource<object>(); + bool startRequested; + bool shutdownRequested; + public Server() { this.handle = ServerSafeHandle.NewServer(GetCompletionQueue(), IntPtr.Zero); @@ -66,71 +68,81 @@ namespace Grpc.Core this.serverShutdownHandler = HandleServerShutdown; } - // only call this before Start() + /// <summary> + /// Adds a service definition to the server. This is how you register + /// handlers for a service with the server. + /// Only call this before Start(). + /// </summary> public void AddServiceDefinition(ServerServiceDefinition serviceDefinition) { - foreach (var entry in serviceDefinition.CallHandlers) + lock (myLock) { - callHandlers.Add(entry.Key, entry.Value); + Preconditions.CheckState(!startRequested); + foreach (var entry in serviceDefinition.CallHandlers) + { + callHandlers.Add(entry.Key, entry.Value); + } } } - // only call before Start() + /// <summary> + /// Add a non-secure port on which server should listen. + /// Only call this before Start(). + /// </summary> public int AddListeningPort(string addr) { - return handle.AddListeningPort(addr); - } - - // only call before Start() - public int AddListeningPort(string addr, ServerCredentials credentials) - { - using (var nativeCredentials = credentials.ToNativeCredentials()) + lock (myLock) { - return handle.AddListeningPort(addr, nativeCredentials); + Preconditions.CheckState(!startRequested); + return handle.AddListeningPort(addr); } } - public void Start() - { - handle.Start(); - - // TODO: this basically means the server is single threaded.... - StartHandlingRpcs(); - } - /// <summary> - /// Requests and handles single RPC call. + /// Add a secure port on which server should listen. + /// Only call this before Start(). /// </summary> - internal void RunRpc() + public int AddListeningPort(string addr, ServerCredentials credentials) { - AllowOneRpc(); - - try + lock (myLock) { - var rpcInfo = newRpcQueue.Take(); - - // Console.WriteLine("Server received RPC " + rpcInfo.Method); - - IServerCallHandler callHandler; - if (!callHandlers.TryGetValue(rpcInfo.Method, out callHandler)) + Preconditions.CheckState(!startRequested); + using (var nativeCredentials = credentials.ToNativeCredentials()) { - callHandler = new NoSuchMethodCallHandler(); + return handle.AddListeningPort(addr, nativeCredentials); } - callHandler.StartCall(rpcInfo.Method, rpcInfo.Call, GetCompletionQueue()); } - catch (Exception e) + } + + /// <summary> + /// Starts the server. + /// </summary> + public void Start() + { + lock (myLock) { - Console.WriteLine("Exception while handling RPC: " + e); + Preconditions.CheckState(!startRequested); + startRequested = true; + + handle.Start(); + AllowOneRpc(); } } /// <summary> /// Requests server shutdown and when there are no more calls being serviced, - /// cleans up used resources. + /// cleans up used resources. The returned task finishes when shutdown procedure + /// is complete. /// </summary> - /// <returns>The async.</returns> public async Task ShutdownAsync() { + lock (myLock) + { + Preconditions.CheckState(startRequested); + Preconditions.CheckState(!shutdownRequested); + shutdownRequested = true; + } + handle.ShutdownAndNotify(serverShutdownHandler); await shutdownTcs.Task; handle.Dispose(); @@ -152,19 +164,43 @@ namespace Grpc.Core handle.Dispose(); } - private async Task StartHandlingRpcs() + /// <summary> + /// Allows one new RPC call to be received by server. + /// </summary> + private void AllowOneRpc() { - while (true) + lock (myLock) { - await Task.Factory.StartNew(RunRpc); + if (!shutdownRequested) + { + handle.RequestCall(GetCompletionQueue(), newServerRpcHandler); + } } } - private void AllowOneRpc() + /// <summary> + /// Selects corresponding handler for given call and handles the call. + /// </summary> + private void InvokeCallHandler(CallSafeHandle call, string method) { - AssertCallOk(handle.RequestCall(GetCompletionQueue(), newServerRpcHandler)); + try + { + IServerCallHandler callHandler; + if (!callHandlers.TryGetValue(method, out callHandler)) + { + callHandler = new NoSuchMethodCallHandler(); + } + callHandler.StartCall(method, call, GetCompletionQueue()); + } + catch (Exception e) + { + Console.WriteLine("Exception while handling RPC: " + e); + } } + /// <summary> + /// Handles the native callback. + /// </summary> private void HandleNewServerRpc(GRPCOpError error, IntPtr batchContextPtr) { try @@ -176,13 +212,16 @@ namespace Grpc.Core // TODO: handle error } - var rpcInfo = new NewRpcInfo(ctx.GetServerRpcNewCall(), ctx.GetServerRpcNewMethod()); + CallSafeHandle call = ctx.GetServerRpcNewCall(); + string method = ctx.GetServerRpcNewMethod(); // after server shutdown, the callback returns with null call - if (!rpcInfo.Call.IsInvalid) + if (!call.IsInvalid) { - newRpcQueue.Add(rpcInfo); + Task.Run(() => InvokeCallHandler(call, method)); } + + AllowOneRpc(); } catch (Exception e) { @@ -190,6 +229,10 @@ namespace Grpc.Core } } + /// <summary> + /// Handles native callback. + /// </summary> + /// <param name="eventPtr"></param> private void HandleServerShutdown(IntPtr eventPtr) { try @@ -202,42 +245,9 @@ namespace Grpc.Core } } - private static void AssertCallOk(GRPCCallError callError) - { - Trace.Assert(callError == GRPCCallError.GRPC_CALL_OK, "Status not GRPC_CALL_OK"); - } - private static CompletionQueueSafeHandle GetCompletionQueue() { return GrpcEnvironment.ThreadPool.CompletionQueue; } - - private struct NewRpcInfo - { - private CallSafeHandle call; - private string method; - - public NewRpcInfo(CallSafeHandle call, string method) - { - this.call = call; - this.method = method; - } - - public CallSafeHandle Call - { - get - { - return this.call; - } - } - - public string Method - { - get - { - return this.method; - } - } - } } } |