diff --git a/zig/src/families/dsv41/dsv41.zig b/zig/src/families/dsv41/dsv41.zig index 04b13a25a7..d672bdec9b 100644 --- a/zig/src/families/dsv41/dsv41.zig +++ b/zig/src/families/dsv41/dsv41.zig @@ -16,6 +16,7 @@ pub const weights = @import("weights.zig"); pub const engram_io = @import("engram_io.zig"); pub const engram_aio = @import("engram_aio.zig"); pub const link = @import("link.zig"); +pub const peer_watch = @import("peer_watch.zig"); pub const comm = @import("comm.zig"); pub const tri = @import("tri.zig"); pub const tri_basic = @import("tri_basic.zig"); @@ -64,6 +65,7 @@ test { refAll(engram_io); refAll(engram_aio); refAll(link); + refAll(peer_watch); refAll(comm); refAll(tri); refAll(tri_basic); diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig index afef27f1cb..23114e348d 100644 --- a/zig/src/families/dsv41/lanes.zig +++ b/zig/src/families/dsv41/lanes.zig @@ -20,6 +20,8 @@ const round = @import("round.zig"); const draft = @import("draft.zig"); const sampling = @import("sampling.zig"); const link = @import("link.zig"); +const peer_watch = @import("peer_watch.zig"); +const rdma = @import("rdma.zig"); const kept_order = @import("kept_order.zig"); const cuda = @import("cuda"); @@ -295,6 +297,8 @@ pub const Lanes = struct { gpa: std.mem.Allocator, m: *Model, peer: ?link.Link, // rank 1 (null: one rank); every follower gets each frame (Model.followers) + watch: ?*peer_watch.Watch = null, // the links to the followers, watched for a rank that dies (peer_watch.zig) + stop_requested: bool = false, // set by the server's stop (native.zig deinitFn): deinit then ends the watch orderly vision: ?*@import("vision.zig").Vision = null, // rank 0's tower (a vision checkpoint served with its kit's vision/) step: u64 = 0, wire: Wire = .{}, @@ -315,8 +319,9 @@ pub const Lanes = struct { snaps: std.ArrayList(*Snap) = .empty, // kept prompt states, oldest first next_snap: u64 = 1, - pub fn init(gpa: std.mem.Allocator, m: *Model) Lanes { + pub fn init(gpa: std.mem.Allocator, m: *Model) !Lanes { var ln: Lanes = .{ .gpa = gpa, .m = m, .peer = if (m.world > 1) m.peer() else null }; + if (m.world > 1) ln.watch = try watchLinks(gpa, m, m.followers()); // the round costs by rows: ROUND_MS[min(rows, len) - 1] (the served table, or TF_DS_ROUND_MS) const t = m.round_ms; for (&ln.wc, 0..) |*c, i| c.* = .{ .width = @intCast(i + 1), .ms = t[@min(i, t.len - 1)] }; @@ -324,6 +329,12 @@ pub const Lanes = struct { return ln; } + /// Ends the peer link watch (Watch.end) before the model frees its links and rings; may exit 75 on `failed`. + pub fn endWatch(self: *Lanes, how: peer_watch.How) void { + if (self.watch) |w| _ = w.end(self.gpa, how); + self.watch = null; + } + /// Tells rank 1 to stop, then frees the host state (the model is the caller's). pub fn deinit(self: *Lanes) void { for (self.snaps.items) |sn| { @@ -331,7 +342,10 @@ pub const Lanes = struct { self.gpa.destroy(sn); } self.snaps.deinit(self.gpa); - if (self.peer != null) self.send(.done) catch {}; + // a SIGTERM stop is orderly; else a follower gone before the done frame, or a done frame that fails, is a fault + const lost = if (self.watch) |w| w.faulted() else false; + const sent = if (self.peer != null) (if (self.send(.done)) |_| true else |_| false) else true; + self.endWatch(if (self.stop_requested or (!lost and sent)) .done else .failed); self.streams.deinit(self.gpa); for (&self.seqs) |*q| q.deinit(self.gpa); self.wire.deinit(self.gpa); @@ -961,6 +975,25 @@ fn refused(err: anyerror) void { /// A follower's verify times with Model.prof (a profile on ranks 1-3: rounds, issue and wall ns summed). pub var follow_prof: struct { rounds: u64 = 0, enqueue: u64 = 0, forward: u64 = 0 } = .{}; +/// A watch on `links`; if it cannot start, the error with TF_DS_EXIT_ON_FAULT=1 (exit 75 needs it), else null. +fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) !?*peer_watch.Watch { + var fds: [3]std.posix.fd_t = undefined; + var n: usize = 0; + for (links) |l| if (l) |x| if (n < fds.len) { + fds[n] = x.fd; + n += 1; + }; + if (n == 0) return null; + // every ring a GPU wait of this rank reads (the same list that Model.synchronize checks) + var rings: [6]*rdma.Ring = undefined; + const grace = peer_watch.faultGrace(); + return peer_watch.Watch.start(gpa, fds[0..n], m.rdmaRings(&rings), m.rank, grace) catch |err| { + std.debug.print("[tensorfold] the peer link watch did not start ({s})\n", .{@errorName(err)}); + if (grace != null) return err; + return null; + }; +} + /// Rank 1 (each of ranks 1-3 on the four-node split): the primitives rank 0 sends, run in order until it says done. pub fn follow(gpa: std.mem.Allocator, m: *Model) !void { const peer = m.peer(); @@ -981,6 +1014,11 @@ pub fn follow(gpa: std.mem.Allocator, m: *Model) !void { while (ki.next()) |b| b.free(); kept.deinit(gpa); } + // a dead rank 0 ends this rank too (peer_watch.zig); started after the kept states, so it ends before they free + const watch = try watchLinks(gpa, m, &.{peer}); + errdefer { + if (watch) |w| _ = w.end(gpa, .failed); + } while (true) { // the next primitive, polled for 20 ms before the read blocks (rank 0 sends it a fraction of a millisecond // after this rank's GPU goes idle in a decode; a sleeping thread would add its core's wake-up to every frame) @@ -1118,7 +1156,10 @@ pub fn follow(gpa: std.mem.Allocator, m: *Model) !void { b.free(); } }, - .done => return, + .done => { + if (watch) |w| _ = w.end(gpa, .done); + return; + }, } } } diff --git a/zig/src/families/dsv41/lanes_main.zig b/zig/src/families/dsv41/lanes_main.zig index 122e26496c..5607c32e8a 100644 --- a/zig/src/families/dsv41/lanes_main.zig +++ b/zig/src/families/dsv41/lanes_main.zig @@ -185,7 +185,7 @@ pub fn main(init: std.process.Init) !u8 { const text = try std.Io.Dir.cwd().readFileAlloc(io, requests_file orelse return error.NoRequests, a, .limited(1 << 30)); const Requests = struct { requests: []Request }; const reqs = (try std.json.parseFromSliceLeaky(Requests, a, text, .{ .ignore_unknown_fields = true })).requests; - var ln = dsv41.lanes.Lanes.init(gpa, m); + var ln = try dsv41.lanes.Lanes.init(gpa, m); defer ln.deinit(); ln.fix_k = fix_k; if (profile) { diff --git a/zig/src/families/dsv41/link.zig b/zig/src/families/dsv41/link.zig index e6a75ecae3..9556b2a1b6 100644 --- a/zig/src/families/dsv41/link.zig +++ b/zig/src/families/dsv41/link.zig @@ -3,6 +3,11 @@ const std = @import("std"); const c = std.c; +/// TCP_USER_TIMEOUT (Linux): the longest time sent data stays not acknowledged before the kernel ends the link. +const tcp_user_timeout: u32 = 18; +/// The same bound as the keepalive (10 s + 3 x 3 s, rounded up); also the end of an idle link's unanswered probes. +const user_timeout_ms: u32 = 20_000; + pub const Link = struct { fd: c.fd_t, @@ -21,6 +26,12 @@ pub const Link = struct { const busy_us: c_int = 50; _ = c.setsockopt(fd, c.IPPROTO.TCP, std.posix.TCP.NODELAY, std.mem.asBytes(&one), @sizeOf(c_int)); _ = c.setsockopt(fd, c.SOL.SOCKET, c.SO.BUSY_POLL, std.mem.asBytes(&busy_us), @sizeOf(c_int)); + // a vanished node (no FIN) shows in about 20 s to a read and to peer_watch.zig: keepalive, TCP_USER_TIMEOUT + _ = c.setsockopt(fd, c.SOL.SOCKET, c.SO.KEEPALIVE, std.mem.asBytes(&one), @sizeOf(c_int)); + for ([_][2]u32{ .{ std.posix.TCP.KEEPIDLE, 10 }, .{ std.posix.TCP.KEEPINTVL, 3 }, .{ std.posix.TCP.KEEPCNT, 3 }, .{ tcp_user_timeout, user_timeout_ms } }) |o| { + const v: c_int = @intCast(o[1]); + _ = c.setsockopt(fd, c.IPPROTO.TCP, o[0], std.mem.asBytes(&v), @sizeOf(c_int)); + } } fn addr(ip: [4]u8, port: u16) c.sockaddr.in { @@ -64,7 +75,8 @@ pub const Link = struct { fn writeAll(l: Link, bytes: []const u8) !void { var done: usize = 0; while (done < bytes.len) { - const n = c.send(l.fd, bytes.ptr + done, bytes.len - done, 0); + // NOSIGNAL: a send to a gone peer fails with PeerDown (a SIGPIPE would end the rank with no exit 75) + const n = c.send(l.fd, bytes.ptr + done, bytes.len - done, c.MSG.NOSIGNAL); if (n <= 0) { if (n < 0 and c.errno(n) == .INTR) continue; return error.PeerDown; @@ -155,6 +167,38 @@ test "addresses parse as four bytes" { try std.testing.expectError(error.BadAddress, parseIp("10.200.30")); } +fn tcpOption(fd: c.fd_t, level: i32, name: u32) !c_int { + var v: c_int = 0; + var len: c.socklen_t = @sizeOf(c_int); + if (c.getsockopt(fd, level, name, &v, &len) != 0) return error.GetSockOptFailed; + return v; +} + +test "both ends of a link find a gone peer within 20 s: keepalive and TCP_USER_TIMEOUT" { + const io = std.testing.io; + const port: u16 = 28400 + @as(u16, @intCast(@mod(std.Io.Timestamp.now(io, .awake).nanoseconds, 500))); + const Peer = struct { + fn run(p: u16, out: *[2]c_int) void { + var l = Link.connect(std.testing.io, .{ 127, 0, 0, 1 }, p, 10) catch return; + defer l.close(); + out[0] = tcpOption(l.fd, c.IPPROTO.TCP, tcp_user_timeout) catch -1; + out[1] = tcpOption(l.fd, c.SOL.SOCKET, c.SO.KEEPALIVE) catch -1; + var buf: [8]u8 = undefined; + _ = l.recv(&buf) catch {}; // until rank 0 closes its end + } + }; + var follower: [2]c_int = .{ 0, 0 }; + const t = try std.Thread.spawn(.{}, Peer.run, .{ port, &follower }); + var l = try Link.listen(.{ 127, 0, 0, 1 }, port); + // rank 0's end (the accepted socket) and rank 1's end (connect) + try std.testing.expectEqual(@as(c_int, user_timeout_ms), try tcpOption(l.fd, c.IPPROTO.TCP, tcp_user_timeout)); + try std.testing.expectEqual(@as(c_int, 1), try tcpOption(l.fd, c.SOL.SOCKET, c.SO.KEEPALIVE)); + try std.testing.expectEqual(@as(c_int, 10), try tcpOption(l.fd, c.IPPROTO.TCP, std.posix.TCP.KEEPIDLE)); + l.close(); + t.join(); + try std.testing.expectEqual([2]c_int{ user_timeout_ms, 1 }, follower); +} + test "frames cross a loopback link in order" { const io = std.testing.io; const port: u16 = 29000 + @as(u16, @intCast(@mod(std.Io.Timestamp.now(io, .awake).nanoseconds, 500))); diff --git a/zig/src/families/dsv41/model.zig b/zig/src/families/dsv41/model.zig index 36dbc93f2d..c70c366d6b 100644 --- a/zig/src/families/dsv41/model.zig +++ b/zig/src/families/dsv41/model.zig @@ -448,6 +448,31 @@ pub const Model = struct { fn synchronize(m: *Model) !void { try m.stream.synchronize(); if (m.neighbor_gather) |*ng| try ng.check(); + try m.ringsOk(); + } + + /// After a stream sync: a ring the peer watch aborted during the work may leave stale peer slots, so it fails. + fn ringsOk(m: *Model) !void { + var buf: [6]*rdma.Ring = undefined; + if (rdma.anyFailed(m.rdmaRings(&buf))) return error.RdmaTransportFailed; + } + + /// Every RDMA ring a GPU wait of this rank reads (TP2, ring2d.zig, neighbor.zig): the peer watch aborts these. + pub fn rdmaRings(m: *Model, buf: *[6]*rdma.Ring) []*rdma.Ring { + var k: usize = 0; + if (m.ring) |r| { + buf[k] = r; + k += 1; + } + if (m.rdma_rings) |rr| for ([_]*rdma.Ring{ rr.all4, rr.col, rr.pair }) |r| { + buf[k] = r; + k += 1; + }; + if (m.neighbor_gather) |ng| for ([_]*rdma.Ring{ ng.first, ng.second }) |r| { + buf[k] = r; + k += 1; + }; + return buf[0..k]; } /// TP2's RDMA ring for the decode gathers: created on both ranks, the queue-pair infos exchanged over NCCL, @@ -831,5 +856,7 @@ pub const Model = struct { const pool = &m.dpool.?; for (q0, slots) |q, s| pool.absorbed[@intCast(s)] = @intCast(q); try dr.pass(&m.eng, &m.ch, pool, tokens, q0, slots, steps, null); + // the drafter syncs its stream itself (draft.zig), so its drafts get the same ring check as synchronize + try m.ringsOk(); } }; diff --git a/zig/src/families/dsv41/native.zig b/zig/src/families/dsv41/native.zig index 194d43d13e..fd121a2e8c 100644 --- a/zig/src/families/dsv41/native.zig +++ b/zig/src/families/dsv41/native.zig @@ -159,7 +159,8 @@ pub fn open(gpa: std.mem.Allocator, io: std.Io, ctx: *const cuda.Context, dir: [ .streams = if (getenv("TF_DS_STREAMS")) |v| std.fmt.parseInt(usize, v, 10) catch return error.BadStreams else model.max_streams_default, }); errdefer st.m.close(); - st.lanes = lanes_mod.Lanes.init(gpa, st.m); + st.lanes = try lanes_mod.Lanes.init(gpa, st.m); + errdefer st.lanes.endWatch(.failed); // a failed start closes the model next: the watch ends first if (getenv("TF_DS_FIX_K")) |t| st.lanes.fix_k = std.fmt.parseInt(usize, t, 10) catch return error.BadFixK; if (getenv("TF_DS_SERVED_K")) |t| st.lanes.served_k = if (std.mem.eql(u8, t, "1") or std.mem.eql(u8, t, "all")) .all else if (std.mem.eql(u8, t, "solo")) .solo else .off; // a vision checkpoint's tower on this rank, from the kit's vision/ (torch's attention cubin, the routing bias) @@ -175,6 +176,7 @@ pub fn open(gpa: std.mem.Allocator, io: std.Io, ctx: *const cuda.Context, dir: [ fn deinitFn(ptr: *anyopaque) void { const st: *State = @ptrCast(@alignCast(ptr)); + st.lanes.stop_requested = true; // the server stops (its SIGTERM, or its start failed after open): an orderly end st.lanes.deinit(); // rank 1 hears done if (st.vision) |v| v.close(); st.m.close(); diff --git a/zig/src/families/dsv41/peer_watch.zig b/zig/src/families/dsv41/peer_watch.zig new file mode 100644 index 0000000000..f33b6729a9 --- /dev/null +++ b/zig/src/families/dsv41/peer_watch.zig @@ -0,0 +1,344 @@ +//! A watch on the ranks' TCP links: a gone peer aborts the RDMA rings, so no request hangs, then optionally exits 75. +const std = @import("std"); +const rdma = @import("rdma.zig"); +const posix = std.posix; +const Allocator = std.mem.Allocator; + +/// POLLRDHUP (Linux): the peer closed its end. +const rdhup: i16 = 0x2000; +/// The peer closed its end, the link failed (keepalive or send), or the descriptor closed (POLLNVAL would repeat). +const gone: i16 = rdhup | posix.POLL.HUP | posix.POLL.ERR | posix.POLL.NVAL; +const max_links = 3; +/// TP2's ring, world 4's three rings (ring2d.zig), or the neighbor gather's two (neighbor.zig). +const max_rings = 6; + +/// How a rank's work with its peers ended, as its caller knows it. +pub const How = enum { + /// rank 0 sent its done frame, or a follower took it: a link that closes after it is expected + done, + /// an error (a failed start, a failed send or receive, a failed round) + failed, +}; + +/// What Watch.end found. +pub const Outcome = enum { + /// the done frame: no fault + orderly, + /// an error while every peer link is still up: this rank's own fault (its caller reports it) + failed, + /// an error with a peer's link gone: with TF_DS_EXIT_ON_FAULT=1, end does not return (exit 75) + peer_fault, +}; + +pub const Watch = struct { + fds: [max_links]posix.fd_t = @splat(-1), + n: usize = 0, + rings: [max_rings]*rdma.Ring = undefined, + n_rings: usize = 0, + rank: u32, + running: std.atomic.Value(bool) = .init(true), + /// Set when the poll found a link's end (before the one-second wait for an orderly stop). + seen: std.atomic.Value(bool) = .init(false), + /// Set when the watch thread acted on a peer gone (it aborted the rings). + lost: std.atomic.Value(bool) = .init(false), + /// Held while a thread aborts the rings (a spin: it is taken at most twice). + rings_busy: std.atomic.Value(bool) = .init(false), + /// The exit 75 on a peer fault, after this grace in ms (faultGrace: TF_DS_EXIT_ON_FAULT=1); null: no exit. + exit_grace: ?u64, + thread: ?std.Thread = null, + + /// A thread that polls `fds` (this rank's peer links) and aborts `rings` when a peer is gone; see faultGrace. + pub fn start(gpa: Allocator, fds: []const posix.fd_t, rings: []const *rdma.Ring, rank: u32, exit_grace: ?u64) !*Watch { + const w = try gpa.create(Watch); + errdefer gpa.destroy(w); + w.* = .{ .rank = rank, .n = @min(fds.len, max_links), .n_rings = @min(rings.len, max_rings), .exit_grace = exit_grace }; + @memcpy(w.fds[0..w.n], fds[0..w.n]); + @memcpy(w.rings[0..w.n_rings], rings[0..w.n_rings]); + openLog(); + w.thread = try std.Thread.spawn(.{}, loop, .{w}); + return w; + } + + /// True when a peer's link has ended (seen by the thread or by a poll now); rank 0 reads it before its done frame. + pub fn faulted(w: *Watch) bool { + return w.seen.load(.acquire) or w.linkGone(); + } + + /// The one teardown, for every way a rank ends, always before the caller frees the rings (Model.close). + pub fn end(w: *Watch, gpa: Allocator, how: How) Outcome { + // on an error, no GPU wait of this rank waits on a gone peer + if (how == .failed) w.abortRings("this rank ends on an error"); + w.running.store(false, .release); + // no thread reads the rings after end returns + if (w.thread) |t| t.join(); + const out: Outcome = switch (how) { + .done => .orderly, + .failed => if (w.faulted()) .peer_fault else .failed, + }; + const rank = w.rank; + const exit_on_fault = w.exit_grace != null; + gpa.destroy(w); + // a normal exit could wait on the gone rank in Model.close + if (out == .peer_fault and exit_on_fault) { + note("[tensorfold] rank {d}: a peer rank is gone; exit 75 (TF_DS_EXIT_ON_FAULT): a supervisor restarts both ranks\n", .{rank}); + std.c._exit(75); + } + return out; + } + + fn abortRings(w: *Watch, why: []const u8) void { + while (w.rings_busy.swap(true, .acquire)) std.atomic.spinLoopHint(); + defer w.rings_busy.store(false, .release); + for (w.rings[0..w.n_rings]) |r| r.abort(why); + } + + /// A peer's link has ended now (a poll that does not wait), maybe before the watch thread saw it. + fn linkGone(w: *Watch) bool { + var pfds: [max_links]posix.pollfd = undefined; + for (pfds[0..w.n], w.fds[0..w.n]) |*p, fd| p.* = .{ .fd = fd, .events = rdhup, .revents = 0 }; + const k = posix.poll(pfds[0..w.n], 0) catch return false; + if (k == 0) return false; + for (pfds[0..w.n]) |p| if (p.revents & gone != 0) return true; + return false; + } + + fn quiet(w: *Watch) bool { + return !w.running.load(.acquire); + } + + /// Waits up to `ms`; true if end came in that time. + fn stopped(w: *Watch, ms: u64) bool { + var left = ms; + while (left > 0) : (left -|= 100) { + if (w.quiet()) return true; + const ts: std.c.timespec = .{ .sec = 0, .nsec = 100 * std.time.ns_per_ms }; + _ = std.c.nanosleep(&ts, null); + } + return w.quiet(); + } + + fn loop(w: *Watch) void { + var pfds: [max_links]posix.pollfd = undefined; + for (pfds[0..w.n], w.fds[0..w.n]) |*p, fd| p.* = .{ .fd = fd, .events = rdhup, .revents = 0 }; + while (!w.quiet()) { + const k = posix.poll(pfds[0..w.n], 200) catch { + sleepMs(200); // no spin on a poll that fails at once + continue; + }; + if (k == 0) continue; + for (pfds[0..w.n]) |p| if (p.revents & gone != 0) { + w.seen.store(true, .release); + return w.peerGone(); + }; + } + } + + /// A peer gone while this rank does not end (an idle rank, or a round whose waits got no error completion). + fn peerGone(w: *Watch) void { + // rank 0's done frame and the close of its link come together: a second for this rank to call end + if (w.stopped(1000)) return; + w.lost.store(true, .release); + // the abort first: no log comes before it + w.abortRings("a peer rank is gone (its link closed)"); + note("[tensorfold] rank {d}: a peer rank's link closed (its process or node is gone); the waits of its {d} RDMA ring(s) are aborted\n", .{ w.rank, w.n_rings }); + const grace_ms = w.exit_grace orelse return; + // a live round fails first: its error goes out, then this rank calls end, or the lane host ends the process + if (w.stopped(grace_ms + 2000)) return; + note("[tensorfold] rank {d}: exit 75 (TF_DS_EXIT_ON_FAULT): a supervisor restarts both ranks\n", .{w.rank}); + // a stream may hold kernels that wait on the gone rank (NCCL's), which a normal exit would join + std.c._exit(75); + } +}; + +/// note's descriptor: stderr's file opened again with O_NONBLOCK (openLog); -1: none; -2: not opened yet. +var log_fd: std.atomic.Value(c_int) = .init(-2); + +/// A new open file description of `fd` with O_NONBLOCK, so fd itself (shared with other threads) stays blocking. +fn nonblockingCopy(fd: c_int) c_int { + var buf: [32]u8 = undefined; + const path = std.fmt.bufPrintSentinel(&buf, "/proc/self/fd/{d}", .{fd}, 0) catch return -1; + return std.c.open(path, .{ .ACCMODE = .WRONLY, .NONBLOCK = true, .APPEND = true, .CLOEXEC = true }); +} + +/// Opens note's descriptor once per process (Watch.start), before any fault path can need it. +fn openLog() void { + if (log_fd.load(.acquire) != -2) return; + const fd = nonblockingCopy(2); + if (log_fd.cmpxchgStrong(-2, if (fd >= 0) fd else -1, .acq_rel, .acquire) != null and fd >= 0) _ = std.c.close(fd); +} + +/// A log line to stderr that never blocks: a full pipe or terminal drops it; the abort and exit never wait on it. +fn note(comptime fmt: []const u8, args: anytype) void { + var buf: [512]u8 = undefined; + const line = std.fmt.bufPrint(&buf, fmt, args) catch return; + const fd = log_fd.load(.acquire); + if (fd >= 0) { + _ = std.c.write(fd, line.ptr, line.len); + return; + } + // a socket stderr cannot open through /proc: MSG_DONTWAIT; on any other descriptor send fails and the line drops + _ = std.c.send(2, line.ptr, line.len, std.c.MSG.DONTWAIT | std.c.MSG.NOSIGNAL); +} + +/// TF_DS_EXIT_ON_FAULT=1: the exit 75 grace in ms (TF_DS_EXIT_GRACE s, default 5, as lane_host.faultExit); else null. +pub fn faultGrace() ?u64 { + const on = std.c.getenv("TF_DS_EXIT_ON_FAULT") orelse return null; + if (!std.mem.eql(u8, std.mem.span(on), "1")) return null; + const g = std.c.getenv("TF_DS_EXIT_GRACE") orelse return 5000; + const s = std.fmt.parseFloat(f64, std.mem.span(g)) catch return 5000; + if (!(s >= 0) or s > 600) return 5000; + return @intFromFloat(s * 1000); +} + +const Link = @import("link.zig").Link; + +// libc calls that the tests need and std.c does not declare +extern "c" fn pipe(fds: *[2]c_int) c_int; +extern "c" fn setenv(name: [*:0]const u8, value: [*:0]const u8, overwrite: c_int) c_int; +extern "c" fn unsetenv(name: [*:0]const u8) c_int; + +fn testPort() u16 { + return 29500 + @as(u16, @intCast(@mod(std.Io.Timestamp.now(std.testing.io, .awake).nanoseconds, 400))); +} + +fn sleepMs(ms: u64) void { + const ts: std.c.timespec = .{ .sec = @intCast(ms / 1000), .nsec = @intCast((ms % 1000) * std.time.ns_per_ms) }; + _ = std.c.nanosleep(&ts, null); +} + +/// The test's peer rank: it connects, then reads frames until "done" or its link ends, or closes after `die_ms`. +const Peer = struct { + port: u16, + die_ms: ?u64, + + fn run(p: Peer) void { + var l = Link.connect(std.testing.io, .{ 127, 0, 0, 1 }, p.port, 10) catch return; + defer l.close(); // the process ends: its end closes + if (p.die_ms) |ms| return sleepMs(ms); + var buf: [64]u8 = undefined; + while (true) { + const f = l.recv(&buf) catch return; + if (std.mem.eql(u8, f, "done")) return; + } + } +}; + +test "a start that fails ends the watch with no fault while the peer is up" { + const port = testPort(); + const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = null }}); + var l = try Link.listen(.{ 127, 0, 0, 1 }, port); + const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0, null); + // native.open fails after Lanes.init: end runs before Model.close frees the rings + try std.testing.expectEqual(Outcome.failed, w.end(std.testing.allocator, .failed)); + l.close(); + t.join(); +} + +test "an idle follower whose rank 0 dies: the watch acts, then end finds the peer fault" { + const port = testPort() + 400; + const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 300 }}); + var l = try Link.listen(.{ 127, 0, 0, 1 }, port); + defer l.close(); + const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1, null); + t.join(); + // no receive runs: the watch alone finds the end (and aborts the rings) after its one-second wait + var waited: u32 = 0; + while (!w.lost.load(.acquire) and waited < 40) : (waited += 1) sleepMs(100); + try std.testing.expect(w.lost.load(.acquire)); + try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed)); +} + +test "a sending leader whose follower dies gets PeerDown, and end finds the peer fault" { + const port = testPort() + 800; + const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 0 }}); + var l = try Link.listen(.{ 127, 0, 0, 1 }, port); + defer l.close(); + const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0, null); + t.join(); + // the first send after the close can still go out; a later one fails (no SIGPIPE: link.zig sends with NOSIGNAL) + var err: ?anyerror = null; + for (0..200) |_| { + l.send("a round's frame") catch |e| { + err = e; + break; + }; + sleepMs(10); + } + try std.testing.expectEqual(@as(?anyerror, error.PeerDown), err); + try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed)); +} + +test "a follower that gets PeerDown calls end before the watch acts, and end still finds the peer fault" { + const port = testPort() + 1200; + const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 200 }}); + var l = try Link.listen(.{ 127, 0, 0, 1 }, port); + defer l.close(); + const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1, null); + var buf: [64]u8 = undefined; + try std.testing.expectError(error.PeerDown, l.recv(&buf)); + // within the watch's one-second wait: the watch has not acted, end acts for it + try std.testing.expect(!w.lost.load(.acquire)); + try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed)); + t.join(); +} + +test "a done frame, then the peer's close: end is orderly" { + const port = testPort() + 1600; + const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = null }}); + var l = try Link.listen(.{ 127, 0, 0, 1 }, port); + defer l.close(); + const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0, null); + try std.testing.expect(!w.faulted()); + try l.send("done"); + t.join(); // the peer took done and closed its end: the watch may have seen it + sleepMs(300); + try std.testing.expect(!w.lost.load(.acquire)); + try std.testing.expectEqual(Outcome.orderly, w.end(std.testing.allocator, .done)); +} + +test "a descriptor that is no longer open counts as a link end (POLLNVAL), not as a poll to repeat" { + var fds: [2]c_int = undefined; + try std.testing.expectEqual(@as(c_int, 0), pipe(&fds)); + defer _ = std.c.close(fds[1]); + const w = try Watch.start(std.testing.allocator, &.{fds[0]}, &.{}, 1, null); + _ = std.c.close(fds[0]); + var waited: u32 = 0; + while (!w.lost.load(.acquire) and waited < 40) : (waited += 1) sleepMs(100); + try std.testing.expect(w.lost.load(.acquire)); + try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed)); +} + +test "the exit 75 comes from Watch.start's grace alone: TF_DS_EXIT_ON_FAULT in the test's environment has no effect" { + try std.testing.expectEqual(@as(c_int, 0), setenv("TF_DS_EXIT_ON_FAULT", "1", 1)); + defer _ = unsetenv("TF_DS_EXIT_ON_FAULT"); + try std.testing.expectEqual(@as(?u64, 5000), faultGrace()); + const port = testPort() + 2000; + const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 0 }}); + var l = try Link.listen(.{ 127, 0, 0, 1 }, port); + defer l.close(); + const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1, null); + t.join(); + sleepMs(100); + // a peer fault with no grace: end returns (the test runner goes on) instead of exit 75 + try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed)); +} + +test "note's descriptor never blocks on a full pipe, and the pipe's own descriptor stays blocking" { + var fds: [2]c_int = undefined; + try std.testing.expectEqual(@as(c_int, 0), pipe(&fds)); + defer _ = std.c.close(fds[0]); + defer _ = std.c.close(fds[1]); + const w = nonblockingCopy(fds[1]); + try std.testing.expect(w >= 0); + defer _ = std.c.close(w); + const chunk: [4096]u8 = @splat('x'); + var full = false; + // 4 MiB at most, more than a pipe holds: a write that would wait fails at once (EAGAIN) + for (0..1024) |_| if (std.c.write(w, &chunk, chunk.len) < 0) { + full = true; + break; + }; + try std.testing.expect(full); + const flags: std.c.O = @bitCast(@as(u32, @intCast(std.c.fcntl(fds[1], std.c.F.GETFL)))); + try std.testing.expect(!flags.NONBLOCK); +} diff --git a/zig/src/families/dsv41/rdma.zig b/zig/src/families/dsv41/rdma.zig index 180393f5dd..ce4ece906c 100644 --- a/zig/src/families/dsv41/rdma.zig +++ b/zig/src/families/dsv41/rdma.zig @@ -511,6 +511,12 @@ pub const Ring = struct { } }; +/// True when one of `rings` failed or was aborted: the work that waited on it may hold stale peer data. +pub fn anyFailed(rings: []const *Ring) bool { + for (rings) |r| if (r.failure() != null) return true; + return false; +} + fn postRecv(ctx: *vabi.Context, qp: *vabi.Qp) Error!void { var wr: vabi.RecvWr = .{}; var bad: ?*vabi.RecvWr = null; @@ -555,3 +561,25 @@ test "a failed neighbor stage releases both local stages' device waits" { try std.testing.expectEqual(@as(u64, 1), first.abortWord().*); try std.testing.expectEqual(@as(u64, 1), second.abortWord().*); } + +test "anyFailed finds a ring that the peer watch aborted" { + var rings: [3]Ring = undefined; + var meta0: [32]u64 align(std.heap.page_size_min) = @splat(0); + var meta1: [32]u64 align(std.heap.page_size_min) = @splat(0); + var meta2: [32]u64 align(std.heap.page_size_min) = @splat(0); + var ptrs: [3]*Ring = undefined; + for (&rings, [_][]u64{ &meta0, &meta1, &meta2 }, &ptrs) |*r, meta, *p| { + r.s.slots = 4; + r.world = 2; + r.meta.mem = @alignCast(std.mem.sliceAsBytes(meta)); + r.failed = .init(false); + r.why = @splat(0); + r.failure_peer = null; + p.* = r; + } + try std.testing.expect(!anyFailed(&ptrs)); + try std.testing.expect(!anyFailed(&.{})); + rings[2].abort("a peer rank is gone (its link closed)"); + try std.testing.expect(anyFailed(&ptrs)); + try std.testing.expect(!anyFailed(ptrs[0..2])); +}