Algorithm-EventsPerSecond

 view release on metacpan or  search on metacpan

lib/Algorithm/EventsPerSecond/Sukkal.pm  view on Meta::CPAN

=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>.

lib/Algorithm/EventsPerSecond/Sukkal.pm  view on Meta::CPAN

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},

lib/Algorithm/EventsPerSecond/Sukkal.pm  view on Meta::CPAN

		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;

lib/Algorithm/EventsPerSecond/Sukkal.pm  view on Meta::CPAN

			%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} );

t/sukkal-lifecycle.t  view on Meta::CPAN

	# intermix the writes on the wire: A marks alpha, B marks beta,
	# back and forth. MARK is fire-and-forget, so nothing to read yet.
	for ( 1 .. 20 ) {
		print $a "MARK alpha\n";
		print $b "MARK beta 2\n";
	}

	# each connection pipelines a query then a PING; replies must come
	# back on the asking socket, in the order that connection asked. A
	# connection's own marks always precede its own query (same buffer,
	# processed in order), so each sees its full count. Cross-connection
	# ordering within a select pass is deliberately not relied on.
	print $a "COUNT alpha\n";
	print $a "PING\n";
	print $b "COUNT beta\n";
	print $b "PING\n";

	is( read_line($a), 'OK 20',   'A: COUNT reflects its own 20 alpha marks' );
	is( read_line($a), 'OK PONG', 'A: PING answered next, in order, on A' );
	is( read_line($b), 'OK 40',   'B: COUNT reflects its own 40 beta marks' );
	is( read_line($b), 'OK PONG', 'B: PING answered next, in order, on B' );

t/sukkal-protocol.t  view on Meta::CPAN

{
	my $c = connect_daemon($d);
	print $c 'MARK ghost 5';    # no newline
	close $c;
	sleep 0.25;
	my ( $hdr, @keys ) = req_multi( $sock, 'KEYS' );
	ok( !grep( { $_ eq 'ghost' } @keys ), 'partial line at EOF is discarded, not applied' );
}

#
# commands pipelined after QUIT are not processed
#
{
	my $c = connect_daemon($d);
	print $c "PING\nQUIT\nPING\n";
	is( read_line($c), 'OK PONG', 'reply to the command before a pipelined QUIT' );
	is( read_line($c), 'OK BYE',  'QUIT acknowledged mid-pipeline' );
	is( read_line($c), undef,     'commands pipelined after QUIT are dropped with the connection' );
}

#