File Coverage

blib/lib/POE/Component/Resolver.pm

Criterion	Covered	Total	%
statement	143	222	64.4
branch	37	94	39.3
condition	12	26	46.1
subroutine	22	33	66.6
pod	5	5	100.0
total	219	380	57.6

package POE::Component::Resolver;
{
  $POE::Component::Resolver::VERSION = '0.921';
}

use warnings;
use strict;

use POE qw(Wheel::Run Filter::Reference);
use Carp qw(croak carp);
use Time::HiRes qw(time);
use Socket qw(
	unpack_sockaddr_in AF_INET AF_INET6
	getnameinfo NI_NUMERICSERV NI_NUMERICHOST
);

use POE::Component::Resolver::Sidecar;

use Exporter;
use base 'Exporter';
our (@EXPORT_OK) = qw(AF_INET AF_INET6);

my $next_alias_index = "aaaaaaaa";

# Determine Perl's location, per perldoc perlvar's treatment of $^X.

use Config;
my $perl_path = $Config{perlpath};
if ($^O ne 'VMS') {
	$perl_path .= $Config{_exe} unless (
		$perl_path =~ /$Config{_exe}$/i
	);
}

# Plain Perl constructor.

sub new {
	my ($class, @args) = @_;

	croak "new() requires an even number of parameters" if @args % 2;
	my %args = @args;

	my $max_resolvers   = delete($args{max_resolvers}) || 8;
	my $idle_timeout    = delete($args{idle_timeout})  || 15;
	my $debug           = delete($args{debug})         || 0;
	my $sidecar_program = delete($args{sidecar_program});

	my $af_order = delete($args{af_order});
	if (defined $af_order and @$af_order) {
		if (ref($af_order) eq "") {
			$af_order = [ $af_order ];
		}
		elsif (ref($af_order) ne "ARRAY") {
			croak "af_order must be a scalar or an array reference";
		}

		my @illegal_afs = grep { ($_ ne AF_INET) && ($_ ne AF_INET6) } @$af_order;
		croak "af_order may only contain AF_INET and/or AF_INET6" if @illegal_afs;
	}
	elsif (exists $ENV{POCO_RESOLVER_IPV}) {
		my %number_to_address_family = ( 4 => AF_INET, 6 => AF_INET6 );
		$af_order = [
			map { $number_to_address_family{$_} }
			($ENV{POCO_RESOLVER_IPV} =~ m/([46])/g)
		];
	}

	unless ($af_order and @$af_order) {
		# Default to IPv4 preference for backward compatibility.
		$af_order = [ AF_INET, AF_INET6 ];
	}

	my @error = sort keys %args;
	croak "unknown new() parameter(s): @error" if @error;

	unless (defined $sidecar_program and length $sidecar_program) {
		if ($^O eq "MSWin32") {
			$sidecar_program = \&POE::Component::Resolver::Sidecar::main;
		}
		else {
			$sidecar_program = [
				$perl_path,
				(map { "-I$_" } @INC),
				'-MPOE::Component::Resolver::Sidecar',
				'-e', 'POE::Component::Resolver::Sidecar->main()'
			];
		}
	}

	my $self = bless {
		alias => "poe_component_resolver_" . $next_alias_index++,
		debug => $debug,
	}, $class;

	POE::Session->create(
		inline_states => {
			_start           => \&_poe_start,
			_stop            => \&_poe_stop,
			_parent          => sub { undef },  # for ASSERT_DEFAULT
			_child           => sub { undef },  # for ASSERT_DEFAULT
			request          => \&_poe_request,
			shutdown         => \&_poe_shutdown,
			cancel           => \&_poe_cancel,
			sidecar_closed   => \&_poe_sidecar_closed,
			sidecar_error    => \&_poe_sidecar_error,
			sidecar_response => \&_poe_sidecar_response,
			sidecar_signal   => \&_poe_sidecar_signal,
			sidecar_eject    => \&_poe_sidecar_eject,
			sidecar_attach   => \&_poe_sidecar_attach,
		},
		heap => {
			af_order        => $af_order,
			alias           => $self->{alias},
			idle_timeout    => $idle_timeout,
			last_request_id => 0,
			max_resolvers   => $max_resolvers,
			requests        => { },
			sidecar_ring    => [ ],
			sidecar_program => $sidecar_program,
			debug           => $debug,
		}
	);

	return $self;
}

