/build/source/nativelink-util/src/fs.rs

Source
// Copyright 2024 The NativeLink Authors. All rights reserved.
//
// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    See LICENSE file for details
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use core::pin::Pin;
use core::sync::atomic::{AtomicUsize, Ordering};
use core::task::{Context, Poll};
use std::fs::{Metadata, Permissions};
use std::io::{IoSlice, Seek};
use std::path::{Path, PathBuf};

use nativelink_error::{Code, Error, ResultExt, make_err};
use rlimit::increase_nofile_limit;
/// We wrap all `tokio::fs` items in our own wrapper so we can limit the number of outstanding
/// open files at any given time. This will greatly reduce the chance we'll hit open file limit
/// issues.
pub use tokio::fs::DirEntry;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeek, AsyncWrite, ReadBuf, SeekFrom, Take};
use tokio::sync::{Semaphore, SemaphorePermit};
use tracing::{error, info, warn};

use crate::spawn_blocking;

/// Default read buffer size when reading to/from disk.
pub const DEFAULT_READ_BUFF_SIZE: usize = 0x4000;

#[derive(Debug)]
pub struct FileSlot {
    // We hold the permit because once it is dropped it goes back into the queue.
    _permit: SemaphorePermit<'static>,
    inner: tokio::fs::File,
}

impl AsRef<tokio::fs::File> for FileSlot {
    fn as_ref(&self) -> &tokio::fs::File {
        &self.inner
    }
}

impl AsMut<tokio::fs::File> for FileSlot {
    fn as_mut(&mut self) -> &mut tokio::fs::File {
        &mut self.inner
    }
}

impl AsyncRead for FileSlot {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<Result<(), tokio::io::Error>> {
        Pin::new(&mut self.inner).poll_read(cx, buf)
    }
}

impl AsyncSeek for FileSlot {
    fn start_seek(mut self: Pin<&mut Self>, position: SeekFrom) -> Result<(), tokio::io::Error> {
        Pin::new(&mut self.inner).start_seek(position)
    }

    fn poll_complete(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Result<u64, tokio::io::Error>> {
        Pin::new(&mut self.inner).poll_complete(cx)
    }
}

impl AsyncWrite for FileSlot {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, tokio::io::Error>> {
        Pin::new(&mut self.inner).poll_write(cx, buf)
    }

