hyperfile 0.10.5

#![cfg(not(feature = "blocking"))]
//! Planning reads apart from doing them: `HyperFile::plan_reads`, `ReadPolicy` and
//! `ReadPlan::execute`, on memory staging.
//!
//! The bytes a plan fetches are checked against `read` of the same ranges, for many
//! ranges and policies, on a file whose blocks are scattered over several objects
//! out of file order — holes, zero blocks, resident blocks, past the end. The
//! requests are checked against `read_plan`, the read path's own planner, for the
//! default policy.
//!
//! ```bash
//! cargo test --test functional_read_planning
//! cargo test --features wal --test functional_read_planning
//! ```

use std::io::ErrorKind;
#[cfg(not(feature = "wal"))]
use std::io::Result;
#[cfg(not(feature = "wal"))]
use std::sync::Arc;
#[cfg(not(feature = "wal"))]
use std::sync::Mutex;

use btree_ondisk::{MemoryBlockLoader, NullNodeCache};
#[cfg(not(feature = "wal"))]
use hyperfile::block_source::BlockSource;
use hyperfile::config::{HyperFileConfigBuilder, HyperFileRuntimeConfig};
use hyperfile::file::file::HyperFile;
use hyperfile::file::flags::{FileFlags, HyperFileFlags};
use hyperfile::file::mode::{FileMode, HyperFileMode};
use hyperfile::file::{HyperTrait, PlannedRead, ReadPolicy};
use hyperfile::staging::config::StagingConfig;
use hyperfile::staging::memory::MemoryStaging;
use hyperfile::staging::Staging;
use tokio::sync::Semaphore;

type MemFile<'a> = HyperFile<'a, MemoryStaging, MemoryBlockLoader<hyperfile::BlockPtr>, NullNodeCache>;

const BS: usize = 4096;
const BLOCKS: usize = 64;

fn runtime() -> HyperFileRuntimeConfig {
    let mut rt = HyperFileRuntimeConfig::default();
    rt.data_cache_dirty_max_bytes_threshold = 1 << 40;
    rt.data_cache_dirty_max_blocks_threshold = (1 << 40) / BS;
    rt.data_cache_dirty_max_flush_interval = u64::MAX / 2;
    rt.segment_buffer_size = 64 << 20;
    // Nothing kept after a flush, so what is resident is only what a test made so.
    rt.data_cache_blocks = 0;
    rt
}

async fn create(name: &str, flags: FileFlags) -> (MemoryStaging, MemFile<'static>) {
    let rt = runtime();
    let staging = MemoryStaging::new(StagingConfig::new_memory(name), rt.clone());
    let cfg = HyperFileConfigBuilder::new().with_staging_config(staging.config()).with_runtime_config(&rt).build();
    let f = MemFile::new(staging.clone(), staging.to_block_loader(), NullNodeCache, cfg,
        HyperFileFlags::from_flags(flags), HyperFileMode::from_mode(FileMode::default_file()))
        .await.expect("create");
    (staging, f)
}

fn block(b: usize, generation: u8) -> Vec<u8> {
    (0..BS).map(|i| (b as u8).wrapping_mul(31) ^ generation ^ (i as u8)).collect()
}

/// 64 blocks written over four flushes in an order that scatters them: each
/// object holds blocks far apart in the file, and file order is not object
/// order. Blocks 20..24 are a hole, 30 an explicit zero block; the last block is
/// half written.
async fn scattered(name: &str) -> (MemoryStaging, MemFile<'static>) {
    let (s, mut f) = create(name, FileFlags::rdwr()).await;
    for generation in 0..4u8 {
        for b in (0..BLOCKS).rev() {
            if b % 4 != generation as usize || (20..24).contains(&b) {
                continue;
            }
            let len = if b == BLOCKS - 1 { BS / 2 } else { BS };
            f.write(b * BS, &block(b, generation)[..len]).await.unwrap();
        }
        f.flush().await.expect("flush");
    }
    f.write_zero(30 * BS, BS).await.unwrap();
    f.flush().await.expect("flush");
    (s, f)
}

async fn executed(f: &mut MemFile<'static>, ranges: &[(u64, u64)], policy: &ReadPolicy) -> Vec<u8> {
    let plan = f.plan_reads(ranges, policy).await.expect("plan");
    let mut out = vec![0xEEu8; plan.len()];
    plan.execute(&mut out, &Semaphore::new(4)).await.expect("execute");
    out
}

