#include "uds_server.h"
#include <cerrno>
#include <cstdio>
#include <cstring>
#include <fcntl.h>
#include <poll.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
namespace swarmd {
UdsServer::~UdsServer() { close_all(); }
bool UdsServer::listen(const char *socket_path) {
_socket_path = socket_path;
unlink(socket_path); // a stale socket file from a crashed run refuses bind
_listen_fd = socket(AF_UNIX, SOCK_STREAM, 0);
if (_listen_fd < 0) return false;
struct sockaddr_un addr;
memset(&addr, 0, sizeof(addr));
addr.sun_family = AF_UNIX;
strncpy(addr.sun_path, socket_path, sizeof(addr.sun_path) - 1);
if (bind(_listen_fd, (struct sockaddr *)&addr, sizeof(addr)) != 0 ||
::listen(_listen_fd, 8) != 0) {
::close(_listen_fd);
_listen_fd = -1;
return false;
}
fcntl(_listen_fd, F_SETFL, O_NONBLOCK);
return true;
}
void UdsServer::close_all() {
for (auto &kv : _clients) ::close(kv.second.fd);
_clients.clear();
if (_listen_fd >= 0) {
::close(_listen_fd);
_listen_fd = -1;
}
if (!_socket_path.empty()) {
unlink(_socket_path.c_str());
_socket_path.clear();
}
}
void UdsServer::_accept() {
for (;;) {
int fd = ::accept(_listen_fd, nullptr, nullptr);
if (fd < 0) return;
fcntl(fd, F_SETFL, O_NONBLOCK);
const uint64_t id = _next_id++;
UdsClient &c = _clients[id];
c.fd = fd;
c.id = id;
if (on_connect) on_connect(id);
}
}
void UdsServer::_read(UdsClient &c) {
uint8_t tmp[65536];
for (;;) {
ssize_t r = ::read(c.fd, tmp, sizeof(tmp));
if (r == 0) { _drop(c.id); return; }
if (r < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) break;
_drop(c.id);
return;
}
c.in.insert(c.in.end(), tmp, tmp + r);
}
// Drain whole frames only; a partial one waits for the next read.
for (;;) {
if (c.in.size() < 4) return;
const uint32_t n = (uint32_t)c.in[0] | ((uint32_t)c.in[1] << 8) |
((uint32_t)c.in[2] << 16) | ((uint32_t)c.in[3] << 24);
if (n > UDS_MAX_FRAME) {
// Refuse loudly rather than resynchronise: a length this large means
// the stream is already misaligned, and guessing where the next
// frame starts invents data.
fprintf(stderr, "[swarmd] client %llu sent length %u > max — dropping it\n",
(unsigned long long)c.id, n);
_drop(c.id);
return;
}
if (c.in.size() < 4 + (size_t)n) return;
_frames_in++;
if (on_frame) on_frame(c.id, c.in.data() + 4, n);
c.in.erase(c.in.begin(), c.in.begin() + 4 + n);
// on_frame may have dropped this client.
if (_clients.find(c.id) == _clients.end()) return;
}
}
void UdsServer::_flush(UdsClient &c) {
while (c.out_sent < c.out.size()) {
ssize_t w = ::write(c.fd, c.out.data() + c.out_sent, c.out.size() - c.out_sent);
if (w > 0) {
// ⛔ A PARTIAL WRITE IS REMEMBERED, NOT DISCARDED. Forgetting the
// remainder is what desynchronises a client's stream.
c.out_sent += (size_t)w;
continue;
}
if (w < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) break;
_drop(c.id);
return;
}
if (c.out_sent == c.out.size()) {
c.out.clear();
c.out_sent = 0;
} else if (c.out_sent > 0) {
c.out.erase(c.out.begin(), c.out.begin() + c.out_sent);
c.out_sent = 0;
}
if (!c.writable && c.out.size() < UDS_CLIENT_HIGH_WATER / 2) c.writable = true;
}
bool UdsServer::send(uint64_t client_id, const uint8_t *data, size_t len, bool *r_writable) {
auto it = _clients.find(client_id);
if (it == _clients.end()) {
if (r_writable) *r_writable = false;
return false;
}
UdsClient &c = it->second;
if (c.out.size() + 4 + len > UDS_CLIENT_HIGH_WATER) {
// ⛔ DROP THE CLIENT RATHER THAN CORRUPT IT. It reconnects and resyncs;
// a client fed a half-frame does not know it has been lied to.
fprintf(stderr, "[swarmd] client %llu is %zu bytes behind — dropping it\n",
(unsigned long long)c.id, c.out.size());
_drop(client_id);
_dropped++;
if (r_writable) *r_writable = false;
return false;
}
const uint32_t n = (uint32_t)len;
const uint8_t hdr[4] = { (uint8_t)(n & 0xff), (uint8_t)((n >> 8) & 0xff),
(uint8_t)((n >> 16) & 0xff), (uint8_t)((n >> 24) & 0xff) };
c.out.insert(c.out.end(), hdr, hdr + 4);
c.out.insert(c.out.end(), data, data + len);
_frames_out++;
_flush(c);
if (_clients.find(client_id) == _clients.end()) {
if (r_writable) *r_writable = false;
return false;
}
if (c.out.size() >= UDS_CLIENT_HIGH_WATER) c.writable = false;
if (r_writable) *r_writable = c.writable;
return true;
}
void UdsServer::broadcast(const uint8_t *data, size_t len, uint64_t except_id) {
std::vector<uint64_t> ids;
ids.reserve(_clients.size());
for (auto &kv : _clients) ids.push_back(kv.first);
for (uint64_t id : ids) {
if (id == except_id) continue;
send(id, data, len);
}
}
void UdsServer::_drop(uint64_t id) {
auto it = _clients.find(id);
if (it == _clients.end()) return;
::close(it->second.fd);
_clients.erase(it);
if (on_disconnect) on_disconnect(id);
}
void UdsServer::poll(int timeout_ms) {
std::vector<struct pollfd> fds;
std::vector<uint64_t> ids;
fds.push_back({ _listen_fd, POLLIN, 0 });
ids.push_back(0);
for (auto &kv : _clients) {
short ev = POLLIN;
if (kv.second.out_sent < kv.second.out.size()) ev |= POLLOUT;
fds.push_back({ kv.second.fd, ev, 0 });
ids.push_back(kv.first);
}
const int r = ::poll(fds.data(), fds.size(), timeout_ms);
if (r <= 0) return;
if (fds[0].revents & POLLIN) _accept();
for (size_t i = 1; i < fds.size(); i++) {
auto it = _clients.find(ids[i]);
if (it == _clients.end()) continue; // dropped earlier in this same pass
if (fds[i].revents & POLLOUT) _flush(it->second);
it = _clients.find(ids[i]);
if (it == _clients.end()) continue;
if (fds[i].revents & (POLLIN | POLLHUP)) _read(it->second);
}
}
} // namespace swarmd