itten/dmf-studio-rnd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fist API trys

Browse Source
This commit is contained in:
JohannesItten
2026-07-07 18:56:15 +03:00
parent 592686b6d7
commit 0b6e9ce46a
4 changed files with 373 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
studio-manager/CMakeLists.txt
+1 -1
View File
@@ -1,4 +1,4 @@
add_executable(dmf-studio-manager main.cpp)
target_compile_features(dmf-studio-manager PRIVATE cxx_std_20)
target_link_libraries(dmf-studio-manager PRIVATE dmf-shared)
target_link_libraries(dmf-studio-manager PRIVATE dmf-shared Crow::Crow)
install(TARGETS dmf-studio-manager RUNTIME DESTINATION bin)
studio-manager/StudioManager.hpp
+283
View File
@@ -0,0 +1,283 @@
#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
studio-manager/main.cpp
+73 -207
View File
@@ -1,190 +1,16 @@
// Studio Manager — launches and monitors node processes for a single-host pipeline.
// Graph is defined in build_graph(). Later: load from graph.json, WebSocket API.
// Node binaries are looked up next to this binary (same directory).
#include <cstdio>
#include <cstdlib>
#include <filesystem>
#include <fstream>
#include <map>
#include <sstream>
#include <mutex>
#include <string>
#include <utility>
#include <vector>
#include <fcntl.h>
#include <sys/wait.h>
#include <unistd.h>
#include <mxl/mxl.h>
#include <mxl/time.h>
#include "Signal.hpp"
#include "FlowGraph.hpp"
#include <thread>
#include <unordered_set>
#include <crow.h>
#include "StudioManager.hpp"
namespace fs = std::filesystem;
// --- UUID generation ---------------------------------------------------------
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)) != sizeof(b)) { perror("read"); exit(1); }
close(fd);
b[6] = (b[6] & 0x0Fu) | 0x40u; // version 4
b[8] = (b[8] & 0x3Fu) | 0x80u; // variant 1
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;
}
// --- Pipeline graph ----------------------------------------------------------
static dmf::FlowGraph load_graph(const std::string& path) {
std::ifstream f(path);
if (!f) {
fprintf(stderr, "[studio-manager] cannot open graph file: %s\n", path.c_str());
exit(1);
}
std::ostringstream ss;
ss << f.rdbuf();
auto j = nlohmann::json::parse(ss.str(), nullptr, /*allow_exceptions=*/false);
if (j.is_discarded()) {
fprintf(stderr, "[studio-manager] invalid JSON in %s\n", path.c_str());
exit(1);
}
dmf::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())
});
}
// Edges sharing the same from_node+from_port reuse the same flow UUID,
// so one writer can be read by multiple consumers (MXL supports N readers per flow).
std::map<std::pair<std::string,std::string>, std::string> flow_ids;
for (const auto& e : j.at("edges")) {
auto from_node = e.at("from").get<std::string>();
auto from_port = e.at("from_port").get<std::string>();
auto key = std::make_pair(from_node, from_port);
if (!flow_ids.count(key)) flow_ids[key] = gen_uuid();
g.edges.push_back({
flow_ids.at(key),
from_node,
from_port,
e.value("to", std::string{}),
e.value("to_port", std::string{}),
e.at("format")
});
}
fprintf(stderr, "[studio-manager] loaded graph from %s (%zu nodes, %zu edges)\n",
path.c_str(), g.nodes.size(), g.edges.size());
return g;
}
static dmf::FlowGraph build_graph() {
dmf::FlowGraph g;
const std::string tp_video_flow = gen_uuid();
const std::string tp_audio_flow = gen_uuid();
const std::string ndi_video_flow = gen_uuid();
const std::string ndi_audio_flow = gen_uuid();
g.nodes = {
{ "testpattern", "testpattern", {{"pattern", "bars"}} },
// { "ndiin", "ndiin", {} },
// { "fakesink", "fakesink", {} },
{ "ndiout", "ndiout", {} },
};
const nlohmann::json video_fmt = {
{"kind","video"}, {"width",1920}, {"height",1080}, {"fps_num",25}, {"fps_den",1}
};
const nlohmann::json audio_fmt = {
{"kind","audio"}, {"sample_rate",48000}, {"channels",2}, {"bit_depth",32}
};
g.edges = {
// { tp_video_flow, "testpattern", "video_flow_id", "fakesink", "video_flow_id", video_fmt },
// { tp_audio_flow, "testpattern", "audio_flow_id", "", "", audio_fmt },
// { ndi_video_flow, "ndiin", "video_flow_id", "ndiout", "video_flow_id", video_fmt },
// { ndi_audio_flow, "ndiin", "audio_flow_id", "", "", audio_fmt },
{ tp_video_flow, "testpattern", "video_flow_id", "ndiout", "video_flow_id", video_fmt },
{ tp_audio_flow, "testpattern", "audio_flow_id", "ndiout", "audio_flow_id", audio_fmt },
};
return g;
}
// --- Process management ------------------------------------------------------
struct NodeProcess {
std::string name;
pid_t pid{-1};
};
// Fork the node binary, passing domain and config via environment variables.
static NodeProcess launch_node(
const std::string& binary,
const std::string& domain,
const nlohmann::json& config)
{
NodeProcess proc;
proc.name = config.value("node_id", binary);
pid_t pid = fork();
if (pid < 0) { perror("fork"); return proc; }
if (pid == 0) {
// Child: set env and exec
setenv("MXL_DOMAIN", domain.c_str(), 1);
setenv("NODE_CONFIG", config.dump().c_str(), 1);
execl(binary.c_str(), binary.c_str(), nullptr);
// execl only returns on error
perror(("execl " + binary).c_str());
_exit(1);
}
proc.pid = pid;
fprintf(stderr, "[studio-manager] launched %s pid=%d\n", proc.name.c_str(), pid);
return proc;
}
// Poll children with WNOHANG; log and clear pid if one has exited.
static void check_children(std::vector<NodeProcess>& nodes) {
for (auto& node : nodes) {
if (node.pid <= 0) continue;
int wstatus = 0;
if (waitpid(node.pid, &wstatus, WNOHANG) == node.pid) {
fprintf(stderr, "[studio-manager] node %s (pid=%d) exited (status=%d)\n",
node.name.c_str(), node.pid, WEXITSTATUS(wstatus));
node.pid = -1;
}
}
}
// Send SIGTERM to all live children, then wait for them.
static void shutdown_children(std::vector<NodeProcess>& nodes) {
fprintf(stderr, "[studio-manager] sending SIGTERM to all nodes\n");
for (auto& node : nodes) {
if (node.pid > 0) {
kill(node.pid, SIGTERM);
}
}
for (auto& node : nodes) {
if (node.pid > 0) {
waitpid(node.pid, nullptr, 0);
fprintf(stderr, "[studio-manager] node %s stopped\n", node.name.c_str());
}
}
}
// -----------------------------------------------------------------------------
int main(int argc, char* argv[]) {
dmf::install_signal_handlers();
// Resolve node binary paths relative to this binary
const fs::path bin_dir = fs::path(argv[0]).parent_path();
#ifdef __APPLE__
@@ -193,47 +19,87 @@ int main(int argc, char* argv[]) {
const std::string domain = "/dev/shm/dmf-studio";
#endif
// Ensure the domain directory exists
std::error_code ec;
fs::create_directories(domain, ec);
if (ec) {
fprintf(stderr, "[studio-manager] cannot create domain dir %s: %s\n",
domain.c_str(), ec.message().c_str());
fprintf(stderr, "[studio-manager] cannot create domain dir: %s\n", ec.message().c_str());
return 1;
}
fprintf(stderr, "[studio-manager] domain: %s\n", domain.c_str());
// Clean up stale flow directories left by any previous crashed run
{
mxlInstance gc = mxlCreateInstance(domain.c_str(), "");
if (gc) {
mxlGarbageCollectFlows(gc);
mxlDestroyInstance(gc);
fprintf(stderr, "[studio-manager] garbage collected stale flows\n");
dmf::StudioManager manager(bin_dir, domain);
if (argc > 1) {
auto r = manager.load_graph_file(argv[1]);
if (r.value("type", "") == "error") {
fprintf(stderr, "[studio-manager] %s\n",
r.value("message", "load failed").c_str());
return 1;
}
}
// --- Build and launch the pipeline graph ---
const dmf::FlowGraph graph = (argc > 1) ? load_graph(argv[1]) : build_graph();
// ── WebSocket server ──────────────────────────────────────────────────────
for (const auto& e : graph.edges)
fprintf(stderr, "[studio-manager] flow %s → %s id=%s\n",
e.from_node.c_str(), e.to_node.c_str(), e.id.c_str());
crow::SimpleApp app;
app.signal_clear(); // let our signal handler manage SIGTERM/SIGINT
std::vector<NodeProcess> nodes;
for (const auto& node : graph.nodes) {
const std::string binary = (bin_dir / ("dmf-node-" + node.type)).string();
nodes.push_back(launch_node(binary, domain, graph.node_config(node.id)));
}
std::mutex ws_mutex;
std::unordered_set<crow::websocket::connection*> clients;
// --- Run until Ctrl+C or SIGTERM ---
fprintf(stderr, "[studio-manager] running — Ctrl+C to stop\n");
while (dmf::g_running.load(std::memory_order_relaxed)) {
check_children(nodes);
mxlSleepForNs(500'000'000); // check every 500 ms
}
// Push status to every connected client (called without manager mutex held).
manager.on_status_change([&](nlohmann::json status) {
std::lock_guard lk(ws_mutex);
const std::string msg = status.dump();
for (auto* c : clients) c->send_text(msg);
});
shutdown_children(nodes);
CROW_WEBSOCKET_ROUTE(app, "/ws")
.onopen([&](crow::websocket::connection& conn) {
{
std::lock_guard lk(ws_mutex);
clients.insert(&conn);
}
// get_status() acquires manager mutex — must NOT hold ws_mutex here
conn.send_text(manager.get_status().dump());
fprintf(stderr, "[studio-manager] client connected\n");
})
.onclose([&](crow::websocket::connection& conn, const std::string&) {
std::lock_guard lk(ws_mutex);
clients.erase(&conn);
fprintf(stderr, "[studio-manager] client disconnected\n");
})
.onmessage([&](crow::websocket::connection& conn, const std::string& data, bool) {
auto j = nlohmann::json::parse(data, nullptr, false);
if (j.is_discarded()) {
conn.send_text(R"({"type":"error","message":"invalid JSON"})");
return;
}
nlohmann::json response;
const std::string type = j.value("type", "");
try {
if (type == "load_graph") response = manager.load_graph(j.at("graph"));
else if (type == "stop_node") response = manager.stop_node(j.at("id"));
else if (type == "start_node") response = manager.start_node(j.at("id"));
else if (type == "get_status") response = manager.get_status();
else response = {{"type","error"},{"message","unknown command: " + type}};
} catch (const std::exception& e) {
response = {{"type","error"},{"message", e.what()}};
}
conn.send_text(response.dump());
});
// ── Monitor thread — detects crashes, stops Crow on shutdown ─────────────
std::thread monitor([&] {
manager.run_monitor();
app.stop();
});
fprintf(stderr, "[studio-manager] WebSocket API at ws://0.0.0.0:7070/ws\n");
app.port(7070).multithreaded().run();
manager.shutdown();
monitor.join();
fprintf(stderr, "[studio-manager] done\n");
return 0;
}