fix(console): bound the pending queue by lines and bytes
The console's pending, not-yet-drained queue was unbounded, so a hidden Console page that kept receiving verbose daemon-log chunks could grow without limit. - Pending lines are now bounded by both a line count and a byte budget. The oldest pending lines are dropped once either is exceeded, while the newest line is always kept so one huge line is never lost. - A formatted RPC result is ingested as one block under a single lock, so lines from other producers can never interleave inside it. The front-sequence accounting is also corrected: a pending line dropped by the cap never entered the visible buffer, so it no longer advances the front sequence. Only a front eviction or a clear does. Tests: the line and byte caps keep the newest lines, a single oversized line survives, a batch appends contiguously, and the front sequence advances only on a front eviction or clear. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -9,27 +9,60 @@
|
|||||||
namespace dragonx {
|
namespace dragonx {
|
||||||
namespace ui {
|
namespace ui {
|
||||||
|
|
||||||
|
void ConsoleModel::trimPendingLocked()
|
||||||
|
{
|
||||||
|
// Keep at least the newest line so one huge line is never dropped on arrival. Dropping the OLDEST pending
|
||||||
|
// lines past max_lines_ is lossless for the view (drain() would evict them immediately); the byte budget
|
||||||
|
// is the only lossy part and only bites while nobody drains (Console page closed).
|
||||||
|
while (pending_.size() > 1 && (pending_.size() > max_lines_ || pending_bytes_ > max_pending_bytes_)) {
|
||||||
|
pending_bytes_ -= pending_.front().text.size();
|
||||||
|
pending_.pop_front();
|
||||||
|
++dropped_;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
void ConsoleModel::ingest(const std::string& text, ConsoleChannel channel, int foldSpan)
|
void ConsoleModel::ingest(const std::string& text, ConsoleChannel channel, int foldSpan)
|
||||||
{
|
{
|
||||||
std::lock_guard<std::mutex> lock(ingest_mutex_);
|
ConsoleModelLine line; // build outside the lock
|
||||||
ConsoleModelLine line;
|
|
||||||
line.text = text;
|
line.text = text;
|
||||||
line.channel = channel;
|
line.channel = channel;
|
||||||
line.foldSpan = foldSpan;
|
line.foldSpan = foldSpan;
|
||||||
|
const std::size_t bytes = line.text.size();
|
||||||
|
std::lock_guard<std::mutex> lock(ingest_mutex_);
|
||||||
pending_.push_back(std::move(line));
|
pending_.push_back(std::move(line));
|
||||||
|
pending_bytes_ += bytes;
|
||||||
|
trimPendingLocked();
|
||||||
|
}
|
||||||
|
|
||||||
|
void ConsoleModel::ingestBatch(std::vector<ConsoleModelLine>&& lines)
|
||||||
|
{
|
||||||
|
if (lines.empty()) return;
|
||||||
|
std::size_t bytes = 0;
|
||||||
|
for (const auto& l : lines) bytes += l.text.size();
|
||||||
|
std::lock_guard<std::mutex> lock(ingest_mutex_); // ONE acquisition: the block stays contiguous
|
||||||
|
for (auto& l : lines) pending_.push_back(std::move(l));
|
||||||
|
pending_bytes_ += bytes;
|
||||||
|
trimPendingLocked();
|
||||||
}
|
}
|
||||||
|
|
||||||
ConsoleModel::DrainResult ConsoleModel::drain()
|
ConsoleModel::DrainResult ConsoleModel::drain()
|
||||||
{
|
{
|
||||||
std::vector<ConsoleModelLine> incoming;
|
std::deque<ConsoleModelLine> incoming;
|
||||||
|
std::size_t dropped = 0;
|
||||||
{
|
{
|
||||||
std::lock_guard<std::mutex> lock(ingest_mutex_);
|
std::lock_guard<std::mutex> lock(ingest_mutex_);
|
||||||
if (pending_.empty()) return {};
|
if (pending_.empty()) return {};
|
||||||
incoming.swap(pending_);
|
incoming.swap(pending_);
|
||||||
|
pending_bytes_ = 0;
|
||||||
|
dropped = dropped_;
|
||||||
|
dropped_ = 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
DrainResult result;
|
DrainResult result;
|
||||||
result.added = incoming.size();
|
result.added = incoming.size();
|
||||||
|
result.dropped = dropped;
|
||||||
|
// A dropped pending line never entered lines_, so it does not advance first_seq_ (the seq of
|
||||||
|
// lines_.front()); only a front eviction below or clear() does. Keeps the wrap cache aligned.
|
||||||
for (auto& line : incoming)
|
for (auto& line : incoming)
|
||||||
lines_.push_back(std::move(line));
|
lines_.push_back(std::move(line));
|
||||||
|
|
||||||
|
|||||||
@@ -45,16 +45,31 @@ class ConsoleModel {
|
|||||||
public:
|
public:
|
||||||
static constexpr std::size_t kDefaultMaxLines = 10000;
|
static constexpr std::size_t kDefaultMaxLines = 10000;
|
||||||
|
|
||||||
explicit ConsoleModel(std::size_t maxLines = kDefaultMaxLines) : max_lines_(maxLines) {}
|
// Pending (not yet drained) text is bounded too: lines beyond max_lines_ would be evicted by drain() anyway,
|
||||||
|
// and a byte budget stops a hidden Console from retaining verbose daemon-log chunks (<=4 KB each) forever.
|
||||||
|
static constexpr std::size_t kDefaultMaxPendingBytes = 4u * 1024u * 1024u;
|
||||||
|
|
||||||
|
explicit ConsoleModel(std::size_t maxLines = kDefaultMaxLines,
|
||||||
|
std::size_t maxPendingBytes = kDefaultMaxPendingBytes)
|
||||||
|
: max_lines_(maxLines), max_pending_bytes_(maxPendingBytes) {}
|
||||||
|
|
||||||
|
std::size_t maxLines() const { return max_lines_; }
|
||||||
|
|
||||||
// Append a line from ANY thread. Cheap: locks only the pending queue, does not touch
|
// Append a line from ANY thread. Cheap: locks only the pending queue, does not touch
|
||||||
// the visible deque. The line becomes visible after the next drain(). `foldSpan` marks a
|
// the visible deque. The line becomes visible after the next drain(). `foldSpan` marks a
|
||||||
// JSON block opener (offset to its matching closer); 0 for ordinary lines.
|
// JSON block opener (offset to its matching closer); 0 for ordinary lines.
|
||||||
void ingest(const std::string& text, ConsoleChannel channel, int foldSpan = 0);
|
void ingest(const std::string& text, ConsoleChannel channel, int foldSpan = 0);
|
||||||
|
|
||||||
|
// Append a contiguous block (e.g. one formatted RPC result) from ANY thread under ONE lock acquisition, so
|
||||||
|
// lines from other producers can never land inside it (foldSpan offsets are relative to the opener).
|
||||||
|
// Never calls the Logger (Logger::write invokes its callback while holding Logger::mutex_, i.e. the lock
|
||||||
|
// order is Logger::mutex_ -> ingest_mutex_; logging from in here would self-deadlock).
|
||||||
|
void ingestBatch(std::vector<ConsoleModelLine>&& lines);
|
||||||
|
|
||||||
struct DrainResult {
|
struct DrainResult {
|
||||||
std::size_t added = 0; // lines appended to the visible deque this drain
|
std::size_t added = 0; // lines appended to the visible deque this drain
|
||||||
std::size_t popped = 0; // lines evicted from the front by the line cap
|
std::size_t popped = 0; // lines evicted from the front by the line cap
|
||||||
|
std::size_t dropped = 0; // oldest pending lines discarded by the pending cap/byte budget since last drain
|
||||||
};
|
};
|
||||||
|
|
||||||
// Main thread only. Move pending lines into the visible deque, then enforce the cap.
|
// Main thread only. Move pending lines into the visible deque, then enforce the cap.
|
||||||
@@ -87,9 +102,14 @@ public:
|
|||||||
std::uint64_t firstSeq() const { return first_seq_; }
|
std::uint64_t firstSeq() const { return first_seq_; }
|
||||||
|
|
||||||
private:
|
private:
|
||||||
|
void trimPendingLocked(); // caller holds ingest_mutex_
|
||||||
|
|
||||||
const std::size_t max_lines_;
|
const std::size_t max_lines_;
|
||||||
|
const std::size_t max_pending_bytes_;
|
||||||
std::deque<ConsoleModelLine> lines_; // visible model — main thread only
|
std::deque<ConsoleModelLine> lines_; // visible model — main thread only
|
||||||
std::vector<ConsoleModelLine> pending_; // guarded by ingest_mutex_
|
std::deque<ConsoleModelLine> pending_; // guarded by ingest_mutex_ (deque: O(1) pop_front when trimming)
|
||||||
|
std::size_t pending_bytes_ = 0; // sum of pending_ text sizes, guarded by ingest_mutex_
|
||||||
|
std::size_t dropped_ = 0; // guarded by ingest_mutex_, reset by drain()
|
||||||
std::mutex ingest_mutex_;
|
std::mutex ingest_mutex_;
|
||||||
std::uint64_t revision_ = 0;
|
std::uint64_t revision_ = 0;
|
||||||
std::uint64_t first_seq_ = 0; // see firstSeq()
|
std::uint64_t first_seq_ = 0; // see firstSeq()
|
||||||
|
|||||||
@@ -2056,12 +2056,14 @@ void ConsoleTab::addFormattedResult(const std::string& result, bool is_error)
|
|||||||
|
|
||||||
// Precompute JSON fold spans over the whole block so each opener line carries the offset
|
// Precompute JSON fold spans over the whole block so each opener line carries the offset
|
||||||
// to its matching closer (the spans are relative, so they stay valid as the model's line
|
// to its matching closer (the spans are relative, so they stay valid as the model's line
|
||||||
// cap evicts from the front). This block is ingested atomically on the main thread.
|
// cap evicts from the front). The block is ingested under one lock (ingestBatch).
|
||||||
std::vector<std::string> texts;
|
std::vector<std::string> texts;
|
||||||
texts.reserve(resultLines.size());
|
texts.reserve(resultLines.size());
|
||||||
for (const auto& rl : resultLines) texts.push_back(rl.text);
|
for (const auto& rl : resultLines) texts.push_back(rl.text);
|
||||||
const std::vector<int> foldSpans = ComputeConsoleFoldSpans(texts);
|
const std::vector<int> foldSpans = ComputeConsoleFoldSpans(texts);
|
||||||
|
|
||||||
|
std::vector<ConsoleModelLine> block;
|
||||||
|
block.reserve(resultLines.size());
|
||||||
for (std::size_t i = 0; i < resultLines.size(); ++i) {
|
for (std::size_t i = 0; i < resultLines.size(); ++i) {
|
||||||
ConsoleChannel channel = ConsoleChannel::None;
|
ConsoleChannel channel = ConsoleChannel::None;
|
||||||
switch (resultLines[i].role) {
|
switch (resultLines[i].role) {
|
||||||
@@ -2072,8 +2074,13 @@ void ConsoleTab::addFormattedResult(const std::string& result, bool is_error)
|
|||||||
case ConsoleResultLineRole::JsonBrace: channel = ConsoleChannel::JsonBrace; break;
|
case ConsoleResultLineRole::JsonBrace: channel = ConsoleChannel::JsonBrace; break;
|
||||||
case ConsoleResultLineRole::Result: break;
|
case ConsoleResultLineRole::Result: break;
|
||||||
}
|
}
|
||||||
model_.ingest(resultLines[i].text, channel, foldSpans[i]);
|
ConsoleModelLine ml;
|
||||||
|
ml.text = resultLines[i].text;
|
||||||
|
ml.channel = channel;
|
||||||
|
ml.foldSpan = foldSpans[i];
|
||||||
|
block.push_back(std::move(ml));
|
||||||
}
|
}
|
||||||
|
model_.ingestBatch(std::move(block));
|
||||||
}
|
}
|
||||||
|
|
||||||
void ConsoleTab::renderStatusHeader(ConsoleCommandExecutor& exec)
|
void ConsoleTab::renderStatusHeader(ConsoleCommandExecutor& exec)
|
||||||
|
|||||||
+31
-7
@@ -7313,19 +7313,22 @@ void testConsoleWrapCache()
|
|||||||
EXPECT_EQ(c2.size(), static_cast<std::size_t>(0));
|
EXPECT_EQ(c2.size(), static_cast<std::size_t>(0));
|
||||||
}
|
}
|
||||||
|
|
||||||
// 7) ConsoleModel::firstSeq: +1 per evicted line, +size() on clear(), untouched by fold toggles.
|
// 7) ConsoleModel::firstSeq advances only when a line leaves the FRONT of lines_ (cap eviction or
|
||||||
|
// clear); a pending line dropped by the byte/line budget never entered lines_ and does not advance
|
||||||
|
// it. Untouched by fold toggles. (cap 3: 0..4 ingested -> pending drops "0","1" before they reach
|
||||||
|
// lines_, so lines_=[2,3,4] with zero front evictions -> firstSeq 0.)
|
||||||
{
|
{
|
||||||
ConsoleModel cm(3);
|
ConsoleModel cm(3);
|
||||||
for (int i = 0; i < 5; ++i) cm.ingest(std::to_string(i), ConsoleChannel::None);
|
for (int i = 0; i < 5; ++i) cm.ingest(std::to_string(i), ConsoleChannel::None);
|
||||||
cm.drain();
|
cm.drain();
|
||||||
EXPECT_EQ(cm.firstSeq(), static_cast<std::uint64_t>(2));
|
EXPECT_EQ(cm.firstSeq(), static_cast<std::uint64_t>(0)); // "2","3","4" land; nothing evicted from front
|
||||||
cm.ingest("{", ConsoleChannel::JsonBrace, 2);
|
cm.ingest("{", ConsoleChannel::JsonBrace, 2);
|
||||||
cm.drain();
|
cm.drain();
|
||||||
EXPECT_EQ(cm.firstSeq(), static_cast<std::uint64_t>(3));
|
EXPECT_EQ(cm.firstSeq(), static_cast<std::uint64_t>(1)); // "{" pushes over cap -> evict "2" from front
|
||||||
cm.toggleCollapsed(2);
|
cm.toggleCollapsed(2);
|
||||||
EXPECT_EQ(cm.firstSeq(), static_cast<std::uint64_t>(3));
|
EXPECT_EQ(cm.firstSeq(), static_cast<std::uint64_t>(1));
|
||||||
cm.clear();
|
cm.clear();
|
||||||
EXPECT_EQ(cm.firstSeq(), static_cast<std::uint64_t>(6));
|
EXPECT_EQ(cm.firstSeq(), static_cast<std::uint64_t>(4)); // + lines_.size() (3) on clear
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -7536,13 +7539,34 @@ void testConsoleModel()
|
|||||||
ConsoleModel m(3); // cap = 3
|
ConsoleModel m(3); // cap = 3
|
||||||
for (int i = 0; i < 5; ++i) m.ingest(std::to_string(i), ConsoleChannel::None);
|
for (int i = 0; i < 5; ++i) m.ingest(std::to_string(i), ConsoleChannel::None);
|
||||||
auto dr = m.drain();
|
auto dr = m.drain();
|
||||||
EXPECT_EQ(dr.added, static_cast<size_t>(5));
|
EXPECT_EQ(dr.added, static_cast<size_t>(3)); // pending is capped at ingest: "0","1" never reach the deque
|
||||||
EXPECT_EQ(dr.popped, static_cast<size_t>(2));
|
EXPECT_EQ(dr.popped, static_cast<size_t>(0));
|
||||||
|
EXPECT_EQ(dr.dropped, static_cast<size_t>(2));
|
||||||
EXPECT_EQ(m.size(), static_cast<size_t>(3));
|
EXPECT_EQ(m.size(), static_cast<size_t>(3));
|
||||||
EXPECT_EQ(m[0].text, std::string("2")); // "0" and "1" evicted
|
EXPECT_EQ(m[0].text, std::string("2")); // "0" and "1" evicted
|
||||||
EXPECT_EQ(m[2].text, std::string("4"));
|
EXPECT_EQ(m[2].text, std::string("4"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Pending byte budget keeps the newest lines; a single oversized line is never dropped.
|
||||||
|
{
|
||||||
|
ConsoleModel m(100, 10);
|
||||||
|
for (int i = 0; i < 5; ++i) m.ingest("abcd", ConsoleChannel::None);
|
||||||
|
auto dr = m.drain();
|
||||||
|
EXPECT_EQ(dr.added, static_cast<size_t>(2));
|
||||||
|
EXPECT_EQ(dr.dropped, static_cast<size_t>(3));
|
||||||
|
ConsoleModel big(100, 10);
|
||||||
|
big.ingest(std::string(50, 'x'), ConsoleChannel::None);
|
||||||
|
EXPECT_EQ(big.drain().added, static_cast<size_t>(1));
|
||||||
|
// ingestBatch appends a block contiguously.
|
||||||
|
std::vector<dragonx::ui::ConsoleModelLine> blk(3);
|
||||||
|
blk[0].text = "B0"; blk[1].text = "B1"; blk[2].text = "B2";
|
||||||
|
ConsoleModel bm;
|
||||||
|
bm.ingestBatch(std::move(blk));
|
||||||
|
bm.drain();
|
||||||
|
EXPECT_EQ(bm.size(), static_cast<size_t>(3));
|
||||||
|
EXPECT_EQ(bm[2].text, std::string("B2"));
|
||||||
|
}
|
||||||
|
|
||||||
// clear() drops the visible lines but not lines still queued for the next drain.
|
// clear() drops the visible lines but not lines still queued for the next drain.
|
||||||
{
|
{
|
||||||
ConsoleModel m;
|
ConsoleModel m;
|
||||||
|
|||||||
Reference in New Issue
Block a user