hyperfile 0.12.0

A file-like interface abstraction layer provides random read and write capabilities for S3
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

//! Compaction on S3: `fs_relocate`, `fs_content_cno` and `fs_content_cno_fast`.
//!
//! `functional_compaction` covers when a publish is marked as moving blocks only.
//! This covers what only S3 shows: relocation reads blocks with the read path's
//! coalescing — a run of blocks in one object is one request, not one per block —
//! and the reader that does not open the file agrees with the one that does.
//!
//! ```bash
//! cargo test --release --test integration_s3_compaction -- --ignored --test-threads=1
//! ```

// The WAL build refuses relocation, and leaves what only that test uses unused.
#![cfg_attr(feature = "wal", allow(unused_imports, dead_code))]

mod common;
use common::*;

use hyperfile::config::HyperFileRuntimeConfig;
use hyperfile::file::flags::FileFlags;
use hyperfile::file::hyper::Hyper;
use hyperfile::file::mode::FileMode;

const BS: usize = 4096;
const MIB: usize = 1 << 20;

fn runtime() -> HyperFileRuntimeConfig {
    let mut r = HyperFileRuntimeConfig::default();
    r.data_cache_dirty_max_bytes_threshold = 1 << 40;
    r.data_cache_dirty_max_blocks_threshold = (1 << 40) / BS;
    r.data_cache_dirty_max_flush_interval = u64::MAX / 2;
    r
}

/// 8 MiB written in one flush, relocated whole and flushed again: few requests
/// for 2048 blocks, the content checkpoint unchanged by both readers, the data
/// the same from a cold handle; then a write moves it on.
#[cfg(not(feature = "wal"))]
#[tokio::test]
#[ignore]
async fn relocation_reads_in_few_requests_and_keeps_the_content_checkpoint() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = TestFile::new(&client).await;
    let data: Vec<u8> = (0..8 * MIB).map(|i| (i / BS) as u8 ^ 0x2a).collect();
    {
        let mut h = Hyper::fs_open_or_create_with_default_opt(&client, tf.uri(), FileFlags::rdwr(), FileMode::default_file())
            .await.expect("create");
        h.fs_write(0, &data).await.expect("write");
        h.fs_flush().await.expect("flush");
        let _ = h.fs_release().await;
    }
    let before = Hyper::fs_content_cno_fast(&client, tf.uri()).await.expect("content_cno_fast");

    let mut h = Hyper::fs_open_opt(&client, tf.uri(), FileFlags::rdwr(), &runtime()).await.expect("open");
    let opened = h.fs_content_cno().expect("content_cno");
    h.read_timing().reset();
    let n = h.fs_relocate(&[(0, 2048)]).await.expect("relocate");
    let reads = h.read_timing().snapshot();
    let relocated_cno = h.fs_flush().await.expect("flush");
    let after_open = h.fs_content_cno().expect("content_cno");
    let _ = h.fs_release().await;
    let after_fast = Hyper::fs_content_cno_fast(&client, tf.uri()).await.expect("content_cno_fast");

    let mut c = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly()).await.expect("reopen");
    let mut back = vec![0u8; data.len()];
    let got = c.fs_read(0, &mut back).await.expect("read");
    back.truncate(got);
    let _ = c.fs_release().await;

    let mut w = Hyper::fs_open(&client, tf.uri(), FileFlags::rdwr()).await.expect("open to write");
    w.fs_write(0, b"changed").await.expect("write");
    w.fs_flush().await.expect("flush");
    let _ = w.fs_release().await;
    let changed = Hyper::fs_content_cno_fast(&client, tf.uri()).await.expect("content_cno_fast");
    tf.cleanup(&client).await;

    assert_eq!(n, 2048, "not every block was relocated");
    assert!(reads.data_gets <= 4, "2048 blocks in one object took {} requests", reads.data_gets);
    assert_eq!(opened, before, "the two readers disagree on an unchanged file");
    assert!(relocated_cno > before, "the relocation did not publish");
    assert_eq!(after_open, before, "content_cno moved on a relocation");
    assert_eq!(after_fast, before, "content_cno_fast moved on a relocation");
    assert!(back == data, "relocated blocks read back differently");
    assert!(changed > after_fast, "content_cno did not move on a write");
}

async fn objects_under(client: &aws_sdk_s3::Client, uri: &str) -> usize {
    let (bucket, prefix) = uri.trim_start_matches("s3://").split_once('/').unwrap();
    let mut n = 0;
    let mut pages = client.list_objects_v2().bucket(bucket).prefix(format!("{}/", prefix)).into_paginator().send();
    while let Some(page) = pages.next().await {
        n += page.expect("list").contents().len();
    }
    n
}

/// A pass larger than the dirty-data limits, handed over in chunks as the cleaner
/// does, is written out as partials while it is marked and published once: one
/// checkpoint, the content checkpoint unchanged by both readers, the data the same
/// from a cold handle.
#[cfg(not(feature = "wal"))]
#[tokio::test]
#[ignore]
async fn a_pass_past_the_limits_is_written_as_partials_and_published_once() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = TestFile::new(&client).await;
    let data: Vec<u8> = (0..8 * MIB).map(|i| (i / BS) as u8 ^ 0x5c).collect();
    {
        let mut h = Hyper::fs_open_or_create_with_default_opt(&client, tf.uri(), FileFlags::rdwr(), FileMode::default_file())
            .await.expect("create");
        h.fs_write(0, &data).await.expect("write");
        h.fs_flush().await.expect("flush");
        let _ = h.fs_release().await;
    }
    let before = Hyper::fs_content_cno_fast(&client, tf.uri()).await.expect("content_cno_fast");
    let objects0 = objects_under(&client, tf.uri()).await;

    let mut rt = runtime();
    rt.parted_segment_enabled = true;
    rt.data_cache_dirty_max_bytes_threshold = 2 * MIB;
    rt.data_cache_dirty_max_blocks_threshold = 2 * MIB / BS;
    rt.data_cache_dirty_min_bytes_to_part = MIB;
    let mut h = Hyper::fs_open_opt(&client, tf.uri(), FileFlags::rdwr(), &rt).await.expect("open");
    let cno0 = h.fs_last_cno();
    let mut marked = 0;
    for first in (0..2048u64).step_by(256) {
        marked += h.fs_relocate(&[(first, 256)]).await.expect("relocate");
    }
    let cno_marking = h.fs_last_cno();
    let objects_marking = objects_under(&client, tf.uri()).await;
    let cno = h.fs_flush().await.expect("flush");
    let _ = h.fs_release().await;
    let after_fast = Hyper::fs_content_cno_fast(&client, tf.uri()).await.expect("content_cno_fast");

    let mut c = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly()).await.expect("reopen");
    let after_open = c.fs_content_cno().expect("content_cno");
    let mut back = vec![0u8; data.len()];
    let got = c.fs_read(0, &mut back).await.expect("read");
    back.truncate(got);
    let _ = c.fs_release().await;
    tf.cleanup(&client).await;

    assert_eq!(marked, 2048);
    assert_eq!(cno_marking, cno0, "a relocate published");
    assert!(objects_marking >= objects0 + 2, "no partials while marking: {} objects, {} before",
        objects_marking, objects0);
    assert_eq!(cno, cno0 + 1, "the pass is one checkpoint");
    assert_eq!(after_fast, before, "content_cno_fast moved on a relocation");
    assert_eq!(after_open, before, "content_cno moved on a relocation");
    assert!(back == data, "relocated blocks read back differently");
}