/// What `read` returns for each range, laid out as a plan lays it: each range at
/// its requested length, zeros past what `read` returned.
async fn read_each(f: &mut MemFile<'static>, ranges: &[(u64, u64)]) -> Vec<u8> {
    let mut all = Vec::new();
    for (off, len) in ranges.iter().copied() {
        let mut buf = vec![0u8; len as usize];
        let n = f.read(off as usize, &mut buf).await.expect("read");
        buf[n..].fill(0);
        all.extend_from_slice(&buf);
    }
    all
}

fn rng(seed: &mut u64) -> u64 {
    *seed ^= *seed << 13;
    *seed ^= *seed >> 7;
    *seed ^= *seed << 17;
    *seed
}

/// Many random sets of ranges, under several policies: the plan fetches what
/// `read` returns, every time.
#[tokio::test]
async fn a_plan_fetches_what_read_returns() {
    let (_s, mut f) = scattered("rp-bytes").await;
    let size = (BLOCKS * BS - BS / 2) as u64;
    let policies = [
        ReadPolicy::default(),
        ReadPolicy { max_gap: 64 * BS, ..Default::default() },
        ReadPolicy { max_gap: 64 * BS, max_get: Some(3 * BS), ..Default::default() },
        ReadPolicy { max_gap: 8 * BS, max_overread: Some(0.25), ..Default::default() },
        ReadPolicy { max_get: Some(1), ..Default::default() },
    ];
    let mut seed = 0x9e3779b97f4a7c15u64;
    for round in 0..60 {
        let n = 1 + (rng(&mut seed) % 6) as usize;
        let ranges: Vec<(u64, u64)> = (0..n).map(|_| {
            let off = rng(&mut seed) % (size + 3 * BS as u64);
            let len = rng(&mut seed) % (6 * BS as u64) + 1;
            (off, len)
        }).collect();
        let want = read_each(&mut f, &ranges).await;
        for (k, p) in policies.iter().enumerate() {
            let got = executed(&mut f, &ranges, p).await;
            assert!(got == want, "round {} policy {} ranges {:?}", round, k, ranges);
        }
    }
}

/// With the default policy and one range, the requests are the read path's own —
/// `read_plan`'s — with one difference: requests to one object that touch are one,
/// though the file does not have them next to each other. A rewrite that moved
/// blocks 5, 6, 17 and 40..43 into one object is three requests to the read path
/// and one here.
#[tokio::test]
async fn the_default_policy_makes_the_read_paths_requests() {
    let (s, mut f) = create("rp-same", FileFlags::rdwr()).await;
    for b in 0..BLOCKS {
        f.write(b * BS, &block(b, 1)).await.unwrap();
        if b % 9 == 8 {
            f.flush().await.unwrap();
        }
    }
    f.flush().await.unwrap();
    // A few blocks rewritten, so runs break where objects change.
    for b in [5usize, 6, 17, 40, 41, 42] {
        f.write(b * BS, &block(b, 2)).await.unwrap();
    }
    f.flush().await.unwrap();
    let _ = s;
    let whole = f.read_plan(0, (BLOCKS * BS) as u64).await.unwrap().iter()
        .filter(|e| matches!(e, PlannedRead::Get { .. })).count();
    let planned = f.plan_reads(&[(0, (BLOCKS * BS) as u64)], &ReadPolicy::default()).await.unwrap().gets().len();
    assert_eq!(planned + 2, whole, "the rewritten blocks' three requests were not one");
    for (off, len) in [(0u64, (BLOCKS * BS) as u64), (3 * BS as u64 + 100, 20 * BS as u64), (39 * BS as u64, 5 * BS as u64)] {
        let want: Vec<(u64, u64, u64)> = f.read_plan(off, len).await.unwrap().into_iter()
            .filter_map(|e| match e { PlannedRead::Get { segid, at, len, .. } => Some((segid.as_cno() ^ (segid.part_id().unwrap_or(0) as u64) << 48, at, len)), _ => None })
            .collect();
        let plan = f.plan_reads(&[(off, len)], &ReadPolicy::default()).await.unwrap();
        let mut got: Vec<(u64, u64, u64)> = plan.gets().iter()
            .map(|g| (g.segid.as_cno() ^ (g.segid.part_id().unwrap_or(0) as u64) << 48, g.at, g.len)).collect();
        // The read path's requests, joined where they touch in one object.
        let mut want_sorted = want.clone();
        want_sorted.sort();
        let mut joined: Vec<(u64, u64, u64)> = Vec::new();
        for (obj, at, n) in want_sorted {
            match joined.last_mut() {
                Some((o, a, l)) if *o == obj && *a + *l == at => *l += n,
                _ => joined.push((obj, at, n)),
            }
        }
        got.sort();
        assert_eq!(got, joined, "range ({}, {})", off, len);
        assert_eq!(plan.overread_bytes(), 0);
    }
}

