Skip to content

Commit f8c2999

Browse files
ThalhammerThalhammer
committed
✨ Add client_cq class
1 parent e9723ab commit f8c2999

File tree

7 files changed

+152
-64
lines changed

7 files changed

+152
-64
lines changed

‎CMakeLists.txt

+3-1
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,8 @@ else()
2424
include(cmake/Fetch_asyncpp.cmake)
2525
endif()
2626

27-
add_library(asyncpp_grpc ${CMAKE_CURRENT_SOURCE_DIR}/src/grpc/server.cpp
27+
add_library(asyncpp_grpc ${CMAKE_CURRENT_SOURCE_DIR}/src/grpc/client_cq.cpp
28+
${CMAKE_CURRENT_SOURCE_DIR}/src/grpc/server.cpp
2829
${CMAKE_CURRENT_SOURCE_DIR}/src/grpc/util.cpp)
2930
target_link_libraries(
3031
asyncpp_grpc PUBLIC asyncpp gRPC::grpc++ protobuf::libprotobuf
@@ -49,6 +50,7 @@ if(ASYNCPP_BUILD_TEST)
4950
asyncpp_grpc-test
5051
${PROTO_SRCS} ${CMAKE_CURRENT_SOURCE_DIR}/test/traits.cpp
5152
${CMAKE_CURRENT_SOURCE_DIR}/test/call.cpp
53+
${CMAKE_CURRENT_SOURCE_DIR}/test/client_cq.cpp
5254
${CMAKE_CURRENT_SOURCE_DIR}/test/task.cpp)
5355
target_link_libraries(asyncpp_grpc-test PRIVATE asyncpp_grpc GTest::gtest
5456
GTest::gtest_main)

‎include/asyncpp/grpc/call.h

+52-28
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
#pragma once
22
#include <asyncpp/detail/std_import.h>
33
#include <asyncpp/grpc/calldata_interface.h>
4+
#include <asyncpp/grpc/client_cq.h>
45
#include <asyncpp/grpc/traits.h>
56
#include <asyncpp/ptr_tag.h>
67
#include <grpcpp/impl/codegen/async_unary_call.h>
@@ -26,20 +27,26 @@ namespace asyncpp::grpc {
2627
unary_call& operator=(const unary_call&) = delete;
2728
unary_call& operator=(unary_call&&) = delete;
2829

29-
auto operator()(const typename traits::request_type& request, typename traits::response_type& response, ::grpc::CompletionQueue* cq) {
30+
auto operator()(const typename traits::request_type& request, typename traits::response_type& response, ::grpc::CompletionQueue* cq = nullptr) {
3031
struct awaiter : calldata_interface {
3132
::grpc::ClientContext* m_context;
3233
typename traits::service_type* m_stub;
3334
const typename traits::request_type& m_request;
3435
typename traits::response_type& m_response;
36+
std::shared_ptr<client_cq> m_default_client_cq{};
3537
::grpc::CompletionQueue* m_cq{};
3638
typename traits::stream_type m_reader{};
3739
::grpc::Status m_status{};
3840
coroutine_handle<> m_handle{};
3941

4042
awaiter(::grpc::ClientContext* context, decltype(m_stub) s, const typename traits::request_type& request,
4143
typename traits::response_type& response, ::grpc::CompletionQueue* cq)
42-
: m_context{context}, m_stub{s}, m_request{request}, m_response{response}, m_cq{cq} {}
44+
: m_context{context}, m_stub{s}, m_request{request}, m_response{response}, m_cq{cq} {
45+
if (m_cq == nullptr) {
46+
m_default_client_cq = client_cq::get_default();
47+
m_cq = &m_default_client_cq->cq();
48+
}
49+
}
4350

4451
void handle_event([[maybe_unused]] size_t evt, bool ok) noexcept override {
4552
assert(evt == 0);
@@ -53,7 +60,7 @@ namespace asyncpp::grpc {
5360
m_handle = h;
5461

5562
m_reader = (m_stub->*FN)(m_context, m_request, m_cq);
56-
m_reader->Finish(&m_response, &m_status, this);
63+
m_reader->Finish(&m_response, &m_status, ptr_tag<0, calldata_interface>(this));
5764
}
5865
::grpc::Status await_resume() noexcept { return m_status; }
5966
};
@@ -77,28 +84,30 @@ namespace asyncpp::grpc {
7784
typename traits::service_type* stub;
7885
typename traits::stream_type stream{};
7986
bool m_need_writes_done{false};
87+
std::shared_ptr<client_cq> default_client_cq{};
88+
::grpc::CompletionQueue* cq{};
8089

8190
void destruct() {
8291
if (stream) {
8392
context->TryCancel();
8493
if constexpr (traits::is_client_streaming) {
8594
if (m_need_writes_done) {
86-
stream->WritesDone(ptr_tag<0>(this));
95+
stream->WritesDone(ptr_tag<0, calldata_interface>(this));
8796
} else {
88-
stream->Finish(&m_exit_status, ptr_tag<1>(this));
97+
stream->Finish(&m_exit_status, ptr_tag<1, calldata_interface>(this));
8998
}
9099
} else
91-
stream->Finish(&m_exit_status, ptr_tag<1>(this));
100+
stream->Finish(&m_exit_status, ptr_tag<1, calldata_interface>(this));
92101
} else {
93102
delete this;
94103
}
95104
}
96105

97106
private:
98107
::grpc::Status m_exit_status{};
99-
void handle_event(size_t evt, bool ok) noexcept override {
108+
void handle_event(size_t evt, [[maybe_unused]] bool ok) noexcept override {
100109
if (evt == 0) {
101-
return stream->Finish(&m_exit_status, ptr_tag<1>(this));
110+
return stream->Finish(&m_exit_status, ptr_tag<1, calldata_interface>(this));
102111
} else {
103112
stream.reset();
104113
context.reset();
@@ -131,17 +140,22 @@ namespace asyncpp::grpc {
131140
::grpc::ClientContext& context() noexcept { return *m_state->context; }
132141
const ::grpc::ClientContext& context() const noexcept { return *m_state->context; }
133142

134-
auto start(const typename traits::request_type& req, ::grpc::CompletionQueue* cq)
143+
auto start(const typename traits::request_type& req, ::grpc::CompletionQueue* cq = nullptr)
135144
requires(!traits::is_client_streaming && traits::is_server_streaming)
136145
{
137146
struct awaiter : calldata_interface {
138147
state* m_state;
139148
const typename traits::request_type& m_request{};
140-
::grpc::CompletionQueue* m_cq{};
141149
coroutine_handle<> m_handle{};
142150
bool m_was_ok = false;
143151

144-
awaiter(state* state, const typename traits::request_type& req, ::grpc::CompletionQueue* cq) : m_state{state}, m_request{req}, m_cq{cq} {}
152+
awaiter(state* state, const typename traits::request_type& req, ::grpc::CompletionQueue* cq) : m_state{state}, m_request{req} {
153+
if (cq == nullptr) {
154+
m_state->default_client_cq = client_cq::get_default();
155+
cq = &m_state->default_client_cq->cq();
156+
}
157+
m_state->cq = cq;
158+
}
145159

146160
void handle_event([[maybe_unused]] size_t evt, bool ok) noexcept override {
147161
assert(evt == 0);
@@ -154,24 +168,29 @@ namespace asyncpp::grpc {
154168
void await_suspend(coroutine_handle<> h) noexcept {
155169
assert(m_state);
156170
m_handle = h;
157-
m_state->stream = (m_state->stub->*FN)(m_state->context.get(), m_request, m_cq, this);
171+
m_state->stream = (m_state->stub->*FN)(m_state->context.get(), m_request, m_state->cq, ptr_tag<0, calldata_interface>(this));
158172
}
159173
bool await_resume() const noexcept { return m_was_ok; }
160174
};
161175
return awaiter{m_state.get(), req, cq};
162176
}
163177

164-
auto start(typename traits::response_type& resp, ::grpc::CompletionQueue* cq)
178+
auto start(typename traits::response_type& resp, ::grpc::CompletionQueue* cq = nullptr)
165179
requires(traits::is_client_streaming && !traits::is_server_streaming)
166180
{
167181
struct awaiter : calldata_interface {
168182
state* m_state;
169183
typename traits::response_type& m_response;
170-
::grpc::CompletionQueue* m_cq{};
171184
coroutine_handle<> m_handle{};
172185
bool m_was_ok = false;
173186

174-
awaiter(state* state, typename traits::response_type& resp, ::grpc::CompletionQueue* cq) : m_state{state}, m_response{resp}, m_cq{cq} {}
187+
awaiter(state* state, typename traits::response_type& resp, ::grpc::CompletionQueue* cq) : m_state{state}, m_response{resp} {
188+
if (cq == nullptr) {
189+
m_state->default_client_cq = client_cq::get_default();
190+
cq = &m_state->default_client_cq->cq();
191+
}
192+
m_state->cq = cq;
193+
}
175194

176195
void handle_event([[maybe_unused]] size_t evt, bool ok) noexcept override {
177196
assert(evt == 0);
@@ -185,23 +204,28 @@ namespace asyncpp::grpc {
185204
void await_suspend(coroutine_handle<> h) noexcept {
186205
assert(m_state);
187206
m_handle = h;
188-
m_state->stream = (m_state->stub->*FN)(m_state->context.get(), &m_response, m_cq, this);
207+
m_state->stream = (m_state->stub->*FN)(m_state->context.get(), &m_response, m_state->cq, ptr_tag<0, calldata_interface>(this));
189208
}
190209
bool await_resume() noexcept { return m_was_ok; }
191210
};
192211
return awaiter{m_state.get(), resp, cq};
193212
}
194213

195-
auto start(::grpc::CompletionQueue* cq)
214+
auto start(::grpc::CompletionQueue* cq = nullptr)
196215
requires(traits::is_client_streaming && traits::is_server_streaming)
197216
{
198217
struct awaiter : calldata_interface {
199218
state* m_state;
200-
::grpc::CompletionQueue* m_cq{};
201219
coroutine_handle<> m_handle{};
202220
bool m_was_ok = false;
203221

204-
awaiter(state* state, ::grpc::CompletionQueue* cq) : m_state{state}, m_cq{cq} {}
222+
awaiter(state* state, ::grpc::CompletionQueue* cq) : m_state{state} {
223+
if (cq == nullptr) {
224+
m_state->default_client_cq = client_cq::get_default();
225+
cq = &m_state->default_client_cq->cq();
226+
}
227+
m_state->cq = cq;
228+
}
205229

206230
void handle_event([[maybe_unused]] size_t evt, bool ok) noexcept override {
207231
assert(evt == 0);
@@ -215,7 +239,7 @@ namespace asyncpp::grpc {
215239
void await_suspend(coroutine_handle<> h) noexcept {
216240
assert(m_state);
217241
m_handle = h;
218-
m_state->stream = (m_state->stub->*FN)(m_state->context.get(), m_cq, this);
242+
m_state->stream = (m_state->stub->*FN)(m_state->context.get(), m_state->cq, ptr_tag<0, calldata_interface>(this));
219243
}
220244
bool await_resume() noexcept { return m_was_ok; }
221245
};
@@ -243,7 +267,7 @@ namespace asyncpp::grpc {
243267
void await_suspend(coroutine_handle<> h) noexcept {
244268
assert(m_state);
245269
m_handle = h;
246-
m_state->stream->Read(m_resp, this);
270+
m_state->stream->Read(m_resp, ptr_tag<0, calldata_interface>(this));
247271
}
248272
bool await_resume() noexcept { return m_was_ok; }
249273
};
@@ -272,7 +296,7 @@ namespace asyncpp::grpc {
272296
void await_suspend(coroutine_handle<> h) noexcept {
273297
assert(m_state);
274298
m_handle = h;
275-
m_state->stream->Write(m_msg, this);
299+
m_state->stream->Write(m_msg, ptr_tag<0, calldata_interface>(this));
276300
}
277301
bool await_resume() noexcept { return m_was_ok; }
278302
};
@@ -302,7 +326,7 @@ namespace asyncpp::grpc {
302326
void await_suspend(coroutine_handle<> h) noexcept {
303327
assert(m_state);
304328
m_handle = h;
305-
m_state->stream->WriteLast(m_msg, ::grpc::WriteOptions{}, this);
329+
m_state->stream->WriteLast(m_msg, ::grpc::WriteOptions{}, ptr_tag<0, calldata_interface>(this));
306330
}
307331
bool await_resume() noexcept { return m_was_ok; }
308332
};
@@ -331,7 +355,7 @@ namespace asyncpp::grpc {
331355
void await_suspend(coroutine_handle<> h) noexcept {
332356
assert(m_state);
333357
m_handle = h;
334-
m_state->stream->WritesDone(this);
358+
m_state->stream->WritesDone(ptr_tag<0, calldata_interface>(this));
335359
}
336360
bool await_resume() noexcept { return m_was_ok; }
337361
};
@@ -350,7 +374,7 @@ namespace asyncpp::grpc {
350374
assert(m_handle);
351375
if (!ok) m_status = ::grpc::Status(::grpc::StatusCode::UNKNOWN, "Event returned ok=false");
352376
switch (evt) {
353-
case 0: return m_state->stream->Finish(&m_status, asyncpp::ptr_tag<1>(this)); break;
377+
case 0: return m_state->stream->Finish(&m_status, asyncpp::ptr_tag<1, calldata_interface>(this)); break;
354378
case 1: m_state->stream.reset(); break;
355379
default: assert(false); break;
356380
}
@@ -362,12 +386,12 @@ namespace asyncpp::grpc {
362386
m_handle = h;
363387
if constexpr (traits::is_client_streaming) {
364388
if (m_state->m_need_writes_done) {
365-
m_state->stream->WritesDone(ptr_tag<0>(this));
389+
m_state->stream->WritesDone(ptr_tag<0, calldata_interface>(this));
366390
} else {
367-
m_state->stream->Finish(&m_status, ptr_tag<1>(this));
391+
m_state->stream->Finish(&m_status, ptr_tag<1, calldata_interface>(this));
368392
}
369393
} else {
370-
m_state->stream->Finish(&m_status, ptr_tag<1>(this));
394+
m_state->stream->Finish(&m_status, ptr_tag<1, calldata_interface>(this));
371395
}
372396
}
373397
::grpc::Status await_resume() noexcept { return m_status; }

‎include/asyncpp/grpc/client_cq.h

+36
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
#pragma once
2+
#include <asyncpp/dispatcher.h>
3+
#include <asyncpp/grpc/calldata_interface.h>
4+
#include <asyncpp/threadsafe_queue.h>
5+
#include <grpcpp/alarm.h>
6+
#include <grpcpp/completion_queue.h>
7+
8+
#include <memory>
9+
#include <thread>
10+
11+
namespace asyncpp::grpc {
12+
class client_cq : public dispatcher, private grpc::calldata_interface {
13+
std::thread m_thread;
14+
threadsafe_queue<std::function<void()>> m_dispatched;
15+
std::atomic_flag m_alarm_set;
16+
::grpc::Alarm m_alarm;
17+
::grpc::CompletionQueue m_cq;
18+
19+
void handle_event(size_t evt, bool ok) noexcept override;
20+
21+
public:
22+
client_cq();
23+
client_cq(const client_cq&) = delete;
24+
client_cq(client_cq&&) = delete;
25+
client_cq& operator=(const client_cq&) = delete;
26+
client_cq& operator=(client_cq&&) = delete;
27+
~client_cq();
28+
29+
void push(std::function<void()> fn) override;
30+
31+
::grpc::CompletionQueue& cq() noexcept { return m_cq; }
32+
const ::grpc::CompletionQueue& cq() const noexcept { return m_cq; }
33+
34+
static std::shared_ptr<client_cq> get_default();
35+
};
36+
} // namespace asyncpp::grpc

‎src/grpc/client_cq.cpp

+43
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
#include <asyncpp/grpc/client_cq.h>
2+
#include <asyncpp/ptr_tag.h>
3+
4+
namespace asyncpp::grpc {
5+
void client_cq::handle_event(size_t, bool) noexcept {
6+
m_alarm_set.clear();
7+
auto task = m_dispatched.pop();
8+
while (task) {
9+
if (*task) (*task)();
10+
task = m_dispatched.pop();
11+
}
12+
}
13+
14+
client_cq::client_cq() {
15+
m_thread = std::thread([this]() {
16+
#ifdef __linux__
17+
pthread_setname_np(pthread_self(), "grpc_client_cq");
18+
#endif
19+
void* tag = nullptr;
20+
bool ok = false;
21+
while (this->m_cq.Next(&tag, &ok)) {
22+
auto [i, t] = ptr_untag<calldata_interface>(tag);
23+
if (!i) continue;
24+
i->handle_event(t, ok);
25+
}
26+
});
27+
}
28+
29+
client_cq::~client_cq() {
30+
m_cq.Shutdown();
31+
if (m_thread.joinable()) m_thread.join();
32+
}
33+
34+
void client_cq::push(std::function<void()> fn) {
35+
m_dispatched.emplace(std::move(fn));
36+
if (!m_alarm_set.test_and_set()) { m_alarm.Set(&m_cq, gpr_time_0(GPR_CLOCK_REALTIME), ptr_tag<0, calldata_interface>(this)); }
37+
}
38+
39+
std::shared_ptr<client_cq> client_cq::get_default() {
40+
static auto instance = std::make_shared<client_cq>();
41+
return instance;
42+
}
43+
} // namespace asyncpp::grpc

0 commit comments

Comments
 (0)