authorgravatar for igor.anic@gmail.comIgor Anić <igor.anic@gmail.com> 2024-03-31 23:57:16+02:00
committergravatar for igor.anic@gmail.comIgor Anić <igor.anic@gmail.com> 2024-03-31 23:57:16+02:00
log254c05a9e117cd4926eb8c3bb1d83e9137b53553
treeeb1dc24ab1c0a866d4e6ffd9d3f17e209fbc2034
parent704660c81a3e53713645528e658881283af013dd

io_uring: simplify copy_cqe logic

First copy as much as we can in this cycle. If there is more needed wrap and start from the buffer 0 position.

1 files changed, 46 insertions(+), 25 deletions(-)

lib/std/os/linux/IoUring.zig+46-25
......@@ -282,19 +282,15 @@ fn copy_cqes_ready(self: *IoUring, cqes: []linux.io_uring_cqe) u32 {
282282 const ready = self.cq_ready();
283283 const count = @min(cqes.len, ready);
284284 const head = self.cq.head.* & self.cq.mask;
285 const tail = (self.cq.head.* +% count) & self.cq.mask;
286
287 if (head < tail) {
288 // head behind tail -> no wrapping
289 @memcpy(cqes[0..count], self.cq.cqes[head..tail]);
290 } else {
291 // head in front of tail -> buffer wraps
292 const two_copies_required: bool = self.cq.cqes.len - head < count;
293 const amount_to_copy_in_first = if (two_copies_required) self.cq.cqes.len - head else count;
294 @memcpy(cqes[0..amount_to_copy_in_first], self.cq.cqes[head .. head + amount_to_copy_in_first]);
295 if (two_copies_required) {
296 @memcpy(cqes[amount_to_copy_in_first..count], self.cq.cqes[0..tail]);
297 }
285
286 // before wrapping
287 const n = @min(self.cq.cqes.len - head, count);
288 @memcpy(cqes[0..n], self.cq.cqes[head..][0..n]);
289
290 if (count > n) {
291 // wrap self.cq.cqes
292 const w = count - n;
293 @memcpy(cqes[n..][0..w], self.cq.cqes[0..w]);
298294 }
299295
300296 self.cq_advance(count);
......@@ -4231,24 +4227,49 @@ fn expect_buf_grp_cqe(
42314227 return cqe;
42324228}
42334229
4234test "failing test for issue 19451" {
4235 var ring = try IoUring.init(2, 0);
4230test "copy_cqes with wrapping sq.cqes buffer" {
4231 if (!is_linux) return error.SkipZigTest;
4232
4233 var ring = IoUring.init(2, 0) catch |err| switch (err) {
4234 error.SystemOutdated => return error.SkipZigTest,
4235 error.PermissionDenied => return error.SkipZigTest,
4236 else => return err,
4237 };
42364238 defer ring.deinit();
42374239
42384240 try testing.expectEqual(2, ring.sq.sqes.len);
42394241 try testing.expectEqual(4, ring.cq.cqes.len);
42404242
4241 for (0..4) |i| {
4242 const sqe = try ring.get_sqe();
4243 sqe.prep_timeout(&.{ .tv_sec = 0, .tv_nsec = 10000 }, 0, 0);
4244 sqe.user_data = i;
4245 _ = try ring.submit();
4243 // submit 2 entries, receive 2 completions
4244 var cqes: [8]linux.io_uring_cqe = undefined;
4245 {
4246 for (0..2) |_| {
4247 const sqe = try ring.get_sqe();
4248 sqe.prep_timeout(&.{ .tv_sec = 0, .tv_nsec = 10000 }, 0, 0);
4249 try testing.expect(try ring.submit() == 1);
4250 }
4251 var cqe_count: u32 = 0;
4252 while (cqe_count < 2) {
4253 cqe_count += try ring.copy_cqes(&cqes, 2 - cqe_count);
4254 }
42464255 }
42474256
4248 var cqe_count: u32 = 0;
4249 while (cqe_count < 4) {
4250 var cqes: [8]linux.io_uring_cqe = undefined;
4251 cqe_count += try ring.copy_cqes(&cqes, 4 - cqe_count);
4257 try testing.expectEqual(2, ring.cq.head.*);
4258
4259 // sq.sqes len is 4, starting at position 2
4260 // every 4 entries submit wraps completion buffer
4261 // we are reading ring.cq.cqes at indexes 2,3,0,1
4262 for (1..1024) |i| {
4263 for (0..4) |_| {
4264 const sqe = try ring.get_sqe();
4265 sqe.prep_timeout(&.{ .tv_sec = 0, .tv_nsec = 10000 }, 0, 0);
4266 try testing.expect(try ring.submit() == 1);
4267 }
4268 var cqe_count: u32 = 0;
4269 while (cqe_count < 4) {
4270 cqe_count += try ring.copy_cqes(&cqes, 4 - cqe_count);
4271 }
4272 try testing.expectEqual(4, cqe_count);
4273 try testing.expectEqual(2 + 4 * i, ring.cq.head.*);
42524274 }
4253 try testing.expectEqual(4, cqe_count);
42544275}