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
//! 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)
}