fandhe_ai/distributed/
error.rs

1//! 分散学習（`fandhe_ai::distributed`）の型付きエラー。
2//!
3//! 役割: autodiff 内部の `distributed::DistributedError`／`DistributedStepError` を、内部 enum を
4//! ペイロードに持たない自己完結の [`DistributedError`] へ写して facade の公開面に出す
5//! （`docs/facade-distributed-host-collective-decision.md` §5・§16。`OnnxError` と同じ方針）。
6//! 写像は `pub(crate)` の `from_inner`／`from_step` に置き、`From` impl は公開しない
7//! （内部型が公開 trait impl として現れるのを避ける）。
8//!
9//! `Display` は固定文・数値・内側エラーの `Display` だけで、ピアから受け取ったバイト列は出さない
10//! （OWASP A09）。
11
12use std::fmt;
13use std::net::SocketAddr;
14
15use fandhe_ai_autodiff::distributed as inner;
16
17use crate::BackendError;
18
19/// 分散学習の設定・集合の確立・集合通信・分散ステップが返す型付きエラー。
20#[derive(Debug)]
21#[non_exhaustive]
22pub enum DistributedError {
23    /// 設定値が不正（固定文の理由）。
24    InvalidConfig {
25        /// 固定文の理由。
26        reason: &'static str,
27    },
28    /// world_size が `1..=max` の範囲外。
29    WorldSizeOutOfRange {
30        /// 指定値。
31        world_size: usize,
32        /// 上限。
33        max: usize,
34    },
35    /// ランク番号が `0..world_size` の範囲外。
36    RankOutOfRange {
37        /// 指定値。
38        rank: usize,
39        /// world_size。
40        world_size: usize,
41    },
42    /// 非 loopback のアドレスが opt-in 無しで指定された。
43    NonLoopbackNotAllowed {
44        /// 該当アドレス。
45        addr: SocketAddr,
46    },
47    /// 同じランク番号が重複した。
48    DuplicateRank {
49        /// 重複したランク。
50        rank: usize,
51    },
52    /// ピアの world_size が自分と違う。
53    WorldSizeMismatch {
54        /// 自分の world_size。
55        expected: usize,
56        /// ピアの申告値。
57        got: usize,
58    },
59    /// リングの相手が想定したランクと違う。
60    UnexpectedRank {
61        /// 想定したランク。
62        expected: usize,
63        /// ピアの申告値。
64        got: usize,
65    },
66    /// ワイヤ上のプロトコル違反（固定文と数値だけの説明。ピアのバイト列は含まない）。
67    Protocol {
68        /// 説明文。
69        message: String,
70    },
71    /// タイムアウト。
72    Timeout {
73        /// 固定文の段階名。
74        phase: &'static str,
75    },
76    /// ピアが切断した。
77    PeerDisconnected,
78    /// その他の I/O エラー。
79    Io(std::io::Error),
80    /// communicator が poisoned（以後の集合通信は拒否される。復旧は範囲外）。
81    Poisoned,
82    /// 勾配の読み出しや optimizer ステップが返したバックエンドエラー。
83    Backend(BackendError),
84    /// 平坦化した勾配の長さがストアの総要素数と一致しない。
85    GradientLengthMismatch {
86        /// ストアの総要素数。
87        expected: usize,
88        /// 平坦化後の長さ。
89        got: usize,
90    },
91    /// 将来 autodiff 側に増える variant の受け皿（現行の全 variant はこれ以外へ写る）。
92    Other {
93        /// 説明文。
94        message: String,
95    },
96}
97
98impl DistributedError {
99    /// 内部の集合通信エラーから写す。
100    pub(crate) fn from_inner(e: inner::DistributedError) -> Self {
101        use inner::DistributedError as E;
102        match e {
103            E::InvalidConfig { reason } => Self::InvalidConfig { reason },
104            E::WorldSizeOutOfRange { world_size, max } => {
105                Self::WorldSizeOutOfRange { world_size, max }
106            }
107            E::RankOutOfRange { rank, world_size } => Self::RankOutOfRange { rank, world_size },
108            E::DuplicateRank { rank } => Self::DuplicateRank { rank },
109            E::WorldSizeMismatch { expected, got } => Self::WorldSizeMismatch { expected, got },
110            E::UnexpectedRank { expected, got } => Self::UnexpectedRank { expected, got },
111            E::NonLoopbackNotAllowed { addr } => Self::NonLoopbackNotAllowed { addr },
112            E::Protocol(v) => Self::Protocol {
113                message: v.to_string(),
114            },
115            E::Timeout { phase } => Self::Timeout { phase },
116            E::PeerDisconnected => Self::PeerDisconnected,
117            E::Io(e) => Self::Io(e),
118            E::Poisoned => Self::Poisoned,
119            other => Self::Other {
120                message: other.to_string(),
121            },
122        }
123    }
124
125    /// 内部の分散ステップエラーから写す（`Collective` は `from_inner` を再帰適用する）。
126    pub(crate) fn from_step(e: inner::DistributedStepError) -> Self {
127        use inner::DistributedStepError as S;
128        match e {
129            S::Collective(c) => Self::from_inner(c),
130            S::Backend(b) => Self::Backend(b),
131            S::GradientLengthMismatch { expected, got } => {
132                Self::GradientLengthMismatch { expected, got }
133            }
134            other => Self::Other {
135                message: other.to_string(),
136            },
137        }
138    }
139}
140
141impl fmt::Display for DistributedError {
142    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
143        match self {
144            Self::InvalidConfig { reason } => write!(f, "invalid distributed config: {reason}"),
145            Self::WorldSizeOutOfRange { world_size, max } => {
146                write!(f, "world_size {world_size} is out of range 1..={max}")
147            }
148            Self::RankOutOfRange { rank, world_size } => {
149                write!(f, "rank {rank} is out of range 0..{world_size}")
150            }
151            Self::NonLoopbackNotAllowed { addr } => write!(
152                f,
153                "non-loopback address {addr} requires explicit opt-in (allow_non_loopback)"
154            ),
155            Self::DuplicateRank { rank } => write!(f, "duplicate rank {rank}"),
156            Self::WorldSizeMismatch { expected, got } => {
157                write!(f, "world_size mismatch (expected {expected}, got {got})")
158            }
159            Self::UnexpectedRank { expected, got } => {
160                write!(f, "unexpected peer rank (expected {expected}, got {got})")
161            }
162            Self::Protocol { message } => write!(f, "protocol violation: {message}"),
163            Self::Timeout { phase } => write!(f, "timed out during {phase}"),
164            Self::PeerDisconnected => write!(f, "peer disconnected"),
165            Self::Io(e) => write!(f, "i/o error: {:?}", e.kind()),
166            Self::Poisoned => write!(f, "communicator is poisoned"),
167            Self::Backend(e) => write!(f, "backend error during distributed step: {e}"),
168            Self::GradientLengthMismatch { expected, got } => write!(
169                f,
170                "flattened gradient length mismatch (expected {expected}, got {got})"
171            ),
172            Self::Other { message } => write!(f, "distributed error: {message}"),
173        }
174    }
175}
176
177impl std::error::Error for DistributedError {
178    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
179        match self {
180            Self::Io(e) => Some(e),
181            Self::Backend(e) => Some(e),
182            _ => None,
183        }
184    }
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190    use inner::DistributedError as E;
191    use inner::DistributedStepError as S;
192
193    fn addr() -> SocketAddr {
194        "192.0.2.1:1".parse().unwrap()
195    }
196
197    #[test]
198    fn every_inner_collective_variant_maps_to_a_dedicated_variant() {
199        type Check = fn(&DistributedError) -> bool;
200        let cases: Vec<(E, Check)> = vec![
201            (E::InvalidConfig { reason: "x" }, |d| {
202                matches!(d, DistributedError::InvalidConfig { reason: "x" })
203            }),
204            (
205                E::WorldSizeOutOfRange {
206                    world_size: 0,
207                    max: 64,
208                },
209                |d| {
210                    matches!(
211                        d,
212                        DistributedError::WorldSizeOutOfRange {
213                            world_size: 0,
214                            max: 64
215                        }
216                    )
217                },
218            ),
219            (
220                E::RankOutOfRange {
221                    rank: 2,
222                    world_size: 2,
223                },
224                |d| {
225                    matches!(
226                        d,
227                        DistributedError::RankOutOfRange {
228                            rank: 2,
229                            world_size: 2
230                        }
231                    )
232                },
233            ),
234            (E::DuplicateRank { rank: 1 }, |d| {
235                matches!(d, DistributedError::DuplicateRank { rank: 1 })
236            }),
237            (
238                E::WorldSizeMismatch {
239                    expected: 2,
240                    got: 3,
241                },
242                |d| {
243                    matches!(
244                        d,
245                        DistributedError::WorldSizeMismatch {
246                            expected: 2,
247                            got: 3
248                        }
249                    )
250                },
251            ),
252            (
253                E::UnexpectedRank {
254                    expected: 1,
255                    got: 2,
256                },
257                |d| {
258                    matches!(
259                        d,
260                        DistributedError::UnexpectedRank {
261                            expected: 1,
262                            got: 2
263                        }
264                    )
265                },
266            ),
267            (E::NonLoopbackNotAllowed { addr: addr() }, |d| {
268                matches!(d, DistributedError::NonLoopbackNotAllowed { .. })
269            }),
270            (
271                E::Protocol(inner::ProtocolViolation::BadMagic),
272                |d| matches!(d, DistributedError::Protocol { message } if message == "protocol violation: bad magic" || message == "bad magic"),
273            ),
274            (E::Timeout { phase: "p" }, |d| {
275                matches!(d, DistributedError::Timeout { phase: "p" })
276            }),
277            (E::PeerDisconnected, |d| {
278                matches!(d, DistributedError::PeerDisconnected)
279            }),
280            (E::Io(std::io::Error::other("x")), |d| {
281                matches!(d, DistributedError::Io(_))
282            }),
283            (E::Poisoned, |d| matches!(d, DistributedError::Poisoned)),
284        ];
285        for (inner_err, check) in cases {
286            let label = format!("{inner_err:?}");
287            let mapped = DistributedError::from_inner(inner_err);
288            assert!(check(&mapped), "{label} の写像が想定外: {mapped:?}");
289            assert!(!matches!(mapped, DistributedError::Other { .. }), "{label}");
290        }
291    }
292
293    #[test]
294    fn step_variants_map_and_collective_recurses() {
295        let m = DistributedError::from_step(S::Collective(E::PeerDisconnected));
296        assert!(matches!(m, DistributedError::PeerDisconnected));
297        let m = DistributedError::from_step(S::GradientLengthMismatch {
298            expected: 3,
299            got: 2,
300        });
301        assert!(matches!(
302            m,
303            DistributedError::GradientLengthMismatch {
304                expected: 3,
305                got: 2
306            }
307        ));
308        let m = DistributedError::from_step(S::Backend(BackendError::InvalidArgument("x".into())));
309        assert!(matches!(m, DistributedError::Backend(_)));
310        assert!(std::error::Error::source(&m).is_some());
311    }
312
313    #[test]
314    fn display_has_no_debug_dump_of_io_error() {
315        let e = DistributedError::Io(std::io::Error::other("secret-detail"));
316        assert!(!e.to_string().contains("secret-detail"));
317    }
318}
fandhe_ai/distributed/error.rs

fandhe_ai/distributed/
error.rs