    fn poll_flush(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Result<(), tokio::io::Error>> {
        Pin::new(&mut self.inner).poll_flush(cx)
    }

    fn poll_shutdown(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Result<(), tokio::io::Error>> {
        Pin::new(&mut self.inner).poll_shutdown(cx)
    }

    fn poll_write_vectored(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        bufs: &[IoSlice<'_>],
    ) -> Poll<Result<usize, tokio::io::Error>> {
        Pin::new(&mut self.inner).poll_write_vectored(cx, bufs)
    }

    fn is_write_vectored(&self) -> bool {
        self.inner.is_write_vectored()
    }
}

// Note: If the default changes make sure you update the documentation in
// `config/cas_server.rs`.
pub const DEFAULT_OPEN_FILE_LIMIT: usize = 24 * 1024; // 24k.
static OPEN_FILE_LIMIT: AtomicUsize = AtomicUsize::new(DEFAULT_OPEN_FILE_LIMIT);
pub static OPEN_FILE_SEMAPHORE: Semaphore = Semaphore::const_new(DEFAULT_OPEN_FILE_LIMIT);

/// Try to acquire a permit from the open file semaphore.
#[inline]
pub async fn get_permit() -> Result<SemaphorePermit<'static>, Error> {
    OPEN_FILE_SEMAPHORE
        .acquire()
        .await
        .map_err(|e| make_err!(Code::Internal0, "Open file semaphore closed {:?}", e))
}
/// Acquire a permit from the open file semaphore and call a raw function.
#[inline]
pub async fn call_with_permit<F, T>(f: F) -> Result<T, Error>
where
    F: FnOnce(SemaphorePermit<'static>) -> Result<T, Error> + Send + 'static,
    T: Send + 'static,
{
    let permit = get_permit().await?0;
    spawn_blocking!("fs_call_with_permit", move || f(permit))
        .await
        .unwrap_or_else(|e| Err(make_err!0(Code::Internal0, "background task failed: {e:?}")))
}

/// Sets the soft nofile limit to `desired_open_file_limit` and adjusts
/// `OPEN_FILE_SEMAPHORE` accordingly.
///
/// # Panics
///
/// If any type conversion fails. This can't happen if `usize` is smaller than
/// `u64`.
pub fn set_open_file_limit(desired_open_file_limit: usize) {
    let new_open_file_limit = {
        match increase_nofile_limit(
            u64::try_from(desired_open_file_limit)
                .expect("desired_open_file_limit is too large to convert to u64."),
        ) {
            Ok(open_file_limit) => {
                info!("set_open_file_limit() assigns new open file limit {open_file_limit}.",);
                usize::try_from(open_file_limit)
                    .expect("open_file_limit is too large to convert to usize.")
            }
            Err(e) => {
                error!(
                    "set_open_file_limit() failed to assign open file limit. Maybe system does not have ulimits, continuing anyway. - {e:?}",
                );
                DEFAULT_OPEN_FILE_LIMIT
            }
        }
    };
    // TODO(jaroeichler): Can we give a better estimate?
    if new_open_file_limit < DEFAULT_OPEN_FILE_LIMIT {
        warn!(
            "The new open file limit ({new_open_file_limit}) is below the recommended value of {DEFAULT_OPEN_FILE_LIMIT}. Consider raising max_open_files.",
        );
    }

    // Use only 80% of the open file limit for permits from OPEN_FILE_SEMAPHORE
    // to give extra room for other file descriptors like sockets, pipes, and
    // other things.
    let reduced_open_file_limit = new_open_file_limit.saturating_sub(new_open_file_limit / 5);
    let previous_open_file_limit = OPEN_FILE_LIMIT.load(Ordering::Acquire);
    // No permit should be acquired yet, so this warning should not occur.
    if (OPEN_FILE_SEMAPHORE.available_permits() + reduced_open_file_limit)
        < previous_open_file_limit
    {
        warn!(
            "There are not enough available permits to remove {previous_open_file_limit} - {reduced_open_file_limit} permits.",
        );
    }
    if previous_open_file_limit <= reduced_open_file_limit {
        OPEN_FILE_LIMIT.fetch_add(
            reduced_open_file_limit - previous_open_file_limit,
            Ordering::Release,
        );
        OPEN_FILE_SEMAPHORE.add_permits(reduced_open_file_limit - previous_open_file_limit);
    } else {
        OPEN_FILE_LIMIT.fetch_sub(
            previous_open_file_limit - reduced_open_file_limit,
            Ordering::Release,
        );
        OPEN_FILE_SEMAPHORE.forget_permits(previous_open_file_limit - reduced_open_file_limit);
    }
}

pub fn get_open_files_for_test() -> usize {
    OPEN_FILE_LIMIT.load(Ordering::Acquire) - OPEN_FILE_SEMAPHORE.available_permits()
}

pub async fn open_file(
    path: impl AsRef<Path>,
    start: u64,
    limit: u64,
) -> Result<Take<FileSlot>, Error> {
    let path = path.as_ref().to_owned();
    let (permit62, os_file62) = call_with_permit(move |permit| {
        let mut os_file =
            std::fs::File::open(&path).err_tip(|| format!("Could not open {}"2, path.display()2))?2;
        if start > 0 {
            os_file
                .seek(SeekFrom::Start(start))
                .err_tip(|| format!("Could not seek to {start} in {}", path.display()))?;
        }
        Ok((permit, os_file))
    })
    .await?2;
    Ok(FileSlot {
        _permit: permit,
        inner: tokio::fs::File::from_std(os_file),
    }
    .take(limit))
}

pub async fn create_file(path: impl AsRef<Path>) -> Result<FileSlot, Error> {
    let path = path.as_ref().to_owned();
    let (permit, os_file) = call_with_permit(move |permit| {
        Ok((
            permit,
            std::fs::File::options()
                .read(true)
                .write(true)
                .create(true)
                .truncate(true)
                .open(&path)
                .err_tip(|| format!("Could not open {}"0, path.display()0))?0,
        ))
    })
    .await?0;
    Ok(FileSlot {
        _permit: permit,
        inner: tokio::fs::File::from_std(os_file),
    })
}

pub async fn hard_link(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> Result<(), Error> {
    let src = src.as_ref().to_owned();
    let dst = dst.as_ref().to_owned();
    call_with_permit(move |_| std::fs::hard_link(src, dst).map_err(Into::<Error>::into)).await
}

pub async fn set_permissions(src: impl AsRef<Path>, perm: Permissions) -> Result<(), Error> {
    let src = src.as_ref().to_owned();
    call_with_permit(move |_| std::fs::set_permissions(src, perm).map_err(Into::<Error>::into))
        .await
}

pub async fn create_dir(path: impl AsRef<Path>) -> Result<(), Error> {
    let path = path.as_ref().to_owned();
    call_with_permit(move |_| std::fs::create_dir(path).map_err(Into::<Error>::into)).await
}

pub async fn create_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
    let path = path.as_ref().to_owned();
    call_with_permit(move |_| std::fs::create_dir_all(path).map_err(Into::<Error>::into)).await
}

#[cfg(target_family = "unix")]
pub async fn symlink(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> Result<(), Error> {
    let src = src.as_ref().to_owned();
    let dst = dst.as_ref().to_owned();
    call_with_permit(move |_| {
        tokio::runtime::Handle::current()
            .block_on(tokio::fs::symlink(src, dst))
            .map_err(Into::<Error>::into)
    })
    .await
}

pub async fn read_link(path: impl AsRef<Path>) -> Result<PathBuf, Error> {
    let path = path.as_ref().to_owned();
    call_with_permit(move |_| std::fs::read_link(path).map_err(Into::<Error>::into)).await
}

#[derive(Debug)]
pub struct ReadDir {
    // We hold the permit because once it is dropped it goes back into the queue.
    permit: SemaphorePermit<'static>,
    inner: tokio::fs::ReadDir,
}

impl ReadDir {
    pub fn into_inner(self) -> (SemaphorePermit<'static>, tokio::fs::ReadDir) {
        (self.permit, self.inner)
    }
}

impl AsRef<tokio::fs::ReadDir> for ReadDir {
    fn as_ref(&self) -> &tokio::fs::ReadDir {
        &self.inner
    }
}

impl AsMut<tokio::fs::ReadDir> for ReadDir {
    fn as_mut(&mut self) -> &mut tokio::fs::ReadDir {
        &mut self.inner
    }
}

pub async fn read_dir(path: impl AsRef<Path>) -> Result<ReadDir, Error> {
    let path = path.as_ref().to_owned();
    let (permit, inner) = call_with_permit(move |permit| {
        Ok((
            permit,
            tokio::runtime::Handle::current()
                .block_on(tokio::fs::read_dir(path))
                .map_err(Into::<Error>::into)?0,
        ))
    })
    .await?0;
    Ok(ReadDir { permit, inner })
}

pub async fn rename(from: impl AsRef<Path>, to: impl AsRef<Path>) -> Result<(), Error> {
    let from = from.as_ref().to_owned();
    let to = to.as_ref().to_owned();
    call_with_permit(move |_| std::fs::rename(from, to).map_err(Into::<Error>::into)).await
}

pub async fn remove_file(path: impl AsRef<Path>) -> Result<(), Error> {
    let path = path.as_ref().to_owned();
    call_with_permit(move |_| std::fs::remove_file(path).map_err(Into::<Error>::into)).await
}

pub async fn canonicalize(path: impl AsRef<Path>) -> Result<PathBuf, Error> {
    let path = path.as_ref().to_owned();
    call_with_permit(move |_| std::fs::canonicalize(path).map_err(Into::<Error>::into)).await
}

pub async fn metadata(path: impl AsRef<Path>) -> Result<Metadata, Error> {
    let path = path.as_ref().to_owned();
    call_with_permit(move |_| std::fs::metadata(path).map_err(Into::<Error>::into)).await
}

pub async fn read(path: impl AsRef<Path>) -> Result<Vec<u8>, Error> {
    let path = path.as_ref().to_owned();
    call_with_permit(move |_| std::fs::read(path).map_err(Into::<Error>::into)).await
}

pub async fn symlink_metadata(path: impl AsRef<Path>) -> Result<Metadata, Error> {
    let path = path.as_ref().to_owned();
    call_with_permit(move |_| std::fs::symlink_metadata(path).map_err(Into::<Error>::into)).await
}

// We can't just use the stock remove_dir_all as it falls over if someone's set readonly
// permissions. This version walks the directories and fixes the permissions where needed
// before deleting everything.
#[cfg(not(target_family = "windows"))]
fn internal_remove_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
    // Because otherwise Windows builds complain about these things not being used
    use std::io::ErrorKind;
    use std::os::unix::fs::PermissionsExt;

    use tracing::debug;
    use walkdir::WalkDir;

    for entry in WalkDir::new23(&path23) {
        let Ok(entry23) = &entry else {
            debug!("Can't get into {entry:?}, assuming already deleted");
            continue;
        };
        let metadata = entry.metadata()?0;
        if metadata.is_dir() {
            match std::fs::remove_dir_all(entry.path()) {
                Ok(()) => {}
                Err(e) if e.kind() == ErrorKind::PermissionDenied => {
                    std::fs::set_permissions(entry.path(), Permissions::from_mode(0o700)).err_tip(
                        || format!("Setting permissions for {}", entry.path().display()),
                    )?;
                }
                e @ Err(_) => e.err_tip(|| format!("Removing {}", entry.path().display()))?,
            }
        } else if metadata.is_file() {
            std::fs::set_permissions(entry.path(), Permissions::from_mode(0o600))
                .err_tip(|| format!("Setting permissions for {}", entry.path().display()))?;
        }
    }

    // should now be safe to delete after we fixed all the permissions in the walk loop
    match std::fs::remove_dir_all(&path) {
        Ok(()) => {}
        Err(e) if e.kind() == ErrorKind::NotFound => {}
        e @ Err(_) => e.err_tip(|| {
            format!(
                "Removing {} after permissions fixes",
                path.as_ref().display()
            )
        })?,
    }
    Ok(())
}

// We can't set the permissions easily in Windows, so just fallback to
// the stock Rust remove_dir_all
#[cfg(target_family = "windows")]
fn internal_remove_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
    std::fs::remove_dir_all(&path)?;
    Ok(())
}

pub async fn remove_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
    let path = path.as_ref().to_owned();
    call_with_permit(move |_| internal_remove_dir_all(path)).await
}

Coverage Report

Created: 2025-11-04 17:54

Line	Count	Source
1		// Copyright 2024 The NativeLink Authors. All rights reserved.
2		//
3		// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
4		// you may not use this file except in compliance with the License.
5		// You may obtain a copy of the License at
6		//
7		// See LICENSE file for details
8		//
9		// Unless required by applicable law or agreed to in writing, software
10		// distributed under the License is distributed on an "AS IS" BASIS,
11		// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12		// See the License for the specific language governing permissions and
13		// limitations under the License.
14
15		use core::pin::Pin;
16		use core::sync::atomic::{AtomicUsize, Ordering};
17		use core::task::{Context, Poll};
18		use std::fs::{Metadata, Permissions};
19		use std::io::{IoSlice, Seek};
20		use std::path::{Path, PathBuf};
21
22		use nativelink_error::{Code, Error, ResultExt, make_err};
23		use rlimit::increase_nofile_limit;
24		/// We wrap all `tokio::fs` items in our own wrapper so we can limit the number of outstanding
25		/// open files at any given time. This will greatly reduce the chance we'll hit open file limit
26		/// issues.
27		pub use tokio::fs::DirEntry;
28		use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeek, AsyncWrite, ReadBuf, SeekFrom, Take};
29		use tokio::sync::{Semaphore, SemaphorePermit};
30		use tracing::{error, info, warn};
31
32		use crate::spawn_blocking;
33
34		/// Default read buffer size when reading to/from disk.
35		pub const DEFAULT_READ_BUFF_SIZE: usize = 0x4000;
36
37		#[derive(Debug)]
38		pub struct FileSlot {
39		// We hold the permit because once it is dropped it goes back into the queue.
40		_permit: SemaphorePermit<'static>,
41		inner: tokio::fs::File,
42		}
43
44		impl AsRef<tokio::fs::File> for FileSlot {
45	122	fn as_ref(&self) -> &tokio::fs::File {
46	122	&self.inner
47	122	}
48		}
49
50		impl AsMut<tokio::fs::File> for FileSlot {
51	2	fn as_mut(&mut self) -> &mut tokio::fs::File {
52	2	&mut self.inner
53	2	}
54		}
55
56		impl AsyncRead for FileSlot {
57	2.57k	fn poll_read(
58	2.57k	mut self: Pin<&mut Self>,
59	2.57k	cx: &mut Context<'_>,
60	2.57k	buf: &mut ReadBuf<'_>,
61	2.57k	) -> Poll<Result<(), tokio::io::Error>> {
62	2.57k	Pin::new(&mut self.inner).poll_read(cx, buf)
63	2.57k	}
64		}
65
66		impl AsyncSeek for FileSlot {
67	19	fn start_seek(mut self: Pin<&mut Self>, position: SeekFrom) -> Result<(), tokio::io::Error> {
68	19	Pin::new(&mut self.inner).start_seek(position)
69	19	}
70
71	57	fn poll_complete(
72	57	mut self: Pin<&mut Self>,
73	57	cx: &mut Context<'_>,
74	57	) -> Poll<Result<u64, tokio::io::Error>> {
75	57	Pin::new(&mut self.inner).poll_complete(cx)
76	57	}
77		}
78
79		impl AsyncWrite for FileSlot {
80	2	fn poll_write(
81	2	mut self: Pin<&mut Self>,
82	2	cx: &mut Context<'_>,
83	2	buf: &[u8],
84	2	) -> Poll<Result<usize, tokio::io::Error>> {
85	2	Pin::new(&mut self.inner).poll_write(cx, buf)
86	2	}
87
88	0	fn poll_flush(
89	0	mut self: Pin<&mut Self>,
90	0	cx: &mut Context<'_>,
91	0	) -> Poll<Result<(), tokio::io::Error>> {
92	0	Pin::new(&mut self.inner).poll_flush(cx)
93	0	}
94
95	0	fn poll_shutdown(
96	0	mut self: Pin<&mut Self>,
97	0	cx: &mut Context<'_>,
98	0	) -> Poll<Result<(), tokio::io::Error>> {
99	0	Pin::new(&mut self.inner).poll_shutdown(cx)
100	0	}
101
102	97	fn poll_write_vectored(
103	97	mut self: Pin<&mut Self>,
104	97	cx: &mut Context<'_>,
105	97	bufs: &[IoSlice<'_>],
106	97	) -> Poll<Result<usize, tokio::io::Error>> {
107	97	Pin::new(&mut self.inner).poll_write_vectored(cx, bufs)
108	97	}
109
110	97	fn is_write_vectored(&self) -> bool {
111	97	self.inner.is_write_vectored()
112	97	}
113		}
114
115		// Note: If the default changes make sure you update the documentation in
116		// `config/cas_server.rs`.
117		pub const DEFAULT_OPEN_FILE_LIMIT: usize = 24 * 1024; // 24k.
118		static OPEN_FILE_LIMIT: AtomicUsize = AtomicUsize::new(DEFAULT_OPEN_FILE_LIMIT);
119		pub static OPEN_FILE_SEMAPHORE: Semaphore = Semaphore::const_new(DEFAULT_OPEN_FILE_LIMIT);
120
121		/// Try to acquire a permit from the open file semaphore.
122		#[inline]
123	25.3k	pub async fn get_permit() -> Result<SemaphorePermit<'static>, Error> {
124	25.3k	OPEN_FILE_SEMAPHORE
125	25.3k	.acquire()
126	25.3k	.await
127	25.3k	.map_err(\|e\| make_err!( Code::Internal0 , "Open file semaphore closed {:?}", e))
128	25.3k	}
129		/// Acquire a permit from the open file semaphore and call a raw function.
130		#[inline]
131	813	pub async fn call_with_permit<F, T>(f: F) -> Result<T, Error>
132	813	where
133	813	F: FnOnce(SemaphorePermit<'static>) -> Result<T, Error> + Send + 'static,
134	813	T: Send + 'static,
135	813	{
136	813	let permit = get_permit().await ?0 ;
137	813	spawn_blocking!("fs_call_with_permit", move \|\| f(permit))
138	813	.await
139	811	.unwrap_or_else(\|e\| Err( make_err!0 ( Code::Internal0 , "background task failed: {e:?}")))
140	811	}
141
142		/// Sets the soft nofile limit to `desired_open_file_limit` and adjusts
143		/// `OPEN_FILE_SEMAPHORE` accordingly.
144		///
145		/// # Panics
146		///
147		/// If any type conversion fails. This can't happen if `usize` is smaller than
148		/// `u64`.
149	0	pub fn set_open_file_limit(desired_open_file_limit: usize) {
150	0	let new_open_file_limit = {
151	0	match increase_nofile_limit(
152	0	u64::try_from(desired_open_file_limit)
153	0	.expect("desired_open_file_limit is too large to convert to u64."),
154	0	) {
155	0	Ok(open_file_limit) => {
156	0	info!("set_open_file_limit() assigns new open file limit {open_file_limit}.",);
157	0	usize::try_from(open_file_limit)
158	0	.expect("open_file_limit is too large to convert to usize.")
159		}
160	0	Err(e) => {
161	0	error!(
162	0	"set_open_file_limit() failed to assign open file limit. Maybe system does not have ulimits, continuing anyway. - {e:?}",
163		);
164	0	DEFAULT_OPEN_FILE_LIMIT
165		}
166		}
167		};
168		// TODO(jaroeichler): Can we give a better estimate?
169	0	if new_open_file_limit < DEFAULT_OPEN_FILE_LIMIT { Branch (169:8): [True: 0, False: 0] Branch (169:8): [Folded - Ignored]
170	0	warn!(
171	0	"The new open file limit ({new_open_file_limit}) is below the recommended value of {DEFAULT_OPEN_FILE_LIMIT}. Consider raising max_open_files.",
172		);
173	0	}
174
175		// Use only 80% of the open file limit for permits from OPEN_FILE_SEMAPHORE
176		// to give extra room for other file descriptors like sockets, pipes, and
177		// other things.
178	0	let reduced_open_file_limit = new_open_file_limit.saturating_sub(new_open_file_limit / 5);
179	0	let previous_open_file_limit = OPEN_FILE_LIMIT.load(Ordering::Acquire);
180		// No permit should be acquired yet, so this warning should not occur.
181	0	if (OPEN_FILE_SEMAPHORE.available_permits() + reduced_open_file_limit) Branch (181:8): [True: 0, False: 0] Branch (181:8): [Folded - Ignored]
182	0	< previous_open_file_limit
183		{
184	0	warn!(
185	0	"There are not enough available permits to remove {previous_open_file_limit} - {reduced_open_file_limit} permits.",
186		);
187	0	}
188	0	if previous_open_file_limit <= reduced_open_file_limit { Branch (188:8): [True: 0, False: 0] Branch (188:8): [Folded - Ignored]
189	0	OPEN_FILE_LIMIT.fetch_add(
190	0	reduced_open_file_limit - previous_open_file_limit,
191	0	Ordering::Release,
192	0	);
193	0	OPEN_FILE_SEMAPHORE.add_permits(reduced_open_file_limit - previous_open_file_limit);
194	0	} else {
195	0	OPEN_FILE_LIMIT.fetch_sub(
196	0	previous_open_file_limit - reduced_open_file_limit,
197	0	Ordering::Release,
198	0	);
199	0	OPEN_FILE_SEMAPHORE.forget_permits(previous_open_file_limit - reduced_open_file_limit);
200	0	}
201	0	}
202
203	20	pub fn get_open_files_for_test() -> usize {
204	20	OPEN_FILE_LIMIT.load(Ordering::Acquire) - OPEN_FILE_SEMAPHORE.available_permits()
205	20	}
206
207	64	pub async fn open_file(
208	64	path: impl AsRef<Path>,
209	64	start: u64,
210	64	limit: u64,
211	64	) -> Result<Take<FileSlot>, Error> {
212	64	let path = path.as_ref().to_owned();
213	64	let ( permit62 , os_file62 ) = call_with_permit(move \|permit\| {
214	62	let mut os_file =
215	64	std::fs::File::open(&path).err_tip(\|\| format!( "Could not open {}"2 , path.display()2 )) ?2 ;
216	62	if start > 0 { Branch (216:12): [True: 0, False: 0] Branch (216:12): [Folded - Ignored] Branch (216:12): [True: 0, False: 1] Branch (216:12): [True: 0, False: 2] Branch (216:12): [True: 0, False: 4] Branch (216:12): [True: 0, False: 6] Branch (216:12): [True: 0, False: 0] Branch (216:12): [True: 0, False: 1] Branch (216:12): [Folded - Ignored] Branch (216:12): [True: 0, False: 0] Branch (216:12): [True: 0, False: 41] Branch (216:12): [True: 0, False: 3] Branch (216:12): [True: 0, False: 4]
217	0	os_file
218	0	.seek(SeekFrom::Start(start))
219	0	.err_tip(\|\| format!("Could not seek to {start} in {}", path.display()))?;
220	62	}
221	62	Ok((permit, os_file))
222	64	})
223	64	.await ?2 ;
224	62	Ok(FileSlot {
225	62	_permit: permit,
226	62	inner: tokio::fs::File::from_std(os_file),
227	62	}
228	62	.take(limit))
229	64	}
230
231	122	pub async fn create_file(path: impl AsRef<Path>) -> Result<FileSlot, Error> {
232	122	let path = path.as_ref().to_owned();
233	122	let (permit, os_file) = call_with_permit(move \|permit\| {
234		Ok((
235	122	permit,
236	122	std::fs::File::options()
237	122	.read(true)
238	122	.write(true)
239	122	.create(true)
240	122	.truncate(true)
241	122	.open(&path)
242	122	.err_tip(\|\| format!( "Could not open {}"0 , path.display()0 )) ?0 ,
243		))
244	122	})
245	122	.await ?0 ;
246	122	Ok(FileSlot {
247	122	_permit: permit,
248	122	inner: tokio::fs::File::from_std(os_file),
249	122	})
250	122	}
251
252	4	pub async fn hard_link(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> Result<(), Error> {
253	4	let src = src.as_ref().to_owned();
254	4	let dst = dst.as_ref().to_owned();
255	4	call_with_permit(move \|_\| std::fs::hard_link(src, dst).map_err(Into::<Error>::into)).await
256	4	}
257
258	1	pub async fn set_permissions(src: impl AsRef<Path>, perm: Permissions) -> Result<(), Error> {
259	1	let src = src.as_ref().to_owned();
260	1	call_with_permit(move \|_\| std::fs::set_permissions(src, perm).map_err(Into::<Error>::into))
261	1	.await
262	1	}
263
264	41	pub async fn create_dir(path: impl AsRef<Path>) -> Result<(), Error> {
265	41	let path = path.as_ref().to_owned();
266	41	call_with_permit(move \|_\| std::fs::create_dir(path).map_err(Into::<Error>::into)).await
267	41	}
268
269	225	pub async fn create_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
270	225	let path = path.as_ref().to_owned();
271	225	call_with_permit(move \|_\| std::fs::create_dir_all(path).map_err(Into::<Error>::into)).await
272	225	}
273
274		#[cfg(target_family = "unix")]
275	1	pub async fn symlink(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> Result<(), Error> {
276	1	let src = src.as_ref().to_owned();
277	1	let dst = dst.as_ref().to_owned();
278	1	call_with_permit(move \|_\| {
279	1	tokio::runtime::Handle::current()
280	1	.block_on(tokio::fs::symlink(src, dst))
281	1	.map_err(Into::<Error>::into)
282	1	})
283	1	.await
284	1	}
285
286	2	pub async fn read_link(path: impl AsRef<Path>) -> Result<PathBuf, Error> {
287	2	let path = path.as_ref().to_owned();
288	2	call_with_permit(move \|_\| std::fs::read_link(path).map_err(Into::<Error>::into)).await
289	2	}
290
291		#[derive(Debug)]
292		pub struct ReadDir {
293		// We hold the permit because once it is dropped it goes back into the queue.
294		permit: SemaphorePermit<'static>,
295		inner: tokio::fs::ReadDir,
296		}
297
298		impl ReadDir {
299	250	pub fn into_inner(self) -> (SemaphorePermit<'static>, tokio::fs::ReadDir) {
300	250	(self.permit, self.inner)
301	250	}
302		}
303
304		impl AsRef<tokio::fs::ReadDir> for ReadDir {
305	0	fn as_ref(&self) -> &tokio::fs::ReadDir {
306	0	&self.inner
307	0	}
308		}
309
310		impl AsMut<tokio::fs::ReadDir> for ReadDir {
311	0	fn as_mut(&mut self) -> &mut tokio::fs::ReadDir {
312	0	&mut self.inner
313	0	}
314		}
315
316	252	pub async fn read_dir(path: impl AsRef<Path>) -> Result<ReadDir, Error> {
317	252	let path = path.as_ref().to_owned();
318	252	let (permit, inner) = call_with_permit(move \|permit\| {
319		Ok((
320	252	permit,
321	252	tokio::runtime::Handle::current()
322	252	.block_on(tokio::fs::read_dir(path))
323	252	.map_err(Into::<Error>::into) ?0 ,
324		))
325	252	})
326	252	.await ?0 ;
327	252	Ok(ReadDir { permit, inner })
328	252	}
329
330	25	pub async fn rename(from: impl AsRef<Path>, to: impl AsRef<Path>) -> Result<(), Error> {
331	25	let from = from.as_ref().to_owned();
332	25	let to = to.as_ref().to_owned();
333	25	call_with_permit(move \|_\| std::fs::rename(from, to).map_err(Into::<Error>::into)).await
334	25	}
335
336	23	pub async fn remove_file(path: impl AsRef<Path>) -> Result<(), Error> {
337	23	let path = path.as_ref().to_owned();
338	23	call_with_permit(move \|_\| std::fs::remove_file(path).map_err(Into::<Error>::into)).await
339	21	}
340
341	2	pub async fn canonicalize(path: impl AsRef<Path>) -> Result<PathBuf, Error> {
342	2	let path = path.as_ref().to_owned();
343	2	call_with_permit(move \|_\| std::fs::canonicalize(path).map_err(Into::<Error>::into)).await
344	2	}
345
346	16	pub async fn metadata(path: impl AsRef<Path>) -> Result<Metadata, Error> {
347	16	let path = path.as_ref().to_owned();
348	16	call_with_permit(move \|_\| std::fs::metadata(path).map_err(Into::<Error>::into)).await
349	16	}
350
351	4	pub async fn read(path: impl AsRef<Path>) -> Result<Vec<u8>, Error> {
352	4	let path = path.as_ref().to_owned();
353	4	call_with_permit(move \|_\| std::fs::read(path).map_err(Into::<Error>::into)).await
354	4	}
355
356	8	pub async fn symlink_metadata(path: impl AsRef<Path>) -> Result<Metadata, Error> {
357	8	let path = path.as_ref().to_owned();
358	8	call_with_permit(move \|_\| std::fs::symlink_metadata(path).map_err(Into::<Error>::into)).await
359	8	}
360
361		// We can't just use the stock remove_dir_all as it falls over if someone's set readonly
362		// permissions. This version walks the directories and fixes the permissions where needed
363		// before deleting everything.
364		#[cfg(not(target_family = "windows"))]
365	23	fn internal_remove_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
366		// Because otherwise Windows builds complain about these things not being used
367		use std::io::ErrorKind;
368		use std::os::unix::fs::PermissionsExt;
369
370		use tracing::debug;
371		use walkdir::WalkDir;
372
373	24	for entry in WalkDir::new23 ( &path23 ) {
374	24	let Ok( entry23 ) = &entry else { Branch (374:13): [Folded - Ignored] Branch (374:13): [True: 1, False: 0] Branch (374:13): [True: 1, False: 1] Branch (374:13): [True: 1, False: 0] Branch (374:13): [True: 0, False: 0] Branch (374:13): [Folded - Ignored] Branch (374:13): [True: 20, False: 0]
375	1	debug!("Can't get into {entry:?}, assuming already deleted");
376	1	continue;
377		};
378	23	let metadata = entry.metadata() ?0 ;
379	23	if metadata.is_dir() { Branch (379:12): [Folded - Ignored] Branch (379:12): [True: 1, False: 0] Branch (379:12): [True: 1, False: 0] Branch (379:12): [True: 1, False: 0] Branch (379:12): [True: 0, False: 0] Branch (379:12): [Folded - Ignored] Branch (379:12): [True: 20, False: 0]
380	23	match std::fs::remove_dir_all(entry.path()) {
381	22	Ok(()) => {}
382	1	Err(e) if e.kind() == ErrorKind::PermissionDenied => { Branch (382:27): [Folded - Ignored] Branch (382:27): [True: 0, False: 0] Branch (382:27): [True: 1, False: 0] Branch (382:27): [True: 0, False: 0] Branch (382:27): [True: 0, False: 0] Branch (382:27): [Folded - Ignored] Branch (382:27): [True: 0, False: 0]
383	1	std::fs::set_permissions(entry.path(), Permissions::from_mode(0o700)).err_tip(
384	0	\|\| format!("Setting permissions for {}", entry.path().display()),
385	0	)?;
386		}
387	0	e @ Err(_) => e.err_tip(\|\| format!("Removing {}", entry.path().display()))?,
388		}
389	0	} else if metadata.is_file() { Branch (389:19): [Folded - Ignored] Branch (389:19): [True: 0, False: 0] Branch (389:19): [True: 0, False: 0] Branch (389:19): [True: 0, False: 0] Branch (389:19): [True: 0, False: 0] Branch (389:19): [Folded - Ignored] Branch (389:19): [True: 0, False: 0]
390	0	std::fs::set_permissions(entry.path(), Permissions::from_mode(0o600))
391	0	.err_tip(\|\| format!("Setting permissions for {}", entry.path().display()))?;
392	0	}
393		}
394
395		// should now be safe to delete after we fixed all the permissions in the walk loop
396	23	match std::fs::remove_dir_all(&path) {
397	1	Ok(()) => {}
398	22	Err(e) if e.kind() == ErrorKind::NotFound => {} Branch (398:19): [Folded - Ignored] Branch (398:19): [True: 1, False: 0] Branch (398:19): [True: 0, False: 0] Branch (398:19): [True: 1, False: 0] Branch (398:19): [True: 0, False: 0] Branch (398:19): [Folded - Ignored] Branch (398:19): [True: 20, False: 0]
399	0	e @ Err(_) => e.err_tip(\|\| {
400	0	format!(
401	0	"Removing {} after permissions fixes",
402	0	path.as_ref().display()
403		)
404	0	})?,
405		}
406	23	Ok(())
407	23	}
408
409		// We can't set the permissions easily in Windows, so just fallback to
410		// the stock Rust remove_dir_all
411		#[cfg(target_family = "windows")]
412		fn internal_remove_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
413		std::fs::remove_dir_all(&path)?;
414		Ok(())
415		}
416
417	23	pub async fn remove_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
418	23	let path = path.as_ref().to_owned();
419	23	call_with_permit(move \|_\| internal_remove_dir_all(path)).await
420	23	}