Struct InMemoryBackend 

pub struct InMemoryBackend { /* private fields */ }

Real in-process backend. Holds counters in a HashMap<(key, window), u64> guarded by a Mutex. Garbage-collects expired windows opportunistically on each incr call.

Implementations

impl InMemoryBackend

pub fn new() -> Self

Examples found in repository

examples/sample_app.rs (line 277)

fn per_account_quota() {
    use std::sync::Arc;

    use subms_rate_limiter::{DistributedLimiter, InMemoryBackend, TestClock};

    println!("\n== distributed-backend: one account quota, two routers ==");
    let clock = Arc::new(TestClock::new());
    let shared = Arc::new(InMemoryBackend::new());
    let window_ns = 1_000_000_000u64; // 1s window, 5 orders per account

    let router_a = DistributedLimiter::with_clock(
        Box::new(SharedBackend(shared.clone())),
        5,
        window_ns,
        Box::new(SharedClock(clock.clone())),
    );
    let router_b = DistributedLimiter::with_clock(
        Box::new(SharedBackend(shared.clone())),
        5,
        window_ns,
        Box::new(SharedClock(clock.clone())),
    );

    let account = "acct-42";
    let mut admitted = 0usize;
    for round in 0..8 {
        let router = if round % 2 == 0 { &router_a } else { &router_b };
        if router.try_acquire(account) {
            admitted += 1;
        }
    }
    println!("  8 orders sprayed across 2 routers: {admitted} admitted (quota 5)");
    assert_eq!(admitted, 5, "the shared quota holds across both routers");
}

examples/perf_features.rs (line 392)

