1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
//! Stall diagnostics for the io_uring reactor: the opt-in
//! `KEVY_DEBUG_STALL_MS` dump that names every connection which can no
//! longer make progress on its own. Split out of [`crate::uring_arm`]
//! so that file stays under the 500-LOC house rule.
use crate::Commands;
use crate::shard::Shard;
use crate::uring_conn::{BigArgState, UringConn};
use kevy_map::KevyMap;
/// Name the big-arg sub-state: which variant, how far the body got, and
/// every flag the cycle waits on. `big_arg=true` alone could not tell a
/// legitimately in-flight read from a wedge — that missing detail cost
/// three rounds of source-reasoning on the deep-pipeline wedge.
fn describe_big_arg(uc: &UringConn) -> String {
match uc.pending_big_arg.as_deref() {
None => String::from("none"),
Some(BigArgState::Frame { frame, total }) => format!("Frame({}/{total})", frame.len()),
Some(BigArgState::BareSetCancelling {
body,
body_len,
crlf_seen,
cancel_acked,
target_canceled,
..
}) => format!(
"Cancelling(body {}/{body_len} crlf={crlf_seen} \
cancel_acked={cancel_acked} target_canceled={target_canceled})",
body.len()
),
Some(BigArgState::BareSetReading {
body,
body_len,
crlf_seen,
..
}) => format!("Reading(body {}/{body_len} crlf={crlf_seen})", body.len()),
}
}
impl<C: Commands> Shard<C> {
/// Print every conn that can no longer make progress on its own —
/// opt-in via `KEVY_DEBUG_STALL_MS=<ms>`, off (one `Option` check on
/// the tick path) otherwise.
///
/// The predicate is "no recv armed and no reason to be visited
/// again": such a conn is invisible to [`Self::uring_arm_conns`],
/// which walks only `arm_pending`, and has no outstanding completion
/// to bring it back. `arm_queued` is reported alongside actual queue
/// membership because a conn whose flag says "already queued" while
/// the queue does not contain it is permanently unreachable —
/// [`Self::mark_arm_pending`] short-circuits on that flag, so every
/// later attempt to wake the conn is a no-op.
///
/// Written for `bench/xshardwedge.sh`, which reproduces exactly that
/// shape. The reactor keeps looping during that wedge (the bounded
/// park wakes on its timeout, which is why threads read 0% CPU rather
/// than spinning) and `CLIENT LIST` — an all-shards fan-out — still
/// answers and still lists the wedged conn, so the shard and its
/// cross-core messaging are fine and the fault is local to one conn.
pub(crate) fn uring_maybe_dump_stalled(
&self,
every: Option<std::time::Duration>,
last: &mut std::time::Instant,
now: std::time::Instant,
io: &KevyMap<u64, UringConn>,
) {
let Some(iv) = every else { return };
if now.duration_since(*last) < iv {
return;
}
*last = now;
// Heartbeat first, unconditionally: without it a silent dump is
// ambiguous between "ran and found nothing" and "never ran", and
// the first capture of this wedge hit exactly that ambiguity.
// The counters are the cross-core ones worth having anyway.
eprintln!(
"kevy: STALLDUMP shard {} conns={} arm_pending={} xshard_inflight={} \
backlog={} dirty={}",
self.id,
self.conns.len(),
self.arm_pending.len(),
self.xshard_inflight,
self.backlog.iter().map(std::collections::VecDeque::len).sum::<usize>(),
self.dirty.len(),
);
for (cid, conn) in self.conns.iter() {
let Some(uc) = io.get(cid) else {
eprintln!("kevy: STALL shard {} conn {cid}: no UringConn entry", self.id);
continue;
};
if uc.recv_armed || uc.write_inflight || uc.closing {
continue;
}
self.dump_stalled_conn(*cid, conn, uc);
}
}
/// One stalled conn's line: every flag its state machine could be
/// waiting on, so the wedge shape is readable without a debugger.
fn dump_stalled_conn(&self, cid: u64, conn: &crate::conn::Conn, uc: &UringConn) {
eprintln!(
"kevy: STALL shard {} conn {cid}: recv_armed=false arm_queued={} \
in_arm_pending={} big_arg={} cancel_pending={} read_pending={} \
rearm_recv={} output={} write_pending={} \
pending_slots={} next_seq={} next_emit={}",
self.id,
uc.arm_queued,
self.arm_pending.contains(&cid),
describe_big_arg(uc),
uc.big_arg_cancel_pending,
uc.big_arg_read_pending,
uc.big_arg_rearm_recv,
!conn.output.is_empty() || !conn.output_arcs.is_empty(),
uc.write_off < uc.write_buf.len() || !uc.write_arcs.is_empty(),
conn.pending.len(),
conn.next_seq,
conn.next_emit,
);
}
}
/// Stall-dump cadence from `KEVY_DEBUG_STALL_MS`; `None` (the default)
/// disables [`Shard::uring_maybe_dump_stalled`] entirely.
pub(crate) fn stall_dump_interval() -> Option<std::time::Duration> {
std::env::var("KEVY_DEBUG_STALL_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|ms| *ms > 0)
.map(std::time::Duration::from_millis)
}