fuse robustness

Ramon Nourequested to merge
rnou/fusefix into master

Merge request reports

include/client/fuse/fuse_client.hpp

+16 −1
Changes for include/client/fuse/fuse_client.hpp: 16 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -72,6 +72,7 @@ extern "C" { 
#include <string>

#include <mutex>

#include <cstdlib>

#include <atomic>



#ifdef __FreeBSD__

#include <sys/socket.h>
@@ -111,7 +112,18 @@ struct _uintptr_to_must_hold_fuse_ino_t_dummy_struct { 


struct Inode {

    std::string path;

    uint64_t lookup_count;

    std::atomic<uint64_t> lookup_count;



    Inode() : lookup_count(0) {}

    Inode(const std::string& p, uint64_t lc) : path(p), lookup_count(lc) {}

    Inode(const Inode& other) : path(other.path), lookup_count(other.lookup_count.load()) {}

    Inode& operator=(const Inode& other) {

        if (this != &other) {

            path = other.path;

            lookup_count.store(other.lookup_count.load());

        }

        return *this;

    }

};



enum {
@@ -135,6 +147,9 @@ struct u_data { 
    int xattr;

    char* mountpoint;

    double timeout;

    double entry_timeout;

    double attr_timeout;

    double negative_timeout;

    int cache;

    int timeout_set;

};

src/client/fuse/fuse_client.cpp

+239 −137

File changed.

Preview size limit exceeded, changes collapsed.

src/client/gkfs_metadata.cpp

+13 −1
Changes for src/client/gkfs_metadata.cpp: 13 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -1165,6 +1165,19 @@ gkfs_opendir(const std::string& path) { 
        ret.second->add("..", gkfs::filemap::FileType::directory);

        for(auto& fut : dcache_futures) {

            auto res = fut.get(); // Wait for the RPC result



            if(res.first != 0) {

                ret.first = res.first;

                LOG(ERROR, "{}() RPC failed with error: {}", __func__,

                    res.first);

                continue;

            }

            if(!res.second) {

                LOG(ERROR, "{}() RPC returned null entries vector", __func__);

                ret.first = EIO;

                continue;

            }



            auto& open_dir = *res.second;

            for(auto& dentry : open_dir) {

                // type returns as unsigned char
@@ -1186,7 +1199,6 @@ gkfs_opendir(const std::string& path) { 
                                                      get<3>(dentry)});

                cnt++;

            }

            ret.first = res.first;

        }

        LOG(DEBUG, "{}() Unpacked dirents for path '{}' counted '{}' entries",

            __func__, path, cnt);

tests/integration/fuse/test_metadata_bench.py

0 → 100644
+98 −0
Changes for tests/integration/fuse/test_metadata_bench.py: 98 added lines, 0 removed lines.
Original line number Diff line number Diff line 
import os

import time

import threading

import pytest

import statistics



def benchmark_metadata_ops(mountdir, num_threads, num_iters):

    def worker(tid, results):

        thread_dir = os.path.join(mountdir, f"bench_t{tid}")

        os.makedirs(thread_dir, exist_ok=True)

        

        start = time.perf_counter()

        for i in range(num_iters):

            # Create/Stat/Remove

            p = os.path.join(thread_dir, f"f{i}")

            with open(p, "w") as f:

                f.write("x")

            os.stat(p)

            os.remove(p)

        end = time.perf_counter()

        results.append(end - start)



    results = []

    threads = []

    for i in range(num_threads):

        t = threading.Thread(target=worker, args=(i, results))

        threads.append(t)

        t.start()

    

    for t in threads:

        t.join()

        

    total_time = max(results)

    ops_per_sec = (num_threads * num_iters * 3) / total_time

    return ops_per_sec



def benchmark_read_latency(mountdir, num_iters):

    p = os.path.join(mountdir, "baseline_file")

    with open(p, "w") as f:

        f.write("data")

    

    # Warm up

    os.stat(p)

    

    latencies = []

    for i in range(num_iters):

        start = time.perf_counter()

        os.stat(p)

        end = time.perf_counter()

        latencies.append(end - start)

        

    return statistics.median(latencies) * 1e6 # in microseconds



def benchmark_write_throughput(mountdir, num_threads, size_mb):

    def worker(tid, results):

        p = os.path.join(mountdir, f"write_bench_t{tid}")

        data = b"x" * 1024 * 1024 # 1MB

        

        start = time.perf_counter()

        with open(p, "wb") as f:

            for _ in range(size_mb):

                f.write(data)

            f.flush()

            os.fsync(f.fileno())

        end = time.perf_counter()

        results.append(end - start)



    results = []

    threads = []

    for i in range(num_threads):

        t = threading.Thread(target=worker, args=(i, results))

        threads.append(t)

        t.start()

    

    for t in threads:

        t.join()

        

    total_time = max(results)

    throughput_mb = (num_threads * size_mb) / total_time

    return throughput_mb



@pytest.mark.parametrize("num_threads", [1, 8])

def test_benchmark_performance(gkfs_daemon, fuse_client, num_threads):

    mountdir = fuse_client.mountdir

    iters = 100

    

    print(f"\n--- Benchmarking with {num_threads} threads ---")

    

    # Run with current settings

    ops = benchmark_metadata_ops(mountdir, num_threads, iters)

    print(f"Metadata Throughput: {ops:.2f} ops/sec")

    

    lat = benchmark_read_latency(mountdir, iters)

    print(f"Stat Latency (Median): {lat:.2f} us")



    if num_threads == 1:

        tp = benchmark_write_throughput(mountdir, 1, 100)

        print(f"Write Throughput (1 thread): {tp:.2f} MB/s")

tests/integration/fuse/test_stress.py

0 → 100644
+45 −0
Changes for tests/integration/fuse/test_stress.py: 45 added lines, 0 removed lines.
Original line number Diff line number Diff line 
import os

import threading

import pytest

import sh



def thread_task(mountdir, thread_id, num_iters):

    thread_dir = os.path.join(mountdir, f"thread_{thread_id}")

    os.makedirs(thread_dir, exist_ok=True)

    

    for i in range(num_iters):

        file_path = os.path.join(thread_dir, f"file_{i}")

        # Create

        with open(file_path, "w") as f:

            f.write("data")

        # Stat

        os.stat(file_path)

        # Unlink

        os.remove(file_path)

        

        # Subdir create/rmdir

        subdir = os.path.join(thread_dir, f"dir_{i}")

        os.mkdir(subdir)

        os.rmdir(subdir)



@pytest.mark.parametrize("num_threads", [10])

@pytest.mark.parametrize("iters_per_thread", [100])

def test_metadata_stress(gkfs_daemon, fuse_client, num_threads, iters_per_thread):

    mountdir = fuse_client.mountdir

    threads = []

    

    for i in range(num_threads):

        t = threading.Thread(target=thread_task, args=(mountdir, i, iters_per_thread))

        threads.append(t)

        t.start()

        

    for t in threads:

        t.join()

        

    # Final check of the root directory

    assert "thread_0" in os.listdir(mountdir)

    

    # Cleanup

    for i in range(num_threads):

        thread_dir = os.path.join(mountdir, f"thread_{i}")

        sh.rm("-rf", thread_dir)