bittensor_core/codec/
batch.rs

1//! Bulk decode entry points — the read-heavy hot loops behind `query_map`
2//! pages and runtime-API batches.
3//!
4//! Decoding is embarrassingly parallel (each item is an independent SCALE
5//! blob against an immutable `Runtime`), so batches above
6//! [`PARALLEL_THRESHOLD`] fan out across rayon workers. Callers on an FFI
7//! boundary release their language's locks around these calls; only
8//! host-object materialization stays on the caller's side.
9
10// Indexing is safe by construction: run_indexed only produces in-bounds
11// indices for the slices captured by each closure.
12#![allow(clippy::indexing_slicing)]
13
14#[cfg(feature = "host")]
15use rayon::prelude::*;
16
17use crate::codec::value::Value;
18use crate::error::CoreError;
19use crate::runtime::type_string::TypeSpec;
20use crate::runtime::{Runtime, StorageInfo};
21
22/// Below this many items the rayon fan-out costs more than it saves.
23pub const PARALLEL_THRESHOLD: usize = 64;
24
25/// One decoded map entry: the free key components and the value.
26pub type DecodedPair = (Vec<Value>, Value);
27
28/// Run `f` over `0..len`, in parallel above the threshold. Without the
29/// `host` feature (e.g. browser WASM, where there are no threads) every
30/// batch runs serially; the API and results are identical.
31#[cfg(feature = "host")]
32fn run_indexed<T: Send>(
33    len: usize,
34    f: impl Fn(usize) -> Result<T, CoreError> + Sync,
35) -> Result<Vec<T>, CoreError> {
36    if len >= PARALLEL_THRESHOLD {
37        (0..len).into_par_iter().map(&f).collect()
38    } else {
39        (0..len).map(f).collect()
40    }
41}
42
43#[cfg(not(feature = "host"))]
44fn run_indexed<T: Send>(
45    len: usize,
46    f: impl Fn(usize) -> Result<T, CoreError> + Sync,
47) -> Result<Vec<T>, CoreError> {
48    (0..len).map(f).collect()
49}
50
51impl Runtime {
52    /// Decode a batch of `(type string, SCALE bytes)` pairs. Consecutive
53    /// duplicate type strings (the common case: one storage item's page)
54    /// resolve to a spec once.
55    pub fn decode_batch(
56        &self,
57        type_strings: &[String],
58        datas: &[Vec<u8>],
59    ) -> Result<Vec<Value>, CoreError> {
60        if type_strings.len() != datas.len() {
61            return Err(CoreError::Codec(
62                "type_strings and datas must have the same length".into(),
63            ));
64        }
65        let mut specs: Vec<TypeSpec> = Vec::with_capacity(type_strings.len());
66        let mut last: Option<(&str, TypeSpec)> = None;
67        for type_string in type_strings {
68            let spec = match &last {
69                Some((s, spec)) if *s == type_string.as_str() => spec.clone(),
70                _ => {
71                    let spec = self.type_spec(type_string)?;
72                    last = Some((type_string.as_str(), spec.clone()));
73                    spec
74                }
75            };
76            specs.push(spec);
77        }
78        run_indexed(datas.len(), |i| {
79            self.decode_spec(&specs[i], &datas[i], true)
80        })
81    }
82
83    /// Decode one page of a storage map: recover the free key components
84    /// from each full storage key (`fixed` leading params were part of the
85    /// queried prefix) and decode each value.
86    pub fn decode_map_page(
87        &self,
88        entry: &StorageInfo,
89        raw_keys: &[Vec<u8>],
90        raw_values: &[Vec<u8>],
91        fixed: usize,
92    ) -> Result<Vec<DecodedPair>, CoreError> {
93        if raw_keys.len() != raw_values.len() {
94            return Err(CoreError::Codec(
95                "raw_keys and raw_values must have the same length".into(),
96            ));
97        }
98        let value_spec = TypeSpec::Id(entry.value_type);
99        run_indexed(raw_keys.len(), |i| {
100            let params = self.decode_storage_key_params(entry, &raw_keys[i], fixed)?;
101            let value = self.decode_spec(&value_spec, &raw_values[i], true)?;
102            Ok((params, value))
103        })
104    }
105
106    /// Like [`Runtime::decode_map_page`], but takes the raw
107    /// `state_queryStorageAt` change tuples (`0x`-hex key/value strings;
108    /// `None` values — keys deleted between the key listing and the value
109    /// fetch — are skipped), so hex parsing also runs in the parallel
110    /// section.
111    pub fn decode_map_changes(
112        &self,
113        entry: &StorageInfo,
114        changes: &[(String, Option<String>)],
115        fixed: usize,
116    ) -> Result<Vec<DecodedPair>, CoreError> {
117        let value_spec = TypeSpec::Id(entry.value_type);
118        let unhex = |s: &str| {
119            hex::decode(s.trim_start_matches("0x"))
120                .map_err(|e| CoreError::Codec(format!("bad hex in changes: {e}")))
121        };
122        let present: Vec<(&str, &str)> = changes
123            .iter()
124            .filter_map(|(k, v)| v.as_deref().map(|v| (k.as_str(), v)))
125            .collect();
126        run_indexed(present.len(), |i| {
127            let (key_hex, value_hex) = present[i];
128            let raw_key = unhex(key_hex)?;
129            let raw_value = unhex(value_hex)?;
130            let params = self.decode_storage_key_params(entry, &raw_key, fixed)?;
131            let value = self.decode_spec(&value_spec, &raw_value, true)?;
132            Ok((params, value))
133        })
134    }
135}
bittensor_core/codec/batch.rs

bittensor_core/codec/
batch.rs
