3 use std
::path
::{Path, PathBuf}
;
4 use std
::io
::{Read, Write}
;
5 use std
::sync
::{Arc, Mutex}
;
6 use std
::os
::unix
::io
::AsRawFd
;
12 pub struct GarbageCollectionStatus
{
13 pub used_bytes
: usize,
14 pub used_chunks
: usize,
15 pub disk_bytes
: usize,
16 pub disk_chunks
: usize,
19 impl Default
for GarbageCollectionStatus
{
20 fn default() -> Self {
21 GarbageCollectionStatus
{
30 /// File system based chunk store
31 pub struct ChunkStore
{
32 name
: String
, // used for error reporting
33 pub (crate) base
: PathBuf
,
36 locker
: Arc
<Mutex
<tools
::ProcessLocker
>>,
39 // TODO: what about sysctl setting vm.vfs_cache_pressure (0 - 100) ?
41 pub fn verify_chunk_size(size
: u64) -> Result
<(), Error
> {
43 static SIZES
: [u64; 7] = [64*1024, 128*1024, 256*1024, 512*1024, 1024*1024, 2048*1024, 4096*1024];
45 if !SIZES
.contains(&size
) {
46 bail
!("Got unsupported chunk size '{}'", size
);
51 fn digest_to_prefix(digest
: &[u8]) -> PathBuf
{
53 let mut buf
= Vec
::<u8>::with_capacity(2+1+2+1);
55 const HEX_CHARS
: &'
static [u8; 16] = b
"0123456789abcdef";
57 buf
.push(HEX_CHARS
[(digest
[0] as usize) >> 4]);
58 buf
.push(HEX_CHARS
[(digest
[0] as usize) &0xf]);
59 buf
.push(HEX_CHARS
[(digest
[1] as usize) >> 4]);
60 buf
.push(HEX_CHARS
[(digest
[1] as usize) & 0xf]);
63 let path
= unsafe { String::from_utf8_unchecked(buf)}
;
70 fn chunk_dir
<P
: AsRef
<Path
>>(path
: P
) -> PathBuf
{
72 let mut chunk_dir
: PathBuf
= PathBuf
::from(path
.as_ref());
73 chunk_dir
.push(".chunks");
78 pub fn create
<P
: Into
<PathBuf
>>(name
: &str, path
: P
) -> Result
<Self, Error
> {
80 let base
: PathBuf
= path
.into();
82 if !base
.is_absolute() {
83 bail
!("expected absolute path - got {:?}", base
);
86 let chunk_dir
= Self::chunk_dir(&base
);
88 if let Err(err
) = std
::fs
::create_dir(&base
) {
89 bail
!("unable to create chunk store '{}' at {:?} - {}", name
, base
, err
);
92 if let Err(err
) = std
::fs
::create_dir(&chunk_dir
) {
93 bail
!("unable to create chunk store '{}' subdir {:?} - {}", name
, chunk_dir
, err
);
96 // create 64*1024 subdirs
97 let mut last_percentage
= 0;
100 let mut l1path
= chunk_dir
.clone();
101 l1path
.push(format
!("{:04x}", i
));
102 if let Err(err
) = std
::fs
::create_dir(&l1path
) {
103 bail
!("unable to create chunk store '{}' subdir {:?} - {}", name
, l1path
, err
);
105 let percentage
= (i
*100)/(64*1024);
106 if percentage
!= last_percentage
{
107 eprintln
!("Percentage done: {}", percentage
);
108 last_percentage
= percentage
;
112 Self::open(name
, base
)
115 pub fn open
<P
: Into
<PathBuf
>>(name
: &str, path
: P
) -> Result
<Self, Error
> {
117 let base
: PathBuf
= path
.into();
119 if !base
.is_absolute() {
120 bail
!("expected absolute path - got {:?}", base
);
123 let chunk_dir
= Self::chunk_dir(&base
);
125 if let Err(err
) = std
::fs
::metadata(&chunk_dir
) {
126 bail
!("unable to open chunk store '{}' at {:?} - {}", name
, chunk_dir
, err
);
129 let mut lockfile_path
= base
.clone();
130 lockfile_path
.push(".lock");
132 let locker
= tools
::ProcessLocker
::new(&lockfile_path
)?
;
135 name
: name
.to_owned(),
139 mutex
: Mutex
::new(false)
143 pub fn touch_chunk(&self, digest
:&[u8]) -> Result
<(), Error
> {
145 let mut chunk_path
= self.chunk_dir
.clone();
146 let prefix
= digest_to_prefix(&digest
);
147 chunk_path
.push(&prefix
);
148 let digest_str
= tools
::digest_to_hex(&digest
);
149 chunk_path
.push(&digest_str
);
151 const UTIME_NOW
: i64 = ((1 << 30) - 1);
152 const UTIME_OMIT
: i64 = ((1 << 30) - 2);
154 let times
: [libc
::timespec
; 2] = [
155 libc
::timespec { tv_sec: 0, tv_nsec: UTIME_NOW }
,
156 libc
::timespec { tv_sec: 0, tv_nsec: UTIME_OMIT }
161 let res
= chunk_path
.with_nix_path(|cstr
| unsafe {
162 libc
::utimensat(-1, cstr
.as_ptr(), ×
[0], libc
::AT_SYMLINK_NOFOLLOW
)
165 if let Err(err
) = nix
::errno
::Errno
::result(res
) {
166 bail
!("updata atime failed for chunk {:?} - {}", chunk_path
, err
);
172 pub fn read_chunk(&self, digest
:&[u8], buffer
: &mut Vec
<u8>) -> Result
<(), Error
> {
174 let mut chunk_path
= self.chunk_dir
.clone();
175 let prefix
= digest_to_prefix(&digest
);
176 chunk_path
.push(&prefix
);
177 let digest_str
= tools
::digest_to_hex(&digest
);
178 chunk_path
.push(&digest_str
);
181 let f
= std
::fs
::File
::open(&chunk_path
)?
;
182 let mut decoder
= zstd
::stream
::Decoder
::new(f
)?
;
184 decoder
.read_to_end(buffer
)?
;
189 pub fn get_chunk_iterator(
191 print_percentage
: bool
,
193 impl Iterator
<Item
= Result
<tools
::fs
::ReadDirEntry
, Error
>> + std
::iter
::FusedIterator
,
197 use nix
::fcntl
::OFlag
;
198 use nix
::sys
::stat
::Mode
;
200 let base_handle
= match Dir
::open(
201 &self.chunk_dir
, OFlag
::O_RDONLY
, Mode
::empty()) {
203 Err(err
) => bail
!("unable to open store '{}' chunk dir {:?} - {}",
204 self.name
, self.chunk_dir
, err
),
207 let mut verbose
= true;
208 let mut last_percentage
= 0;
210 Ok((0..0x10000).filter_map(move |index
| {
211 if print_percentage
{
212 let percentage
= (index
* 100) / 0x10000;
213 if last_percentage
!= percentage
{
214 last_percentage
= percentage
;
215 eprintln
!("percentage done: {}", percentage
);
218 let subdir
: &str = &format
!("{:04x}", index
);
219 match tools
::fs
::read_subdir(base_handle
.as_raw_fd(), subdir
) {
222 eprintln
!("Error iterating through chunks: {}", e
);
227 Ok(iter
) => Some(iter
),
232 // Check that the file name is actually a hash! (64 hex digits)
233 let entry
= match entry
{
234 Err(_
) => return true, // pass errors onwards
235 Ok(ref entry
) => entry
,
237 let bytes
= entry
.file_name().to_bytes();
238 if bytes
.len() != 64 {
242 if !b
.is_ascii_hexdigit() {
250 pub fn oldest_writer(&self) -> Option
<i64> {
251 tools
::ProcessLocker
::oldest_shared_lock(self.locker
.clone())
254 pub fn sweep_unused_chunks(
256 oldest_writer
: Option
<i64>,
257 status
: &mut GarbageCollectionStatus
258 ) -> Result
<(), Error
> {
259 use nix
::sys
::stat
::fstatat
;
261 let now
= unsafe { libc::time(std::ptr::null_mut()) }
;
263 let mut min_atime
= now
- 3600*24; // at least 24h (see mount option relatime)
265 if let Some(stamp
) = oldest_writer
{
266 if stamp
< min_atime
{
271 min_atime
-= 300; // add 5 mins gap for safety
273 for entry
in self.get_chunk_iterator(true)?
{
275 tools
::fail_on_shutdown()?
;
277 let (dirfd
, entry
) = match entry
{
278 Ok(entry
) => (entry
.parent_fd(), entry
),
279 Err(_
) => continue, // ignore errors
282 let file_type
= match entry
.file_type() {
283 Some(file_type
) => file_type
,
284 None
=> bail
!("unsupported file system type on chunk store '{}'", self.name
),
286 if file_type
!= nix
::dir
::Type
::File
{
290 let filename
= entry
.file_name();
292 let lock
= self.mutex
.lock();
294 if let Ok(stat
) = fstatat(dirfd
, filename
, nix
::fcntl
::AtFlags
::AT_SYMLINK_NOFOLLOW
) {
295 let age
= now
- stat
.st_atime
;
296 //println!("FOUND {} {:?}", age/(3600*24), filename);
297 if stat
.st_atime
< min_atime
{
298 println
!("UNLINK {} {:?}", age
/(3600*24), filename
);
299 let res
= unsafe { libc::unlinkat(dirfd, filename.as_ptr(), 0) }
;
301 let err
= nix
::Error
::last();
303 "unlink chunk {:?} failed on store '{}' - {}",
310 status
.disk_chunks
+= 1;
311 status
.disk_bytes
+= stat
.st_size
as usize;
319 pub fn insert_chunk(&self, chunk
: &[u8]) -> Result
<(bool
, [u8; 32], u64), Error
> {
321 // fixme: use Sha512/256 when available
322 let digest
= sha
::sha256(chunk
);
323 let (new
, csize
) = self.insert_chunk_noverify(&digest
, chunk
)?
;
324 Ok((new
, digest
, csize
))
327 pub fn insert_chunk_noverify(
331 ) -> Result
<(bool
, u64), Error
> {
333 //println!("DIGEST {}", tools::digest_to_hex(&digest));
335 let mut chunk_path
= self.chunk_dir
.clone();
336 let prefix
= digest_to_prefix(digest
);
337 chunk_path
.push(&prefix
);
338 let digest_str
= tools
::digest_to_hex(digest
);
339 chunk_path
.push(&digest_str
);
341 let lock
= self.mutex
.lock();
343 if let Ok(metadata
) = std
::fs
::metadata(&chunk_path
) {
344 if metadata
.is_file() {
345 return Ok((true, metadata
.len()));
347 bail
!("Got unexpected file type on store '{}' for chunk {}", self.name
, digest_str
);
351 let mut tmp_path
= chunk_path
.clone();
352 tmp_path
.set_extension("tmp");
354 let f
= std
::fs
::File
::create(&tmp_path
)?
;
356 let mut encoder
= zstd
::stream
::Encoder
::new(f
, 1)?
;
358 encoder
.write_all(chunk
)?
;
359 let f
= encoder
.finish()?
;
361 if let Err(err
) = std
::fs
::rename(&tmp_path
, &chunk_path
) {
362 if let Err(_
) = std
::fs
::remove_file(&tmp_path
) { /* ignore */ }
364 "Atomic rename on store '{}' failed for chunk {} - {}",
371 // fixme: is there a better way to get the compressed size?
372 let stat
= nix
::sys
::stat
::fstat(f
.as_raw_fd())?
;
373 let compressed_size
= stat
.st_size
as u64;
375 //println!("PATH {:?}", chunk_path);
379 Ok((false, compressed_size
))
382 pub fn relative_path(&self, path
: &Path
) -> PathBuf
{
384 let mut full_path
= self.base
.clone();
385 full_path
.push(path
);
389 pub fn base_path(&self) -> PathBuf
{
393 pub fn try_shared_lock(&self) -> Result
<tools
::ProcessLockSharedGuard
, Error
> {
394 tools
::ProcessLocker
::try_shared_lock(self.locker
.clone())
397 pub fn try_exclusive_lock(&self) -> Result
<tools
::ProcessLockExclusiveGuard
, Error
> {
398 tools
::ProcessLocker
::try_exclusive_lock(self.locker
.clone())
404 fn test_chunk_store1() {
406 let mut path
= std
::fs
::canonicalize(".").unwrap(); // we need absulute path
407 path
.push(".testdir");
409 if let Err(_e
) = std
::fs
::remove_dir_all(".testdir") { /* ignore */ }
411 let chunk_store
= ChunkStore
::open("test", &path
);
412 assert
!(chunk_store
.is_err());
414 let chunk_store
= ChunkStore
::create("test", &path
).unwrap();
415 let (exists
, _
, _
) = chunk_store
.insert_chunk(&[0u8, 1u8]).unwrap();
418 let (exists
, _
, _
) = chunk_store
.insert_chunk(&[0u8, 1u8]).unwrap();
422 let chunk_store
= ChunkStore
::create("test", &path
);
423 assert
!(chunk_store
.is_err());
425 if let Err(_e
) = std
::fs
::remove_dir_all(".testdir") { /* ignore */ }