/// Blocks far apart in the file and next to each other in an object are one
/// request when asked for together — across ranges — and several when asked for
/// one range at a time.
#[tokio::test]
async fn ranges_far_apart_in_the_file_join_in_the_object() {
    let (s, mut f) = create("rp-cross", FileFlags::rdwr()).await;
    // One flush holding only these: the object has them side by side.
    for b in [40usize, 3, 27] {
        f.write(b * BS, &block(b, 7)).await.unwrap();
    }
    f.flush().await.unwrap();
    let ranges = [(3 * BS as u64, BS as u64), (27 * BS as u64, BS as u64), (40 * BS as u64, BS as u64)];

    let plan = f.plan_reads(&ranges, &ReadPolicy::default()).await.unwrap();
    assert_eq!(plan.gets().len(), 1, "{:?}", plan.gets());
    assert_eq!(plan.gets()[0].len, 3 * BS as u64);
    assert_eq!(plan.gets()[0].targets.len(), 3);

    let one_by_one: usize = {
        let mut n = 0;
        for r in ranges.iter() {
            n += f.read_plan(r.0, r.1).await.unwrap().iter().filter(|e| matches!(e, PlannedRead::Get { .. })).count();
        }
        n
    };
    assert_eq!(one_by_one, 3);

    s.read_timing().reset();
    let mut out = vec![0u8; plan.len()];
    plan.execute(&mut out, &Semaphore::new(4)).await.unwrap();
    assert_eq!(s.read_timing().snapshot().data_gets, 1, "one request was planned, more were made");
    let want: Vec<u8> = [3usize, 27, 40].iter().flat_map(|b| block(*b, 7)).collect();
    assert!(out == want);

    // Asked in another order, the bytes go where the ranges say.
    let rev: Vec<(u64, u64)> = ranges.iter().rev().copied().collect();
    let got = executed(&mut f, &rev, &ReadPolicy::default()).await;
    let want: Vec<u8> = [40usize, 27, 3].iter().flat_map(|b| block(*b, 7)).collect();
    assert!(got == want);
}

/// A gap is fetched only within `max_gap`, a request stays within `max_get`, and
/// the gaps fetched stay within `max_overread`.
#[tokio::test]
async fn the_policy_bounds_gaps_requests_and_overread() {
    let (_s, mut f) = create("rp-policy", FileFlags::rdwr()).await;
    for b in 0..16usize {
        f.write(b * BS, &block(b, 3)).await.unwrap();
    }
    f.flush().await.unwrap();
    // Blocks 0, 2, 3, 8: gaps of one block and of four.
    let ranges: Vec<(u64, u64)> = [0u64, 2, 3, 8].iter().map(|b| (b * BS as u64, BS as u64)).collect();
    let bs = BS as u64;

    let p = f.plan_reads(&ranges, &ReadPolicy::default()).await.unwrap();
    assert_eq!(p.gets().iter().map(|g| g.len).collect::<Vec<_>>(), vec![bs, 2 * bs, bs]);

    let p = f.plan_reads(&ranges, &ReadPolicy { max_gap: BS, ..Default::default() }).await.unwrap();
    assert_eq!(p.gets().iter().map(|g| g.len).collect::<Vec<_>>(), vec![4 * bs, bs]);
    assert_eq!(p.overread_bytes(), bs);

    let p = f.plan_reads(&ranges, &ReadPolicy { max_gap: 4 * BS, ..Default::default() }).await.unwrap();
    assert_eq!(p.gets().iter().map(|g| g.len).collect::<Vec<_>>(), vec![9 * bs]);
    assert_eq!(p.overread_bytes(), 5 * bs);
    assert_eq!(p.needed_bytes(), 4 * bs);

    // Budget for the small gap only.
    let p = f.plan_reads(&ranges, &ReadPolicy { max_gap: 4 * BS, max_overread: Some(0.25), ..Default::default() }).await.unwrap();
    assert_eq!(p.gets().iter().map(|g| g.len).collect::<Vec<_>>(), vec![4 * bs, bs]);

    let p = f.plan_reads(&ranges, &ReadPolicy { max_gap: 4 * BS, max_get: Some(4 * BS), ..Default::default() }).await.unwrap();
    assert!(p.gets().iter().all(|g| g.len <= 4 * bs), "{:?}", p.gets());

    // A long run split for concurrency.
    let p = f.plan_reads(&[(0, 16 * bs)], &ReadPolicy { max_get: Some(4 * BS), ..Default::default() }).await.unwrap();
    assert_eq!(p.gets().len(), 4);
}

