| /* |
| * |
| * Copyright 2017 gRPC authors. |
| * |
| * Licensed under the Apache License, Version 2.0 (the "License"); |
| * you may not use this file except in compliance with the License. |
| * You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| * |
| */ |
| |
| #include <grpc/support/port_platform.h> |
| |
| #include "src/core/lib/iomgr/call_combiner.h" |
| |
| #include <inttypes.h> |
| |
| #include <grpc/support/log.h> |
| #include "src/core/lib/debug/stats.h" |
| #include "src/core/lib/profiling/timers.h" |
| |
| grpc_core::TraceFlag grpc_call_combiner_trace(false, "call_combiner"); |
| |
| static grpc_error* decode_cancel_state_error(gpr_atm cancel_state) { |
| if (cancel_state & 1) { |
| return (grpc_error*)(cancel_state & ~static_cast<gpr_atm>(1)); |
| } |
| return GRPC_ERROR_NONE; |
| } |
| |
| static gpr_atm encode_cancel_state_error(grpc_error* error) { |
| return static_cast<gpr_atm>(1) | (gpr_atm)error; |
| } |
| |
| #ifdef GRPC_TSAN_ENABLED |
| static void tsan_closure(void* user_data, grpc_error* error) { |
| grpc_call_combiner* call_combiner = |
| static_cast<grpc_call_combiner*>(user_data); |
| // We ref-count the lock, and check if it's already taken. |
| // If it was taken, we should do nothing. Otherwise, we will mark it as |
| // locked. Note that if two different threads try to do this, only one of |
| // them will be able to mark the lock as acquired, while they both run their |
| // callbacks. In such cases (which should never happen for call_combiner), |
| // TSAN will correctly produce an error. |
| // |
| // TODO(soheil): This only covers the callbacks scheduled by |
| // grpc_call_combiner_(start|finish). If in the future, a |
| // callback gets scheduled using other mechanisms, we will need |
| // to add APIs to externally lock call combiners. |
| grpc_core::RefCountedPtr<grpc_call_combiner::TsanLock> lock = |
| call_combiner->tsan_lock; |
| bool prev = false; |
| if (lock->taken.compare_exchange_strong(prev, true)) { |
| TSAN_ANNOTATE_RWLOCK_ACQUIRED(&lock->taken, true); |
| } else { |
| lock.reset(); |
| } |
| GRPC_CLOSURE_RUN(call_combiner->original_closure, GRPC_ERROR_REF(error)); |
| if (lock != nullptr) { |
| TSAN_ANNOTATE_RWLOCK_RELEASED(&lock->taken, true); |
| bool prev = true; |
| GPR_ASSERT(lock->taken.compare_exchange_strong(prev, false)); |
| } |
| } |
| #endif |
| |
| static void call_combiner_sched_closure(grpc_call_combiner* call_combiner, |
| grpc_closure* closure, |
| grpc_error* error) { |
| #ifdef GRPC_TSAN_ENABLED |
| call_combiner->original_closure = closure; |
| GRPC_CLOSURE_SCHED(&call_combiner->tsan_closure, error); |
| #else |
| GRPC_CLOSURE_SCHED(closure, error); |
| #endif |
| } |
| |
| void grpc_call_combiner_init(grpc_call_combiner* call_combiner) { |
| gpr_atm_no_barrier_store(&call_combiner->cancel_state, 0); |
| gpr_atm_no_barrier_store(&call_combiner->size, 0); |
| gpr_mpscq_init(&call_combiner->queue); |
| #ifdef GRPC_TSAN_ENABLED |
| GRPC_CLOSURE_INIT(&call_combiner->tsan_closure, tsan_closure, call_combiner, |
| grpc_schedule_on_exec_ctx); |
| #endif |
| } |
| |
| void grpc_call_combiner_destroy(grpc_call_combiner* call_combiner) { |
| gpr_mpscq_destroy(&call_combiner->queue); |
| GRPC_ERROR_UNREF(decode_cancel_state_error(call_combiner->cancel_state)); |
| } |
| |
| #ifndef NDEBUG |
| #define DEBUG_ARGS , const char *file, int line |
| #define DEBUG_FMT_STR "%s:%d: " |
| #define DEBUG_FMT_ARGS , file, line |
| #else |
| #define DEBUG_ARGS |
| #define DEBUG_FMT_STR |
| #define DEBUG_FMT_ARGS |
| #endif |
| |
| void grpc_call_combiner_start(grpc_call_combiner* call_combiner, |
| grpc_closure* closure, |
| grpc_error* error DEBUG_ARGS, |
| const char* reason) { |
| GPR_TIMER_SCOPE("call_combiner_start", 0); |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, |
| "==> grpc_call_combiner_start() [%p] closure=%p [" DEBUG_FMT_STR |
| "%s] error=%s", |
| call_combiner, closure DEBUG_FMT_ARGS, reason, |
| grpc_error_string(error)); |
| } |
| size_t prev_size = static_cast<size_t>( |
| gpr_atm_full_fetch_add(&call_combiner->size, (gpr_atm)1)); |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, " size: %" PRIdPTR " -> %" PRIdPTR, prev_size, |
| prev_size + 1); |
| } |
| GRPC_STATS_INC_CALL_COMBINER_LOCKS_SCHEDULED_ITEMS(); |
| if (prev_size == 0) { |
| GRPC_STATS_INC_CALL_COMBINER_LOCKS_INITIATED(); |
| |
| GPR_TIMER_MARK("call_combiner_initiate", 0); |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, " EXECUTING IMMEDIATELY"); |
| } |
| // Queue was empty, so execute this closure immediately. |
| call_combiner_sched_closure(call_combiner, closure, error); |
| } else { |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, " QUEUING"); |
| } |
| // Queue was not empty, so add closure to queue. |
| closure->error_data.error = error; |
| gpr_mpscq_push(&call_combiner->queue, |
| reinterpret_cast<gpr_mpscq_node*>(closure)); |
| } |
| } |
| |
| void grpc_call_combiner_stop(grpc_call_combiner* call_combiner DEBUG_ARGS, |
| const char* reason) { |
| GPR_TIMER_SCOPE("call_combiner_stop", 0); |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, |
| "==> grpc_call_combiner_stop() [%p] [" DEBUG_FMT_STR "%s]", |
| call_combiner DEBUG_FMT_ARGS, reason); |
| } |
| size_t prev_size = static_cast<size_t>( |
| gpr_atm_full_fetch_add(&call_combiner->size, (gpr_atm)-1)); |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, " size: %" PRIdPTR " -> %" PRIdPTR, prev_size, |
| prev_size - 1); |
| } |
| GPR_ASSERT(prev_size >= 1); |
| if (prev_size > 1) { |
| while (true) { |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, " checking queue"); |
| } |
| bool empty; |
| grpc_closure* closure = reinterpret_cast<grpc_closure*>( |
| gpr_mpscq_pop_and_check_end(&call_combiner->queue, &empty)); |
| if (closure == nullptr) { |
| // This can happen either due to a race condition within the mpscq |
| // code or because of a race with grpc_call_combiner_start(). |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, " queue returned no result; checking again"); |
| } |
| continue; |
| } |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, " EXECUTING FROM QUEUE: closure=%p error=%s", |
| closure, grpc_error_string(closure->error_data.error)); |
| } |
| call_combiner_sched_closure(call_combiner, closure, |
| closure->error_data.error); |
| break; |
| } |
| } else if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, " queue empty"); |
| } |
| } |
| |
| void grpc_call_combiner_set_notify_on_cancel(grpc_call_combiner* call_combiner, |
| grpc_closure* closure) { |
| GRPC_STATS_INC_CALL_COMBINER_SET_NOTIFY_ON_CANCEL(); |
| while (true) { |
| // Decode original state. |
| gpr_atm original_state = gpr_atm_acq_load(&call_combiner->cancel_state); |
| grpc_error* original_error = decode_cancel_state_error(original_state); |
| // If error is set, invoke the cancellation closure immediately. |
| // Otherwise, store the new closure. |
| if (original_error != GRPC_ERROR_NONE) { |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, |
| "call_combiner=%p: scheduling notify_on_cancel callback=%p " |
| "for pre-existing cancellation", |
| call_combiner, closure); |
| } |
| GRPC_CLOSURE_SCHED(closure, GRPC_ERROR_REF(original_error)); |
| break; |
| } else { |
| if (gpr_atm_full_cas(&call_combiner->cancel_state, original_state, |
| (gpr_atm)closure)) { |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, "call_combiner=%p: setting notify_on_cancel=%p", |
| call_combiner, closure); |
| } |
| // If we replaced an earlier closure, invoke the original |
| // closure with GRPC_ERROR_NONE. This allows callers to clean |
| // up any resources they may be holding for the callback. |
| if (original_state != 0) { |
| closure = (grpc_closure*)original_state; |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, |
| "call_combiner=%p: scheduling old cancel callback=%p", |
| call_combiner, closure); |
| } |
| GRPC_CLOSURE_SCHED(closure, GRPC_ERROR_NONE); |
| } |
| break; |
| } |
| } |
| // cas failed, try again. |
| } |
| } |
| |
| void grpc_call_combiner_cancel(grpc_call_combiner* call_combiner, |
| grpc_error* error) { |
| GRPC_STATS_INC_CALL_COMBINER_CANCELLED(); |
| while (true) { |
| gpr_atm original_state = gpr_atm_acq_load(&call_combiner->cancel_state); |
| grpc_error* original_error = decode_cancel_state_error(original_state); |
| if (original_error != GRPC_ERROR_NONE) { |
| GRPC_ERROR_UNREF(error); |
| break; |
| } |
| if (gpr_atm_full_cas(&call_combiner->cancel_state, original_state, |
| encode_cancel_state_error(error))) { |
| if (original_state != 0) { |
| grpc_closure* notify_on_cancel = (grpc_closure*)original_state; |
| if (grpc_call_combiner_trace.enabled()) { |
| gpr_log(GPR_INFO, |
| "call_combiner=%p: scheduling notify_on_cancel callback=%p", |
| call_combiner, notify_on_cancel); |
| } |
| GRPC_CLOSURE_SCHED(notify_on_cancel, GRPC_ERROR_REF(error)); |
| } |
| break; |
| } |
| // cas failed, try again. |
| } |
| } |