Skip to content

iocp: add poll, through the AFD driver #240

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
endel wants to merge 1 commit into
base: main
Choose a base branch
from
Open

iocp: add poll, through the AFD driver #240

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
127 changes: 127 additions & 0 deletions src/backend/iocp.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,9 @@ pub const Loop = struct {
/// Heap of timers.
timers: TimerHeap = .{ .context = {} },

/// The AFD driver handle poll requests go through, opened on first use.
afd_handle: ?windows.HANDLE = null,

/// Cached time
cached_now: u64,

Expand Down Expand Up @@ -97,6 +100,7 @@ pub const Loop = struct {
/// Deinitialize the loop, this closes the handle to the Completion Port. Any events that were
/// unprocessed are lost -- their callbacks will never be called.
pub fn deinit(self: *Loop) void {
if (self.afd_handle) |h| windows.CloseHandle(h);
windows.CloseHandle(self.iocp_handle);
}

Expand Down Expand Up @@ -726,6 +730,20 @@ pub const Loop = struct {
break :action .{ .submitted = {} };
},

.poll => |*v| action: {
const afd_handle = self.afd() catch |err| break :action .{ .result = .{ .poll = err } };
const base = windows.afd.baseHandle(v.fd) catch |err| break :action .{ .result = .{ .poll = err } };
v.info = .{
.timeout = std.math.maxInt(i64),
.number_of_handles = 1,
.exclusive = 0,
.handles = .{.{ .handle = base, .events = v.events, .status = .SUCCESS }},
};
windows.afd.poll(afd_handle, &v.info, &completion.overlapped) catch |err|
break :action .{ .result = .{ .poll = err } };
break :action .{ .submitted = {} };
},

.timer => |*v| action: {
v.c = completion;
self.timers.insert(v);
Expand Down Expand Up @@ -844,6 +862,12 @@ pub const Loop = struct {
}
},

.poll => {
if (completion.flags.state == .active) {
cancel_result.?.* = windows.afd.cancel(self.afd_handle.?, &completion.overlapped);
}
},

else => @panic("Not implemented"),
}
}
Expand All @@ -864,6 +888,17 @@ pub const Loop = struct {
};
}

/// The AFD handle, opened and associated with the port the first time a
/// poll needs it.
fn afd(self: *Loop) PollError!windows.HANDLE {
if (self.afd_handle) |h| return h;
const h = try windows.afd.open();
errdefer windows.CloseHandle(h);
self.associate_fd(h) catch return error.Unexpected;
self.afd_handle = h;
return h;
}

/// Associate a handler to the internal completion port.
/// This has to be done only once per handle so we delegate the responsibility to the caller.
pub fn associate_fd(self: Loop, fd: windows.HANDLE) !void {
Expand Down Expand Up @@ -1179,6 +1214,12 @@ pub const Completion = struct {
.async_wait => .{ .async_wait = {} },

.job_object => self.result.?,

.poll => .{ .poll = switch (windows.afd.completionStatus(&self.overlapped)) {
.SUCCESS => {},
.CANCELLED => error.Canceled,
else => |status| std.os.windows.unexpectedStatus(status),
} },
};
}

Expand Down Expand Up @@ -1248,8 +1289,20 @@ pub const OperationType = enum {

/// Receive a notification from a job object associated with a completion port
job_object,

/// Wait for a socket to become ready, through the AFD driver.
poll,
};

/// What a read poll waits for: data, or anything that means a read will no
/// longer block.
pub const poll_read_events = windows.afd.POLL_RECEIVE |
windows.afd.POLL_ACCEPT |
windows.afd.POLL_DISCONNECT |
windows.afd.POLL_ABORT |
windows.afd.POLL_LOCAL_CLOSE |
windows.afd.POLL_CONNECT_FAIL;

/// All the supported operations of this event loop. These are always
/// backend-specific and therefore the structure and types change depending
/// on the underlying system in use. The high level operations are
Expand Down Expand Up @@ -1345,6 +1398,16 @@ pub const Operation = union(OperationType) {
/// Do not use this, it is used internally.
associated: bool = false,
},

poll: struct {
fd: windows.HANDLE,

/// AFD_POLL_* events to wait for.
events: u32,

/// The request while AFD holds it. Do not use this, it is used internally.
info: windows.afd.PollInfo = undefined,
},
};

/// The result type based on the operation type. For a callback, the
Expand All @@ -1367,12 +1430,18 @@ pub const Result = union(OperationType) {
cancel: CancelError!void,
async_wait: AsyncError!void,
job_object: JobObjectError!JobObjectResult,
poll: PollError!void,
};

pub const CancelError = error{
Unexpected,
};

pub const PollError = error{
Canceled,
Unexpected,
};

