diff --git a/src/backend/epoll.zig b/src/backend/epoll.zig index 25c0a931..bec44407 100644 --- a/src/backend/epoll.zig +++ b/src/backend/epoll.zig @@ -11,6 +11,12 @@ const heap = @import("../heap.zig"); const ThreadPool = @import("../ThreadPool.zig"); const Async = @import("../main.zig").Epoll.Async; +/// Sending on a socket whose peer has gone fails with EPIPE, and also raises +/// SIGPIPE, which kills the process by default. A library can't ignore the +/// signal for its whole process, so every send asks the kernel not to raise +/// it; the caller still gets error.BrokenPipe. +const send_flags: u32 = linux.MSG.NOSIGNAL; + /// In Zig 0.16, `std.posix.epoll_create1`, `epoll_ctl`, and `epoll_wait` /// were removed. These helpers replicate the old wrappers using the raw /// Linux syscall layer (`std.os.linux`). @@ -1074,13 +1080,13 @@ pub const Completion = struct { .send => |*op| .{ .send = switch (op.buffer) { - .slice => |v| xev_posix.send(op.fd, v, 0), - .array => |*v| xev_posix.send(op.fd, v.array[0..v.len], 0), + .slice => |v| xev_posix.send(op.fd, v, send_flags), + .array => |*v| xev_posix.send(op.fd, v.array[0..v.len], send_flags), }, }, .sendmsg => |*op| .{ - .sendmsg = if (xev_posix.sendmsg(op.fd, op.msghdr, 0)) |v| + .sendmsg = if (xev_posix.sendmsg(op.fd, op.msghdr, send_flags)) |v| v else |err| err, diff --git a/src/backend/io_uring.zig b/src/backend/io_uring.zig index 6891bc3a..4bbfd557 100644 --- a/src/backend/io_uring.zig +++ b/src/backend/io_uring.zig @@ -14,6 +14,12 @@ const noopCallback = looppkg.NoopCallback(@This()); pub const ShutdownHow = std.Io.net.ShutdownHow; +/// Sending on a socket whose peer has gone fails with EPIPE, and also raises +/// SIGPIPE, which kills the process by default. A library can't ignore the +/// signal for its whole process, so every send asks the kernel not to raise +/// it; the completion still gets error.BrokenPipe. +const send_flags: u32 = linux.MSG.NOSIGNAL; + /// True if this backend is available on this platform. pub fn available() bool { if (comptime builtin.os.tag != .linux) return false; @@ -479,13 +485,13 @@ pub const Loop = struct { .array => |*buf| sqe.prep_send( v.fd, buf.array[0..buf.len], - 0, + send_flags, ), .slice => |buf| sqe.prep_send( v.fd, buf, - 0, + send_flags, ), }, @@ -497,7 +503,7 @@ pub const Loop = struct { sqe.prep_sendmsg( v.fd, v.msghdr, - 0, + send_flags, ); }, diff --git a/src/backend/kqueue.zig b/src/backend/kqueue.zig index a78249dd..568ecbb6 100644 --- a/src/backend/kqueue.zig +++ b/src/backend/kqueue.zig @@ -24,6 +24,13 @@ const noopCallback = looppkg.NoopCallback(@This()); const log = std.log.scoped(.libxev_kqueue); +/// Sending on a socket whose peer has gone fails with EPIPE, and also raises +/// SIGPIPE, which kills the process by default. A library can't ignore the +/// signal for its whole process, so every send asks the kernel not to raise +/// it; the completion still gets error.BrokenPipe. Darwin and FreeBSD both +/// define MSG_NOSIGNAL; SO_NOSIGPIPE only covers sockets their owner set it on. +const send_flags: u32 = if (@hasDecl(posix.MSG, "NOSIGNAL")) posix.MSG.NOSIGNAL else 0; + /// True if this backend is available on this platform. pub fn available() bool { return switch (builtin.os.tag) { @@ -1175,15 +1182,15 @@ pub const Completion = struct { .send => |*op| .{ .send = switch (op.buffer) { - .slice => |v| xev_posix.send(op.fd, v, 0) catch |err| mapWriteError(err), - .array => |*v| xev_posix.send(op.fd, v.array[0..v.len], 0) catch |err| mapWriteError(err), + .slice => |v| xev_posix.send(op.fd, v, send_flags) catch |err| mapWriteError(err), + .array => |*v| xev_posix.send(op.fd, v.array[0..v.len], send_flags) catch |err| mapWriteError(err), }, }, .sendto => |*op| .{ .sendto = switch (op.buffer) { - .slice => |v| xev_posix.sendto(op.fd, v, 0, &op.addr.any, op.addr.getOsSockLen()) catch |err| mapWriteError(err), - .array => |*v| xev_posix.sendto(op.fd, v.array[0..v.len], 0, &op.addr.any, op.addr.getOsSockLen()) catch |err| mapWriteError(err), + .slice => |v| xev_posix.sendto(op.fd, v, send_flags, &op.addr.any, op.addr.getOsSockLen()) catch |err| mapWriteError(err), + .array => |*v| xev_posix.sendto(op.fd, v.array[0..v.len], send_flags, &op.addr.any, op.addr.getOsSockLen()) catch |err| mapWriteError(err), }, }, @@ -1530,6 +1537,8 @@ fn mapReadError(err: anyerror) ReadError { fn mapWriteError(err: anyerror) WriteError { return switch (err) { error.AccessDenied, error.PermissionDenied => error.PermissionDenied, + error.BrokenPipe => error.BrokenPipe, + error.ConnectionResetByPeer => error.ConnectionResetByPeer, else => error.Unexpected, }; } @@ -1731,6 +1740,8 @@ pub const ReadError = KEventError || error{ pub const WriteError = KEventError || error{ Canceled, + BrokenPipe, + ConnectionResetByPeer, PermissionDenied, Unexpected, }; diff --git a/src/watcher/tcp.zig b/src/watcher/tcp.zig index 040378f8..8fd65612 100644 --- a/src/watcher/tcp.zig +++ b/src/watcher/tcp.zig @@ -694,6 +694,139 @@ fn TCPTests(comptime xev: type, comptime Impl: type) type { try testing.expect(server_closed); } + test "TCP: a send that fails with EPIPE raises no SIGPIPE" { + // We have no way to get a socket in WASI from a WASI context. + if (builtin.os.tag == .wasi) return error.SkipZigTest; + // No SIGPIPE on Windows. + if (builtin.os.tag == .windows) return error.SkipZigTest; + + const testing = std.testing; + + // The test runner's std.Io.Threaded already catches SIGPIPE, so + // the process wouldn't die of it here: count it instead. + const Sigpipe = struct { + var raised = std.atomic.Value(u32).init(0); + fn handler(_: posix.SIG) callconv(.c) void { + _ = raised.fetchAdd(1, .monotonic); + } + }; + var old_action: posix.Sigaction = undefined; + posix.sigaction(.PIPE, &.{ + .handler = .{ .handler = Sigpipe.handler }, + .mask = posix.sigemptyset(), + .flags = 0, + }, &old_action); + defer posix.sigaction(.PIPE, &old_action, null); + Sigpipe.raised.store(0, .monotonic); + + var tpool = ThreadPool.init(.{}); + defer tpool.deinit(); + defer tpool.shutdown(); + var loop = try xev.Loop.init(.{ .thread_pool = &tpool }); + defer loop.deinit(); + + var address = try std.Io.net.IpAddress.parse("127.0.0.1", 0); + const server = try Impl.init(address); + try server.bind(address); + try server.listen(1); + var internal_addr = net.Address.fromIpAddress(address); + var sock_len = internal_addr.getOsSockLen(); + try xev_posix.getsockname(if (xev.dynamic) server.fd() else server.fd, &internal_addr.any, &sock_len); + address = internal_addr.toIpAddress(); + const client = try Impl.init(address); + + var c_accept: xev.Completion = undefined; + var c_connect: xev.Completion = undefined; + var server_conn: ?Impl = null; + server.accept(&loop, &c_accept, ?Impl, &server_conn, (struct { + fn callback( + ud: ?*?Impl, + _: *xev.Loop, + _: *xev.Completion, + r: xev.AcceptError!Impl, + ) xev.CallbackAction { + ud.?.* = r catch unreachable; + return .disarm; + } + }).callback); + var connected = false; + client.connect(&loop, &c_connect, address, bool, &connected, (struct { + fn callback( + ud: ?*bool, + _: *xev.Loop, + _: *xev.Completion, + _: Impl, + r: xev.ConnectError!void, + ) xev.CallbackAction { + _ = r catch unreachable; + ud.?.* = true; + return .disarm; + } + }).callback); + try loop.run(.until_done); + try testing.expect(server_conn != null); + try testing.expect(connected); + + // With its sending side shut, the socket fails every send with + // EPIPE, as one does once the peer has reset the connection. + var shut = false; + client.shutdown(&loop, &c_connect, bool, &shut, (struct { + fn callback( + ud: ?*bool, + _: *xev.Loop, + _: *xev.Completion, + _: Impl, + r: xev.ShutdownError!void, + ) xev.CallbackAction { + _ = r catch unreachable; + ud.?.* = true; + return .disarm; + } + }).callback); + try loop.run(.until_done); + try testing.expect(shut); + + // Without MSG_NOSIGNAL this send also raises SIGPIPE, which kills + // a process that hasn't caught or ignored it. + const Result = xev.WriteError!usize; + var result: ?Result = null; + var send_buf = [_]u8{ 1, 2, 3 }; + client.write(&loop, &c_connect, .{ .slice = &send_buf }, ?Result, &result, (struct { + fn callback( + ud: ?*?Result, + _: *xev.Loop, + _: *xev.Completion, + _: Impl, + _: xev.WriteBuffer, + r: Result, + ) xev.CallbackAction { + ud.?.* = r; + return .disarm; + } + }).callback); + try loop.run(.until_done); + try testing.expectError(error.BrokenPipe, result.?); + try testing.expectEqual(@as(u32, 0), Sigpipe.raised.load(.monotonic)); + + const Closed = struct { + fn callback( + _: ?*void, + _: *xev.Loop, + _: *xev.Completion, + _: Impl, + r: xev.CloseError!void, + ) xev.CallbackAction { + _ = r catch unreachable; + return .disarm; + } + }; + var c_close: [3]xev.Completion = undefined; + server.close(&loop, &c_close[0], void, null, Closed.callback); + server_conn.?.close(&loop, &c_close[1], void, null, Closed.callback); + client.close(&loop, &c_close[2], void, null, Closed.callback); + try loop.run(.until_done); + } + // Potentially flaky - this test could hang if the sender is unable to // write everything to the socket for whatever reason // (e.g. incorrectly sized buffer on the receiver side), or if the