From 61e256a5e4037cfb29c303cf9f43379cf303aa9a Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:14:47 +0200 Subject: [PATCH 1/6] Bound the hydration wait and close the open() registration race First, the job was registered by the socket thread only after the send succeeded. The job is now inserted before the message is posted, so no waiter can observe the absence of a job it just posted. Second, SharedMap now carries a condition variable and SharedMap::waitForJob() blocks on it with a deadline. Also made _transfer_id atomic. --- INTEGRATION.md | 14 +++++ src/openvfsfuse/config.json | 3 +- src/openvfsfuse/main.cpp | 30 ++++++++-- src/openvfsfuse/openvfsfuse.cpp | 96 ++++++++++++-------------------- src/openvfsfuse/openvfsfuse.h | 5 ++ src/openvfsfuse/sharedmap.cpp | 68 +++++++++++++++++----- src/openvfsfuse/sharedmap.h | 32 +++++++++++ src/openvfsfuse/socketthread.cpp | 21 ++++--- src/openvfsfuse/socketthread.h | 3 +- 9 files changed, 184 insertions(+), 88 deletions(-) diff --git a/INTEGRATION.md b/INTEGRATION.md index 4c24364..f7d063b 100644 --- a/INTEGRATION.md +++ b/INTEGRATION.md @@ -15,6 +15,16 @@ The openvfs binary accepts the following command-line parameters: The FUSE layer must be unmounted when the desktop client shuts down or when a sync connection is removed. +## Configuration File + +The file passed via `-i` is JSON. All keys are optional; a missing key falls back to the built-in default. + +* `ignoreApps.byName`: List of absolute executable paths that must not trigger a hydration. A matching caller receives `EPERM` from `open()` instead of the file being downloaded. +* `ignoreApps.endsWith`: List of suffixes matched against the calling executable's path, for the same purpose. This is the distribution-independent way to name a binary, and `-thumbnailer` catches the freedesktop thumbnailer convention in one entry. +* `hydrationTimeoutSeconds`: How long `open()` blocks waiting for the client to hydrate a file before giving up and returning `ETIMEDOUT`. Defaults to `300`. Raise it if large files are legitimately transferred over slow links; the point of the bound is to keep an unresponsive client from wedging the calling application indefinitely. + +The shipped `config.json` is a cross-desktop starting point, not an exhaustive list. Any indexer, thumbnailer, or antivirus that opens files as a matter of routine will otherwise download the whole sync root. + ## Socket API Communication OpenVFS communicates with the desktop client over a local, unauthenticated Unix socket — the so-called socket API. This bidirectional channel is already used by both OpenCloud and Nextcloud clients for file manager integration. @@ -49,3 +59,7 @@ Expected response fields: On a successful response, the file is considered hydrated and its content is available locally. +Messages are newline-delimited on a stream socket, so a reply may be split across several writes and several replies may be batched into one. openVFS reassembles them; the client need not size its replies to any particular limit. + +If the client reports an error, `open()` fails with `EIO`. If it does not answer within `hydrationTimeoutSeconds`, `open()` fails with `ETIMEDOUT`. + diff --git a/src/openvfsfuse/config.json b/src/openvfsfuse/config.json index 61e5191..3b77c3b 100644 --- a/src/openvfsfuse/config.json +++ b/src/openvfsfuse/config.json @@ -34,5 +34,6 @@ "clamscan", "freshclam" ] - } + }, + "hydrationTimeoutSeconds": 300 } diff --git a/src/openvfsfuse/main.cpp b/src/openvfsfuse/main.cpp index d414f01..cae0040 100644 --- a/src/openvfsfuse/main.cpp +++ b/src/openvfsfuse/main.cpp @@ -5,6 +5,7 @@ #include "openvfsfuse.h" #include +#include #include #include #include @@ -24,6 +25,7 @@ constexpr int MaxFuseArgs = 32; const std::string ConfigIgnoreAppsStr = "ignoreApps"; const std::string ConfigByNameStr = "byName"; const std::string ConfigEndsWith = "endsWith"; +const std::string ConfigHydrationTimeoutSecondsStr = "hydrationTimeoutSeconds"; void usage(char *name) { @@ -72,10 +74,30 @@ std::optional processArgs(int argc, char *argv[]) break; case 'i': { std::ifstream ifs(optarg); - json data = json::parse(ifs); - - out.appsNoHydrateFull = data[ConfigIgnoreAppsStr][ConfigByNameStr].get>(); - out.appsNoHydrateEndsWith = data[ConfigIgnoreAppsStr][ConfigEndsWith].get>(); + if (!ifs) { + std::cerr << "Failed to open config file " << optarg << std::endl; + return {}; + } + json data; + try { + data = json::parse(ifs); + } catch (const json::exception &e) { + std::cerr << "Failed to parse config file " << optarg << ": " << e.what() << std::endl; + return {}; + } + + // Every setting is optional so that a config file written for an + // older version keeps working. + const auto ignoreApps = data.value(ConfigIgnoreAppsStr, json::object()); + out.appsNoHydrateFull = ignoreApps.value(ConfigByNameStr, std::vector{}); + out.appsNoHydrateEndsWith = ignoreApps.value(ConfigEndsWith, std::vector{}); + + const auto timeoutSeconds = data.value(ConfigHydrationTimeoutSecondsStr, 0); + if (timeoutSeconds > 0) { + out.hydrationTimeout = std::chrono::seconds(timeoutSeconds); + } else if (timeoutSeconds < 0) { + std::cerr << ConfigHydrationTimeoutSecondsStr << " must be positive, keeping the default" << std::endl; + } break; } case 'o': diff --git a/src/openvfsfuse/openvfsfuse.cpp b/src/openvfsfuse/openvfsfuse.cpp index 6ffa3cf..7e1881f 100644 --- a/src/openvfsfuse/openvfsfuse.cpp +++ b/src/openvfsfuse/openvfsfuse.cpp @@ -19,6 +19,7 @@ #include "sharedmap.h" #include "socketthread.h" +#include #include #include #include @@ -41,7 +42,6 @@ #include #endif -using namespace std::chrono_literals; using json = nlohmann::json; namespace { @@ -55,6 +55,7 @@ class VFSFuseContext , _rootHandle(open(_mountPoint.c_str(), 0)) , _appsNoHydrateFull(args.appsNoHydrateFull) , _appsNoHydrateEndsWith(args.appsNoHydrateEndsWith) + , _hydrationTimeout(args.hydrationTimeout) , _debugEnabled(args.debugEnabled) { assert(!_instance); @@ -103,6 +104,9 @@ class VFSFuseContext bool debugEnabled() const { return _debugEnabled; } + /// How long open() waits for the desktop client to hydrate a file. + auto hydrationTimeout() const { return _hydrationTimeout; } + private: static VFSFuseContext *_instance; std::filesystem::path _mountPoint; @@ -110,6 +114,7 @@ class VFSFuseContext int _rootHandle; std::vector _appsNoHydrateFull; std::vector _appsNoHydrateEndsWith; + std::chrono::milliseconds _hydrationTimeout; bool _debugEnabled; }; @@ -179,7 +184,10 @@ std::string getcallername(fuse_context *context) static SharedMap _jobs; static SocketThread _socketThread("SocketThread", _jobs); -static int _transfer_id{12}; +// FUSE dispatches requests from several threads, so the transfer id handed to +// the client must be incremented atomically -- two opens racing for the same id +// would share a single job entry. +static std::atomic _transfer_id{12}; void openvfsfuse_log(const std::string &path, const char *action, int returncode, const char *format, ...) @@ -575,69 +583,39 @@ static int openVFSfuse_open(const char *orig_path, struct fuse_file_info *fi) openvfsfuse_log(path, "open", 0, "Requesting hydration %s: %d", path.c_str(), msgData->id); + // Register the job *before* handing the request over. PostMsg() only + // queues, so the socket thread may well have sent the request and + // received the answer before this thread runs again. A waiter must + // never be able to observe the absence of a job it just posted. + _jobs.insert(msgData->id, HydJob{.state = HydJobState::Running}); + // push hydration request to the thread that handles the communication to the client - _socketThread.PostMsg(msgData); - - // the socketThread now talks to the client, which downloads the file for us. - // Here in this thread we enter a loop and wait for results - int cnt{0}; - int state{1}; - const auto MaxCnt{20}; - std::chrono::duration waitTime{10ms}; - std::chrono::duration dur{30ms}; - - HydJob hj; - - while (state == 1 && cnt++ < MaxCnt) { - std::this_thread::sleep_for(waitTime); // sleep for some time - waitTime += dur; - dur += waitTime; - // first: waittime: 10ms, dur: 30ms - // second:waittime: 40ms, dur: 70ms - // third: waittime: 110ms, dur: 180ms - // forth: waittime: 290ms, dur: 470ms - // fifth: waittime: 760ms, dur: 1230ms - // ... - - // check shared map and see if the id has changed to 0, which means success - // the value is changed in the other thread and fetched here - - if (!_jobs.get(msgData->id, hj)) { - // The job is no longer there :-/ - openvfsfuse_log(path, "open", 1, "Job queue does not have job %d", msgData->id); - state = -1; - } else { - state = hj.state; - openvfsfuse_log(path, "open", 1, "Found in job queue %d", state); - - // With all the state values except 1, the loop is left - if (state == 0) { - // success! - openvfsfuse_log(path, "open", 1, "Sucessfully finished job %d", msgData->id); - } else if (state == 1) { - // still running - } else if (state == -1) { - // fail - openvfsfuse_log(path, "open", 0, "Failed job %d", msgData->id); - } else if (state == 2) { - // timeout - openvfsfuse_log(path, "open", 0, "Job %d timed out", msgData->id); - } - } + if (!_socketThread.PostMsg(msgData)) { + _jobs.remove(msgData->id); + openvfsfuse_log(path, "open", 0, "Could not queue hydration request %d, shutting down", msgData->id); + return -EIO; } + // The socketThread now talks to the client, which downloads the file + // for us. Block here until the client answers or the timeout expires. + const auto result = _jobs.waitForJob(msgData->id, VFSFuseContext::instance().hydrationTimeout()); + // remove the job regardless of the result _jobs.remove(msgData->id); - if (state == -1 || state == 2) { - // Fail, job with ID was errornous - openvfsfuse_log(path, "open", 1, "ERROR while retrieving: %d", state); - return -ENOENT; - } - - if (cnt >= MaxCnt) { - openvfsfuse_log(path, "open", MaxCnt, "TIMEOUT - no answer from client"); - return -ENOENT; + switch (result) { + case HydJobResult::Succeeded: + openvfsfuse_log(path, "open", 0, "Sucessfully finished job %d", msgData->id); + break; + case HydJobResult::Failed: + openvfsfuse_log(path, "open", 0, "Failed job %d", msgData->id); + return -EIO; + case HydJobResult::TimedOut: + openvfsfuse_log(path, "open", 0, "TIMEOUT - no answer from client for job %d", msgData->id); + return -ETIMEDOUT; + case HydJobResult::Lost: + openvfsfuse_log(path, "open", 0, "Job queue does not have job %d", msgData->id); + return -EIO; } openvfsfuse_log(path, "open", 0, "-- open finished"); diff --git a/src/openvfsfuse/openvfsfuse.h b/src/openvfsfuse/openvfsfuse.h index 78b3c67..28ec585 100644 --- a/src/openvfsfuse/openvfsfuse.h +++ b/src/openvfsfuse/openvfsfuse.h @@ -12,6 +12,7 @@ #define _X_SOURCE 500 #endif +#include #include #include @@ -24,6 +25,10 @@ struct openVFSfuse_Args std::vector fuseArgv; std::vector appsNoHydrateFull; // these apps are not permitted to cause a dehydration std::vector appsNoHydrateEndsWith; + // Upper bound for how long open() blocks waiting for the desktop client to + // hydrate a file. A large file over a slow link is legitimately slow, an + // unresponsive client is not -- so this is configurable. + std::chrono::milliseconds hydrationTimeout = std::chrono::seconds(300); bool debugEnabled = false; // checked in the central logging function if logging is enabled }; diff --git a/src/openvfsfuse/sharedmap.cpp b/src/openvfsfuse/sharedmap.cpp index 3c9bf5e..df63b4f 100644 --- a/src/openvfsfuse/sharedmap.cpp +++ b/src/openvfsfuse/sharedmap.cpp @@ -21,12 +21,18 @@ #include "sharedmap.h" -SharedMap::SharedMap() { } +SharedMap::SharedMap() + : _pid(0) +{ +} void SharedMap::insert(int key, const HydJob &value) { - std::lock_guard lock(_mutex); - _data[key] = value; + { + std::lock_guard lock(_mutex); + _data[key] = value; + } + _cv.notify_all(); } bool SharedMap::get(int key, HydJob &outValue) @@ -42,25 +48,57 @@ bool SharedMap::get(int key, HydJob &outValue) bool SharedMap::set(int key, const HydJob &value) { - std::lock_guard lock(_mutex); + bool found{false}; + { + std::lock_guard lock(_mutex); - auto it = _data.find(key); - if (it != _data.end()) { - _data[key] = value; - return true; + auto it = _data.find(key); + if (it != _data.end()) { + it->second = value; + found = true; + } } - return false; + if (found) { + _cv.notify_all(); + } + return found; +} + +HydJobResult SharedMap::waitForJob(int key, std::chrono::milliseconds timeout) +{ + std::unique_lock lock(_mutex); + + const bool settled = _cv.wait_for(lock, timeout, [this, key] { + const auto it = _data.find(key); + return it == _data.end() || it->second.state != HydJobState::Running; + }); + + if (!settled) { + return HydJobResult::TimedOut; + } + + const auto it = _data.find(key); + if (it == _data.end()) { + return HydJobResult::Lost; + } + return it->second.state == HydJobState::Success ? HydJobResult::Succeeded : HydJobResult::Failed; } bool SharedMap::remove(int id) { - std::lock_guard lock(_mutex); - auto it = _data.find(id); - if (it != _data.end()) { - _data.erase(id); - return true; + bool found{false}; + { + std::lock_guard lock(_mutex); + auto it = _data.find(id); + if (it != _data.end()) { + _data.erase(it); + found = true; + } } - return false; + if (found) { + _cv.notify_all(); + } + return found; } void SharedMap::setDesktopClientPid(long pid) diff --git a/src/openvfsfuse/sharedmap.h b/src/openvfsfuse/sharedmap.h index c43151c..660ba1a 100644 --- a/src/openvfsfuse/sharedmap.h +++ b/src/openvfsfuse/sharedmap.h @@ -22,18 +22,40 @@ #ifndef SHAREDMAP_H #define SHAREDMAP_H +#include +#include #include #include #include #include #include +/** + * The values a hydration job can take. A job starts out Running and is moved to + * one of the other states by the socket thread. + */ +namespace HydJobState { +constexpr int Success = 0; +constexpr int Running = 1; +constexpr int Failed = -1; +} + struct HydJob { public: int state; }; +/** + * Outcome of waiting for a hydration job to leave the Running state. + */ +enum class HydJobResult { + Succeeded, ///< the client reported the file as hydrated + Failed, ///< the client reported an error, or the request could not be sent + TimedOut, ///< the client did not answer within the timeout + Lost, ///< the job vanished from the map, which should not happen +}; + class SharedMap { public: @@ -44,6 +66,15 @@ class SharedMap bool remove(int id); bool set(int key, const HydJob &value); + /** + * Block until the job identified by @p key leaves the Running state, or + * until @p timeout has elapsed. + * + * The wait is bounded by wall-clock time rather than by a number of polls, + * and a state change is observed as soon as the socket thread publishes it. + */ + HydJobResult waitForJob(int key, std::chrono::milliseconds timeout); + void printAll(); void setDesktopClientPid(long pid); long desktopClientPid(); @@ -51,6 +82,7 @@ class SharedMap private: std::map _data; std::mutex _mutex; + std::condition_variable _cv; long _pid; }; diff --git a/src/openvfsfuse/socketthread.cpp b/src/openvfsfuse/socketthread.cpp index 1ba8174..092d873 100644 --- a/src/openvfsfuse/socketthread.cpp +++ b/src/openvfsfuse/socketthread.cpp @@ -160,6 +160,9 @@ bool SocketThread::socketSendMsg(std::shared_ptr msgData) if (value < 0) { perror("write"); close(_socket); + // Mark the descriptor as gone so we do not keep reading from and + // writing to a closed fd on every subsequent tick. + _socket = -1; return false; } @@ -266,9 +269,9 @@ void SocketThread::handleReceivedMsg(const std::string &msg) } if (id > 0) { - int res{-1}; // Default set to fail + int res{HydJobState::Failed}; // Default set to fail if (status == "OK") { - res = 0; // good! + res = HydJobState::Success; // good! } else { cout << "ERROR from socket API for Id" << id << endl; } @@ -366,10 +369,10 @@ void SocketThread::ExitThread() //---------------------------------------------------------------------------- // PostMsg //---------------------------------------------------------------------------- -void SocketThread::PostMsg(std::shared_ptr data) +bool SocketThread::PostMsg(std::shared_ptr data) { if (m_exit.load()) - return; + return false; assert(m_thread); // Create a new ThreadMsg @@ -379,6 +382,7 @@ void SocketThread::PostMsg(std::shared_ptr data) std::unique_lock lk(m_mutex); m_queue.push(threadMsg); m_cv.notify_one(); + return true; } //---------------------------------------------------------------------------- @@ -439,11 +443,12 @@ void SocketThread::Process() if (!socketSendMsg(msgData)) { cout << "Failed to send msg " << msgData->id << ": " << msgData->msg.c_str() << endl; - } else { + // The job was registered by the caller before it posted the + // message, so a failure to send has to be published as such. + // Otherwise the caller waits out its full timeout for a + // request that never left the machine. if (msgData->id > 0) { - const HydJob hj{.state = 1}; - _sharedMap.insert(msgData->id, hj); - cout << "Storing sent message ID" << msgData->id << endl; + _sharedMap.set(msgData->id, HydJob{.state = HydJobState::Failed}); } } break; diff --git a/src/openvfsfuse/socketthread.h b/src/openvfsfuse/socketthread.h index 93b81ed..7b3a6bf 100644 --- a/src/openvfsfuse/socketthread.h +++ b/src/openvfsfuse/socketthread.h @@ -72,7 +72,8 @@ class SocketThread /// Add a message to the thread queue /// @param[in] data - thread specific message information - void PostMsg(std::shared_ptr msg); + /// @return False if the thread is shutting down and the message was dropped. + bool PostMsg(std::shared_ptr msg); /// Get size of thread message queue. size_t GetQueueSize(); From c0d8e20d2a6f793d89b83522b884abab29ed3f5c Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:35:15 +0200 Subject: [PATCH 2/6] Add a test for the socket framing and the hydration wait --- src/openvfsfuse/tests/CMakeLists.txt | 13 ++ src/openvfsfuse/tests/socketthreadtest.cpp | 169 +++++++++++++++++++++ 2 files changed, 182 insertions(+) create mode 100644 src/openvfsfuse/tests/socketthreadtest.cpp diff --git a/src/openvfsfuse/tests/CMakeLists.txt b/src/openvfsfuse/tests/CMakeLists.txt index 9af17b0..b1e847c 100644 --- a/src/openvfsfuse/tests/CMakeLists.txt +++ b/src/openvfsfuse/tests/CMakeLists.txt @@ -3,3 +3,16 @@ add_executable(strtools_test ../strtools.cpp ) add_test(NAME strtools_test COMMAND strtools_test) + +# The socket and job handling live in the openvfsfuse executable, so the test +# compiles those sources directly rather than linking a library. +add_executable(socketthreadtest + socketthreadtest.cpp + ../socketthread.cpp + ../sharedmap.cpp + ../strtools.cpp +) +target_include_directories(socketthreadtest PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/..) +target_link_libraries(socketthreadtest PRIVATE nlohmann_json::nlohmann_json Threads::Threads) +add_test(NAME socketthreadtest COMMAND socketthreadtest) + diff --git a/src/openvfsfuse/tests/socketthreadtest.cpp b/src/openvfsfuse/tests/socketthreadtest.cpp new file mode 100644 index 0000000..c0725ad --- /dev/null +++ b/src/openvfsfuse/tests/socketthreadtest.cpp @@ -0,0 +1,169 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +// SPDX-FileCopyrightText: 2026 toxicphreAK + +/* + * Drives SocketThread and SharedMap over a real AF_UNIX socket, standing in for + * the desktop client. Covers the stream framing and the hydration wait, both of + * which are timing dependent and regress silently. + */ + +#include "socketthread.h" +#include "sharedmap.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std::chrono_literals; + +namespace { + +int failures = 0; + +void check(bool ok, const std::string &what) +{ + std::cerr << (ok ? "ok : " : "FAIL : ") << what << std::endl; + if (!ok) { + ++failures; + } +} + +/// A socket path has to fit into sockaddr_un::sun_path, so keep it short and +/// out of the (potentially deeply nested) build directory. +std::string makeSocketPath() +{ + char tmpl[] = "/tmp/openvfs-test-XXXXXX"; + const char *dir = mkdtemp(tmpl); + if (!dir) { + std::cerr << "Failed to create a temporary directory" << std::endl; + std::exit(2); + } + return std::string(dir) + "/s"; +} + +int listenOn(const std::string &path) +{ + const int fd = socket(AF_UNIX, SOCK_STREAM, 0); + sockaddr_un addr{}; + addr.sun_family = AF_UNIX; + std::strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1); + if (bind(fd, reinterpret_cast(&addr), sizeof(addr)) != 0 || listen(fd, 1) != 0) { + std::cerr << "Failed to listen on " << path << ": " << std::strerror(errno) << std::endl; + std::exit(2); + } + return fd; +} + +void writeAll(int fd, const std::string &data) +{ + size_t offset = 0; + while (offset < data.size()) { + const ssize_t n = write(fd, data.data() + offset, data.size() - offset); + if (n <= 0) { + return; + } + offset += static_cast(n); + } +} + +std::string hydrateResult(int id, const std::string &argumentsBody) +{ + return "V2/HYDRATE_FILE_RESULT:{\"id\":\"" + std::to_string(id) + "\",\"arguments\":{" + argumentsBody + "}}\n"; +} + +} + +int main() +{ + const std::string socketPath = makeSocketPath(); + const int server = listenOn(socketPath); + + SharedMap jobs; + SocketThread socketThread("TestSocketThread", jobs); + socketThread.CreateThread(socketPath); + + const int client = accept(server, nullptr, nullptr); + if (client < 0) { + std::cerr << "SocketThread did not connect" << std::endl; + return 2; + } + + // Register a job the way openVFSfuse_open() does, before posting it. + const auto post = [&](int id) { + auto data = std::make_shared(); + data->msg = "V2/HYDRATE_FILE"; + data->file = "/tmp/some/file"; + data->id = id; + jobs.insert(id, HydJob{.state = HydJobState::Running}); + socketThread.PostMsg(data); + }; + + // A message split across two writes must not be acted on until it is complete. + writeAll(client, "VERSION:1.2.3:2"); + std::this_thread::sleep_for(750ms); + check(jobs.desktopClientPid() == 0, "an incomplete message is not dispatched"); + writeAll(client, ":4242\n"); + std::this_thread::sleep_for(750ms); + check(jobs.desktopClientPid() == 4242, "a message split across two writes is reassembled"); + + // A reply well past any single read buffer must arrive as one message. An + // error string carrying a path and a description clears 1 KB easily. + post(101); + writeAll(client, hydrateResult(101, "\"error\":\"" + std::string(8000, 'E') + "\"")); + check(jobs.waitForJob(101, 10s) == HydJobResult::Failed, "an 8 KB reply is parsed as a single message"); + + // Several replies batched into one write must all be dispatched. + post(102); + post(103); + writeAll(client, hydrateResult(102, "\"status\":\"OK\"") + hydrateResult(103, "\"status\":\"FAILED\"")); + check(jobs.waitForJob(102, 10s) == HydJobResult::Succeeded, "first of two batched replies is dispatched"); + check(jobs.waitForJob(103, 10s) == HydJobResult::Failed, "second of two batched replies is dispatched"); + + // ... and the stream is still in sync after all of the above. + post(104); + writeAll(client, hydrateResult(104, "\"status\":\"OK\"")); + check(jobs.waitForJob(104, 10s) == HydJobResult::Succeeded, "the stream stays in sync"); + + // A silent client must time out, bounded by wall-clock time. + post(105); + const auto beforeTimeout = std::chrono::steady_clock::now(); + const auto timedOut = jobs.waitForJob(105, 500ms); + const auto waited = std::chrono::steady_clock::now() - beforeTimeout; + check(timedOut == HydJobResult::TimedOut, "a silent client is reported as a timeout"); + check(waited >= 450ms && waited < 30s, "the timeout is bounded by wall-clock time"); + + // A reply arriving later must be observed without waiting out a long backoff. + post(106); + std::thread late([&] { + std::this_thread::sleep_for(300ms); + writeAll(client, hydrateResult(106, "\"status\":\"OK\"")); + }); + const auto beforeLate = std::chrono::steady_clock::now(); + const auto lateResult = jobs.waitForJob(106, 60s); + const auto latency = std::chrono::steady_clock::now() - beforeLate; + late.join(); + check(lateResult == HydJobResult::Succeeded, "a late reply is picked up"); + check(latency < 10s, "a late reply is not delayed by a growing backoff"); + + // A caller must never be left waiting for a message that was never queued. + socketThread.ExitThread(); + auto dropped = std::make_shared(); + dropped->msg = "V2/HYDRATE_FILE"; + dropped->id = 107; + check(!socketThread.PostMsg(dropped), "PostMsg reports messages dropped during shutdown"); + + close(client); + close(server); + std::error_code ec; + std::filesystem::remove_all(std::filesystem::path(socketPath).parent_path(), ec); + + std::cerr << (failures == 0 ? "All checks passed" : std::to_string(failures) + " check(s) failed") << std::endl; + return failures == 0 ? 0 : 1; +} From 2fd38eb008798ace2ea493fe17562f01e6f43c62 Mon Sep 17 00:00:00 2001 From: Klaas Freitag Date: Mon, 12 Oct 2026 17:56:52 +0200 Subject: [PATCH 3/6] Clean the code a bit, eliminate if chain --- src/openvfsfuse/openvfsfuse.cpp | 113 +++++++++++++++---------------- src/openvfsfuse/socketthread.cpp | 2 +- 2 files changed, 54 insertions(+), 61 deletions(-) diff --git a/src/openvfsfuse/openvfsfuse.cpp b/src/openvfsfuse/openvfsfuse.cpp index 7e1881f..f5c83fd 100644 --- a/src/openvfsfuse/openvfsfuse.cpp +++ b/src/openvfsfuse/openvfsfuse.cpp @@ -184,9 +184,7 @@ std::string getcallername(fuse_context *context) static SharedMap _jobs; static SocketThread _socketThread("SocketThread", _jobs); -// FUSE dispatches requests from several threads, so the transfer id handed to -// the client must be incremented atomically -- two opens racing for the same id -// would share a single job entry. +// FUSE dispatches requests from several threads, so the transfer id needs to be atomic static std::atomic _transfer_id{12}; @@ -562,66 +560,61 @@ static int openVFSfuse_open(const char *orig_path, struct fuse_file_info *fi) if (callerBelongsToProcess(fuse_get_context()->pid, _jobs.desktopClientPid())) { openvfsfuse_log(path, "open", res, "Desktop client tries to access, bypassing!"); } else { - if (const auto attribs = OpenVFS::PlaceHolderAttributes::fromAttributes(path)) { - if (attribs->state == OpenVFS::Constants::States::DeHydrated) { - // the file is virtual. It will be hydrated if the calling instance - // is not on the ignore list - - // ignore list of apps that must not cause a hydration - if (VFSFuseContext::instance().blockedToOpen(opener)) { - openvfsfuse_log(path, "open", 0, "blocked hydration for not permitted app %s", opener.c_str()); - return -EPERM; - } - - // Create message to send to worker thread 1 - std::shared_ptr msgData(new MsgData()); - msgData->msg = "V2/HYDRATE_FILE"; - msgData->file = VFSFuseContext::instance().getExternalPath(attribs->absolutePath); - msgData->fileId = attribs->fileId; - msgData->requester = getcallername(fuse_get_context()); - msgData->id = ++_transfer_id; // attention: _transfer_id is global and can change! - - openvfsfuse_log(path, "open", 0, "Requesting hydration %s: %d", path.c_str(), msgData->id); - - // Register the job *before* handing the request over. PostMsg() only - // queues, so the socket thread may well have sent the request and - // received the answer before this thread runs again. A waiter must - // never be able to observe the absence of a job it just posted. - _jobs.insert(msgData->id, HydJob{.state = HydJobState::Running}); - - // push hydration request to the thread that handles the communication to the client - if (!_socketThread.PostMsg(msgData)) { - _jobs.remove(msgData->id); - openvfsfuse_log(path, "open", 0, "Could not queue hydration request %d, shutting down", msgData->id); - return -EIO; - } - - // The socketThread now talks to the client, which downloads the file - // for us. Block here until the client answers or the timeout expires. - const auto result = _jobs.waitForJob(msgData->id, VFSFuseContext::instance().hydrationTimeout()); - - // remove the job regardless of the result + const auto attribs = OpenVFS::PlaceHolderAttributes::fromAttributes(path); + if (attribs && attribs->state == OpenVFS::Constants::States::DeHydrated) { + // the file is virtual. It will be hydrated if the calling instance + // is not on the ignore list + + // ignore list of apps that must not cause a hydration + if (VFSFuseContext::instance().blockedToOpen(opener)) { + openvfsfuse_log(path, "open", 0, "blocked hydration for not permitted app %s", opener.c_str()); + return -EPERM; + } + + // Create message to send to worker thread 1 + std::shared_ptr msgData(new MsgData()); + msgData->msg = "V2/HYDRATE_FILE"; + msgData->file = VFSFuseContext::instance().getExternalPath(attribs->absolutePath); + msgData->fileId = attribs->fileId; + msgData->requester = getcallername(fuse_get_context()); + msgData->id = ++_transfer_id; // attention: _transfer_id is global and can change! + + openvfsfuse_log(path, "open", 0, "Requesting hydration %s: %d", path.c_str(), msgData->id); + + // Register the job *before* handing the request over. PostMsg() only + // queues, so the socket thread may well have sent the request and + // received the answer before this thread runs again. A waiter must + // never be able to observe the absence of a job it just posted. + _jobs.insert(msgData->id, HydJob{.state = HydJobState::Running}); + + // push hydration request to the thread that handles the communication to the client + if (!_socketThread.PostMsg(msgData)) { _jobs.remove(msgData->id); + openvfsfuse_log(path, "open", 0, "Could not queue hydration request %d, shutting down", msgData->id); + return -EIO; + } - switch (result) { - case HydJobResult::Succeeded: - openvfsfuse_log(path, "open", 0, "Sucessfully finished job %d", msgData->id); - break; - case HydJobResult::Failed: - openvfsfuse_log(path, "open", 0, "Failed job %d", msgData->id); - return -EIO; - case HydJobResult::TimedOut: - openvfsfuse_log(path, "open", 0, "TIMEOUT - no answer from client for job %d", msgData->id); - return -ETIMEDOUT; - case HydJobResult::Lost: - openvfsfuse_log(path, "open", 0, "Job queue does not have job %d", msgData->id); - return -EIO; - } - - openvfsfuse_log(path, "open", 0, "-- open finished"); + // The socketThread now talks to the client, which downloads the file + // for us. Block here until the client answers or the timeout expires. + const auto result = _jobs.waitForJob(msgData->id, VFSFuseContext::instance().hydrationTimeout()); + + // remove the job regardless of the result + _jobs.remove(msgData->id); + + switch (result) { + case HydJobResult::Succeeded: + openvfsfuse_log(path, "open", 0, "Sucessfully finished job %d", msgData->id); + break; + case HydJobResult::Failed: + openvfsfuse_log(path, "open", 0, "Failed job %d", msgData->id); + return -EIO; + case HydJobResult::TimedOut: + openvfsfuse_log(path, "open", 0, "TIMEOUT - no answer from client for job %d", msgData->id); + return -ETIMEDOUT; + case HydJobResult::Lost: + openvfsfuse_log(path, "open", 0, "Job queue does not have job %d", msgData->id); + return -EIO; } - } else { - openvfsfuse_log(path, "open", 0, "Not a placeholder file"); } } // File is not dehydrated and it is just going to be opened diff --git a/src/openvfsfuse/socketthread.cpp b/src/openvfsfuse/socketthread.cpp index 092d873..9bb339b 100644 --- a/src/openvfsfuse/socketthread.cpp +++ b/src/openvfsfuse/socketthread.cpp @@ -141,7 +141,7 @@ bool SocketThread::socketSendMsg(std::shared_ptr msgData) std::string msg{msgData->msg}; // prepare the OpenCloud message to be sent over the wire - if (msg.empty()) { + if (msg.empty() || _socket == -1) { return false; } From 1dc29320681a740d2af6ffdb6e885cbf57a04865 Mon Sep 17 00:00:00 2001 From: Klaas Freitag Date: Tue, 20 Oct 2026 14:29:22 +0200 Subject: [PATCH 4/6] Few cleanups and error handling to address review comments --- src/openvfsfuse/openvfsfuse.cpp | 8 +++-- src/openvfsfuse/socketthread.cpp | 57 +++++++++++++++++++++++--------- src/openvfsfuse/socketthread.h | 3 +- 3 files changed, 48 insertions(+), 20 deletions(-) diff --git a/src/openvfsfuse/openvfsfuse.cpp b/src/openvfsfuse/openvfsfuse.cpp index f5c83fd..fd4728a 100644 --- a/src/openvfsfuse/openvfsfuse.cpp +++ b/src/openvfsfuse/openvfsfuse.cpp @@ -603,7 +603,7 @@ static int openVFSfuse_open(const char *orig_path, struct fuse_file_info *fi) switch (result) { case HydJobResult::Succeeded: - openvfsfuse_log(path, "open", 0, "Sucessfully finished job %d", msgData->id); + openvfsfuse_log(path, "open", 0, "Successfully finished job %d", msgData->id); break; case HydJobResult::Failed: openvfsfuse_log(path, "open", 0, "Failed job %d", msgData->id); @@ -802,8 +802,10 @@ int initializeOpenVFSFuse(openVFSfuse_Args &openVFSArgs) openVFSfuse_oper.removexattr = openVFSfuse_removexattr; assert(!openVFSArgs.socketPath.empty()); - _socketThread.CreateThread(openVFSArgs.socketPath); - + if (!_socketThread.CreateThread(openVFSArgs.socketPath)) { + std::cerr << "Unable to create the socket thread" << std::endl; + return -1; + } std::cout << "openVFSfuse starting: PID:" << getpid(); std::cout << ", Socket:" << openVFSArgs.socketPath << std::endl; diff --git a/src/openvfsfuse/socketthread.cpp b/src/openvfsfuse/socketthread.cpp index 9bb339b..8f880c7 100644 --- a/src/openvfsfuse/socketthread.cpp +++ b/src/openvfsfuse/socketthread.cpp @@ -72,6 +72,7 @@ SocketThread::SocketThread(const std::string &threadName, SharedMap &map) , m_exit(false) , m_timerExit(false) , THREAD_NAME(threadName) + , _socket(-1) , _sharedMap(map) { } @@ -89,6 +90,8 @@ SocketThread::~SocketThread() //---------------------------------------------------------------------------- bool SocketThread::CreateThread(const std::string& socketPath) { + bool re{true}; + if (!m_thread) { m_threadStartFuture = m_threadStartPromise.get_future(); @@ -99,33 +102,41 @@ bool SocketThread::CreateThread(const std::string& socketPath) // Wait for the thread to enter the Process method m_threadStartFuture.get(); + _socketPath = socketPath; - initSocket(socketPath); + re = initSocket(); } - return true; + return re; } -int SocketThread::initSocket(const std::string& socketPath) +bool SocketThread::initSocket() { - assert(!socketPath.empty()); + assert(!_socketPath.empty()); _socket = socket(AF_UNIX, SOCK_STREAM, 0); if (_socket < 0) { - return -1; + return false; } // Put the socket in non-blocking mode: if (fcntl(_socket, F_SETFL, fcntl(_socket, F_GETFL) | O_NONBLOCK) < 0) { // handle error + perror("fcntl"); + close(_socket); + _socket = -1; + return false; } struct sockaddr_un addr{}; addr.sun_family = AF_UNIX; - std::strncpy(addr.sun_path, socketPath.c_str(), sizeof(addr.sun_path) - 1); + std::strncpy(addr.sun_path, _socketPath.c_str(), sizeof(addr.sun_path) - 1); if (connect(_socket, (struct sockaddr *)&addr, sizeof(addr)) < 0) { - return -1; + perror("connect"); + close(_socket); + _socket = -1; + return false; } /* send a welcome message to the SocketAPI */ @@ -133,7 +144,7 @@ int SocketThread::initSocket(const std::string& socketPath) msgData->msg = "VERSION"; PostMsg(msgData); - return 1; + return true; } bool SocketThread::socketSendMsg(std::shared_ptr msgData) @@ -156,9 +167,12 @@ bool SocketThread::socketSendMsg(std::shared_ptr msgData) // send message ssize_t value; - value = write(_socket, msg.c_str(), msg.size()); + // the msg buffer is sent to the socket, and SIGPIPE is ignored. If SIGPIPE is not ignored, + // it could bring down the entire fuse layer in case the peer closes the socket, which would + // generate the signal + value = send(_socket, msg.c_str(), msg.size(), MSG_NOSIGNAL); if (value < 0) { - perror("write"); + perror("send"); close(_socket); // Mark the descriptor as gone so we do not keep reading from and // writing to a closed fd on every subsequent tick. @@ -173,7 +187,14 @@ bool SocketThread::socketSendMsg(std::shared_ptr msgData) void SocketThread::processSocketInput() { - assert(_socket >= 0); + if (_socket < 0) { + // the socket can only be -1 if the connection was closed by the peer. + // But this fuse layer is completely controlled by the client, which is + // the peer, so every reconnect attempt will fail. + // Also, if the client goes down and closes the socket, this fuse is also + // closed down. + return; + } // The socket is a SOCK_STREAM and carries no message boundaries: a single // read may return a fragment of a message, several messages at once, or @@ -341,12 +362,14 @@ void SocketThread::SetThreadName([[maybe_unused]] std::thread::native_handle_typ //---------------------------------------------------------------------------- void SocketThread::ExitThread() { - if (!m_thread) + if (!m_thread) { return; + } // Create a new ThreadMsg std::shared_ptr threadMsg(new ThreadMsg(MSG_EXIT_THREAD, nullptr)); + m_exit.store(true); // Put exit thread message into the queue { lock_guard lock(m_mutex); @@ -354,7 +377,6 @@ void SocketThread::ExitThread() m_cv.notify_one(); } - m_exit.store(true); m_thread->join(); // Clear the queue if anything added while waiting for join @@ -371,17 +393,20 @@ void SocketThread::ExitThread() //---------------------------------------------------------------------------- bool SocketThread::PostMsg(std::shared_ptr data) { - if (m_exit.load()) + if (m_exit.load()) { return false; + } assert(m_thread); // Create a new ThreadMsg std::shared_ptr threadMsg(new ThreadMsg(MSG_POST_USER_DATA, data)); // Add user data msg to queue and notify worker thread - std::unique_lock lk(m_mutex); + + lock_guard lock(m_mutex); m_queue.push(threadMsg); m_cv.notify_one(); + return true; } @@ -397,7 +422,7 @@ void SocketThread::TimerThread() std::shared_ptr threadMsg(new ThreadMsg(MSG_TIMER, nullptr)); // Add timer msg to queue and notify worker thread - std::unique_lock lk(m_mutex); + lock_guard lock(m_mutex); m_queue.push(threadMsg); m_cv.notify_one(); } diff --git a/src/openvfsfuse/socketthread.h b/src/openvfsfuse/socketthread.h index 7b3a6bf..dee5296 100644 --- a/src/openvfsfuse/socketthread.h +++ b/src/openvfsfuse/socketthread.h @@ -87,7 +87,7 @@ class SocketThread SocketThread &operator=(const SocketThread &) = delete; /// Initialise the socket to communicate with the client - int initSocket(const std::string& socketPath); + bool initSocket(); bool socketSendMsg(std::shared_ptr); @@ -111,6 +111,7 @@ class SocketThread std::atomic m_exit; std::atomic m_timerExit; const std::string THREAD_NAME; + std::string _socketPath; // Promise and future to synchronize thread start std::promise m_threadStartPromise; From 2552709c8692e7330362c70c835a92c62d731329 Mon Sep 17 00:00:00 2001 From: Klaas Freitag Date: Tue, 20 Oct 2026 15:12:58 +0200 Subject: [PATCH 5/6] Fix a potential loss of message when shutting down --- src/openvfsfuse/socketthread.cpp | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/openvfsfuse/socketthread.cpp b/src/openvfsfuse/socketthread.cpp index 8f880c7..f761745 100644 --- a/src/openvfsfuse/socketthread.cpp +++ b/src/openvfsfuse/socketthread.cpp @@ -369,10 +369,11 @@ void SocketThread::ExitThread() // Create a new ThreadMsg std::shared_ptr threadMsg(new ThreadMsg(MSG_EXIT_THREAD, nullptr)); - m_exit.store(true); // Put exit thread message into the queue { + // Set under the queue mutex so PostMsg() can't enqueue behind the exit message lock_guard lock(m_mutex); + m_exit.store(true); m_queue.push(threadMsg); m_cv.notify_one(); } @@ -393,6 +394,7 @@ void SocketThread::ExitThread() //---------------------------------------------------------------------------- bool SocketThread::PostMsg(std::shared_ptr data) { + lock_guard lock(m_mutex); if (m_exit.load()) { return false; } @@ -403,7 +405,6 @@ bool SocketThread::PostMsg(std::shared_ptr data) // Add user data msg to queue and notify worker thread - lock_guard lock(m_mutex); m_queue.push(threadMsg); m_cv.notify_one(); From 582a15f103d8e083a247447f2b163bbc3a3912e9 Mon Sep 17 00:00:00 2001 From: Klaas Freitag Date: Tue, 20 Oct 2026 17:48:04 +0200 Subject: [PATCH 6/6] Use enum class for HydJobState and some cleanups --- src/openvfsfuse/openvfsfuse.cpp | 2 +- src/openvfsfuse/sharedmap.cpp | 7 +++-- src/openvfsfuse/sharedmap.h | 35 +++++++++++++++------- src/openvfsfuse/socketthread.cpp | 8 ++--- src/openvfsfuse/tests/socketthreadtest.cpp | 2 +- 5 files changed, 34 insertions(+), 20 deletions(-) diff --git a/src/openvfsfuse/openvfsfuse.cpp b/src/openvfsfuse/openvfsfuse.cpp index fd4728a..1b1913d 100644 --- a/src/openvfsfuse/openvfsfuse.cpp +++ b/src/openvfsfuse/openvfsfuse.cpp @@ -585,7 +585,7 @@ static int openVFSfuse_open(const char *orig_path, struct fuse_file_info *fi) // queues, so the socket thread may well have sent the request and // received the answer before this thread runs again. A waiter must // never be able to observe the absence of a job it just posted. - _jobs.insert(msgData->id, HydJob{.state = HydJobState::Running}); + _jobs.insert(msgData->id, HydJob{._state = HydJobState::Running}); // push hydration request to the thread that handles the communication to the client if (!_socketThread.PostMsg(msgData)) { diff --git a/src/openvfsfuse/sharedmap.cpp b/src/openvfsfuse/sharedmap.cpp index df63b4f..64420dd 100644 --- a/src/openvfsfuse/sharedmap.cpp +++ b/src/openvfsfuse/sharedmap.cpp @@ -18,6 +18,7 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ +#include #include "sharedmap.h" @@ -70,7 +71,7 @@ HydJobResult SharedMap::waitForJob(int key, std::chrono::milliseconds timeout) const bool settled = _cv.wait_for(lock, timeout, [this, key] { const auto it = _data.find(key); - return it == _data.end() || it->second.state != HydJobState::Running; + return it == _data.end() || it->second._state != HydJobState::Running; }); if (!settled) { @@ -81,7 +82,7 @@ HydJobResult SharedMap::waitForJob(int key, std::chrono::milliseconds timeout) if (it == _data.end()) { return HydJobResult::Lost; } - return it->second.state == HydJobState::Success ? HydJobResult::Succeeded : HydJobResult::Failed; + return it->second._state == HydJobState::Success ? HydJobResult::Succeeded : HydJobResult::Failed; } bool SharedMap::remove(int id) @@ -117,6 +118,6 @@ void SharedMap::printAll() { std::lock_guard lock(_mutex); for (auto &[k, v] : _data) { - std::cout << k << ": " << v.state << "\n"; + std::cout << k << ": " << v.stateStr() << "\n"; } } diff --git a/src/openvfsfuse/sharedmap.h b/src/openvfsfuse/sharedmap.h index 660ba1a..94fa0ce 100644 --- a/src/openvfsfuse/sharedmap.h +++ b/src/openvfsfuse/sharedmap.h @@ -24,30 +24,42 @@ #include #include -#include #include #include #include -#include /** - * The values a hydration job can take. A job starts out Running and is moved to + * The states a hydration job can be in. A job starts out Running and is moved to * one of the other states by the socket thread. */ -namespace HydJobState { -constexpr int Success = 0; -constexpr int Running = 1; -constexpr int Failed = -1; -} +enum class HydJobState { Success = 0, Running = 1, Failed = -1 }; -struct HydJob +/** + * The HydJob class represents a hydration job that is stored in the job queue + * and passes data between fuse- and worker thread + */ +class HydJob { public: - int state; + HydJobState _state; + + std::string stateStr() const + { + switch (_state) { + case HydJobState::Success: + return "success"; + case HydJobState::Failed: + return "failed"; + case HydJobState::Running: + return "running"; + default: + return "undef"; + } + } }; /** - * Outcome of waiting for a hydration job to leave the Running state. + * Outcome of a hydration job after it left the Running state. */ enum class HydJobResult { Succeeded, ///< the client reported the file as hydrated @@ -56,6 +68,7 @@ enum class HydJobResult { Lost, ///< the job vanished from the map, which should not happen }; + class SharedMap { public: diff --git a/src/openvfsfuse/socketthread.cpp b/src/openvfsfuse/socketthread.cpp index f761745..3a03d40 100644 --- a/src/openvfsfuse/socketthread.cpp +++ b/src/openvfsfuse/socketthread.cpp @@ -290,20 +290,20 @@ void SocketThread::handleReceivedMsg(const std::string &msg) } if (id > 0) { - int res{HydJobState::Failed}; // Default set to fail + auto res{HydJobState::Failed}; // Default set to fail if (status == "OK") { res = HydJobState::Success; // good! } else { cout << "ERROR from socket API for Id" << id << endl; } - const HydJob hj{.state = res}; + const HydJob hj{._state = res}; bool ok = _sharedMap.set(id, hj); if (!ok) { // the id could not be set. That means, the job was not inserted. cout << "Job not found:" << id << endl; } else { - cout << "Setting Job ID " << id << " to result " << res << endl; + cout << "Setting Job ID " << id << " to result " << hj.stateStr() << endl; } } } else if (msgType == "VERSION") { @@ -474,7 +474,7 @@ void SocketThread::Process() // Otherwise the caller waits out its full timeout for a // request that never left the machine. if (msgData->id > 0) { - _sharedMap.set(msgData->id, HydJob{.state = HydJobState::Failed}); + _sharedMap.set(msgData->id, HydJob{._state = HydJobState::Failed}); } } break; diff --git a/src/openvfsfuse/tests/socketthreadtest.cpp b/src/openvfsfuse/tests/socketthreadtest.cpp index c0725ad..a64d4a5 100644 --- a/src/openvfsfuse/tests/socketthreadtest.cpp +++ b/src/openvfsfuse/tests/socketthreadtest.cpp @@ -101,7 +101,7 @@ int main() data->msg = "V2/HYDRATE_FILE"; data->file = "/tmp/some/file"; data->id = id; - jobs.insert(id, HydJob{.state = HydJobState::Running}); + jobs.insert(id, HydJob{._state = HydJobState::Running}); socketThread.PostMsg(data); };