Skip to content

Ergonomics: Treat PollCompletionQueueError::CompletionQueueEmpty as an empty iterator to simplify completion polling #86

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vtta wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Ergonomics: Treat PollCompletionQueueError::CompletionQueueEmpty as an empty iterator to simplify completion polling #86

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bb826b9
feat(ibverbs): allow getting the lid from PortAttr
vtta Nov 20, 2025
76455a2
chore: better naming
vtta Dec 2, 2025
250519c
refactor(ibverbs): ergonomically treat CompletionQueueEmpty as an emp…
vtta Dec 3, 2025
cd39297
chore: fix a sq/rq typo
vtta Dec 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
143 changes: 65 additions & 78 deletions examples/rc_pingpong.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -326,89 +326,76 @@ fn main() -> anyhow::Result<()> {
// poll for the completion
{
loop {
match cq.start_poll() {
Ok(mut poller) => {
while let Some(wc) = poller.next() {
if wc.status() != WorkCompletionStatus::Success as u32 {
panic!(
"Failed status {:#?} ({}) for wr_id {}",
Into::<WorkCompletionStatus>::into(wc.status()),
wc.status(),
wc.wr_id()
);
}
match wc.wr_id() {
SEND_WR_ID => {
scnt += 1;
outstanding_send = false;
},
RECV_WR_ID => {
rcnt += 1;
rout -= 1;

// Post more receives if the receive side credit is low
if rout <= rx_depth / 2 {
let to_post = rx_depth - rout;
for _ in 0..to_post {
let mut guard = qp.start_post_recv();
let recv_handle = guard.construct_wr(RECV_WR_ID);
unsafe {
recv_handle.setup_sge(
recv_mr.lkey(),
recv_data.as_mut_ptr() as _,
args.size,
);
};
guard.post().unwrap();
}
rout += to_post;
}

if args.ts {
let timestamp = wc.completion_timestamp();
if ts_param.last_completion_with_timestamp != 0 {
let delta: u64 = if timestamp >= ts_param.completion_recv_prev_time {
timestamp - ts_param.completion_recv_prev_time
} else {
completion_timestamp_mask - ts_param.completion_recv_prev_time
+ timestamp
+ 1
};

ts_param.completion_recv_max_time_delta =
ts_param.completion_recv_max_time_delta.max(delta);
ts_param.completion_recv_min_time_delta =
ts_param.completion_recv_min_time_delta.min(delta);
ts_param.completion_recv_total_time_delta += delta;
ts_param.completion_with_time_iters += 1;
}

ts_param.completion_recv_prev_time = timestamp;
ts_param.last_completion_with_timestamp = 1;
} else {
ts_param.last_completion_with_timestamp = 0;
}
},
_ => {
panic!("Unknown error!");
},
for wc in cq.iter()? {
if wc.status() != WorkCompletionStatus::Success as u32 {
panic!(
"Failed status {:#?} ({}) for wr_id {}",
Into::<WorkCompletionStatus>::into(wc.status()),
wc.status(),
wc.wr_id()
);
}
match wc.wr_id() {
SEND_WR_ID => {
scnt += 1;
outstanding_send = false;
},
RECV_WR_ID => {
rcnt += 1;
rout -= 1;

// Post more receives if the receive side credit is low
if rout <= rx_depth / 2 {
let to_post = rx_depth - rout;
for _ in 0..to_post {
let mut guard = qp.start_post_recv();
let recv_handle = guard.construct_wr(RECV_WR_ID);
unsafe {
recv_handle.setup_sge(recv_mr.lkey(), recv_data.as_mut_ptr() as _, args.size);
};
guard.post().unwrap();
}
rout += to_post;
}

if scnt < args.iter && !outstanding_send {
// Post another send if we haven't reached the iteration limit
let mut guard = qp.start_post_send();
let send_handle = guard.construct_wr(SEND_WR_ID, WorkRequestFlags::Signaled).setup_send();
unsafe {
send_handle.setup_sge(send_mr.lkey(), send_data.as_ptr() as _, args.size);
if args.ts {
let timestamp = wc.completion_timestamp();
if ts_param.last_completion_with_timestamp != 0 {
let delta: u64 = if timestamp >= ts_param.completion_recv_prev_time {
timestamp - ts_param.completion_recv_prev_time
} else {
completion_timestamp_mask - ts_param.completion_recv_prev_time + timestamp + 1
};

ts_param.completion_recv_max_time_delta =
ts_param.completion_recv_max_time_delta.max(delta);
ts_param.completion_recv_min_time_delta =
ts_param.completion_recv_min_time_delta.min(delta);
ts_param.completion_recv_total_time_delta += delta;
ts_param.completion_with_time_iters += 1;
}
guard.post()?;
outstanding_send = true;

ts_param.completion_recv_prev_time = timestamp;
ts_param.last_completion_with_timestamp = 1;
} else {
ts_param.last_completion_with_timestamp = 0;
}
},
_ => {
panic!("Unknown error!");
},
}

if scnt < args.iter && !outstanding_send {
// Post another send if we haven't reached the iteration limit
let mut guard = qp.start_post_send();
let send_handle = guard.construct_wr(SEND_WR_ID, WorkRequestFlags::Signaled).setup_send();
unsafe {
send_handle.setup_sge(send_mr.lkey(), send_data.as_ptr() as _, args.size);
}
},
Err(_) => {
continue;
},
guard.post()?;
outstanding_send = true;
}
}

// Check if we're done
Expand Down
45 changes: 19 additions & 26 deletions examples/rc_pingpong_split.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -457,32 +457,25 @@ fn main() -> anyhow::Result<()> {
let mut need_post_send = false;

{
match ctx.cq.start_poll() {
Ok(mut poller) => {
while let Some(wc) = poller.next() {
ctx.parse_single_work_completion(
&wc,
&mut ts_param,
&mut scnt,
&mut rcnt,
&mut outstanding_send,
&mut rout,
rx_depth,
&mut need_post_recv,
&mut to_post_recv,
args.ts,
);

// Record that we need to post a send later
if scnt < args.iter && !outstanding_send {
need_post_send = true;
outstanding_send = true;
}
}
},
Err(_) => {
continue;
},
for wc in ctx.cq.iter()? {
ctx.parse_single_work_completion(
&wc,
&mut ts_param,
&mut scnt,
&mut rcnt,
&mut outstanding_send,
&mut rout,
rx_depth,
&mut need_post_recv,
&mut to_post_recv,
args.ts,
);

// Record that we need to post a send later
if scnt < args.iter && !outstanding_send {
need_post_send = true;
outstanding_send = true;
}
}
}

Expand Down
Loading
Loading