diff --git a/.env.example b/.env.example index 5f07b8c97af5..9017c9591fac 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 +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..911ca58016a1 --- /dev/null +++ b/.github/services/hf3fs/README.md @@ -0,0 +1,91 @@ +# 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. +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. + +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. + +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 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 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. 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. + +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. + +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..4e4f26abc7c3 --- /dev/null +++ b/.github/services/hf3fs/bench_direct.c @@ -0,0 +1,101 @@ +/* + * 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/.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..b29696709d72 100644 --- a/.github/workflows/test_behavior_core.yml +++ b/.github/workflows/test_behavior_core.yml @@ -41,7 +41,8 @@ jobs: uses: ./.github/actions/setup with: need-protoc: true - need-rocksdb: 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 }} - name: Setup 1Password Connect @@ -58,3 +59,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..2005604ba2ba --- /dev/null +++ b/core/services/hf3fs/README.md @@ -0,0 +1,71 @@ +# 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 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 +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_matrix.rs b/core/services/hf3fs/examples/bench_matrix.rs new file mode 100644 index 000000000000..a19a87a61772 --- /dev/null +++ b/core/services/hf3fs/examples/bench_matrix.rs @@ -0,0 +1,326 @@ +// 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::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; + + 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 = 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", 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()); + 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?; + eprintln!( + "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!( + "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 >= 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)?; + 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(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) + .chunk(io_bs) + .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) + .chunk(io_bs) + .concurrent(1) + .await?; + written = 0; + } + } else { + operator + .write_with(&path, payload.clone()) + .chunk(io_bs) + .concurrent(1) + .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!( + "{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, + (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(()) + } +} 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..8a4124119886 --- /dev/null +++ b/core/services/hf3fs/src/driver.rs @@ -0,0 +1,506 @@ +// 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::{OwnedSemaphorePermit, Semaphore, 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, + permits: Arc, +} + +// 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 {} + +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, + permits: Arc::new(Semaphore::new(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, + _permit: Option, +} + +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, + slot: Option, + reply: oneshot::Sender>, +} + +struct Pending { + request: Request, + slot: Slot, +} + +#[derive(Clone)] +pub(crate) struct IoDriver { + tx: mpsc::Sender, + pub block_size: usize, + read: bool, + write_pool: Option>, +} + +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(worker.pool.clone())).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)?; + 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), + }) + } + + 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(); + 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, + slot, + 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, mut request: Request) -> Option { + if request.reply.is_closed() { + return None; + } + 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) }; + 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 { + request.slot = Some(slot); + 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) { + self.pool.permits.close(); + 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..8bdc3c394da7 --- /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. +//! + +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..278849d9ae93 --- /dev/null +++ b/core/services/hf3fs/tests/real_service.rs @@ -0,0 +1,246 @@ +// 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(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(block_size); + 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(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); + 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(4096); + 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(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(); + 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(4096); + 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(4096); + 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); +} + +#[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(); +} 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")] 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",