sub DESTROY {
	my $self = shift;

	# Can't resolve the session: it must already be gone.
	return unless $poe_kernel->alias_resolve($self->{alias});

	carp " destroying $self->{alias}" if $self->{debug};

	$poe_kernel->call($self->{alias}, "shutdown");
}

sub _poe_stop {
	my $heap = $_[HEAP];
	carp " stopping $heap->{alias}" if $heap->{debug};
}

sub shutdown {
	my $self = shift;

	# Can't resolve the session: it must already be gone.
	return unless $poe_kernel->alias_resolve($self->{alias});

	carp " got shutdown request for $self->{alias}" if $self->{debug};

	$poe_kernel->call($self->{alias}, "shutdown");
}

# Internal POE event handler to release all resources owned by the
# hidden POE::Session and then shut it down.  It's an event handler so
# that this code can run "within" the POE::Session.

sub _poe_shutdown {
	my ($kernel, $heap) = @_[KERNEL, HEAP];

	$heap->{shutdown} = 1;

	$kernel->alias_remove($heap->{alias});

	_poe_wipe_sidecars($heap);

	foreach my $request (values %{$heap->{requests}}) {
		$kernel->post(
			$request->{sender},
			$request->{event},
			'component shut down',
			[ ],
			{ map { $_ => $request->{$_} } qw(host service misc) },
		);

		warn " $heap->{alias} --refcount for sender $request->{sender}" if (
			$heap->{debug}
		);

		$kernel->refcount_decrement($request->{sender}, __PACKAGE__);
	}

	$heap->{requests} = {};

	# No more sidecars to eject.
	$kernel->delay(sidecar_eject => undef);
}

# POE event handler to accept a request from some other session.  The
# public Perl resolve() method forwards into this.  This runs "within"
# the session so the resources it creates are properly owned.

sub _poe_request {
	my ($kernel, $heap, $host, $service, $hints, $event, $misc) = @_[
		KERNEL, HEAP, ARG0..ARG4
	];

	return if $heap->{shutdown};

	my $request_id = ++$heap->{last_request_id};
	my $sender_id  = $_[SENDER]->ID();

	warn " $heap->{alias} ++refcount for sender $sender_id" if (
		$heap->{debug}
	);

	$kernel->refcount_increment($sender_id, __PACKAGE__);

	_poe_setup_sidecar_ring($kernel, $heap);

	my $next_sidecar = pop @{$heap->{sidecar_ring}};
	unshift @{$heap->{sidecar_ring}}, $next_sidecar;

	$next_sidecar->put( [ $request_id, $host, $service, $hints ] );

	$heap->{requests}{$request_id} = {
		begin       => time(),
		host        => $host,
		service     => $service,
		hints       => $hints,
		sender      => $sender_id,
		event       => $event,
		misc        => $misc,
		sidecar_id  => $next_sidecar->ID(),
	};

	# No ejecting until we're done.
	$kernel->delay(sidecar_eject => undef);

	return $request_id;
}

# The user wishes to cancel a DNS request that may still be in
# progress.  This can happen in places like PoCo::Client::HTTP when
# the HTTP request times out before the DNS request is done.
#
# The public cancel() API forwards the cancelation request into the
# POE::Session managing requests via POE::Kernel's call() method.

sub cancel {
	my ($self, $request_id) = @_;
	return $poe_kernel->call($self->{alias}, "cancel", $request_id);
}

# The inside-POE cancelation code.  It must run within POE so that the
# proper resources are removed from the correct session.

sub _poe_cancel {
	my ($kernel, $heap, $request_id) = @_[KERNEL, HEAP, ARG0];

	return unless exists $heap->{requests}{$request_id};

	my $request = $heap->{requests}{$request_id};
	_sidecar_cleanup($kernel, $heap, $request->{sidecar_id});
}

# POE _start handler.  Initialize the session and start sidecar
# processes, which are owned and managed by that session.

