Algorithm-EventsPerSecond
view release on metacpan or search on metacpan
When a "work that uses the Library" uses material from a header file
that is part of the Library, the object code for the work may be a
derivative work of the Library even though the source code is not.
Whether this is true is especially significant if the work can be
linked without the Library, or if the work is itself a library. The
threshold for this to be true is not precisely defined by law.
If such an object file uses only numerical parameters, data
structure layouts and accessors, and small macros and small inline
functions (ten lines or less in length), then the use of the object
file is unrestricted, regardless of whether it is legally a derivative
work. (Executables containing this object code plus portions of the
Library will still fall under Section 6.)
Otherwise, if the work is a derivative of the Library, you may
distribute the object code for the work under the terms of Section 6.
Any executables containing that work also fall under Section 6,
whether or not they are linked directly with the Library itself.
6. As an exception to the Sections above, you may also combine or
Makefile.PL view on Meta::CPAN
use 5.006;
use strict;
use warnings;
use ExtUtils::MakeMaker;
use Config;
# IF_OPT sets the -O value used to compile the XS backend (default -O3).
# IF_ARCH optionally sets the target arch, e.g. IF_ARCH=native for
# -march=native. Both accept a bare value or a full compiler flag.
my $optimize = defined $ENV{IF_OPT} && length $ENV{IF_OPT} ? $ENV{IF_OPT} : '3';
$optimize = "-O$optimize" unless $optimize =~ /^-/;
my $arch_flags = '';
if ( defined $ENV{IF_ARCH} && length $ENV{IF_ARCH} ) {
$arch_flags = $ENV{IF_ARCH};
$arch_flags = "-march=$arch_flags" unless $arch_flags =~ /^-/;
}
# The XS backend is optional: without a working compiler (or with
# PUREPERL_ONLY=1) the module installs as pure Perl and
# Algorithm::EventsPerSecond falls back automatically.
my $pureperl_only = grep { /^PUREPERL_ONLY=(\S+)$/ && $1 } @ARGV;
my $can_xs = !$pureperl_only
lib/Algorithm/EventsPerSecond.pm view on Meta::CPAN
Algorithm::EventsPerSecond keeps per-second counts in a fixed-size ring buffer and
reports the average event rate over the most recent N seconds (the
"window"). Memory use is constant regardless of event volume, and both
C<mark> and C<rate> are O(1) averaged out over time.
=head1 METHODS
=head2 new( window => $seconds )
Construct a meter. C<window> is the length of the averaging window in
seconds and defaults to 60.
=cut
sub new {
my ( $class, %args ) = @_;
my $window = $args{window} // 60;
die "window must be a positive integer" unless $window =~ /^\d+$/ && $window > 0;
lib/Algorithm/EventsPerSecond.pm view on Meta::CPAN
=head2 total
Lifetime count of all events ever recorded, regardless of window.
=cut
sub total { $_[0]->{total} }
=head2 window
The configured window length in seconds.
=cut
sub window { $_[0]->{window} }
=head2 reset
Clear all counts and restart the clock. Returns the meter object.
=cut
lib/Algorithm/EventsPerSecond/Sukkal.pm view on Meta::CPAN
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.
lib/Algorithm/EventsPerSecond/Sukkal.pm view on Meta::CPAN
=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"
lib/Algorithm/EventsPerSecond/Sukkal.pm view on Meta::CPAN
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.
lib/Algorithm/EventsPerSecond/Sukkal.pm view on Meta::CPAN
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 {
lib/Algorithm/EventsPerSecond/Sukkal.pm view on Meta::CPAN
return $self->_send( $c, "ERR unknown command\n" );
} ## end sub _command
sub _send {
my ( $self, $c, $data ) = @_;
if ( $c->{wbuf} eq '' ) {
my $n = syswrite $c->{fh}, $data;
if ( defined $n ) {
if ( $n == length $data ) {
$self->_drop($c) if $c->{closing};
return;
}
$data = substr $data, $n;
} elsif ( !( $!{EAGAIN} || $!{EWOULDBLOCK} || $!{EINTR} ) ) {
return $self->_drop($c);
}
} ## end if ( $c->{wbuf} eq '' )
$c->{wbuf} .= $data;
return $self->_drop($c) if length $c->{wbuf} > _WBUF_MAX;
$self->{wsel}->add( $c->{fh} );
return;
} ## end sub _send
sub _flush {
my ( $self, $c ) = @_;
my $n = syswrite $c->{fh}, $c->{wbuf};
if ( !defined $n ) {
return if $!{EAGAIN} || $!{EWOULDBLOCK} || $!{EINTR};
lib/Algorithm/EventsPerSecond/Sukkal.pm view on Meta::CPAN
}
delete @{$self}{qw(rsel wsel listener_fd)};
return;
} ## end sub _shutdown
=head1 PROTOCOL
The protocol is line-based over a unix stream socket. Lines end in
C<\n> (a trailing C<\r> is tolerated) and hold whitespace-separated
tokens; commands are case-insensitive. Keys are any non-whitespace,
non-control bytes up to L</max_key_length> long. Replies are a single
C<OK ...> or C<ERR ...> line, except L</KEYS> and L</DUMP>, which are
multi-line. Commands may be pipelined freely; replies come back in
order.
=head2 MARK <key> [<count>]
Record one event, or C<count> events, against C<key>, creating the key
if it is new. Nothing is replied on success
so writers never have to read; malformed input or hitting
L</max_keys> replies C<ERR ...>.
src_bin/iqbi-damiq view on Meta::CPAN
use 5.006;
use strict;
use warnings;
use Getopt::Long qw(GetOptions);
use Pod::Usage qw(pod2usage);
use Algorithm::EventsPerSecond;
use Algorithm::EventsPerSecond::Sukkal;
my (
$socket, $window, $max_keys, $max_key_length, $idle_timeout, $sweep_interval,
$max_clients, $mode, $daemonize, $pidfile, $require_xs,
);
GetOptions(
'socket|s=s' => \$socket,
'require-xs|x' => \$require_xs,
'window|w=i' => \$window,
'max-keys=i' => \$max_keys,
'max-key-length=i' => \$max_key_length,
'idle-timeout=i' => \$idle_timeout,
'sweep-interval=i' => \$sweep_interval,
'max-clients=i' => \$max_clients,
'mode|m=s' => \$mode,
'daemonize|d' => \$daemonize,
'pidfile|p=s' => \$pidfile,
'help|h' => sub { pod2usage( -exitval => 0, -verbose => 1 ) },
'version|v' => sub {
print "iqbi-damiq (Algorithm::EventsPerSecond::Sukkal) " . "$Algorithm::EventsPerSecond::Sukkal::VERSION\n";
exit 0;
src_bin/iqbi-damiq view on Meta::CPAN
pod2usage( -exitval => 2, -message => "iqbi-damiq: --socket is required\n" )
unless defined $socket;
die "iqbi-damiq: --require-xs given but the " . Algorithm::EventsPerSecond->backend . " backend is loaded, not XS\n"
if $require_xs && Algorithm::EventsPerSecond->backend ne 'XS';
my $sukkal = Algorithm::EventsPerSecond::Sukkal->new(
socket => $socket,
( defined $window ? ( window => $window ) : () ),
( defined $max_keys ? ( max_keys => $max_keys ) : () ),
( defined $max_key_length ? ( max_key_length => $max_key_length ) : () ),
( defined $idle_timeout ? ( idle_timeout => $idle_timeout ) : () ),
( defined $sweep_interval ? ( sweep_interval => $sweep_interval ) : () ),
( defined $max_clients ? ( max_clients => $max_clients ) : () ),
( defined $mode ? ( socket_mode => $mode ) : () ),
);
if ($daemonize) {
require POSIX;
defined( my $pid = fork ) or die "iqbi-damiq: cannot fork: $!\n";
src_bin/iqbi-damiq view on Meta::CPAN
=item -w, --window SECONDS
Averaging window used for every meter. Default 60.
=item --max-keys N
Maximum distinct keys tracked at once; 0 for unlimited. Default
100000.
=item --max-key-length N
Maximum key length in bytes. Default 255.
=item --idle-timeout SECONDS
Evict keys unmarked for this long; must be at least the window.
Default is twice the window.
=item --sweep-interval SECONDS
Seconds between eviction sweeps. Default 30.
t/sukkal-protocol.t view on Meta::CPAN
if $^O eq 'MSWin32';
}
use lib 't/lib';
use Time::HiRes qw(sleep);
use Sukkal_TestUtil qw(spawn_daemon connect_daemon read_line req req_multi stop_daemon);
$SIG{PIPE} = 'IGNORE';
alarm 120; # watchdog: a hung daemon must fail the test, not the harness
my $d = spawn_daemon( window => 5, max_key_length => 10 );
my $sock = connect_daemon($d);
#
# a command split across two writes is reassembled from the read buffer
#
print $sock 'MARK fr';
sleep 0.25; # let the daemon read the fragment on its own
print $sock "ag 3\nCOUNT frag\n";
is( read_line($sock), 'OK 3', 'command fragmented across writes is reassembled' );
t/sukkal-protocol.t view on Meta::CPAN
#
# count boundaries: 15 digits is the documented ceiling
#
print $sock "MARK big 100000000000000\n";
is( req( $sock, 'TOTAL big' ), 'OK 100000000000000', '15-digit count accepted' );
like( req( $sock, 'MARK big 1000000000000000' ), qr/^ERR bad count/, '16-digit count rejected' );
like( req( $sock, 'MARK big 0' ), qr/^ERR bad count/, 'MARK count of 0 rejected' );
like( req( $sock, 'MARKRATE big 0' ), qr/^ERR bad count/, 'MARKRATE count of 0 rejected' );
#
# key length boundary at max_key_length (10 here)
#
print $sock 'MARK ', 'k' x 10, "\n";
is( req( $sock, 'COUNT ' . 'k' x 10 ), 'OK 1', 'key exactly at max_key_length accepted' );
like( req( $sock, 'MARK ' . 'k' x 11 ), qr/^ERR bad key/, 'key one over max_key_length rejected' );
like( req( $sock, 'COUNT ' . 'k' x 11 ), qr/^ERR bad key/, 'query with an oversized key rejected' );
#
# keys may hold any high-bit bytes
#
my $key = "caf\xC3\xA9";
print $sock "MARK $key\n";
is( req( $sock, "COUNT $key" ), 'OK 1', 'high-bit bytes are legal in keys' );
#
t/sukkal-protocol.t view on Meta::CPAN
}
#
# a single line larger than the read buffer ceiling is rejected and
# the connection dropped
#
{
my $c = connect_daemon($d);
my $junk = 'x' x ( 2 * 1024 * 1024 ); # 2 MB, no newline
my $off = 0;
while ( $off < length $junk ) {
my $n = syswrite $c, $junk, 65536, $off;
last unless defined $n; # daemon dropped us mid-write
$off += $n;
}
is( read_line($c), 'ERR line too long', 'oversized line without a newline is rejected' );
is( read_line($c), undef, 'and the connection is dropped' );
}
# the daemon survived all of the above
is( req( $sock, 'PING' ), 'OK PONG', 'daemon still serving after the abuse' );
( run in 0.903 second using v1.01-cache-2.11-cpan-995e09ba956 )