JohannesItten
284 lines
8.9 KiB
C++
284 lines
8.9 KiB
C++
#pragma once
|
|
|
|
#include <atomic>
|
|
#include <cstdio>
|
|
#include <filesystem>
|
|
#include <fstream>
|
|
#include <functional>
|
|
#include <map>
|
|
#include <mutex>
|
|
#include <sstream>
|
|
#include <stdexcept>
|
|
#include <string>
|
|
#include <utility>
|
|
#include <vector>
|
|
#include <fcntl.h>
|
|
#include <signal.h>
|
|
#include <sys/wait.h>
|
|
#include <unistd.h>
|
|
#include <mxl/mxl.h>
|
|
#include <mxl/time.h>
|
|
#include <nlohmann/json.hpp>
|
|
#include "FlowGraph.hpp"
|
|
#include "Signal.hpp"
|
|
|
|
namespace fs = std::filesystem;
|
|
|
|
namespace dmf {
|
|
|
|
enum class NodeState { Running, Stopped, Crashed };
|
|
|
|
struct NodeProcess {
|
|
std::string id;
|
|
std::string type;
|
|
pid_t pid{-1};
|
|
NodeState state{NodeState::Stopped};
|
|
};
|
|
|
|
class StudioManager {
|
|
public:
|
|
StudioManager(fs::path bin_dir, std::string domain)
|
|
: bin_dir_(std::move(bin_dir)), domain_(std::move(domain)) {}
|
|
|
|
~StudioManager() { shutdown(); }
|
|
|
|
// Replace the running pipeline with a new graph.
|
|
// Stops all current nodes, then starts the new ones.
|
|
nlohmann::json load_graph(const nlohmann::json& graph_json) {
|
|
nlohmann::json s;
|
|
std::string err;
|
|
{
|
|
std::lock_guard lk(mutex_);
|
|
stop_all_locked();
|
|
garbage_collect_locked();
|
|
try {
|
|
graph_ = parse_graph(graph_json);
|
|
start_all_locked();
|
|
} catch (const std::exception& e) {
|
|
err = e.what();
|
|
}
|
|
s = status_locked();
|
|
}
|
|
notify(s);
|
|
if (!err.empty())
|
|
return {{"type", "error"}, {"message", err}};
|
|
return s;
|
|
}
|
|
|
|
nlohmann::json load_graph_file(const std::string& path) {
|
|
std::ifstream f(path);
|
|
if (!f) return {{"type", "error"}, {"message", "cannot open: " + path}};
|
|
std::ostringstream ss;
|
|
ss << f.rdbuf();
|
|
auto j = nlohmann::json::parse(ss.str(), nullptr, false);
|
|
if (j.is_discarded())
|
|
return {{"type", "error"}, {"message", "invalid JSON in: " + path}};
|
|
return load_graph(j);
|
|
}
|
|
|
|
nlohmann::json stop_node(const std::string& id) {
|
|
nlohmann::json s;
|
|
{
|
|
std::lock_guard lk(mutex_);
|
|
for (auto& p : processes_) {
|
|
if (p.id == id && p.pid > 0) {
|
|
kill(p.pid, SIGTERM);
|
|
waitpid(p.pid, nullptr, 0);
|
|
p.pid = -1;
|
|
p.state = NodeState::Stopped;
|
|
fprintf(stderr, "[studio-manager] stopped node %s\n", id.c_str());
|
|
break;
|
|
}
|
|
}
|
|
s = status_locked();
|
|
}
|
|
notify(s);
|
|
return s;
|
|
}
|
|
|
|
nlohmann::json start_node(const std::string& id) {
|
|
nlohmann::json s;
|
|
{
|
|
std::lock_guard lk(mutex_);
|
|
for (auto& p : processes_) {
|
|
if (p.id == id && p.pid <= 0) {
|
|
launch_locked(p);
|
|
break;
|
|
}
|
|
}
|
|
s = status_locked();
|
|
}
|
|
notify(s);
|
|
return s;
|
|
}
|
|
|
|
nlohmann::json get_status() {
|
|
std::lock_guard lk(mutex_);
|
|
return status_locked();
|
|
}
|
|
|
|
// Register a callback invoked (without mutex held) whenever node state changes.
|
|
// Safe to call send_text() from inside the callback.
|
|
void on_status_change(std::function<void(nlohmann::json)> cb) {
|
|
std::lock_guard lk(mutex_);
|
|
status_cb_ = std::move(cb);
|
|
}
|
|
|
|
void shutdown() {
|
|
std::lock_guard lk(mutex_);
|
|
stop_all_locked();
|
|
}
|
|
|
|
// Blocks until g_running is false. Call from a dedicated thread.
|
|
void run_monitor() {
|
|
while (dmf::g_running.load(std::memory_order_relaxed)) {
|
|
mxlSleepForNs(500'000'000);
|
|
check_children();
|
|
}
|
|
}
|
|
|
|
private:
|
|
fs::path bin_dir_;
|
|
std::string domain_;
|
|
FlowGraph graph_;
|
|
std::vector<NodeProcess> processes_;
|
|
std::mutex mutex_;
|
|
std::function<void(nlohmann::json)> status_cb_;
|
|
|
|
// ── locked helpers — call only while holding mutex_ ──────────────────────
|
|
|
|
void stop_all_locked() {
|
|
for (auto& p : processes_)
|
|
if (p.pid > 0) kill(p.pid, SIGTERM);
|
|
for (auto& p : processes_) {
|
|
if (p.pid > 0) {
|
|
waitpid(p.pid, nullptr, 0);
|
|
fprintf(stderr, "[studio-manager] stopped node %s\n", p.id.c_str());
|
|
p.pid = -1;
|
|
p.state = NodeState::Stopped;
|
|
}
|
|
}
|
|
processes_.clear();
|
|
}
|
|
|
|
void garbage_collect_locked() {
|
|
mxlInstance gc = mxlCreateInstance(domain_.c_str(), "");
|
|
if (gc) {
|
|
mxlGarbageCollectFlows(gc);
|
|
mxlDestroyInstance(gc);
|
|
}
|
|
}
|
|
|
|
void start_all_locked() {
|
|
processes_.clear();
|
|
for (const auto& node : graph_.nodes) {
|
|
processes_.push_back({node.id, node.type, -1, NodeState::Stopped});
|
|
launch_locked(processes_.back());
|
|
}
|
|
}
|
|
|
|
void launch_locked(NodeProcess& p) {
|
|
const std::string binary = (bin_dir_ / ("dmf-node-" + p.type)).string();
|
|
const nlohmann::json cfg = graph_.node_config(p.id);
|
|
|
|
pid_t pid = fork();
|
|
if (pid < 0) { perror("fork"); return; }
|
|
if (pid == 0) {
|
|
setenv("MXL_DOMAIN", domain_.c_str(), 1);
|
|
setenv("NODE_CONFIG", cfg.dump().c_str(), 1);
|
|
execl(binary.c_str(), binary.c_str(), nullptr);
|
|
perror(("execl " + binary).c_str());
|
|
_exit(1);
|
|
}
|
|
p.pid = pid;
|
|
p.state = NodeState::Running;
|
|
fprintf(stderr, "[studio-manager] launched %s pid=%d\n", p.id.c_str(), pid);
|
|
}
|
|
|
|
nlohmann::json status_locked() const {
|
|
auto nodes = nlohmann::json::array();
|
|
for (const auto& p : processes_) {
|
|
const char* state =
|
|
p.state == NodeState::Running ? "running" :
|
|
p.state == NodeState::Crashed ? "crashed" : "stopped";
|
|
nodes.push_back({{"id", p.id}, {"type", p.type},
|
|
{"pid", p.pid}, {"state", state}});
|
|
}
|
|
return {{"type", "status"}, {"nodes", nodes}};
|
|
}
|
|
|
|
// ── unlocked helpers ─────────────────────────────────────────────────────
|
|
|
|
void notify(const nlohmann::json& s) {
|
|
// Called WITHOUT mutex_ held so the callback can safely call back into us.
|
|
if (status_cb_) status_cb_(s);
|
|
}
|
|
|
|
void check_children() {
|
|
bool changed = false;
|
|
nlohmann::json s;
|
|
{
|
|
std::lock_guard lk(mutex_);
|
|
for (auto& p : processes_) {
|
|
if (p.pid <= 0) continue;
|
|
int ws = 0;
|
|
if (waitpid(p.pid, &ws, WNOHANG) == p.pid) {
|
|
fprintf(stderr, "[studio-manager] node %s (pid=%d) exited (status=%d)\n",
|
|
p.id.c_str(), p.pid, WEXITSTATUS(ws));
|
|
p.pid = -1;
|
|
p.state = NodeState::Crashed;
|
|
changed = true;
|
|
}
|
|
}
|
|
if (changed) s = status_locked();
|
|
}
|
|
if (changed) notify(s);
|
|
}
|
|
|
|
// ── static helpers ───────────────────────────────────────────────────────
|
|
|
|
static FlowGraph parse_graph(const nlohmann::json& j) {
|
|
FlowGraph g;
|
|
for (const auto& n : j.at("nodes"))
|
|
g.nodes.push_back({
|
|
n.at("id").get<std::string>(),
|
|
n.at("type").get<std::string>(),
|
|
n.value("params", nlohmann::json::object())
|
|
});
|
|
|
|
std::map<std::pair<std::string,std::string>, std::string> flow_ids;
|
|
for (const auto& e : j.at("edges")) {
|
|
auto from = e.at("from").get<std::string>();
|
|
auto port = e.at("from_port").get<std::string>();
|
|
auto key = std::make_pair(from, port);
|
|
if (!flow_ids.count(key)) flow_ids[key] = gen_uuid();
|
|
g.edges.push_back({
|
|
flow_ids.at(key),
|
|
from, port,
|
|
e.value("to", std::string{}),
|
|
e.value("to_port", std::string{}),
|
|
e.at("format")
|
|
});
|
|
}
|
|
return g;
|
|
}
|
|
|
|
static std::string gen_uuid() {
|
|
uint8_t b[16];
|
|
int fd = open("/dev/urandom", O_RDONLY);
|
|
if (fd < 0) { perror("open /dev/urandom"); exit(1); }
|
|
if (read(fd, b, sizeof(b)) != (ssize_t)sizeof(b)) { perror("read"); exit(1); }
|
|
close(fd);
|
|
b[6] = (b[6] & 0x0Fu) | 0x40u;
|
|
b[8] = (b[8] & 0x3Fu) | 0x80u;
|
|
char s[37];
|
|
snprintf(s, sizeof(s),
|
|
"%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x",
|
|
b[0],b[1],b[2],b[3], b[4],b[5], b[6],b[7],
|
|
b[8],b[9], b[10],b[11],b[12],b[13],b[14],b[15]);
|
|
return s;
|
|
}
|
|
};
|
|
|
|
} // namespace dmf
|