sub _poe_start {
	my ($kernel, $heap) = @_[KERNEL, HEAP];

	carp " starting $heap->{alias}" if $heap->{debug};

	$kernel->alias_set($heap->{alias});

	#_poe_setup_sidecar_ring($kernel, $heap);

	undef;
}

# Internal helper sub.  Make sure the apprpriate number of sidecar
# resolvers are running at any given time.

sub _poe_setup_sidecar_ring {
	my ($kernel, $heap) = @_;

	return if $heap->{shutdown};

	while (scalar(keys %{$heap->{sidecar}}) < $heap->{max_resolvers}) {
		my $sidecar = POE::Wheel::Run->new(
			StdioFilter  => POE::Filter::Reference->new(),
			StdoutEvent  => 'sidecar_response',
			StderrEvent  => 'sidecar_error',
			CloseEvent   => 'sidecar_closed',
			Program      => $heap->{sidecar_program},
		);

		$heap->{sidecar}{$sidecar->PID}   = $sidecar;
		$heap->{sidecar_id}{$sidecar->ID} = $sidecar;
		push @{$heap->{sidecar_ring}}, $sidecar;

		$kernel->sig_child($sidecar->PID(), "sidecar_signal");
	}
}

# Internal helper sub to replay pending requests when their associated
# sidecars are destroyed.

sub _poe_replay_pending {
	my ($kernel, $heap) = @_;

	while (my ($request_id, $request) = each %{$heap->{requests}}) {

		# This request is riding in an existing sidecar.
		# No need to replay it.
		next if exists $heap->{sidecar_id}{$request->{sidecar_id}};

		# Give the request to a new sidecar.
		my $next_sidecar = pop @{$heap->{sidecar_ring}};
		unshift @{$heap->{sidecar_ring}}, $next_sidecar;

		$request->{sidecar_id} = $next_sidecar->ID();

		$next_sidecar->put(
			[
				$request_id, $request->{host}, $request->{service}, $request->{hints}
			]
		);
	}
}

# Internal event handler to briefly defer replaying requests until any
# responses in the queue have had time to be delivered.  This prevents
# us from replaying requests that may already have answers.

sub _poe_sidecar_attach {
	my ($kernel, $heap) = @_[KERNEL, HEAP];

	# Nothing to do if we don't have requests.
	return unless scalar keys %{$heap->{requests}};

	# Requests exist.
	_poe_setup_sidecar_ring($kernel, $heap);
	_poe_replay_pending($kernel, $heap);
}

# Plain public Perl method.  Begin resolving something.

sub resolve {
	my ($self, @args) = @_;

	croak "resolve() requires an even number of parameters" if @args % 2;
	my %args = @args;

	my $host = delete $args{host};
	croak "resolve() requires a host" unless defined $host and length $host;

	my $service = delete $args{service};
	croak "resolve() requires a service" unless (
		defined $service and length $service
	);

	my $misc = delete $args{misc};
	$misc = "" unless defined $misc;

	my $hints = delete $args{hints};
	$hints ||= { };

	my $event = delete $args{event};
	$event = "resolver_response" unless defined $event and length $event;

	my @error = sort keys %args;
	croak "unknown resolve() parameter(s): @error" if @error;

	my $result;
	croak "resolve() on shutdown resolver" unless (
		$result = $poe_kernel->call(
			$self->{alias}, "request", $host, $service, $hints, $event, $misc
		)
	);

	carp " $self->{alias} request for host($host) service($service)" if (
		$self->{debug}
	);

	return $result;
}

# A sidecar process has produced an error or warning.  Pass it
# through to the parent process' console.

sub _poe_sidecar_error {
	warn __PACKAGE__, " error in getaddrinfo subprocess: $_[ARG0]\n";
}

# A sidecar process has closed its standard output.  We will receive
# no more responses from it.  Clean up the sidecar's resources, and
# start new ones if necessary.

sub _poe_sidecar_closed {
	my ($kernel, $heap, $wheel_id) = @_[KERNEL, HEAP, ARG0];

	# Don't bother checking for pending requests if we've shut down.
	return if $heap->{shutdown};

	_sidecar_cleanup($kernel, $heap, $wheel_id);
}

