weaselab/weaseldb
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add arena to RenderPlan

Browse Source
This commit is contained in:
Andrew Noyes
2025-09-03 10:43:11 -04:00
parent a30020e960
commit 8763daca8e
1 changed files with 67 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files

322
src/metric.cpp
View File

@@ -19,7 +19,6 @@
#include <type_traits> #include <type_traits>
#include <unordered_map> #include <unordered_map>
#include <unordered_set> #include <unordered_set>
#include <variant>
#include <vector> #include <vector>
#include <immintrin.h> #include <immintrin.h>
@@ -631,11 +630,11 @@ struct Metric {
// lookups // lookups
struct CounterLabelData { struct CounterLabelData {
LabelsKey labels_key; LabelsKey labels_key;
std::vector<Counter::State *> thread_states; // Pre-resolved pointers ArenaVector<Counter::State *> thread_states; // Pre-resolved pointers
Counter::State *global_state; // Pre-resolved global state pointer Counter::State *global_state; // Pre-resolved global state pointer
CounterLabelData(const LabelsKey &key) CounterLabelData(const LabelsKey &key, ArenaAllocator &arena)
: labels_key(key), global_state(nullptr) {} : labels_key(key), thread_states(&arena), global_state(nullptr) {}
}; };
struct GaugeLabelData { struct GaugeLabelData {
@@ -648,19 +647,22 @@ struct Metric {
struct HistogramLabelData { struct HistogramLabelData {
LabelsKey labels_key; LabelsKey labels_key;
std::vector<Histogram::State *> thread_states; // Pre-resolved pointers ArenaVector<Histogram::State *> thread_states; // Pre-resolved pointers
Histogram::State *global_state; // Pre-resolved global state pointer Histogram::State *global_state; // Pre-resolved global state pointer
size_t bucket_count; // Cache bucket count from family size_t bucket_count; // Cache bucket count from family
HistogramLabelData(const LabelsKey &key) HistogramLabelData(const LabelsKey &key, ArenaAllocator &arena)
: labels_key(key), global_state(nullptr), bucket_count(0) {} : labels_key(key), thread_states(&arena), global_state(nullptr),
bucket_count(0) {}
}; };
// Pre-computed data for each family type, built once and reused // Pre-computed data for each family type, built once and reused
struct LabelSets { struct LabelSets {
std::vector<std::vector<CounterLabelData>> counter_data; ArenaVector<ArenaVector<CounterLabelData>> counter_data;
std::vector<std::vector<GaugeLabelData>> gauge_data; ArenaVector<ArenaVector<GaugeLabelData>> gauge_data;
std::vector<std::vector<HistogramLabelData>> histogram_data; ArenaVector<ArenaVector<HistogramLabelData>> histogram_data;
explicit LabelSets(ArenaAllocator &arena)
: counter_data(&arena), gauge_data(&arena), histogram_data(&arena) {}
}; };
// Instruction types for the execute phase // Instruction types for the execute phase
@@ -737,70 +739,20 @@ struct Metric {
static_assert(std::is_trivially_destructible_v<AggregateCounter>); static_assert(std::is_trivially_destructible_v<AggregateCounter>);
static_assert(std::is_trivially_destructible_v<AggregateGauge>); static_assert(std::is_trivially_destructible_v<AggregateGauge>);
static_assert(std::is_trivially_destructible_v<AggregateHistogram>); static_assert(std::is_trivially_destructible_v<AggregateHistogram>);
// Copy constructor and assignment
RenderInstruction(const RenderInstruction &other) : type(other.type) {
switch (type) {
case InstructionType::CALL_COUNTER_CALLBACK:
new (&counter_callback) CallCounterCallback(other.counter_callback);
break;
case InstructionType::CALL_GAUGE_CALLBACK:
new (&gauge_callback) CallGaugeCallback(other.gauge_callback);
break;
case InstructionType::AGGREGATE_COUNTER:
new (&aggregate_counter) AggregateCounter(other.aggregate_counter);
break;
case InstructionType::AGGREGATE_GAUGE:
new (&aggregate_gauge) AggregateGauge(other.aggregate_gauge);
break;
case InstructionType::AGGREGATE_HISTOGRAM:
new (&aggregate_histogram)
AggregateHistogram(other.aggregate_histogram);
break;
}
}
RenderInstruction &operator=(const RenderInstruction &other) {
if (this != &other) {
// All union members are trivially destructible, so no need to call
// destructor. Just reconstruct with new type and data.
type = other.type;
switch (type) {
case InstructionType::CALL_COUNTER_CALLBACK:
new (&counter_callback) CallCounterCallback(other.counter_callback);
break;
case InstructionType::CALL_GAUGE_CALLBACK:
new (&gauge_callback) CallGaugeCallback(other.gauge_callback);
break;
case InstructionType::AGGREGATE_COUNTER:
new (&aggregate_counter) AggregateCounter(other.aggregate_counter);
break;
case InstructionType::AGGREGATE_GAUGE:
new (&aggregate_gauge) AggregateGauge(other.aggregate_gauge);
break;
case InstructionType::AGGREGATE_HISTOGRAM:
new (&aggregate_histogram)
AggregateHistogram(other.aggregate_histogram);
break;
}
}
return *this;
}
}; };
// Three-phase rendering system // Three-phase rendering system
struct RenderPlan { struct RenderPlan {
ArenaVector<std::string_view> static_text; ArenaAllocator arena;
ArenaVector<RenderInstruction> instructions; ArenaVector<std::string_view> static_text{&arena};
ArenaVector<RenderInstruction> instructions{&arena};
RenderPlan(ArenaAllocator *arena)
: static_text(arena), instructions(arena) {}
}; };
// Phase 1: Compile phase - generate static text and instructions // Phase 1: Compile phase - generate static text and instructions
static RenderPlan compile_render_plan(ArenaAllocator &arena, static RenderPlan compile_render_plan() {
const LabelSets &label_sets) { RenderPlan plan;
RenderPlan plan(&arena);
Metric::LabelSets label_sets = Metric::build_label_sets(plan.arena);
// Helper function to append an additional label to existing Prometheus // Helper function to append an additional label to existing Prometheus
// format // format
@@ -818,7 +770,7 @@ struct Metric {
if (base_format.empty()) { if (base_format.empty()) {
// Create new format: {key="value"} // Create new format: {key="value"}
size_t required_size = 2 + key_value_size; // {} size_t required_size = 2 + key_value_size; // {}
char *buf = arena.allocate<char>(required_size); char *buf = plan.arena.allocate<char>(required_size);
char *p = buf; char *p = buf;
*p++ = '{'; *p++ = '{';
std::memcpy(p, key.data(), key.length()); std::memcpy(p, key.data(), key.length());
@@ -851,7 +803,7 @@ struct Metric {
// Append to existing format: {existing,key="value"} // Append to existing format: {existing,key="value"}
size_t required_size = base_format.length() + 1 + size_t required_size = base_format.length() + 1 +
key_value_size; // comma + key="value", replace } key_value_size; // comma + key="value", replace }
char *buf = arena.allocate<char>(required_size); char *buf = plan.arena.allocate<char>(required_size);
char *p = buf; char *p = buf;
// Copy everything except the closing } // Copy everything except the closing }
std::memcpy(p, base_format.data(), base_format.length() - 1); std::memcpy(p, base_format.data(), base_format.length() - 1);
@@ -894,7 +846,7 @@ struct Metric {
for (const auto &[name, family] : get_counter_families()) { for (const auto &[name, family] : get_counter_families()) {
// Add HELP line // Add HELP line
auto help_line = format( auto help_line = format(
arena, "%s# HELP %.*s %.*s\n# TYPE %.*s counter", plan.arena, "%s# HELP %.*s %.*s\n# TYPE %.*s counter",
is_first_static ? "" : "\n", static_cast<int>(name.length()), is_first_static ? "" : "\n", static_cast<int>(name.length()),
name.data(), static_cast<int>(family->help.length()), name.data(), static_cast<int>(family->help.length()),
family->help.data(), static_cast<int>(name.length()), name.data()); family->help.data(), static_cast<int>(name.length()), name.data());
@@ -903,13 +855,11 @@ struct Metric {
// Callback instructions and static text // Callback instructions and static text
for (const auto &[labels_key, callback] : family->callbacks) { for (const auto &[labels_key, callback] : family->callbacks) {
plan.instructions.push_back(CallCounterCallback{&callback}); plan.instructions.push_back(CallCounterCallback{&callback});
plan.static_text.push_back(arena_copy_string( plan.static_text.push_back(format(
format(arena, "%.*s\n%.*s%.*s ", static_cast<int>(help_line.size()), plan.arena, "%.*s\n%.*s%.*s ", static_cast<int>(help_line.size()),
help_line.data(), static_cast<int>(name.length()), help_line.data(), static_cast<int>(name.length()), name.data(),
name.data(), static_cast<int>(labels_key.prometheus_format.length()),
static_cast<int>(labels_key.prometheus_format.length()), labels_key.prometheus_format.data()));
labels_key.prometheus_format.data()),
arena));
help_line = ""; help_line = "";
} }
@@ -918,13 +868,11 @@ struct Metric {
for (const auto &data : family_data) { for (const auto &data : family_data) {
plan.instructions.push_back( plan.instructions.push_back(
AggregateCounter{data.thread_states, data.global_state}); AggregateCounter{data.thread_states, data.global_state});
plan.static_text.push_back(arena_copy_string( plan.static_text.push_back(format(
format(arena, "%.*s\n%.*s%.*s ", static_cast<int>(help_line.size()), plan.arena, "%.*s\n%.*s%.*s ", static_cast<int>(help_line.size()),
help_line.data(), static_cast<int>(name.length()), help_line.data(), static_cast<int>(name.length()), name.data(),
name.data(), static_cast<int>(data.labels_key.prometheus_format.length()),
static_cast<int>(data.labels_key.prometheus_format.length()), data.labels_key.prometheus_format.data()));
data.labels_key.prometheus_format.data()),
arena));
help_line = ""; help_line = "";
} }
} }
@@ -934,7 +882,7 @@ struct Metric {
for (const auto &[name, family] : get_gauge_families()) { for (const auto &[name, family] : get_gauge_families()) {
// Add HELP line // Add HELP line
auto help_line = format( auto help_line = format(
arena, "%s# HELP %.*s %.*s\n# TYPE %.*s gauge", plan.arena, "%s# HELP %.*s %.*s\n# TYPE %.*s gauge",
is_first_static ? "" : "\n", static_cast<int>(name.length()), is_first_static ? "" : "\n", static_cast<int>(name.length()),
name.data(), static_cast<int>(family->help.length()), name.data(), static_cast<int>(family->help.length()),
family->help.data(), static_cast<int>(name.length()), name.data()); family->help.data(), static_cast<int>(name.length()), name.data());
@@ -943,13 +891,11 @@ struct Metric {
// Callback instructions and static text // Callback instructions and static text
for (const auto &[labels_key, callback] : family->callbacks) { for (const auto &[labels_key, callback] : family->callbacks) {
plan.instructions.push_back(CallCounterCallback{&callback}); plan.instructions.push_back(CallCounterCallback{&callback});
plan.static_text.push_back(arena_copy_string( plan.static_text.push_back(format(
format(arena, "%.*s\n%.*s%.*s ", static_cast<int>(help_line.size()), plan.arena, "%.*s\n%.*s%.*s ", static_cast<int>(help_line.size()),
help_line.data(), static_cast<int>(name.length()), help_line.data(), static_cast<int>(name.length()), name.data(),
name.data(), static_cast<int>(labels_key.prometheus_format.length()),
static_cast<int>(labels_key.prometheus_format.length()), labels_key.prometheus_format.data()));
labels_key.prometheus_format.data()),
arena));
help_line = ""; help_line = "";
} }
@@ -957,13 +903,11 @@ struct Metric {
const auto &family_data = label_sets.gauge_data[gauge_family_idx++]; const auto &family_data = label_sets.gauge_data[gauge_family_idx++];
for (const auto &data : family_data) { for (const auto &data : family_data) {
plan.instructions.push_back(AggregateGauge{data.instance_state}); plan.instructions.push_back(AggregateGauge{data.instance_state});
plan.static_text.push_back(arena_copy_string( plan.static_text.push_back(format(
format(arena, "%.*s\n%.*s%.*s ", static_cast<int>(help_line.size()), plan.arena, "%.*s\n%.*s%.*s ", static_cast<int>(help_line.size()),
help_line.data(), static_cast<int>(name.length()), help_line.data(), static_cast<int>(name.length()), name.data(),
name.data(), static_cast<int>(data.labels_key.prometheus_format.length()),
static_cast<int>(data.labels_key.prometheus_format.length()), data.labels_key.prometheus_format.data()));
data.labels_key.prometheus_format.data()),
arena));
help_line = ""; help_line = "";
} }
} }
@@ -972,7 +916,7 @@ struct Metric {
size_t histogram_family_idx = 0; size_t histogram_family_idx = 0;
for (const auto &[name, family] : get_histogram_families()) { for (const auto &[name, family] : get_histogram_families()) {
auto help_line = format( auto help_line = format(
arena, "%s# HELP %.*s %.*s\n# TYPE %.*s histogram", plan.arena, "%s# HELP %.*s %.*s\n# TYPE %.*s histogram",
is_first_static ? "" : "\n", static_cast<int>(name.length()), is_first_static ? "" : "\n", static_cast<int>(name.length()),
name.data(), static_cast<int>(family->help.length()), name.data(), static_cast<int>(family->help.length()),
family->help.data(), static_cast<int>(name.length()), name.data()); family->help.data(), static_cast<int>(name.length()), name.data());
@@ -986,42 +930,38 @@ struct Metric {
// Static text for explicit buckets // Static text for explicit buckets
for (size_t i = 0; i < data.bucket_count; ++i) { for (size_t i = 0; i < data.bucket_count; ++i) {
auto bucket_value = static_format(arena, family->buckets[i]); auto bucket_value = static_format(plan.arena, family->buckets[i]);
auto labels = append_label_to_format( auto labels = append_label_to_format(
data.labels_key.prometheus_format, "le", bucket_value); data.labels_key.prometheus_format, "le", bucket_value);
plan.static_text.push_back(arena_copy_string( plan.static_text.push_back(
format(arena, "%.*s\n%.*s_bucket%.*s ", format(plan.arena, "%.*s\n%.*s_bucket%.*s ",
static_cast<int>(help_line.size()), help_line.data(), static_cast<int>(help_line.size()), help_line.data(),
static_cast<int>(name.length()), name.data(), static_cast<int>(name.length()), name.data(),
static_cast<int>(labels.length()), labels.data()), static_cast<int>(labels.length()), labels.data()));
arena));
help_line = ""; help_line = "";
} }
// Static text for +Inf bucket // Static text for +Inf bucket
auto inf_labels = append_label_to_format( auto inf_labels = append_label_to_format(
data.labels_key.prometheus_format, "le", "+Inf"); data.labels_key.prometheus_format, "le", "+Inf");
plan.static_text.push_back(arena_copy_string( plan.static_text.push_back(
format(arena, "\n%.*s_bucket%.*s ", static_cast<int>(name.length()), format(plan.arena, "\n%.*s_bucket%.*s ",
name.data(), static_cast<int>(inf_labels.length()), static_cast<int>(name.length()), name.data(),
inf_labels.data()), static_cast<int>(inf_labels.length()), inf_labels.data()));
arena));
// Static text for sum // Static text for sum
plan.static_text.push_back(arena_copy_string( plan.static_text.push_back(
format(arena, "\n%.*s_sum%.*s ", static_cast<int>(name.length()), format(plan.arena, "\n%.*s_sum%.*s ",
name.data(), static_cast<int>(name.length()), name.data(),
static_cast<int>(data.labels_key.prometheus_format.length()), static_cast<int>(data.labels_key.prometheus_format.length()),
data.labels_key.prometheus_format.data()), data.labels_key.prometheus_format.data()));
arena));
// Static text for count // Static text for count
plan.static_text.push_back(arena_copy_string( plan.static_text.push_back(
format(arena, "\n%.*s_count%.*s ", static_cast<int>(name.length()), format(plan.arena, "\n%.*s_count%.*s ",
name.data(), static_cast<int>(name.length()), name.data(),
static_cast<int>(data.labels_key.prometheus_format.length()), static_cast<int>(data.labels_key.prometheus_format.length()),
data.labels_key.prometheus_format.data()), data.labels_key.prometheus_format.data()));
arena));
} }
} }
@@ -1149,7 +1089,7 @@ struct Metric {
// Build label sets once for reuse in both phases // Build label sets once for reuse in both phases
static LabelSets build_label_sets(ArenaAllocator &arena) { static LabelSets build_label_sets(ArenaAllocator &arena) {
LabelSets label_sets; LabelSets label_sets{arena};
// Build counter data with pre-resolved pointers // Build counter data with pre-resolved pointers
for (const auto &[name, family] : Metric::get_counter_families()) { for (const auto &[name, family] : Metric::get_counter_families()) {
@@ -1170,9 +1110,9 @@ struct Metric {
} }
// Pre-resolve all pointers for each label set // Pre-resolve all pointers for each label set
std::vector<CounterLabelData> family_data; ArenaVector<CounterLabelData> family_data{&arena};
for (const auto &labels_key : all_labels) { for (const auto &labels_key : all_labels) {
CounterLabelData data(labels_key); CounterLabelData data{labels_key, arena};
// Pre-resolve thread-local state pointers // Pre-resolve thread-local state pointers
for (const auto &[thread_id, per_thread] : family->per_thread_state) { for (const auto &[thread_id, per_thread] : family->per_thread_state) {
@@ -1197,7 +1137,7 @@ struct Metric {
// Build gauge data with pre-resolved pointers // Build gauge data with pre-resolved pointers
for (const auto &[name, family] : Metric::get_gauge_families()) { for (const auto &[name, family] : Metric::get_gauge_families()) {
std::vector<GaugeLabelData> family_data; ArenaVector<GaugeLabelData> family_data{&arena};
// Gauges iterate directly over instances // Gauges iterate directly over instances
for (const auto &[labels_key, instance] : family->instances) { for (const auto &[labels_key, instance] : family->instances) {
@@ -1228,9 +1168,9 @@ struct Metric {
} }
// Pre-resolve all pointers for each label set // Pre-resolve all pointers for each label set
std::vector<HistogramLabelData> family_data; ArenaVector<HistogramLabelData> family_data{&arena};
for (const auto &labels_key : all_labels) { for (const auto &labels_key : all_labels) {
HistogramLabelData data(labels_key); HistogramLabelData data(labels_key, arena);
data.bucket_count = family->buckets.size(); // Cache bucket count data.bucket_count = family->buckets.size(); // Cache bucket count
// Pre-resolve thread-local state pointers // Pre-resolve thread-local state pointers
@@ -1594,141 +1534,13 @@ union MetricValue {
uint64_t as_uint64; uint64_t as_uint64;
}; };
// Legacy function kept for reference - will be replaced
static ArenaVector<MetricValue>
compute_metric_values_legacy(ArenaAllocator &arena,
const Metric::LabelSets &label_sets) {
ArenaVector<MetricValue> values(&arena);
// Compute counter values - ITERATION ORDER MUST MATCH FORMAT PHASE
size_t counter_family_idx = 0;
for (const auto &[name, family] : Metric::get_counter_families()) {
// Callback values
for (const auto &[labels_key, callback] : family->callbacks) {
auto value = callback();
values.push_back({.as_double = value});
}
// Use pre-computed data with resolved pointers - no hash lookups!
const auto &family_data = label_sets.counter_data[counter_family_idx++];
for (const auto &data : family_data) {
double total_value = 0.0;
// Sum thread-local values using pre-resolved pointers
for (auto *state_ptr : data.thread_states) {
// Atomic read to match atomic store in Counter::inc()
double value;
__atomic_load(&state_ptr->value, &value, __ATOMIC_RELAXED);
total_value += value;
}
// Add global accumulated value using pre-resolved pointer
if (data.global_state) {
total_value += data.global_state->value;
}
values.push_back({.as_double = total_value});
}
}
// Compute gauge values - ITERATION ORDER MUST MATCH FORMAT PHASE
size_t gauge_family_idx = 0;
for (const auto &[name, family] : Metric::get_gauge_families()) {
// Callback values
for (const auto &[labels_key, callback] : family->callbacks) {
auto value = callback();
values.push_back({.as_double = value});
}
// Use pre-computed data with resolved pointers - no hash lookups!
const auto &family_data = label_sets.gauge_data[gauge_family_idx++];
for (const auto &data : family_data) {
auto value = std::bit_cast<double>(
data.instance_state->value.load(std::memory_order_relaxed));
values.push_back({.as_double = value});
}
}
// Compute histogram values - ITERATION ORDER MUST MATCH FORMAT PHASE
size_t histogram_family_idx = 0;
for ([[maybe_unused]] const auto &[_name, _family] :
Metric::get_histogram_families()) {
// Use pre-computed data with resolved pointers - no hash lookups!
const auto &family_data = label_sets.histogram_data[histogram_family_idx++];
for (const auto &data : family_data) {
size_t bucket_count = data.bucket_count; // Use cached bucket count
uint64_t *total_counts_data = arena.allocate<uint64_t>(bucket_count);
std::memset(total_counts_data, 0, bucket_count * sizeof(uint64_t));
std::span<uint64_t> total_counts(total_counts_data, bucket_count);
double total_sum = 0.0;
uint64_t total_observations = 0;
// Sum thread-local values using pre-resolved pointers
for (auto *instance : data.thread_states) {
// Extract data under lock - minimize critical section
uint64_t *counts_snapshot = arena.allocate<uint64_t>(bucket_count);
double sum_snapshot;
uint64_t observations_snapshot;
{
std::lock_guard<std::mutex> lock(instance->mutex);
for (size_t i = 0; i < instance->counts.size(); ++i) {
counts_snapshot[i] = instance->counts[i];
}
sum_snapshot = instance->sum;
observations_snapshot = instance->observations;
}
// Add to totals
for (size_t i = 0; i < bucket_count; ++i) {
total_counts[i] += counts_snapshot[i];
}
total_sum += sum_snapshot;
total_observations += observations_snapshot;
}
// Add global accumulated value using pre-resolved pointer
if (data.global_state) {
auto *global_state = data.global_state;
for (size_t i = 0; i < global_state->counts.size(); ++i) {
total_counts[i] += global_state->counts[i];
}
total_sum += global_state->sum;
total_observations += global_state->observations;
}
// Store histogram values
// Store explicit bucket counts
for (size_t i = 0; i < total_counts.size(); ++i) {
values.push_back({.as_uint64 = total_counts[i]});
}
// Store +Inf bucket (total observations)
values.push_back({.as_uint64 = total_observations});
// Store sum
values.push_back({.as_double = total_sum});
// Store count
values.push_back({.as_uint64 = total_observations});
}
}
return values;
}
// Forward declaration
std::span<std::string_view> render_legacy(ArenaAllocator &arena);
// New three-phase render implementation // New three-phase render implementation
std::span<std::string_view> render(ArenaAllocator &arena) { std::span<std::string_view> render(ArenaAllocator &arena) {
// Hold lock throughout all phases to prevent registry changes // Hold lock throughout all phases to prevent registry changes
std::unique_lock<std::mutex> _{Metric::mutex}; std::unique_lock<std::mutex> _{Metric::mutex};
// Build label sets once for all phases
Metric::LabelSets label_sets = Metric::build_label_sets(arena);
// Phase 1: Compile - generate static text and instructions // Phase 1: Compile - generate static text and instructions
Metric::RenderPlan plan = Metric::compile_render_plan(arena, label_sets); Metric::RenderPlan plan = Metric::compile_render_plan();
// Phase 2: Execute - run instructions and generate dynamic text // Phase 2: Execute - run instructions and generate dynamic text
ArenaVector<std::string_view> dynamic_text = ArenaVector<std::string_view> dynamic_text =