fn main() -> io::Result<()> {
    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("..")
        .join(".subms")
        .join("features")
        .join("rust.json");
    let existing = std::fs::read_to_string(&path).unwrap_or_default();
    let mut manifest = SubMsFeatureManifest::load_str("rust", &existing);
    // Stamp the box these numbers came from. The bench runs wherever it is
    // invoked, so an unstamped manifest is indistinguishable from a fleet
    // capture; the renderer will not publish one it cannot attribute.
    let (source, instance) = SubMsP99Source::from_env();
    manifest.set_p99_source(source, instance.as_deref());

    // The baseline: a plain `try_acquire` on the base GCRA limiter, cycled
    // over a fleet the same size as the canonical sweep point so it pays the
    // same cache footprint the features do. Every call is granted; GCRA's
    // reject path returns before the CAS, so the grant is the dearer branch
    // and the conservative baseline.
    let base = measure(
        || {
            (0..CANON_N)
                .map(|_| RateLimiter::new(BASE_RATE, BASE_BURST))
                .collect::<Vec<_>>()
        },
        |v, i| v[i % v.len()].try_acquire(),
        OPS,
        BATCH,
    );
    let base_p50 = base.p50;
    eprintln!(
        "base try_acquire over {CANON_N} limiters: p50={base_p50}ns p99={}ns accept={:.0}%",
        base.p99,
        base.accept * 100.0
    );
    // One hammered limiter, for context only. It is not the classifier's base:
    // comparing a fleet-cycling feature against a single hot limiter would
    // charge the feature for the cache misses the workload shape causes.
    let hot = measure(
        || RateLimiter::new(BASE_RATE, (2 * OPS) as u64),
        |r, _| r.try_acquire(),
        OPS,
        BATCH,
    );
    eprintln!(
        "base try_acquire on 1 hot limiter: p50={}ns p99={}ns (context only)",
        hot.p50, hot.p99
    );

    // ---------- token-bucket: mutex-guarded refill + batch drain ----------
    #[cfg(feature = "token-bucket")]
    {
        use subms_rate_limiter::TokenBucket;

        fn fleet(n: usize) -> Vec<TokenBucket> {
            let v: Vec<TokenBucket> = (0..n)
                .map(|_| TokenBucket::with_clock(TB_CAP, TB_RATE, Box::new(SteppingClock::new())))
                .collect();
            for (j, b) in v.iter().enumerate() {
                for _ in 0..PRE_DRAIN + (j & 1) {
                    b.try_acquire(1);
                }
            }
            v
        }

        let (rows, canon) = sweep("token-bucket/try_acquire", &SIZES, |n| {
            measure(
                || fleet(n),
                |v, i| v[i % v.len()].try_acquire(1),
                OPS,
                BATCH,
            )
        });
        let (cat, reason) = classify_feature(&rows, Some(base_p50), None);

        let avail = measure(
            || fleet(CANON_N),
            |v, i| {
                let _ = v[i % v.len()].available();
                true
            },
            OPS,
            BATCH,
        );
        let mut p99 = BTreeMap::new();
        p99.insert("try_acquire".to_string(), canon.p99);
        p99.insert("available".to_string(), avail.p99);
        manifest.set_feature("token-bucket", cat, &p99, &reason);
    }

    // ---------- hierarchical: child AND parent must both grant ----------
    #[cfg(feature = "hierarchical")]
    {
        use subms_rate_limiter::HierarchicalLimiter;

        // Swept on the CHILD COUNT, which is the only thing this feature can
        // scale. It is not a parent CHAIN - the source holds one parent and a
        // flat `Vec` of children, and a call is a `Vec` index plus a fixed
        // three bucket operations - so the cost is expected to be flat and the
        // sweep is what says so rather than a reading of the code.
        fn hier(n: usize) -> HierarchicalLimiter {
            let h = HierarchicalLimiter::with_clock_fn(
                HIER_PARENT_CAP,
                HIER_PARENT_RATE,
                n,
                TB_CAP,
                TB_RATE,
                || Box::new(SteppingClock::new()),
            );
            for c in 0..h.num_children() {
                for _ in 0..PRE_DRAIN + (c & 1) {
                    h.try_acquire(c, 1);
                }
            }
            h
        }

        let (rows, canon) = sweep("hierarchical/try_acquire", &SIZES, |n| {
            measure(
                || hier(n),
                |h, i| h.try_acquire(i % h.num_children(), 1),
                OPS,
                BATCH,
            )
        });
        let (cat, reason) = classify_feature(&rows, Some(base_p50), None);

        let mut p99 = BTreeMap::new();
        p99.insert("try_acquire".to_string(), canon.p99);
        manifest.set_feature("hierarchical", cat, &p99, &reason);
    }

    // ---------- distributed-backend: fixed-window counters ----------
    #[cfg(feature = "distributed-backend")]
    {
        use subms_rate_limiter::{DistributedLimiter, InMemoryBackend};

        // Keys are built once, outside every timed region: formatting one
        // inside the loop would put string construction in the measurement.
        // Fixed width so key length is not a second variable.
        let keys: Vec<String> = (0..DIST_CANON).map(|i| format!("key-{i:06}")).collect();

        let prefilled = |n: usize| {
            let d = DistributedLimiter::with_clock(
                Box::new(InMemoryBackend::new()),
                DIST_LIMIT,
                DIST_WINDOW_NS,
                Box::new(SteppingClock::new()),
            );
            for k in &keys[..n] {
                d.try_acquire(k);
            }
            d
        };

        let (rows, canon) = sweep("distributed-backend/try_acquire", &DIST_SIZES, |n| {
            measure(
                || prefilled(n),
                |d, i| d.try_acquire(&keys[i % DIST_HOT]),
                DIST_OPS,
                1,
            )
        });
        let (cat, reason) = classify_feature(&rows, Some(base_p50), None);

        let mut p99 = BTreeMap::new();
        p99.insert("try_acquire".to_string(), canon.p99);
        manifest.set_feature("distributed-backend", cat, &p99, &reason);
    }

    // ---------- metrics: counters around the same bucket ----------
    #[cfg(feature = "metrics")]
    {
        use subms_rate_limiter::MeteredTokenBucket;

        fn fleet(n: usize) -> Vec<MeteredTokenBucket> {
            let v: Vec<MeteredTokenBucket> = (0..n)
                .map(|_| {
                    MeteredTokenBucket::with_clock(TB_CAP, MET_RATE, Box::new(SteppingClock::new()))
                })
                .collect();
            for (j, b) in v.iter().enumerate() {
                for _ in 0..PRE_DRAIN + (j & 1) {
                    b.try_acquire(1);
                }
            }
            v
        }

        let (rows, canon) = sweep("metrics/try_acquire", &SIZES, |n| {
            measure(
                || fleet(n),
                |v, i| v[i % v.len()].try_acquire(1),
                OPS,
                BATCH,
            )
        });
        let (cat, reason) = classify_feature(&rows, Some(base_p50), None);

        let snap = measure(
            || fleet(CANON_N),
            |v, i| {
                let _ = v[i % v.len()].snapshot();
                true
            },
            OPS,
            BATCH,
        );
        let mut p99 = BTreeMap::new();
        p99.insert("try_acquire".to_string(), canon.p99);
        p99.insert("snapshot".to_string(), snap.p99);
        manifest.set_feature("metrics", cat, &p99, &reason);
    }

    // ---------- keyed: one GCRA limiter per key, sharded ----------
    #[cfg(feature = "keyed")]
    {
        use subms_rate_limiter::KeyedRateLimiter;

        // Swept on the LIVE KEY COUNT, the only axis this feature scales on.
        // A hash map lookup is expected to read flat; what the sweep is really
        // testing is that the sharded lock does not turn into the bottleneck as
        // the key set outgrows cache.
        //
        // Keys are formatted once, outside every timed region, and the clock is
        // driven rather than read: a fixed `now` of 0 with a burst wide enough
        // to cover every timed call keeps each op on the grant branch, which is
        // the dearer one (a reject returns before the map write).
        let keys: Vec<String> = (0..CANON_N).map(|i| format!("key-{i:06}")).collect();
        let burst = (OPS / SIZES[0] + 2) as u64;

        let (rows, canon) = sweep("keyed/try_acquire", &SIZES, |n| {
            measure(
                || KeyedRateLimiter::new(BASE_RATE, burst),
                |k, i| matches!(k.try_acquire_at(0, &keys[i % n], 1), Acquire::Ok),
                OPS,
                BATCH,
            )
        });
        let (cat, reason) = classify_feature(&rows, Some(base_p50), None);

        let mut p99 = BTreeMap::new();
        p99.insert("try_acquire".to_string(), canon.p99);
        manifest.set_feature("keyed", cat, &p99, &reason);
    }

    std::fs::create_dir_all(path.parent().unwrap())?;
    std::fs::write(&path, manifest.to_json())?;
    io::stdout().write_all(manifest.to_json().as_bytes())?;
    Ok(())
}

Trait Implementations

impl Backend for InMemoryBackend

fn incr(&self, key: &str, window_start_ns: u64, ttl_ns: u64) -> u64

Increment the counter at key for window_start_ns. Returns the new count after the bump. Must be atomic across concurrent callers.

fn read(&self, key: &str, window_start_ns: u64) -> u64

Read the current counter without bumping. Returns 0 if the (key, window) pair is unknown or expired.

impl Default for InMemoryBackend

fn default() -> Self

Returns the “default value” for a type.