Google Git
Sign in
chromium / external / github.com / grpc / grpc / 64b422091e9eec82951402f0b04d053db521c4e2 / . / src / core / lib / surface / completion_queue.cc
blob: 70bae71fdb09afc31745a73fb298938075ea9bea [file] [log] [blame]
/*
*
* Copyright 2015-2016 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/surface/completion_queue.h"
#include <inttypes.h>
#include <stdio.h>
#include <algorithm>
#include <atomic>
#include <new>
#include <string>
#include <utility>
#include <vector>
#include "absl/strings/str_format.h"
#include "absl/strings/str_join.h"
#include <grpc/grpc.h>
#include <grpc/impl/codegen/gpr_types.h>
#include <grpc/support/alloc.h>
#include <grpc/support/atm.h>
#include <grpc/support/log.h>
#include <grpc/support/sync.h>
#include "src/core/lib/debug/stats.h"
#include "src/core/lib/gpr/spinlock.h"
#include "src/core/lib/gpr/tls.h"
#include "src/core/lib/gprpp/atomic_utils.h"
#include "src/core/lib/gprpp/debug_location.h"
#include "src/core/lib/gprpp/ref_counted.h"
#include "src/core/lib/gprpp/time.h"
#include "src/core/lib/iomgr/closure.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/iomgr/executor.h"
#include "src/core/lib/iomgr/iomgr.h"
#include "src/core/lib/iomgr/pollset.h"
#include "src/core/lib/profiling/timers.h"
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/event_string.h"
grpc_core::TraceFlag grpc_trace_operation_failures(false, "op_failure");
grpc_core::DebugOnlyTraceFlag grpc_trace_pending_tags(false, "pending_tags");
grpc_core::DebugOnlyTraceFlag grpc_trace_cq_refcount(false, "cq_refcount");
namespace {
// Specifies a cq thread local cache.
// The first event that occurs on a thread
// with a cq cache will go into that cache, and
// will only be returned on the thread that initialized the cache.
// NOTE: Only one event will ever be cached.
GPR_THREAD_LOCAL(grpc_cq_completion*) g_cached_event;
GPR_THREAD_LOCAL(grpc_completion_queue*) g_cached_cq;
struct plucker {
grpc_pollset_worker** worker;
void* tag;
};
struct cq_poller_vtable {
bool can_get_pollset;
bool can_listen;
size_t (*size)(void);
void (*init)(grpc_pollset* pollset, gpr_mu** mu);
grpc_error_handle (*kick)(grpc_pollset* pollset,
grpc_pollset_worker* specific_worker);
grpc_error_handle (*work)(grpc_pollset* pollset, grpc_pollset_worker** worker,
grpc_core::Timestamp deadline);
void (*shutdown)(grpc_pollset* pollset, grpc_closure* closure);
void (*destroy)(grpc_pollset* pollset);
};
typedef struct non_polling_worker {
gpr_cv cv;
bool kicked;
struct non_polling_worker* next;
struct non_polling_worker* prev;
} non_polling_worker;
struct non_polling_poller {
gpr_mu mu;
bool kicked_without_poller;
non_polling_worker* root;
grpc_closure* shutdown;
};
size_t non_polling_poller_size(void) { return sizeof(non_polling_poller); }
void non_polling_poller_init(grpc_pollset* pollset, gpr_mu** mu) {
non_polling_poller* npp = reinterpret_cast<non_polling_poller*>(pollset);
gpr_mu_init(&npp->mu);
*mu = &npp->mu;
}
void non_polling_poller_destroy(grpc_pollset* pollset) {
non_polling_poller* npp = reinterpret_cast<non_polling_poller*>(pollset);
gpr_mu_destroy(&npp->mu);
}
grpc_error_handle non_polling_poller_work(grpc_pollset* pollset,
grpc_pollset_worker** worker,
grpc_core::Timestamp deadline) {
non_polling_poller* npp = reinterpret_cast<non_polling_poller*>(pollset);
if (npp->shutdown) return GRPC_ERROR_NONE;
if (npp->kicked_without_poller) {
npp->kicked_without_poller = false;
return GRPC_ERROR_NONE;
}
non_polling_worker w;
gpr_cv_init(&w.cv);
if (worker != nullptr) *worker = reinterpret_cast<grpc_pollset_worker*>(&w);
if (npp->root == nullptr) {
npp->root = w.next = w.prev = &w;
} else {
w.next = npp->root;
w.prev = w.next->prev;
w.next->prev = w.prev->next = &w;
}
w.kicked = false;
gpr_timespec deadline_ts = deadline.as_timespec(GPR_CLOCK_MONOTONIC);
while (!npp->shutdown && !w.kicked &&
!gpr_cv_wait(&w.cv, &npp->mu, deadline_ts)) {
}
grpc_core::ExecCtx::Get()->InvalidateNow();
if (&w == npp->root) {
npp->root = w.next;
if (&w == npp->root) {
if (npp->shutdown) {
grpc_core::ExecCtx::Run(DEBUG_LOCATION, npp->shutdown, GRPC_ERROR_NONE);
}
npp->root = nullptr;
}
}
w.next->prev = w.prev;
w.prev->next = w.next;
gpr_cv_destroy(&w.cv);
if (worker != nullptr) *worker = nullptr;
return GRPC_ERROR_NONE;
}
grpc_error_handle non_polling_poller_kick(
grpc_pollset* pollset, grpc_pollset_worker* specific_worker) {
non_polling_poller* p = reinterpret_cast<non_polling_poller*>(pollset);
if (specific_worker == nullptr) {
specific_worker = reinterpret_cast<grpc_pollset_worker*>(p->root);
}
if (specific_worker != nullptr) {
non_polling_worker* w =
reinterpret_cast<non_polling_worker*>(specific_worker);
if (!w->kicked) {
w->kicked = true;
gpr_cv_signal(&w->cv);
}
} else {
p->kicked_without_poller = true;
}
return GRPC_ERROR_NONE;
}
void non_polling_poller_shutdown(grpc_pollset* pollset, grpc_closure* closure) {
non_polling_poller* p = reinterpret_cast<non_polling_poller*>(pollset);
GPR_ASSERT(closure != nullptr);
p->shutdown = closure;
if (p->root == nullptr) {
grpc_core::ExecCtx::Run(DEBUG_LOCATION, closure, GRPC_ERROR_NONE);
} else {
non_polling_worker* w = p->root;
do {
gpr_cv_signal(&w->cv);
w = w->next;
} while (w != p->root);
}
}
const cq_poller_vtable g_poller_vtable_by_poller_type[] = {
/* GRPC_CQ_DEFAULT_POLLING */
{true, true, grpc_pollset_size, grpc_pollset_init, grpc_pollset_kick,
grpc_pollset_work, grpc_pollset_shutdown, grpc_pollset_destroy},
/* GRPC_CQ_NON_LISTENING */
{true, false, grpc_pollset_size, grpc_pollset_init, grpc_pollset_kick,
grpc_pollset_work, grpc_pollset_shutdown, grpc_pollset_destroy},
/* GRPC_CQ_NON_POLLING */
{false, false, non_polling_poller_size, non_polling_poller_init,
non_polling_poller_kick, non_polling_poller_work,
non_polling_poller_shutdown, non_polling_poller_destroy},
};
} // namespace
struct cq_vtable {
grpc_cq_completion_type cq_completion_type;
size_t data_size;
void (*init)(void* data, grpc_completion_queue_functor* shutdown_callback);
void (*shutdown)(grpc_completion_queue* cq);
void (*destroy)(void* data);
bool (*begin_op)(grpc_completion_queue* cq, void* tag);
void (*end_op)(grpc_completion_queue* cq, void* tag, grpc_error_handle error,
void (*done)(void* done_arg, grpc_cq_completion* storage),
void* done_arg, grpc_cq_completion* storage, bool internal);
grpc_event (*next)(grpc_completion_queue* cq, gpr_timespec deadline,
void* reserved);
grpc_event (*pluck)(grpc_completion_queue* cq, void* tag,
gpr_timespec deadline, void* reserved);
};
namespace {
/* Queue that holds the cq_completion_events. Internally uses
* MultiProducerSingleConsumerQueue (a lockfree multiproducer single consumer
* queue). It uses a queue_lock to support multiple consumers.
* Only used in completion queues whose completion_type is GRPC_CQ_NEXT */
class CqEventQueue {
public:
CqEventQueue() = default;
~CqEventQueue() = default;
/* Note: The counter is not incremented/decremented atomically with push/pop.
* The count is only eventually consistent */
intptr_t num_items() const {
return num_queue_items_.load(std::memory_order_relaxed);
}
bool Push(grpc_cq_completion* c);
grpc_cq_completion* Pop();
private:
/* Spinlock to serialize consumers i.e pop() operations */
gpr_spinlock queue_lock_ = GPR_SPINLOCK_INITIALIZER;
grpc_core::MultiProducerSingleConsumerQueue queue_;
/* A lazy counter of number of items in the queue. This is NOT atomically
incremented/decremented along with push/pop operations and hence is only
eventually consistent */
std::atomic<intptr_t> num_queue_items_{0};
};
struct cq_next_data {
~cq_next_data() {
GPR_ASSERT(queue.num_items() == 0);
#ifndef NDEBUG
if (pending_events.load(std::memory_order_acquire) != 0) {
gpr_log(GPR_ERROR, "Destroying CQ without draining it fully.");
}
#endif
}
/** Completed events for completion-queues of type GRPC_CQ_NEXT */
CqEventQueue queue;
/** Counter of how many things have ever been queued on this completion queue
useful for avoiding locks to check the queue */
std::atomic<intptr_t> things_queued_ever{0};
/** Number of outstanding events (+1 if not shut down)
Initial count is dropped by grpc_completion_queue_shutdown */
std::atomic<intptr_t> pending_events{1};
/** 0 initially. 1 once we initiated shutdown */
bool shutdown_called = false;
};
struct cq_pluck_data {
cq_pluck_data() {
completed_tail = &completed_head;
completed_head.next = reinterpret_cast<uintptr_t>(completed_tail);
}
~cq_pluck_data() {
GPR_ASSERT(completed_head.next ==
reinterpret_cast<uintptr_t>(&completed_head));
#ifndef NDEBUG
if (pending_events.load(std::memory_order_acquire) != 0) {
gpr_log(GPR_ERROR, "Destroying CQ without draining it fully.");
}
#endif
}
/** Completed events for completion-queues of type GRPC_CQ_PLUCK */
grpc_cq_completion completed_head;
grpc_cq_completion* completed_tail;
/** Number of pending events (+1 if we're not shutdown).
Initial count is dropped by grpc_completion_queue_shutdown. */
std::atomic<intptr_t> pending_events{1};
/** Counter of how many things have ever been queued on this completion queue
useful for avoiding locks to check the queue */
std::atomic<intptr_t> things_queued_ever{0};
/** 0 initially. 1 once we completed shutting */
/* TODO: (sreek) This is not needed since (shutdown == 1) if and only if
* (pending_events == 0). So consider removing this in future and use
* pending_events */
std::atomic<bool> shutdown{false};
/** 0 initially. 1 once we initiated shutdown */
bool shutdown_called = false;
int num_pluckers = 0;
plucker pluckers[GRPC_MAX_COMPLETION_QUEUE_PLUCKERS];
};
struct cq_callback_data {
explicit cq_callback_data(grpc_completion_queue_functor* shutdown_callback)
: shutdown_callback(shutdown_callback) {}
~cq_callback_data() {
#ifndef NDEBUG
if (pending_events.load(std::memory_order_acquire) != 0) {
gpr_log(GPR_ERROR, "Destroying CQ without draining it fully.");
}
#endif
}
/** No actual completed events queue, unlike other types */
/** Number of pending events (+1 if we're not shutdown).
Initial count is dropped by grpc_completion_queue_shutdown. */
std::atomic<intptr_t> pending_events{1};
/** 0 initially. 1 once we initiated shutdown */
bool shutdown_called = false;
/** A callback that gets invoked when the CQ completes shutdown */
grpc_completion_queue_functor* shutdown_callback;
};
} // namespace
/* Completion queue structure */
struct grpc_completion_queue {
/** Once owning_refs drops to zero, we will destroy the cq */
grpc_core::RefCount owning_refs;
gpr_mu* mu;
const cq_vtable* vtable;
const cq_poller_vtable* poller_vtable;
#ifndef NDEBUG
void** outstanding_tags;
size_t outstanding_tag_count;
size_t outstanding_tag_capacity;
#endif
grpc_closure pollset_shutdown_done;
int num_polls;
};
/* Forward declarations */
static void cq_finish_shutdown_next(grpc_completion_queue* cq);
static void cq_finish_shutdown_pluck(grpc_completion_queue* cq);
static void cq_finish_shutdown_callback(grpc_completion_queue* cq);
static void cq_shutdown_next(grpc_completion_queue* cq);
static void cq_shutdown_pluck(grpc_completion_queue* cq);
static void cq_shutdown_callback(grpc_completion_queue* cq);
static bool cq_begin_op_for_next(grpc_completion_queue* cq, void* tag);
static bool cq_begin_op_for_pluck(grpc_completion_queue* cq, void* tag);
static bool cq_begin_op_for_callback(grpc_completion_queue* cq, void* tag);
// A cq_end_op function is called when an operation on a given CQ with
// a given tag has completed. The storage argument is a reference to the
// space reserved for this completion as it is placed into the corresponding
// queue. The done argument is a callback that will be invoked when it is
// safe to free up that storage. The storage MUST NOT be freed until the
// done callback is invoked.
static void cq_end_op_for_next(
grpc_completion_queue* cq, void* tag, grpc_error_handle error,
void (*done)(void* done_arg, grpc_cq_completion* storage), void* done_arg,
grpc_cq_completion* storage, bool internal);
static void cq_end_op_for_pluck(
grpc_completion_queue* cq, void* tag, grpc_error_handle error,
void (*done)(void* done_arg, grpc_cq_completion* storage), void* done_arg,
grpc_cq_completion* storage, bool internal);
static void cq_end_op_for_callback(
grpc_completion_queue* cq, void* tag, grpc_error_handle error,
void (*done)(void* done_arg, grpc_cq_completion* storage), void* done_arg,
grpc_cq_completion* storage, bool internal);
static grpc_event cq_next(grpc_completion_queue* cq, gpr_timespec deadline,
void* reserved);
static grpc_event cq_pluck(grpc_completion_queue* cq, void* tag,
gpr_timespec deadline, void* reserved);
// Note that cq_init_next and cq_init_pluck do not use the shutdown_callback
static void cq_init_next(void* data,
grpc_completion_queue_functor* shutdown_callback);
static void cq_init_pluck(void* data,
grpc_completion_queue_functor* shutdown_callback);
static void cq_init_callback(void* data,
grpc_completion_queue_functor* shutdown_callback);
static void cq_destroy_next(void* data);
static void cq_destroy_pluck(void* data);
static void cq_destroy_callback(void* data);
/* Completion queue vtables based on the completion-type */
static const cq_vtable g_cq_vtable[] = {
/* GRPC_CQ_NEXT */
{GRPC_CQ_NEXT, sizeof(cq_next_data), cq_init_next, cq_shutdown_next,
cq_destroy_next, cq_begin_op_for_next, cq_end_op_for_next, cq_next,
nullptr},
/* GRPC_CQ_PLUCK */
{GRPC_CQ_PLUCK, sizeof(cq_pluck_data), cq_init_pluck, cq_shutdown_pluck,
cq_destroy_pluck, cq_begin_op_for_pluck, cq_end_op_for_pluck, nullptr,
cq_pluck},
/* GRPC_CQ_CALLBACK */
{GRPC_CQ_CALLBACK, sizeof(cq_callback_data), cq_init_callback,
cq_shutdown_callback, cq_destroy_callback, cq_begin_op_for_callback,
cq_end_op_for_callback, nullptr, nullptr},
};
#define DATA_FROM_CQ(cq) ((void*)((cq) + 1))
#define POLLSET_FROM_CQ(cq) \
((grpc_pollset*)((cq)->vtable->data_size + (char*)DATA_FROM_CQ(cq)))
grpc_core::TraceFlag grpc_cq_pluck_trace(false, "queue_pluck");
#define GRPC_SURFACE_TRACE_RETURNED_EVENT(cq, event) \
do { \
if (GRPC_TRACE_FLAG_ENABLED(grpc_api_trace) && \
(GRPC_TRACE_FLAG_ENABLED(grpc_cq_pluck_trace) || \
(event)->type != GRPC_QUEUE_TIMEOUT)) { \
gpr_log(GPR_INFO, "RETURN_EVENT[%p]: %s", cq, \
grpc_event_string(event).c_str()); \
} \
} while (0)
static void on_pollset_shutdown_done(void* arg, grpc_error_handle error);
void grpc_cq_global_init() {}
void grpc_completion_queue_thread_local_cache_init(grpc_completion_queue* cq) {
if (g_cached_cq == nullptr) {
g_cached_event = nullptr;
g_cached_cq = cq;
}
}
int grpc_completion_queue_thread_local_cache_flush(grpc_completion_queue* cq,
void** tag, int* ok) {
grpc_cq_completion* storage = g_cached_event;
int ret = 0;
if (storage != nullptr && g_cached_cq == cq) {
*tag = storage->tag;
grpc_core::ExecCtx exec_ctx;
*ok = (storage->next & static_cast<uintptr_t>(1)) == 1;
storage->done(storage->done_arg, storage);
ret = 1;
cq_next_data* cqd = static_cast<cq_next_data*> DATA_FROM_CQ(cq);
if (cqd->pending_events.fetch_sub(1, std::memory_order_acq_rel) == 1) {
GRPC_CQ_INTERNAL_REF(cq, "shutting_down");
gpr_mu_lock(cq->mu);
cq_finish_shutdown_next(cq);
gpr_mu_unlock(cq->mu);
GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down");
}
}
g_cached_event = nullptr;
g_cached_cq = nullptr;
return ret;
}
bool CqEventQueue::Push(grpc_cq_completion* c) {
queue_.Push(
reinterpret_cast<grpc_core::MultiProducerSingleConsumerQueue::Node*>(c));
return num_queue_items_.fetch_add(1, std::memory_order_relaxed) == 0;
}
grpc_cq_completion* CqEventQueue::Pop() {
grpc_cq_completion* c = nullptr;
if (gpr_spinlock_trylock(&queue_lock_)) {
GRPC_STATS_INC_CQ_EV_QUEUE_TRYLOCK_SUCCESSES();
bool is_empty = false;
c = reinterpret_cast<grpc_cq_completion*>(queue_.PopAndCheckEnd(&is_empty));
gpr_spinlock_unlock(&queue_lock_);
if (c == nullptr && !is_empty) {
GRPC_STATS_INC_CQ_EV_QUEUE_TRANSIENT_POP_FAILURES();
}
} else {
GRPC_STATS_INC_CQ_EV_QUEUE_TRYLOCK_FAILURES();
}
if (c) {
num_queue_items_.fetch_sub(1, std::memory_order_relaxed);
}
return c;
}
grpc_completion_queue* grpc_completion_queue_create_internal(
grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type,
grpc_completion_queue_functor* shutdown_callback) {
GPR_TIMER_SCOPE("grpc_completion_queue_create_internal", 0);
grpc_completion_queue* cq;
GRPC_API_TRACE(
"grpc_completion_queue_create_internal(completion_type=%d, "
"polling_type=%d)",
2, (completion_type, polling_type));
const cq_vtable* vtable = &g_cq_vtable[completion_type];
const cq_poller_vtable* poller_vtable =
&g_poller_vtable_by_poller_type[polling_type];
grpc_core::ExecCtx exec_ctx;
GRPC_STATS_INC_CQS_CREATED();
cq = static_cast<grpc_completion_queue*>(
gpr_zalloc(sizeof(grpc_completion_queue) + vtable->data_size +
poller_vtable->size()));
cq->vtable = vtable;
cq->poller_vtable = poller_vtable;
/* One for destroy(), one for pollset_shutdown */
new (&cq->owning_refs) grpc_core::RefCount(2);
poller_vtable->init(POLLSET_FROM_CQ(cq), &cq->mu);
vtable->init(DATA_FROM_CQ(cq), shutdown_callback);
GRPC_CLOSURE_INIT(&cq->pollset_shutdown_done, on_pollset_shutdown_done, cq,
grpc_schedule_on_exec_ctx);
return cq;
}
static void cq_init_next(void* data,
grpc_completion_queue_functor* /*shutdown_callback*/) {
new (data) cq_next_data();
}
static void cq_destroy_next(void* data) {
cq_next_data* cqd = static_cast<cq_next_data*>(data);
cqd->~cq_next_data();
}
static void cq_init_pluck(
void* data, grpc_completion_queue_functor* /*shutdown_callback*/) {
new (data) cq_pluck_data();
}
static void cq_destroy_pluck(void* data) {
cq_pluck_data* cqd = static_cast<cq_pluck_data*>(data);
cqd->~cq_pluck_data();
}
static void cq_init_callback(void* data,
grpc_completion_queue_functor* shutdown_callback) {
new (data) cq_callback_data(shutdown_callback);
}
static void cq_destroy_callback(void* data) {
cq_callback_data* cqd = static_cast<cq_callback_data*>(data);
cqd->~cq_callback_data();
}
grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue* cq) {
return cq->vtable->cq_completion_type;
}
int grpc_get_cq_poll_num(grpc_completion_queue* cq) {
int cur_num_polls;
gpr_mu_lock(cq->mu);
cur_num_polls = cq->num_polls;
gpr_mu_unlock(cq->mu);
return cur_num_polls;
}
#ifndef NDEBUG
void grpc_cq_internal_ref(grpc_completion_queue* cq, const char* reason,
const char* file, int line) {
grpc_core::DebugLocation debug_location(file, line);
#else
void grpc_cq_internal_ref(grpc_completion_queue* cq) {
grpc_core::DebugLocation debug_location;
const char* reason = nullptr;
#endif
cq->owning_refs.Ref(debug_location, reason);
}
static void on_pollset_shutdown_done(void* arg, grpc_error_handle /*error*/) {
grpc_completion_queue* cq = static_cast<grpc_completion_queue*>(arg);
GRPC_CQ_INTERNAL_UNREF(cq, "pollset_destroy");
}
#ifndef NDEBUG
void grpc_cq_internal_unref(grpc_completion_queue* cq, const char* reason,
const char* file, int line) {
grpc_core::DebugLocation debug_location(file, line);
#else
void grpc_cq_internal_unref(grpc_completion_queue* cq) {
grpc_core::DebugLocation debug_location;
const char* reason = nullptr;
#endif
if (GPR_UNLIKELY(cq->owning_refs.Unref(debug_location, reason))) {
cq->vtable->destroy(DATA_FROM_CQ(cq));
cq->poller_vtable->destroy(POLLSET_FROM_CQ(cq));
#ifndef NDEBUG
gpr_free(cq->outstanding_tags);
#endif
gpr_free(cq);
}
}
#ifndef NDEBUG
static void cq_check_tag(grpc_completion_queue* cq, void* tag, bool lock_cq) {
int found = 0;
if (lock_cq) {
gpr_mu_lock(cq->mu);
}
for (int i = 0; i < static_cast<int>(cq->outstanding_tag_count); i++) {
if (cq->outstanding_tags[i] == tag) {
cq->outstanding_tag_count--;
std::swap(cq->outstanding_tags[i],
cq->outstanding_tags[cq->outstanding_tag_count]);
found = 1;
break;
}
}
if (lock_cq) {
gpr_mu_unlock(cq->mu);
}
GPR_ASSERT(found);
}
#else
static void cq_check_tag(grpc_completion_queue* /*cq*/, void* /*tag*/,
bool /*lock_cq*/) {}
#endif
static bool cq_begin_op_for_next(grpc_completion_queue* cq, void* /*tag*/) {
cq_next_data* cqd = static_cast<cq_next_data*> DATA_FROM_CQ(cq);
return grpc_core::IncrementIfNonzero(&cqd->pending_events);
}
static bool cq_begin_op_for_pluck(grpc_completion_queue* cq, void* /*tag*/) {
cq_pluck_data* cqd = static_cast<cq_pluck_data*> DATA_FROM_CQ(cq);
return grpc_core::IncrementIfNonzero(&cqd->pending_events);
}
static bool cq_begin_op_for_callback(grpc_completion_queue* cq, void* /*tag*/) {
cq_callback_data* cqd = static_cast<cq_callback_data*> DATA_FROM_CQ(cq);
return grpc_core::IncrementIfNonzero(&cqd->pending_events);
}
bool grpc_cq_begin_op(grpc_completion_queue* cq, void* tag) {
#ifndef NDEBUG
gpr_mu_lock(cq->mu);
if (cq->outstanding_tag_count == cq->outstanding_tag_capacity) {
cq->outstanding_tag_capacity =
std::max(size_t(4), 2 * cq->outstanding_tag_capacity);
cq->outstanding_tags = static_cast<void**>(gpr_realloc(
cq->outstanding_tags,
sizeof(*cq->outstanding_tags) * cq->outstanding_tag_capacity));
}
cq->outstanding_tags[cq->outstanding_tag_count++] = tag;
gpr_mu_unlock(cq->mu);
#endif
return cq->vtable->begin_op(cq, tag);
}
/* Queue a GRPC_OP_COMPLETED operation to a completion queue (with a
* completion
* type of GRPC_CQ_NEXT) */
static void cq_end_op_for_next(
grpc_completion_queue* cq, void* tag, grpc_error_handle error,
void (*done)(void* done_arg, grpc_cq_completion* storage), void* done_arg,
grpc_cq_completion* storage, bool /*internal*/) {
GPR_TIMER_SCOPE("cq_end_op_for_next", 0);
if (GRPC_TRACE_FLAG_ENABLED(grpc_api_trace) ||
(GRPC_TRACE_FLAG_ENABLED(grpc_trace_operation_failures) &&
error != GRPC_ERROR_NONE)) {
std::string errmsg = grpc_error_std_string(error);
GRPC_API_TRACE(
"cq_end_op_for_next(cq=%p, tag=%p, error=%s, "
"done=%p, done_arg=%p, storage=%p)",
6, (cq, tag, errmsg.c_str(), done, done_arg, storage));
if (GRPC_TRACE_FLAG_ENABLED(grpc_trace_operation_failures) &&
error != GRPC_ERROR_NONE) {
gpr_log(GPR_INFO, "Operation failed: tag=%p, error=%s", tag,
errmsg.c_str());
}
}
cq_next_data* cqd = static_cast<cq_next_data*> DATA_FROM_CQ(cq);
int is_success = (error == GRPC_ERROR_NONE);
storage->tag = tag;
storage->done = done;
storage->done_arg = done_arg;
storage->next = static_cast<uintptr_t>(is_success);
cq_check_tag(cq, tag, true); /* Used in debug builds only */
if (g_cached_cq == cq && g_cached_event == nullptr) {
g_cached_event = storage;
} else {
/* Add the completion to the queue */
bool is_first = cqd->queue.Push(storage);
cqd->things_queued_ever.fetch_add(1, std::memory_order_relaxed);
/* Since we do not hold the cq lock here, it is important to do an 'acquire'
load here (instead of a 'no_barrier' load) to match with the release
store
(done via pending_events.fetch_sub(1, ACQ_REL)) in cq_shutdown_next
*/
if (cqd->pending_events.load(std::memory_order_acquire) != 1) {
/* Only kick if this is the first item queued */
if (is_first) {
gpr_mu_lock(cq->mu);
grpc_error_handle kick_error =
cq->poller_vtable->kick(POLLSET_FROM_CQ(cq), nullptr);
gpr_mu_unlock(cq->mu);
if (kick_error != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Kick failed: %s",
grpc_error_std_string(kick_error).c_str());
GRPC_ERROR_UNREF(kick_error);
}
}
if (cqd->pending_events.fetch_sub(1, std::memory_order_acq_rel) == 1) {
GRPC_CQ_INTERNAL_REF(cq, "shutting_down");
gpr_mu_lock(cq->mu);
cq_finish_shutdown_next(cq);
gpr_mu_unlock(cq->mu);
GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down");
}
} else {
GRPC_CQ_INTERNAL_REF(cq, "shutting_down");
cqd->pending_events.store(0, std::memory_order_release);
gpr_mu_lock(cq->mu);
cq_finish_shutdown_next(cq);
gpr_mu_unlock(cq->mu);
GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down");
}
}
GRPC_ERROR_UNREF(error);
}
/* Queue a GRPC_OP_COMPLETED operation to a completion queue (with a
* completion
* type of GRPC_CQ_PLUCK) */
static void cq_end_op_for_pluck(
grpc_completion_queue* cq, void* tag, grpc_error_handle error,
void (*done)(void* done_arg, grpc_cq_completion* storage), void* done_arg,
grpc_cq_completion* storage, bool /*internal*/) {
GPR_TIMER_SCOPE("cq_end_op_for_pluck", 0);
cq_pluck_data* cqd = static_cast<cq_pluck_data*> DATA_FROM_CQ(cq);
int is_success = (error == GRPC_ERROR_NONE);
if (GRPC_TRACE_FLAG_ENABLED(grpc_api_trace) ||
(GRPC_TRACE_FLAG_ENABLED(grpc_trace_operation_failures) &&
error != GRPC_ERROR_NONE)) {
std::string errmsg = grpc_error_std_string(error).c_str();
GRPC_API_TRACE(
"cq_end_op_for_pluck(cq=%p, tag=%p, error=%s, "
"done=%p, done_arg=%p, storage=%p)",
6, (cq, tag, errmsg.c_str(), done, done_arg, storage));
if (GRPC_TRACE_FLAG_ENABLED(grpc_trace_operation_failures) &&
error != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Operation failed: tag=%p, error=%s", tag,
errmsg.c_str());
}
}
storage->tag = tag;
storage->done = done;
storage->done_arg = done_arg;
storage->next = reinterpret_cast<uintptr_t>(&cqd->completed_head) |
static_cast<uintptr_t>(is_success);
gpr_mu_lock(cq->mu);
cq_check_tag(cq, tag, false); /* Used in debug builds only */
/* Add to the list of completions */
cqd->things_queued_ever.fetch_add(1, std::memory_order_relaxed);
cqd->completed_tail->next =
reinterpret_cast<uintptr_t>(storage) | (1u & cqd->completed_tail->next);
cqd->completed_tail = storage;
if (cqd->pending_events.fetch_sub(1, std::memory_order_acq_rel) == 1) {
cq_finish_shutdown_pluck(cq);
gpr_mu_unlock(cq->mu);
} else {
grpc_pollset_worker* pluck_worker = nullptr;
for (int i = 0; i < cqd->num_pluckers; i++) {
if (cqd->pluckers[i].tag == tag) {
pluck_worker = *cqd->pluckers[i].worker;
break;
}
}
grpc_error_handle kick_error =
cq->poller_vtable->kick(POLLSET_FROM_CQ(cq), pluck_worker);
gpr_mu_unlock(cq->mu);
if (kick_error != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Kick failed: %s",
grpc_error_std_string(kick_error).c_str());
GRPC_ERROR_UNREF(kick_error);
}
}
GRPC_ERROR_UNREF(error);
}
static void functor_callback(void* arg, grpc_error_handle error) {
auto* functor = static_cast<grpc_completion_queue_functor*>(arg);
functor->functor_run(functor, error == GRPC_ERROR_NONE);
}
/* Complete an event on a completion queue of type GRPC_CQ_CALLBACK */
static void cq_end_op_for_callback(
grpc_completion_queue* cq, void* tag, grpc_error_handle error,
void (*done)(void* done_arg, grpc_cq_completion* storage), void* done_arg,
grpc_cq_completion* storage, bool internal) {
GPR_TIMER_SCOPE("cq_end_op_for_callback", 0);
cq_callback_data* cqd = static_cast<cq_callback_data*> DATA_FROM_CQ(cq);
if (GRPC_TRACE_FLAG_ENABLED(grpc_api_trace) ||
(GRPC_TRACE_FLAG_ENABLED(grpc_trace_operation_failures) &&
error != GRPC_ERROR_NONE)) {
std::string errmsg = grpc_error_std_string(error);
GRPC_API_TRACE(
"cq_end_op_for_callback(cq=%p, tag=%p, error=%s, "
"done=%p, done_arg=%p, storage=%p)",
6, (cq, tag, errmsg.c_str(), done, done_arg, storage));
if (GRPC_TRACE_FLAG_ENABLED(grpc_trace_operation_failures) &&
error != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Operation failed: tag=%p, error=%s", tag,
errmsg.c_str());
}
}
// The callback-based CQ isn't really a queue at all and thus has no need
// for reserved storage. Invoke the done callback right away to release it.
done(done_arg, storage);
cq_check_tag(cq, tag, true); /* Used in debug builds only */
if (cqd->pending_events.fetch_sub(1, std::memory_order_acq_rel) == 1) {
cq_finish_shutdown_callback(cq);
}
// If possible, schedule the callback onto an existing thread-local
// ApplicationCallbackExecCtx, which is a work queue. This is possible for:
// 1. The callback is internally-generated and there is an ACEC available
// 2. The callback is marked inlineable and there is an ACEC available
// 3. We are already running in a background poller thread (which always has
// an ACEC available at the base of the stack).
auto* functor = static_cast<grpc_completion_queue_functor*>(tag);
if (((internal || functor->inlineable) &&
grpc_core::ApplicationCallbackExecCtx::Available()) ||
grpc_iomgr_is_any_background_poller_thread()) {
grpc_core::ApplicationCallbackExecCtx::Enqueue(functor,
(error == GRPC_ERROR_NONE));
GRPC_ERROR_UNREF(error);
return;
}
// Schedule the callback on a closure if not internal or triggered
// from a background poller thread.
grpc_core::Executor::Run(
GRPC_CLOSURE_CREATE(functor_callback, functor, nullptr), error);
}
void grpc_cq_end_op(grpc_completion_queue* cq, void* tag,
grpc_error_handle error,
void (*done)(void* done_arg, grpc_cq_completion* storage),
void* done_arg, grpc_cq_completion* storage,
bool internal) {
cq->vtable->end_op(cq, tag, error, done, done_arg, storage, internal);
}
struct cq_is_finished_arg {
gpr_atm last_seen_things_queued_ever;
grpc_completion_queue* cq;
grpc_core::Timestamp deadline;
grpc_cq_completion* stolen_completion;
void* tag; /* for pluck */
bool first_loop;
};
class ExecCtxNext : public grpc_core::ExecCtx {
public:
explicit ExecCtxNext(void* arg)
: ExecCtx(0), check_ready_to_finish_arg_(arg) {}
bool CheckReadyToFinish() override {
cq_is_finished_arg* a =
static_cast<cq_is_finished_arg*>(check_ready_to_finish_arg_);
grpc_completion_queue* cq = a->cq;
cq_next_data* cqd = static_cast<cq_next_data*> DATA_FROM_CQ(cq);
GPR_ASSERT(a->stolen_completion == nullptr);
intptr_t current_last_seen_things_queued_ever =
cqd->things_queued_ever.load(std::memory_order_relaxed);
if (current_last_seen_things_queued_ever !=
a->last_seen_things_queued_ever) {
a->last_seen_things_queued_ever =
cqd->things_queued_ever.load(std::memory_order_relaxed);
/* Pop a cq_completion from the queue. Returns NULL if the queue is empty
* might return NULL in some cases even if the queue is not empty; but
* that
* is ok and doesn't affect correctness. Might effect the tail latencies a
* bit) */
a->stolen_completion = cqd->queue.Pop();
if (a->stolen_completion != nullptr) {
return true;
}
}
return !a->first_loop && a->deadline < grpc_core::ExecCtx::Get()->Now();
}
private:
void* check_ready_to_finish_arg_;
};
#ifndef NDEBUG
static void dump_pending_tags(grpc_completion_queue* cq) {
if (!GRPC_TRACE_FLAG_ENABLED(grpc_trace_pending_tags)) return;
std::vector<std::string> parts;
parts.push_back("PENDING TAGS:");
gpr_mu_lock(cq->mu);
for (size_t i = 0; i < cq->outstanding_tag_count; i++) {
parts.push_back(absl::StrFormat(" %p", cq->outstanding_tags[i]));
}
gpr_mu_unlock(cq->mu);
gpr_log(GPR_DEBUG, "%s", absl::StrJoin(parts, "").c_str());
}
#else
static void dump_pending_tags(grpc_completion_queue* /*cq*/) {}
#endif
static grpc_event cq_next(grpc_completion_queue* cq, gpr_timespec deadline,
void* reserved) {
GPR_TIMER_SCOPE("grpc_completion_queue_next", 0);
grpc_event ret;
cq_next_data* cqd = static_cast<cq_next_data*> DATA_FROM_CQ(cq);
GRPC_API_TRACE(
"grpc_completion_queue_next("
"cq=%p, "
"deadline=gpr_timespec { tv_sec: %" PRId64
", tv_nsec: %d, clock_type: %d }, "
"reserved=%p)",
5,
(cq, deadline.tv_sec, deadline.tv_nsec, (int)deadline.clock_type,
reserved));
GPR_ASSERT(!reserved);
dump_pending_tags(cq);
GRPC_CQ_INTERNAL_REF(cq, "next");
grpc_core::Timestamp deadline_millis =
grpc_core::Timestamp::FromTimespecRoundUp(deadline);
cq_is_finished_arg is_finished_arg = {
cqd->things_queued_ever.load(std::memory_order_relaxed),
cq,
deadline_millis,
nullptr,
nullptr,
true};
ExecCtxNext exec_ctx(&is_finished_arg);
for (;;) {
grpc_core::Timestamp iteration_deadline = deadline_millis;
if (is_finished_arg.stolen_completion != nullptr) {
grpc_cq_completion* c = is_finished_arg.stolen_completion;
is_finished_arg.stolen_completion = nullptr;
ret.type = GRPC_OP_COMPLETE;
ret.success = c->next & 1u;
ret.tag = c->tag;
c->done(c->done_arg, c);
break;
}
grpc_cq_completion* c = cqd->queue.Pop();
if (c != nullptr) {
ret.type = GRPC_OP_COMPLETE;
ret.success = c->next & 1u;
ret.tag = c->tag;
c->done(c->done_arg, c);
break;
} else {
/* If c == NULL it means either the queue is empty OR in an transient
inconsistent state. If it is the latter, we shold do a 0-timeout poll
so that the thread comes back quickly from poll to make a second
attempt at popping. Not doing this can potentially deadlock this
thread forever (if the deadline is infinity) */
if (cqd->queue.num_items() > 0) {
iteration_deadline = grpc_core::Timestamp::ProcessEpoch();
}
}
if (cqd->pending_events.load(std::memory_order_acquire) == 0) {
/* Before returning, check if the queue has any items left over (since
MultiProducerSingleConsumerQueue::Pop() can sometimes return NULL
even if the queue is not empty. If so, keep retrying but do not
return GRPC_QUEUE_SHUTDOWN */
if (cqd->queue.num_items() > 0) {
/* Go to the beginning of the loop. No point doing a poll because
(cq->shutdown == true) is only possible when there is no pending
work (i.e cq->pending_events == 0) and any outstanding completion
events should have already been queued on this cq */
continue;
}
ret.type = GRPC_QUEUE_SHUTDOWN;
ret.success = 0;
break;
}
if (!is_finished_arg.first_loop &&
grpc_core::ExecCtx::Get()->Now() >= deadline_millis) {
ret.type = GRPC_QUEUE_TIMEOUT;
ret.success = 0;
dump_pending_tags(cq);
break;
}
/* The main polling work happens in grpc_pollset_work */
gpr_mu_lock(cq->mu);
cq->num_polls++;
grpc_error_handle err = cq->poller_vtable->work(
POLLSET_FROM_CQ(cq), nullptr, iteration_deadline);
gpr_mu_unlock(cq->mu);
if (err != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Completion queue next failed: %s",
grpc_error_std_string(err).c_str());
GRPC_ERROR_UNREF(err);
if (err == GRPC_ERROR_CANCELLED) {
ret.type = GRPC_QUEUE_SHUTDOWN;
} else {
ret.type = GRPC_QUEUE_TIMEOUT;
}
ret.success = 0;
dump_pending_tags(cq);
break;
}
is_finished_arg.first_loop = false;
}
if (cqd->queue.num_items() > 0 &&
cqd->pending_events.load(std::memory_order_acquire) > 0) {
gpr_mu_lock(cq->mu);
(void)cq->poller_vtable->kick(POLLSET_FROM_CQ(cq), nullptr);
gpr_mu_unlock(cq->mu);
}
GRPC_SURFACE_TRACE_RETURNED_EVENT(cq, &ret);
GRPC_CQ_INTERNAL_UNREF(cq, "next");
GPR_ASSERT(is_finished_arg.stolen_completion == nullptr);
return ret;
}
/* Finishes the completion queue shutdown. This means that there are no more
completion events / tags expected from the completion queue
- Must be called under completion queue lock
- Must be called only once in completion queue's lifetime
- grpc_completion_queue_shutdown() MUST have been called before calling
this function */
static void cq_finish_shutdown_next(grpc_completion_queue* cq) {
cq_next_data* cqd = static_cast<cq_next_data*> DATA_FROM_CQ(cq);
GPR_ASSERT(cqd->shutdown_called);
GPR_ASSERT(cqd->pending_events.load(std::memory_order_relaxed) == 0);
cq->poller_vtable->shutdown(POLLSET_FROM_CQ(cq), &cq->pollset_shutdown_done);
}
static void cq_shutdown_next(grpc_completion_queue* cq) {
cq_next_data* cqd = static_cast<cq_next_data*> DATA_FROM_CQ(cq);
/* Need an extra ref for cq here because:
* We call cq_finish_shutdown_next() below, that would call pollset shutdown.
* Pollset shutdown decrements the cq ref count which can potentially destroy
* the cq (if that happens to be the last ref).
* Creating an extra ref here prevents the cq from getting destroyed while
* this function is still active */
GRPC_CQ_INTERNAL_REF(cq, "shutting_down");
gpr_mu_lock(cq->mu);
if (cqd->shutdown_called) {
gpr_mu_unlock(cq->mu);
GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down");
return;
}
cqd->shutdown_called = true;
/* Doing acq/release fetch_sub here to match with
* cq_begin_op_for_next and cq_end_op_for_next functions which read/write
* on this counter without necessarily holding a lock on cq */
if (cqd->pending_events.fetch_sub(1, std::memory_order_acq_rel) == 1) {
cq_finish_shutdown_next(cq);
}
gpr_mu_unlock(cq->mu);
GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down");
}
grpc_event grpc_completion_queue_next(grpc_completion_queue* cq,
gpr_timespec deadline, void* reserved) {
return cq->vtable->next(cq, deadline, reserved);
}
static int add_plucker(grpc_completion_queue* cq, void* tag,
grpc_pollset_worker** worker) {
cq_pluck_data* cqd = static_cast<cq_pluck_data*> DATA_FROM_CQ(cq);
if (cqd->num_pluckers == GRPC_MAX_COMPLETION_QUEUE_PLUCKERS) {
return 0;
}
cqd->pluckers[cqd->num_pluckers].tag = tag;
cqd->pluckers[cqd->num_pluckers].worker = worker;
cqd->num_pluckers++;
return 1;
}
static void del_plucker(grpc_completion_queue* cq, void* tag,
grpc_pollset_worker** worker) {
cq_pluck_data* cqd = static_cast<cq_pluck_data*> DATA_FROM_CQ(cq);
for (int i = 0; i < cqd->num_pluckers; i++) {
if (cqd->pluckers[i].tag == tag && cqd->pluckers[i].worker == worker) {
cqd->num_pluckers--;
std::swap(cqd->pluckers[i], cqd->pluckers[cqd->num_pluckers]);
return;
}
}
GPR_UNREACHABLE_CODE(return );
}
class ExecCtxPluck : public grpc_core::ExecCtx {
public:
explicit ExecCtxPluck(void* arg)
: ExecCtx(0), check_ready_to_finish_arg_(arg) {}
bool CheckReadyToFinish() override {
cq_is_finished_arg* a =
static_cast<cq_is_finished_arg*>(check_ready_to_finish_arg_);
grpc_completion_queue* cq = a->cq;
cq_pluck_data* cqd = static_cast<cq_pluck_data*> DATA_FROM_CQ(cq);
GPR_ASSERT(a->stolen_completion == nullptr);
gpr_atm current_last_seen_things_queued_ever =
cqd->things_queued_ever.load(std::memory_order_relaxed);
if (current_last_seen_things_queued_ever !=
a->last_seen_things_queued_ever) {
gpr_mu_lock(cq->mu);
a->last_seen_things_queued_ever =
cqd->things_queued_ever.load(std::memory_order_relaxed);
grpc_cq_completion* c;
grpc_cq_completion* prev = &cqd->completed_head;
while ((c = reinterpret_cast<grpc_cq_completion*>(
prev->next & ~static_cast<uintptr_t>(1))) !=
&cqd->completed_head) {
if (c->tag == a->tag) {
prev->next = (prev->next & static_cast<uintptr_t>(1)) |
(c->next & ~static_cast<uintptr_t>(1));
if (c == cqd->completed_tail) {
cqd->completed_tail = prev;
}
gpr_mu_unlock(cq->mu);
a->stolen_completion = c;
return true;
}
prev = c;
}
gpr_mu_unlock(cq->mu);
}
return !a->first_loop && a->deadline < grpc_core::ExecCtx::Get()->Now();
}
private:
void* check_ready_to_finish_arg_;
};
static grpc_event cq_pluck(grpc_completion_queue* cq, void* tag,
gpr_timespec deadline, void* reserved) {
GPR_TIMER_SCOPE("grpc_completion_queue_pluck", 0);
grpc_event ret;
grpc_cq_completion* c;
grpc_cq_completion* prev;
grpc_pollset_worker* worker = nullptr;
cq_pluck_data* cqd = static_cast<cq_pluck_data*> DATA_FROM_CQ(cq);
if (GRPC_TRACE_FLAG_ENABLED(grpc_cq_pluck_trace)) {
GRPC_API_TRACE(
"grpc_completion_queue_pluck("
"cq=%p, tag=%p, "
"deadline=gpr_timespec { tv_sec: %" PRId64
", tv_nsec: %d, clock_type: %d }, "
"reserved=%p)",
6,
(cq, tag, deadline.tv_sec, deadline.tv_nsec, (int)deadline.clock_type,
reserved));
}
GPR_ASSERT(!reserved);
dump_pending_tags(cq);
GRPC_CQ_INTERNAL_REF(cq, "pluck");
gpr_mu_lock(cq->mu);
grpc_core::Timestamp deadline_millis =
grpc_core::Timestamp::FromTimespecRoundUp(deadline);
cq_is_finished_arg is_finished_arg = {
cqd->things_queued_ever.load(std::memory_order_relaxed),
cq,
deadline_millis,
nullptr,
tag,
true};
ExecCtxPluck exec_ctx(&is_finished_arg);
for (;;) {
if (is_finished_arg.stolen_completion != nullptr) {
gpr_mu_unlock(cq->mu);
c = is_finished_arg.stolen_completion;
is_finished_arg.stolen_completion = nullptr;
ret.type = GRPC_OP_COMPLETE;
ret.success = c->next & 1u;
ret.tag = c->tag;
c->done(c->done_arg, c);
break;
}
prev = &cqd->completed_head;
while ((c = reinterpret_cast<grpc_cq_completion*>(
prev->next & ~static_cast<uintptr_t>(1))) !=
&cqd->completed_head) {
if (c->tag == tag) {
prev->next = (prev->next & static_cast<uintptr_t>(1)) |
(c->next & ~static_cast<uintptr_t>(1));
if (c == cqd->completed_tail) {
cqd->completed_tail = prev;
}
gpr_mu_unlock(cq->mu);
ret.type = GRPC_OP_COMPLETE;
ret.success = c->next & 1u;
ret.tag = c->tag;
c->done(c->done_arg, c);
goto done;
}
prev = c;
}
if (cqd->shutdown.load(std::memory_order_relaxed)) {
gpr_mu_unlock(cq->mu);
ret.type = GRPC_QUEUE_SHUTDOWN;
ret.success = 0;
break;
}
if (!add_plucker(cq, tag, &worker)) {
gpr_log(GPR_DEBUG,
"Too many outstanding grpc_completion_queue_pluck calls: maximum "
"is %d",
GRPC_MAX_COMPLETION_QUEUE_PLUCKERS);
gpr_mu_unlock(cq->mu);
/* TODO(ctiller): should we use a different result here */
ret.type = GRPC_QUEUE_TIMEOUT;
ret.success = 0;
dump_pending_tags(cq);
break;
}
if (!is_finished_arg.first_loop &&
grpc_core::ExecCtx::Get()->Now() >= deadline_millis) {
del_plucker(cq, tag, &worker);
gpr_mu_unlock(cq->mu);
ret.type = GRPC_QUEUE_TIMEOUT;
ret.success = 0;
dump_pending_tags(cq);
break;
}
cq->num_polls++;
grpc_error_handle err =
cq->poller_vtable->work(POLLSET_FROM_CQ(cq), &worker, deadline_millis);
if (err != GRPC_ERROR_NONE) {
del_plucker(cq, tag, &worker);
gpr_mu_unlock(cq->mu);
gpr_log(GPR_ERROR, "Completion queue pluck failed: %s",
grpc_error_std_string(err).c_str());
GRPC_ERROR_UNREF(err);
ret.type = GRPC_QUEUE_TIMEOUT;
ret.success = 0;
dump_pending_tags(cq);
break;
}
is_finished_arg.first_loop = false;
del_plucker(cq, tag, &worker);
}
done:
GRPC_SURFACE_TRACE_RETURNED_EVENT(cq, &ret);
GRPC_CQ_INTERNAL_UNREF(cq, "pluck");
GPR_ASSERT(is_finished_arg.stolen_completion == nullptr);
return ret;
}
grpc_event grpc_completion_queue_pluck(grpc_completion_queue* cq, void* tag,
gpr_timespec deadline, void* reserved) {
return cq->vtable->pluck(cq, tag, deadline, reserved);
}
static void cq_finish_shutdown_pluck(grpc_completion_queue* cq) {
cq_pluck_data* cqd = static_cast<cq_pluck_data*> DATA_FROM_CQ(cq);
GPR_ASSERT(cqd->shutdown_called);
GPR_ASSERT(!cqd->shutdown.load(std::memory_order_relaxed));
cqd->shutdown.store(true, std::memory_order_relaxed);
cq->poller_vtable->shutdown(POLLSET_FROM_CQ(cq), &cq->pollset_shutdown_done);
}
/* NOTE: This function is almost exactly identical to cq_shutdown_next() but
* merging them is a bit tricky and probably not worth it */
static void cq_shutdown_pluck(grpc_completion_queue* cq) {
cq_pluck_data* cqd = static_cast<cq_pluck_data*> DATA_FROM_CQ(cq);
/* Need an extra ref for cq here because:
* We call cq_finish_shutdown_pluck() below, that would call pollset shutdown.
* Pollset shutdown decrements the cq ref count which can potentially destroy
* the cq (if that happens to be the last ref).
* Creating an extra ref here prevents the cq from getting destroyed while
* this function is still active */
GRPC_CQ_INTERNAL_REF(cq, "shutting_down (pluck cq)");
gpr_mu_lock(cq->mu);
if (cqd->shutdown_called) {
gpr_mu_unlock(cq->mu);
GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (pluck cq)");
return;
}
cqd->shutdown_called = true;
if (cqd->pending_events.fetch_sub(1, std::memory_order_acq_rel) == 1) {
cq_finish_shutdown_pluck(cq);
}
gpr_mu_unlock(cq->mu);
GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (pluck cq)");
}
static void cq_finish_shutdown_callback(grpc_completion_queue* cq) {
cq_callback_data* cqd = static_cast<cq_callback_data*> DATA_FROM_CQ(cq);
auto* callback = cqd->shutdown_callback;
GPR_ASSERT(cqd->shutdown_called);
cq->poller_vtable->shutdown(POLLSET_FROM_CQ(cq), &cq->pollset_shutdown_done);
if (grpc_iomgr_is_any_background_poller_thread()) {
grpc_core::ApplicationCallbackExecCtx::Enqueue(callback, true);
return;
}
// Schedule the callback on a closure if not internal or triggered
// from a background poller thread.
grpc_core::Executor::Run(
GRPC_CLOSURE_CREATE(functor_callback, callback, nullptr),
GRPC_ERROR_NONE);
}
static void cq_shutdown_callback(grpc_completion_queue* cq) {
cq_callback_data* cqd = static_cast<cq_callback_data*> DATA_FROM_CQ(cq);
/* Need an extra ref for cq here because:
* We call cq_finish_shutdown_callback() below, which calls pollset shutdown.
* Pollset shutdown decrements the cq ref count which can potentially destroy
* the cq (if that happens to be the last ref).
* Creating an extra ref here prevents the cq from getting destroyed while
* this function is still active */
GRPC_CQ_INTERNAL_REF(cq, "shutting_down (callback cq)");
gpr_mu_lock(cq->mu);
if (cqd->shutdown_called) {
gpr_mu_unlock(cq->mu);
GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (callback cq)");
return;
}
cqd->shutdown_called = true;
if (cqd->pending_events.fetch_sub(1, std::memory_order_acq_rel) == 1) {
gpr_mu_unlock(cq->mu);
cq_finish_shutdown_callback(cq);
} else {
gpr_mu_unlock(cq->mu);
}
GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (callback cq)");
}
/* Shutdown simply drops a ref that we reserved at creation time; if we drop
to zero here, then enter shutdown mode and wake up any waiters */
void grpc_completion_queue_shutdown(grpc_completion_queue* cq) {
GPR_TIMER_SCOPE("grpc_completion_queue_shutdown", 0);
grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
grpc_core::ExecCtx exec_ctx;
GRPC_API_TRACE("grpc_completion_queue_shutdown(cq=%p)", 1, (cq));
cq->vtable->shutdown(cq);
}
void grpc_completion_queue_destroy(grpc_completion_queue* cq) {
GPR_TIMER_SCOPE("grpc_completion_queue_destroy", 0);
GRPC_API_TRACE("grpc_completion_queue_destroy(cq=%p)", 1, (cq));
grpc_completion_queue_shutdown(cq);
grpc_core::ExecCtx exec_ctx;
GRPC_CQ_INTERNAL_UNREF(cq, "destroy");
}
grpc_pollset* grpc_cq_pollset(grpc_completion_queue* cq) {
return cq->poller_vtable->can_get_pollset ? POLLSET_FROM_CQ(cq) : nullptr;
}
bool grpc_cq_can_listen(grpc_completion_queue* cq) {
return cq->poller_vtable->can_listen;
}