1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
|
/*
* Copyright 2015 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 <grpcpp/completion_queue.h>
#include <memory>
#include <grpc/grpc.h>
#include <grpc/support/log.h>
#include <grpcpp/impl/grpc_library.h>
#include <grpcpp/support/time.h>
namespace grpc {
static internal::GrpcLibraryInitializer g_gli_initializer;
// 'CompletionQueue' constructor can safely call GrpcLibraryCodegen(false) here
// i.e not have GrpcLibraryCodegen call grpc_init(). This is because, to create
// a 'grpc_completion_queue' instance (which is being passed as the input to
// this constructor), one must have already called grpc_init().
CompletionQueue::CompletionQueue(grpc_completion_queue* take)
: GrpcLibraryCodegen(false), cq_(take) {
InitialAvalanching();
}
void CompletionQueue::Shutdown() {
g_gli_initializer.summon();
CompleteAvalanching();
}
void CompletionQueue::CompleteAvalanching() {
// Check if this was the last avalanching operation
if (gpr_atm_no_barrier_fetch_add(&avalanches_in_flight_,
static_cast<gpr_atm>(-1)) == 1) {
grpc_completion_queue_shutdown(cq_);
}
}
CompletionQueue::NextStatus CompletionQueue::AsyncNextInternal(
void** tag, bool* ok, gpr_timespec deadline) {
for (;;) {
auto ev = grpc_completion_queue_next(cq_, deadline, nullptr);
switch (ev.type) {
case GRPC_QUEUE_TIMEOUT:
return TIMEOUT;
case GRPC_QUEUE_SHUTDOWN:
return SHUTDOWN;
case GRPC_OP_COMPLETE:
auto core_cq_tag = static_cast<internal::CompletionQueueTag*>(ev.tag);
*ok = ev.success != 0;
*tag = core_cq_tag;
if (core_cq_tag->FinalizeResult(tag, ok)) {
return GOT_EVENT;
}
break;
}
}
}
CompletionQueue::CompletionQueueTLSCache::CompletionQueueTLSCache(
CompletionQueue* cq)
: cq_(cq), flushed_(false) {
grpc_completion_queue_thread_local_cache_init(cq_->cq_);
}
CompletionQueue::CompletionQueueTLSCache::~CompletionQueueTLSCache() {
GPR_ASSERT(flushed_);
}
bool CompletionQueue::CompletionQueueTLSCache::Flush(void** tag, bool* ok) {
int res = 0;
void* res_tag;
flushed_ = true;
if (grpc_completion_queue_thread_local_cache_flush(cq_->cq_, &res_tag,
&res)) {
auto core_cq_tag = static_cast<internal::CompletionQueueTag*>(res_tag);
*ok = res == 1;
if (core_cq_tag->FinalizeResult(tag, ok)) {
return true;
}
}
return false;
}
} // namespace grpc
|