sub _sidecar_cleanup {
	my ($kernel, $heap, $wheel_id) = @_;

	my $sidecar = delete $heap->{sidecar_id}{$wheel_id};
	if (defined $sidecar) {
		$sidecar->kill(9);
		delete $heap->{sidecar}{$sidecar->PID()};
	}

	# Remove the sidecar from the rotation.
	my $i = @{$heap->{sidecar_ring}};
	while ($i--) {
		next unless $heap->{sidecar_ring}[$i]->ID() == $wheel_id;
		splice(@{$heap->{sidecar_ring}}, $i, 1);
		last;
	}

	_poe_setup_sidecar_ring($kernel, $heap);
	_poe_replay_pending($kernel, $heap) if scalar keys %{$heap->{requests}};
}

# A sidecar has produced a response.  Pass it back to the original
# caller of resolve().  If we've run out of requests, briefly defer a
# partial shutdown.  We don't need all those sidecar processes if we
# might be done.

sub _poe_sidecar_response {
	my ($kernel, $heap, $response_rec) = @_[KERNEL, HEAP, ARG0];
	my ($request_id, $error, $addresses) = @$response_rec;

	my $request_rec = delete $heap->{requests}{$request_id};
	return unless defined $request_rec;

	if (defined $heap->{af_order}) {
		my @filtered_addresses;
		foreach my $af_filter (@{$heap->{af_order}}) {
			push @filtered_addresses, grep { $_->{family} == $af_filter } @$addresses;
		}
		$addresses = \@filtered_addresses;
	}

	$kernel->post(
		$request_rec->{sender}, $request_rec->{event},
		$error, $addresses,
		{ map { $_ => $request_rec->{$_} } qw(host service misc) },
	);

	warn " $heap->{alias} --refcount for sender $request_rec->{sender}" if (
		$heap->{debug}
	);

	$kernel->refcount_decrement($request_rec->{sender}, __PACKAGE__);

	# No more requests?  Consder detaching sidecar.
	$kernel->delay(sidecar_eject => $heap->{idle_timeout}) unless (
		scalar keys %{$heap->{requests}}
	);
}

# A sidecar process has exited.  Clean up its resources, and attach a
# replacement sidecar if there are requests.

sub _poe_sidecar_signal {
	my ($heap, $pid) = @_[HEAP, ARG1];

	return unless exists $heap->{sidecar}{$pid};
	my $sidecar = delete $heap->{sidecar}{$pid};
	my $sidecar_id = $sidecar->ID();
	delete $heap->{sidecar_id}{$sidecar_id};

	# Remove the sidecar from the rotation.
	my $i = @{$heap->{sidecar_ring}};
	while ($i--) {
		next unless $heap->{sidecar_ring}[$i]->ID() == $sidecar_id;
		splice(@{$heap->{sidecar_ring}}, 1, 1);
		last;
	}

	$_[KERNEL]->yield("sidecar_attach") if scalar keys %{$heap->{requests}};

	undef;
}

# Event handler to defer wiping out all sidecars.  This allows for
# lazy cleanup, which may eliminate thrashing in some situations.

sub _poe_sidecar_eject {
	my ($kernel, $heap) = @_[KERNEL, HEAP];
	_poe_wipe_sidecars($heap) unless scalar keys %{$heap->{requests}};
}

# Internal helper sub to synchronously wipe out all sidecars.

sub _poe_wipe_sidecars {
	my $heap = shift;

	return unless @{$heap->{sidecar_ring}};

	foreach my $sidecar (@{$heap->{sidecar_ring}}) {
		$sidecar->kill(-9);
	}

	$heap->{sidecar}      = {};
	$heap->{sidecar_id}   = {};
	$heap->{sidecar_ring} = [];
}

sub unpack_addr {
	my ($self, $address_rec) = @_;

	# [rt.cpan.org 76314] Untaint the address.
	my ($input_addr) = ($address_rec->{addr} =~ /\A(.*)\z/s);

	my ($error, $address, $port) = (
		(getnameinfo $input_addr, NI_NUMERICHOST | NI_NUMERICSERV)[0,1]
	);

	return if $error;
	return($address, $port) if wantarray();
	return $address;
}

1;

__END__