blob: 6b5759a036f5f118903d614737e83ac62a54c518 [file] [log] [blame]
/*
*
* 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.
}
}