Urma event mode - #6
Closed
Gzure wants to merge 137 commits into
Closed
Conversation
added 30 commits
June 4, 2026 17:29
- Add urma_socket_t class implementing ib_socket_t-compatible interface - Add urma_device.hpp with device discovery and management - Add urma_buffer.hpp with buffer pool for memory registration - Add YLT_ENABLE_URMA CMake option (default OFF) - Integrate URMA into socket_wrapper_t visitor dispatch - Add init_urma() method to coro_rpc_server This enables coro_rpc to use URMA as an alternative RDMA transport on Kunpeng hardware for improved performance.
…flict with URMA library type - Rename wrapper class to urma_device_wrapper_t - Use ::urma_device_t to explicitly refer to URMA library type - Add backward compatibility alias using urma_device_t = urma_device_wrapper_t - Fix all references to URMA library functions to use :: scope resolution
- Rename local urma_buffer_t to urma_buf_t to avoid conflict with URMA library type - Fix sge.length to sge.len per URMA API - Remove lkey from subview() as URMA uses UBVA addressing
- Reorganize urma_socket.hpp based on URMA documentation - Use urma_buf_t with urma_target_seg_t instead of lkey - Add proper JFC/JFR/Jetty creation flow - Add register_buffer/unregister_buffer for segment management - Fix urma_sge_t field usage (len instead of length) Note: Still needs fixes for API details (alloc_jfc, jfc_cfg, etc)
Known issues remaining: - urma_query_jetty signature incorrect - urma_import_jetty parameters need urma_rjetty_t structure - ASIO coroutine compatibility issues (await_ready) - Socket wrapper interface mismatches
- Fix urma_query_jetty call - access Jetty ID directly via jetty_->jetty_id.id
- Fix urma_import_jetty parameters - properly initialize urma_rjetty_t with
trans_mode, type, tp_type, and flag fields
- Fix executor type mismatch - use async_simple::CurrentExecutor{} and proper cast
- Add missing urma_socket_t methods: prepare_accept, get_remote_address,
get_local_address, get_remote_qp_num, get_local_qp_num
- Add urma_md5_header and urma_md5_first_header for protocol identification
- Fix waiting_write_over - stub out until Future awaiting is fixed
- Initialize socket properly in shared state with placement new
Note: urma_example.cpp still needs integration fixes for coro_rpc_client
Summary of changes committed: - urma_socket.hpp: Fix URMA CTP API usage (jetty_id access, import_jetty params) - Add missing methods: prepare_accept, get_remote/local_address, get_qp_num - Add urma_md5_header constants for protocol identification Remaining issues: - urma_example.cpp needs URMA config/variant integration with coro_rpc_client - get_global_urma_device() needs to accept configuration parameters Next steps documented in .claude/plans/urma-ctp-next-steps.md
- Add executor_type, lowest_layer_type, next_layer() for ASIO compatibility - Add async_read_some, async_write_some, cancel methods - Add async_connect overloads to support endpoint-based connection - Fix prepare_accept socket initialization with std::optional - Add urma_socket_t overload for async_connect in coro_io.hpp - Update urma_example to use new get_global_urma_device API
- Add template connect(endpoint) method for URMA CTP mode - Delegate TCP handshake to coro_io::async_connect(socket, endpoint) - Remove special urma async_connect overload from coro_io.hpp - URMA now works like ib_socket: connect(endpoint) handles full flow
idle_poll_interval_ is now a runtime value (was constexpr 5us). Default 5us, configurable via URMA_RPC_POLL_INTERVAL env var (in us). URMA_RPC_POLL_INTERVAL=5 # 5us (default) URMA_RPC_POLL_INTERVAL=50 # 50us URMA_RPC_POLL_INTERVAL=0 # 0us (busy spin)
SIGBUS root cause: wait_urma_write_completion called socket.poll_completion_once() in a spin loop, racing with event_loop's poll_completion on the same shared state (send_callbacks_, recv_queue_, recv_callback_). Two concurrent poll_completion calls can double-free buffers or access freed memory -> SIGBUS. Fix: revert to callback_awaitor only (no direct poll). The send completion is handled exclusively by event_loop's poll_completion. Also fix idle_poll_interval_ default from 5000us back to 5us.
…request Previously benchmark_rpc_call was only recorded in urma_benchmark's issue_rpc_call. Mooncake and other users of coro_rpc_client didn't get total RPC latency profiling. Add profiling in send_request() itself: records from request start (send_request_for_impl) through response deserialize, covering the full client-side RPC round-trip for all callers.
send_request returns Lazy<Lazy<T>> (double wrapped), so co_await in send_request can't record after deserialize. Move profiling to call() which does co_await co_await send_request (double unwrap), recording benchmark_rpc_call after the full RPC round-trip completes.
send_request returns Lazy<Lazy<T>> so can't co_await internally. Pass rpc_begin to deserialize_rpc_result which runs inside the inner Lazy and records benchmark_rpc_call before each co_return. Covers both call() and send_request() paths (Mooncake uses send_request directly via client_pool).
Record response payload size (body + attachment) for benchmark_rpc_call so it shows up in payload-size buckets in the tree output.
All profiling stages now carry payload size for bucketed statistics: - client_recv_header: response payload size (body + attachment) - server_read_header: request payload size (length + attach_length) - server_serialize_response: response body + attachment size - server_response_queue: 0 (too short to matter) - urma read/write stages: actual transfer length - connect/handshake stages: 0
Add buckets() calls after parent() for: - benchmark.rpc_call (total) - client.send_request / recv_header / recv_payload - server.read_header / read_payload / dispatch Each shows per-payload-size-bucket avg/p50/p99/max under the parent stage line. Stages without meaningful payload variation (connect, serialize, response_queue) are left as single-line summary.
benchmark_rpc_call now buckets by request body size (not response), so PutStart/PutEnd/GetReplicaList with different request sizes show up in different buckets consistently. Other stages remain as-is: request stages use request size, response stages use response size.
…size Replace payload-size bucketing for benchmark_rpc_call with per-function- name grouping. Each RPC function (PutStart, PutEnd, GetReplicaList, etc.) gets its own latency statistics line. New APIs: record_rpc_call_by_name(func_name, duration_ns) record_rpc_call_by_name_since(func_name, begin_ns) func_name stored in control_t::last_func_name, passed to deserialize_rpc_result which records by name. urma_benchmark's issue_rpc_call no longer needs manual profiling - send_request -> deserialize_rpc_result handles it automatically.
get_func_name<func>() may return empty string on some GCC versions or for member function pointers. Fall back to __PRETTY_FUNCTION__ and extract the function name after the last :: separator.
record_rpc_call_by_name now increments counters[benchmark_rpc_call] so has() returns true. print() computes total + per-func-name stats from the func-name map directly, not from the bucket-based merged array.
Remove the record_since fallback that caused payload-size bucketing to still appear. If func_name is empty, use 'rpc_call' as the name.
|
for detail, goto summary download Artifacts |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why
What is changing
Example