hyperfile 0.11.3

#![cfg(all(target_os = "linux", not(any(feature = "blocking", feature = "wal"))))]
//! What sending ahead costs the thread that writes, on memory staging.
//!
//! Sending ahead does work on the writer's thread that a flush would otherwise
//! do later: choosing the blocks, copying them for the upload, setting their
//! pointers. It should cost about that, not a scan of everything already sent
//! for every piece. Measured as the writing thread's CPU time, against the same
//! writes under `PartedSegment`, best of three after a warm-up, with room for
//! noise. Linux only: the thread's CPU time is read with `CLOCK_THREAD_CPUTIME_ID`.
//!
//! One test, alone in its binary, so that nothing else runs on the machine's
//! cores while it measures. Run it in release:
//!
//! ```bash
//! cargo test --release --test functional_stream_cpu
//! ```

use btree_ondisk::{MemoryBlockLoader, NullNodeCache};
use hyperfile::config::{HyperFileConfigBuilder, HyperFileMetaConfig, HyperFileRuntimeConfig};
use hyperfile::file::file::HyperFile;
use hyperfile::file::flags::{FileFlags, HyperFileFlags};
use hyperfile::file::mode::{FileMode, HyperFileMode};
use hyperfile::file::HyperTrait;
use hyperfile::meta_format::BlockPtrFormat;
use hyperfile::staging::config::StagingConfig;
use hyperfile::staging::memory::MemoryStaging;
use hyperfile::staging::Staging;
use std::sync::Arc;

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

const BS: usize = 4096;
const MIB: usize = 1 << 20;
const TOTAL: usize = 512 * MIB;

fn thread_cpu() -> std::time::Duration {
    let mut t = libc::timespec { tv_sec: 0, tv_nsec: 0 };
    // SAFETY: a valid clock id and a valid out pointer.
    unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &mut t) };
    std::time::Duration::new(t.tv_sec as u64, t.tv_nsec as u32)
}

struct Src(Vec<u8>);

impl hyperfile::block_source::BlockSource for Src {
    fn read_at(&self, off: u64, buf: &mut [u8]) -> std::io::Result<()> {
        buf.copy_from_slice(&self.0[off as usize..off as usize + buf.len()]);
        Ok(())
    }
}

/// The writing thread's CPU time for `TOTAL` bytes in 16 MiB calls, by value or
/// by reference, in `fmt`, sending ahead at the default target.
async fn writer_cpu(fmt: BlockPtrFormat, by_ref: bool, src: &Arc<Src>, chunk: &[u8]) -> std::time::Duration {
    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.segment_buffer_size = 1 << 40;
    rt.segment_mpu_max_concurrency = 8;
    let mut meta = HyperFileMetaConfig::default();
    meta.block_ptr_format = fmt;
    let s = MemoryStaging::new(StagingConfig::new_memory("cpu"), rt.clone());
    let cfg = HyperFileConfigBuilder::new().with_staging_config(s.config()).with_meta_config(&meta).with_runtime_config(&rt).build();
    let mut f = MemFile::new(s.clone(), s.to_block_loader(), NullNodeCache, cfg, HyperFileFlags::from_flags(FileFlags::rdwr()),
        HyperFileMode::from_mode(FileMode::default_file())).await.expect("create");
    if by_ref {
        f.set_block_source(Some(src.clone())).unwrap();
    }
    let mut spent = std::time::Duration::ZERO;
    let mut off = 0;
    while off < TOTAL {
        let t = thread_cpu();
        if by_ref { f.write_ref(off, chunk.len()).await.unwrap(); } else { f.write(off, chunk).await.unwrap(); }
        spent += thread_cpu() - t;
        off += chunk.len();
        // Room for what was sent to land, as a writer paced by its producer gives.
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    }
    f.flush().await.expect("flush");
    spent
}

async fn best_of_three(fmt: BlockPtrFormat, by_ref: bool, src: &Arc<Src>, chunk: &[u8]) -> std::time::Duration {
    let mut best = std::time::Duration::MAX;
    for _ in 0..3 {
        best = best.min(writer_cpu(fmt, by_ref, src, chunk).await);
    }
    best
}

/// Sending ahead costs the writer a few times what writing into the dirty tier
/// does, not the many times a scan of every block already sent, at each piece,
/// made it. By value: the copy for the upload and the pointers set are the
/// writer's now, and come to about twice the write. By reference: no copy.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn sending_ahead_costs_the_writer_about_what_it_does() {
    let chunk: Vec<u8> = (0..16 * MIB).map(|i| (i.wrapping_mul(2654435761) >> 13) as u8).collect();
    let src = Arc::new(Src((0..TOTAL).map(|i| (i.wrapping_mul(2654435761) >> 13) as u8).collect()));
    // Warm-up: the allocator and the page cache.
    let _ = writer_cpu(BlockPtrFormat::PartedSegment, false, &src, &chunk).await;
    let value_parted = best_of_three(BlockPtrFormat::PartedSegment, false, &src, &chunk).await;
    let value_stream = best_of_three(BlockPtrFormat::StreamSegment, false, &src, &chunk).await;
    let ref_parted = best_of_three(BlockPtrFormat::PartedSegment, true, &src, &chunk).await;
    let ref_stream = best_of_three(BlockPtrFormat::StreamSegment, true, &src, &chunk).await;
    let gib = TOTAL as f64 / (1u64 << 30) as f64;
    let per = |d: std::time::Duration| d.as_secs_f64() / gib;
    println!("writer CPU, s/GiB: by value {:.2} parted, {:.2} stream; by reference {:.2} parted, {:.2} stream",
        per(value_parted), per(value_stream), per(ref_parted), per(ref_stream));
    assert!(value_stream.as_secs_f64() <= 3.0 * value_parted.as_secs_f64(),
        "by value, sending ahead cost the writer {:.2} s/GiB against {:.2}", per(value_stream), per(value_parted));
    // A write by reference costs little, so the bound is in time rather than a
    // multiple: what is left is choosing blocks and setting their pointers.
    assert!(per(ref_stream) <= per(ref_parted) + 0.25,
        "by reference, sending ahead cost the writer {:.2} s/GiB against {:.2}", per(ref_stream), per(ref_parted));
}