use std::{
fs::{self, File, OpenOptions},
io::{self, Read, Seek, SeekFrom, Write},
panic::{self, AssertUnwindSafe},
path::{Path, PathBuf},
sync::Arc,
};
use crate::{
durability::requires_parent_dir_sync_after_rename,
error::{Error, Result},
options::DurabilityMode,
storage::StorageReadBuffer,
};
use super::{
PLATFORM_IO_FAILED, PlatformIoBackendKind, PlatformIoBackendMatrix, PlatformIoParentCreation,
PlatformIoPublishPlan, PlatformIoTaskClass, ScheduledPlatformIoTask,
};
#[cfg_attr(feature = "platform-io-native", allow(dead_code))]
pub(super) fn matrix() -> PlatformIoBackendMatrix {
#[cfg(all(
any(unix, windows),
not(all(target_arch = "wasm32", target_os = "unknown"))
))]
{
use PlatformIoTaskClass::ThreadPoolManagedAsync;
PlatformIoBackendMatrix {
kind: PlatformIoBackendKind::ThreadPoolManaged,
length_lookup: ThreadPoolManagedAsync,
owned_random_read: ThreadPoolManagedAsync,
optional_whole_object_read: ThreadPoolManagedAsync,
temp_write_rename_publish: ThreadPoolManagedAsync,
append_object_open: ThreadPoolManagedAsync,
append: ThreadPoolManagedAsync,
persist: ThreadPoolManagedAsync,
wal_rewrite: ThreadPoolManagedAsync,
object_delete: ThreadPoolManagedAsync,
directory_create: ThreadPoolManagedAsync,
directory_sync: ThreadPoolManagedAsync,
directory_listing: ThreadPoolManagedAsync,
writer_lease_acquire: ThreadPoolManagedAsync,
}
}
#[cfg(any(
not(any(unix, windows)),
all(target_arch = "wasm32", target_os = "unknown")
))]
{
use PlatformIoTaskClass::Unsupported;
PlatformIoBackendMatrix {
kind: PlatformIoBackendKind::UnsupportedFallback,
length_lookup: Unsupported,
owned_random_read: Unsupported,
optional_whole_object_read: Unsupported,
temp_write_rename_publish: Unsupported,
append_object_open: Unsupported,
append: Unsupported,
persist: Unsupported,
wal_rewrite: Unsupported,
object_delete: Unsupported,
directory_create: Unsupported,
directory_sync: Unsupported,
directory_listing: Unsupported,
writer_lease_acquire: Unsupported,
}
}
}
pub(super) fn run_worker(receiver: crossbeam_channel::Receiver<ScheduledPlatformIoTask>) {
for mut scheduled in receiver {
let task = scheduled.take_task();
let completion = task.failure_completion();
if task.mark_execution_class(scheduled.class).is_err() {
completion.complete();
scheduled
.state
.store(PLATFORM_IO_FAILED, std::sync::atomic::Ordering::Release);
scheduled.finish();
continue;
}
if panic::catch_unwind(AssertUnwindSafe(|| task.run_thread_pool())).is_err() {
completion.complete();
scheduled
.state
.store(PLATFORM_IO_FAILED, std::sync::atomic::Ordering::Release);
}
scheduled.finish();
}
}
pub(super) fn len(path: PathBuf) -> Result<u64> {
Ok(fs::metadata(path)?.len())
}
pub(super) fn read_exact_at_owned(
path: PathBuf,
offset: usize,
len: usize,
) -> Result<StorageReadBuffer> {
let mut file = File::open(path)?;
file.seek(SeekFrom::Start(platform_offset(offset)?))?;
let mut buffer = vec![0; len];
file.read_exact(&mut buffer)?;
Ok(StorageReadBuffer::from_vec(offset, buffer))
}
pub(super) fn read_optional(path: &Path, max_bytes: usize) -> Result<Option<Arc<[u8]>>> {
let file = match File::open(path) {
Ok(file) => file,
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(Error::Io(error)),
};
let len = file.metadata()?.len();
let len = usize::try_from(len).map_err(|_| Error::Corruption {
message: format!("object {} length exceeds usize", path.display()),
})?;
if len > max_bytes {
return Err(Error::Corruption {
message: format!(
"object {} length {len} exceeds maximum {max_bytes}",
path.display()
),
});
}
let read_limit = u64::try_from(max_bytes)
.unwrap_or(u64::MAX)
.saturating_add(1);
let mut bytes = Vec::with_capacity(len);
file.take(read_limit).read_to_end(&mut bytes)?;
if bytes.len() > max_bytes {
return Err(Error::Corruption {
message: format!(
"object {} grew beyond maximum {max_bytes} while it was read",
path.display()
),
});
}
Ok(Some(Arc::from(bytes)))
}
pub(super) fn publish(plan: &PlatformIoPublishPlan) -> Result<()> {
if plan.create_parent == PlatformIoParentCreation::CreateAll
&& let Some(parent) = plan.temporary_path.parent()
{
fs::create_dir_all(parent)?;
}
{
let mut file = File::create(&plan.temporary_path)?;
file.write_all(&plan.bytes)?;
persist_file(&file, plan.durability)?;
}
fs::rename(&plan.temporary_path, &plan.path)?;
if requires_parent_dir_sync_after_rename(plan.durability) {
sync_parent_directory(&plan.path)?;
}
Ok(())
}
pub(super) fn open_append(path: PathBuf) -> Result<()> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
OpenOptions::new()
.append(true)
.create(true)
.open(path)
.map(drop)
.map_err(Error::Io)
}
pub(super) fn append(path: PathBuf, bytes: &[u8], durability: DurabilityMode) -> Result<()> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
let mut file = OpenOptions::new().append(true).create(true).open(path)?;
file.write_all(bytes)?;
persist_file(&file, durability)
}
pub(super) fn persist_path(path: PathBuf, durability: DurabilityMode) -> Result<()> {
if !requires_sync(durability) {
return Ok(());
}
let file = OpenOptions::new().read(true).write(true).open(path)?;
persist_file(&file, durability)
}
pub(super) fn delete_path(path: PathBuf) -> Result<()> {
match fs::remove_file(path) {
Ok(()) => Ok(()),
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(Error::Io(error)),
}
}
pub(super) fn create_dir_all(path: PathBuf) -> Result<()> {
fs::create_dir_all(path).map_err(Error::Io)
}
pub(super) fn list_file_paths(path: PathBuf) -> Result<Vec<PathBuf>> {
let mut paths = Vec::new();
for entry in fs::read_dir(path)? {
let entry = entry?;
if entry.file_type()?.is_file() {
paths.push(entry.path());
}
}
paths.sort_unstable();
Ok(paths)
}
#[cfg(all(
any(unix, windows),
not(all(target_arch = "wasm32", target_os = "unknown"))
))]
pub(super) fn acquire_writer_lease(path: &Path, owner: &[u8]) -> Result<File> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
let mut file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(path)
.map_err(Error::Io)?;
if !fs4::fs_std::FileExt::try_lock_exclusive(&file).map_err(Error::Io)? {
return Err(Error::lease_unavailable(path.display().to_string()));
}
file.set_len(0)?;
file.seek(SeekFrom::Start(0))?;
if let Err(error) = file.write_all(owner) {
return Err(Error::Io(error));
}
if let Err(error) = file.flush() {
return Err(Error::Io(error));
}
Ok(file)
}
fn persist_file(file: &File, durability: DurabilityMode) -> Result<()> {
crate::durability::sync_file_for_durability(file, durability)
}
fn requires_sync(durability: DurabilityMode) -> bool {
crate::durability::requires_file_sync(durability)
}
fn sync_parent_directory(path: &Path) -> Result<()> {
let Some(parent) = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
else {
return Ok(());
};
sync_directory(parent.to_path_buf())
}
#[cfg(all(unix, not(target_os = "macos")))]
pub(super) fn sync_directory(path: PathBuf) -> Result<()> {
let file = File::open(path)?;
file.sync_all().map_err(Error::Io)
}
#[cfg(target_os = "macos")]
pub(super) fn sync_directory(path: PathBuf) -> Result<()> {
let file = File::open(path)?;
file.sync_all().map_err(Error::Io)
}
#[cfg(windows)]
pub(super) fn sync_directory(path: PathBuf) -> Result<()> {
use std::os::windows::fs::OpenOptionsExt;
const FILE_FLAG_BACKUP_SEMANTICS: u32 = 0x0200_0000;
const FILE_SHARE_READ: u32 = 0x0000_0001;
const FILE_SHARE_WRITE: u32 = 0x0000_0002;
const FILE_SHARE_DELETE: u32 = 0x0000_0004;
let file = match OpenOptions::new()
.read(true)
.share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE)
.custom_flags(FILE_FLAG_BACKUP_SEMANTICS)
.open(path)
{
Ok(file) => file,
Err(error) if crate::durability::is_windows_directory_sync_permission_denied(&error) => {
return Ok(());
}
Err(error) => return Err(Error::Io(error)),
};
crate::durability::finish_windows_directory_sync(file.sync_all())
}
#[cfg(not(any(unix, windows)))]
pub(super) fn sync_directory(_path: PathBuf) -> Result<()> {
Err(Error::unsupported_backend(
"platform I/O thread-pool storage",
))
}
fn platform_offset(offset: usize) -> Result<u64> {
u64::try_from(offset).map_err(|_| Error::invalid_options("platform I/O offset overflow"))
}