Skip to content

Commit 5f3bd02

Browse files
poor-circlepoor-circle
authored
[doc] add coro_rpc doc (#681)
* add coro_rpc doc * fix doc * update en doc * fix code
1 parent f786c5a commit 5f3bd02

22 files changed

+1525
-309
lines changed

‎include/ylt/coro_rpc/impl/context.hpp

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -130,10 +130,12 @@ class context_base {
130130
/*finish here*/
131131
self_->status_ = context_status::finish_response;
132132
}
133-
const context_info_t<rpc_protocol> *get_context() const noexcept {
133+
const context_info_t<rpc_protocol> *get_context_info() const noexcept {
134+
return self_.get();
135+
}
136+
context_info_t<rpc_protocol> *get_context_info() noexcept {
134137
return self_.get();
135138
}
136-
context_info_t<rpc_protocol> *get_context() noexcept { return self_.get(); }
137139
};
138140

139141
template <typename T>

‎include/ylt/coro_rpc/impl/coro_rpc_client.hpp

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,11 @@ class client_pool;
7373

7474
namespace coro_rpc {
7575

76+
inline uint64_t get_global_client_id() {
77+
static std::atomic<uint64_t> cid = 0;
78+
return cid.fetch_add(1, std::memory_order::relaxed);
79+
}
80+
7681
#ifdef GENERATE_BENCHMARK_DATA
7782
std::string benchmark_file_path = "./";
7883
#endif
@@ -105,7 +110,7 @@ struct async_rpc_result_value_t {
105110
async_rpc_result_value_t(T &&result) : result_(std::move(result)) {}
106111
T &result() noexcept { return result_; }
107112
const T &result() const noexcept { return result_; }
108-
std::string_view attachment() const noexcept {
113+
std::string_view get_attachment() const noexcept {
109114
return buffer_.resp_attachment_buf_;
110115
}
111116
resp_body release_buffer() { return std::move(buffer_); }
@@ -155,12 +160,12 @@ class coro_rpc_client {
155160
const inline static rpc_error connect_error = {errc::io_error,
156161
"client has been closed"};
157162
struct config {
158-
uint32_t client_id = 0;
163+
uint64_t client_id = get_global_client_id();
159164
std::chrono::milliseconds timeout_duration =
160165
std::chrono::milliseconds{5000};
161166
std::string host;
162167
std::string port;
163-
bool enable_tcp_no_delay_ = true;
168+
bool enable_tcp_no_delay = true;
164169
#ifdef YLT_ENABLE_SSL
165170
std::filesystem::path ssl_cert_path;
166171
std::string ssl_domain;
@@ -172,7 +177,7 @@ class coro_rpc_client {
172177
* @param io_context asio io_context, async event handler
173178
*/
174179
coro_rpc_client(asio::io_context::executor_type executor,
175-
uint32_t client_id = 0)
180+
uint64_t client_id = get_global_client_id())
176181
: control_(std::make_shared<control_t>(executor, false)),
177182
timer_(std::make_unique<coro_io::period_timer>(executor)) {
178183
config_.client_id = client_id;
@@ -184,7 +189,7 @@ class coro_rpc_client {
184189
*/
185190
coro_rpc_client(
186191
coro_io::ExecutorWrapper<> *executor = coro_io::get_global_executor(),
187-
uint32_t client_id = 0)
192+
uint64_t client_id = get_global_client_id())
188193
: control_(
189194
std::make_shared<control_t>(executor->get_asio_executor(), false)),
190195
timer_(std::make_unique<coro_io::period_timer>(
@@ -424,7 +429,7 @@ class coro_rpc_client {
424429
ELOGV(WARN, "client_id %d connect timeout", config_.client_id);
425430
co_return errc::timed_out;
426431
}
427-
if (config_.enable_tcp_no_delay_ == true) {
432+
if (config_.enable_tcp_no_delay == true) {
428433
control_->socket_.set_option(asio::ip::tcp::no_delay(true), ec);
429434
}
430435

@@ -738,7 +743,7 @@ class coro_rpc_client {
738743
call<func>(std::forward<Args>(args)...));
739744
}
740745
#endif
741-
746+
private:
742747
template <auto func, typename... Args>
743748
async_simple::coro::Lazy<rpc_error> send_request_for_impl(
744749
auto duration, uint32_t &id, coro_io::period_timer &timer,

‎include/ylt/coro_rpc/impl/coro_rpc_server.hpp

Lines changed: 41 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
#include <memory>
3030
#include <mutex>
3131
#include <system_error>
32+
#include <thread>
3233
#include <unordered_map>
3334
#include <vector>
3435
#include <ylt/easylog.hpp>
@@ -68,29 +69,36 @@ class coro_rpc_server_base {
6869
* TODO: add doc
6970
* @param thread_num the number of io_context.
7071
* @param port the server port to listen.
72+
* @param listen address of server
7173
* @param conn_timeout_duration client connection timeout. 0 for no timeout.
7274
* default no timeout.
75+
* @param is_enable_tcp_no_delay is tcp socket allow
7376
*/
74-
coro_rpc_server_base(size_t thread_num, unsigned short port,
77+
coro_rpc_server_base(size_t thread_num = std::thread::hardware_concurrency(),
78+
unsigned short port = 9001,
7579
std::string address = "0.0.0.0",
7680
std::chrono::steady_clock::duration
77-
conn_timeout_duration = std::chrono::seconds(0))
81+
conn_timeout_duration = std::chrono::seconds(0),
82+
bool is_enable_tcp_no_delay = true)
7883
: pool_(thread_num),
7984
acceptor_(pool_.get_executor()->get_asio_executor()),
8085
port_(port),
8186
conn_timeout_duration_(conn_timeout_duration),
82-
flag_{stat::init} {
87+
flag_{stat::init},
88+
is_enable_tcp_no_delay_(is_enable_tcp_no_delay) {
8389
init_address(std::move(address));
8490
}
8591

86-
coro_rpc_server_base(size_t thread_num,
87-
std::string address /* = "0.0.0.0:9001" */,
92+
coro_rpc_server_base(size_t thread_num = std::thread::hardware_concurrency(),
93+
std::string address = "0.0.0.0:9001",
8894
std::chrono::steady_clock::duration
89-
conn_timeout_duration = std::chrono::seconds(0))
95+
conn_timeout_duration = std::chrono::seconds(0),
96+
bool is_enable_tcp_no_delay = true)
9097
: pool_(thread_num),
9198
acceptor_(pool_.get_executor()->get_asio_executor()),
9299
conn_timeout_duration_(conn_timeout_duration),
93-
flag_{stat::init} {
100+
flag_{stat::init},
101+
is_enable_tcp_no_delay_(is_enable_tcp_no_delay) {
94102
init_address(std::move(address));
95103
}
96104

@@ -99,7 +107,13 @@ class coro_rpc_server_base {
99107
acceptor_(pool_.get_executor()->get_asio_executor()),
100108
port_(config.port),
101109
conn_timeout_duration_(config.conn_timeout_duration),
102-
flag_{stat::init} {
110+
flag_{stat::init},
111+
is_enable_tcp_no_delay_(config.is_enable_tcp_no_delay) {
112+
#ifdef YLT_ENABLE_SSL
113+
if (config.ssl_config) {
114+
init_ssl_context_helper(context_, config.ssl_config.value());
115+
}
116+
#endif
103117
init_address(config.address);
104118
}
105119

@@ -109,7 +123,7 @@ class coro_rpc_server_base {
109123
}
110124

111125
#ifdef YLT_ENABLE_SSL
112-
void init_ssl_context(const ssl_configure &conf) {
126+
void init_ssl(const ssl_configure &conf) {
113127
use_ssl_ = init_ssl_context_helper(context_, conf);
114128
}
115129
#endif
@@ -122,19 +136,19 @@ class coro_rpc_server_base {
122136
* @return error code if start failed, otherwise block until server stop.
123137
*/
124138
[[nodiscard]] coro_rpc::err_code start() noexcept {
125-
auto ret = async_start();
126-
if (ret) {
127-
ret.value().wait();
128-
return ret.value().value();
129-
}
130-
else {
131-
return ret.error();
132-
}
139+
return async_start().get();
133140
}
134141

135-
[[nodiscard]] coro_rpc::expected<async_simple::Future<coro_rpc::err_code>,
136-
coro_rpc::err_code>
137-
async_start() noexcept {
142+
private:
143+
async_simple::Future<coro_rpc::err_code> make_error_future(
144+
coro_rpc::err_code &&err) {
145+
async_simple::Promise<coro_rpc::err_code> p;
146+
p.setValue(std::move(err));
147+
return p.getFuture();
148+
}
149+
150+
public:
151+
async_simple::Future<coro_rpc::err_code> async_start() noexcept {
138152
{
139153
std::unique_lock lock(start_mtx_);
140154
if (flag_ != stat::init) {
@@ -144,8 +158,8 @@ class coro_rpc_server_base {
144158
else if (flag_ == stat::stop) {
145159
ELOGV(INFO, "has stoped");
146160
}
147-
return coro_rpc::unexpected<coro_rpc::err_code>{
148-
coro_rpc::errc::server_has_ran};
161+
return make_error_future(
162+
coro_rpc::err_code{coro_rpc::errc::server_has_ran});
149163
}
150164
errc_ = listen();
151165
if (!errc_) {
@@ -177,7 +191,7 @@ class coro_rpc_server_base {
177191
return std::move(future);
178192
}
179193
else {
180-
return coro_rpc::unexpected<coro_rpc::err_code>{errc_};
194+
return make_error_future(coro_rpc::err_code{errc_});
181195
}
182196
}
183197

@@ -387,7 +401,9 @@ class coro_rpc_server_base {
387401

388402
int64_t conn_id = ++conn_id_;
389403
ELOGV(INFO, "new client conn_id %d coming", conn_id);
390-
socket.set_option(asio::ip::tcp::no_delay(true), error);
404+
if (is_enable_tcp_no_delay_) {
405+
socket.set_option(asio::ip::tcp::no_delay(true), error);
406+
}
391407
auto conn = std::make_shared<coro_connection>(executor, std::move(socket),
392408
conn_timeout_duration_);
393409
conn->set_quit_callback(
@@ -459,6 +475,7 @@ class coro_rpc_server_base {
459475

460476
std::atomic<uint16_t> port_;
461477
std::string address_;
478+
bool is_enable_tcp_no_delay_;
462479
coro_rpc::err_code errc_ = {};
463480
std::chrono::steady_clock::duration conn_timeout_duration_;
464481

‎include/ylt/coro_rpc/impl/default_config/coro_rpc_config.hpp

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,19 +25,22 @@
2525

2626
namespace coro_rpc {
2727

28-
namespace config {
29-
struct coro_rpc_config_base {
30-
uint16_t port = 8801;
28+
struct config_base {
29+
bool is_enable_tcp_no_delay = true;
30+
uint16_t port = 9001;
3131
unsigned thread_num = std::thread::hardware_concurrency();
3232
std::chrono::steady_clock::duration conn_timeout_duration =
3333
std::chrono::seconds{0};
34+
std::string address = "0.0.0.0";
35+
#ifdef YLT_ENABLE_SSL
36+
std::optional<ssl_configure> ssl_config = std::nullopt;
37+
#endif
3438
};
3539

36-
struct coro_rpc_default_config : public coro_rpc_config_base {
40+
struct config_t : public config_base {
3741
using rpc_protocol = coro_rpc::protocol::coro_rpc_protocol;
3842
using executor_pool_t = coro_io::io_context_pool;
3943
};
40-
} // namespace config
4144

42-
using coro_rpc_server = coro_rpc_server_base<config::coro_rpc_default_config>;
45+
using coro_rpc_server = coro_rpc_server_base<config_t>;
4346
} // namespace coro_rpc

‎src/coro_io/tests/test_channel.cpp

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ TEST_CASE("test RR") {
2323
async_simple::coro::syncAwait([]() -> async_simple::coro::Lazy<void> {
2424
coro_rpc::coro_rpc_server server(1, 8801);
2525
auto res = server.async_start();
26-
REQUIRE_MESSAGE(res, "server start failed");
26+
REQUIRE_MESSAGE(!res.hasResult(), "server start failed");
2727
auto hosts =
2828
std::vector<std::string_view>{"127.0.0.1:8801", "localhost:8801"};
2929
auto channel = coro_io::channel<coro_rpc::coro_rpc_client>::create(hosts);
@@ -62,10 +62,10 @@ TEST_CASE("test WRR") {
6262

6363
coro_rpc::coro_rpc_server server1(1, 8801);
6464
auto res = server1.async_start();
65-
REQUIRE_MESSAGE(res, "server start failed");
65+
REQUIRE_MESSAGE(!res.hasResult(), "server start failed");
6666
coro_rpc::coro_rpc_server server2(1, 8802);
6767
auto res2 = server2.async_start();
68-
REQUIRE_MESSAGE(res2, "server start failed");
68+
REQUIRE_MESSAGE(!res2.hasResult(), "server start failed");
6969

7070
async_simple::coro::syncAwait([]() -> async_simple::coro::Lazy<void> {
7171
auto hosts =
@@ -119,7 +119,7 @@ TEST_CASE("test Random") {
119119
async_simple::coro::syncAwait([]() -> async_simple::coro::Lazy<void> {
120120
coro_rpc::coro_rpc_server server(1, 8801);
121121
auto res = server.async_start();
122-
REQUIRE_MESSAGE(res, "server start failed");
122+
REQUIRE_MESSAGE(!res.hasResult(), "server start failed");
123123
auto hosts =
124124
std::vector<std::string_view>{"127.0.0.1:8801", "localhost:8801"};
125125
auto channel = coro_io::channel<coro_rpc::coro_rpc_client>::create(
@@ -148,7 +148,7 @@ TEST_CASE("test single host") {
148148
async_simple::coro::syncAwait([]() -> async_simple::coro::Lazy<void> {
149149
coro_rpc::coro_rpc_server server(1, 8801);
150150
auto res = server.async_start();
151-
REQUIRE_MESSAGE(res, "server start failed");
151+
REQUIRE_MESSAGE(!res.hasResult(), "server start failed");
152152
auto hosts = std::vector<std::string_view>{"127.0.0.1:8801"};
153153
auto channel = coro_io::channel<coro_rpc::coro_rpc_client>::create(hosts);
154154
for (int i = 0; i < 100; ++i) {
@@ -168,7 +168,7 @@ TEST_CASE("test send_request config") {
168168
async_simple::coro::syncAwait([]() -> async_simple::coro::Lazy<void> {
169169
coro_rpc::coro_rpc_server server(1, 9813);
170170
auto res = server.async_start();
171-
REQUIRE_MESSAGE(res, "server start failed");
171+
REQUIRE_MESSAGE(!res.hasResult(), "server start failed");
172172
auto hosts = std::vector<std::string_view>{"127.0.0.1:9813"};
173173
auto channel = coro_io::channel<coro_rpc::coro_rpc_client>::create(hosts);
174174
for (int i = 0; i < 100; ++i) {

‎src/coro_io/tests/test_client_pool.cpp

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@ TEST_CASE("test client pool") {
8787
async_simple::coro::syncAwait([]() -> async_simple::coro::Lazy<void> {
8888
coro_rpc::coro_rpc_server server(1, 8801);
8989
auto is_started = server.async_start();
90-
REQUIRE(is_started);
90+
REQUIRE(is_started.hasResult() == false);
9191
auto pool = coro_io::client_pool<coro_rpc::coro_rpc_client>::create(
9292
"127.0.0.1:8801", {.max_connection = 100,
9393
.idle_timeout = 300ms,
@@ -114,7 +114,7 @@ TEST_CASE("test idle timeout yield") {
114114
async_simple::coro::syncAwait([]() -> async_simple::coro::Lazy<void> {
115115
coro_rpc::coro_rpc_server server(1, 8801);
116116
auto is_started = server.async_start();
117-
REQUIRE(is_started);
117+
REQUIRE(!is_started.hasResult());
118118
auto pool = coro_io::client_pool<coro_rpc::coro_rpc_client>::create(
119119
"127.0.0.1:8801", {.max_connection = 100,
120120
.idle_queue_per_max_clear_count = 1,
@@ -142,7 +142,7 @@ TEST_CASE("test reconnect") {
142142
async_simple::Promise<async_simple::Unit> p;
143143
coro_io::sleep_for(700ms).start([&server, &p](auto &&) {
144144
auto server_is_started = server.async_start();
145-
REQUIRE(server_is_started);
145+
REQUIRE(!server_is_started.hasResult());
146146
});
147147

148148
auto res = co_await event(100, *pool, cv, lock);
@@ -177,7 +177,7 @@ TEST_CASE("test reconnect retry wait time exclude reconnect cost time") {
177177
async_simple::Promise<async_simple::Unit> p;
178178
coro_io::sleep_for(350ms).start([&server, &p](auto &&) {
179179
auto server_is_started = server.async_start();
180-
REQUIRE(server_is_started);
180+
REQUIRE(!server_is_started.hasResult());
181181
});
182182
auto res = co_await event<mock_client>(100, *pool, cv, lock);
183183
CHECK(res);
@@ -196,7 +196,7 @@ TEST_CASE("test collect_free_client") {
196196
async_simple::coro::syncAwait([]() -> async_simple::coro::Lazy<void> {
197197
coro_rpc::coro_rpc_server server(1, 8801);
198198
auto is_started = server.async_start();
199-
REQUIRE(is_started);
199+
REQUIRE(!is_started.hasResult());
200200
auto pool = coro_io::client_pool<coro_rpc::coro_rpc_client>::create(
201201
"127.0.0.1:8801", {.max_connection = 100, .idle_timeout = 300ms});
202202

‎src/coro_rpc/benchmark/data_gen.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ int main() {
3131
coro_rpc::coro_rpc_server server(std::thread::hardware_concurrency(), 0);
3232
register_handlers(server);
3333
auto started = server.async_start();
34-
if (!started) {
34+
if (started.hasResult()) {
3535
ELOGV(ERROR, "server started failed");
3636
return -1;
3737
}
@@ -118,7 +118,7 @@ int main() {
118118

119119
server.stop();
120120

121-
started->wait();
121+
started.wait();
122122

123123
pool.stop();
124124
thd.join();

‎src/coro_rpc/examples/base_examples/rpc_service.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ void async_echo_by_callback(
5252
/* rpc function runs in global io thread pool */
5353
coro_io::post([conn, data]() mutable {
5454
/* send work to global non-io thread pool */
55-
auto *ctx = conn.get_context();
55+
auto *ctx = conn.get_context_info();
5656
conn.response_msg(data); /*response here*/
5757
}).start([](auto &&) {
5858
});

‎src/coro_rpc/examples/base_examples/server.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ int main() {
3939
server2.register_handler<echo>();
4040
// async start server
4141
auto res = server2.async_start();
42-
assert(res.has_value());
42+
assert(!res.hasResult());
4343

4444
// sync start server & sync await server stop
4545
return !server.start();

0 commit comments

Comments
 (0)