3 use std
::path
::{Path, PathBuf}
;
5 use std
::sync
::{Arc, Mutex}
;
6 use std
::os
::unix
::io
::AsRawFd
;
11 use crate::server
::WorkerTask
;
13 #[derive(Clone, Serialize)]
14 pub struct GarbageCollectionStatus
{
15 pub upid
: Option
<String
>,
16 pub index_file_count
: usize,
17 pub index_data_bytes
: u64,
19 pub disk_chunks
: usize,
20 pub removed_bytes
: u64,
21 pub removed_chunks
: usize,
24 impl Default
for GarbageCollectionStatus
{
25 fn default() -> Self {
26 GarbageCollectionStatus
{
38 /// File system based chunk store
39 pub struct ChunkStore
{
40 name
: String
, // used for error reporting
41 pub (crate) base
: PathBuf
,
44 locker
: Arc
<Mutex
<tools
::ProcessLocker
>>,
47 // TODO: what about sysctl setting vm.vfs_cache_pressure (0 - 100) ?
49 pub fn verify_chunk_size(size
: usize) -> Result
<(), Error
> {
51 static SIZES
: [usize; 7] = [64*1024, 128*1024, 256*1024, 512*1024, 1024*1024, 2048*1024, 4096*1024];
53 if !SIZES
.contains(&size
) {
54 bail
!("Got unsupported chunk size '{}'", size
);
59 fn digest_to_prefix(digest
: &[u8]) -> PathBuf
{
61 let mut buf
= Vec
::<u8>::with_capacity(2+1+2+1);
63 const HEX_CHARS
: &[u8; 16] = b
"0123456789abcdef";
65 buf
.push(HEX_CHARS
[(digest
[0] as usize) >> 4]);
66 buf
.push(HEX_CHARS
[(digest
[0] as usize) &0xf]);
67 buf
.push(HEX_CHARS
[(digest
[1] as usize) >> 4]);
68 buf
.push(HEX_CHARS
[(digest
[1] as usize) & 0xf]);
71 let path
= unsafe { String::from_utf8_unchecked(buf)}
;
78 fn chunk_dir
<P
: AsRef
<Path
>>(path
: P
) -> PathBuf
{
80 let mut chunk_dir
: PathBuf
= PathBuf
::from(path
.as_ref());
81 chunk_dir
.push(".chunks");
86 pub fn create
<P
: Into
<PathBuf
>>(name
: &str, path
: P
) -> Result
<Self, Error
> {
88 let base
: PathBuf
= path
.into();
90 if !base
.is_absolute() {
91 bail
!("expected absolute path - got {:?}", base
);
94 let chunk_dir
= Self::chunk_dir(&base
);
96 if let Err(err
) = std
::fs
::create_dir_all(&base
) {
97 bail
!("unable to create chunk store '{}' at {:?} - {}", name
, base
, err
);
100 if let Err(err
) = std
::fs
::create_dir_all(&chunk_dir
) {
101 bail
!("unable to create chunk store '{}' subdir {:?} - {}", name
, chunk_dir
, err
);
104 // create 64*1024 subdirs
105 let mut last_percentage
= 0;
107 for i
in 0..64*1024 {
108 let mut l1path
= chunk_dir
.clone();
109 l1path
.push(format
!("{:04x}", i
));
110 if let Err(err
) = std
::fs
::create_dir(&l1path
) {
111 bail
!("unable to create chunk store '{}' subdir {:?} - {}", name
, l1path
, err
);
113 let percentage
= (i
*100)/(64*1024);
114 if percentage
!= last_percentage
{
115 eprintln
!("Percentage done: {}", percentage
);
116 last_percentage
= percentage
;
120 Self::open(name
, base
)
123 pub fn open
<P
: Into
<PathBuf
>>(name
: &str, path
: P
) -> Result
<Self, Error
> {
125 let base
: PathBuf
= path
.into();
127 if !base
.is_absolute() {
128 bail
!("expected absolute path - got {:?}", base
);
131 let chunk_dir
= Self::chunk_dir(&base
);
133 if let Err(err
) = std
::fs
::metadata(&chunk_dir
) {
134 bail
!("unable to open chunk store '{}' at {:?} - {}", name
, chunk_dir
, err
);
137 let mut lockfile_path
= base
.clone();
138 lockfile_path
.push(".lock");
140 let locker
= tools
::ProcessLocker
::new(&lockfile_path
)?
;
143 name
: name
.to_owned(),
147 mutex
: Mutex
::new(false)
151 pub fn touch_chunk(&self, digest
: &[u8; 32]) -> Result
<(), Error
> {
153 let (chunk_path
, _digest_str
) = self.chunk_path(digest
);
155 const UTIME_NOW
: i64 = ((1 << 30) - 1);
156 const UTIME_OMIT
: i64 = ((1 << 30) - 2);
158 let times
: [libc
::timespec
; 2] = [
159 libc
::timespec { tv_sec: 0, tv_nsec: UTIME_NOW }
,
160 libc
::timespec { tv_sec: 0, tv_nsec: UTIME_OMIT }
165 let res
= chunk_path
.with_nix_path(|cstr
| unsafe {
166 libc
::utimensat(-1, cstr
.as_ptr(), ×
[0], libc
::AT_SYMLINK_NOFOLLOW
)
169 if let Err(err
) = nix
::errno
::Errno
::result(res
) {
170 bail
!("updata atime failed for chunk {:?} - {}", chunk_path
, err
);
176 pub fn read_chunk(&self, digest
: &[u8; 32]) -> Result
<DataBlob
, Error
> {
178 let (chunk_path
, digest_str
) = self.chunk_path(digest
);
179 let mut file
= std
::fs
::File
::open(&chunk_path
)
182 "store '{}', unable to read chunk '{}' - {}",
189 DataBlob
::load(&mut file
)
192 pub fn get_chunk_iterator(
195 impl Iterator
<Item
= (Result
<tools
::fs
::ReadDirEntry
, Error
>, usize)> + std
::iter
::FusedIterator
,
199 use nix
::fcntl
::OFlag
;
200 use nix
::sys
::stat
::Mode
;
202 let base_handle
= Dir
::open(&self.chunk_dir
, OFlag
::O_RDONLY
, Mode
::empty())
205 "unable to open store '{}' chunk dir {:?} - {}",
212 let mut done
= false;
213 let mut inner
: Option
<tools
::fs
::ReadDir
> = None
;
215 let mut percentage
= 0;
216 Ok(std
::iter
::from_fn(move || {
222 if let Some(ref mut inner
) = inner
{
225 // skip files if they're not a hash
226 let bytes
= entry
.file_name().to_bytes();
227 if bytes
.len() != 64 {
230 if !bytes
.iter().all(u8::is_ascii_hexdigit
) {
233 return Some((Ok(entry
), percentage
));
236 // stop after first error
238 // and pass the error through:
239 return Some((Err(err
), percentage
));
241 None
=> (), // open next directory
252 let subdir
: &str = &format
!("{:04x}", at
);
253 percentage
= (at
* 100) / 0x10000;
255 match tools
::fs
::read_subdir(base_handle
.as_raw_fd(), subdir
) {
261 Err(ref err
) if err
.as_errno() == Some(nix
::errno
::Errno
::ENOENT
) => {
262 // non-existing directories are okay, just keep going:
266 // other errors are fatal, so end our iteration
268 // and pass the error through:
269 return Some((Err(format_err
!("unable to read subdir '{}' - {}", subdir
, err
)), percentage
));
276 pub fn oldest_writer(&self) -> Option
<i64> {
277 tools
::ProcessLocker
::oldest_shared_lock(self.locker
.clone())
280 pub fn sweep_unused_chunks(
282 oldest_writer
: Option
<i64>,
283 status
: &mut GarbageCollectionStatus
,
284 worker
: Arc
<WorkerTask
>,
285 ) -> Result
<(), Error
> {
286 use nix
::sys
::stat
::fstatat
;
288 let now
= unsafe { libc::time(std::ptr::null_mut()) }
;
290 let mut min_atime
= now
- 3600*24; // at least 24h (see mount option relatime)
292 if let Some(stamp
) = oldest_writer
{
293 if stamp
< min_atime
{
298 min_atime
-= 300; // add 5 mins gap for safety
300 let mut last_percentage
= 0;
301 let mut chunk_count
= 0;
303 for (entry
, percentage
) in self.get_chunk_iterator()?
{
304 if last_percentage
!= percentage
{
305 last_percentage
= percentage
;
306 worker
.log(format
!("percentage done: {}, chunk count: {}", percentage
, chunk_count
));
309 tools
::fail_on_shutdown()?
;
311 let (dirfd
, entry
) = match entry
{
312 Ok(entry
) => (entry
.parent_fd(), entry
),
313 Err(err
) => bail
!("chunk iterator on chunk store '{}' failed - {}", self.name
, err
),
316 let file_type
= match entry
.file_type() {
317 Some(file_type
) => file_type
,
318 None
=> bail
!("unsupported file system type on chunk store '{}'", self.name
),
320 if file_type
!= nix
::dir
::Type
::File
{
326 let filename
= entry
.file_name();
328 let lock
= self.mutex
.lock();
330 if let Ok(stat
) = fstatat(dirfd
, filename
, nix
::fcntl
::AtFlags
::AT_SYMLINK_NOFOLLOW
) {
331 let age
= now
- stat
.st_atime
;
332 //println!("FOUND {} {:?}", age/(3600*24), filename);
333 if stat
.st_atime
< min_atime
{
334 println
!("UNLINK {} {:?}", age
/(3600*24), filename
);
335 let res
= unsafe { libc::unlinkat(dirfd, filename.as_ptr(), 0) }
;
337 let err
= nix
::Error
::last();
339 "unlink chunk {:?} failed on store '{}' - {}",
345 status
.removed_chunks
+= 1;
346 status
.removed_bytes
+= stat
.st_size
as u64;
348 status
.disk_chunks
+= 1;
349 status
.disk_bytes
+= stat
.st_size
as u64;
362 ) -> Result
<(bool
, u64), Error
> {
364 //println!("DIGEST {}", proxmox::tools::digest_to_hex(digest));
366 let (chunk_path
, digest_str
) = self.chunk_path(digest
);
368 let lock
= self.mutex
.lock();
370 if let Ok(metadata
) = std
::fs
::metadata(&chunk_path
) {
371 if metadata
.is_file() {
372 return Ok((true, metadata
.len()));
374 bail
!("Got unexpected file type on store '{}' for chunk {}", self.name
, digest_str
);
378 let mut tmp_path
= chunk_path
.clone();
379 tmp_path
.set_extension("tmp");
381 let mut file
= std
::fs
::File
::create(&tmp_path
)?
;
383 let raw_data
= chunk
.raw_data();
384 let encoded_size
= raw_data
.len() as u64;
386 file
.write_all(raw_data
)?
;
388 if let Err(err
) = std
::fs
::rename(&tmp_path
, &chunk_path
) {
389 if let Err(_
) = std
::fs
::remove_file(&tmp_path
) { /* ignore */ }
391 "Atomic rename on store '{}' failed for chunk {} - {}",
400 Ok((false, encoded_size
))
403 pub fn chunk_path(&self, digest
:&[u8; 32]) -> (PathBuf
, String
) {
404 let mut chunk_path
= self.chunk_dir
.clone();
405 let prefix
= digest_to_prefix(digest
);
406 chunk_path
.push(&prefix
);
407 let digest_str
= proxmox
::tools
::digest_to_hex(digest
);
408 chunk_path
.push(&digest_str
);
409 (chunk_path
, digest_str
)
412 pub fn relative_path(&self, path
: &Path
) -> PathBuf
{
414 let mut full_path
= self.base
.clone();
415 full_path
.push(path
);
419 pub fn base_path(&self) -> PathBuf
{
423 pub fn try_shared_lock(&self) -> Result
<tools
::ProcessLockSharedGuard
, Error
> {
424 tools
::ProcessLocker
::try_shared_lock(self.locker
.clone())
427 pub fn try_exclusive_lock(&self) -> Result
<tools
::ProcessLockExclusiveGuard
, Error
> {
428 tools
::ProcessLocker
::try_exclusive_lock(self.locker
.clone())
434 fn test_chunk_store1() {
436 let mut path
= std
::fs
::canonicalize(".").unwrap(); // we need absulute path
437 path
.push(".testdir");
439 if let Err(_e
) = std
::fs
::remove_dir_all(".testdir") { /* ignore */ }
441 let chunk_store
= ChunkStore
::open("test", &path
);
442 assert
!(chunk_store
.is_err());
444 let chunk_store
= ChunkStore
::create("test", &path
).unwrap();
446 let (chunk
, digest
) = super::DataChunkBuilder
::new(&[0u8, 1u8]).build().unwrap();
448 let (exists
, _
) = chunk_store
.insert_chunk(&chunk
, &digest
).unwrap();
451 let (exists
, _
) = chunk_store
.insert_chunk(&chunk
, &digest
).unwrap();
455 let chunk_store
= ChunkStore
::create("test", &path
);
456 assert
!(chunk_store
.is_err());
458 if let Err(_e
) = std
::fs
::remove_dir_all(".testdir") { /* ignore */ }