skippy_protocol/binary/
codec.rs

1use std::io::{self, Read, Write};
2
3use super::{
4    MAX_STAGE_ACTIVATION_BYTES, MAX_STAGE_CHAT_SAMPLING_METADATA_BYTES,
5    MAX_STAGE_DECODED_ACTIVATION_BYTES, MAX_STAGE_LOGIT_BIAS, MAX_STAGE_PREDICTED_TOKENS,
6    MAX_STAGE_SIDEBAND_VALUES, MAX_STAGE_STATE_IMPORT_BYTES, READY_MAGIC, STAGE_STATE_VERSION,
7    StageLogitBias, StageNativeMtpDraft, StageReply, StageReplyStats, StageReplyWindow,
8    StageSamplingConfig, StageStateHeader, StageWireMessage, WireActivationDType, WireMessageKind,
9    WireReplyKind,
10    activation::{
11        activation_decoded_f32_bytes_with_state_flags, activation_wire_bytes_with_state_flags,
12    },
13    invalid_data, invalid_input,
14};
15
16pub fn send_ready(mut writer: impl Write) -> io::Result<()> {
17    write_i32(&mut writer, READY_MAGIC)
18}
19
20pub fn recv_ready(mut reader: impl Read) -> io::Result<()> {
21    let magic = read_i32(&mut reader)?;
22    if magic != READY_MAGIC {
23        return Err(invalid_data("stage ready magic mismatch"));
24    }
25    Ok(())
26}
27
28pub fn send_reply_ack(mut writer: impl Write) -> io::Result<()> {
29    send_reply_ack_with_stats(&mut writer, StageReplyStats::default())
30}
31
32pub fn send_reply_ack_with_stats(mut writer: impl Write, stats: StageReplyStats) -> io::Result<()> {
33    send_reply_message(
34        &mut writer,
35        &StageReply {
36            kind: WireReplyKind::Ack,
37            predicted: 0,
38            predicted_tokens: Vec::new(),
39            native_mtp_draft: None,
40            window: StageReplyWindow::default(),
41            stats,
42        },
43    )
44}
45
46pub fn send_reply_predicted(mut writer: impl Write, predicted: i32) -> io::Result<()> {
47    send_reply_predicted_with_stats(&mut writer, predicted, StageReplyStats::default())
48}
49
50pub fn send_reply_predicted_with_stats(
51    mut writer: impl Write,
52    predicted: i32,
53    stats: StageReplyStats,
54) -> io::Result<()> {
55    send_reply_predicted_with_tokens_window_and_stats(
56        &mut writer,
57        predicted,
58        &[predicted],
59        StageReplyWindow::default(),
60        stats,
61    )
62}
63
64pub fn send_reply_predicted_with_tokens_and_stats(
65    mut writer: impl Write,
66    predicted: i32,
67    predicted_tokens: &[i32],
68    stats: StageReplyStats,
69) -> io::Result<()> {
70    send_reply_predicted_with_tokens_window_and_stats(
71        &mut writer,
72        predicted,
73        predicted_tokens,
74        StageReplyWindow::default(),
75        stats,
76    )
77}
78
79pub fn send_reply_predicted_with_tokens_window_and_stats(
80    mut writer: impl Write,
81    predicted: i32,
82    predicted_tokens: &[i32],
83    window: StageReplyWindow,
84    stats: StageReplyStats,
85) -> io::Result<()> {
86    if predicted_tokens.len() > MAX_STAGE_PREDICTED_TOKENS {
87        return Err(invalid_input("too many predicted tokens"));
88    }
89    write_reply_header(
90        &mut writer,
91        WireReplyKind::PredictedToken,
92        predicted,
93        predicted_tokens,
94        window,
95    )?;
96    write_native_mtp_draft(&mut writer, None)?;
97    write_reply_stats(&mut writer, stats)
98}
99
100pub fn send_reply_predicted_tokens_with_stats(
101    mut writer: impl Write,
102    predicted_tokens: &[i32],
103    stats: StageReplyStats,
104) -> io::Result<()> {
105    send_reply_predicted_tokens_with_window_and_stats(
106        &mut writer,
107        predicted_tokens,
108        StageReplyWindow::default(),
109        stats,
110    )
111}
112
113pub fn send_reply_predicted_tokens_with_window_and_stats(
114    mut writer: impl Write,
115    predicted_tokens: &[i32],
116    window: StageReplyWindow,
117    stats: StageReplyStats,
118) -> io::Result<()> {
119    if predicted_tokens.len() > MAX_STAGE_PREDICTED_TOKENS {
120        return Err(invalid_input("too many predicted tokens"));
121    }
122    let predicted = predicted_tokens.first().copied().unwrap_or(0);
123    write_reply_header(
124        &mut writer,
125        WireReplyKind::PredictedTokens,
126        predicted,
127        predicted_tokens,
128        window,
129    )?;
130    write_native_mtp_draft(&mut writer, None)?;
131    write_reply_stats(&mut writer, stats)
132}
133
134pub fn send_reply_message(mut writer: impl Write, reply: &StageReply) -> io::Result<()> {
135    if reply.predicted_tokens.len() > MAX_STAGE_PREDICTED_TOKENS {
136        return Err(invalid_input("too many predicted tokens"));
137    }
138    write_reply_header(
139        &mut writer,
140        reply.kind,
141        reply.predicted,
142        &reply.predicted_tokens,
143        reply.window,
144    )?;
145    write_native_mtp_draft(&mut writer, reply.native_mtp_draft.as_ref())?;
146    write_reply_stats(&mut writer, reply.stats)
147}
148
149pub fn recv_reply(mut reader: impl Read) -> io::Result<StageReply> {
150    let kind = WireReplyKind::try_from(read_i32(&mut reader)?)?;
151    let predicted = read_i32(&mut reader)?;
152    let predicted_count = checked_i32_len(
153        read_i32(&mut reader)?,
154        MAX_STAGE_PREDICTED_TOKENS,
155        "negative predicted token count",
156        "predicted token count exceeds maximum",
157    )?;
158    let mut predicted_tokens = Vec::with_capacity(predicted_count);
159    for _ in 0..predicted_count {
160        predicted_tokens.push(read_i32(&mut reader)?);
161    }
162    let window = read_reply_window(&mut reader)?;
163    let native_mtp_draft = read_native_mtp_draft(&mut reader)?;
164    let stats = read_reply_stats(&mut reader)?;
165    Ok(StageReply {
166        kind,
167        predicted,
168        predicted_tokens,
169        native_mtp_draft,
170        window,
171        stats,
172    })
173}
174
175fn write_reply_header(
176    mut writer: impl Write,
177    kind: WireReplyKind,
178    predicted: i32,
179    predicted_tokens: &[i32],
180    window: StageReplyWindow,
181) -> io::Result<()> {
182    write_i32(&mut writer, kind as i32)?;
183    write_i32(&mut writer, predicted)?;
184    write_i32(
185        &mut writer,
186        i32::try_from(predicted_tokens.len())
187            .map_err(|_| invalid_input("too many predicted tokens"))?,
188    )?;
189    for token in predicted_tokens {
190        write_i32(&mut writer, *token)?;
191    }
192    write_reply_window(&mut writer, window)
193}
194
195fn write_native_mtp_draft(
196    mut writer: impl Write,
197    draft: Option<&StageNativeMtpDraft>,
198) -> io::Result<()> {
199    let Some(draft) = draft else {
200        return write_i32(&mut writer, 0);
201    };
202    if draft.token_ids.len() > MAX_STAGE_PREDICTED_TOKENS {
203        return Err(invalid_input("too many native MTP draft tokens"));
204    }
205    write_i32(&mut writer, 1)?;
206    write_i32(
207        &mut writer,
208        i32::try_from(draft.token_ids.len())
209            .map_err(|_| invalid_input("too many native MTP draft tokens"))?,
210    )?;
211    for token in &draft.token_ids {
212        write_i32(&mut writer, *token)?;
213    }
214    write_i64(&mut writer, draft.proposal_compute_us)
215}
216
217fn read_native_mtp_draft(mut reader: impl Read) -> io::Result<Option<StageNativeMtpDraft>> {
218    match read_i32(&mut reader)? {
219        0 => Ok(None),
220        1 => {
221            let token_count = checked_i32_len(
222                read_i32(&mut reader)?,
223                MAX_STAGE_PREDICTED_TOKENS,
224                "negative native MTP draft token count",
225                "native MTP draft token count exceeds maximum",
226            )?;
227            let mut token_ids = Vec::with_capacity(token_count);
228            for _ in 0..token_count {
229                token_ids.push(read_i32(&mut reader)?);
230            }
231            Ok(Some(StageNativeMtpDraft {
232                token_ids,
233                proposal_compute_us: read_i64(&mut reader)?,
234            }))
235        }
236        _ => Err(invalid_data("unknown native MTP draft reply marker")),
237    }
238}
239
240pub fn write_stage_message(
241    mut writer: impl Write,
242    message: &StageWireMessage,
243    dtype: WireActivationDType,
244) -> io::Result<()> {
245    // Wire v4 fixed prefix, little-endian:
246    // kind, pos_start, token_count, token_sideband_count, position_sideband_count (5 x i32);
247    // StageStateHeader (10 x i32); request_id, session_id (2 x u64);
248    // optional StageSamplingConfig follows when state_flags::SAMPLING is set.
249    // Token sideband, raw StateImport bytes, or activation bytes follow this
250    // prefix, so prefill overhead stays independent of ID string length.
251    write_i32(&mut writer, message.kind as i32)?;
252    write_i32(&mut writer, message.pos_start)?;
253    write_i32(&mut writer, message.token_count)?;
254    if message.tokens.len() > MAX_STAGE_SIDEBAND_VALUES {
255        return Err(invalid_input("too many tokens"));
256    }
257    write_i32(
258        &mut writer,
259        i32::try_from(message.tokens.len()).map_err(|_| invalid_input("too many tokens"))?,
260    )?;
261    if message.positions.len() > MAX_STAGE_SIDEBAND_VALUES {
262        return Err(invalid_input("too many position sideband values"));
263    }
264    write_i32(
265        &mut writer,
266        i32::try_from(message.positions.len())
267            .map_err(|_| invalid_input("too many position sideband values"))?,
268    )?;
269
270    let mut state = message.state;
271    state.reserved = dtype as i32;
272    if message.sampling.is_some() {
273        state.flags |= super::state_flags::SAMPLING;
274    } else {
275        state.flags &= !super::state_flags::SAMPLING;
276    }
277    if message.chat_sampling_metadata.is_some() {
278        state.flags |= super::state_flags::CHAT_SAMPLING_METADATA;
279    } else {
280        state.flags &= !super::state_flags::CHAT_SAMPLING_METADATA;
281    }
282    write_state_header(&mut writer, state)?;
283    write_u64(&mut writer, message.request_id)?;
284    write_u64(&mut writer, message.session_id)?;
285    if let Some(sampling) = message.sampling.as_ref() {
286        write_sampling_config(&mut writer, sampling)?;
287    }
288    if let Some(metadata) = message.chat_sampling_metadata.as_ref() {
289        let bytes = metadata.as_bytes();
290        if bytes.len() > MAX_STAGE_CHAT_SAMPLING_METADATA_BYTES {
291            return Err(invalid_input("chat sampling metadata is too large"));
292        }
293        write_u32(
294            &mut writer,
295            u32::try_from(bytes.len())
296                .map_err(|_| invalid_input("chat sampling metadata is too large"))?,
297        )?;
298        writer.write_all(bytes)?;
299    }
300
301    if message.kind == WireMessageKind::StateImport {
302        let raw_byte_count = usize::try_from(message.token_count)
303            .map_err(|_| invalid_input("state import raw byte count mismatch"))?;
304        if raw_byte_count != message.raw_bytes.len() {
305            return Err(invalid_input("state import raw byte count mismatch"));
306        }
307        if raw_byte_count > MAX_STAGE_STATE_IMPORT_BYTES {
308            return Err(invalid_input("state import raw byte count exceeds maximum"));
309        }
310        writer.write_all(&message.raw_bytes)?;
311        return Ok(());
312    }
313    for token in &message.tokens {
314        write_i32(&mut writer, *token)?;
315    }
316    for position in &message.positions {
317        write_i32(&mut writer, *position)?;
318    }
319    writer.write_all(&message.activation)?;
320    Ok(())
321}
322
323pub fn read_stage_message(mut reader: impl Read, n_embd: i32) -> io::Result<StageWireMessage> {
324    let kind = WireMessageKind::try_from(read_i32(&mut reader)?)?;
325    let pos_start = read_i32(&mut reader)?;
326    let token_count = read_i32(&mut reader)?;
327    let token_sideband_count = read_i32(&mut reader)?;
328    let position_sideband_count = read_i32(&mut reader)?;
329    let state = read_state_header(&mut reader)?;
330    if state.version != STAGE_STATE_VERSION {
331        return Err(invalid_data("unsupported stage state version"));
332    }
333    let request_id = read_u64(&mut reader)?;
334    let session_id = read_u64(&mut reader)?;
335    let sampling = if (state.flags & super::state_flags::SAMPLING) != 0 {
336        Some(read_sampling_config(&mut reader)?)
337    } else {
338        None
339    };
340    let chat_sampling_metadata = if (state.flags & super::state_flags::CHAT_SAMPLING_METADATA) != 0
341    {
342        let len = checked_u32_len(
343            read_u32(&mut reader)?,
344            MAX_STAGE_CHAT_SAMPLING_METADATA_BYTES,
345            "chat sampling metadata length exceeds maximum",
346        )?;
347        let mut bytes = vec![0_u8; len];
348        reader.read_exact(&mut bytes)?;
349        Some(
350            String::from_utf8(bytes)
351                .map_err(|_| invalid_data("chat sampling metadata is not UTF-8"))?,
352        )
353    } else {
354        None
355    };
356    let dtype = state.dtype()?;
357    if kind == WireMessageKind::Stop {
358        return Ok(StageWireMessage {
359            kind,
360            pos_start,
361            token_count,
362            state,
363            request_id,
364            session_id,
365            sampling,
366            chat_sampling_metadata,
367            tokens: Vec::new(),
368            positions: Vec::new(),
369            activation: Vec::new(),
370            raw_bytes: Vec::new(),
371        });
372    }
373    if token_count < 0 || token_sideband_count < 0 || position_sideband_count < 0 {
374        return Err(invalid_data("negative wire count"));
375    }
376    let token_sideband_count = checked_i32_len(
377        token_sideband_count,
378        MAX_STAGE_SIDEBAND_VALUES,
379        "negative wire count",
380        "token sideband count exceeds maximum",
381    )?;
382    let position_sideband_count = checked_i32_len(
383        position_sideband_count,
384        MAX_STAGE_SIDEBAND_VALUES,
385        "negative wire count",
386        "position sideband count exceeds maximum",
387    )?;
388    if kind == WireMessageKind::StateImport {
389        let raw_byte_count = checked_i32_len(
390            token_count,
391            MAX_STAGE_STATE_IMPORT_BYTES,
392            "negative wire count",
393            "state import byte count exceeds maximum",
394        )?;
395        let mut raw_bytes = vec![0; raw_byte_count];
396        reader.read_exact(&mut raw_bytes)?;
397        return Ok(StageWireMessage {
398            kind,
399            pos_start,
400            token_count,
401            state,
402            request_id,
403            session_id,
404            sampling,
405            chat_sampling_metadata,
406            tokens: Vec::new(),
407            positions: Vec::new(),
408            activation: Vec::new(),
409            raw_bytes,
410        });
411    }
412
413    let mut tokens = Vec::with_capacity(token_sideband_count);
414    for _ in 0..token_sideband_count {
415        tokens.push(read_i32(&mut reader)?);
416    }
417    let mut positions = Vec::with_capacity(position_sideband_count);
418    for _ in 0..position_sideband_count {
419        positions.push(read_i32(&mut reader)?);
420    }
421    let activation_bytes =
422        if state.source_stage_index < 0 || kind.is_activationless_prefix_cache_control() {
423            0
424        } else {
425            activation_wire_bytes_with_state_flags(dtype, token_count, n_embd, state.flags)?
426        };
427    if activation_bytes > MAX_STAGE_ACTIVATION_BYTES {
428        return Err(invalid_data(
429            "activation payload byte count exceeds maximum",
430        ));
431    }
432    if activation_bytes > 0 {
433        let decoded_activation_bytes =
434            activation_decoded_f32_bytes_with_state_flags(token_count, n_embd, state.flags)?;
435        if decoded_activation_bytes > MAX_STAGE_DECODED_ACTIVATION_BYTES {
436            return Err(invalid_data(
437                "decoded activation payload byte count exceeds maximum",
438            ));
439        }
440    }
441    let mut activation = vec![0; activation_bytes];
442    if activation_bytes > 0 {
443        reader.read_exact(&mut activation)?;
444    }
445    Ok(StageWireMessage {
446        kind,
447        pos_start,
448        token_count,
449        state,
450        request_id,
451        session_id,
452        sampling,
453        chat_sampling_metadata,
454        tokens,
455        positions,
456        activation,
457        raw_bytes: Vec::new(),
458    })
459}
460
461fn checked_i32_len(
462    value: i32,
463    max: usize,
464    negative_message: &'static str,
465    too_large_message: &'static str,
466) -> io::Result<usize> {
467    if value < 0 {
468        return Err(invalid_data(negative_message));
469    }
470    let value = usize::try_from(value).map_err(|_| invalid_data(too_large_message))?;
471    if value > max {
472        return Err(invalid_data(too_large_message));
473    }
474    Ok(value)
475}
476
477fn checked_u32_len(value: u32, max: usize, too_large_message: &'static str) -> io::Result<usize> {
478    let value = usize::try_from(value).map_err(|_| invalid_data(too_large_message))?;
479    if value > max {
480        return Err(invalid_data(too_large_message));
481    }
482    Ok(value)
483}
484
485fn write_state_header(mut writer: impl Write, state: StageStateHeader) -> io::Result<()> {
486    write_i32(&mut writer, state.version)?;
487    write_i32(&mut writer, state.seq_id)?;
488    write_i32(&mut writer, state.phase)?;
489    write_i32(&mut writer, state.flags)?;
490    write_i32(&mut writer, state.checkpoint_generation)?;
491    write_i32(&mut writer, state.prompt_token_count)?;
492    write_i32(&mut writer, state.decode_step)?;
493    write_i32(&mut writer, state.current_token)?;
494    write_i32(&mut writer, state.source_stage_index)?;
495    write_i32(&mut writer, state.reserved)
496}
497
498fn read_state_header(mut reader: impl Read) -> io::Result<StageStateHeader> {
499    Ok(StageStateHeader {
500        version: read_i32(&mut reader)?,
501        seq_id: read_i32(&mut reader)?,
502        phase: read_i32(&mut reader)?,
503        flags: read_i32(&mut reader)?,
504        checkpoint_generation: read_i32(&mut reader)?,
505        prompt_token_count: read_i32(&mut reader)?,
506        decode_step: read_i32(&mut reader)?,
507        current_token: read_i32(&mut reader)?,
508        source_stage_index: read_i32(&mut reader)?,
509        reserved: read_i32(&mut reader)?,
510    })
511}
512
513fn write_sampling_config(mut writer: impl Write, sampling: &StageSamplingConfig) -> io::Result<()> {
514    write_u32(&mut writer, sampling.flags)?;
515    write_u32(&mut writer, sampling.seed)?;
516    write_f32(&mut writer, sampling.temperature)?;
517    write_f32(&mut writer, sampling.top_p)?;
518    write_i32(&mut writer, sampling.top_k)?;
519    write_f32(&mut writer, sampling.min_p)?;
520    write_f32(&mut writer, sampling.presence_penalty)?;
521    write_f32(&mut writer, sampling.frequency_penalty)?;
522    write_f32(&mut writer, sampling.repeat_penalty)?;
523    write_i32(&mut writer, sampling.penalty_last_n)?;
524    let count = sampling.logit_bias.len().min(MAX_STAGE_LOGIT_BIAS);
525    write_u32(&mut writer, count as u32)?;
526    for bias in sampling.logit_bias.iter().take(count) {
527        write_i32(&mut writer, bias.token_id)?;
528        write_f32(&mut writer, bias.bias)?;
529    }
530    Ok(())
531}
532
533fn read_sampling_config(mut reader: impl Read) -> io::Result<StageSamplingConfig> {
534    let mut sampling = StageSamplingConfig {
535        flags: read_u32(&mut reader)?,
536        seed: read_u32(&mut reader)?,
537        temperature: read_f32(&mut reader)?,
538        top_p: read_f32(&mut reader)?,
539        top_k: read_i32(&mut reader)?,
540        min_p: read_f32(&mut reader)?,
541        presence_penalty: read_f32(&mut reader)?,
542        frequency_penalty: read_f32(&mut reader)?,
543        repeat_penalty: read_f32(&mut reader)?,
544        penalty_last_n: read_i32(&mut reader)?,
545        logit_bias: Vec::new(),
546    };
547    let logit_bias_count = usize::try_from(read_u32(&mut reader)?)
548        .map_err(|_| invalid_data("logit bias count overflows usize"))?;
549    if logit_bias_count > MAX_STAGE_LOGIT_BIAS {
550        return Err(invalid_data("logit bias count exceeds maximum"));
551    }
552    sampling.logit_bias.reserve(logit_bias_count);
553    for _ in 0..logit_bias_count {
554        sampling.logit_bias.push(StageLogitBias {
555            token_id: read_i32(&mut reader)?,
556            bias: read_f32(&mut reader)?,
557        });
558    }
559    Ok(sampling)
560}
561
562const REPLY_STATS_FIELD_COUNT: usize = 23;
563const REPLY_STATS_WIRE_BYTES: usize = REPLY_STATS_FIELD_COUNT * std::mem::size_of::<i64>();
564
565fn write_reply_stats(mut writer: impl Write, stats: StageReplyStats) -> io::Result<()> {
566    let fields = reply_stats_fields(stats);
567    let mut bytes = [0_u8; REPLY_STATS_WIRE_BYTES];
568    for (chunk, value) in bytes
569        .chunks_exact_mut(std::mem::size_of::<i64>())
570        .zip(fields)
571    {
572        chunk.copy_from_slice(&value.to_le_bytes());
573    }
574    writer.write_all(&bytes)
575}
576
577fn read_reply_stats(mut reader: impl Read) -> io::Result<StageReplyStats> {
578    let mut bytes = [0_u8; REPLY_STATS_WIRE_BYTES];
579    reader.read_exact(&mut bytes)?;
580    let mut fields = [0_i64; REPLY_STATS_FIELD_COUNT];
581    for (field, chunk) in fields
582        .iter_mut()
583        .zip(bytes.chunks_exact(std::mem::size_of::<i64>()))
584    {
585        *field = i64::from_le_bytes(chunk.try_into().expect("i64 chunk size"));
586    }
587    Ok(reply_stats_from_fields(fields))
588}
589
590fn write_reply_window(mut writer: impl Write, window: StageReplyWindow) -> io::Result<()> {
591    write_i32(&mut writer, window.window_id)
592}
593
594fn read_reply_window(mut reader: impl Read) -> io::Result<StageReplyWindow> {
595    Ok(StageReplyWindow {
596        window_id: read_i32(&mut reader)?,
597    })
598}
599
600fn reply_stats_fields(stats: StageReplyStats) -> [i64; REPLY_STATS_FIELD_COUNT] {
601    [
602        stats.kv_lookup_hits,
603        stats.kv_lookup_misses,
604        stats.kv_lookup_errors,
605        stats.kv_imported_pages,
606        stats.kv_imported_tokens,
607        stats.kv_recorded_pages,
608        stats.kv_recorded_bytes,
609        stats.kv_hit_stage_mask,
610        stats.kv_record_stage_mask,
611        stats.verify_window_compute_us,
612        stats.verify_window_forward_write_us,
613        stats.verify_window_downstream_wait_us,
614        stats.verify_window_total_us,
615        stats.verify_window_stage_count,
616        stats.verify_window_request_count,
617        stats.verify_window_token_count,
618        stats.verify_window_max_tokens,
619        stats.prefill_edge_write_us_max,
620        stats.prefill_edge_wait_us_max,
621        stats.prefill_edge_total_us_max,
622        stats.prefill_edge_stage_index,
623        stats.prefill_edge_activation_bytes_max,
624        stats.prefill_edge_observation_count,
625    ]
626}
627
628fn reply_stats_from_fields(fields: [i64; REPLY_STATS_FIELD_COUNT]) -> StageReplyStats {
629    StageReplyStats {
630        kv_lookup_hits: fields[0],
631        kv_lookup_misses: fields[1],
632        kv_lookup_errors: fields[2],
633        kv_imported_pages: fields[3],
634        kv_imported_tokens: fields[4],
635        kv_recorded_pages: fields[5],
636        kv_recorded_bytes: fields[6],
637        kv_hit_stage_mask: fields[7],
638        kv_record_stage_mask: fields[8],
639        verify_window_compute_us: fields[9],
640        verify_window_forward_write_us: fields[10],
641        verify_window_downstream_wait_us: fields[11],
642        verify_window_total_us: fields[12],
643        verify_window_stage_count: fields[13],
644        verify_window_request_count: fields[14],
645        verify_window_token_count: fields[15],
646        verify_window_max_tokens: fields[16],
647        prefill_edge_write_us_max: fields[17],
648        prefill_edge_wait_us_max: fields[18],
649        prefill_edge_total_us_max: fields[19],
650        prefill_edge_stage_index: fields[20],
651        prefill_edge_activation_bytes_max: fields[21],
652        prefill_edge_observation_count: fields[22],
653    }
654}
655
656fn read_i32(mut reader: impl Read) -> io::Result<i32> {
657    let mut bytes = [0_u8; 4];
658    reader.read_exact(&mut bytes)?;
659    Ok(i32::from_le_bytes(bytes))
660}
661
662fn write_i32(mut writer: impl Write, value: i32) -> io::Result<()> {
663    writer.write_all(&value.to_le_bytes())
664}
665
666fn read_i64(mut reader: impl Read) -> io::Result<i64> {
667    let mut bytes = [0_u8; 8];
668    reader.read_exact(&mut bytes)?;
669    Ok(i64::from_le_bytes(bytes))
670}
671
672fn write_i64(mut writer: impl Write, value: i64) -> io::Result<()> {
673    writer.write_all(&value.to_le_bytes())
674}
675
676fn read_u32(mut reader: impl Read) -> io::Result<u32> {
677    let mut bytes = [0_u8; 4];
678    reader.read_exact(&mut bytes)?;
679    Ok(u32::from_le_bytes(bytes))
680}
681
682fn write_u32(mut writer: impl Write, value: u32) -> io::Result<()> {
683    writer.write_all(&value.to_le_bytes())
684}
685
686fn read_f32(mut reader: impl Read) -> io::Result<f32> {
687    let mut bytes = [0_u8; 4];
688    reader.read_exact(&mut bytes)?;
689    Ok(f32::from_le_bytes(bytes))
690}
691
692fn write_f32(mut writer: impl Write, value: f32) -> io::Result<()> {
693    writer.write_all(&value.to_le_bytes())
694}
695
696fn read_u64(mut reader: impl Read) -> io::Result<u64> {
697    let mut bytes = [0_u8; 8];
698    reader.read_exact(&mut bytes)?;
699    Ok(u64::from_le_bytes(bytes))
700}
701
702fn write_u64(mut writer: impl Write, value: u64) -> io::Result<()> {
703    writer.write_all(&value.to_le_bytes())
704}
skippy_protocol/binary/codec.rs

skippy_protocol/binary/
codec.rs
