Skip to content

Socket api across read 2 #73

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dragotin merged 3 commits into from
Oct 12, 2026
Merged

Socket api across read 2 #73

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4295aa2
Reassemble socket API messages across reads
toxicphreAK Aug 18, 2026
c5cd4de
Simplify the socket read loop
dragotin Oct 9, 2026
fb26ff9
Applied review feedback
dragotin Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
183 changes: 114 additions & 69 deletions src/openvfsfuse/socketthread.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
#include "sharedmap.h"
#include "strtools.h"

#include <cerrno>
#include <cstring>
#include <fcntl.h>
#include <iostream>
Expand All @@ -42,6 +43,14 @@ using namespace std;
#define MSG_POST_USER_DATA 2
#define MSG_TIMER 3

namespace {
/// Upper bound for the receive buffer. A message from the socket API is a
/// single JSON line and stays far below this; anything larger means the peer
/// is not speaking the protocol.
constexpr size_t MaxRxBufferSize = 1024 * 100;
constexpr size_t SingleReadBufferSize = 1024 * 4;
}

using json = nlohmann::json;

struct ThreadMsg
Expand Down Expand Up @@ -159,86 +168,126 @@ bool SocketThread::socketSendMsg(std::shared_ptr<MsgData> msgData)
// openvfsfuse_log(socket_path.c_str(), "socket send", value, "Message: %s", msg.c_str());
}

std::string SocketThread::readSocket()
void SocketThread::processSocketInput()
{
// read answer FIXME: Split messages by \n and keep the rest
char buf[1024];
ssize_t n = read(_socket, buf, sizeof(buf) - 1);
if (n <= 0)
return std::string();
return std::string(buf, n);
assert(_socket >= 0);

// 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
// both. Accumulate into _rxBuffer and only dispatch complete lines.
char buf[SingleReadBufferSize];

bool contRead{true};
ssize_t n;
while (contRead) {
n = read(_socket, buf, sizeof(buf));

if (n > 0) {
// all good, we read something and continue to do so
_rxBuffer.append(buf, n);
} else if (n == 0) {
// Peer closed the connection. Whatever is left in the buffer will
// be interpreted - and remaining incompleted reads will be erased.
std::cerr << "Socket closed by peer" << std::endl;
contRead = false;
} else if (n < 0) {
// error condition
if (errno == EINTR) {
// continue reading
} else {
// stop reading in all other error conditions
contRead = false;

// EAGAIN means that no data is available at the moment, which is
// not an issue for non blocking sockets. All other errnos are.
if (errno != EAGAIN

@TheOneRing TheOneRing Oct 12, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not that used to raw unix sockets could you add a comment that mentions that EAGAIN just mean, "nothing to read here, please go on" .

And would a while(true) { break; } loop be more conventional?

@dragotin dragotin Oct 12, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, please let me keep that state var 🙏

#if EAGAIN != EWOULDBLOCK
&& errno != EWOULDBLOCK
#endif
) {
// a serious errno was set
perror("socket read");
}
}
}
}

// handle the contents of _rxBuffer here
size_t newpos, pos{0};
while ((newpos = _rxBuffer.find('\n', pos)) != std::string::npos) {
handleReceivedMsg(_rxBuffer.substr(pos, newpos - pos));
pos = newpos + 1;
}
_rxBuffer.erase(0, pos); // clear everything that was processed

// A peer that never sends a newline must not be able to grow our buffer
// without bound.
// Also, if n == 0 because that are useless fragments after connection closed
if (n == 0 || _rxBuffer.size() > MaxRxBufferSize) {
std::cerr << "Discarding " << _rxBuffer.size() << " bytes from the socket API" << std::endl;
_rxBuffer.clear();
}
}

