Merge pull request #16288 from yashykt/combinernotify

Explictly Flush exec_ctx after resetting call_combiner_set_notify_on_…

src/core/lib/iomgr/call_combiner.h

@@ -102,7 +102,10 @@ void grpc_call_combiner_stop(grpc_call_combiner* call_combiner,
 /// If \a closure is NULL, then no closure will be invoked on
 /// cancellation; this effectively unregisters the previously set closure.
 /// However, most filters will not need to explicitly unregister their
-/// callbacks, as this is done automatically when the call is destroyed.
+/// callbacks, as this is done automatically when the call is destroyed. Filters
+/// that schedule the cancellation closure on ExecCtx do not need to take a ref
+/// on the call stack to guarantee closure liveness. This is done by explicitly
+/// flushing ExecCtx after the unregistration during call destruction.
 void grpc_call_combiner_set_notify_on_cancel(grpc_call_combiner* call_combiner,
                                              grpc_closure* closure);
 

src/core/lib/security/transport/client_auth_filter.cc

@@ -167,7 +167,6 @@ static void cancel_get_request_metadata(void* arg, grpc_error* error) {
     grpc_call_credentials_cancel_get_request_metadata(
         calld->creds, &calld->md_array, GRPC_ERROR_REF(error));
   }
-  GRPC_CALL_STACK_UNREF(calld->owning_call, "cancel_get_request_metadata");
 }
 
 static void send_security_metadata(grpc_call_element* elem,
@@ -222,7 +221,6 @@ static void send_security_metadata(grpc_call_element* elem,
     GRPC_ERROR_UNREF(error);
   } else {
     // Async return; register cancellation closure with call combiner.
-    GRPC_CALL_STACK_REF(calld->owning_call, "cancel_get_request_metadata");
     grpc_call_combiner_set_notify_on_cancel(
         calld->call_combiner,
         GRPC_CLOSURE_INIT(&calld->get_request_metadata_cancel_closure,
@@ -265,7 +263,6 @@ static void cancel_check_call_host(void* arg, grpc_error* error) {
         chand->security_connector, &calld->async_result_closure,
         GRPC_ERROR_REF(error));
   }
-  GRPC_CALL_STACK_UNREF(calld->owning_call, "cancel_check_call_host");
 }
 
 static void auth_start_transport_stream_op_batch(
@@ -318,7 +315,6 @@ static void auth_start_transport_stream_op_batch(
         GRPC_ERROR_UNREF(error);
       } else {
         // Async return; register cancellation closure with call combiner.
-        GRPC_CALL_STACK_REF(calld->owning_call, "cancel_check_call_host");
         grpc_call_combiner_set_notify_on_cancel(
             calld->call_combiner,
             GRPC_CLOSURE_INIT(&calld->check_call_host_cancel_closure,

src/core/lib/security/transport/server_auth_filter.cc

@@ -156,7 +156,6 @@ static void cancel_call(void* arg, grpc_error* error) {
     on_md_processing_done_inner(elem, nullptr, 0, nullptr, 0,
                                 GRPC_ERROR_REF(error));
   }
-  GRPC_CALL_STACK_UNREF(calld->owning_call, "cancel_call");
 }
 
 static void recv_initial_metadata_ready(void* arg, grpc_error* error) {
@@ -168,7 +167,6 @@ static void recv_initial_metadata_ready(void* arg, grpc_error* error) {
     if (chand->creds != nullptr && chand->creds->processor.process != nullptr) {
       // We're calling out to the application, so we need to make sure
       // to drop the call combiner early if we get cancelled.
-      GRPC_CALL_STACK_REF(calld->owning_call, "cancel_call");
       GRPC_CLOSURE_INIT(&calld->cancel_closure, cancel_call, elem,
                         grpc_schedule_on_exec_ctx);
       grpc_call_combiner_set_notify_on_cancel(calld->call_combiner,

src/core/lib/surface/call.cc

@@ -613,8 +613,11 @@ void grpc_call_unref(grpc_call* c) {
     // Unset the call combiner cancellation closure.  This has the
     // effect of scheduling the previously set cancellation closure, if
     // any, so that it can release any internal references it may be
-    // holding to the call stack.
+    // holding to the call stack. Also flush the closures on exec_ctx so that
+    // filters that schedule cancel notification closures on exec_ctx do not
+    // need to take a ref of the call stack to guarantee closure liveness.
     grpc_call_combiner_set_notify_on_cancel(&c->call_combiner, nullptr);
+    grpc_core::ExecCtx::Get()->Flush();
   }
   GRPC_CALL_INTERNAL_UNREF(c, "destroy");
 }