/// The plan fetches what the blocks held when it was made, though the handle
/// writes and publishes over them before it runs.
#[tokio::test]
async fn a_plan_fetches_what_was_there_when_it_was_made() {
    let (_s, mut f) = create("rp-snapshot", FileFlags::rdwr()).await;
    for b in 0..8usize {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    f.flush().await.unwrap();
    let plan = f.plan_reads(&[(0, 8 * BS as u64)], &ReadPolicy::default()).await.unwrap();
    for b in 0..8usize {
        f.write(b * BS, &block(b, 9)).await.unwrap();
    }
    f.flush().await.unwrap();
    let mut out = vec![0u8; plan.len()];
    plan.execute(&mut out, &Semaphore::new(2)).await.unwrap();
    let before: Vec<u8> = (0..8).flat_map(|b| block(b, 1)).collect();
    assert!(out == before);
}

/// Resident blocks are copied into the plan, and need no request.
#[tokio::test]
async fn resident_blocks_are_copied_in_and_need_no_request() {
    let (s, mut f) = create("rp-resident", FileFlags::rdwr()).await;
    for b in 0..8usize {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    f.flush().await.unwrap();
    f.write(2 * BS, &block(2, 5)).await.unwrap();
    let plan = f.plan_reads(&[(0, 8 * BS as u64)], &ReadPolicy::default()).await.unwrap();
    assert_eq!(plan.local_bytes(), BS as u64);
    assert_eq!(plan.gets().len(), 2, "{:?}", plan.gets());
    // Changed again after planning: the plan has what it copied.
    f.write(2 * BS, &block(2, 6)).await.unwrap();
    s.read_timing().reset();
    let mut out = vec![0u8; plan.len()];
    plan.execute(&mut out, &Semaphore::new(2)).await.unwrap();
    assert_eq!(s.read_timing().snapshot().data_gets, 2);
    assert!(out[2 * BS..3 * BS] == block(2, 5)[..]);
    assert!(out[..BS] == block(0, 1)[..]);
}

/// The buffer must be the plan's length, and a handle not open for reading
/// cannot plan.
#[tokio::test]
async fn misuse_is_refused() {
    let (_s, mut f) = create("rp-misuse", FileFlags::rdwr()).await;
    f.write(0, &block(0, 1)).await.unwrap();
    f.flush().await.unwrap();
    let plan = f.plan_reads(&[(0, BS as u64)], &ReadPolicy::default()).await.unwrap();
    let mut short = vec![0u8; BS - 1];
    assert_eq!(plan.execute(&mut short, &Semaphore::new(1)).await.unwrap_err().kind(), ErrorKind::InvalidInput);

    let (_s, mut w) = create("rp-wronly", FileFlags::wronly()).await;
    w.write(0, &block(0, 1)).await.unwrap();
    let e = w.plan_reads(&[(0, BS as u64)], &ReadPolicy::default()).await.unwrap_err();
    assert_eq!(e.raw_os_error(), Some(libc_ebadf()));
}

fn libc_ebadf() -> i32 {
    9
}

/// A source over a plain buffer.
#[cfg(not(feature = "wal"))]
struct Buf(Mutex<Vec<u8>>);

#[cfg(not(feature = "wal"))]
impl BlockSource for Buf {
    fn read_at(&self, off: u64, buf: &mut [u8]) -> Result<()> {
        let d = self.0.lock().unwrap();
        buf.copy_from_slice(&d[off as usize..off as usize + buf.len()]);
        Ok(())
    }
}

/// Blocks written by reference are read from the source when the plan runs.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn blocks_written_by_reference_come_from_the_source() {
    let (_s, mut f) = create("rp-ref", FileFlags::rdwr()).await;
    for b in 0..8usize {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    f.flush().await.unwrap();
    let src: Vec<u8> = (0..8).flat_map(|b| block(b, 4)).collect();
    f.set_block_source(Some(Arc::new(Buf(Mutex::new(src.clone()))))).unwrap();
    f.write_ref(2 * BS, 3 * BS).await.unwrap();
    let ranges = [(0u64, 8 * BS as u64), (3 * BS as u64 + 7, 100)];
    let want = read_each(&mut f, &ranges).await;
    let got = executed(&mut f, &ranges, &ReadPolicy { max_gap: 8 * BS, ..Default::default() }).await;
    assert!(got == want);
    assert!(got[2 * BS..5 * BS] == src[2 * BS..5 * BS]);
}