void SocketThread::handleReceivedMsg(const std::string &rawmsg)
void SocketThread::handleReceivedMsg(const std::string &msg)
{
if (rawmsg.empty()) {
cout << "Received Message empty" << endl;
string msgType, msgAttr;
if (msg.empty()) {
return;
}

auto copies = StrTools::split(rawmsg, 0x000A);
cout << "Handle single message " << msg << endl;

for (const string &msg : copies) {
string msgType, msgAttr;
if (msg.empty()) {
continue;
}
size_t found = msg.find(':');
if (found != string::npos) {
msgType = msg.substr(0, found);
msgAttr = msg.substr(found + 1, string::npos);
} else {
std::cerr << "Invalid message format: " << msg << std::endl;
return;
}

cout << "Handle single message " << msg << endl;
if (msgType == "V2/HYDRATE_FILE_RESULT") {
int id = -1;
std::string status;

size_t found = msg.find(':');
if (found != string::npos) {
msgType = msg.substr(0, found);
msgAttr = msg.substr(found + 1, string::npos);
} else {
std::cerr << "Invalid message format: " << msg << std::endl;
continue;
try {
const auto j = json::parse(msgAttr);
id = std::stoi(j["id"].get<string>());
const auto arguments = j["arguments"].get<json>();
if (arguments.contains("error")) {
std::cerr << "Error from socket API for Id " << id << ": " << arguments["error"].get<string>() << std::endl;
} else {
status = arguments["status"].get<string>();
}
} catch (json::exception &e) {
std::cerr << "Invalid JSON message: " << msgAttr << e.what() << std::endl;
return;
}

// FIXME: Think if splitting by newline makes sense

if (msgType == "V2/HYDRATE_FILE_RESULT") {
int id = -1;
std::string status;

try {
const auto j = json::parse(msgAttr);
id = std::stoi(j["id"].get<string>());
const auto arguments = j["arguments"].get<json>();
if (arguments.contains("error")) {
std::cerr << "Error from socket API for Id " << id << ": " << arguments["error"].get<string>() << std::endl;
} else {
status = arguments["status"].get<string>();
}
} catch (json::exception &e) {
std::cerr << "Invalid JSON message: " << msgAttr << e.what() << std::endl;
continue;
if (id > 0) {
int res{-1}; // Default set to fail
if (status == "OK") {
res = 0; // good!
} else {
cout << "ERROR from socket API for Id" << id << endl;
}

if (id > 0) {
int res{-1}; // Default set to fail
if (status == "OK") {
res = 0; // good!
} else {
cout << "ERROR from socket API for Id" << id << endl;
}

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;
}
}
} else if (msgType == "VERSION") {
vector<string> attribs = StrTools::split(msgAttr, ':');
if (attribs.size() == 3) {
cout << "Got PID of the Desktop Client: " << attribs.at(2) << endl;
_sharedMap.setDesktopClientPid(std::stol(attribs.at(2)));
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;
}
}
} else if (msgType == "VERSION") {
vector<string> attribs = StrTools::split(msgAttr, ':');
if (attribs.size() == 3) {
cout << "Got PID of the Desktop Client: " << attribs.at(2) << endl;
_sharedMap.setDesktopClientPid(std::stol(attribs.at(2)));
}
}
}

Expand Down Expand Up @@ -402,11 +451,7 @@ void SocketThread::Process()

case MSG_TIMER: {
// cout << "Timer expired on " << THREAD_NAME << endl;
const std::string msg = readSocket();
if (!msg.empty()) {
cout << "Message received: " << msg << endl;
handleReceivedMsg(msg);
}
processSocketInput();
break;
}

Expand Down
9 changes: 8 additions & 1 deletion src/openvfsfuse/socketthread.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,10 @@ class SocketThread
int initSocket(const std::string& socketPath);

bool socketSendMsg(std::shared_ptr<MsgData>);
std::string readSocket();

/// Drain everything readable from the socket into _rxBuffer and dispatch
/// every complete (newline terminated) message it contains.
void processSocketInput();
void handleReceivedMsg(const std::string &msg);

/// Entry point for the worker thread
Expand All @@ -114,6 +117,10 @@ class SocketThread

std::atomic<int> _socket;

/// Receive buffer holding the bytes read from the socket that do not form
/// a complete message yet. Only touched by the worker thread.
std::string _rxBuffer;

SharedMap &_sharedMap;
};

Expand Down
Loading