queue_dryrun.cpp
Rama main · réplica de lectura del repositorio original.
// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0
#include "core_queue_sink.h"
#include "queue_flac_metadata.h"
#include "queue_lms_cli.h"
#include "queue_report.h"
#include "third_party/yeney-core/core/player.h"
#include <algorithm>
#include <atomic>
#include <cmath>
#include <csignal>
#include <ctime>
#include <cstdlib>
#include <filesystem>
#include <fcntl.h>
#include <fstream>
#include <iomanip>
#include <iostream>
#include <sstream>
#include <sys/file.h>
#include <thread>
#include <unistd.h>
namespace {
std::atomic_bool stopped{false}, resetMeasurement{false};
extern "C" void measureSignal(int) { resetMeasurement = true; }
extern "C" void stopSignal(int) { stopped = true; }
} // namespace
int main(int argc, char **argv) {
std::signal(SIGINT, stopSignal);
std::signal(SIGTERM, stopSignal);
std::signal(SIGHUP, stopSignal);
std::string server;
unsigned seconds = 900;
double speed = 1;
bool keep = false, transcoded = false, controlled = false;
for (int i = 1; i < argc; ++i) {
std::string a = argv[i];
if (a == "--help") {
std::cout << "yeney-queue-dryrun --server <ip> [--seconds 900] [--speed 1.0] "
"[--keep] [--transcoded]\nOwner-run only; select Queue Test "
"(YeneY) in "
"LMS. No Sonos or room-service access.\n";
return 0;
}
if (a == "--suite-controlled")
controlled = true;
else if (a == "--transcoded")
transcoded = true;
else if (a == "--keep")
keep = true;
else if (a == "--server" && i + 1 < argc)
server = argv[++i];
else if (a.rfind("--server=", 0) == 0)
server = a.substr(9);
else if (a == "--speed" && i + 1 < argc) {
try {
std::string value = argv[++i];
size_t used = 0;
speed = std::stod(value, &used);
if (used != value.size() || !std::isfinite(speed) || speed <= 0)
return 2;
} catch (...) {
return 2;
}
} else if (a == "--seconds" && i + 1 < argc) {
try {
std::string value = argv[++i];
if (value.empty() || value.find_first_not_of("0123456789") != std::string::npos)
return 2;
auto n = std::stoull(value);
if (!n || n > UINT32_MAX)
return 2;
seconds = unsigned(n);
} catch (...) {
return 2;
}
} else {
std::cerr << "Invalid option: " << a << '\n';
return 2;
}
}
if (server.empty() || server.find(':') != std::string::npos || !seconds) {
std::cerr << "--server <IPv4 address or hostname> is required; never "
"append :9000\n";
return 2;
}
const auto env = [](const char *key, const std::string &fallback) {
const char *value = std::getenv(key);
return value && *value ? std::string(value) : fallback;
};
const std::string tempRoot = env("TMPDIR", "/var/tmp");
const std::string cache = env("YENEY_TEST_CACHE", env("HOME", ".") + "/.cache/yeney-test");
const std::filesystem::path reports = std::filesystem::path(cache) / "dryrun";
std::filesystem::create_directories(reports);
int lock = open("/var/tmp/yeney-queue-dryrun.lock", O_RDWR | O_CREAT | O_CLOEXEC | O_NOFOLLOW, 0600);
if (lock < 0 || flock(lock, LOCK_EX | LOCK_NB)) {
std::cerr << "Another dry-run owns the fixed test player\n";
return 1;
}
auto time = std::time(nullptr);
std::tm local{};
localtime_r(&time, &local);
char stamp[32];
std::strftime(stamp, sizeof(stamp), "%Y%m%d-%H%M%S", &local);
const std::string runName = std::string("yeney-queue-dryrun-") + stamp + "-" + std::to_string(getpid());
const auto reportDir = reports / runName;
std::filesystem::create_directory(reportDir);
std::vector<std::filesystem::path> oldReports;
for (const auto &entry : std::filesystem::directory_iterator(reports))
if (entry.is_directory()) oldReports.push_back(entry.path());
std::sort(oldReports.begin(), oldReports.end(), [](const auto &a, const auto &b) {
return std::filesystem::last_write_time(a) > std::filesystem::last_write_time(b);
});
for (size_t i = 5; i < oldReports.size(); ++i) std::filesystem::remove_all(oldReports[i]);
const std::string base = (reportDir / "report").string();
const std::string storeRoot = keep ? (reportDir / "items").string() : tempRoot + "/" + runName + "-store";
std::ofstream log(base + ".log");
try {
queue_audio::Config storage;
storage.root = storeRoot;
queue_audio::CoreQueueSink sink("QueueTest", storage, 30, speed);
yeney::Config cfg;
cfg.server = server;
cfg.sourceIsTranscoded = [transcoded](uint64_t) { return transcoded; };
cfg.name = "Queue Test (YeneY)";
cfg.mac = {0x02, 0x59, 0x51, 0x54, 0, 1};
std::mutex logMutex;
cfg.log = [&](const std::string &s) {
std::lock_guard<std::mutex> lock(logMutex);
log << s << std::endl;
};
queue_audio::FlacMetadata flacMetadata(server, cfg.log);
cfg.observeCommand = [&](const yeney::Command &command) {
if (command.letter == 'f' || command.letter == 'q')
flacMetadata.reset();
};
cfg.flacSourceInfo = [&](uint64_t id, std::optional<yeney::FlacSourceInfo> info) {
auto rows = sink.snapshot();
const auto audible = sink.audibleOccurrence();
std::optional<uint64_t> audibleIndex;
for (const auto &r : rows)
if (r.source.occurrence == audible)
audibleIndex = r.expectedIndex;
for (const auto &r : rows)
if (r.source.occurrence == id)
return flacMetadata.resolve(r.expectedIndex, info, audibleIndex);
cfg.log("FLAC metadata sealed fallback: occurrence identity unavailable");
return info;
};
yeney::Player player(cfg, sink);
if (controlled)
std::signal(SIGUSR1, measureSignal);
std::exception_ptr playerError;
std::thread core([&] {
try {
player.run(stopped);
} catch (...) {
playerError = std::current_exception();
stopped = true;
}
});
struct JoinOnExit {
std::thread &thread;
~JoinOnExit() {
stopped = true;
if (thread.joinable())
thread.join();
}
} joinOnExit{core};
std::cout << "Play an album to Queue Test (YeneY), first crossfade off, "
"then on.\nReport: "
<< base << ".md" << std::endl;
using Clock = std::chrono::steady_clock;
auto start = Clock::now(), last = start;
double progress = 0;
uint64_t active = 0;
std::vector<uint64_t> completed;
struct CliResult {
uint64_t id;
std::vector<std::string> values;
double elapsed, stamp;
};
std::mutex cliMutex;
std::vector<CliResult> cliResults;
std::atomic<uint64_t> cliActive{0};
std::atomic<double> cliElapsed{0};
std::atomic_bool cliStop{false};
std::thread cli([&] {
auto next = Clock::now();
auto nextRaw = next;
while (!cliStop) {
if (Clock::now() >= next) {
next += std::chrono::seconds(1);
const auto id = cliActive.load();
const double elapsed = cliElapsed.load(), stamp = sink.secondsSinceStart();
std::string raw;
auto values = queue_audio::pollLms(server, 9090, &raw);
{
std::lock_guard<std::mutex> lock(logMutex);
if (Clock::now() >= nextRaw) {
log << "LMS CLI raw: " << raw << std::endl;
nextRaw = Clock::now() + std::chrono::minutes(1);
}
log << "LMS CLI poll stamp=" << sink.secondsSinceStart() << " audible=" << id;
if (values.size() == 3)
log << " index=" << values[0] << " time=" << values[1] << " title=" << values[2];
else
log << " unavailable";
log << std::endl;
}
if (id && id == cliActive.load()) {
std::lock_guard<std::mutex> lock(cliMutex);
cliResults.push_back({id, std::move(values), elapsed, stamp});
}
}
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
});
struct CliJoin {
std::thread &thread;
std::atomic_bool &stop;
~CliJoin() {
stop = true;
if (thread.joinable())
thread.join();
}
} cliJoin{cli, cliStop};
double measurementStart = 0;
auto measuredRows = [&] {
auto all = sink.snapshot();
if (measurementStart > 0) {
std::vector<uint64_t> owners;
for (const auto &r : all)
if (r.source.occurrence < UINT64_MAX / 2 && r.strm >= measurementStart)
owners.push_back(r.source.occurrence);
all.erase(std::remove_if(all.begin(), all.end(),
[&](const auto &r) {
if (r.source.occurrence < UINT64_MAX / 2)
return r.strm < measurementStart;
return std::find(owners.begin(), owners.end(),
UINT64_MAX - r.source.occurrence) ==
owners.end();
}),
all.end());
}
return all;
};
size_t reportedJoins = SIZE_MAX;
auto nextProgress = Clock::now();
while (!stopped && Clock::now() - start < std::chrono::seconds(seconds)) {
auto now = Clock::now();
if (resetMeasurement.exchange(false)) {
measurementStart = sink.secondsSinceStart();
reportedJoins = SIZE_MAX;
std::cout << "Measurement ready" << std::endl;
}
double dt = std::chrono::duration<double>(now - last).count();
last = now;
auto rows = sink.snapshot();
std::sort(rows.begin(), rows.end(),
[](const auto &a, const auto &b) { return a.first < b.first; });
for (const auto &r : rows) {
if (!r.item ||
std::find(completed.begin(), completed.end(), r.source.occurrence) != completed.end())
continue;
std::optional<uint64_t> length;
try {
length = r.item->contentLength();
} catch (...) {
// A playlist replacement cancels the old active asset.
// Let the new occurrence become the next audible epoch.
if (active == r.source.occurrence) {
active = 0;
progress = 0;
}
continue;
}
if (!length || !r.item->producedFrames())
break;
if (!active) {
active = r.source.occurrence;
progress = 0;
sink.audibleFrom(active, r.first);
dt = 0;
}
if (r.source.occurrence != active)
continue;
if (!sink.isPaused())
progress += dt * speed * r.source.rate;
const uint64_t total = (*length - 44) / 6, available = r.item->producedFrames();
if (progress > available)
progress = available; // real underrun, never advance through absent PCM
auto played = std::min<uint64_t>(uint64_t(progress), total);
sink.advanceAudible(r.first + played);
if (played == total) {
sink.audibleEnd(active);
completed.push_back(active);
active = 0;
progress = 0;
}
break;
}
const auto audibleId = active ? sink.audibleOccurrence() : 0;
for (const auto &r : rows)
if (r.source.occurrence == audibleId && r.source.rate) {
const auto frame = sink.audibleFrames();
cliElapsed = frame > r.first + r.prefixFrames
? double(frame - r.first - r.prefixFrames) / r.source.rate
: 0;
}
cliActive = audibleId;
std::vector<CliResult> samples;
{
std::lock_guard<std::mutex> lock(cliMutex);
samples.swap(cliResults);
}
for (const auto &sample : samples) {
auto it = std::find_if(rows.begin(), rows.end(),
[&](const auto &r) { return r.source.occurrence == sample.id; });
if (it == rows.end())
continue;
queue_audio::QueueRecord::LmsSample value;
value.stamp = sample.stamp;
value.audibleElapsed = sample.elapsed;
if (sample.values.size() == 3) {
value.index = sample.values[0];
value.time = sample.values[1];
value.title = sample.values[2];
try {
size_t used = 0;
double displayed = std::stod(value.time, &used);
if (used == value.time.size() && std::isfinite(displayed))
value.timeErrorWall = (displayed - sample.elapsed) / speed;
} catch (...) {
}
value.matches = value.index == std::to_string(it->expectedIndex);
}
sink.recordLms(sample.id, value);
std::lock_guard<std::mutex> lock(logMutex);
log << "LMS CLI stamp=" << value.stamp << " audible=" << sample.id
<< " expected_index=" << it->expectedIndex << " index=" << value.index
<< " time=" << value.time << " title=" << value.title << " lms_matches_audible="
<< (value.matches ? (*value.matches ? "true" : "false") : "unknown") << std::endl;
}
if (now >= nextProgress) {
std::ostringstream md, tsv;
auto verdict = queue_audio::writeReadAheadReport(measuredRows(), md, tsv, speed);
if (verdict.pairs != reportedJoins) {
reportedJoins = verdict.pairs;
std::cout << "Progress: completed_joins=" << reportedJoins << std::endl;
}
nextProgress = now + std::chrono::milliseconds(100);
}
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
cliStop = true;
cli.join();
try {
sink.waitIdle();
} catch (const std::exception &e) {
std::lock_guard<std::mutex> lock(logMutex);
log << "worker fault: " << e.what() << std::endl;
}
auto rows = measuredRows();
// Snapshot before Player shutdown flushes pending assets; timestamps remain
// explicit.
stopped = true;
core.join();
try {
sink.waitIdle();
} catch (const std::exception &e) {
log << "worker fault: " << e.what() << '\n';
}
std::ofstream tsv(base + ".tsv"), report(base + ".md");
auto verdict =
queue_audio::writeReadAheadReport(rows, report, tsv, speed, sink.wallSecondsSinceStart());
std::cout << verdict.line() << "; worst lead " << verdict.worst << " s\n";
if (playerError)
std::rethrow_exception(playerError);
} catch (const std::exception &e) {
std::cerr << e.what() << '\n';
if (!keep) {
std::error_code ignored;
std::filesystem::remove_all(storeRoot, ignored);
}
close(lock);
return 1;
}
if (!keep)
std::filesystem::remove_all(storeRoot);
close(lock);
return 0;
}