#include "session_daemon.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 session {
const char *PROTOCOL = "blender-bridge-session/v1";
// The UDS framing. 4-byte little-endian length, then that many bytes. It is NOT
// the session wire format — it is the boundary marker a byte stream does not
// have. blender-bridge/session/wire.py encodes what goes inside.
//
// ⚠ A stream socket coalesces writes, so two frames can arrive in one read and
// one frame across two. Anything that assumes read() == one message works right
// up until it does not, under load, once. Hence the per-client buffer.
static const uint32_t MAX_FRAME = 8 * 1024 * 1024;
SessionDaemon::SessionDaemon() {}
SessionDaemon::~SessionDaemon() { stop(); }
bool SessionDaemon::start(const uint8_t *topic, const char *bootstrap,
const char *socket_path) {
_topic.assign(topic, topic + 32);
_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) {
close(_listen_fd);
_listen_fd = -1;
return false;
}
if (listen(_listen_fd, 8) != 0) {
close(_listen_fd);
_listen_fd = -1;
return false;
}
fcntl(_listen_fd, F_SETFL, O_NONBLOCK);
_t.set_mux_callbacks(
[this](int32_t id, Protomux *mux) { _on_mux_ready(id, mux); },
[this](int32_t id) { _on_mux_gone(id); });
Ed25519KeyPair kp = NoiseIK::generate_keypair();
const char *bs[1] = { bootstrap };
_t.start_dht(kp, bs, 1);
// ⛔ DO NOT swarm_join() HERE. start_dht() is ASYNCHRONOUS: it starts the UV
// thread and returns before the loop's async handle exists. Joining straight
// away segfaults inside uv_async_send() — not an error, a crash, with no
// output at all because it happens before the first printf. The join is
// deferred to poll(), which also keeps start() non-blocking so the socket is
// bindable immediately and a client can connect before the swarm is up.
//
// test_session_channel.cpp got this right by waiting on is_dht_active()
// before joining; the first version of this daemon did not carry that over.
_started = true;
return true;
}
void SessionDaemon::stop() {
if (!_started) return;
_started = false;
for (auto &kv : _locals) close(kv.first);
_locals.clear();
if (_listen_fd >= 0) {
close(_listen_fd);
_listen_fd = -1;
}
if (!_socket_path.empty()) unlink(_socket_path.c_str());
}
void SessionDaemon::_on_mux_ready(int32_t id, Protomux *mux) {
if (!mux) return;
std::lock_guard<std::mutex> lk(_m);
if (_channels.count(id)) return;
ProtomuxChannel *ch = mux->create_channel(PROTOCOL, _topic);
if (ch == nullptr) return; // already open on this mux
// One slot: the slot IS the message type, which is why what crosses here is a
// BODY and the type only exists on the local hop, where there are no slots.
ch->add_message([this](const uint8_t *data, size_t len, ProtomuxChannel *) {
// Arrives on the UDX thread. Writing to a local socket from here is safe
// (the fds are only closed under _m) and keeps the hop short.
_broadcast_to_local(data, len, -1);
});
_channels[id] = ch;
ch->open();
}
void SessionDaemon::_on_mux_gone(int32_t id) {
std::lock_guard<std::mutex> lk(_m);
_channels.erase(id);
}
void SessionDaemon::_broadcast_to_peers(const uint8_t *data, size_t len) {
std::lock_guard<std::mutex> lk(_m);
for (auto &kv : _channels) kv.second->send(0, data, len);
_to_peers++;
}
void SessionDaemon::_broadcast_to_local(const uint8_t *data, size_t len, int except_fd) {
uint8_t hdr[4];
uint32_t n = (uint32_t)len;
hdr[0] = n & 0xff;
hdr[1] = (n >> 8) & 0xff;
hdr[2] = (n >> 16) & 0xff;
hdr[3] = (n >> 24) & 0xff;
std::lock_guard<std::mutex> lk(_m);
for (auto &kv : _locals) {
if (kv.first == except_fd) continue;
if (write(kv.first, hdr, 4) != 4) continue;
if (write(kv.first, data, len) != (ssize_t)len) continue;
}
_to_local++;
}
void SessionDaemon::_accept_local() {
for (;;) {
int fd = accept(_listen_fd, nullptr, nullptr);
if (fd < 0) return;
fcntl(fd, F_SETFL, O_NONBLOCK);
std::lock_guard<std::mutex> lk(_m);
_locals[fd] = Local();
}
}
void SessionDaemon::_read_local(int fd) {
uint8_t tmp[65536];
for (;;) {
ssize_t r = read(fd, tmp, sizeof(tmp));
if (r == 0) { _drop_local(fd); return; }
if (r < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) break;
_drop_local(fd);
return;
}
std::vector<uint8_t> *buf;
{
std::lock_guard<std::mutex> lk(_m);
auto it = _locals.find(fd);
if (it == _locals.end()) return;
buf = &it->second.buf;
buf->insert(buf->end(), tmp, tmp + r);
}
}
// Drain whole frames only. A partial one stays buffered for the next read.
for (;;) {
std::vector<uint8_t> frame;
{
std::lock_guard<std::mutex> lk(_m);
auto it = _locals.find(fd);
if (it == _locals.end()) return;
std::vector<uint8_t> &b = it->second.buf;
if (b.size() < 4) return;
uint32_t n = (uint32_t)b[0] | ((uint32_t)b[1] << 8) |
((uint32_t)b[2] << 16) | ((uint32_t)b[3] << 24);
if (n > MAX_FRAME) { // a length that large is a framing bug, not a frame
fprintf(stderr, "[session] client sent length %u > max, dropping it\n", n);
b.clear();
return;
}
if (b.size() < 4 + (size_t)n) return;
frame.assign(b.begin() + 4, b.begin() + 4 + n);
b.erase(b.begin(), b.begin() + 4 + n);
}
_broadcast_to_peers(frame.data(), frame.size());
// A second local client on the same daemon should see it too, or two
// Blenders on one machine would be invisible to each other.
_broadcast_to_local(frame.data(), frame.size(), fd);
}
}
void SessionDaemon::_drop_local(int fd) {
std::lock_guard<std::mutex> lk(_m);
auto it = _locals.find(fd);
if (it == _locals.end()) return;
close(fd);
_locals.erase(it);
}
void SessionDaemon::poll(int timeout_ms) {
// The deferred join (see start()). is_dht_active() is the signal that the UV
// loop and its async handle exist.
if (!_joined && _t.is_dht_active()) {
_t.swarm_join(_topic.data(), true, true);
_joined = true;
}
std::vector<struct pollfd> fds;
fds.push_back({ _listen_fd, POLLIN, 0 });
{
std::lock_guard<std::mutex> lk(_m);
for (auto &kv : _locals) fds.push_back({ kv.first, POLLIN, 0 });
}
int r = ::poll(fds.data(), fds.size(), timeout_ms);
if (r <= 0) return;
if (fds[0].revents & POLLIN) _accept_local();
for (size_t i = 1; i < fds.size(); i++) {
if (fds[i].revents & (POLLIN | POLLHUP)) _read_local(fds[i].fd);
}
}
int SessionDaemon::peer_count() {
std::lock_guard<std::mutex> lk(_m);
return (int)_channels.size();
}
int SessionDaemon::local_client_count() {
std::lock_guard<std::mutex> lk(_m);
return (int)_locals.size();
}
} // namespace session