view release on metacpan or  search on metacpan

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

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

	$self->{listener}    = $listener;
	$self->{listener_fd} = fileno $listener;
	return;
} ## end sub _listen

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

	while ( my $fh = $self->{listener}->accept ) {
		if ( $self->{max_clients}
			&& keys %{ $self->{conns} } >= $self->{max_clients} )
		{
			close $fh;
			next;
		}
		$fh->blocking(0);
		my $id = fileno $fh;
		$self->{conns}{$id} = {
			fh      => $fh,
			id      => $id,
			rbuf    => '',
			wbuf    => '',
			closing => 0,
		};
		$self->{rsel}->add($fh);
	} ## end while ( my $fh = $self->{listener}->accept )
	return;
} ## end sub _accept

sub _read_client {
	my ( $self, $fh ) = @_;
	my $id = fileno $fh;
	my $c  = $self->{conns}{$id} or return;

	my $eof;
	# bounded so one firehose client cannot starve the rest of the loop
	for ( 1 .. 16 ) {
		my $n = sysread $fh, my $chunk, _READ_CHUNK;
		if ( !defined $n ) {
			last if $!{EAGAIN} || $!{EWOULDBLOCK} || $!{EINTR};
			$eof = 1;
			last;
		}
		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 {