this repo has no description
op: add readv support
+58
+36
src/Kqueue.zig
+36
src/Kqueue.zig
···
344
344
}
345
345
},
346
346
347
+
.readv => |req| {
348
+
self.in_flight.push(task);
349
+
const kevent = evSet(@intCast(req.fd), EVFILT.READ, EV.ADD | EV.ONESHOT, task);
350
+
try self.submission_queue.append(self.gpa, kevent);
351
+
},
352
+
347
353
.recv => |req| {
348
354
self.in_flight.push(task);
349
355
const kevent = evSet(@intCast(req.fd), EVFILT.READ, EV.ADD | EV.ONESHOT, task);
···
485
491
const kevent = evSet(@intCast(cancel_req.fd), EVFILT.WRITE, EV.DELETE, task);
486
492
try self.submission_queue.append(self.gpa, kevent);
487
493
}
494
+
},
495
+
496
+
.readv => |cancel_req| {
497
+
self.in_flight.remove(task);
498
+
task.result = .{ .readv = error.Canceled };
499
+
const kevent = evSet(
500
+
@intCast(cancel_req.fd),
501
+
EVFILT.READ,
502
+
EV.DELETE,
503
+
task,
504
+
);
505
+
try self.submission_queue.append(self.gpa, kevent);
488
506
},
489
507
490
508
.recv => |cancel_req| {
···
654
672
// async tasks. These can be handled synchronously in a cancel all
655
673
.accept,
656
674
.poll,
675
+
.readv,
657
676
.recv,
658
677
.write,
659
678
.writev,
···
700
719
.msg_ring => .{ .msg_ring = error.Canceled },
701
720
.noop => unreachable,
702
721
.poll => .{ .poll = error.Canceled },
722
+
.readv => .{ .readv = error.Canceled },
703
723
.recv => .{ .recv = error.Canceled },
704
724
.socket => .{ .socket = error.Canceled },
705
725
.statx => .{ .statx = error.Canceled },
···
783
803
const err = unexpectedError(dataToE(event.data));
784
804
task.result = .{ .poll = err };
785
805
} else task.result = .{ .poll = {} };
806
+
return task.callback(rt, task.*);
807
+
},
808
+
809
+
.readv => |req| {
810
+
defer self.releaseTask(rt, task);
811
+
self.in_flight.remove(task);
812
+
if (event.flags & EV.ERROR != 0) {
813
+
// Interpret data as an errno
814
+
const err = unexpectedError(dataToE(event.data));
815
+
task.result = .{ .readv = err };
816
+
return task.callback(rt, task.*);
817
+
}
818
+
if (posix.readv(req.fd, req.vecs)) |n|
819
+
task.result = .{ .readv = n }
820
+
else |_|
821
+
task.result = .{ .readv = error.Unexpected };
786
822
return task.callback(rt, task.*);
787
823
},
788
824
+2
src/Mock.zig
+2
src/Mock.zig
···
19
19
msg_ring_cb: ?*const fn (*io.Task) io.Result = null,
20
20
noop_cb: ?*const fn (*io.Task) io.Result = null,
21
21
poll_cb: ?*const fn (*io.Task) io.Result = null,
22
+
readv_cb: ?*const fn (*io.Task) io.Result = null,
22
23
recv_cb: ?*const fn (*io.Task) io.Result = null,
23
24
socket_cb: ?*const fn (*io.Task) io.Result = null,
24
25
statx_cb: ?*const fn (*io.Task) io.Result = null,
···
69
70
.msg_ring => if (self.msg_ring_cb) |cb| cb(task) else return error.NoMockCallback,
70
71
.noop => if (self.noop_cb) |cb| cb(task) else return error.NoMockCallback,
71
72
.poll => if (self.poll_cb) |cb| cb(task) else return error.NoMockCallback,
73
+
.readv => if (self.readv_cb) |cb| cb(task) else return error.NoMockCallback,
72
74
.recv => if (self.recv_cb) |cb| cb(task) else return error.NoMockCallback,
73
75
.socket => if (self.socket_cb) |cb| cb(task) else return error.NoMockCallback,
74
76
.statx => if (self.statx_cb) |cb| cb(task) else return error.NoMockCallback,
+14
src/Uring.zig
+14
src/Uring.zig
···
225
225
self.prepDeadline(task, sqe);
226
226
},
227
227
228
+
.readv => |req| {
229
+
const sqe = self.getSqe();
230
+
sqe.prep_readv(req.fd, req.vecs, 0);
231
+
sqe.user_data = @intFromPtr(task);
232
+
self.prepDeadline(task, sqe);
233
+
},
234
+
228
235
// user* is only sent internally between rings and higher level wrappers
229
236
.userfd, .usermsg, .userptr => unreachable,
230
237
}
···
349
356
350
357
.statx => |req| .{ .statx = switch (cqeToE(cqe.res)) {
351
358
.SUCCESS => req.result,
359
+
.INVAL => io.ResultError.Invalid,
360
+
.CANCELED => io.ResultError.Canceled,
361
+
else => |e| unexpectedError(e),
362
+
} },
363
+
364
+
.readv => .{ .readv = switch (cqeToE(cqe.res)) {
365
+
.SUCCESS => @intCast(cqe.res),
352
366
.INVAL => io.ResultError.Invalid,
353
367
.CANCELED => io.ResultError.Canceled,
354
368
else => |e| unexpectedError(e),
+6
src/main.zig
+6
src/main.zig
···
493
493
socket,
494
494
connect,
495
495
statx,
496
+
readv,
496
497
497
498
/// userfd is meant to send file descriptors between Ring instances (using msgRing)
498
499
userfd,
···
546
547
path: [:0]const u8,
547
548
result: *Statx, // this will be filled in by the op
548
549
},
550
+
readv: struct {
551
+
fd: posix.fd_t,
552
+
vecs: []const posix.iovec,
553
+
},
549
554
550
555
userfd,
551
556
usermsg,
···
567
572
socket: ResultError!posix.fd_t,
568
573
connect: ResultError!void,
569
574
statx: ResultError!*Statx,
575
+
readv: ResultError!usize,
570
576
571
577
userfd: anyerror!posix.fd_t,
572
578
usermsg: u16,