view release on metacpan or  search on metacpan

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

package Algorithm::EventsPerSecond::Sukkal;

use 5.006;
use strict;
use warnings;

use Errno qw(EAGAIN EWOULDBLOCK EINTR);
use IO::Select;
use IO::Socket::UNIX;
use Socket qw(SOCK_STREAM);
use Algorithm::EventsPerSecond;

=encoding utf8

=head1 NAME

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,

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

			}
			$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};
		return $self->_drop($c);
	}
	substr( $c->{wbuf}, 0, $n ) = '';
	if ( $c->{wbuf} eq '' ) {
		$self->{wsel}->remove( $c->{fh} );
		$self->_drop($c) if $c->{closing};
	}
	return;
} ## end sub _flush

sub _drop {
	my ( $self, $c ) = @_;
	my $fh = $c->{fh};
	$self->{rsel}->remove($fh);
	$self->{wsel}->remove($fh);
	delete $self->{conns}{ $c->{id} };
	close $fh;
	return;
}

sub _sweep {
	my ($self) = @_;
	my $meters = $self->{meters};
	my $cutoff = time() - $self->{idle_timeout};
	delete @$meters{ grep { $meters->{$_}{seen} < $cutoff } keys %$meters };
	return;
}

sub _shutdown {
	my ($self) = @_;
	$self->_drop($_) for values %{ $self->{conns} };
	if ( $self->{listener} ) {
		close delete $self->{listener};
		unlink $self->{socket};
	}
	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 ...>.

=head2 RATE <key>

Reply C<OK n> with the key's events per second averaged over the
window. Unknown keys read as C<OK 0>.

=head2 MARKRATE <key> [<count>]

Record one event, or C<count> events, against C<key> exactly as
L</MARK> would, then reply C<OK n> with the key's rate as L</RATE>
would — a mark and a query in a single round trip. Rejects with
C<ERR ...> under the same conditions as L</MARK>.

=head2 COUNT <key>

Reply C<OK n> with the number of events inside the window. Unknown
keys read as C<OK 0>.

=head2 TOTAL <key>

Reply C<OK n> with the key's lifetime event count. Unknown (or
evicted) keys read as C<OK 0>.

=head2 STATS [<key>]

With a key, reply C<OK rate=n count=n total=n window=n> for it. With
no key, reply the daemon's own statistics: tracked keys, connected
clients, the daemon-wide mark rate and totals, uptime, window, and
which L<Algorithm::EventsPerSecond> backend is loaded.

=head2 KEYS

Reply C<OK n>, then one key per line, then C<END>.

=head2 DUMP

Reply C<OK n>, then C<< <key> <rate> <count> <total> >> per line, then
C<END>. Note each row costs an O(window) scan, so on huge key counts
with long windows prefer targeted queries.

=head2 RESET <key>

Zero the key's meter and lifetime total, as
L<Algorithm::EventsPerSecond/reset>. Replies C<OK>.

=head2 DEL <key>