chunky-bits 0.2.0

Chunky Bits is simple, unmanaged, distributed HTTP object store tool.
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

use std::{
    error::Error,
    sync::Arc,
};

use chunky_bits::{
    error::FileWriteError,
    file::{
        new_profiler,
        FileReference,
        Location,
        LocationContext,
        WeightedLocation,
    },
};
use futures::stream::StreamExt;
use tempfile::{
    tempdir,
    TempDir,
};
use tokio::io::{
    repeat,
    AsyncReadExt,
};

#[tokio::test(flavor = "multi_thread", worker_threads = 10)]
async fn test_file_write() -> Result<(), Box<dyn Error>> {
    let mut tasks = Vec::new();
    for data in 1..=3 {
        for parity in 1..=3 {
            tasks.push(tokio::task::spawn(async move {
                // Set the length to be something not divisible by (d+p)==5
                let length = (1 << 23) + 7;
                let chunk_size = 1 << 20;
                let mut reader = repeat(0).take(length);
                let file_ref = FileReference::write_builder()
                    .chunk_size(chunk_size)
                    .data_chunks(data)
                    .parity_chunks(parity)
                    .write(&mut reader)
                    .await?;
                assert_eq!(file_ref.length.unwrap(), length);
                let part_size: u64 = (chunk_size as u64) * (data as u64);
                assert_eq!(
                    file_ref.parts.len() as u64,
                    (length + part_size - 1) / part_size,
                );
                Ok::<_, FileWriteError>(())
            }));
        }
    }
    for task in tasks.drain(..) {
        task.await??;
    }
    Ok(())
}

#[tokio::test]
async fn not_enough_writers() -> Result<(), Box<dyn Error>> {
    let mut directories: Vec<TempDir> = vec![tempdir()?, tempdir()?, tempdir()?, tempdir()?];
    let locations: Vec<WeightedLocation> = directories
        .iter()
        .map(|dir| Location::from(dir.path()).into())
        .collect();
    let locations = Arc::new(locations);
    for data in 1..=3 {
        for parity in 1..=3 {
            let length = 1;
            let chunk_size = 1 << 20;
            let mut reader = repeat(0).take(length);
            let result = FileReference::write_builder()
                .destination(Arc::new(locations.clone()))
                .chunk_size(chunk_size)
                .data_chunks(data)
                .parity_chunks(parity)
                .write(&mut reader)
                .await;
            if (data + parity) > locations.len() {
                match result {
                    Err(FileWriteError::NotEnoughWriters) => {},
                    Ok(_) => {
                        panic!(
                            "Wrote {} chunks to {} locations",
                            (data + parity),
                            locations.len(),
                        );
                    },
                    Err(err) => {
                        panic!(
                            "Could not write {} chunks to {} locations: {}",
                            (data + parity),
                            locations.len(),
                            err,
                        );
                    },
                }
            } else if let Err(err) = result {
                panic!(
                    "Could not write {} chunks to {} locations: {}",
                    (data + parity),
                    locations.len(),
                    err,
                );
            }
        }
    }
    for dir in directories.drain(..) {
        dir.close()?;
    }
    Ok(())
}

#[tokio::test]
async fn test_read_profiler() -> Result<(), Box<dyn Error>> {
    let directories: Vec<TempDir> = vec![tempdir()?, tempdir()?, tempdir()?, tempdir()?];
    let locations: Vec<WeightedLocation> = directories
        .iter()
        .map(|dir| Location::from(dir.path()).into())
        .collect();
    let length = (1 << 23) + 7;
    let mut reader = repeat(0).take(length);
    let file_ref = FileReference::write_builder()
        .data_chunks(2)
        .parity_chunks(1)
        .destination(locations)
        .write(&mut reader)
        .await?;
    let (profiler, reporter) = new_profiler();
    let location_context: LocationContext = LocationContext::builder().profiler(profiler).build();
    let mut stream = file_ref
        .read_builder_owned()
        .location_context(location_context)
        .stream_reader_owned();
    while let Some(_) = stream.next().await {}
    let report = reporter.profile().await;
    let average_duration = report.average_read_duration().unwrap();
    assert!(average_duration.as_nanos() > 0);
    assert!(average_duration.as_millis() < 1000);
    assert!(report.average_write_duration() == None);
    Ok(())
}