From 3ac9645b368187d6305d1b1d66bc67ef4892ad30 Mon Sep 17 00:00:00 2001 From: MrCroxx Date: Mon, 26 Oct 2026 15:36:13 +0800 Subject: [PATCH 1/6] feat(services): add 3FS USRBIO support and real service CI --- .env.example | 3 + .github/scripts/test_behavior/plan.py | 10 +- .github/scripts/test_behavior/test_plan.py | 13 + .github/services/hf3fs/README.md | 30 ++ .github/services/hf3fs/single_node/action.yml | 70 +++ .github/services/hf3fs/single_node/setup.sh | 58 +++ .github/workflows/test_behavior_core.yml | 20 +- core/Cargo.lock | 16 + core/Cargo.toml | 2 + core/services/hf3fs/Cargo.toml | 49 ++ core/services/hf3fs/README.md | 68 +++ core/services/hf3fs/examples/bench_read.rs | 81 +++ core/services/hf3fs/src/backend.rs | 264 ++++++++++ core/services/hf3fs/src/config.rs | 55 ++ core/services/hf3fs/src/driver.rs | 473 ++++++++++++++++++ core/services/hf3fs/src/ffi.rs | 135 +++++ core/services/hf3fs/src/lib.rs | 39 ++ core/services/hf3fs/src/reader.rs | 69 +++ core/services/hf3fs/src/writer.rs | 147 ++++++ core/services/hf3fs/tests/real_service.rs | 193 +++++++ core/src/lib.rs | 5 + 21 files changed, 1798 insertions(+), 2 deletions(-) create mode 100644 .github/services/hf3fs/README.md create mode 100644 .github/services/hf3fs/single_node/action.yml create mode 100644 .github/services/hf3fs/single_node/setup.sh create mode 100644 core/services/hf3fs/Cargo.toml create mode 100644 core/services/hf3fs/README.md create mode 100644 core/services/hf3fs/examples/bench_read.rs create mode 100644 core/services/hf3fs/src/backend.rs create mode 100644 core/services/hf3fs/src/config.rs create mode 100644 core/services/hf3fs/src/driver.rs create mode 100644 core/services/hf3fs/src/ffi.rs create mode 100644 core/services/hf3fs/src/lib.rs create mode 100644 core/services/hf3fs/src/reader.rs create mode 100644 core/services/hf3fs/src/writer.rs create mode 100644 core/services/hf3fs/tests/real_service.rs diff --git a/.env.example b/.env.example index 5f07b8c97af5..09266282f92e 100644 --- a/.env.example +++ b/.env.example @@ -2,6 +2,9 @@ # fs OPENDAL_FS_ROOT=/path/to/dir OPENDAL_FS_ATOMIC_WRITE_DIR=/path/to/tempdir +# hf3fs (existing directory on a real 3FS mount) +OPENDAL_HF3FS_ROOT=/path/to/3fs/testdir +OPENDAL_HF3FS_LIBRARY_PATH=/path/to/libhf3fs_api_shared.so # cos OPENDAL_COS_BUCKET=opendal-testing-1318209832 OPENDAL_COS_ENDPOINT=https://cos.ap-singapore.myqcloud.com diff --git a/.github/scripts/test_behavior/plan.py b/.github/scripts/test_behavior/plan.py index 342b611db3c1..5f842fd7517a 100755 --- a/.github/scripts/test_behavior/plan.py +++ b/.github/scripts/test_behavior/plan.py @@ -285,6 +285,9 @@ def generate_language_binding_cases( # Bindings may be treated as parallel requests, so we need to disable it for all languages. cases = [v for v in cases if v["service"] != "aliyun_drive"] + # 3FS currently has a Rust-only fixture that requires Ubuntu 22.04's RXE module. + cases = [v for v in cases if v["service"] != "hf3fs"] + # Remove invalid cases for java. if language == "java": cases = [v for v in cases if v["service"] not in [ @@ -387,7 +390,12 @@ def plan(changed_files: list[str]) -> dict[str, Any]: if len(core_cases) > 0: jobs["components"]["core"] = True - jobs["core"].append({"os": "ubuntu-latest", "cases": core_cases}) + linux_cases = [v for v in core_cases if v["service"] != "hf3fs"] + hf3fs_cases = [v for v in core_cases if v["service"] == "hf3fs"] + if linux_cases: + jobs["core"].append({"os": "ubuntu-latest", "cases": linux_cases}) + if hf3fs_cases: + jobs["core"].append({"os": "ubuntu-22.04", "cases": hf3fs_cases}) # fs is the only services need to run upon windows, let's hard code it here. if "fs" in [v["service"] for v in core_cases]: diff --git a/.github/scripts/test_behavior/test_plan.py b/.github/scripts/test_behavior/test_plan.py index 8d7ed33956ef..bfabf7fdd4d2 100644 --- a/.github/scripts/test_behavior/test_plan.py +++ b/.github/scripts/test_behavior/test_plan.py @@ -54,6 +54,19 @@ def test_core_services_hdfs_native_mapping(self): self.assertTrue("hdfs_native" in cases) self.assertFalse("fs" in cases) + def test_hf3fs_uses_ubuntu_22(self): + result = plan(["core/services/hf3fs/src/lib.rs"]) + self.assertEqual(result["core"], [{ + "os": "ubuntu-22.04", + "cases": [{ + "service": "hf3fs", + "feature": "services-hf3fs", + "setups": ["single_node"], + }], + }]) + for language in ["java", "python", "ruby", "nodejs", "go", "c", "cpp", "dotnet"]: + self.assertEqual(result[f"binding_{language}"], []) + def test_group_core_cases_by_service(self): cases = [ { diff --git a/.github/services/hf3fs/README.md b/.github/services/hf3fs/README.md new file mode 100644 index 000000000000..6bac1f51f2b3 --- /dev/null +++ b/.github/services/hf3fs/README.md @@ -0,0 +1,30 @@ +# 3FS behavior tests + +The `single_node` setup starts a real, single-replica 3FS cluster on a hosted +Ubuntu 22.04 runner. It uses SoftRoCE, FoundationDB, and ordinary data directories; +it requires neither a physical RDMA adapter nor a dedicated disk. Ubuntu 24.04's +current Azure kernel disables RXE, so the behavior planner selects Ubuntu 22.04. + +CI caches the pinned 3FS build, runs the focused USRBIO tests and the standard +behavior suite, and stops the cluster and removes the RXE device afterward. +The first build is expensive; subsequent jobs restore the compiled backend. + +To test an existing Linux deployment, use a real 3FS FUSE mount and the matching +USRBIO client library. + +Create a dedicated test directory on the mount. Set `OPENDAL_HF3FS_ROOT` to its +absolute path and `OPENDAL_HF3FS_LIBRARY_PATH` to `libhf3fs_api_shared.so`. +Set `OPENDAL_TEST=hf3fs` and `OPENDAL_DISABLE_RANDOM_ROOT=true`, since the service +requires its root to exist before construction. + +From `core/`, run: + +```shell +cargo test --test behavior --features tests,services-hf3fs +cargo test -p opendal-service-hf3fs --test real_service -- --ignored +cargo run --release -p opendal-service-hf3fs --example bench_read +``` + +The benchmark compares 1 MiB streaming reads at concurrency 1 and 16 against +`Fs` on the same file and mount. It warms each path and reports four measured +passes. Results depend on client configuration, caching, and cluster load. diff --git a/.github/services/hf3fs/single_node/action.yml b/.github/services/hf3fs/single_node/action.yml new file mode 100644 index 000000000000..7ee480a31dfb --- /dev/null +++ b/.github/services/hf3fs/single_node/action.yml @@ -0,0 +1,70 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# 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. + +name: single_node +description: Test 3FS with one storage node, one replica, and SoftRoCE + +runs: + using: composite + steps: + - uses: actions/cache/restore@v5 + id: cache + with: + path: ${{ runner.temp }}/3fs + key: hf3fs-ubuntu22-${{ hashFiles('.github/services/hf3fs/single_node/setup.sh') }} + - name: Prepare 3FS + shell: bash + env: + CACHE_HIT: ${{ steps.cache.outputs.cache-hit }} + run: bash .github/services/hf3fs/single_node/setup.sh + - uses: actions/cache/save@v5 + if: steps.cache.outputs.cache-hit != 'true' + with: + path: ${{ runner.temp }}/3fs + key: hf3fs-ubuntu22-${{ hashFiles('.github/services/hf3fs/single_node/setup.sh') }} + - name: Start 3FS + shell: bash + run: | + source_dir="$RUNNER_TEMP/3fs" + cluster="$RUNNER_TEMP/hf3fs-cluster" + sed -i '/\[ib_devices\]/a device_filter = ["opendal_rxe"]' \ + "$source_dir"/tests/fuse/config/*launcher.toml "$source_dir"/tests/fuse/config/admin_cli.toml + sed -i "s/start_level = 'NONE'/start_level = 'WARN'/" "$source_dir"/tests/fuse/config/*.toml + sudo systemd-run --unit=opendal-hf3fs --uid="$(id -u)" \ + --working-directory="$RUNNER_TEMP" \ + --property=LimitMEMLOCK=infinity --property=TimeoutStopSec=10s \ + bash "$source_dir/tests/fuse/run.sh" "$source_dir/build/bin" "$cluster" + for _ in $(seq 1 120); do + if mountpoint -q "$cluster/mnt" && [[ -d "$cluster/mnt/test" ]]; then + break + fi + if ! systemctl is-active --quiet opendal-hf3fs; then + sudo journalctl -u opendal-hf3fs --no-pager + exit 1 + fi + sleep 1 + done + mountpoint "$cluster/mnt" + test -d "$cluster/mnt/test" + echo "OPENDAL_HF3FS_ROOT=$cluster/mnt/test" >> "$GITHUB_ENV" + echo "OPENDAL_HF3FS_LIBRARY_PATH=$source_dir/build/src/lib/api/libhf3fs_api_shared.so" >> "$GITHUB_ENV" + echo "OPENDAL_DISABLE_RANDOM_ROOT=true" >> "$GITHUB_ENV" + echo "RUST_TEST_THREADS=2" >> "$GITHUB_ENV" + - name: Test USRBIO + shell: bash + working-directory: core + run: cargo test --locked -p opendal-service-hf3fs --test real_service -- --ignored diff --git a/.github/services/hf3fs/single_node/setup.sh b/.github/services/hf3fs/single_node/setup.sh new file mode 100644 index 000000000000..dcb43e6c84ec --- /dev/null +++ b/.github/services/hf3fs/single_node/setup.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# 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. + +set -euo pipefail + +sudo apt-get update -qq +sudo apt-get install -y "linux-modules-extra-$(uname -r)" rdma-core ibverbs-utils \ + cmake ninja-build libuv1-dev liblz4-dev liblzma-dev libdouble-conversion-dev \ + libdwarf-dev libunwind-dev libaio-dev libgflags-dev libgoogle-glog-dev \ + libgtest-dev libgmock-dev clang-14 clang-format-14 lld-14 libgoogle-perftools-dev \ + libssl-dev gcc-12 g++-12 libboost-all-dev meson libibverbs-dev libnuma-dev \ + libevent-dev libfmt-dev autoconf automake libtool gettext fuse3 +sudo modprobe rdma_rxe +device=$(ip -4 route show default | awk 'NR == 1 {print $5}') +sudo rdma link add opendal_rxe type rxe netdev "$device" +ibv_devinfo -d opendal_rxe + +curl -fL --retry 3 -o "$RUNNER_TEMP/fdb.deb" \ + https://github.com/apple/foundationdb/releases/download/7.1.61/foundationdb-clients_7.1.61-1_amd64.deb +sudo dpkg -i "$RUNNER_TEMP/fdb.deb" +curl -fL --retry 3 -o "$RUNNER_TEMP/fdbserver" \ + https://github.com/apple/foundationdb/releases/download/7.1.61/fdbserver.x86_64 +sudo install -m 755 "$RUNNER_TEMP/fdbserver" /usr/local/bin/fdbserver + +git clone -q --depth 1 --branch fuse-3.16.2 https://github.com/libfuse/libfuse.git "$RUNNER_TEMP/libfuse" +meson setup "$RUNNER_TEMP/libfuse/build" "$RUNNER_TEMP/libfuse" -Dexamples=false -Dtests=false +ninja -C "$RUNNER_TEMP/libfuse/build" +sudo ninja -C "$RUNNER_TEMP/libfuse/build" install +sudo ldconfig + +if [[ "$CACHE_HIT" == true ]]; then + exit 0 +fi + +git clone -q https://github.com/deepseek-ai/3FS.git "$RUNNER_TEMP/3fs" +cd "$RUNNER_TEMP/3fs" +git checkout 22fca04564c7cc230fd8b9523b8b92864e1dad47 +git submodule update --init --recursive --depth 1 --jobs 4 +./patches/apply.sh +cmake -S . -B build -G Ninja -DCMAKE_C_COMPILER=clang-14 -DCMAKE_CXX_COMPILER=clang++-14 \ + -DCMAKE_BUILD_TYPE=Release -DCMAKE_C_FLAGS_RELEASE='-O1 -DNDEBUG' \ + -DCMAKE_CXX_FLAGS_RELEASE='-O1 -DNDEBUG' -DSHUFFLE_METHOD=g++11 +cmake --build build -j 3 --target mgmtd_main meta_main storage_main hf3fs_fuse_main admin_cli hf3fs_api_shared diff --git a/.github/workflows/test_behavior_core.yml b/.github/workflows/test_behavior_core.yml index cabd6a89aa17..c55366adacc1 100644 --- a/.github/workflows/test_behavior_core.yml +++ b/.github/workflows/test_behavior_core.yml @@ -41,7 +41,7 @@ jobs: uses: ./.github/actions/setup with: need-protoc: true - need-rocksdb: true + need-rocksdb: ${{ matrix.cases.service != 'hf3fs' }} github-token: ${{ secrets.GITHUB_TOKEN }} - name: Setup 1Password Connect @@ -58,3 +58,21 @@ jobs: setups: ${{ join(matrix.cases.setups, ',') }} service: ${{ matrix.cases.service }} feature: ${{ matrix.cases.feature }} + + - name: Stop 3FS + if: always() && matrix.cases.service == 'hf3fs' + shell: bash + run: | + sudo journalctl -u opendal-hf3fs --no-pager > "$RUNNER_TEMP/hf3fs-journal.log" 2>&1 || true + sudo systemctl stop opendal-hf3fs || true + fusermount3 -u "$RUNNER_TEMP/hf3fs-cluster/mnt" 2>/dev/null || true + sudo rdma link delete opendal_rxe 2>/dev/null || true + + - name: Upload 3FS logs + if: always() && matrix.cases.service == 'hf3fs' + uses: actions/upload-artifact@v7 + with: + name: hf3fs-logs + path: | + ${{ runner.temp }}/hf3fs-journal.log + ${{ runner.temp }}/hf3fs-cluster/log diff --git a/core/Cargo.lock b/core/Cargo.lock index 8c49a8e5a441..1e5621e69dad 100644 --- a/core/Cargo.lock +++ b/core/Cargo.lock @@ -5732,6 +5732,7 @@ dependencies = [ "opendal-service-hdfs", "opendal-service-hdfs-native", "opendal-service-hf", + "opendal-service-hf3fs", "opendal-service-http", "opendal-service-ipfs", "opendal-service-ipmfs", @@ -6573,6 +6574,21 @@ dependencies = [ "tokio", ] +[[package]] +name = "opendal-service-hf3fs" +version = "0.59.5" +dependencies = [ + "bytes", + "futures", + "libc", + "libloading 0.8.9", + "opendal-core", + "opendal-service-fs", + "serde", + "tempfile", + "tokio", +] + [[package]] name = "opendal-service-http" version = "0.59.5" diff --git a/core/Cargo.toml b/core/Cargo.toml index d4f9c523c6b9..9ae42fc006ae 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -177,6 +177,7 @@ services-gridfs = ["dep:opendal-service-gridfs"] services-hdfs = ["dep:opendal-service-hdfs"] services-hdfs-native = ["dep:opendal-service-hdfs-native"] services-hf = ["dep:opendal-service-hf"] +services-hf3fs = ["dep:opendal-service-hf3fs"] services-http = ["dep:opendal-service-http"] services-huggingface = ["services-hf"] services-ipfs = ["dep:opendal-service-ipfs"] @@ -301,6 +302,7 @@ opendal-service-gridfs = { path = "services/gridfs", version = "0.59.5", optiona opendal-service-hdfs = { path = "services/hdfs", version = "0.59.5", optional = true, default-features = false } opendal-service-hdfs-native = { path = "services/hdfs-native", version = "0.59.5", optional = true, default-features = false } opendal-service-hf = { path = "services/hf", version = "0.59.5", optional = true, default-features = false } +opendal-service-hf3fs = { path = "services/hf3fs", version = "0.59.5", optional = true, default-features = false } opendal-service-http = { path = "services/http", version = "0.59.5", optional = true, default-features = false } opendal-service-ipfs = { path = "services/ipfs", version = "0.59.5", optional = true, default-features = false } opendal-service-ipmfs = { path = "services/ipmfs", version = "0.59.5", optional = true, default-features = false } diff --git a/core/services/hf3fs/Cargo.toml b/core/services/hf3fs/Cargo.toml new file mode 100644 index 000000000000..8b9eff180974 --- /dev/null +++ b/core/services/hf3fs/Cargo.toml @@ -0,0 +1,49 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# 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. + +[package] +description = "Apache OpenDAL Hf3fs service implementation" +name = "opendal-service-hf3fs" +readme = "README.md" + +authors = { workspace = true } +edition = { workspace = true } +homepage = { workspace = true } +license = { workspace = true } +repository = { workspace = true } +rust-version = { workspace = true } +version = { workspace = true } + +[package.metadata.docs.rs] +all-features = true + +[target.'cfg(target_os = "linux")'.dependencies] +bytes = { workspace = true } +libc = "0.2" +libloading = "0.8" +opendal-core = { path = "../../core", version = "0.59.5", default-features = false } +opendal-service-fs = { path = "../fs", version = "0.59.5", default-features = false } +serde = { workspace = true, features = ["derive"] } +tokio = { workspace = true, features = ["fs", "rt", "sync"] } + +[dev-dependencies] +futures = { workspace = true } +opendal-core = { path = "../../core", version = "0.59.5", default-features = false, features = [ + "executors-tokio", +] } +tempfile = "3" +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] } diff --git a/core/services/hf3fs/README.md b/core/services/hf3fs/README.md new file mode 100644 index 000000000000..5161af01053f --- /dev/null +++ b/core/services/hf3fs/README.md @@ -0,0 +1,68 @@ +# Apache OpenDAL™ 3FS Service + +`Hf3fs` reads and writes through the 3FS USRBIO shared-memory interface. Metadata +operations use the mounted filesystem. It requires Linux, a running 3FS FUSE +client, and a compatible `libhf3fs_api_shared.so` implementing the public +[`hf3fs_usrbio.h` API](https://github.com/deepseek-ai/3FS/blob/main/src/lib/api/hf3fs_usrbio.h). +The library is loaded at runtime; building OpenDAL does not require a 3FS SDK. + +Enable `services-hf3fs` in `opendal`, or use the split crates: + +```rust,no_run +use opendal_core::{Operator, Result}; +use opendal_service_hf3fs::Hf3fs; + +fn operator() -> Result { + Operator::new(Hf3fs::default() + .root("/mnt/3fs/data") + .library_path("/opt/3fs/lib/libhf3fs_api_shared.so")) +} +``` + +The root must be an existing absolute directory on 3FS. The service discovers +its mount point. By default, the dynamic loader searches for +`libhf3fs_api_shared.so`; `library_path` selects a specific installation. +Use the client library distributed with your 3FS deployment. + +The service supports stat, range reads, streaming writes, create directory, +list, delete, recursive delete, and rename. Writes create parent directories +and replace existing files; `if_not_exists` rejects an existing file. +Closing a writer waits for writes and synchronizes the file. Writes are not +atomic, and failed or aborted writes can leave partial files. Append, native +copy, presign, and rollback on abort are unsupported. Operations require a +Tokio runtime. The operation context supplies the executor for concurrent writes. + +## Memory and concurrency + +Each operator owns a read driver and a write driver. Each driver submits and +reaps batches on its own thread. `queue_depth` defaults to 32 requests per ring; +`block_size` defaults to 1 MiB. Shared memory uses approximately +`3 * queue_depth * block_size`, plus the rings, and request queues are bounded. +Cloned operators share these resources. + +Reads return shared-memory-backed buffers without copying while fewer than +`queue_depth` slots remain retained by callers. Additional completions copy to +ordinary buffers so collecting large files cannot exhaust the I/O pool and +deadlock. Buffer clones retain their slot until the last clone is dropped, +even after the operator is dropped. Writes copy their input into shared memory. + +Cancelling an awaiting future does not cancel submitted native I/O. The driver +retains the file and buffer until completion. If submission or completion +collection fails fatally, the driver stops and reports errors. Since USRBIO +provides no cancellation fence, resources still involved in native I/O remain +allocated until process exit. Rebuild the operator after a driver failure. +A stalled client can also keep its driver and in-flight resources alive; +operation timeouts do not release native buffers. + +## Testing on 3FS + +Set `OPENDAL_HF3FS_ROOT` to an existing, writable test directory on a real 3FS +mount. Optionally set `OPENDAL_HF3FS_LIBRARY_PATH`. From `core/`, run: + +```shell +cargo test -p opendal-service-hf3fs --test real_service -- --ignored +OPENDAL_TEST=hf3fs OPENDAL_DISABLE_RANDOM_ROOT=true cargo test --test behavior --features tests,services-hf3fs +``` + +The focused tests create and remove their own subdirectories. Use a dedicated +test root for the behavior suite. Neither test command needs a mock service. diff --git a/core/services/hf3fs/examples/bench_read.rs b/core/services/hf3fs/examples/bench_read.rs new file mode 100644 index 000000000000..08c289473092 --- /dev/null +++ b/core/services/hf3fs/examples/bench_read.rs @@ -0,0 +1,81 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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. + +//! Compare streaming reads through Fs and USRBIO on the same 3FS file. +#[cfg(target_os = "linux")] +use std::time::Instant; + +#[cfg(target_os = "linux")] +use futures::TryStreamExt; +#[cfg(target_os = "linux")] +use opendal_core::{Operator, Result}; +#[cfg(target_os = "linux")] +use opendal_service_hf3fs::Hf3fs; + +#[cfg(not(target_os = "linux"))] +fn main() { + eprintln!("The 3FS benchmark requires Linux."); +} + +#[cfg(target_os = "linux")] +#[tokio::main] +async fn main() -> Result<()> { + let root = std::env::var("OPENDAL_HF3FS_ROOT").expect("set OPENDAL_HF3FS_ROOT"); + let directory = tempfile::tempdir_in(root).expect("create benchmark directory"); + let root = directory.path().to_str().unwrap(); + let mut builder = Hf3fs::default().root(root); + if let Ok(library) = std::env::var("OPENDAL_HF3FS_LIBRARY_PATH") { + builder = builder.library_path(&library); + } + let hf3fs = Operator::new(builder)?; + let fs = Operator::new(opendal_service_fs::Fs::default().root(root))?; + fs.write("data", vec![0x5a; 64 * 1024 * 1024]).await?; + + for concurrent in [1, 16] { + for (name, operator) in [("fs", &fs), ("hf3fs", &hf3fs)] { + // Warm the same file before measuring four complete streaming reads. + read(operator, concurrent).await?; + let start = Instant::now(); + let mut bytes = 0; + for _ in 0..4 { + bytes += read(operator, concurrent).await?; + } + println!( + "{name} concurrency={concurrent}: {:.1} MiB/s", + bytes as f64 / start.elapsed().as_secs_f64() / 1048576.0 + ); + } + } + Ok(()) +} + +#[cfg(target_os = "linux")] +async fn read(operator: &Operator, concurrent: usize) -> Result { + let mut stream = operator + .reader_with("data") + .chunk(1024 * 1024) + .concurrent(concurrent) + .await? + .into_stream(..) + .await?; + let mut size = 0; + while let Some(buffer) = stream.try_next().await? { + size += buffer.len(); + std::hint::black_box(buffer); + } + Ok(size) +} diff --git a/core/services/hf3fs/src/backend.rs b/core/services/hf3fs/src/backend.rs new file mode 100644 index 000000000000..30f644158b47 --- /dev/null +++ b/core/services/hf3fs/src/backend.rs @@ -0,0 +1,264 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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 std::ffi::CString; +use std::os::unix::ffi::OsStrExt; +use std::path::{Component, Path, PathBuf}; +use std::sync::Arc; + +use opendal_core::raw::*; +use opendal_core::*; + +use crate::HF3FS_SCHEME; +use crate::config::Hf3fsConfig; +use crate::driver::IoDriver; +use crate::ffi::Api; +use crate::reader::Hf3fsReader; +use crate::writer::Hf3fsWriter; + +/// Access 3FS through USRBIO for data and the mounted filesystem for metadata. +/// +/// Requires Linux, a running 3FS FUSE client, and its compatible USRBIO shared +/// library. The root must already exist. Reads retain shared memory while space +/// permits; writes copy into shared memory. See the crate documentation for +/// memory limits and failure semantics. +#[derive(Debug, Default)] +pub struct Hf3fsBuilder { + pub(crate) config: Hf3fsConfig, +} + +impl Hf3fsBuilder { + /// Set the existing absolute directory used as the storage root. + pub fn root(mut self, root: &str) -> Self { + self.config.root = Some(root.into()); + self + } + + /// Set the path to the compatible `libhf3fs_api_shared.so` library. + /// Defaults to the system dynamic loader's search path. + pub fn library_path(mut self, path: &str) -> Self { + self.config.library_path = Some(path.into()); + self + } + + /// Set the positive maximum number of in-flight requests per ring. + /// Defaults to 32. Each operator owns one read ring and one write ring. + pub fn queue_depth(mut self, depth: usize) -> Self { + self.config.queue_depth = Some(depth); + self + } + + /// Set the positive maximum request size in bytes. Defaults to 1 MiB. + /// Shared memory usage is approximately `3 * queue_depth * block_size`. + pub fn block_size(mut self, size: usize) -> Self { + self.config.block_size = Some(size); + self + } +} + +impl Builder for Hf3fsBuilder { + type Config = Hf3fsConfig; + + fn build(self) -> Result { + let depth = self.config.queue_depth.unwrap_or(32); + let block_size = self.config.block_size.unwrap_or(1024 * 1024); + if depth == 0 + || depth >= i32::MAX as usize + || block_size == 0 + || depth + .checked_mul(3) + .and_then(|n| n.checked_mul(block_size)) + .is_none_or(|n| n > isize::MAX as usize) + { + return Err(Error::new( + ErrorKind::ConfigInvalid, + "invalid USRBIO queue depth or block size", + )); + } + let root = self + .config + .root + .as_deref() + .filter(|p| Path::new(p).is_absolute()) + .ok_or_else(|| { + Error::new(ErrorKind::ConfigInvalid, "an absolute 3FS root is required") + })?; + let root = Path::new(root).canonicalize().map_err(new_std_io_error)?; + if !root.is_dir() { + return Err(Error::new( + ErrorKind::ConfigInvalid, + "3FS root must be a directory", + )); + } + let api = Api::load(Path::new( + self.config + .library_path + .as_deref() + .unwrap_or("libhf3fs_api_shared.so"), + ))?; + let c_root = CString::new(root.as_os_str().as_bytes()) + .map_err(|e| Error::new(ErrorKind::ConfigInvalid, "root contains NUL").set_source(e))?; + let mount = api.mount_point(&c_root)?; + let read = IoDriver::start(api.clone(), mount.clone(), depth, block_size, true)?; + let write = IoDriver::start(api.clone(), mount, depth, block_size, false)?; + let metadata = Arc::new( + opendal_service_fs::Fs::default() + .root(&root.to_string_lossy()) + .build()?, + ); + Ok(Hf3fsBackend { + core: Arc::new(Hf3fsCore { + root, + api, + read, + write, + }), + metadata, + }) + } +} + +pub(crate) struct Hf3fsCore { + pub root: PathBuf, + pub api: Arc, + pub read: IoDriver, + pub write: IoDriver, +} + +impl Hf3fsCore { + pub fn path(&self, path: &str) -> Result { + let path = Path::new(path.trim_end_matches('/')); + if path.components().any(|c| { + matches!( + c, + Component::ParentDir | Component::RootDir | Component::Prefix(_) + ) + }) { + return Err(Error::new( + ErrorKind::NotFound, + "path escapes the configured root", + )); + } + Ok(self.root.join(path)) + } +} + +struct Hf3fsBackend { + core: Arc, + metadata: Servicer, +} + +impl std::fmt::Debug for Hf3fsBackend { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Hf3fsBackend") + .field("root", &self.core.root) + .finish_non_exhaustive() + } +} + +impl Service for Hf3fsBackend { + type Reader = oio::PositionReader; + type Writer = Hf3fsWriter; + type Lister = oio::Lister; + type Deleter = oio::Deleter; + type Copier = (); + type Composer = (); + + fn info(&self) -> ServiceInfo { + ServiceInfo::new(HF3FS_SCHEME, self.core.root.to_string_lossy(), "") + } + + fn capability(&self) -> Capability { + Capability { + stat: true, + read: true, + write: true, + write_can_empty: true, + write_can_multi: true, + write_with_if_not_exists: true, + create_dir: true, + list: true, + delete: true, + delete_with_recursive: true, + rename: true, + shared: true, + ..Default::default() + } + } + + fn read(&self, _: &OperationContext, path: &str, _: OpRead) -> Result { + Ok(oio::PositionReader::new(Hf3fsReader { + core: self.core.clone(), + path: self.core.path(path)?, + }) + .with_max_buf_size(self.core.read.block_size)) + } + + fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result { + Ok(Hf3fsWriter::new( + self.core.clone(), + self.core.path(path)?, + ctx.executor().clone(), + args, + )) + } + + async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result { + self.metadata.stat(ctx, path, args).await + } + + async fn create_dir( + &self, + ctx: &OperationContext, + path: &str, + args: OpCreateDir, + ) -> Result { + self.metadata.create_dir(ctx, path, args).await + } + + fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result { + self.metadata.list(ctx, path, args) + } + + fn delete(&self, ctx: &OperationContext) -> Result { + self.metadata.delete(ctx) + } + + fn copy(&self, _: &OperationContext, _: &str, _: &str, _: OpCopy) -> Result { + Err(Error::new( + ErrorKind::Unsupported, + "3FS native copy is unsupported", + )) + } + + async fn presign(&self, _: &OperationContext, _: &str, _: OpPresign) -> Result { + Err(Error::new( + ErrorKind::Unsupported, + "3FS presign is unsupported", + )) + } + + async fn rename( + &self, + ctx: &OperationContext, + from: &str, + to: &str, + args: OpRename, + ) -> Result { + self.metadata.rename(ctx, from, to, args).await + } +} diff --git a/core/services/hf3fs/src/config.rs b/core/services/hf3fs/src/config.rs new file mode 100644 index 000000000000..5b84af46d03e --- /dev/null +++ b/core/services/hf3fs/src/config.rs @@ -0,0 +1,55 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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 opendal_core::{Configurator, OperatorUri, Result}; +use serde::{Deserialize, Serialize}; + +use crate::Hf3fs; + +/// Configuration for 3FS USRBIO access on Linux. +#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default)] +#[non_exhaustive] +pub struct Hf3fsConfig { + /// Existing absolute directory on a mounted 3FS instance. Required. + pub root: Option, + /// USRBIO shared library path. Defaults to `libhf3fs_api_shared.so`. + pub library_path: Option, + /// Maximum in-flight requests per read or write ring. Defaults to 32. + pub queue_depth: Option, + /// Maximum bytes per request. Defaults to 1 MiB. + /// Shared memory usage is approximately `3 * queue_depth * block_size`. + pub block_size: Option, +} + +impl Configurator for Hf3fsConfig { + type Builder = Hf3fs; + + fn from_uri(uri: &OperatorUri) -> Result { + let mut options = uri.options().clone(); + if let Some(root) = uri.root().filter(|v| !v.is_empty()) { + options + .entry("root".into()) + .or_insert_with(|| format!("/{root}")); + } + Self::from_iter(options) + } + + fn into_builder(self) -> Self::Builder { + Hf3fs { config: self } + } +} diff --git a/core/services/hf3fs/src/driver.rs b/core/services/hf3fs/src/driver.rs new file mode 100644 index 000000000000..81d8a381586a --- /dev/null +++ b/core/services/hf3fs/src/driver.rs @@ -0,0 +1,473 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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 std::ffi::{CStr, CString}; +use std::fs::File; +use std::os::fd::AsRawFd; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use bytes::Bytes; +use opendal_core::raw::new_std_io_error; +use opendal_core::{Buffer, Error, ErrorKind, Result}; +use tokio::sync::{mpsc, oneshot}; + +use crate::ffi::{Api, Cqe, Ior, Iov}; + +pub(crate) struct RegisteredFile { + pub file: File, + api: Arc, +} + +impl RegisteredFile { + pub fn new(api: Arc, file: File) -> Result> { + // Registration borrows the descriptor until deregistration in Drop. + let result = unsafe { (api.reg_fd)(file.as_raw_fd(), 0) }; + // Unlike other USRBIO functions, registration returns positive errno. + if result > 0 { + return Err(new_std_io_error(std::io::Error::from_raw_os_error(result))); + } + Ok(Arc::new(Self { file, api })) + } +} + +impl Drop for RegisteredFile { + fn drop(&mut self) { + // Every prepared request owns an Arc, including cancelled callers. + unsafe { (self.api.dereg_fd)(self.file.as_raw_fd()) }; + } +} + +struct Slots { + free: Vec, + retained: usize, +} + +struct Pool { + api: Arc, + iov: Iov, + block_size: usize, + retain_limit: usize, + slots: Mutex, +} + +// The mapping is stable. Only the driver writes to exclusively checked-out +// slots; completed, retained slots are immutable until their final owner drops. +unsafe impl Send for Pool {} +unsafe impl Sync for Pool {} + +impl Pool { + fn new( + api: Arc, + mount: &CStr, + depth: usize, + block_size: usize, + read: bool, + ) -> Result> { + let count = if read { depth * 2 } else { depth }; + // Zero is a valid initial state for the C struct, which the API fills. + let mut iov = unsafe { std::mem::zeroed() }; + let result = unsafe { + (api.iovcreate)(&mut iov, mount.as_ptr(), count * block_size, block_size, -1) + }; + if result < 0 { + return Err(new_std_io_error(std::io::Error::from_raw_os_error(-result))); + } + Ok(Arc::new(Self { + api, + iov, + block_size, + retain_limit: depth, + slots: Mutex::new(Slots { + free: (0..count).collect(), + retained: 0, + }), + })) + } +} + +impl Drop for Pool { + fn drop(&mut self) { + // No ring, pending request, or returned Bytes can still use this mapping. + unsafe { (self.api.iovdestroy)(&mut self.iov) }; + } +} + +struct Slot { + pool: Arc, + index: usize, + len: usize, + retained: bool, +} + +impl AsRef<[u8]> for Slot { + fn as_ref(&self) -> &[u8] { + // Only completed reads use this view. A slot never crosses an Iov block. + unsafe { + std::slice::from_raw_parts( + self.pool.iov.base.add(self.index * self.pool.block_size), + self.len, + ) + } + } +} + +impl Slot { + fn into_buffer(mut self) -> Buffer { + let mut slots = self.pool.slots.lock().unwrap(); + if slots.retained < self.pool.retain_limit { + slots.retained += 1; + self.retained = true; + drop(slots); + Bytes::from_owner(self).into() + } else { + drop(slots); + // Reserve a full ring's worth of slots for progress even when an + // application collects a file larger than the shared-memory pool. + Bytes::copy_from_slice(self.as_ref()).into() + } + } +} + +impl Drop for Slot { + fn drop(&mut self) { + let mut slots = self.pool.slots.lock().unwrap(); + if self.retained { + slots.retained -= 1; + } + slots.free.push(self.index); + } +} + +struct Request { + file: Arc, + offset: usize, + size: usize, + data: Option, + reply: oneshot::Sender>, +} + +struct Pending { + request: Request, + slot: Slot, +} + +#[derive(Debug, Clone)] +pub(crate) struct IoDriver { + tx: mpsc::Sender, + pub block_size: usize, + read: bool, +} + +impl IoDriver { + pub fn start( + api: Arc, + mount: CString, + depth: usize, + block_size: usize, + read: bool, + ) -> Result { + let (tx, rx) = mpsc::channel(depth); + let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel(1); + std::thread::Builder::new() + .name(if read { "hf3fs-read" } else { "hf3fs-write" }.into()) + .spawn( + move || match Worker::new(api, &mount, depth, block_size, read, rx) { + Ok(mut worker) => { + if ready_tx.send(Ok(())).is_ok() + && let Err(error) = worker.run() + { + worker.failure = error.to_string(); + } + } + Err(error) => { + let _ = ready_tx.send(Err(error)); + } + }, + ) + .map_err(new_std_io_error)?; + ready_rx.recv().map_err(|e| { + Error::new(ErrorKind::Unexpected, "USRBIO driver initialization failed").set_source(e) + })??; + Ok(Self { + tx, + block_size, + read, + }) + } + + pub async fn submit( + &self, + file: Arc, + offset: u64, + size: usize, + data: Option, + ) -> Result<(usize, Buffer)> { + if size == 0 + || size > self.block_size + || self.read != data.is_none() + || data.as_ref().is_some_and(|buffer| buffer.len() != size) + { + return Err(Error::new( + ErrorKind::Unexpected, + "invalid USRBIO buffer size or direction", + )); + } + let offset = usize::try_from(offset).map_err(|e| { + Error::new( + ErrorKind::RangeNotSatisfied, + "file offset exceeds platform limits", + ) + .set_source(e) + })?; + let (reply, result) = oneshot::channel(); + self.tx + .send(Request { + file, + offset, + size, + data, + reply, + }) + .await + .map_err(|_| Error::new(ErrorKind::Unexpected, "USRBIO driver stopped"))?; + result + .await + .map_err(|_| Error::new(ErrorKind::Unexpected, "USRBIO driver stopped"))? + } +} + +struct Worker { + pool: Arc, + ring: Option, + read: bool, + depth: usize, + rx: mpsc::Receiver, + pending: Vec>, + active: usize, + failure: String, +} + +impl Worker { + fn new( + api: Arc, + mount: &CStr, + depth: usize, + block_size: usize, + read: bool, + rx: mpsc::Receiver, + ) -> Result { + let pool = Pool::new(api.clone(), mount, depth, block_size, read)?; + let mut ring = unsafe { std::mem::zeroed() }; + // io_depth=0 dispatches partial batches immediately, including one I/O. + // Leave headroom for a completion whose submission slot is not reclaimed yet. + let result = unsafe { + (api.iorcreate4)( + &mut ring, + mount.as_ptr(), + (depth + 1) as i32, + read, + 0, + 0, + -1, + 0, + ) + }; + if result < 0 { + return Err(new_std_io_error(std::io::Error::from_raw_os_error(-result))); + } + Ok(Self { + pool, + ring: Some(ring), + read, + depth, + rx, + pending: (0..depth * 2).map(|_| None).collect(), + active: 0, + failure: "USRBIO driver stopped with pending I/O".into(), + }) + } + + fn prepare(&mut self, request: Request) -> Option { + if request.reply.is_closed() { + return None; + } + let index = self + .pool + .slots + .lock() + .unwrap() + .free + .pop() + .expect("reserved I/O slot"); + let slot = Slot { + pool: self.pool.clone(), + index, + len: request.size, + retained: false, + }; + // The slot is exclusive, and the request fits within one Iov block. + let ptr = unsafe { self.pool.iov.base.add(index * self.pool.block_size) }; + if let Some(data) = &request.data { + let dst = unsafe { std::slice::from_raw_parts_mut(ptr, request.size) }; + let mut offset = 0; + for part in data.clone() { + dst[offset..offset + part.len()].copy_from_slice(&part); + offset += part.len(); + } + } + let result = unsafe { + (self.pool.api.prep_io)( + self.ring.as_ref().unwrap(), + &self.pool.iov, + self.read, + ptr.cast(), + request.file.file.as_raw_fd(), + request.offset, + request.size as u64, + index as *const _, + ) + }; + if result == -libc::EAGAIN { + return Some(request); + } else if result < 0 { + let _ = request + .reply + .send(Err(new_std_io_error(std::io::Error::from_raw_os_error( + -result, + )))); + } else { + // A prepared I/O may start before submit_ios; retain all resources. + self.pending[index] = Some(Pending { request, slot }); + self.active += 1; + } + None + } + + fn run(&mut self) -> Result<()> { + let mut cqes: Vec = (0..self.depth) + .map(|_| unsafe { std::mem::zeroed() }) + .collect(); + let mut waiting = None; + loop { + let mut prepared = false; + // A completed request can leave its submission slot occupied until + // an earlier batch finishes. Drain before retrying a full ring. + while self.active < self.depth && (waiting.is_none() || self.active == 0) { + let request = if waiting.is_some() { + waiting.take() + } else if self.active == 0 { + self.rx.blocking_recv() + } else { + self.rx.try_recv().ok() + }; + let Some(request) = request else { break }; + waiting = self.prepare(request); + prepared = true; + if waiting.is_some() { + break; + } + } + if self.active == 0 { + if waiting.is_some() { + std::thread::sleep(Duration::from_millis(1)); + continue; + } + return Ok(()); + } + if prepared { + let result = unsafe { (self.pool.api.submit_ios)(self.ring.as_ref().unwrap()) }; + if result < 0 { + return Err(new_std_io_error(std::io::Error::from_raw_os_error(-result))); + } + } + // A bounded wait lets newly queued requests join an outstanding batch. + let deadline = (SystemTime::now() + Duration::from_millis(1)) + .duration_since(UNIX_EPOCH) + .unwrap(); + let timeout = libc::timespec { + tv_sec: deadline.as_secs() as _, + tv_nsec: deadline.subsec_nanos() as _, + }; + let count = unsafe { + (self.pool.api.wait_for_ios)( + self.ring.as_ref().unwrap(), + cqes.as_mut_ptr(), + cqes.len() as i32, + 1, + &timeout, + ) + }; + if count == -libc::ETIMEDOUT || count == -libc::EINTR { + continue; + } + if count < 0 { + return Err(new_std_io_error(std::io::Error::from_raw_os_error(-count))); + } + for cqe in &cqes[..count as usize] { + let index = cqe.userdata as usize; + let Pending { request, mut slot } = + self.pending[index].take().expect("known USRBIO completion"); + self.active -= 1; + let result = if cqe.result < 0 { + Err(new_std_io_error(std::io::Error::from_raw_os_error( + -cqe.result as i32, + ))) + } else if cqe.result as u64 > request.size as u64 { + Err(Error::new( + ErrorKind::Unexpected, + "USRBIO completion exceeds request size", + )) + } else { + let size = cqe.result as usize; + slot.len = size; + let buffer = if self.read && size > 0 { + slot.into_buffer() + } else { + Buffer::new() + }; + Ok((size, buffer)) + }; + let _ = request.reply.send(result); + } + } + } +} + +impl Drop for Worker { + fn drop(&mut self) { + let Some(mut ring) = self.ring.take() else { + return; + }; + if self.active == 0 { + unsafe { (self.pool.api.iordestroy)(&mut ring) }; + } else { + // USRBIO has no cancellation fence. On a fatal driver error, native + // I/O may still touch these resources: quarantine them until exit. + // Dropping a caller future alone never enters this path. + for pending in self.pending.iter_mut().filter_map(Option::take) { + let _ = pending + .request + .reply + .send(Err(Error::new(ErrorKind::Unexpected, self.failure.clone()))); + std::mem::forget((pending.request.file, pending.slot)); + } + // The ring's native allocation remains live; leaked slots retain + // the shared mapping and library needed by the in-flight requests. + } + } +} diff --git a/core/services/hf3fs/src/ffi.rs b/core/services/hf3fs/src/ffi.rs new file mode 100644 index 000000000000..4ebfb06e9a72 --- /dev/null +++ b/core/services/hf3fs/src/ffi.rs @@ -0,0 +1,135 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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. + +//! Minimal bindings to the public `hf3fs_usrbio.h` ABI. +//! https://github.com/deepseek-ai/3FS/blob/22fca04564c7cc230fd8b9523b8b92864e1dad47/src/lib/api/hf3fs_usrbio.h + +use std::ffi::{CStr, c_char, c_int, c_void}; +use std::path::Path; +use std::sync::Arc; + +use libloading::Library; +use opendal_core::{Error, ErrorKind, Result}; + +#[repr(C)] +pub(crate) struct Iov { + pub base: *mut u8, + handle: *mut c_void, + id: [c_char; 16], + mount_point: [c_char; 256], + size: usize, + block_size: usize, + numa: c_int, +} + +#[repr(C)] +pub(crate) struct Ior { + iov: Iov, + handle: *mut c_void, + mount_point: [c_char; 256], + for_read: bool, + io_depth: c_int, + priority: c_int, + timeout: c_int, + flags: u64, +} + +#[derive(Clone, Copy)] +#[repr(C)] +pub(crate) struct Cqe { + pub index: i32, + reserved: i32, + pub result: i64, + pub userdata: *const c_void, +} + +pub(crate) struct Api { + pub extract_mount_point: unsafe extern "C" fn(*mut c_char, c_int, *const c_char) -> c_int, + pub iovcreate: unsafe extern "C" fn(*mut Iov, *const c_char, usize, usize, c_int) -> c_int, + pub iovdestroy: unsafe extern "C" fn(*mut Iov), + pub iorcreate4: unsafe extern "C" fn( + *mut Ior, + *const c_char, + c_int, + bool, + c_int, + c_int, + c_int, + u64, + ) -> c_int, + pub iordestroy: unsafe extern "C" fn(*mut Ior), + pub reg_fd: unsafe extern "C" fn(c_int, u64) -> c_int, + pub dereg_fd: unsafe extern "C" fn(c_int), + pub prep_io: unsafe extern "C" fn( + *const Ior, + *const Iov, + bool, + *mut c_void, + c_int, + usize, + u64, + *const c_void, + ) -> c_int, + pub submit_ios: unsafe extern "C" fn(*const Ior) -> c_int, + pub wait_for_ios: + unsafe extern "C" fn(*const Ior, *mut Cqe, c_int, c_int, *const libc::timespec) -> c_int, + _library: Library, +} + +impl Api { + pub fn load(path: &Path) -> Result> { + // The configured library must implement the public 3FS C ABI. Keep it + // loaded for every registered file, ring, and outstanding buffer. + let load = || unsafe { + let library = Library::new(path)?; + Ok::<_, libloading::Error>(Self { + extract_mount_point: *library.get(b"hf3fs_extract_mount_point\0")?, + iovcreate: *library.get(b"hf3fs_iovcreate\0")?, + iovdestroy: *library.get(b"hf3fs_iovdestroy\0")?, + iorcreate4: *library.get(b"hf3fs_iorcreate4\0")?, + iordestroy: *library.get(b"hf3fs_iordestroy\0")?, + reg_fd: *library.get(b"hf3fs_reg_fd\0")?, + dereg_fd: *library.get(b"hf3fs_dereg_fd\0")?, + prep_io: *library.get(b"hf3fs_prep_io\0")?, + submit_ios: *library.get(b"hf3fs_submit_ios\0")?, + wait_for_ios: *library.get(b"hf3fs_wait_for_ios\0")?, + _library: library, + }) + }; + load().map(Arc::new).map_err(|e| { + Error::new( + ErrorKind::ConfigInvalid, + "failed to load 3FS USRBIO library", + ) + .set_source(e) + }) + } + + pub fn mount_point(&self, root: &CStr) -> Result { + let mut mount = [0; 256]; + // Both strings are bounded and the root is NUL terminated. + let n = unsafe { (self.extract_mount_point)(mount.as_mut_ptr(), 256, root.as_ptr()) }; + if n <= 0 || n > 256 { + return Err(Error::new( + ErrorKind::ConfigInvalid, + "root is not on a supported 3FS mount", + )); + } + // The API includes the terminating NUL in its return value. + Ok(unsafe { CStr::from_ptr(mount.as_ptr()) }.to_owned()) + } +} diff --git a/core/services/hf3fs/src/lib.rs b/core/services/hf3fs/src/lib.rs new file mode 100644 index 000000000000..fdc2bccdc7fb --- /dev/null +++ b/core/services/hf3fs/src/lib.rs @@ -0,0 +1,39 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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. + +#![doc = include_str!("../README.md")] +#![cfg(target_os = "linux")] +#![deny(missing_docs)] + +mod backend; +mod config; +mod driver; +mod ffi; +mod reader; +mod writer; + +pub use backend::Hf3fsBuilder as Hf3fs; +pub use config::Hf3fsConfig; + +/// URI scheme for the 3FS USRBIO service. +pub const HF3FS_SCHEME: &str = "hf3fs"; + +/// Register 3FS for [`opendal_core::Operator::from_uri`] and +/// [`opendal_core::Operator::via_iter`]. +pub fn register_hf3fs_service(registry: &opendal_core::OperatorRegistry) { + registry.register::(HF3FS_SCHEME); +} diff --git a/core/services/hf3fs/src/reader.rs b/core/services/hf3fs/src/reader.rs new file mode 100644 index 000000000000..d79d29929ca8 --- /dev/null +++ b/core/services/hf3fs/src/reader.rs @@ -0,0 +1,69 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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 std::path::PathBuf; +use std::sync::Arc; + +use opendal_core::raw::*; +use opendal_core::{Buffer, Result}; + +use crate::backend::Hf3fsCore; +use crate::driver::{IoDriver, RegisteredFile}; + +pub(crate) struct Hf3fsReader { + pub core: Arc, + pub path: PathBuf, +} + +pub(crate) struct Hf3fsReadHandle { + file: Arc, + driver: IoDriver, +} + +impl oio::PositionRead for Hf3fsReader { + type Handle = Hf3fsReadHandle; + + async fn open(&self) -> Result { + let api = self.core.api.clone(); + let path = self.path.clone(); + let file = tokio::task::spawn_blocking(move || { + RegisteredFile::new(api, std::fs::File::open(path).map_err(new_std_io_error)?) + }) + .await + .map_err(new_task_join_error)??; + Ok(Hf3fsReadHandle { + file, + driver: self.core.read.clone(), + }) + } + + async fn read_at(handle: &Self::Handle, offset: u64, size: usize) -> Result { + if size == 0 { + return Ok(Buffer::new()); + } + let (_, buffer) = handle + .driver + .submit( + handle.file.clone(), + offset, + size.min(handle.driver.block_size), + None, + ) + .await?; + Ok(buffer) + } +} diff --git a/core/services/hf3fs/src/writer.rs b/core/services/hf3fs/src/writer.rs new file mode 100644 index 000000000000..5e5ba87338ab --- /dev/null +++ b/core/services/hf3fs/src/writer.rs @@ -0,0 +1,147 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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 std::path::PathBuf; +use std::sync::Arc; + +use bytes::Buf; +use opendal_core::raw::*; +use opendal_core::*; + +use crate::backend::Hf3fsCore; +use crate::driver::{IoDriver, RegisteredFile}; + +pub(crate) struct Hf3fsWriter { + core: Arc, + path: PathBuf, + executor: Executor, + options: OpWrite, + inner: Option>, +} + +impl Hf3fsWriter { + pub fn new(core: Arc, path: PathBuf, executor: Executor, options: OpWrite) -> Self { + Self { + core, + path, + executor, + options, + inner: None, + } + } + + async fn inner(&mut self) -> Result<&mut oio::PositionWriter> { + if self.inner.is_none() { + let path = self.path.clone(); + let api = self.core.api.clone(); + let exclusive = self.options.if_not_exists(); + let file = tokio::task::spawn_blocking(move || { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent).map_err(new_std_io_error)?; + } + let file = std::fs::OpenOptions::new() + .write(true) + .create(true) + .create_new(exclusive) + .truncate(true) + .open(path) + .map_err(|e| { + if exclusive && e.kind() == std::io::ErrorKind::AlreadyExists { + Error::new(ErrorKind::ConditionNotMatch, "file already exists") + .set_source(e) + } else { + new_std_io_error(e) + } + })?; + RegisteredFile::new(api, file) + }) + .await + .map_err(new_task_join_error)??; + self.inner = Some(oio::PositionWriter::new( + self.executor.clone(), + Hf3fsWriteHandle { + file, + driver: self.core.write.clone(), + }, + self.options.concurrent(), + )); + } + Ok(self.inner.as_mut().expect("writer initialized")) + } +} + +impl oio::Write for Hf3fsWriter { + async fn write(&mut self, buffer: Buffer) -> Result<()> { + self.inner().await?.write(buffer).await + } + + async fn close(&mut self) -> Result { + self.inner().await?.close().await + } + + async fn abort(&mut self) -> Result<()> { + Err(Error::new( + ErrorKind::Unsupported, + "3FS writes cannot be rolled back", + )) + } +} + +struct Hf3fsWriteHandle { + file: Arc, + driver: IoDriver, +} + +impl oio::PositionWrite for Hf3fsWriteHandle { + async fn write_all_at(&self, mut offset: u64, mut buffer: Buffer) -> Result<()> { + while !buffer.is_empty() { + let size = buffer.len().min(self.driver.block_size); + let (written, _) = self + .driver + .submit(self.file.clone(), offset, size, Some(buffer.slice(..size))) + .await?; + if written == 0 { + return Err(new_std_io_error(std::io::ErrorKind::WriteZero.into())); + } + offset += written as u64; + buffer.advance(written); + } + Ok(()) + } + + async fn close(&self, _: u64) -> Result { + let file = self.file.clone(); + tokio::task::spawn_blocking(move || { + file.file.sync_all().map_err(new_std_io_error)?; + let stat = file.file.metadata().map_err(new_std_io_error)?; + let mut metadata = MetadataBuilder::file(stat.len()); + metadata.last_modified(Timestamp::try_from( + stat.modified().map_err(new_std_io_error)?, + )?); + Ok(metadata.build()) + }) + .await + .map_err(new_task_join_error)? + } + + async fn abort(&self) -> Result<()> { + Err(Error::new( + ErrorKind::Unsupported, + "3FS writes cannot be rolled back", + )) + } +} diff --git a/core/services/hf3fs/tests/real_service.rs b/core/services/hf3fs/tests/real_service.rs new file mode 100644 index 000000000000..407552c91245 --- /dev/null +++ b/core/services/hf3fs/tests/real_service.rs @@ -0,0 +1,193 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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. + +#![cfg(target_os = "linux")] + +use std::time::Duration; + +use futures::TryStreamExt; +use opendal_core::{ErrorKind, Operator}; +use opendal_service_hf3fs::Hf3fs; + +fn fixture() -> (tempfile::TempDir, Operator) { + let root = std::env::var("OPENDAL_HF3FS_ROOT") + .expect("set OPENDAL_HF3FS_ROOT to a real 3FS test directory"); + let dir = tempfile::tempdir_in(root).unwrap(); + let mut builder = Hf3fs::default() + .root(dir.path().to_str().unwrap()) + .queue_depth(2) + .block_size(4096); + if let Ok(library) = std::env::var("OPENDAL_HF3FS_LIBRARY_PATH") { + builder = builder.library_path(&library); + } + (dir, Operator::new(builder).unwrap()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires a real 3FS mount and USRBIO library"] +async fn read_write_ranges_and_metadata() { + let (dir, op) = fixture(); + let data: Vec = (0..131_137).map(|i| (i % 251) as u8).collect(); + op.write("nested/data", data.clone()).await.unwrap(); + assert_eq!(std::fs::read(dir.path().join("nested/data")).unwrap(), data); + assert_eq!( + op.stat("nested/data").await.unwrap().content_length(), + data.len() as u64 + ); + + // Retain more than the entire shared-memory pool without stalling progress. + let buffer = tokio::time::timeout(Duration::from_secs(10), op.read("nested/data")) + .await + .unwrap() + .unwrap(); + assert_eq!(buffer.to_vec(), data); + let mut stream = op + .reader("nested/data") + .await + .unwrap() + .into_stream(..) + .await + .unwrap(); + let mut streamed = Vec::new(); + tokio::time::timeout(Duration::from_secs(10), async { + while let Some(buffer) = stream.try_next().await.unwrap() { + streamed.extend(buffer.to_vec()); + } + }) + .await + .unwrap(); + assert_eq!(streamed, data); + assert_eq!( + op.read_with("nested/data") + .range(17..9001) + .await + .unwrap() + .to_vec(), + data[17..9001] + ); + assert_eq!( + op.read_with("nested/data") + .range(131_130..) + .await + .unwrap() + .to_vec(), + data[131_130..] + ); + + op.rename("nested/data", "nested/renamed").await.unwrap(); + assert!( + op.list("nested/") + .await + .unwrap() + .iter() + .any(|e| e.path() == "nested/renamed") + ); + op.delete("nested/renamed").await.unwrap(); + op.delete("nested/renamed").await.unwrap(); + assert_eq!( + op.stat("nested/renamed").await.unwrap_err().kind(), + ErrorKind::NotFound + ); + assert_eq!(buffer.to_vec(), data); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires a real 3FS mount and USRBIO library"] +async fn retained_buffers_survive_reuse_and_operator_drop() { + let (dir, op) = fixture(); + std::fs::write(dir.path().join("data"), vec![0x31; 4096]).unwrap(); + let first = op.read("data").await.unwrap(); + let clone = first.clone(); + let second = op.read("data").await.unwrap(); + for byte in 0..16u8 { + op.write("data", vec![byte; 4096]).await.unwrap(); + assert_eq!(op.read("data").await.unwrap().to_vec(), vec![byte; 4096]); + } + drop(first); + drop(op); + assert_eq!(clone.to_vec(), vec![0x31; 4096]); + assert_eq!(second.to_vec(), vec![0x31; 4096]); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires a real 3FS mount and USRBIO library"] +async fn concurrent_reads_and_cancelled_callers() { + let (dir, op) = fixture(); + let data: Vec = (0..65_537).map(|i| (i % 251) as u8).collect(); + std::fs::write(dir.path().join("data"), &data).unwrap(); + let reader = op.reader("data").await.unwrap(); + let mut tasks = tokio::task::JoinSet::new(); + for i in 0..64 { + let reader = reader.clone(); + let task = tasks.spawn(async move { reader.read(0..65_537).await }); + if i % 2 == 0 { + tokio::task::yield_now().await; + task.abort(); + } + } + tokio::time::timeout(Duration::from_secs(20), async { + while let Some(result) = tasks.join_next().await { + match result { + Ok(result) => assert_eq!(result.unwrap().to_vec(), data), + Err(error) => assert!(error.is_cancelled()), + } + } + }) + .await + .unwrap(); + assert_eq!(op.read("data").await.unwrap().to_vec(), data); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires a real 3FS mount and USRBIO library"] +async fn overwrite_empty_and_exclusive_writes() { + let (dir, op) = fixture(); + op.write("data", vec![0x41; 9000]).await.unwrap(); + let err = op + .write_with("data", "replacement") + .if_not_exists(true) + .await + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + assert_eq!(op.read("data").await.unwrap().len(), 9000); + op.write("data", "short").await.unwrap(); + assert_eq!(std::fs::read(dir.path().join("data")).unwrap(), b"short"); + op.write("data", "").await.unwrap(); + assert_eq!(op.stat("data").await.unwrap().content_length(), 0); + assert!(op.read("data").await.unwrap().is_empty()); + op.create_dir("empty/").await.unwrap(); + assert!(op.stat("empty/").await.unwrap().is_dir()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires a real 3FS mount and USRBIO library"] +async fn concurrent_streaming_writes() { + let (dir, op) = fixture(); + let mut writer = op + .writer_with("data") + .concurrent(4) + .chunk(4096) + .await + .unwrap(); + for byte in 0..32u8 { + writer.write(vec![byte; 4096]).await.unwrap(); + } + writer.close().await.unwrap(); + let expected: Vec = (0..32u8).flat_map(|byte| vec![byte; 4096]).collect(); + assert_eq!(std::fs::read(dir.path().join("data")).unwrap(), expected); + assert_eq!(op.read("data").await.unwrap().to_vec(), expected); +} diff --git a/core/src/lib.rs b/core/src/lib.rs index 75097fe746fc..6097917e703c 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -164,6 +164,9 @@ fn init_default_registry_inner(registry: &OperatorRegistry) { #[cfg(feature = "services-hf")] opendal_service_hf::register_hf_service(registry); + #[cfg(all(feature = "services-hf3fs", target_os = "linux"))] + opendal_service_hf3fs::register_hf3fs_service(registry); + #[cfg(feature = "services-ipfs")] opendal_service_ipfs::register_ipfs_service(registry); @@ -345,6 +348,8 @@ pub mod services { pub use opendal_service_hdfs_native::*; #[cfg(feature = "services-hf")] pub use opendal_service_hf::*; + #[cfg(all(feature = "services-hf3fs", target_os = "linux"))] + pub use opendal_service_hf3fs::*; #[cfg(feature = "services-http")] pub use opendal_service_http::*; #[cfg(feature = "services-ipfs")] From 8421123accfb8e6a2dcbcb3d709d8a982aeb2277 Mon Sep 17 00:00:00 2001 From: MrCroxx Date: Mon, 26 Oct 2026 15:42:17 +0800 Subject: [PATCH 2/6] docs: simplify hf3fs environment example heading --- .env.example | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.env.example b/.env.example index 09266282f92e..9017c9591fac 100644 --- a/.env.example +++ b/.env.example @@ -2,7 +2,7 @@ # fs OPENDAL_FS_ROOT=/path/to/dir OPENDAL_FS_ATOMIC_WRITE_DIR=/path/to/tempdir -# hf3fs (existing directory on a real 3FS mount) +# hf3fs OPENDAL_HF3FS_ROOT=/path/to/3fs/testdir OPENDAL_HF3FS_LIBRARY_PATH=/path/to/libhf3fs_api_shared.so # cos From 373ddced24841c1a4c73cf362761feae7c259563 Mon Sep 17 00:00:00 2001 From: MrCroxx Date: Mon, 26 Oct 2026 17:08:17 +0800 Subject: [PATCH 3/6] perf(hf3fs): add read and write benchmark matrix --- .github/services/hf3fs/README.md | 37 +++ core/services/hf3fs/examples/bench_matrix.rs | 305 +++++++++++++++++++ 2 files changed, 342 insertions(+) create mode 100644 core/services/hf3fs/examples/bench_matrix.rs diff --git a/.github/services/hf3fs/README.md b/.github/services/hf3fs/README.md index 6bac1f51f2b3..f9cb450f761b 100644 --- a/.github/services/hf3fs/README.md +++ b/.github/services/hf3fs/README.md @@ -28,3 +28,40 @@ cargo run --release -p opendal-service-hf3fs --example bench_read The benchmark compares 1 MiB streaming reads at concurrency 1 and 16 against `Fs` on the same file and mount. It warms each path and reports four measured passes. Results depend on client configuration, caching, and cluster load. + +For the read/write matrix, run: + +```shell +cargo run --release -p opendal-service-hf3fs --example bench_matrix > matrix.csv +``` + +The matrix uses 4 KiB, 64 KiB, 1 MiB, and 4 MiB application blocks with 1, 4, +16, and 64 concurrent workers. Each worker issues one application operation at +a time. Both services use the same 1 GiB input file. `Hf3fs` uses a queue depth +of 64 per direction and a native block size matching the application block. +`Fs` can split large application reads into smaller native requests. + +Workloads include sequential range reads, random range reads, streaming writes, +single-block object overwrites, and mixed random reads/overwrites with 70% or +50% reads by operation count. Mixed workloads use separate input and output +files; they do not test random writes within an existing file. Overwrites include +open, truncate, write, close, and fsync. Streaming writers close and synchronize +every 64 MiB and at the end; their write latency measures API admission, while +throughput includes final synchronization. These latency columns are not device +completion latencies. + +The default run preserves each service's normal cache behavior. Reopening the +file on some 3FS deployments invalidates its kernel page cache; initial prewarming +does not guarantee cache hits for new readers. Set `BENCH_WARM_READS=true` to +prewarm the whole input through an existing `Fs` reader after all workers open +the file. USRBIO bypasses that file cache. Neither profile isolates uncached +transport performance. The benchmark uses constant payloads and checks returned +lengths and sampled data. CPU usage covers the benchmark process, excluding the +FUSE process and servers. + +Each case runs for 2 seconds and repeats 3 times, reversing service order on +alternate rounds. CSV rows contain counts, throughput, IOPS, latency percentiles, +and CPU usage. Set `BENCH_SECONDS`, `BENCH_ROUNDS`, `BENCH_BS` (bytes), +`BENCH_CONCURRENCY`, or `BENCH_WORKLOADS` to override the defaults; the last three +accept comma-separated lists. Short cases and shared-cluster load limit the +precision of tail latencies. Temporary files are removed after a successful run. diff --git a/core/services/hf3fs/examples/bench_matrix.rs b/core/services/hf3fs/examples/bench_matrix.rs new file mode 100644 index 000000000000..54df0278ad2d --- /dev/null +++ b/core/services/hf3fs/examples/bench_matrix.rs @@ -0,0 +1,305 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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. + +//! Compare OpenDAL read, write, and mixed workloads on an existing 3FS mount. +#[cfg(not(target_os = "linux"))] +fn main() { + eprintln!("The 3FS benchmark requires Linux."); +} + +#[cfg(target_os = "linux")] +#[tokio::main(worker_threads = 8)] +async fn main() -> opendal_core::Result<()> { + bench::run().await +} + +#[cfg(target_os = "linux")] +mod bench { + use std::sync::Arc; + use std::time::{Duration, Instant}; + + use futures::TryStreamExt; + use opendal_core::{Buffer, Operator, Result}; + use opendal_service_hf3fs::Hf3fs; + use tokio::sync::Barrier; + + const FILE_SIZE: usize = 1024 * 1024 * 1024; + const SEGMENT_SIZE: usize = 64 * 1024 * 1024; + + #[derive(Default)] + struct Samples { + reads: Vec, + writes: Vec, + closes: Vec, + } + + #[derive(Clone, Copy)] + enum Workload { + Read, + RandomRead, + StreamWrite, + Overwrite, + Mixed70, + Mixed50, + } + + impl Workload { + fn name(self) -> &'static str { + match self { + Self::Read => "seqread", + Self::RandomRead => "randread", + Self::StreamWrite => "seqwrite", + Self::Overwrite => "overwrite", + Self::Mixed70 => "mixed70", + Self::Mixed50 => "mixed50", + } + } + + fn read_percent(self) -> usize { + match self { + Self::Read | Self::RandomRead => 100, + Self::StreamWrite | Self::Overwrite => 0, + Self::Mixed70 => 70, + Self::Mixed50 => 50, + } + } + } + + fn setting(name: &str, default: T) -> T { + std::env::var(name).map_or(default, |v| v.parse().ok().expect(name)) + } + + fn percentiles(values: &mut [f64]) -> [f64; 3] { + if values.is_empty() { + return [0.0; 3]; + } + values.sort_unstable_by(f64::total_cmp); + [50, 95, 99].map(|p| values[((values.len() * p).div_ceil(100) - 1).min(values.len() - 1)]) + } + + fn cpu_seconds() -> f64 { + let mut usage = std::mem::MaybeUninit::::uninit(); + // getrusage initializes the output on success and does not retain it. + let result = unsafe { libc::getrusage(libc::RUSAGE_SELF, usage.as_mut_ptr()) }; + assert_eq!(result, 0); + let usage = unsafe { usage.assume_init() }; + (usage.ru_utime.tv_sec + usage.ru_stime.tv_sec) as f64 + + (usage.ru_utime.tv_usec + usage.ru_stime.tv_usec) as f64 / 1e6 + } + + pub async fn run() -> Result<()> { + let root = std::env::var("OPENDAL_HF3FS_ROOT").expect("set OPENDAL_HF3FS_ROOT"); + let directory = tempfile::tempdir_in(root).expect("create benchmark directory"); + let root = directory.path().to_str().unwrap(); + let fs = Operator::new(opendal_service_fs::Fs::default().root(root))?; + let seconds = setting("BENCH_SECONDS", 2.0_f64); + let rounds = setting("BENCH_ROUNDS", 3_usize); + let warm_reads = setting("BENCH_WARM_READS", false); + assert!(seconds.is_finite() && seconds > 0.0 && rounds > 0); + let duration = Duration::from_secs_f64(seconds); + let block_sizes = std::env::var("BENCH_BS").unwrap_or("4096,65536,1048576,4194304".into()); + let concurrencies = std::env::var("BENCH_CONCURRENCY").unwrap_or("1,4,16,64".into()); + let selected = std::env::var("BENCH_WORKLOADS") + .unwrap_or("seqread,randread,seqwrite,overwrite,mixed70,mixed50".into()); + let payload: Buffer = vec![0x5a; 4 * 1024 * 1024].into(); + let mut writer = fs.writer("input").await?; + for _ in 0..FILE_SIZE / payload.len() { + writer.write(payload.clone()).await?; + } + writer.close().await?; + let mut warmup = fs.reader("input").await?.into_stream(..).await?; + while let Some(buffer) = warmup.try_next().await? { + std::hint::black_box(buffer); + } + eprintln!( + "file_bytes={FILE_SIZE} seconds={seconds} rounds={rounds} runtime_threads=8 queue_depth=64; warm_after_open={warm_reads}; Fs uses its default buffered read path; seqwrite latency measures write() admission; close/fsync is included in elapsed time" + ); + println!( + "round,service,workload,bs,concurrency,elapsed_s,read_ops,write_ops,read_mib_s,write_mib_s,iops,read_p50_us,read_p95_us,read_p99_us,write_p50_us,write_p95_us,write_p99_us,close_count,close_p50_us,close_p95_us,close_p99_us,cpu_cores" + ); + for round in 0..rounds { + for size in block_sizes.split(',') { + let bs: usize = size.parse().expect("BENCH_BS"); + assert!(bs > 0 && bs <= payload.len() && FILE_SIZE.is_multiple_of(bs)); + let mut builder = Hf3fs::default().root(root).queue_depth(64).block_size(bs); + if let Ok(path) = std::env::var("OPENDAL_HF3FS_LIBRARY_PATH") { + builder = builder.library_path(&path); + } + let hf3fs = Operator::new(builder)?; + // Warm the native data path as well; it does not use the kernel page cache. + let mut warmup = hf3fs + .reader_with("input") + .chunk(4 * 1024 * 1024) + .concurrent(16) + .await? + .into_stream(0..SEGMENT_SIZE as u64) + .await?; + while let Some(buffer) = warmup.try_next().await? { + std::hint::black_box(buffer); + } + for concurrency in concurrencies.split(',') { + let concurrency: usize = concurrency.parse().expect("BENCH_CONCURRENCY"); + assert!(concurrency > 0 && concurrency <= 64); + for workload in [ + Workload::Read, + Workload::RandomRead, + Workload::StreamWrite, + Workload::Overwrite, + Workload::Mixed70, + Workload::Mixed50, + ] { + if !selected.split(',').any(|v| v == workload.name()) { + continue; + } + let order = if round % 2 == 0 { + [("fs", &fs), ("hf3fs", &hf3fs)] + } else { + [("hf3fs", &hf3fs), ("fs", &fs)] + }; + for (name, operator) in order { + let barrier = Arc::new(Barrier::new(concurrency + 1)); + let mut tasks = Vec::new(); + let mut warm_reader = None; + for worker in 0..concurrency { + let operator = operator.clone(); + let barrier = barrier.clone(); + let payload = payload.slice(..bs); + let reader = operator + .reader_with("input") + .chunk(bs) + .concurrent(1) + .await?; + reader.read(0..bs as u64).await?; + warm_reader = Some(reader.clone()); + let path = format!("output-{worker}"); + let mut writer = operator.writer_with(&path).concurrent(1).await?; + tasks.push(tokio::spawn(async move { + let mut samples = Samples::default(); + let mut rng = worker as u64 + 1; + let mut sequence = worker * (FILE_SIZE / bs / concurrency); + let mut written = 0; + barrier.wait().await; + let deadline = Instant::now() + duration; + while Instant::now() < deadline { + let read = + (sequence + worker) % 10 < workload.read_percent() / 10; + let start = Instant::now(); + if read { + rng ^= rng << 13; + rng ^= rng >> 7; + rng ^= rng << 17; + let block = if matches!(workload, Workload::Read) { + sequence + } else { + rng as usize + } % (FILE_SIZE / bs); + let offset = (block * bs) as u64; + let buffer = + reader.read(offset..offset + bs as u64).await?; + assert_eq!(buffer.len(), bs); + assert_eq!(buffer.clone().next().unwrap()[0], 0x5a); + std::hint::black_box(buffer); + samples.reads.push(start.elapsed().as_secs_f64() * 1e6); + } else if matches!(workload, Workload::StreamWrite) { + writer.write(payload.clone()).await?; + samples + .writes + .push(start.elapsed().as_secs_f64() * 1e6); + written += bs; + if written >= SEGMENT_SIZE { + let start = Instant::now(); + writer.close().await?; + samples + .closes + .push(start.elapsed().as_secs_f64() * 1e6); + writer = operator + .writer_with(&path) + .concurrent(1) + .await?; + written = 0; + } + } else { + operator.write(&path, payload.clone()).await?; + samples + .writes + .push(start.elapsed().as_secs_f64() * 1e6); + } + sequence += 1; + } + if written > 0 { + let start = Instant::now(); + writer.close().await?; + samples.closes.push(start.elapsed().as_secs_f64() * 1e6); + } + Ok::<_, opendal_core::Error>(samples) + })); + } + if warm_reads && name == "fs" && workload.read_percent() > 0 { + // Reopening a FUSE file can invalidate its page cache. Warm + // through an existing handle after all workers have opened it. + let mut stream = warm_reader.unwrap().into_stream(..).await?; + while let Some(buffer) = stream.try_next().await? { + std::hint::black_box(buffer); + } + } + let cpu = cpu_seconds(); + let start = Instant::now(); + barrier.wait().await; + let mut results = Vec::new(); + for task in tasks { + results.push(task.await.expect("benchmark worker panicked")?); + } + let elapsed = start.elapsed().as_secs_f64(); + let cpu = (cpu_seconds() - cpu) / elapsed; + let mut samples = Samples::default(); + for sample in results { + samples.reads.extend(sample.reads); + samples.writes.extend(sample.writes); + samples.closes.extend(sample.closes); + } + let r = samples.reads.len(); + let w = samples.writes.len(); + let close_count = samples.closes.len(); + let [r50, r95, r99] = percentiles(&mut samples.reads); + let [w50, w95, w99] = percentiles(&mut samples.writes); + let [c50, c95, c99] = percentiles(&mut samples.closes); + println!( + "{round},{name},{},{bs},{concurrency},{elapsed:.6},{r},{w},{:.3},{:.3},{:.3},{r50:.3},{r95:.3},{r99:.3},{w50:.3},{w95:.3},{w99:.3},{close_count},{c50:.3},{c95:.3},{c99:.3},{cpu:.3}", + workload.name(), + r as f64 * bs as f64 / elapsed / 1048576.0, + w as f64 * bs as f64 / elapsed / 1048576.0, + (r + w) as f64 / elapsed + ); + if w > 0 { + for worker in 0..concurrency { + let path = format!("output-{worker}"); + let buffer = operator + .read_with(&path) + .range(0..bs.min(4096) as u64) + .await?; + assert!(buffer.to_bytes().iter().all(|v| *v == 0x5a)); + operator.delete(&path).await?; + } + } + } + } + } + } + } + Ok(()) + } +} From 90bb5ace53f990a1b652417932f7dc5b1edd60e1 Mon Sep 17 00:00:00 2001 From: MrCroxx Date: Tue, 27 Oct 2026 11:42:43 +0800 Subject: [PATCH 4/6] bench(hf3fs): align cache policy and I/O chunks with Fs --- .github/services/hf3fs/README.md | 70 +++++++++----- .github/services/hf3fs/bench_direct.c | 99 ++++++++++++++++++++ core/services/hf3fs/examples/bench_matrix.rs | 75 +++++++++------ 3 files changed, 193 insertions(+), 51 deletions(-) create mode 100644 .github/services/hf3fs/bench_direct.c diff --git a/.github/services/hf3fs/README.md b/.github/services/hf3fs/README.md index f9cb450f761b..531cd03e485f 100644 --- a/.github/services/hf3fs/README.md +++ b/.github/services/hf3fs/README.md @@ -29,39 +29,61 @@ The benchmark compares 1 MiB streaming reads at concurrency 1 and 16 against `Fs` on the same file and mount. It warms each path and reports four measured passes. Results depend on client configuration, caching, and cluster load. -For the read/write matrix, run: +For the read/write matrix, build the example from `core/`. The default profile +requires direct I/O on the FUSE path. `Fs` has no direct-I/O builder option, so a +Linux preload helper adds `O_DIRECT` only to files below the dedicated benchmark +root. It also verifies each intercepted descriptor's flags. This helper is for +benchmarking on a real 3FS mount, not for application use. ```shell -cargo run --release -p opendal-service-hf3fs --example bench_matrix > matrix.csv +cargo build --release -p opendal-service-hf3fs --example bench_matrix +cc -shared -fPIC -O2 -Wall -Wextra -Werror \ + ../.github/services/hf3fs/bench_direct.c -o /tmp/opendal-bench-direct.so -ldl -pthread +OPENDAL_BENCH_DIRECT_ROOT="$OPENDAL_HF3FS_ROOT" \ + LD_PRELOAD=/tmp/opendal-bench-direct.so \ + target/release/examples/bench_matrix > matrix.csv ``` -The matrix uses 4 KiB, 64 KiB, 1 MiB, and 4 MiB application blocks with 1, 4, -16, and 64 concurrent workers. Each worker issues one application operation at -a time. Both services use the same 1 GiB input file. `Hf3fs` uses a queue depth -of 64 per direction and a native block size matching the application block. -`Fs` can split large application reads into smaller native requests. +The benchmark refuses to label a run `direct` unless the helper covers its root. +For a new client deployment, verify the data-file opens with `strace -f -e openat` +and check page residency with `mincore` before interpreting results. The helper +intercepts libc `open` and `open64`; it does not cover arbitrary direct syscalls +or relative `openat` calls. It prints the verified read/write open counts on exit. +Direct I/O bypasses the client's kernel page cache, not storage-server caches. + +Both services use one Operator, the same 1 GiB input file, application block +sizes of 4 KiB, 64 KiB, 1 MiB, and 4 MiB, and 1, 4, 16, or 64 concurrent workers. +Each worker issues one application operation at a time, with read and write +concurrency set to one. Both paths split application operations into service +chunks of `min(BS, 1 MiB)`; `Hf3fs` uses that native block size and queue depth 64 +per direction. Kernel and server request splitting can still differ. Internal +threads, copies, and ring scheduling remain properties of each implementation; +the benchmark does not add Operators to compensate for an implementation limit. +Each worker performs one untimed input read before the start barrier. Workloads include sequential range reads, random range reads, streaming writes, single-block object overwrites, and mixed random reads/overwrites with 70% or 50% reads by operation count. Mixed workloads use separate input and output files; they do not test random writes within an existing file. Overwrites include open, truncate, write, close, and fsync. Streaming writers close and synchronize -every 64 MiB and at the end; their write latency measures API admission, while -throughput includes final synchronization. These latency columns are not device -completion latencies. +every 64 MiB and at the end. Both services buffer writes, so streaming write +latency measures API admission; throughput includes all writes and final +synchronization. These latency columns are not device completion latencies. + +For application-level cache comparisons, run without the preload and explicitly +set `BENCH_CACHE=buffered` or `BENCH_CACHE=hot`. The latter warms the input through +an existing `Fs` reader after all workers open the file. USRBIO bypasses that +file cache in every profile. Keep buffered/hot results separate from the direct +comparison; they do not isolate data-path performance. CSV rows include the +cache profile and service chunk size to preserve these distinctions. -The default run preserves each service's normal cache behavior. Reopening the -file on some 3FS deployments invalidates its kernel page cache; initial prewarming -does not guarantee cache hits for new readers. Set `BENCH_WARM_READS=true` to -prewarm the whole input through an existing `Fs` reader after all workers open -the file. USRBIO bypasses that file cache. Neither profile isolates uncached -transport performance. The benchmark uses constant payloads and checks returned -lengths and sampled data. CPU usage covers the benchmark process, excluding the -FUSE process and servers. +Each case runs for 2 seconds and repeats 4 times. Rounds must be a positive even +number so each service runs first equally often. Set `BENCH_SECONDS`, +`BENCH_ROUNDS`, `BENCH_BS` (bytes), `BENCH_CONCURRENCY`, or `BENCH_WORKLOADS` to +override the defaults; the last three accept comma-separated lists. Use longer +cases and additional rounds to confirm differences found in the matrix. -Each case runs for 2 seconds and repeats 3 times, reversing service order on -alternate rounds. CSV rows contain counts, throughput, IOPS, latency percentiles, -and CPU usage. Set `BENCH_SECONDS`, `BENCH_ROUNDS`, `BENCH_BS` (bytes), -`BENCH_CONCURRENCY`, or `BENCH_WORKLOADS` to override the defaults; the last three -accept comma-separated lists. Short cases and shared-cluster load limit the -precision of tail latencies. Temporary files are removed after a successful run. +The benchmark uses constant payloads and checks returned lengths and sampled +data. CPU usage covers only the benchmark process, excluding the FUSE process +and servers. Short cases and shared-cluster load limit the precision of tail +latencies. Temporary files are removed after a successful run. diff --git a/.github/services/hf3fs/bench_direct.c b/.github/services/hf3fs/bench_direct.c new file mode 100644 index 000000000000..ab8b8c1585ad --- /dev/null +++ b/.github/services/hf3fs/bench_direct.c @@ -0,0 +1,99 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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. + +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +static pthread_once_t initialized = PTHREAD_ONCE_INIT; +static int (*next_open)(const char *, int, ...); +static int (*next_open64)(const char *, int, ...); +static char *root; +static size_t root_len; +static _Atomic unsigned long reads, writes; + +static void initialize(void) { + next_open = dlsym(RTLD_NEXT, "open"); + next_open64 = dlsym(RTLD_NEXT, "open64"); + const char *value = getenv("OPENDAL_BENCH_DIRECT_ROOT"); + if (!next_open || !next_open64 || !value || !(root = realpath(value, NULL))) { + fputs("direct I/O helper requires OPENDAL_BENCH_DIRECT_ROOT\n", stderr); + _exit(1); + } + root_len = strlen(root); +} + +/* Queried by the benchmark so a missing preload cannot silently use page cache. */ +int opendal_bench_direct_for(const char *path) { + pthread_once(&initialized, initialize); + return strncmp(path, root, root_len) == 0 && path[root_len] == '/'; +} + +static int open_file(const char *path, int flags, mode_t mode, int large) { + pthread_once(&initialized, initialize); + int direct = opendal_bench_direct_for(path) && !(flags & (O_DIRECTORY | O_PATH)); + int fd = (large ? next_open64 : next_open)(path, flags | (direct ? O_DIRECT : 0), mode); + if (fd >= 0 && direct) { + int actual = fcntl(fd, F_GETFL); + if (actual < 0 || !(actual & O_DIRECT)) { + close(fd); + errno = EIO; + return -1; + } + if ((flags & O_ACCMODE) == O_RDONLY) { + atomic_fetch_add(&reads, 1); + } else { + atomic_fetch_add(&writes, 1); + } + } + return fd; +} + +int open(const char *path, int flags, ...) { + mode_t mode = 0; + if ((flags & O_CREAT) || (flags & O_TMPFILE) == O_TMPFILE) { + va_list args; + va_start(args, flags); + mode = va_arg(args, mode_t); + va_end(args); + } + return open_file(path, flags, mode, 0); +} + +int open64(const char *path, int flags, ...) { + mode_t mode = 0; + if ((flags & O_CREAT) || (flags & O_TMPFILE) == O_TMPFILE) { + va_list args; + va_start(args, flags); + mode = va_arg(args, mode_t); + va_end(args); + } + return open_file(path, flags, mode, 1); +} + +__attribute__((destructor)) static void report(void) { + fprintf(stderr, "direct_io: verified_read_opens=%lu verified_write_opens=%lu\n", + atomic_load(&reads), atomic_load(&writes)); +} diff --git a/core/services/hf3fs/examples/bench_matrix.rs b/core/services/hf3fs/examples/bench_matrix.rs index 54df0278ad2d..a19a87a61772 100644 --- a/core/services/hf3fs/examples/bench_matrix.rs +++ b/core/services/hf3fs/examples/bench_matrix.rs @@ -29,10 +29,12 @@ async fn main() -> opendal_core::Result<()> { #[cfg(target_os = "linux")] mod bench { + use std::ffi::CString; use std::sync::Arc; use std::time::{Duration, Instant}; use futures::TryStreamExt; + use libloading::os::unix::Library; use opendal_core::{Buffer, Operator, Result}; use opendal_service_hf3fs::Hf3fs; use tokio::sync::Barrier; @@ -104,12 +106,28 @@ mod bench { pub async fn run() -> Result<()> { let root = std::env::var("OPENDAL_HF3FS_ROOT").expect("set OPENDAL_HF3FS_ROOT"); let directory = tempfile::tempdir_in(root).expect("create benchmark directory"); - let root = directory.path().to_str().unwrap(); + let root = std::fs::canonicalize(directory.path()).expect("canonical benchmark root"); + let root = root.to_str().unwrap(); + let cache = std::env::var("BENCH_CACHE").unwrap_or("direct".into()); + assert!(matches!(cache.as_str(), "direct" | "buffered" | "hot")); + // The preload helper exports this probe and checks the covered directory. + let direct = unsafe { + Library::this() + .get:: libc::c_int>( + b"opendal_bench_direct_for\0", + ) + .is_ok_and(|probe| probe(CString::new(root).unwrap().as_ptr()) != 0) + }; + assert_eq!( + cache == "direct", + direct, + "BENCH_CACHE=direct requires bench_direct.so covering OPENDAL_HF3FS_ROOT; buffered/hot require no preload" + ); let fs = Operator::new(opendal_service_fs::Fs::default().root(root))?; let seconds = setting("BENCH_SECONDS", 2.0_f64); - let rounds = setting("BENCH_ROUNDS", 3_usize); - let warm_reads = setting("BENCH_WARM_READS", false); - assert!(seconds.is_finite() && seconds > 0.0 && rounds > 0); + let rounds = setting("BENCH_ROUNDS", 4_usize); + let warm_reads = cache == "hot"; + assert!(seconds.is_finite() && seconds > 0.0 && rounds > 0 && rounds.is_multiple_of(2)); let duration = Duration::from_secs_f64(seconds); let block_sizes = std::env::var("BENCH_BS").unwrap_or("4096,65536,1048576,4194304".into()); let concurrencies = std::env::var("BENCH_CONCURRENCY").unwrap_or("1,4,16,64".into()); @@ -121,36 +139,30 @@ mod bench { writer.write(payload.clone()).await?; } writer.close().await?; - let mut warmup = fs.reader("input").await?.into_stream(..).await?; - while let Some(buffer) = warmup.try_next().await? { - std::hint::black_box(buffer); - } eprintln!( - "file_bytes={FILE_SIZE} seconds={seconds} rounds={rounds} runtime_threads=8 queue_depth=64; warm_after_open={warm_reads}; Fs uses its default buffered read path; seqwrite latency measures write() admission; close/fsync is included in elapsed time" + "file_bytes={FILE_SIZE} seconds={seconds} rounds={rounds} runtime_threads=8 queue_depth=64; cache={cache}; each service has one Operator; native chunks are at most 1 MiB; seqwrite latency measures write() admission; close/fsync is included in elapsed time" ); println!( - "round,service,workload,bs,concurrency,elapsed_s,read_ops,write_ops,read_mib_s,write_mib_s,iops,read_p50_us,read_p95_us,read_p99_us,write_p50_us,write_p95_us,write_p99_us,close_count,close_p50_us,close_p95_us,close_p99_us,cpu_cores" + "cache,round,service,workload,bs,io_bs,concurrency,elapsed_s,read_ops,write_ops,read_mib_s,write_mib_s,iops,read_p50_us,read_p95_us,read_p99_us,write_p50_us,write_p95_us,write_p99_us,close_count,close_p50_us,close_p95_us,close_p99_us,cpu_cores" ); for round in 0..rounds { for size in block_sizes.split(',') { let bs: usize = size.parse().expect("BENCH_BS"); - assert!(bs > 0 && bs <= payload.len() && FILE_SIZE.is_multiple_of(bs)); - let mut builder = Hf3fs::default().root(root).queue_depth(64).block_size(bs); + assert!( + bs >= 4096 + && bs.is_multiple_of(4096) + && bs <= payload.len() + && FILE_SIZE.is_multiple_of(bs) + ); + let io_bs = bs.min(1024 * 1024); + let mut builder = Hf3fs::default() + .root(root) + .queue_depth(64) + .block_size(io_bs); if let Ok(path) = std::env::var("OPENDAL_HF3FS_LIBRARY_PATH") { builder = builder.library_path(&path); } let hf3fs = Operator::new(builder)?; - // Warm the native data path as well; it does not use the kernel page cache. - let mut warmup = hf3fs - .reader_with("input") - .chunk(4 * 1024 * 1024) - .concurrent(16) - .await? - .into_stream(0..SEGMENT_SIZE as u64) - .await?; - while let Some(buffer) = warmup.try_next().await? { - std::hint::black_box(buffer); - } for concurrency in concurrencies.split(',') { let concurrency: usize = concurrency.parse().expect("BENCH_CONCURRENCY"); assert!(concurrency > 0 && concurrency <= 64); @@ -180,13 +192,17 @@ mod bench { let payload = payload.slice(..bs); let reader = operator .reader_with("input") - .chunk(bs) + .chunk(io_bs) .concurrent(1) .await?; reader.read(0..bs as u64).await?; warm_reader = Some(reader.clone()); let path = format!("output-{worker}"); - let mut writer = operator.writer_with(&path).concurrent(1).await?; + let mut writer = operator + .writer_with(&path) + .chunk(io_bs) + .concurrent(1) + .await?; tasks.push(tokio::spawn(async move { let mut samples = Samples::default(); let mut rng = worker as u64 + 1; @@ -228,12 +244,17 @@ mod bench { .push(start.elapsed().as_secs_f64() * 1e6); writer = operator .writer_with(&path) + .chunk(io_bs) .concurrent(1) .await?; written = 0; } } else { - operator.write(&path, payload.clone()).await?; + operator + .write_with(&path, payload.clone()) + .chunk(io_bs) + .concurrent(1) + .await?; samples .writes .push(start.elapsed().as_secs_f64() * 1e6); @@ -278,7 +299,7 @@ mod bench { let [w50, w95, w99] = percentiles(&mut samples.writes); let [c50, c95, c99] = percentiles(&mut samples.closes); println!( - "{round},{name},{},{bs},{concurrency},{elapsed:.6},{r},{w},{:.3},{:.3},{:.3},{r50:.3},{r95:.3},{r99:.3},{w50:.3},{w95:.3},{w99:.3},{close_count},{c50:.3},{c95:.3},{c99:.3},{cpu:.3}", + "{cache},{round},{name},{},{bs},{io_bs},{concurrency},{elapsed:.6},{r},{w},{:.3},{:.3},{:.3},{r50:.3},{r95:.3},{r99:.3},{w50:.3},{w95:.3},{w99:.3},{close_count},{c50:.3},{c95:.3},{c99:.3},{cpu:.3}", workload.name(), r as f64 * bs as f64 / elapsed / 1048576.0, w as f64 * bs as f64 / elapsed / 1048576.0, From f8bf69bc670f2addfce729e5046a6db397abd6a8 Mon Sep 17 00:00:00 2001 From: MrCroxx Date: Tue, 27 Oct 2026 14:21:53 +0800 Subject: [PATCH 5/6] fix(hf3fs): address CI documentation and code generation checks --- .github/services/hf3fs/README.md | 2 + .github/services/hf3fs/bench_direct.c | 34 ++++++++-------- .github/workflows/test_behavior_core.yml | 1 + core/services/hf3fs/src/ffi.rs | 2 +- dev/src/generate/dotnet.rs | 2 +- dev/src/generate/java.rs | 2 +- dev/src/generate/python.rs | 2 +- website/data/services.json | 51 ++++++++++++++++++++++++ 8 files changed, 76 insertions(+), 20 deletions(-) diff --git a/.github/services/hf3fs/README.md b/.github/services/hf3fs/README.md index 531cd03e485f..911ca58016a1 100644 --- a/.github/services/hf3fs/README.md +++ b/.github/services/hf3fs/README.md @@ -8,6 +8,8 @@ current Azure kernel disables RXE, so the behavior planner selects Ubuntu 22.04. CI caches the pinned 3FS build, runs the focused USRBIO tests and the standard behavior suite, and stops the cluster and removes the RXE device afterward. The first build is expensive; subsequent jobs restore the compiled backend. +The job skips the common setup's standalone RocksDB build because the Rust +service does not depend on it. 3FS builds its own bundled RocksDB. To test an existing Linux deployment, use a real 3FS FUSE mount and the matching USRBIO client library. diff --git a/.github/services/hf3fs/bench_direct.c b/.github/services/hf3fs/bench_direct.c index ab8b8c1585ad..4e4f26abc7c3 100644 --- a/.github/services/hf3fs/bench_direct.c +++ b/.github/services/hf3fs/bench_direct.c @@ -1,19 +1,21 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// 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. +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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. + */ #define _GNU_SOURCE #include diff --git a/.github/workflows/test_behavior_core.yml b/.github/workflows/test_behavior_core.yml index c55366adacc1..b29696709d72 100644 --- a/.github/workflows/test_behavior_core.yml +++ b/.github/workflows/test_behavior_core.yml @@ -41,6 +41,7 @@ jobs: uses: ./.github/actions/setup with: need-protoc: true + # 3FS builds its bundled RocksDB; its Rust service does not use this copy. need-rocksdb: ${{ matrix.cases.service != 'hf3fs' }} github-token: ${{ secrets.GITHUB_TOKEN }} diff --git a/core/services/hf3fs/src/ffi.rs b/core/services/hf3fs/src/ffi.rs index 4ebfb06e9a72..8bdc3c394da7 100644 --- a/core/services/hf3fs/src/ffi.rs +++ b/core/services/hf3fs/src/ffi.rs @@ -16,7 +16,7 @@ // under the License. //! Minimal bindings to the public `hf3fs_usrbio.h` ABI. -//! https://github.com/deepseek-ai/3FS/blob/22fca04564c7cc230fd8b9523b8b92864e1dad47/src/lib/api/hf3fs_usrbio.h +//! use std::ffi::{CStr, c_char, c_int, c_void}; use std::path::Path; diff --git a/dev/src/generate/dotnet.rs b/dev/src/generate/dotnet.rs index 5cb57af7b80d..7a184a92d420 100644 --- a/dev/src/generate/dotnet.rs +++ b/dev/src/generate/dotnet.rs @@ -324,7 +324,7 @@ fn metadata_field(f: metadata::MetadataField) -> MetadataField { fn enabled_service(srv: &str) -> bool { match srv { // not enabled in bindings/dotnet/Cargo.toml - "foundationdb" | "foyer" | "ftp" | "hdfs" | "rocksdb" => false, + "foundationdb" | "foyer" | "ftp" | "hdfs" | "hf3fs" | "rocksdb" => false, _ => true, } } diff --git a/dev/src/generate/java.rs b/dev/src/generate/java.rs index c25ea75af50f..d26d5492976b 100644 --- a/dev/src/generate/java.rs +++ b/dev/src/generate/java.rs @@ -25,7 +25,7 @@ use std::path::PathBuf; fn enabled_service(srv: &str) -> bool { match srv { // not enabled in bindings/java/Cargo.toml - "foundationdb" | "ftp" | "hdfs" | "rocksdb" | "tikv" => false, + "foundationdb" | "ftp" | "hdfs" | "hf3fs" | "rocksdb" | "tikv" => false, _ => true, } } diff --git a/dev/src/generate/python.rs b/dev/src/generate/python.rs index 963048f2d0e8..ae8dcc81839b 100644 --- a/dev/src/generate/python.rs +++ b/dev/src/generate/python.rs @@ -29,7 +29,7 @@ fn enabled_service(srv: &str) -> bool { // not enabled in bindings/python/Cargo.toml "etcd" | "foundationdb" | "hdfs" | "rocksdb" | "tikv" | "github" | "cloudflare_kv" | "monoiofs" | "dbfs" | "surrealdb" | "d1" | "opfs" | "compfs" | "lakefs" | "pcloud" - | "vercel_blob" | "foyer" | "goosefs" => false, + | "vercel_blob" | "foyer" | "goosefs" | "hf3fs" => false, _ => true, } } diff --git a/website/data/services.json b/website/data/services.json index 1156882f31f3..8e5df21b2423 100644 --- a/website/data/services.json +++ b/website/data/services.json @@ -2357,6 +2357,57 @@ } ] }, + { + "name": "hf3fs", + "scheme": "hf3fs", + "groups": [ + "General" + ], + "configs": [ + { + "name": "root", + "type": "string", + "required": false, + "group": "General", + "comments": "Existing absolute directory on a mounted 3FS instance. Required." + }, + { + "name": "library_path", + "type": "string", + "required": false, + "group": "General", + "comments": "USRBIO shared library path. Defaults to `libhf3fs_api_shared.so`." + }, + { + "name": "queue_depth", + "type": "integer", + "required": false, + "group": "General", + "comments": "Maximum in-flight requests per read or write ring. Defaults to 32." + }, + { + "name": "block_size", + "type": "integer", + "required": false, + "group": "General", + "comments": "Maximum bytes per request. Defaults to 1 MiB.\nShared memory usage is approximately `3 * queue_depth * block_size`." + } + ], + "examples": [ + { + "binding": "rust", + "language": "rust", + "minimal": "use opendal::Operator;\n\nlet operator = Operator::via_iter(\"hf3fs\", [\n])?;", + "full": "use opendal::Operator;\n\nlet operator = Operator::via_iter(\"hf3fs\", [\n // Existing absolute directory on a mounted 3FS instance. Required.\n // (\"root\".to_string(), \"...\".to_string()),\n\n // USRBIO shared library path. Defaults to `libhf3fs_api_shared.so`.\n // (\"library_path\".to_string(), \"...\".to_string()),\n\n // Maximum in-flight requests per read or write ring. Defaults to 32.\n // (\"queue_depth\".to_string(), \"1000\".to_string()),\n\n // Maximum bytes per request. Defaults to 1 MiB.\n // Shared memory usage is approximately `3 * queue_depth * block_size`.\n // (\"block_size\".to_string(), \"1000\".to_string()),\n])?;" + }, + { + "binding": "go", + "language": "go", + "minimal": "import (\n\t\"github.com/apache/opendal-go-services/hf3fs\"\n\topendal \"github.com/apache/opendal/bindings/go\"\n)\n\noperator, err := opendal.NewOperator(hf3fs.Scheme, opendal.OperatorOptions{\n})", + "full": "import (\n\t\"github.com/apache/opendal-go-services/hf3fs\"\n\topendal \"github.com/apache/opendal/bindings/go\"\n)\n\noperator, err := opendal.NewOperator(hf3fs.Scheme, opendal.OperatorOptions{\n\t// Existing absolute directory on a mounted 3FS instance. Required.\n\t// \"root\": \"...\",\n\n\t// USRBIO shared library path. Defaults to `libhf3fs_api_shared.so`.\n\t// \"library_path\": \"...\",\n\n\t// Maximum in-flight requests per read or write ring. Defaults to 32.\n\t// \"queue_depth\": \"1000\",\n\n\t// Maximum bytes per request. Defaults to 1 MiB.\n\t// Shared memory usage is approximately `3 * queue_depth * block_size`.\n\t// \"block_size\": \"1000\",\n})" + } + ] + }, { "name": "http", "scheme": "http", From 9978ceab4fb72413e08956dadd0270fb93ca3ed8 Mon Sep 17 00:00:00 2001 From: MrCroxx Date: Tue, 27 Oct 2026 14:36:49 +0800 Subject: [PATCH 6/6] perf(hf3fs): copy writes before driver submission --- core/services/hf3fs/README.md | 5 +- core/services/hf3fs/src/driver.rs | 95 +++++++++++++++-------- core/services/hf3fs/tests/real_service.rs | 67 ++++++++++++++-- 3 files changed, 128 insertions(+), 39 deletions(-) diff --git a/core/services/hf3fs/README.md b/core/services/hf3fs/README.md index 5161af01053f..2005604ba2ba 100644 --- a/core/services/hf3fs/README.md +++ b/core/services/hf3fs/README.md @@ -44,7 +44,10 @@ Reads return shared-memory-backed buffers without copying while fewer than `queue_depth` slots remain retained by callers. Additional completions copy to ordinary buffers so collecting large files cannot exhaust the I/O pool and deadlock. Buffer clones retain their slot until the last clone is dropped, -even after the operator is dropped. Writes copy their input into shared memory. +even after the operator is dropped. Writes reserve a shared-memory slot and copy +their input on the calling task before submitting it to the driver. Concurrent +tasks can copy in parallel on a multithreaded runtime; the write pool remains +bounded by `queue_depth`. Cancelling an awaiting future does not cancel submitted native I/O. The driver retains the file and buffer until completion. If submission or completion diff --git a/core/services/hf3fs/src/driver.rs b/core/services/hf3fs/src/driver.rs index 81d8a381586a..8a4124119886 100644 --- a/core/services/hf3fs/src/driver.rs +++ b/core/services/hf3fs/src/driver.rs @@ -24,7 +24,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use bytes::Bytes; use opendal_core::raw::new_std_io_error; use opendal_core::{Buffer, Error, ErrorKind, Result}; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc, oneshot}; use crate::ffi::{Api, Cqe, Ior, Iov}; @@ -63,10 +63,11 @@ struct Pool { block_size: usize, retain_limit: usize, slots: Mutex, + permits: Arc, } -// The mapping is stable. Only the driver writes to exclusively checked-out -// slots; completed, retained slots are immutable until their final owner drops. +// The mapping is stable. Each slot has one owner until I/O completes; +// completed, retained read slots are immutable until their final owner drops. unsafe impl Send for Pool {} unsafe impl Sync for Pool {} @@ -92,6 +93,7 @@ impl Pool { iov, block_size, retain_limit: depth, + permits: Arc::new(Semaphore::new(depth)), slots: Mutex::new(Slots { free: (0..count).collect(), retained: 0, @@ -112,6 +114,7 @@ struct Slot { index: usize, len: usize, retained: bool, + _permit: Option, } impl AsRef<[u8]> for Slot { @@ -157,7 +160,7 @@ struct Request { file: Arc, offset: usize, size: usize, - data: Option, + slot: Option, reply: oneshot::Sender>, } @@ -166,11 +169,12 @@ struct Pending { slot: Slot, } -#[derive(Debug, Clone)] +#[derive(Clone)] pub(crate) struct IoDriver { tx: mpsc::Sender, pub block_size: usize, read: bool, + write_pool: Option>, } impl IoDriver { @@ -188,7 +192,7 @@ impl IoDriver { .spawn( move || match Worker::new(api, &mount, depth, block_size, read, rx) { Ok(mut worker) => { - if ready_tx.send(Ok(())).is_ok() + if ready_tx.send(Ok(worker.pool.clone())).is_ok() && let Err(error) = worker.run() { worker.failure = error.to_string(); @@ -200,13 +204,14 @@ impl IoDriver { }, ) .map_err(new_std_io_error)?; - ready_rx.recv().map_err(|e| { + let pool = ready_rx.recv().map_err(|e| { Error::new(ErrorKind::Unexpected, "USRBIO driver initialization failed").set_source(e) })??; Ok(Self { tx, block_size, read, + write_pool: (!read).then_some(pool), }) } @@ -235,12 +240,42 @@ impl IoDriver { .set_source(e) })?; let (reply, result) = oneshot::channel(); + let slot = if let Some(data) = data { + let pool = self.write_pool.as_ref().expect("write pool"); + let permit = pool + .permits + .clone() + .acquire_owned() + .await + .map_err(|_| Error::new(ErrorKind::Unexpected, "USRBIO driver stopped"))?; + let index = pool.slots.lock().unwrap().free.pop().expect("write slot"); + let slot = Slot { + pool: pool.clone(), + index, + len: size, + retained: false, + _permit: Some(permit), + }; + // Copy on the caller's thread before handing the exclusive slot to + // the driver, so concurrent writers do not serialize their copies. + let dst = unsafe { + std::slice::from_raw_parts_mut(pool.iov.base.add(index * pool.block_size), size) + }; + let mut offset = 0; + for part in data { + dst[offset..offset + part.len()].copy_from_slice(&part); + offset += part.len(); + } + Some(slot) + } else { + None + }; self.tx .send(Request { file, offset, size, - data, + slot, reply, }) .await @@ -302,34 +337,30 @@ impl Worker { }) } - fn prepare(&mut self, request: Request) -> Option { + fn prepare(&mut self, mut request: Request) -> Option { if request.reply.is_closed() { return None; } - let index = self - .pool - .slots - .lock() - .unwrap() - .free - .pop() - .expect("reserved I/O slot"); - let slot = Slot { - pool: self.pool.clone(), - index, - len: request.size, - retained: false, - }; + let slot = request.slot.take().unwrap_or_else(|| { + let index = self + .pool + .slots + .lock() + .unwrap() + .free + .pop() + .expect("reserved I/O slot"); + Slot { + pool: self.pool.clone(), + index, + len: request.size, + retained: false, + _permit: None, + } + }); + let index = slot.index; // The slot is exclusive, and the request fits within one Iov block. let ptr = unsafe { self.pool.iov.base.add(index * self.pool.block_size) }; - if let Some(data) = &request.data { - let dst = unsafe { std::slice::from_raw_parts_mut(ptr, request.size) }; - let mut offset = 0; - for part in data.clone() { - dst[offset..offset + part.len()].copy_from_slice(&part); - offset += part.len(); - } - } let result = unsafe { (self.pool.api.prep_io)( self.ring.as_ref().unwrap(), @@ -343,6 +374,7 @@ impl Worker { ) }; if result == -libc::EAGAIN { + request.slot = Some(slot); return Some(request); } else if result < 0 { let _ = request @@ -450,6 +482,7 @@ impl Worker { impl Drop for Worker { fn drop(&mut self) { + self.pool.permits.close(); let Some(mut ring) = self.ring.take() else { return; }; diff --git a/core/services/hf3fs/tests/real_service.rs b/core/services/hf3fs/tests/real_service.rs index 407552c91245..278849d9ae93 100644 --- a/core/services/hf3fs/tests/real_service.rs +++ b/core/services/hf3fs/tests/real_service.rs @@ -23,14 +23,14 @@ use futures::TryStreamExt; use opendal_core::{ErrorKind, Operator}; use opendal_service_hf3fs::Hf3fs; -fn fixture() -> (tempfile::TempDir, Operator) { +fn fixture(block_size: usize) -> (tempfile::TempDir, Operator) { let root = std::env::var("OPENDAL_HF3FS_ROOT") .expect("set OPENDAL_HF3FS_ROOT to a real 3FS test directory"); let dir = tempfile::tempdir_in(root).unwrap(); let mut builder = Hf3fs::default() .root(dir.path().to_str().unwrap()) .queue_depth(2) - .block_size(4096); + .block_size(block_size); if let Ok(library) = std::env::var("OPENDAL_HF3FS_LIBRARY_PATH") { builder = builder.library_path(&library); } @@ -40,7 +40,7 @@ fn fixture() -> (tempfile::TempDir, Operator) { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[ignore = "requires a real 3FS mount and USRBIO library"] async fn read_write_ranges_and_metadata() { - let (dir, op) = fixture(); + let (dir, op) = fixture(4096); let data: Vec = (0..131_137).map(|i| (i % 251) as u8).collect(); op.write("nested/data", data.clone()).await.unwrap(); assert_eq!(std::fs::read(dir.path().join("nested/data")).unwrap(), data); @@ -108,7 +108,7 @@ async fn read_write_ranges_and_metadata() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[ignore = "requires a real 3FS mount and USRBIO library"] async fn retained_buffers_survive_reuse_and_operator_drop() { - let (dir, op) = fixture(); + let (dir, op) = fixture(4096); std::fs::write(dir.path().join("data"), vec![0x31; 4096]).unwrap(); let first = op.read("data").await.unwrap(); let clone = first.clone(); @@ -126,7 +126,7 @@ async fn retained_buffers_survive_reuse_and_operator_drop() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[ignore = "requires a real 3FS mount and USRBIO library"] async fn concurrent_reads_and_cancelled_callers() { - let (dir, op) = fixture(); + let (dir, op) = fixture(4096); let data: Vec = (0..65_537).map(|i| (i % 251) as u8).collect(); std::fs::write(dir.path().join("data"), &data).unwrap(); let reader = op.reader("data").await.unwrap(); @@ -155,7 +155,7 @@ async fn concurrent_reads_and_cancelled_callers() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[ignore = "requires a real 3FS mount and USRBIO library"] async fn overwrite_empty_and_exclusive_writes() { - let (dir, op) = fixture(); + let (dir, op) = fixture(4096); op.write("data", vec![0x41; 9000]).await.unwrap(); let err = op .write_with("data", "replacement") @@ -176,7 +176,7 @@ async fn overwrite_empty_and_exclusive_writes() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[ignore = "requires a real 3FS mount and USRBIO library"] async fn concurrent_streaming_writes() { - let (dir, op) = fixture(); + let (dir, op) = fixture(4096); let mut writer = op .writer_with("data") .concurrent(4) @@ -191,3 +191,56 @@ async fn concurrent_streaming_writes() { assert_eq!(std::fs::read(dir.path().join("data")).unwrap(), expected); assert_eq!(op.read("data").await.unwrap().to_vec(), expected); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "requires a real 3FS mount and USRBIO library"] +async fn concurrent_writes_and_cancelled_callers() { + let (dir, op) = fixture(1024 * 1024); + let mut tasks = tokio::task::JoinSet::new(); + let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(33)); + let mut cancelled = Vec::new(); + for i in 0..32 { + let op = op.clone(); + let barrier = barrier.clone(); + let task = tasks.spawn(async move { + let path = format!("concurrent/{i}"); + let data: Vec = (0..2 * 1024 * 1024 + 137) + .map(|j| ((i + j) % 251) as u8) + .collect(); + barrier.wait().await; + op.write_with(&path, data.clone()) + .chunk(1024 * 1024) + .concurrent(4) + .await + .unwrap(); + (path, data) + }); + if i % 2 == 0 { + cancelled.push(task); + } + } + barrier.wait().await; + tokio::time::sleep(Duration::from_millis(10)).await; + for task in cancelled { + task.abort(); + } + tokio::time::timeout(Duration::from_secs(30), async { + while let Some(result) = tasks.join_next().await { + match result { + Ok((path, data)) => { + assert_eq!(std::fs::read(dir.path().join(path)).unwrap(), data); + } + Err(error) => assert!(error.is_cancelled()), + } + } + op.write("after-cancellation", vec![0x73; 65_537]) + .await + .unwrap(); + assert_eq!( + op.read("after-cancellation").await.unwrap().to_vec(), + vec![0x73; 65_537] + ); + }) + .await + .unwrap(); +}