script/bench-outbox-dispatcher
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 JSON::MaybeXS qw(encode_json);
use List::Util qw(max);
use Time::HiRes qw(time);
use lib 'lib';
use lib 't/lib';
use GPForum::Command::Usage;
use GPForum::Service::Outbox::Dispatcher;
use GPForum::Test::OutboxBenchmarkFunctions;
use GPForum::Test::OutboxClock;
use GPForum::Test::OutboxCreateResultSet;
use GPForum::Test::OutboxResultSet;
use GPForum::Test::OutboxSchema;
use GPForum::Test::OutboxTransport;
use GPForum::X::Usage;
our $VERSION = '0.001';
const my @DEFAULT_WORKER_COUNTS => ( 1, 2, 4, 8 );
const my @DEFAULT_MESSAGE_COUNTS => (1_000);
const my $DEFAULT_BATCH_SIZE => 100;
const my $MILLISECONDS => 1_000;
const my $MINIMUM_ELAPSED_SEC => 0.000_001;
const my $ROUND_HALF_UP => 0.5;
const my $P95 => 0.95;
const my $READY_AT => '2026-05-23T12:00:00Z';
my $cli_options;
try {
$cli_options = _options(@ARGV);
}
catch ($error) {
croak $error if !GPForum::Command::Usage->is_usage($error);
exit GPForum::Command::Usage->error( undef, "$error" );
};
exit GPForum::Command::Usage->help( \*STDOUT, _usage() )
if $cli_options->{help};
my $benchmark = _benchmark_report($cli_options);
if ( defined $cli_options->{artifact} ) {
_write_artifact( $cli_options->{artifact}, $benchmark );
}
if ( $cli_options->{json} ) {
print encode_json($benchmark), "\n" or croak 'failed to write JSON report';
}
else {
_print_text_report($benchmark);
}
exit( $benchmark->{status} eq 'ok' ? 0 : 1 );
sub _benchmark_report {
my ($options) = @_;
my @results;
for my $message_count ( @{ $options->{message_counts} } ) {
for my $worker_count ( @{ $options->{worker_counts} } ) {
push @results,
_run_benchmark( $message_count, $worker_count,
$options->{batch_size} );
}
}
return {
status => ( grep { $_->{status} ne 'ok' } @results ) ? 'fail' : 'ok',
mode => 'fixture',
batch_size => $options->{batch_size},
message_counts => $options->{message_counts},
worker_counts => $options->{worker_counts},
results => \@results,
};
}
sub _run_benchmark {
my ( $message_count, $worker_count, $batch_size ) = @_;
my @rows = map {
_ready_row(
'bench-' . $message_count . q{-} . $worker_count . q{-} . $_ )
} ( 1 .. $message_count );
my $functions =
GPForum::Test::OutboxBenchmarkFunctions->new( queue => \@rows );
my $transport = GPForum::Test::OutboxTransport->new;
my @workers = _dispatchers( $functions, $transport, $worker_count );
my $started = time;
my $drain = _drain_pool( $batch_size, @workers );
my $elapsed = time - $started;
my $delivered = scalar @{ $transport->delivered };
my $duplicates = _duplicate_count( $transport->delivered );
my $lost = $message_count - $delivered + $duplicates;
# Each message is acknowledged on its own now, as soon as it is
# delivered: one acknowledgement a message, where there was one a batch.
my $unacknowledged = $message_count - $drain->{acknowledged};
my $failed = $duplicates || $lost || $unacknowledged;
my $safe_time =
$elapsed > $MINIMUM_ELAPSED_SEC
? $elapsed
: $MINIMUM_ELAPSED_SEC;
return {
status => $failed ? 'fail' : 'ok',
acknowledged => $drain->{acknowledged},
claimed_batches => $drain->{nonempty_claims},
workers => $worker_count,
messages => $message_count,
batch_size => $batch_size,
delivered => $delivered,
lost => $lost,
duplicates => $duplicates,
dispatch_calls => $drain->{dispatch_calls},
nonempty_claims => $drain->{nonempty_claims},
elapsed_ms => $elapsed * $MILLISECONDS,
msg_s => $delivered / $safe_time,
p95_claim_ms => _percentile( $drain->{claim_ms}, $P95 ),
max_claim_ms => max( 0, @{ $drain->{claim_ms} } ),
};
}
sub _drain_pool {
my ( $batch_size, @workers ) = @_;
my @claim_ms;
my $acknowledged = 0;
my $dispatch_calls = 0;
my $nonempty_claims = 0;
while (1) {
my $selected = 0;
for my $worker (@workers) {
my $started = time;
my $summary = $worker->dispatch_pending($batch_size);
my $elapsed = time - $started;
push @claim_ms, $elapsed * $MILLISECONDS;
$dispatch_calls++;
if ( $summary->{selected} ) {
$nonempty_claims++;
}
$selected += $summary->{selected};
$acknowledged += $summary->{acknowledged} // 0;
}
last if !$selected;
}
return {
acknowledged => $acknowledged,
claim_ms => \@claim_ms,
dispatch_calls => $dispatch_calls,
nonempty_claims => $nonempty_claims,
};
}
sub _dispatchers {
my ( $functions, $transport, $worker_count ) = @_;
return
map { _dispatcher( $functions, $transport, 'bench-worker-' . $_ ) }
( 1 .. $worker_count );
}
# Every worker claims from the one queue, through the functions the
# dispatcher calls on PostgreSQL.
sub _dispatcher {
my ( $functions, $transport, $worker_id ) = @_;
return GPForum::Service::Outbox::Dispatcher->new(
functions => $functions,
schema => GPForum::Test::OutboxSchema->new(
outbox_resultset => GPForum::Test::OutboxResultSet->new,
dead_letter_resultset => GPForum::Test::OutboxCreateResultSet->new,
),
transport => $transport,
clock => GPForum::Test::OutboxClock->new,
worker_id => $worker_id,
);
}
sub _ready_row {
my ($outbox_id) = @_;
return {
outbox_id => $outbox_id,
attempt_count => 0,
payload => {},
next_attempt_at => $READY_AT,
created_at => $READY_AT,
};
}
sub _duplicate_count {
my ($values) = @_;
my %seen;
my $duplicates = 0;
for my $value ( @{$values} ) {
if ( $seen{$value} ) {
$duplicates++;
}
$seen{$value} = 1;
}
return $duplicates;
}
sub _percentile {
my ( $values, $percentile ) = @_;
return 0 if !@{$values};
my @sorted = sort { $a <=> $b } @{$values};
my $index = int( ( @sorted - 1 ) * $percentile + $ROUND_HALF_UP );
return $sorted[$index];
}
sub _print_text_report {
my ($report) = @_;
for my $result ( @{ $report->{results} } ) {
printf
'outbox_dispatch_benchmark status=%s workers=%d messages=%d batch_size=%d delivered=%d lost=%d duplicates=%d acknowledged=%d claimed_batches=%d elapsed_ms=%.3f msg_s=%.3f p95_claim_ms=%.3f max_claim_ms=%.3f'
. "\n",
@{$result}{
qw(status workers messages batch_size delivered lost duplicates acknowledged claimed_batches elapsed_ms msg_s p95_claim_ms max_claim_ms)
}
or croak 'failed to write outbox benchmark result';
}
return;
}
sub _write_artifact {
my ( $artifact, $report ) = @_;
open my $handle, '>', $artifact
or croak "failed to write artifact $artifact";
print {$handle} encode_json($report), "\n"
or croak "failed to write artifact $artifact";
close $handle
or croak "failed to close artifact $artifact";
return;
}
sub _options {
my (@arguments) = @_;
my $usage = _usage();
my $counts = sub {
my ($value) = @_;
my @values = split /,/msx, $value // q{};
if ( !@values ) {
GPForum::X::Usage->throw( message => $usage );
}
return [
map {
GPForum::Command::Usage->option_number( $_,
'positive_integer', $usage )
} @values
];
};
return GPForum::Command::Usage->parse_options(
\@arguments,
{
artifact => undef,
batch_size => $DEFAULT_BATCH_SIZE,
help => 0,
json => 0,
message_counts => [@DEFAULT_MESSAGE_COUNTS],
worker_counts => [@DEFAULT_WORKER_COUNTS],
},
{
usage => $usage,
switches =>
{ '--help' => { help => 1 }, '--json' => { json => 1 } },
numbers =>
{ '--batch-size' => [ batch_size => 'positive_integer' ] },
values => {
'--artifact' => sub {
my ( $parsed, $value ) = @_;
$parsed->{artifact} =
GPForum::Command::Usage->option_value( $value, qr/./msx,
$usage );
},
'--messages' => sub {
my ( $parsed, $value ) = @_;
$parsed->{message_counts} = $counts->($value);
},
'--workers' => sub {
my ( $parsed, $value ) = @_;
$parsed->{worker_counts} = $counts->($value);
},
},
},
);
}
sub _usage {
return
'Usage: script/bench-outbox-dispatcher [--json] [--artifact FILE] [--messages 1000,10000,100000] [--workers 1,2,4,8] [--batch-size N]';
}