asio_service support stream mode #527
Conversation
1. set TCP_NODELAY = true this can be set after the socket open. So we can't set it in constructor. (async_connect will open the socket) 2. does the order of calling all the when_done matter when error occur? in write path when_done(error triggered) -> close_socket(in destructor) in read path close_socket(directly call) -> when_done(error triggered) 3. keep the old path for all request if stream_mode_ = false already addressed all the comment.
src/asio_service.cxx
Outdated
| ptr<req_msg> get_req() { | ||
| return req; | ||
| } | ||
|
|
||
| rpc_handler& get_when_done() { | ||
| return when_done; | ||
| } | ||
|
|
||
| uint64_t get_timeout_ms() { | ||
| return timeout_ms; | ||
| } | ||
| private: |
There was a problem hiding this comment.
This is struct. Remove these functions and directly access members.
src/asio_service.cxx
Outdated
| @@ -150,6 +150,31 @@ asio_service::meta_cb_params req_to_params(req_msg* req, resp_msg* resp) { | |||
|
|
|||
| // === ASIO Abstraction === | |||
| // (to switch SSL <-> unsecure on-the-fly) | |||
| struct pending_req_pkg { | |||
| public: | |||
src/asio_service.cxx
Outdated
| ptr<req_msg> req; | ||
| rpc_handler when_done; | ||
| uint64_t timeout_ms; |
There was a problem hiding this comment.
Naming convention for member variables is using _ suffix: req_, when_done_, and timeout_ms_.
src/asio_service.cxx
Outdated
| rpc_handler& when_done, | ||
| uint64_t send_timeout_ms) { |
src/asio_service.cxx
Outdated
| rpc_handler& when_done, | ||
| uint64_t send_timeout_ms) { |
src/asio_service.cxx
Outdated
| // handle read queue in reverse | ||
| { | ||
| auto_lock(pending_read_reqs_lock_); | ||
| for (auto rit = pending_read_reqs_.rbegin(); rit != pending_read_reqs_.rend(); ++rit) { |
tests/CMakeLists.txt
Outdated
| add_executable(asio_service_stream_test | ||
| unit/asio_service_stream_test.cxx | ||
| ${EXAMPLES_SRC}/logger.cc | ||
| ${EXAMPLES_SRC}/in_memory_log_store.cxx) | ||
| add_dependencies(asio_service_stream_test | ||
| static_lib | ||
| build_ssl_key) | ||
| target_link_libraries(asio_service_stream_test | ||
| ${BUILD_DIR}/${LIBRARY_OUTPUT_NAME} | ||
| ${LIBRARIES}) |
| int my_id; | ||
| int port; | ||
| std::atomic<ulong> response_log_index; | ||
| ulong send_log_index = 0; | ||
| ptr<asio_service> asio_svc; | ||
| ptr<rpc_client> my_client; | ||
| ptr<rpc_listener> my_listener; | ||
| ptr<logger_wrapper> my_log_wrapper; | ||
| ptr<logger> my_log; | ||
| ptr<stream_msg_handler> my_msg_handler; |
There was a problem hiding this comment.
Use _ suffix for all member variable names.
| SimpleLogger* ll = my_log_wrapper->getLogger(); | ||
| _log_info(ll, "resp log index not match, resp: %ld, current: %ld", | ||
| resp->get_next_idx(), get_next_log_index()); |
There was a problem hiding this comment.
This is the whole point of this test (checking whether request and response are delivered in order). Rather than leaving a log, there should be a boolean flag to set if failure happens, and check CHK_FALSE at the end.
There was a problem hiding this comment.
If entering into this branch, this response_log_index will not increase. So it will fail at CHK_EQ(count, s.get_resp_log_index());. Those logs are just for debugging purpose.
| SimpleLogger* ll = my_log_wrapper->getLogger(); | ||
| _log_info(ll, "req log index not match, req: %ld, current: %ld", | ||
| req.get_last_log_idx(), streamed_log_index.load()); |
There was a problem hiding this comment.
Ditto, it should set a boolean flag and make test failed.
|
Also, can you add into |
already addressed all the comment.