view release on metacpan or  search on metacpan

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