Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor: refine device build #745

Merged
merged 3 commits into from
Sep 28, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 2 additions & 6 deletions examples/hybrid.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.

use foyer::{DirectFsDeviceOptionsBuilder, Engine, HybridCache, HybridCacheBuilder};
use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
Expand All @@ -21,11 +21,7 @@ async fn main() -> anyhow::Result<()> {
let hybrid: HybridCache<u64, String> = HybridCacheBuilder::new()
.memory(64 * 1024 * 1024)
.storage(Engine::Large) // use large object disk cache engine only
.with_device_config(
DirectFsDeviceOptionsBuilder::new(dir.path())
.with_capacity(256 * 1024 * 1024)
.build(),
)
.with_device_options(DirectFsDeviceOptions::new(dir.path()).with_capacity(256 * 1024 * 1024))
.build()
.await?;

Expand Down
9 changes: 4 additions & 5 deletions examples/hybrid_full.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ use std::sync::Arc;
use anyhow::Result;
use chrono::Datelike;
use foyer::{
DirectFsDeviceOptionsBuilder, Engine, FifoPicker, HybridCache, HybridCacheBuilder, LargeEngineOptions, LruConfig,
DirectFsDeviceOptions, Engine, FifoPicker, HybridCache, HybridCacheBuilder, LargeEngineOptions, LruConfig,
RateLimitPicker, RecoverMode, RuntimeConfig, SmallEngineOptions, TokioRuntimeConfig, TombstoneLogConfigBuilder,
};
use tempfile::tempdir;
Expand All @@ -36,11 +36,10 @@ async fn main() -> Result<()> {
.with_hash_builder(ahash::RandomState::default())
.with_weighter(|_key, value: &String| value.len())
.storage(Engine::Mixed(0.1))
.with_device_config(
DirectFsDeviceOptionsBuilder::new(dir.path())
.with_device_options(
DirectFsDeviceOptions::new(dir.path())
.with_capacity(64 * 1024 * 1024)
.with_file_size(4 * 1024 * 1024)
.build(),
.with_file_size(4 * 1024 * 1024),
)
.with_flush(true)
.with_recover_mode(RecoverMode::Quiet)
Expand Down
8 changes: 2 additions & 6 deletions examples/tail_based_tracing.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@

use std::time::Duration;

use foyer::{DirectFsDeviceOptionsBuilder, Engine, HybridCache, HybridCacheBuilder};
use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};

#[cfg(feature = "jaeger")]
fn init_jaeger_exporter() {
Expand Down Expand Up @@ -71,11 +71,7 @@ async fn main() -> anyhow::Result<()> {
let hybrid: HybridCache<u64, String> = HybridCacheBuilder::new()
.memory(64 * 1024 * 1024)
.storage(Engine::Large)
.with_device_config(
DirectFsDeviceOptionsBuilder::new(dir.path())
.with_capacity(256 * 1024 * 1024)
.build(),
)
.with_device_options(DirectFsDeviceOptions::new(dir.path()).with_capacity(256 * 1024 * 1024))
.build()
.await?;

Expand Down
20 changes: 9 additions & 11 deletions foyer-bench/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,9 @@
use bytesize::ByteSize;
use clap::{builder::PossibleValuesParser, ArgGroup, Parser};
use foyer::{
Compression, DirectFileDeviceOptionsBuilder, DirectFsDeviceOptionsBuilder, Engine, FifoConfig, FifoPicker,
HybridCache, HybridCacheBuilder, InvalidRatioPicker, LargeEngineOptions, LfuConfig, LruConfig, RateLimitPicker,
RecoverMode, RuntimeConfig, S3FifoConfig, SmallEngineOptions, TokioRuntimeConfig, TracingConfig,
Compression, DirectFileDeviceOptions, DirectFsDeviceOptions, Engine, FifoConfig, FifoPicker, HybridCache,
HybridCacheBuilder, InvalidRatioPicker, LargeEngineOptions, LfuConfig, LruConfig, RateLimitPicker, RecoverMode,
RuntimeConfig, S3FifoConfig, SmallEngineOptions, TokioRuntimeConfig, TracingConfig,
};
use futures::future::join_all;
use itertools::Itertools;
Expand Down Expand Up @@ -462,17 +462,15 @@
.storage(args.engine);

builder = match (args.file.as_ref(), args.dir.as_ref()) {
(Some(file), None) => builder.with_device_config(
DirectFileDeviceOptionsBuilder::new(file)
(Some(file), None) => builder.with_device_options(
DirectFileDeviceOptions::new(file)

Check warning on line 466 in foyer-bench/src/main.rs

View check run for this annotation

Codecov / codecov/patch

foyer-bench/src/main.rs#L465-L466

Added lines #L465 - L466 were not covered by tests
.with_capacity(args.disk.as_u64() as _)
.with_region_size(args.region_size.as_u64() as _)
.build(),
.with_region_size(args.region_size.as_u64() as _),

Check warning on line 468 in foyer-bench/src/main.rs

View check run for this annotation

Codecov / codecov/patch

foyer-bench/src/main.rs#L468

Added line #L468 was not covered by tests
),
(None, Some(dir)) => builder.with_device_config(
DirectFsDeviceOptionsBuilder::new(dir)
(None, Some(dir)) => builder.with_device_options(
DirectFsDeviceOptions::new(dir)
.with_capacity(args.disk.as_u64() as _)
.with_file_size(args.region_size.as_u64() as _)
.build(),
.with_file_size(args.region_size.as_u64() as _),
),
_ => unreachable!(),
};
Expand Down
83 changes: 42 additions & 41 deletions foyer-storage/src/device/direct_file.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,36 +22,21 @@ use foyer_common::{asyncify::asyncify_with_runtime, bits};
use fs4::free_space;
use serde::{Deserialize, Serialize};

use super::{Dev, DevExt, DevOptions, RegionId};
use super::{Dev, DevExt, RegionId};
use crate::{
device::ALIGN,
error::{Error, Result},
IoBytes, IoBytesMut, Runtime,
};

/// Options for the direct file device.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DirectFileDeviceOptions {
/// Path of the direct file device.
pub path: PathBuf,
/// Capacity of the direct file device.
pub capacity: usize,
/// Region size of the direct file device.
pub region_size: usize,
}

/// A device that uses a single direct i/o file.
#[derive(Debug, Clone)]
pub struct DirectFileDevice {
file: Arc<File>,

pub struct DirectFileDeviceConfig {
path: PathBuf,
capacity: usize,
region_size: usize,

runtime: Runtime,
}

impl DevOptions for DirectFileDeviceOptions {
impl DirectFileDeviceConfig {
fn verify(&self) -> Result<()> {
if self.region_size == 0 || self.region_size % ALIGN != 0 {
return Err(anyhow::anyhow!(
Expand All @@ -74,6 +59,17 @@ impl DevOptions for DirectFileDeviceOptions {
}
}

/// A device that uses a single direct i/o file.
#[derive(Debug, Clone)]
pub struct DirectFileDevice {
file: Arc<File>,

capacity: usize,
region_size: usize,

runtime: Runtime,
}

impl DirectFileDevice {
/// Positioned write API for the direct file device.
#[fastrace::trace(name = "foyer::storage::device::direct_file::pwrite")]
Expand Down Expand Up @@ -160,7 +156,7 @@ impl DirectFileDevice {
}

impl Dev for DirectFileDevice {
type Options = DirectFileDeviceOptions;
type Config = DirectFileDeviceConfig;

fn capacity(&self) -> usize {
self.capacity
Expand All @@ -171,7 +167,7 @@ impl Dev for DirectFileDevice {
}

#[fastrace::trace(name = "foyer::storage::device::direct_file::open")]
async fn open(options: Self::Options, runtime: Runtime) -> Result<Self> {
async fn open(options: Self::Config, runtime: Runtime) -> Result<Self> {
options.verify()?;

let dir = options
Expand Down Expand Up @@ -254,19 +250,19 @@ impl Dev for DirectFileDevice {
}
}

/// [`DirectFileDeviceOptionsBuilder`] is used to build the options for the direct fs device.
/// [`DirectFileDeviceOptions`] is used to build the options for the direct fs device.
///
/// The direct fs device uses a directory in a file system to store the data of disk cache.
///
/// It uses direct I/O to reduce buffer copy and page cache pollution if supported.
#[derive(Debug)]
pub struct DirectFileDeviceOptionsBuilder {
pub struct DirectFileDeviceOptions {
path: PathBuf,
capacity: Option<usize>,
region_size: Option<usize>,
}

impl DirectFileDeviceOptionsBuilder {
impl DirectFileDeviceOptions {
const DEFAULT_FILE_SIZE: usize = 64 * 1024 * 1024;

/// Use the given file path as the direct file device path.
Expand Down Expand Up @@ -297,27 +293,31 @@ impl DirectFileDeviceOptionsBuilder {
self.region_size = Some(region_size);
self
}
}

/// Build the options of the direct file device with the given arguments.
pub fn build(self) -> DirectFileDeviceOptions {
let path = self.path;
impl From<DirectFileDeviceOptions> for DirectFileDeviceConfig {
fn from(options: DirectFileDeviceOptions) -> Self {
let path = options.path;

let align_v = |value: usize, align: usize| value - value % align;

let capacity = self.capacity.unwrap_or({
let capacity = options.capacity.unwrap_or({
// Create an empty directory before to get free space.
let dir = path.parent().expect("path must point to a file").to_path_buf();
create_dir_all(&dir).unwrap();
free_space(&dir).unwrap() as usize / 10 * 8
});
let capacity = align_v(capacity, ALIGN);

let region_size = self.region_size.unwrap_or(Self::DEFAULT_FILE_SIZE).min(capacity);
let region_size = options
.region_size
.unwrap_or(DirectFileDeviceOptions::DEFAULT_FILE_SIZE)
.min(capacity);
let region_size = align_v(region_size, ALIGN);

let capacity = align_v(capacity, region_size);

DirectFileDeviceOptions {
DirectFileDeviceConfig {
path,
capacity,
region_size,
Expand All @@ -335,38 +335,39 @@ mod tests {
fn test_options_builder() {
let dir = tempfile::tempdir().unwrap();

let options = DirectFileDeviceOptionsBuilder::new(dir.path().join("test-direct-file")).build();
let config: DirectFileDeviceConfig = DirectFileDeviceOptions::new(dir.path().join("test-direct-file")).into();

tracing::debug!("{options:?}");
tracing::debug!("{config:?}");

options.verify().unwrap();
config.verify().unwrap();
}

#[test_log::test]

fn test_options_builder_noent() {
let dir = tempfile::tempdir().unwrap();

let options = DirectFileDeviceOptionsBuilder::new(dir.path().join("noent").join("test-direct-file")).build();
let config: DirectFileDeviceConfig =
DirectFileDeviceOptions::new(dir.path().join("noent").join("test-direct-file")).into();

tracing::debug!("{options:?}");
tracing::debug!("{config:?}");

options.verify().unwrap();
config.verify().unwrap();
}

#[test_log::test(tokio::test)]
async fn test_direct_file_device_io() {
let dir = tempfile::tempdir().unwrap();
let runtime = Runtime::current();

let options = DirectFileDeviceOptionsBuilder::new(dir.path().join("test-direct-file"))
let config: DirectFileDeviceConfig = DirectFileDeviceOptions::new(dir.path().join("test-direct-file"))
.with_capacity(4 * 1024 * 1024)
.with_region_size(1024 * 1024)
.build();
.into();

tracing::debug!("{options:?}");
tracing::debug!("{config:?}");

let device = DirectFileDevice::open(options.clone(), runtime.clone()).await.unwrap();
let device = DirectFileDevice::open(config.clone(), runtime.clone()).await.unwrap();

let mut buf = IoBytesMut::with_capacity(64 * 1024);
buf.extend(repeat_n(b'x', 64 * 1024 - 100));
Expand All @@ -381,7 +382,7 @@ mod tests {

drop(device);

let device = DirectFileDevice::open(options, runtime).await.unwrap();
let device = DirectFileDevice::open(config, runtime).await.unwrap();

let b = device.read(0, 4096, 64 * 1024 - 100).await.unwrap().freeze();
assert_eq!(buf, b);
Expand Down
Loading
Loading