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.

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

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

sub run {
	my ($self) = @_;

	die "already running\n" if $self->{running};

	$self->_listen;
	local $SIG{PIPE} = 'IGNORE';

	my $rsel = $self->{rsel} = IO::Select->new( $self->{listener} );
	my $wsel = $self->{wsel} = IO::Select->new;

	$self->{running}    = 1;
	$self->{next_sweep} = time() + $self->{sweep_interval};

	while ( $self->{running} ) {
		my $timeout = $self->{next_sweep} - time();
		$timeout = 0 if $timeout < 0;

		my ( $r, $w ) = IO::Select->select( $rsel, $wsel, undef, $timeout );

		for my $fh ( @{ $r || [] } ) {
			if ( fileno($fh) == $self->{listener_fd} ) {
				$self->_accept;
			} else {
				$self->_read_client($fh);
			}
		}

		for my $fh ( @{ $w || [] } ) {
			# a connection dropped during the read pass may still be in
			# this list; its handle is closed, so fileno is undef
			my $id = fileno $fh;
			next unless defined $id && $self->{conns}{$id};
			$self->_flush( $self->{conns}{$id} );
		}

		if ( time() >= $self->{next_sweep} ) {
			$self->_sweep;
			$self->{next_sweep} = time() + $self->{sweep_interval};
		}
	} ## end while ( $self->{running} )

	$self->_shutdown;
	return $self;
} ## end sub run

=head2 stop

Ask a running daemon to shut down. Safe to call from a signal handler;
the L</run> loop notices on its next wakeup. Returns the daemon
object.

=cut

sub stop {
	my ($self) = @_;
	$self->{running} = 0;
	return $self;
}

sub _listen {
	my ($self) = @_;
	my $path = $self->{socket};

	if ( -e $path ) {
		die "$path exists and is not a socket\n" unless -S _;
		my $probe = IO::Socket::UNIX->new( Type => SOCK_STREAM, Peer => $path );
		die "something is already listening on $path\n" if $probe;
		unlink $path or die "cannot remove stale socket $path: $!\n";
	}

	my $listener = IO::Socket::UNIX->new(
		Type   => SOCK_STREAM,
		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};