Merge pull request #1472 from ctiller/bye-bye-completion-queue-pie

C Core API cleanup.

9 files changed, 39 insertions, 85 deletions

diff --git a/src/csharp/Grpc.Core.Tests/PInvokeTest.cs b/src/csharp/Grpc.Core.Tests/PInvokeTest.cs
index 3beffc3955..26f87660df 100644
--- a/src/csharp/Grpc.Core.Tests/PInvokeTest.cs
+++ b/src/csharp/Grpc.Core.Tests/PInvokeTest.cs
@@ -134,7 +134,7 @@ namespace Grpc.Core.Tests
                 });
         }
 
-        private void Handler(GRPCOpError op, IntPtr ptr)
+        private void Handler(bool success, IntPtr ptr)
         {
             counter++;
         }
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
index 3532f7347a..9bb918d53d 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
@@ -274,7 +274,7 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handler for unary response completion.
         /// </summary>
-        private void HandleUnaryResponse(bool wasError, BatchContextSafeHandleNotOwned ctx)
+        private void HandleUnaryResponse(bool success, BatchContextSafeHandleNotOwned ctx)
         {
             lock (myLock)
             {
@@ -284,7 +284,7 @@ namespace Grpc.Core.Internal
                 ReleaseResourcesIfPossible();
             }
 
-            if (wasError)
+            if (!success)
             {
                 unaryResponseTcs.SetException(new RpcException(new Status(StatusCode.Internal, "Internal error occured.")));
                 return;
@@ -307,7 +307,7 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handles receive status completion for calls with streaming response.
         /// </summary>
-        private void HandleFinished(bool wasError, BatchContextSafeHandleNotOwned ctx)
+        private void HandleFinished(bool success, BatchContextSafeHandleNotOwned ctx)
         {
             var status = ctx.GetReceivedStatus();
 
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
index fc5bee40e2..b4f4edb17a 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
@@ -287,13 +287,12 @@ namespace Grpc.Core.Internal
         /// </summary>
         protected CompletionCallbackDelegate CreateBatchCompletionCallback(Action<bool, BatchContextSafeHandleNotOwned> handler)
         {
-            return new CompletionCallbackDelegate((error, batchContextPtr) =>
+            return new CompletionCallbackDelegate((success, batchContextPtr) =>
             {
                 try
                 {
                     var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
-                    bool wasError = (error != GRPCOpError.GRPC_OP_OK);
-                    handler(wasError, ctx);
+                    handler(success, ctx);
                 }
                 catch (Exception e)
                 {
@@ -305,7 +304,7 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handles send completion.
         /// </summary>
-        private void HandleSendFinished(bool wasError, BatchContextSafeHandleNotOwned ctx)
+        private void HandleSendFinished(bool success, BatchContextSafeHandleNotOwned ctx)
         {
             AsyncCompletionDelegate<object> origCompletionDelegate = null;
             lock (myLock)
@@ -316,7 +315,7 @@ namespace Grpc.Core.Internal
                 ReleaseResourcesIfPossible();
             }
 
-            if (wasError)
+            if (!success)
             {
                 FireCompletion(origCompletionDelegate, null, new OperationFailedException("Send failed"));
             }
@@ -329,7 +328,7 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handles halfclose completion.
         /// </summary>
-        private void HandleHalfclosed(bool wasError, BatchContextSafeHandleNotOwned ctx)
+        private void HandleHalfclosed(bool success, BatchContextSafeHandleNotOwned ctx)
         {
             AsyncCompletionDelegate<object> origCompletionDelegate = null;
             lock (myLock)
@@ -341,7 +340,7 @@ namespace Grpc.Core.Internal
                 ReleaseResourcesIfPossible();
             }
 
-            if (wasError)
+            if (!success)
             {
                 FireCompletion(origCompletionDelegate, null, new OperationFailedException("Halfclose failed"));
             }
@@ -354,7 +353,7 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handles streaming read completion.
         /// </summary>
-        private void HandleReadFinished(bool wasError, BatchContextSafeHandleNotOwned ctx)
+        private void HandleReadFinished(bool success, BatchContextSafeHandleNotOwned ctx)
         {
             var payload = ctx.GetReceivedMessage();
 
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
index 171d0c799d..1f0335e4e6 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
@@ -121,7 +121,7 @@ namespace Grpc.Core.Internal
         /// <summary>
         /// Handles the server side close completion.
         /// </summary>
-        private void HandleFinishedServerside(bool wasError, BatchContextSafeHandleNotOwned ctx)
+        private void HandleFinishedServerside(bool success, BatchContextSafeHandleNotOwned ctx)
         {
             bool cancelled = ctx.GetReceivedCloseOnServerCancelled();
 
diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
index c97a3bc2b1..491b8414ec 100644
--- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
@@ -37,7 +37,7 @@ using Grpc.Core.Utils;
 
 namespace Grpc.Core.Internal
 {
-    internal delegate void CompletionCallbackDelegate(GRPCOpError error, IntPtr batchContextPtr);
+    internal delegate void CompletionCallbackDelegate(bool success, IntPtr batchContextPtr);
     
     /// <summary>
     /// grpc_call from <grpc/grpc.h>
diff --git a/src/csharp/Grpc.Core/Internal/Enums.cs b/src/csharp/Grpc.Core/Internal/Enums.cs
index 94a2fd1784..2b4f6cae0c 100644
--- a/src/csharp/Grpc.Core/Internal/Enums.cs
+++ b/src/csharp/Grpc.Core/Internal/Enums.cs
@@ -70,45 +70,12 @@ namespace Grpc.Core.Internal
     internal enum GRPCCompletionType
     {
         /* Shutting down */
-        GRPC_QUEUE_SHUTDOWN,
+        GRPC_QUEUE_SHUTDOWN, 
 
-        /* operation completion */
-        GRPC_OP_COMPLETE,
-
-        /* A read has completed */
-        GRPC_READ,
-
-        /* A write has been accepted by flow control */
-        GRPC_WRITE_ACCEPTED,
-
-        /* writes_done or write_status has been accepted */
-        GRPC_FINISH_ACCEPTED,
-
-        /* The metadata array sent by server received at client */
-        GRPC_CLIENT_METADATA_READ,
-
-        /* An RPC has finished. The event contains status.
-         * On the server this will be OK or Cancelled. */
-        GRPC_FINISHED,
-
-        /* A new RPC has arrived at the server */
-        GRPC_SERVER_RPC_NEW,
-
-        /* The server has finished shutting down */
-        GRPC_SERVER_SHUTDOWN,
+        /* No event before timeout */
+        GRPC_QUEUE_TIMEOUT,  
 
-        /* must be last, forces users to include a default: case */
-        GRPC_COMPLETION_DO_NOT_USE
-    }
-
-    /// <summary>
-    /// grpc_op_error from grpc/grpc.h
-    /// </summary>
-    internal enum GRPCOpError
-    {
-        /* everything went ok */
-        GRPC_OP_OK = 0,
-        /* something failed, we don't know what */
-        GRPC_OP_ERROR
+        /* operation completion */
+        GRPC_OP_COMPLETE
     }
 }
diff --git a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
index 8080643d8c..731ea2be81 100644
--- a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
@@ -40,7 +40,7 @@ using Grpc.Core.Utils;
 namespace Grpc.Core.Internal
 {
     // TODO: we need to make sure that the delegates are not collected before invoked.
-    internal delegate void ServerShutdownCallbackDelegate(IntPtr eventPtr);
+    //internal delegate void ServerShutdownCallbackDelegate(bool success);
 
     /// <summary>
     /// grpc_server from grpc/grpc.h
@@ -65,9 +65,8 @@ namespace Grpc.Core.Internal
         [DllImport("grpc_csharp_ext.dll")]
         static extern void grpcsharp_server_shutdown(ServerSafeHandle server);
 
-        // TODO: get rid of the old callback style
-        [DllImport("grpc_csharp_ext.dll", EntryPoint = "grpcsharp_server_shutdown_and_notify")]
-        static extern void grpcsharp_server_shutdown_and_notify_CALLBACK(ServerSafeHandle server, [MarshalAs(UnmanagedType.FunctionPtr)] ServerShutdownCallbackDelegate callback);
+        [DllImport("grpc_csharp_ext.dll")]
+        static extern void grpcsharp_server_shutdown_and_notify_callback(ServerSafeHandle server, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
 
         [DllImport("grpc_csharp_ext.dll")]
         static extern void grpcsharp_server_destroy(IntPtr server);
@@ -101,9 +100,9 @@ namespace Grpc.Core.Internal
             grpcsharp_server_shutdown(this);
         }
 
-        public void ShutdownAndNotify(ServerShutdownCallbackDelegate callback)
+        public void ShutdownAndNotify(CompletionCallbackDelegate callback)
         {
-            grpcsharp_server_shutdown_and_notify_CALLBACK(this, callback);
+            grpcsharp_server_shutdown_and_notify_callback(this, callback);
         }
 
         public void RequestCall(CompletionQueueSafeHandle cq, CompletionCallbackDelegate callback)
diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs
index 0df46bb25b..4a7abbb33f 100644
--- a/src/csharp/Grpc.Core/Server.cs
+++ b/src/csharp/Grpc.Core/Server.cs
@@ -54,7 +54,7 @@ namespace Grpc.Core
 
         // TODO(jtattermusch) : make sure the delegate doesn't get garbage collected while
         // native callbacks are in the completion queue.
-        readonly ServerShutdownCallbackDelegate serverShutdownHandler;
+        readonly CompletionCallbackDelegate serverShutdownHandler;
         readonly CompletionCallbackDelegate newServerRpcHandler;
 
         readonly ServerSafeHandle handle;
@@ -222,16 +222,13 @@ namespace Grpc.Core
         /// <summary>
         /// Handles the native callback.
         /// </summary>
-        private void HandleNewServerRpc(GRPCOpError error, IntPtr batchContextPtr)
+        private void HandleNewServerRpc(bool success, IntPtr batchContextPtr)
         {
             try
             {
                 var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
 
-                if (error != GRPCOpError.GRPC_OP_OK)
-                {
-                    // TODO: handle error
-                }
+                // TODO: handle error
 
                 CallSafeHandle call = ctx.GetServerRpcNewCall();
                 string method = ctx.GetServerRpcNewMethod();
@@ -253,8 +250,7 @@ namespace Grpc.Core
         /// <summary>
         /// Handles native callback.
         /// </summary>
-        /// <param name="eventPtr"></param>
-        private void HandleServerShutdown(IntPtr eventPtr)
+        private void HandleServerShutdown(bool success, IntPtr batchContextPtr)
         {
             try
             {
diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c
index 6cefacece2..cea23f019e 100644
--- a/src/csharp/ext/grpc_csharp_ext.c
+++ b/src/csharp/ext/grpc_csharp_ext.c
@@ -63,8 +63,7 @@ grpc_byte_buffer *string_to_byte_buffer(const char *buffer, size_t len) {
   return bb;
 }
 
-typedef void(GPR_CALLTYPE *callback_funcptr)(grpc_op_error op_error,
-                                             void *batch_context);
+typedef void(GPR_CALLTYPE *callback_funcptr)(gpr_int32 success, void *batch_context);
 
 /*
  * Helper to maintain lifetime of batch op inputs and store batch op outputs.
@@ -308,27 +307,18 @@ grpcsharp_completion_queue_destroy(grpc_completion_queue *cq) {
 
 GPR_EXPORT grpc_completion_type GPR_CALLTYPE
 grpcsharp_completion_queue_next_with_callback(grpc_completion_queue *cq) {
-  grpc_event *ev;
+  grpc_event ev;
   grpcsharp_batch_context *batch_context;
   grpc_completion_type t;
-  void(GPR_CALLTYPE * callback)(grpc_event *);
 
   ev = grpc_completion_queue_next(cq, gpr_inf_future);
-  t = ev->type;
-  if (t == GRPC_OP_COMPLETE && ev->tag) {
+  t = ev.type;
+  if (t == GRPC_OP_COMPLETE && ev.tag) {
     /* NEW API handler */
-    batch_context = (grpcsharp_batch_context *)ev->tag;
-    batch_context->callback(ev->data.op_complete, batch_context);
+    batch_context = (grpcsharp_batch_context *)ev.tag;
+    batch_context->callback((gpr_int32) ev.success, batch_context);
     grpcsharp_batch_context_destroy(batch_context);
-  } else if (ev->tag) {
-    /* call the callback in ev->tag */
-    /* C forbids to cast object pointers to function pointers, so
-     * we cast to intptr first.
-     */
-    callback = (void(GPR_CALLTYPE *)(grpc_event *))(gpr_intptr)ev->tag;
-    (*callback)(ev);
   }
-  grpc_event_finish(ev);
 
   /* return completion type to allow some handling for events that have no
    * tag - such as GRPC_QUEUE_SHUTDOWN
@@ -692,8 +682,11 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_server_shutdown(grpc_server *server) {
 }
 
 GPR_EXPORT void GPR_CALLTYPE
-grpcsharp_server_shutdown_and_notify(grpc_server *server, void *tag) {
-  grpc_server_shutdown_and_notify(server, tag);
+grpcsharp_server_shutdown_and_notify_callback(grpc_server *server,
+                                              callback_funcptr callback) {
+  grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+  ctx->callback = callback;
+  grpc_server_shutdown_and_notify(server, ctx);
 }
 
 GPR_EXPORT void GPR_CALLTYPE grpcsharp_server_destroy(grpc_server *server) {
@@ -797,7 +790,7 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_redirect_log(grpcsharp_log_func func) {
 /* For testing */
 GPR_EXPORT void GPR_CALLTYPE
 grpcsharp_test_callback(callback_funcptr callback) {
-  callback(GRPC_OP_OK, NULL);
+  callback(1, NULL);
 }
 
 /* For testing */