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..1b1913d 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,8 @@ 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 needs to be atomic +static std::atomic _transfer_id{12}; void openvfsfuse_log(const std::string &path, const char *action, int returncode, const char *format, ...) @@ -554,96 +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); - - // 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); - } - } - } - - // remove the job regardless of the result - _jobs.remove(msgData->id); + 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; + } - if (state == -1 || state == 2) { - // Fail, job with ID was errornous - openvfsfuse_log(path, "open", 1, "ERROR while retrieving: %d", state); - return -ENOENT; - } + // 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! - if (cnt >= MaxCnt) { - openvfsfuse_log(path, "open", MaxCnt, "TIMEOUT - no answer from client"); - return -ENOENT; - } + openvfsfuse_log(path, "open", 0, "Requesting hydration %s: %d", path.c_str(), msgData->id); - openvfsfuse_log(path, "open", 0, "-- open finished"); + // 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 + _jobs.remove(msgData->id); + + switch (result) { + case HydJobResult::Succeeded: + 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); + 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 @@ -831,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/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..64420dd 100644 --- a/src/openvfsfuse/sharedmap.cpp +++ b/src/openvfsfuse/sharedmap.cpp @@ -18,15 +18,22 @@ * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ +#include #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 +49,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) @@ -79,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 c43151c..94fa0ce 100644 --- a/src/openvfsfuse/sharedmap.h +++ b/src/openvfsfuse/sharedmap.h @@ -22,18 +22,53 @@ #ifndef SHAREDMAP_H #define SHAREDMAP_H -#include +#include +#include #include #include #include -#include -struct HydJob +/** + * 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. + */ +enum class HydJobState { Success = 0, Running = 1, Failed = -1 }; + +/** + * 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 a hydration job after it left 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 +79,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 +95,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..3a03d40 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) @@ -141,7 +152,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; } @@ -156,10 +167,16 @@ 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. + _socket = -1; return false; } @@ -170,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 @@ -266,20 +290,20 @@ void SocketThread::handleReceivedMsg(const std::string &msg) } if (id > 0) { - int res{-1}; // Default set to fail + auto 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; } - 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") { @@ -338,20 +362,22 @@ 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)); // 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(); } - m_exit.store(true); m_thread->join(); // Clear the queue if anything added while waiting for join @@ -366,19 +392,23 @@ void SocketThread::ExitThread() //---------------------------------------------------------------------------- // PostMsg //---------------------------------------------------------------------------- -void SocketThread::PostMsg(std::shared_ptr data) +bool SocketThread::PostMsg(std::shared_ptr data) { - if (m_exit.load()) - return; + lock_guard lock(m_mutex); + 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); + m_queue.push(threadMsg); m_cv.notify_one(); + + return true; } //---------------------------------------------------------------------------- @@ -393,7 +423,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(); } @@ -439,11 +469,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..dee5296 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(); @@ -86,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); @@ -110,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; 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..a64d4a5 --- /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; +}