/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 (permit63, os_file63) = 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> {
    // TODO: add a test for #2051: deadlock with large number of files
    let _permit = get_permit().await?0;
    tokio::fs::symlink(src, dst).await.map_err(Into::into)
}

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-18 18:43

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	88	fn as_ref(&self) -> &tokio::fs::File {
46	88	&self.inner
47	88	}
48		}
49
50		impl AsMut<tokio::fs::File> for FileSlot {
51	0	fn as_mut(&mut self) -> &mut tokio::fs::File {
52	0	&mut self.inner
53	0	}
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	796	pub async fn call_with_permit<F, T>(f: F) -> Result<T, Error>
132	796	where
133	796	F: FnOnce(SemaphorePermit<'static>) -> Result<T, Error> + Send + 'static,
134	796	T: Send + 'static,
135	796	{
136	796	let permit = get_permit().await ?0 ;
137	796	spawn_blocking!("fs_call_with_permit", move \|\| f(permit))
138	796	.await
139	794	.unwrap_or_else(\|e\| Err( make_err!0 ( Code::Internal0 , "background task failed: {e:?}")))
140	794	}
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	18	pub fn get_open_files_for_test() -> usize {
204	18	OPEN_FILE_LIMIT.load(Ordering::Acquire) - OPEN_FILE_SEMAPHORE.available_permits()
205	18	}
206
207	65	pub async fn open_file(
208	65	path: impl AsRef<Path>,
209	65	start: u64,
210	65	limit: u64,
211	65	) -> Result<Take<FileSlot>, Error> {
212	65	let path = path.as_ref().to_owned();
213	65	let ( permit63 , os_file63 ) = call_with_permit(move \|permit\| {
214	63	let mut os_file =
215	65	std::fs::File::open(&path).err_tip(\|\| format!( "Could not open {}"2 , path.display()2 )) ?2 ;
216	63	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: 1] 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	63	}
221	63	Ok((permit, os_file))
222	65	})
223	65	.await ?2 ;
224	63	Ok(FileSlot {
225	63	_permit: permit,
226	63	inner: tokio::fs::File::from_std(os_file),
227	63	}
228	63	.take(limit))
229	65	}
230
231	88	pub async fn create_file(path: impl AsRef<Path>) -> Result<FileSlot, Error> {
232	88	let path = path.as_ref().to_owned();
233	88	let (permit, os_file) = call_with_permit(move \|permit\| {
234		Ok((
235	88	permit,
236	88	std::fs::File::options()
237	88	.read(true)
238	88	.write(true)
239	88	.create(true)
240	88	.truncate(true)
241	88	.open(&path)
242	88	.err_tip(\|\| format!( "Could not open {}"0 , path.display()0 )) ?0 ,
243		))
244	88	})
245	88	.await ?0 ;
246	88	Ok(FileSlot {
247	88	_permit: permit,
248	88	inner: tokio::fs::File::from_std(os_file),
249	88	})
250	88	}
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	241	pub async fn create_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
270	241	let path = path.as_ref().to_owned();
271	241	call_with_permit(move \|_\| std::fs::create_dir_all(path).map_err(Into::<Error>::into)).await
272	241	}
273
274		#[cfg(target_family = "unix")]
275	1	pub async fn symlink(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> Result<(), Error> {
276		// TODO: add a test for #2051: deadlock with large number of files
277	1	let _permit = get_permit().await ?0 ;
278	1	tokio::fs::symlink(src, dst).await.map_err(Into::into)
279	1	}
280
281	2	pub async fn read_link(path: impl AsRef<Path>) -> Result<PathBuf, Error> {
282	2	let path = path.as_ref().to_owned();
283	2	call_with_permit(move \|_\| std::fs::read_link(path).map_err(Into::<Error>::into)).await
284	2	}
285
286		#[derive(Debug)]
287		pub struct ReadDir {
288		// We hold the permit because once it is dropped it goes back into the queue.
289		permit: SemaphorePermit<'static>,
290		inner: tokio::fs::ReadDir,
291		}
292
293		impl ReadDir {
294	284	pub fn into_inner(self) -> (SemaphorePermit<'static>, tokio::fs::ReadDir) {
295	284	(self.permit, self.inner)
296	284	}
297		}
298
299		impl AsRef<tokio::fs::ReadDir> for ReadDir {
300	0	fn as_ref(&self) -> &tokio::fs::ReadDir {
301	0	&self.inner
302	0	}
303		}
304
305		impl AsMut<tokio::fs::ReadDir> for ReadDir {
306	0	fn as_mut(&mut self) -> &mut tokio::fs::ReadDir {
307	0	&mut self.inner
308	0	}
309		}
310
311	286	pub async fn read_dir(path: impl AsRef<Path>) -> Result<ReadDir, Error> {
312	286	let path = path.as_ref().to_owned();
313	286	let (permit, inner) = call_with_permit(move \|permit\| {
314		Ok((
315	286	permit,
316	286	tokio::runtime::Handle::current()
317	286	.block_on(tokio::fs::read_dir(path))
318	286	.map_err(Into::<Error>::into) ?0 ,
319		))
320	286	})
321	286	.await ?0 ;
322	286	Ok(ReadDir { permit, inner })
323	286	}
324
325	9	pub async fn rename(from: impl AsRef<Path>, to: impl AsRef<Path>) -> Result<(), Error> {
326	9	let from = from.as_ref().to_owned();
327	9	let to = to.as_ref().to_owned();
328	9	call_with_permit(move \|_\| std::fs::rename(from, to).map_err(Into::<Error>::into)).await
329	9	}
330
331	7	pub async fn remove_file(path: impl AsRef<Path>) -> Result<(), Error> {
332	7	let path = path.as_ref().to_owned();
333	7	call_with_permit(move \|_\| std::fs::remove_file(path).map_err(Into::<Error>::into)).await
334	5	}
335
336	2	pub async fn canonicalize(path: impl AsRef<Path>) -> Result<PathBuf, Error> {
337	2	let path = path.as_ref().to_owned();
338	2	call_with_permit(move \|_\| std::fs::canonicalize(path).map_err(Into::<Error>::into)).await
339	2	}
340
341	15	pub async fn metadata(path: impl AsRef<Path>) -> Result<Metadata, Error> {
342	15	let path = path.as_ref().to_owned();
343	15	call_with_permit(move \|_\| std::fs::metadata(path).map_err(Into::<Error>::into)).await
344	15	}
345
346	4	pub async fn read(path: impl AsRef<Path>) -> Result<Vec<u8>, Error> {
347	4	let path = path.as_ref().to_owned();
348	4	call_with_permit(move \|_\| std::fs::read(path).map_err(Into::<Error>::into)).await
349	4	}
350
351	8	pub async fn symlink_metadata(path: impl AsRef<Path>) -> Result<Metadata, Error> {
352	8	let path = path.as_ref().to_owned();
353	8	call_with_permit(move \|_\| std::fs::symlink_metadata(path).map_err(Into::<Error>::into)).await
354	8	}
355
356		// We can't just use the stock remove_dir_all as it falls over if someone's set readonly
357		// permissions. This version walks the directories and fixes the permissions where needed
358		// before deleting everything.
359		#[cfg(not(target_family = "windows"))]
360	23	fn internal_remove_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
361		// Because otherwise Windows builds complain about these things not being used
362		use std::io::ErrorKind;
363		use std::os::unix::fs::PermissionsExt;
364
365		use tracing::debug;
366		use walkdir::WalkDir;
367
368	24	for entry in WalkDir::new23 ( &path23 ) {
369	24	let Ok( entry23 ) = &entry else { Branch (369:13): [Folded - Ignored] Branch (369:13): [True: 1, False: 0] Branch (369:13): [True: 1, False: 1] Branch (369:13): [True: 1, False: 0] Branch (369:13): [True: 0, False: 0] Branch (369:13): [Folded - Ignored] Branch (369:13): [True: 20, False: 0]
370	1	debug!("Can't get into {entry:?}, assuming already deleted");
371	1	continue;
372		};
373	23	let metadata = entry.metadata() ?0 ;
374	23	if metadata.is_dir() { Branch (374:12): [Folded - Ignored] Branch (374:12): [True: 1, False: 0] Branch (374:12): [True: 1, False: 0] Branch (374:12): [True: 1, False: 0] Branch (374:12): [True: 0, False: 0] Branch (374:12): [Folded - Ignored] Branch (374:12): [True: 20, False: 0]
375	23	match std::fs::remove_dir_all(entry.path()) {
376	22	Ok(()) => {}
377	1	Err(e) if e.kind() == ErrorKind::PermissionDenied => { Branch (377:27): [Folded - Ignored] Branch (377:27): [True: 0, False: 0] Branch (377:27): [True: 1, False: 0] Branch (377:27): [True: 0, False: 0] Branch (377:27): [True: 0, False: 0] Branch (377:27): [Folded - Ignored] Branch (377:27): [True: 0, False: 0]
378	1	std::fs::set_permissions(entry.path(), Permissions::from_mode(0o700)).err_tip(
379	0	\|\| format!("Setting permissions for {}", entry.path().display()),
380	0	)?;
381		}
382	0	e @ Err(_) => e.err_tip(\|\| format!("Removing {}", entry.path().display()))?,
383		}
384	0	} else if metadata.is_file() { Branch (384:19): [Folded - Ignored] Branch (384:19): [True: 0, False: 0] Branch (384:19): [True: 0, False: 0] Branch (384:19): [True: 0, False: 0] Branch (384:19): [True: 0, False: 0] Branch (384:19): [Folded - Ignored] Branch (384:19): [True: 0, False: 0]
385	0	std::fs::set_permissions(entry.path(), Permissions::from_mode(0o600))
386	0	.err_tip(\|\| format!("Setting permissions for {}", entry.path().display()))?;
387	0	}
388		}
389
390		// should now be safe to delete after we fixed all the permissions in the walk loop
391	23	match std::fs::remove_dir_all(&path) {
392	1	Ok(()) => {}
393	22	Err(e) if e.kind() == ErrorKind::NotFound => {} Branch (393:19): [Folded - Ignored] Branch (393:19): [True: 1, False: 0] Branch (393:19): [True: 0, False: 0] Branch (393:19): [True: 1, False: 0] Branch (393:19): [True: 0, False: 0] Branch (393:19): [Folded - Ignored] Branch (393:19): [True: 20, False: 0]
394	0	e @ Err(_) => e.err_tip(\|\| {
395	0	format!(
396	0	"Removing {} after permissions fixes",
397	0	path.as_ref().display()
398		)
399	0	})?,
400		}
401	23	Ok(())
402	23	}
403
404		// We can't set the permissions easily in Windows, so just fallback to
405		// the stock Rust remove_dir_all
406		#[cfg(target_family = "windows")]
407		fn internal_remove_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
408		std::fs::remove_dir_all(&path)?;
409		Ok(())
410		}
411
412	23	pub async fn remove_dir_all(path: impl AsRef<Path>) -> Result<(), Error> {
413	23	let path = path.as_ref().to_owned();
414	23	call_with_permit(move \|_\| internal_remove_dir_all(path)).await
415	23	}