script/performance-outbox.pl
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env perl
# SPDX-FileCopyrightText: 2026 GPForum Contributors
# SPDX-License-Identifier: BSD-3-Clause
package main;
use v5.40;
use Carp qw(croak);
use Const::Fast;
use English qw(-no_match_vars);
use File::Temp qw(tempdir);
use Getopt::Long qw(GetOptionsFromArray);
use JSON::MaybeXS qw(decode_json encode_json);
use POSIX qw(_exit);
use Time::HiRes qw(time);
use lib 'lib';
use lib 't/lib';
use GPForum::Benchmark::PostgresMetrics
qw(snapshot process_snapshot local_backend_pid);
use GPForum::Command::Migrate;
use GPForum::Service::Outbox::Dispatcher;
use GPForum::Test::OutboxTransport;
use GPForum::Test::PostgresHarness;
our $VERSION = '0.001';
const my $MAX_MESSAGES => 100_000;
const my $MAX_REPETITIONS => 5;
const my $BATCH_SIZE => 100;
const my $MILLISECONDS => 1_000;
const my $MIN_DURATION => 0.000_001;
const my $P50 => 0.50;
const my $P95 => 0.95;
const my $P99 => 0.99;
const my $ROUND_HALF => 0.5;
# The fixture benchmark remains script/bench-outbox-dispatcher. This adds
# the PostgreSQL costs it cannot measure, using the same real Dispatcher
# and recording transport, and real worker processes on independent DBI
# connections. CREATE DATABASE and migrations are outside the timed span.
exit main(@ARGV);
sub main (@arguments) {
my $options = _options(@arguments);
return _usage( $options->{usage} ) if defined $options->{usage};
my %options = %{$options};
return _benchmark( \%options );
}
sub _options (@arguments) {
my %options = ( messages => 1_000, workers => '1,2', repetitions => 3 );
GetOptionsFromArray( \@arguments, \%options,
'messages=i', 'workers=s', 'repetitions=i', 'json', 'help' )
or return { usage => 2 };
return { usage => 0 } if $options{help};
return { usage => 2 }
if @arguments
|| $options{messages} < 1
|| $options{messages} > $MAX_MESSAGES
|| $options{repetitions} < 1
|| $options{repetitions} > $MAX_REPETITIONS
|| $options{workers} !~ /\A[1248](?:,[1248])*\z/msx;
return \%options;
}
sub _benchmark ($option) {
my %options = %{$option};
my $admin_dsn = $ENV{GPFORUM_DATABASE_DSN} // croak
'GPFORUM_DATABASE_DSN is required for an owned benchmark database';
my @workers = split /,/msx, $options{workers};
my @children;
my $database;
my @results;
my $failure;
local $SIG{TERM} = sub { die "outbox benchmark interrupted\n"; };
local $SIG{INT} = sub { die "outbox benchmark interrupted\n"; };
try {
$database = GPForum::Test::PostgresHarness::create_database($admin_dsn);
local $ENV{GPFORUM_DATABASE_DSN} = $database->{dsn};
croak 'owned benchmark database migrations failed'
if GPForum::Test::PostgresHarness::quietly(
sub { return GPForum::Command::Migrate->new->run('--apply'); } );
_collect( $database, \%options, \@workers, \@children, \@results );
}
catch ($error) {
$failure = "$error";
};
for my $pid (@children) {
kill 'TERM', $pid;
waitpid $pid, 0;
}
if ($database) {
try {
GPForum::Test::PostgresHarness::drop_database($database);
}
catch ($error) {
$failure = ( $failure // q{} ) . " cleanup failed: $error";
};
}
my @failed = grep { $_->{status} ne 'ok' } @results;
my $status = $failure || @failed ? 'fail' : 'ok';
my $report = {
mode => 'postgresql',
status => $status,
results => \@results,
dataset =>
{ messages => $options{messages}, batch_size => $BATCH_SIZE },
limitations => [
'No-op transport excludes real handlers and external delivery.',
'CPU/RSS samples cover local Unix-socket backends and workers; remote backend CPU is unavailable. RSS includes shared mappings; ps CPU has centisecond precision.',
'I/O and WAL counters include cluster background activity; track_io_timing off makes I/O duration unavailable.',
'Batch percentiles include claim, renewal, dispatch and acknowledgement.',
'Timed wall span includes worker connection startup; migrations and seeding excluded.',
],
( $failure ? ( error => $failure ) : () ),
};
print encode_json($report), "\n" or croak 'cannot write outbox evidence';
return $status eq 'ok' ? 0 : 1;
}
sub _collect ( $database, $options, $workers, $children, $results ) {
for my $repetition ( 1 .. $options->{repetitions} ) {
# Alternating order reduces monotonic host/cache bias.
my @order = $repetition % 2 ? @{$workers} : reverse @{$workers};
for my $count (@order) {
# Keep each finished measurement in the caller's buffer before
# starting the next one: a caught signal or database error must
# not discard evidence from completed repetitions.
push @{$results},
_run( $database, $options->{messages},
$count, $repetition, $children );
}
}
return;
}
sub _run ( $database, $messages, $workers, $repetition, $children ) {
my $dbh = $database->{dbh};
$dbh->do('TRUNCATE outbox_messages');
$dbh->do(
q{INSERT INTO outbox_messages (outbox_id, event_id, queue, job_type,}
. q{ idempotency_key, payload, available_at, created_at, next_attempt_at)}
. q{ SELECT md5('message-' || n)::uuid, md5('event-' || n)::uuid,}
. q{ 'domain', 'benchmark.noop', 'performance-' || n, '{}'::jsonb,}
. q{ timestamptz '2026-01-01', timestamptz '2026-01-01',}
. q{ timestamptz '2026-01-01' FROM generate_series(1, ?::integer) n},
undef, $messages
);
$dbh->do('ANALYZE outbox_messages');
my $directory = tempdir( CLEANUP => 1 );
pipe my $barrier_read, my $barrier_write or croak "pipe: $OS_ERROR";
@{$children} = ();
for my $worker ( 1 .. $workers ) {
my $pid = _spawn( $database, $worker, $directory,
$barrier_read, $barrier_write );
push @{$children}, $pid;
}
close $barrier_read or croak "close barrier: $OS_ERROR";
my $before = snapshot($dbh);
my $started = time;
print {$barrier_write} 'x' x $workers or croak "release barrier: $OS_ERROR";
close $barrier_write or croak "close barrier: $OS_ERROR";
my $failed = 0;
for my $pid ( @{$children} ) {
waitpid $pid, 0;
$failed ||= $CHILD_ERROR != 0;
}
@{$children} = ();
my $elapsed = time - $started;
my $summary = _summary( $dbh, $messages, $workers, $repetition,
$directory, $elapsed, $failed );
$summary->{before} = $before;
$summary->{after} = snapshot($dbh);
$summary->{wal_bytes_inserted} = $dbh->selectrow_array(
'SELECT pg_wal_lsn_diff(?::pg_lsn, ?::pg_lsn)',
undef,
$summary->{after}{lsn},
$summary->{before}{lsn}
);
return $summary;
}
sub _spawn ( $database, $worker, $directory, $barrier_read, $barrier_write ) {
my $pid = fork;
croak "fork: $OS_ERROR" if !defined $pid;
return $pid if $pid;
# Only the parent owns database cleanup. A group timeout must not unwind
# an inherited parent stack in children and race DROP DATABASE.
local $SIG{TERM} = 'DEFAULT';
local $SIG{INT} = 'DEFAULT';
close $barrier_write or _exit(1);
# Never destruct an inherited parent DBI socket after a fork.
$database->{dbh}{InactiveDestroy} = 1;
$database->{admin_dbh}{InactiveDestroy} = 1;
my $token;
# Unbuffered: read() can prefetch several tokens into one child's
# PerlIO buffer, leaving other workers at EOF despite a complete release.
sysread $barrier_read, $token, 1 or _exit(1);
close $barrier_read or _exit(1);
my $report;
try {
$report = _worker($worker);
}
catch ($error) {
$report = { status => 'fail', error => "$error" };
};
open my $output, '>', "$directory/$worker.json" or _exit(1);
print {$output} encode_json($report) or _exit(1);
close $output or _exit(1);
_exit( $report->{status} eq 'ok' ? 0 : 1 );
return undef;
}
sub _summary ( $dbh, $messages, $workers, $repetition,
$directory, $elapsed, $failed )
{
my @delivered;
my @resources;
my @batch_ms;
my $acknowledged = 0;
my @errors;
for my $worker ( 1 .. $workers ) {
my $report = _read_report("$directory/$worker.json");
push @delivered, @{ $report->{delivered} // [] };
push @resources, $report->{resources};
push @batch_ms, @{ $report->{batch_ms} // [] };
$acknowledged += $report->{acknowledged} // 0;
if ( $report->{error} ) {
push @errors, $report->{error};
}
}
my %seen;
my $duplicates = grep { $seen{$_}++ } @delivered;
my $lost = $messages - scalar keys %seen;
my ($done) = $dbh->selectrow_array(
q{SELECT count(*) FROM outbox_messages WHERE status = 'done'});
my $incomplete = $acknowledged != $messages || $done != $messages;
return {
status => $failed
|| @errors
|| $duplicates
|| $lost
|| $incomplete ? 'fail' : 'ok',
workers => $workers,
worker_resources => \@resources,
repetition => $repetition,
messages => $messages,
duration_seconds => $elapsed,
messages_per_second => @delivered / ( $elapsed || $MIN_DURATION ),
delivered => scalar @delivered,
acknowledged => $acknowledged,
done => $done,
lost => $lost,
duplicates => $duplicates,
p50_batch_ms => _percentile( \@batch_ms, $P50 ),
p95_batch_ms => _percentile( \@batch_ms, $P95 ),
p99_batch_ms => _percentile( \@batch_ms, $P99 ),
errors => \@errors,
};
}
sub _read_report ($path) {
return { error => 'worker did not produce evidence' } if !-f $path;
open my $input, '<', $path or croak "read worker report: $OS_ERROR";
my $report =
decode_json( do { local $INPUT_RECORD_SEPARATOR = undef; <$input> } );
close $input or croak "close report: $OS_ERROR";
return $report;
}
sub _worker ($number) {
my $schema = GPForum::Test::PostgresHarness::connect_schema();
my $transport = GPForum::Test::OutboxTransport->new;
my $dispatcher = GPForum::Service::Outbox::Dispatcher->new(
schema => $schema,
transport => $transport,
worker_id => "performance-worker-$number",
);
my @batch_ms;
my $dbh = $schema->storage->dbh;
my $backend = local_backend_pid($dbh);
my $pg_before = process_snapshot($backend);
my @cpu_before = times;
my $acknowledged = 0;
while (1) {
my $started = time;
my $batch = $dispatcher->dispatch_pending($BATCH_SIZE);
last if !$batch->{selected};
push @batch_ms, ( time - $started ) * $MILLISECONDS;
$acknowledged += $batch->{acknowledged};
}
my @cpu_after = times;
my $pg_after = process_snapshot($backend);
my $resources = {
application_cpu_seconds => $cpu_after[0] +
$cpu_after[1] -
$cpu_before[0] -
$cpu_before[1],
application_process => process_snapshot($PROCESS_ID),
postgres_before => $pg_before,
postgres_after => $pg_after,
postgres_cpu_seconds => $pg_before && $pg_after
? $pg_after->{cpu_seconds} - $pg_before->{cpu_seconds}
: undef,
};
$schema->storage->disconnect;
return {
resources => $resources,
status => 'ok',
delivered => $transport->delivered,
batch_ms => \@batch_ms,
acknowledged => $acknowledged
};
}
sub _percentile ( $values, $quantile ) {
return undef if !@{$values};
my @sorted = sort { $a <=> $b } @{$values};
return $sorted[ int( ( @sorted - 1 ) * $quantile + $ROUND_HALF ) ];
}
sub _usage ($code) {
print 'Usage: script/performance-outbox.pl [--json] [--messages N]'
. " [--workers 1,2,4] [--repetitions 1..5]\n"
or croak 'cannot write usage';
return $code;
}