2 use std
::os
::unix
::io
::AsRawFd
;
3 use std
::path
::{Path, PathBuf}
;
4 use std
::sync
::{Arc, Mutex}
;
6 use anyhow
::{bail, format_err, Error}
;
8 use proxmox
::tools
::fs
::{CreateOptions, create_path, create_dir}
;
9 use proxmox_sys
::process_locker
::{ProcessLocker, ProcessLockSharedGuard, ProcessLockExclusiveGuard}
;
10 use proxmox_sys
::worker_task_context
::WorkerTaskContext
;
11 use proxmox_sys
::task_log
;
12 use pbs_api_types
::GarbageCollectionStatus
;
17 /// File system based chunk store
18 pub struct ChunkStore
{
19 name
: String
, // used for error reporting
20 pub (crate) base
: PathBuf
,
23 locker
: Arc
<Mutex
<ProcessLocker
>>,
26 // TODO: what about sysctl setting vm.vfs_cache_pressure (0 - 100) ?
28 pub fn verify_chunk_size(size
: usize) -> Result
<(), Error
> {
30 static SIZES
: [usize; 7] = [64*1024, 128*1024, 256*1024, 512*1024, 1024*1024, 2048*1024, 4096*1024];
32 if !SIZES
.contains(&size
) {
33 bail
!("Got unsupported chunk size '{}'", size
);
38 fn digest_to_prefix(digest
: &[u8]) -> PathBuf
{
40 let mut buf
= Vec
::<u8>::with_capacity(2+1+2+1);
42 const HEX_CHARS
: &[u8; 16] = b
"0123456789abcdef";
44 buf
.push(HEX_CHARS
[(digest
[0] as usize) >> 4]);
45 buf
.push(HEX_CHARS
[(digest
[0] as usize) &0xf]);
46 buf
.push(HEX_CHARS
[(digest
[1] as usize) >> 4]);
47 buf
.push(HEX_CHARS
[(digest
[1] as usize) & 0xf]);
50 let path
= unsafe { String::from_utf8_unchecked(buf)}
;
57 fn chunk_dir
<P
: AsRef
<Path
>>(path
: P
) -> PathBuf
{
59 let mut chunk_dir
: PathBuf
= PathBuf
::from(path
.as_ref());
60 chunk_dir
.push(".chunks");
65 pub fn base(&self) -> &Path
{
69 pub fn create
<P
>(name
: &str, path
: P
, uid
: nix
::unistd
::Uid
, gid
: nix
::unistd
::Gid
, worker
: Option
<&dyn WorkerTaskContext
>) -> Result
<Self, Error
>
74 let base
: PathBuf
= path
.into();
76 if !base
.is_absolute() {
77 bail
!("expected absolute path - got {:?}", base
);
80 let chunk_dir
= Self::chunk_dir(&base
);
82 let options
= CreateOptions
::new()
86 let default_options
= CreateOptions
::new();
88 match create_path(&base
, Some(default_options
), Some(options
.clone())) {
89 Err(err
) => bail
!("unable to create chunk store '{}' at {:?} - {}", name
, base
, err
),
90 Ok(res
) => if ! res { nix::unistd::chown(&base, Some(uid), Some(gid))? }
,
93 if let Err(err
) = create_dir(&chunk_dir
, options
.clone()) {
94 bail
!("unable to create chunk store '{}' subdir {:?} - {}", name
, chunk_dir
, err
);
97 // create lock file with correct owner/group
98 let lockfile_path
= Self::lockfile_path(&base
);
99 proxmox
::tools
::fs
::replace_file(lockfile_path
, b
"", options
.clone(), false)?
;
101 // create 64*1024 subdirs
102 let mut last_percentage
= 0;
104 for i
in 0..64*1024 {
105 let mut l1path
= chunk_dir
.clone();
106 l1path
.push(format
!("{:04x}", i
));
107 if let Err(err
) = create_dir(&l1path
, options
.clone()) {
108 bail
!("unable to create chunk store '{}' subdir {:?} - {}", name
, l1path
, err
);
110 let percentage
= (i
*100)/(64*1024);
111 if percentage
!= last_percentage
{
112 if let Some(worker
) = worker
{
113 task_log
!(worker
, "Chunkstore create: {}%", percentage
)
115 last_percentage
= percentage
;
119 Self::open(name
, base
)
122 fn lockfile_path
<P
: Into
<PathBuf
>>(base
: P
) -> PathBuf
{
123 let mut lockfile_path
: PathBuf
= base
.into();
125 lockfile_path
.push(".lock");
130 pub fn open
<P
: Into
<PathBuf
>>(name
: &str, base
: P
) -> Result
<Self, Error
> {
132 let base
: PathBuf
= base
.into();
134 if !base
.is_absolute() {
135 bail
!("expected absolute path - got {:?}", base
);
138 let chunk_dir
= Self::chunk_dir(&base
);
140 if let Err(err
) = std
::fs
::metadata(&chunk_dir
) {
141 bail
!("unable to open chunk store '{}' at {:?} - {}", name
, chunk_dir
, err
);
144 let lockfile_path
= Self::lockfile_path(&base
);
146 let locker
= ProcessLocker
::new(&lockfile_path
)?
;
149 name
: name
.to_owned(),
153 mutex
: Mutex
::new(())
157 pub fn touch_chunk(&self, digest
: &[u8; 32]) -> Result
<(), Error
> {
158 self.cond_touch_chunk(digest
, true)?
;
162 pub fn cond_touch_chunk(&self, digest
: &[u8; 32], fail_if_not_exist
: bool
) -> Result
<bool
, Error
> {
163 let (chunk_path
, _digest_str
) = self.chunk_path(digest
);
164 self.cond_touch_path(&chunk_path
, fail_if_not_exist
)
167 pub fn cond_touch_path(&self, path
: &Path
, fail_if_not_exist
: bool
) -> Result
<bool
, Error
> {
168 const UTIME_NOW
: i64 = (1 << 30) - 1;
169 const UTIME_OMIT
: i64 = (1 << 30) - 2;
171 let times
: [libc
::timespec
; 2] = [
172 libc
::timespec { tv_sec: 0, tv_nsec: UTIME_NOW }
,
173 libc
::timespec { tv_sec: 0, tv_nsec: UTIME_OMIT }
178 let res
= path
.with_nix_path(|cstr
| unsafe {
179 let tmp
= libc
::utimensat(-1, cstr
.as_ptr(), ×
[0], libc
::AT_SYMLINK_NOFOLLOW
);
180 nix
::errno
::Errno
::result(tmp
)
183 if let Err(err
) = res
{
184 if !fail_if_not_exist
&& err
.as_errno() == Some(nix
::errno
::Errno
::ENOENT
) {
188 bail
!("update atime failed for chunk/file {:?} - {}", path
, err
);
194 pub fn get_chunk_iterator(
197 impl Iterator
<Item
= (Result
<pbs_tools
::fs
::ReadDirEntry
, Error
>, usize, bool
)> + std
::iter
::FusedIterator
,
201 use nix
::fcntl
::OFlag
;
202 use nix
::sys
::stat
::Mode
;
204 let base_handle
= Dir
::open(&self.chunk_dir
, OFlag
::O_RDONLY
, Mode
::empty())
207 "unable to open store '{}' chunk dir {:?} - {}",
214 let mut done
= false;
215 let mut inner
: Option
<pbs_tools
::fs
::ReadDir
> = None
;
217 let mut percentage
= 0;
218 Ok(std
::iter
::from_fn(move || {
224 if let Some(ref mut inner
) = inner
{
227 // skip files if they're not a hash
228 let bytes
= entry
.file_name().to_bytes();
229 if bytes
.len() != 64 && bytes
.len() != 64 + ".0.bad".len() {
232 if !bytes
.iter().take(64).all(u8::is_ascii_hexdigit
) {
236 let bad
= bytes
.ends_with(b
".bad");
237 return Some((Ok(entry
), percentage
, bad
));
240 // stop after first error
242 // and pass the error through:
243 return Some((Err(err
), percentage
, false));
245 None
=> (), // open next directory
256 let subdir
: &str = &format
!("{:04x}", at
);
257 percentage
= (at
* 100) / 0x10000;
259 match pbs_tools
::fs
::read_subdir(base_handle
.as_raw_fd(), subdir
) {
265 Err(ref err
) if err
.as_errno() == Some(nix
::errno
::Errno
::ENOENT
) => {
266 // non-existing directories are okay, just keep going:
270 // other errors are fatal, so end our iteration
272 // and pass the error through:
273 return Some((Err(format_err
!("unable to read subdir '{}' - {}", subdir
, err
)), percentage
, false));
280 pub fn oldest_writer(&self) -> Option
<i64> {
281 ProcessLocker
::oldest_shared_lock(self.locker
.clone())
284 pub fn sweep_unused_chunks(
287 phase1_start_time
: i64,
288 status
: &mut GarbageCollectionStatus
,
289 worker
: &dyn WorkerTaskContext
,
290 ) -> Result
<(), Error
> {
291 use nix
::sys
::stat
::fstatat
;
292 use nix
::unistd
::{unlinkat, UnlinkatFlags}
;
294 let mut min_atime
= phase1_start_time
- 3600*24; // at least 24h (see mount option relatime)
296 if oldest_writer
< min_atime
{
297 min_atime
= oldest_writer
;
300 min_atime
-= 300; // add 5 mins gap for safety
302 let mut last_percentage
= 0;
303 let mut chunk_count
= 0;
305 for (entry
, percentage
, bad
) in self.get_chunk_iterator()?
{
306 if last_percentage
!= percentage
{
307 last_percentage
= percentage
;
310 "processed {}% ({} chunks)",
316 worker
.check_abort()?
;
317 worker
.fail_on_shutdown()?
;
319 let (dirfd
, entry
) = match entry
{
320 Ok(entry
) => (entry
.parent_fd(), entry
),
321 Err(err
) => bail
!("chunk iterator on chunk store '{}' failed - {}", self.name
, err
),
324 let file_type
= match entry
.file_type() {
325 Some(file_type
) => file_type
,
326 None
=> bail
!("unsupported file system type on chunk store '{}'", self.name
),
328 if file_type
!= nix
::dir
::Type
::File
{
334 let filename
= entry
.file_name();
336 let lock
= self.mutex
.lock();
338 if let Ok(stat
) = fstatat(dirfd
, filename
, nix
::fcntl
::AtFlags
::AT_SYMLINK_NOFOLLOW
) {
339 if stat
.st_atime
< min_atime
{
340 //let age = now - stat.st_atime;
341 //println!("UNLINK {} {:?}", age/(3600*24), filename);
342 if let Err(err
) = unlinkat(Some(dirfd
), filename
, UnlinkatFlags
::NoRemoveDir
) {
344 status
.still_bad
+= 1;
347 "unlinking chunk {:?} failed on store '{}' - {}",
354 status
.removed_bad
+= 1;
356 status
.removed_chunks
+= 1;
358 status
.removed_bytes
+= stat
.st_size
as u64;
359 } else if stat
.st_atime
< oldest_writer
{
361 status
.still_bad
+= 1;
363 status
.pending_chunks
+= 1;
365 status
.pending_bytes
+= stat
.st_size
as u64;
368 status
.disk_chunks
+= 1;
370 status
.disk_bytes
+= stat
.st_size
as u64;
383 ) -> Result
<(bool
, u64), Error
> {
385 //println!("DIGEST {}", proxmox::tools::digest_to_hex(digest));
387 let (chunk_path
, digest_str
) = self.chunk_path(digest
);
389 let lock
= self.mutex
.lock();
391 if let Ok(metadata
) = std
::fs
::metadata(&chunk_path
) {
392 if metadata
.is_file() {
393 self.touch_chunk(digest
)?
;
394 return Ok((true, metadata
.len()));
396 bail
!("Got unexpected file type on store '{}' for chunk {}", self.name
, digest_str
);
400 let mut tmp_path
= chunk_path
.clone();
401 tmp_path
.set_extension("tmp");
403 let mut file
= std
::fs
::File
::create(&tmp_path
).map_err(|err
| {
405 "creating temporary chunk on store '{}' failed for {} - {}",
412 let raw_data
= chunk
.raw_data();
413 let encoded_size
= raw_data
.len() as u64;
415 file
.write_all(raw_data
).map_err(|err
| {
417 "writing temporary chunk on store '{}' failed for {} - {}",
424 if let Err(err
) = std
::fs
::rename(&tmp_path
, &chunk_path
) {
425 if std
::fs
::remove_file(&tmp_path
).is_err() { /* ignore */ }
427 "Atomic rename on store '{}' failed for chunk {} - {}",
436 Ok((false, encoded_size
))
439 pub fn chunk_path(&self, digest
:&[u8; 32]) -> (PathBuf
, String
) {
440 let mut chunk_path
= self.chunk_dir
.clone();
441 let prefix
= digest_to_prefix(digest
);
442 chunk_path
.push(&prefix
);
443 let digest_str
= proxmox
::tools
::digest_to_hex(digest
);
444 chunk_path
.push(&digest_str
);
445 (chunk_path
, digest_str
)
448 pub fn relative_path(&self, path
: &Path
) -> PathBuf
{
450 let mut full_path
= self.base
.clone();
451 full_path
.push(path
);
455 pub fn name(&self) -> &str {
459 pub fn base_path(&self) -> PathBuf
{
463 pub fn try_shared_lock(&self) -> Result
<ProcessLockSharedGuard
, Error
> {
464 ProcessLocker
::try_shared_lock(self.locker
.clone())
467 pub fn try_exclusive_lock(&self) -> Result
<ProcessLockExclusiveGuard
, Error
> {
468 ProcessLocker
::try_exclusive_lock(self.locker
.clone())
474 fn test_chunk_store1() {
476 let mut path
= std
::fs
::canonicalize(".").unwrap(); // we need absolute path
477 path
.push(".testdir");
479 if let Err(_e
) = std
::fs
::remove_dir_all(".testdir") { /* ignore */ }
481 let chunk_store
= ChunkStore
::open("test", &path
);
482 assert
!(chunk_store
.is_err());
484 let user
= nix
::unistd
::User
::from_uid(nix
::unistd
::Uid
::current()).unwrap().unwrap();
485 let chunk_store
= ChunkStore
::create("test", &path
, user
.uid
, user
.gid
, None
).unwrap();
487 let (chunk
, digest
) = crate::data_blob
::DataChunkBuilder
::new(&[0u8, 1u8]).build().unwrap();
489 let (exists
, _
) = chunk_store
.insert_chunk(&chunk
, &digest
).unwrap();
492 let (exists
, _
) = chunk_store
.insert_chunk(&chunk
, &digest
).unwrap();
496 let chunk_store
= ChunkStore
::create("test", &path
, user
.uid
, user
.gid
, None
);
497 assert
!(chunk_store
.is_err());
499 if let Err(_e
) = std
::fs
::remove_dir_all(".testdir") { /* ignore */ }