Algorithm-EventsPerSecond
view release on metacpan or search on metacpan
lib/Algorithm/EventsPerSecond/Sukkal.pm view on Meta::CPAN
Algorithm::EventsPerSecond::Sukkal - A unix-socket daemon serving per-key sliding-window event rates.
=head1 VERSION
Version 0.1.0
=cut
our $VERSION = '0.1.0';
# per-connection buffer ceilings: a single line may not span more than
# _RBUF_MAX, and a client that stops reading is dropped once _WBUF_MAX
# of replies have queued up
use constant {
_RBUF_MAX => 1024 * 1024,
_WBUF_MAX => 8 * 1024 * 1024,
_READ_CHUNK => 65536,
};
=head1 SYNOPSIS
use Algorithm::EventsPerSecond::Sukkal;
my $sukkal = Algorithm::EventsPerSecond::Sukkal->new(
socket => '/var/run/iqbi-damiq.sock',
window => 60,
);
$SIG{TERM} = $SIG{INT} = sub { $sukkal->stop };
$sukkal->run; # blocks until stop()
Then, from any client:
use IO::Socket::UNIX;
my $sock = IO::Socket::UNIX->new(
Type => SOCK_STREAM,
Peer => '/var/run/iqbi-damiq.sock',
);
print $sock "MARK requests\n"; # fire and forget
print $sock "MARK errors 3\n";
print $sock "RATE requests\n";
my $reply = <$sock>; # "OK 41.2\n"
print $sock "MARKRATE requests\n"; # mark and rate in one call
my $rate = <$sock>; # "OK 41.3\n"
=head1 DESCRIPTION
A sukkal is the vizier-messenger of a Mesopotamian court: petitioners
speak to it, and it relays word of them to the throne. This sukkal
listens on a unix stream socket, records events marked against
arbitrary client-chosen keys, and answers queries about their rates.
Each key gets its own L<Algorithm::EventsPerSecond> meter, so C<mark>
stays O(1) and memory per key is constant regardless of event volume.
The daemon is a single process driven by a non-blocking select loop;
no non-core modules are required. Marks arriving back-to-back on a
connection are coalesced per key and applied with a single C<mark($n)>
call, so the hot path is dominated by socket reads and line parsing,
not by the meters.
Keys that go idle longer than L</idle_timeout> are evicted by a
periodic sweep. Because the timeout is never shorter than the window,
an evicted key by definition has zero events inside the window, so
queries for it correctly read as zero; the only state lost is its
lifetime L</TOTAL>.
The bundled launcher script is L<iqbi-damiq>, "She said 'it is fine!'".
=head1 METHODS
=head2 new( socket => $path, %options )
Construct a daemon. Nothing is bound until L</run> is called.
=over 4
=item socket
Path of the unix socket to listen on. Required. A stale socket file
left by a dead daemon is removed automatically; a live listener on the
same path is an error.
=item window
Averaging window in seconds for every meter, as in
L<Algorithm::EventsPerSecond/new>. Defaults to 60. Each key's memory
scales linearly with the window; see L</MEMORY USAGE>.
=item max_keys
Maximum number of distinct keys tracked at once. Marks for new keys
beyond the limit are rejected with an error reply. 0 means unlimited.
Defaults to 100000. This is the daemon's memory ceiling: worst case
is C<max_keys> live meters, each of a size fixed by the window; see
L</MEMORY USAGE>.
=item max_key_length
Maximum key length in bytes. Keys may be any non-whitespace,
non-control bytes. Defaults to 255.
=item idle_timeout
Seconds a key may go unmarked before the sweep evicts it. Must be at
least C<window>. Defaults to twice the window.
=item sweep_interval
Seconds between eviction sweeps. Defaults to 30.
=item max_clients
Maximum simultaneous client connections; further connections are
closed immediately. 0 means unlimited, the default.
=item listen_backlog
The listen(2) backlog. Defaults to 128.
=item socket_mode
Octal permission string, e.g. C<'0770'>, applied to the socket file
after binding. By default the process umask decides.
=back
=cut
sub new {
my ( $class, %args ) = @_;
my $self = {
socket => $args{socket},
window => $args{window} // 60,
max_keys => $args{max_keys} // 100_000,
max_key_length => $args{max_key_length} // 255,
sweep_interval => $args{sweep_interval} // 30,
max_clients => $args{max_clients} // 0,
listen_backlog => $args{listen_backlog} // 128,
};
die "socket path required\n"
unless defined $self->{socket} && length $self->{socket};
for my $opt (qw(window max_key_length sweep_interval listen_backlog)) {
die "$opt must be a positive integer\n"
unless $self->{$opt} =~ /^\d+$/ && $self->{$opt} > 0;
}
for my $opt (qw(max_keys max_clients)) {
die "$opt must be a non-negative integer\n"
unless $self->{$opt} =~ /^\d+$/;
}
$self->{idle_timeout} = $args{idle_timeout} // $self->{window} * 2;
die "idle_timeout must be an integer >= window\n"
unless $self->{idle_timeout} =~ /^\d+$/
&& $self->{idle_timeout} >= $self->{window};
if ( defined $args{socket_mode} ) {
die "socket_mode must be an octal string, e.g. '0770'\n"
unless $args{socket_mode} =~ /^0?[0-7]{3}$/;
$self->{socket_mode} = oct $args{socket_mode};
}
$self->{meters} = {}; # key => { m => meter, seen => epoch }
$self->{conns} = {}; # fd => { fh, id, rbuf, wbuf, closing }
$self->{running} = 0;
$self->{started} = time();
$self->{self_meter} = Algorithm::EventsPerSecond->new( window => $self->{window} );
$self->{key_re} = qr/^[\x21-\x7E\x80-\xFF]{1,$self->{max_key_length}}$/;
return bless $self, $class;
} ## end sub new
=head2 run
Bind the socket and serve until L</stop> is called (typically from a
signal handler; signals interrupt the select and are honored
promptly). On return the socket file has been unlinked and all client
connections closed. Dies if the socket cannot be bound.
=cut
lib/Algorithm/EventsPerSecond/Sukkal.pm view on Meta::CPAN
Local => $path,
Listen => $self->{listen_backlog},
) or die "cannot listen on $path: $!\n";
$listener->blocking(0);
chmod $self->{socket_mode}, $path if defined $self->{socket_mode};
$self->{listener} = $listener;
$self->{listener_fd} = fileno $listener;
return;
} ## end sub _listen
sub _accept {
my ($self) = @_;
while ( my $fh = $self->{listener}->accept ) {
if ( $self->{max_clients}
&& keys %{ $self->{conns} } >= $self->{max_clients} )
{
close $fh;
next;
}
$fh->blocking(0);
my $id = fileno $fh;
$self->{conns}{$id} = {
fh => $fh,
id => $id,
rbuf => '',
wbuf => '',
closing => 0,
};
$self->{rsel}->add($fh);
} ## end while ( my $fh = $self->{listener}->accept )
return;
} ## end sub _accept
sub _read_client {
my ( $self, $fh ) = @_;
my $id = fileno $fh;
my $c = $self->{conns}{$id} or return;
my $eof;
# bounded so one firehose client cannot starve the rest of the loop
for ( 1 .. 16 ) {
my $n = sysread $fh, my $chunk, _READ_CHUNK;
if ( !defined $n ) {
last if $!{EAGAIN} || $!{EWOULDBLOCK} || $!{EINTR};
$eof = 1;
last;
}
if ( $n == 0 ) { $eof = 1; last }
$c->{rbuf} .= $chunk;
last if $n < _READ_CHUNK;
} ## end for ( 1 .. 16 )
if ( length $c->{rbuf} > _RBUF_MAX ) {
$self->_send( $c, "ERR line too long\n" );
return $self->_drop($c);
}
$self->_process($c);
$self->_drop($c) if $eof && $self->{conns}{$id};
return;
} ## end sub _read_client
sub _process {
my ( $self, $c ) = @_;
my $buf = $c->{rbuf};
my $pos = 0;
my %pending;
my $pending_new = 0;
while ( ( my $nl = index $buf, "\n", $pos ) >= 0 ) {
my $line = substr $buf, $pos, $nl - $pos;
$pos = $nl + 1;
$line =~ s/\r\z//;
my ( $cmd, $key, $extra ) = split ' ', $line, 3;
next unless defined $cmd;
$cmd = uc $cmd;
$extra =~ s/\s+\z// if defined $extra;
# hot path: marks are coalesced per key and applied in one
# mark($n) call, either just before the next query or at the
# end of the buffer
if ( $cmd eq 'MARK' ) {
my $count = 1;
if ( defined $extra ) {
if ( $extra =~ /^\d{1,15}$/ && $extra > 0 ) {
$count = $extra;
} else {
$self->_send( $c, "ERR bad count\n" );
next;
}
}
if ( !defined $key || $key !~ $self->{key_re} ) {
$self->_send( $c, "ERR bad key\n" );
next;
}
if ( !exists $self->{meters}{$key} && !exists $pending{$key} ) {
if ( $self->{max_keys}
&& keys( %{ $self->{meters} } ) + $pending_new >= $self->{max_keys} )
{
$self->_send( $c, "ERR key limit reached\n" );
next;
}
$pending_new++;
}
$pending{$key} += $count;
next;
} ## end if ( $cmd eq 'MARK' )
if (%pending) {
$self->_apply_marks( \%pending );
%pending = ();
$pending_new = 0;
}
$self->_command( $c, $cmd, $key, $extra );
last if $c->{closing};
} ## end while ( ( my $nl = index $buf, "\n", $pos ) >=...)
$self->_apply_marks( \%pending ) if %pending;
$c->{rbuf} = $pos ? substr( $buf, $pos ) : $buf;
return;
} ## end sub _process
sub _apply_marks {
my ( $self, $pending ) = @_;
my $meters = $self->{meters};
my $now = time();
my $n = 0;
for my $key ( keys %$pending ) {
my $e = $meters->{$key} ||= { m => Algorithm::EventsPerSecond->new( window => $self->{window} ) };
$e->{m}->mark( $pending->{$key} );
$e->{seen} = $now;
$n += $pending->{$key};
}
$self->{self_meter}->mark($n);
return;
} ## end sub _apply_marks
sub _command {
my ( $self, $c, $cmd, $key, $extra ) = @_;
my $meters = $self->{meters};
if ( $cmd eq 'RATE' || $cmd eq 'COUNT' || $cmd eq 'TOTAL' ) {
return $self->_send( $c, "ERR bad key\n" )
unless defined $key && $key =~ $self->{key_re};
my $e = $meters->{$key};
my $v
= !$e ? 0
: $cmd eq 'RATE' ? sprintf( '%.6g', $e->{m}->rate )
: $cmd eq 'COUNT' ? $e->{m}->count
: $e->{m}->total;
return $self->_send( $c, "OK $v\n" );
} ## end if ( $cmd eq 'RATE' || $cmd eq 'COUNT' || ...)
if ( $cmd eq 'MARKRATE' ) {
my $count = 1;
if ( defined $extra ) {
return $self->_send( $c, "ERR bad count\n" )
unless $extra =~ /^\d{1,15}$/ && $extra > 0;
$count = $extra;
}
return $self->_send( $c, "ERR bad key\n" )
unless defined $key && $key =~ $self->{key_re};
if ( !exists $meters->{$key}
&& $self->{max_keys}
&& keys(%$meters) >= $self->{max_keys} )
{
return $self->_send( $c, "ERR key limit reached\n" );
}
$self->_apply_marks( { $key => $count } );
return $self->_send( $c, sprintf( "OK %.6g\n", $meters->{$key}{m}->rate ) );
} ## end if ( $cmd eq 'MARKRATE' )
if ( $cmd eq 'STATS' ) {
if ( defined $key ) {
return $self->_send( $c, "ERR bad key\n" )
unless $key =~ $self->{key_re};
my $e = $meters->{$key};
my ( $rate, $count, $total )
= $e
? ( sprintf( '%.6g', $e->{m}->rate ), $e->{m}->count, $e->{m}->total )
( run in 1.806 second using v1.01-cache-2.11-cpan-600a1bdf6e4 )