view release on metacpan or  search on metacpan

lib/Algorithm/ToNumberMunger.pm  view on Meta::CPAN

part, so C<< parts => ['rate', 'count'] >> costs exactly two commands.

=item * C<on_error> - C<'die'> (default) croaks the write when the daemon is
unreachable or replies C<ERR>; a number is returned instead as a quiet fallback.
Note C<0> is indistinguishable from a genuinely idle key, so quiet fallback
biases the column -- loud is the default on purpose.

=item * C<timeout> - per-operation socket timeout in seconds (default 5,
best-effort via C<SO_RCVTIMEO>/C<SO_SNDTIMEO>), so a wedged daemon cannot hang a
writer forever.

=back

Semantics worth knowing: a marked read B<includes the event just marked>; keys
have whitespace/control bytes replaced with C<_> to satisfy the daemon's key
rules; connections are made lazily on first use and kept open (reconnecting
transparently after a fork or an error), so compiling a plan -- including the
eager validation in C<write_info> -- needs no running daemon. Each eps column
costs one unix-socket round trip per row; the multi-output form exists so
rate+count of the same key costs one round trip, not two.

=cut

# Default socket path of the iqbi-damiq daemon.
our $EPS_SOCKET = '/var/run/iqbi-damiq.sock';

# Persistent daemon connections, keyed by socket path, shared by every eps
# munger in the process. Entries record the pid that opened them so a forked
# writer transparently reopens instead of sharing a socket with its parent.
# Connections are made lazily on first use -- never at munger build time, so a
# plan can compile (eager validation) with no daemon running.
my %EPS_CONN;

sub _eps_conn {
	my ( $path, $timeout ) = @_;
	my $c = $EPS_CONN{$path};
	return $c->{fh} if $c && $c->{pid} == $$;

	require Socket;
	require IO::Socket::UNIX;
	my $fh = IO::Socket::UNIX->new(
		Type => Socket::SOCK_STREAM(),
		Peer => $path,
	) or die "cannot connect to iqbi-damiq at $path: $!\n";

	# Best-effort read/write timeouts so a wedged daemon cannot hang a writer.
	eval {
		my $tv = pack( 'l!l!', $timeout, 0 );
		setsockopt( $fh, Socket::SOL_SOCKET(), Socket::SO_RCVTIMEO(), $tv );
		setsockopt( $fh, Socket::SOL_SOCKET(), Socket::SO_SNDTIMEO(), $tv );
	};

	$EPS_CONN{$path} = { fh => $fh, pid => $$ };
	return $fh;
} ## end sub _eps_conn

# One pipelined transaction: send $cmd (possibly several lines) and read
# $nreplies "OK n" lines, one per command sent. The munger only ever sends
# commands that reply exactly once -- MARKRATE (which marks AND returns the
# rate in one go), RATE, COUNT, TOTAL; never a bare MARK, whose reply-only-on-
# error behavior would let a failure desynchronize the reply stream. Dies on
# ERR, EOF, or timeout; the caller still drops the cached connection on error
# as belt and braces.
sub _eps_txn {
	my ( $path, $timeout, $cmd, $nreplies ) = @_;
	my $fh = _eps_conn( $path, $timeout );
	print {$fh} $cmd or die "write to iqbi-damiq failed: $!\n";
	my @out;
	for ( 1 .. $nreplies ) {
		my $reply = <$fh>;
		die "iqbi-damiq closed the connection (or timed out)\n"
			unless defined $reply;
		$reply =~ /\AOK (\S+)/
			or die "iqbi-damiq replied: $reply";
		push @out, $1 + 0;
	}
	return @out;
} ## end sub _eps_txn

# Validate the spec keys shared by the scalar and multi-output eps builders.
sub _eps_spec {
	my ( $spec, $where ) = @_;

	my $socket = defined $spec->{socket} ? $spec->{socket} : $EPS_SOCKET;
	croak "eps munger$where: 'socket' must be a non-empty path"
		unless length $socket;

	my $prefix = defined $spec->{prefix} ? $spec->{prefix} : '';
	croak "eps munger$where: 'prefix' may not contain whitespace or control " . 'characters'
		if $prefix =~ /[\s[:cntrl:]]/;

	my $mark = exists $spec->{mark} ? ( $spec->{mark} ? 1 : 0 ) : 1;

	my $timeout = defined $spec->{timeout} ? $spec->{timeout} : 5;
	croak "eps munger$where: 'timeout' must be a positive number"
		unless looks_like_number($timeout) && $timeout > 0;

	my $on_error = defined $spec->{on_error} ? $spec->{on_error} : 'die';
	croak "eps munger$where: 'on_error' must be 'die' or a number"
		unless $on_error eq 'die' || looks_like_number($on_error);

	return ( $socket, $prefix, $mark, $timeout, $on_error );
} ## end sub _eps_spec

my %EPS_READ = map { $_ => 1 } qw(rate count total);

sub _build_eps {
	my ( $spec, $where ) = @_;

	croak "eps munger$where: 'parts' is for the multi-output form (needs " . "'into'); use 'read' for a single column"
		if defined $spec->{parts};

	my ( $socket, $prefix, $mark, $timeout, $on_error ) = _eps_spec( $spec, $where );

	my $read = defined $spec->{read} ? $spec->{read} : 'rate';
	croak "eps munger$where: unknown read '$read' (known: " . join( ', ', sort keys %EPS_READ ) . ')'
		unless $EPS_READ{$read};

	# Command plan, fixed at build time. The common case -- mark and read the
	# rate -- is the daemon's single MARKRATE command. mark+count/total rides
	# MARKRATE too (its rate reply is discarded) so marking failures come back