pub const AcceptError = error{
AddressFamilyNotSupported,
ProcessFdQuotaExceeded,
Expand Down Expand Up @@ -2349,6 +2418,64 @@ test "iocp: recv cancellation" {
try testing.expectError(error.Canceled, recv_result.recv);
}

test "iocp: poll" {
const mem = std.mem;
const testing = std.testing;

var loop = try Loop.init(.{});
defer loop.deinit();

const address = try net.Address.parseIp4("127.0.0.1", 3132);
const socket = try windows.WSASocketW(windows.ws2_32.AF.INET, windows.ws2_32.SOCK.DGRAM, windows.ws2_32.IPPROTO.UDP, null, 0, windows.ws2_32.WSA_FLAG_OVERLAPPED);
defer iocpClose(socket);
try iocpSetsockopt(asSocket(socket), windows.ws2_32.SOL.SOCKET, windows.ws2_32.SO.REUSEADDR, &mem.toBytes(@as(c_int, 1)));
try iocpBind(asSocket(socket), &address.any, address.getOsSockLen());

const Poll = struct {
result: ?(PollError!void) = null,

fn callback(ud: ?*anyopaque, _: *Loop, _: *Completion, r: Result) CallbackAction {
const self: *@This() = @ptrCast(@alignCast(ud.?));
self.result = r.poll;
return .disarm;
}
};

// Nothing to read, so the poll stays pending until it is canceled.
var canceled: Poll = .{};
var c_canceled: Completion = .{
.op = .{ .poll = .{ .fd = socket, .events = poll_read_events } },
.userdata = &canceled,
.callback = Poll.callback,
};
loop.add(&c_canceled);
try loop.run(.no_wait);
try testing.expect(canceled.result == null);

var c_cancel: Completion = .{ .op = .{ .cancel = .{ .c = &c_canceled } } };
loop.add(&c_cancel);
try loop.run(.until_done);
try testing.expectError(error.Canceled, canceled.result.?);

// A datagram arriving completes it.
var readable: Poll = .{};
var c_readable: Completion = .{
.op = .{ .poll = .{ .fd = socket, .events = poll_read_events } },
.userdata = &readable,
.callback = Poll.callback,
};
loop.add(&c_readable);
try loop.run(.no_wait);
try testing.expect(readable.result == null);

const sender = try windows.WSASocketW(windows.ws2_32.AF.INET, windows.ws2_32.SOCK.DGRAM, windows.ws2_32.IPPROTO.UDP, null, 0, windows.ws2_32.WSA_FLAG_OVERLAPPED);
defer iocpClose(sender);
try testing.expectEqual(4, windows.ws2_32.sendto(asSocket(sender), "ping", 4, 0, &address.any, @intCast(address.getOsSockLen())));

try loop.run(.until_done);
try readable.result.?;
}

test "iocp: accept cancellation" {
const mem = std.mem;
const testing = std.testing;
Expand Down
21 changes: 14 additions & 7 deletions src/watcher/stream.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,12 +60,12 @@ pub fn Shared(comptime xev: type) type {
pub const PollError = switch (xev.backend) {
.io_uring,
.epoll,
.iocp,
=> xev.Sys.PollError,

.kqueue,
=> xev.ReadError,

.iocp,
.wasi_poll,
=> error{},
};
Expand All @@ -75,8 +75,9 @@ pub fn Shared(comptime xev: type) type {
read = switch (xev.backend) {
.io_uring => std.posix.POLL.IN,
.epoll => std.os.linux.EPOLL.IN,
.iocp => xev.Sys.poll_read_events,
.kqueue => 0, // doesn't matter
.iocp, .wasi_poll => 0, // invalid
.wasi_poll => 0, // invalid
},

fn fromResult(
Expand All @@ -86,6 +87,7 @@ pub fn Shared(comptime xev: type) type {
return switch (xev.backend) {
.io_uring,
.epoll,
.iocp,
=> if (result.poll) |_|
@enumFromInt(c.op.poll.events)
else |err|
Expand All @@ -96,7 +98,6 @@ pub fn Shared(comptime xev: type) type {
else => unreachable,
},

.iocp,
.wasi_poll,
=> @compileError("poll not supported on this backend"),
};
Expand Down Expand Up @@ -221,8 +222,8 @@ fn Pollable(comptime xev: type, comptime T: type, comptime options: Options) typ

// Do not add the methods for poll if the backend doesn't support it.
switch (xev.backend) {
.io_uring, .epoll, .kqueue => {},
.iocp, .wasi_poll => return struct {},
.io_uring, .epoll, .kqueue, .iocp => {},
.wasi_poll => return struct {},
}

return struct {
Expand Down Expand Up @@ -263,6 +264,13 @@ fn Pollable(comptime xev: type, comptime T: type, comptime options: Options) typ
},
} },

.iocp => .{ .poll = .{
.fd = self.fd,
.events = switch (event) {
.read => xev.Sys.poll_read_events,
},
} },

.kqueue => switch (options.read) {
.none => unreachable,

Expand All @@ -277,7 +285,6 @@ fn Pollable(comptime xev: type, comptime T: type, comptime options: Options) typ
} },
},

.iocp,
.wasi_poll,
=> @compileError("poll not supported on this backend"),
},
Expand All @@ -292,11 +299,11 @@ fn Pollable(comptime xev: type, comptime T: type, comptime options: Options) typ
const fd: Self = switch (xev.backend) {
.io_uring,
.epoll,
.iocp,
=> T.initFd(c_inner.op.poll.fd),

.kqueue => T.initFd(c_inner.op.read.fd),

.iocp,
.wasi_poll,
=> @compileError("poll not supported on this backend"),
};
Expand Down
Loading