File Coverage

blib/lib/POE/Component/Client/Keepalive.pm

Criterion	Covered	Total	%
statement	363	439	82.6
branch	83	122	68.0
condition	27	51	52.9
subroutine	49	54	90.7
pod	5	5	100.0
total	527	671	78.5

package POE::Component::Client::Keepalive;
# vim: ts=2 sw=2 expandtab
$POE::Component::Client::Keepalive::VERSION = '0.272';
use warnings;
use strict;

use Carp qw(croak);
use Errno qw(ETIMEDOUT EBADF);
use Socket qw(SOL_SOCKET SO_LINGER);

use POE;
use POE::Wheel::SocketFactory;
use POE::Component::Connection::Keepalive;
use POE::Component::Resolver;
use Net::IP::Minimal qw(ip_is_ipv4);

my $ssl_available;
eval {
  require POE::Component::SSLify;
  $ssl_available = 1;
};

use constant DEBUG => 0;

use constant {
  DEBUG_DNS        => (DEBUG || 0),
  DEBUG_DEALLOCATE => (DEBUG || 0),
};

use constant TCP_PROTO => scalar(getprotobyname "tcp") || (
  die "getprotobyname('tcp') failed: $!"
);

# Manage connection request IDs.

my $current_id = 0;
my %active_req_ids;

sub _allocate_req_id {
  while (1) {
    last unless exists $active_req_ids{++$current_id};
  }
  return $active_req_ids{$current_id} = $current_id;
}

sub _free_req_id {
  my $id = shift;
  delete $active_req_ids{$id};
}

my $default_resolver;
my $instances = 0;

# The connection manager uses a number of data structures, most of
# them arrays.  These constants define offsets into those arrays, and
# the comments document them.

use constant {         # @$self = (
  SF_POOL       => 0,  #   \%socket_pool,
  SF_QUEUE      => 1,  #   \@request_queue,
  SF_USED       => 2,  #   \%sockets_in_use,
  SF_WHEELS     => 3,  #   \%wheels_by_id,
  SF_USED_EACH  => 4,  #   \%count_by_triple,
  SF_MAX_OPEN   => 5,  #   $max_open_count,
  SF_MAX_HOST   => 6,  #   $max_per_host,
  SF_SOCKETS    => 7,  #   \%socket_xref,
  SF_KEEPALIVE  => 8,  #   $keep_alive_secs,
  SF_TIMEOUT    => 9,  #   $default_request_timeout,
  SF_RESOLVER   => 10, #   $poco_client_dns_object,
  SF_SHUTDOWN   => 11, #   $shutdown_flag,
  SF_REQ_INDEX  => 12, #   \%request_id_to_wheel_id,
  SF_BIND_ADDR  => 13, #   $bind_address,
  SF_ALIAS      => 14, #   $embedded_session_alias
};                     # );

use constant {       # $socket_xref{$socket} = [
  SK_KEY       => 0, #   $conn_key,
  SK_TIMER     => 1, #   $idle_timer,
};                   # ];

# $count_by_triple{$conn_key} = $conn_count;

use constant {        # $wheels_by_id{$wheel_id} = [
  WHEEL_WHEEL   => 0, #   $wheel_object,
  WHEEL_REQUEST => 1, #   $request,
};                    # ];

# $socket_pool{$conn_key}{$socket} = $socket;

use constant {      # $sockets_in_use{$socket} = (
  USED_SOCKET => 0, #   $socket_handle,
  USED_TIME   => 1, #   $allocation_time,
  USED_KEY    => 2, #   $conn_key,
};                  # );

# @request_queue = (
#   $request,
#   $request,
#   ....
# );

use constant {          # $request = [
  RQ_SESSION     => 0,  #   $request_session,
  RQ_EVENT       => 1,  #   $request_event,
  RQ_SCHEME      => 2,  #   $request_scheme,
  RQ_ADDRESS     => 3,  #   $request_address,
  RQ_IP          => 4,  #   $request_ip,
  RQ_PORT        => 5,  #   $request_port,
  RQ_CONN_KEY    => 6,  #   $request_connection_key,
  RQ_CONTEXT     => 7,  #   $request_context,
  RQ_TIMEOUT     => 8,  #   $request_timeout,
  RQ_START       => 9,  #   $request_start_time,
  RQ_TIMER_ID    => 10, #   $request_timer_id,
  RQ_WHEEL_ID    => 11, #   $request_wheel_id,
  RQ_ACTIVE      => 12, #   $request_is_active,
  RQ_ID          => 13, #   $request_id,
  RQ_ADDR_FAM    => 14, #   $request_address_family,
  RQ_FOR_SCHEME  => 15, #   $...
  RQ_FOR_ADDRESS => 16, #   $...
  RQ_FOR_PORT    => 17, #   $...
  RQ_RESOLVER_ID => 18, #   $resolver_request_id,
};                      # ];

# Create a connection manager.

sub new {
  my $class = shift;
  croak "new() needs an even number of parameters" if @_ % 2;
  my %args = @_;

  my $max_per_host = delete($args{max_per_host}) || 4;
  my $max_open     = delete($args{max_open})     || 128;
  my $keep_alive   = delete($args{keep_alive})   || 15;
  my $timeout      = delete($args{timeout})      || 120;
  my $resolver     = delete($args{resolver});
  my $bind_address = delete($args{bind_address});

  my @unknown = sort keys %args;
  if (@unknown) {
    croak "new() doesn't accept: @unknown";
  }

  my $alias = "POE::Component::Client::Keepalive::" . ++$current_id;

  my $self = bless [
    { },                # SF_POOL
    [ ],                # SF_QUEUE
    { },                # SF_USED
    { },                # SF_WHEELS
    { },                # SF_USED_EACH
    $max_open,          # SF_MAX_OPEN
    $max_per_host,      # SF_MAX_HOST
    { },                # SF_SOCKETS
    $keep_alive,        # SF_KEEPALIVE
    $timeout,           # SF_TIMEOUT
    undef,              # SF_RESOLVER
    undef,              # SF_SHUTDOWN
    undef,              # SF_REQ_INDEX
    $bind_address,      # SF_BIND_ADDR
    undef,              # SF_ALIAS
  ], $class;

  $default_resolver = $resolver if (
    $resolver and eval { $resolver->isa('POE::Component::Resolver') }
  );

  $self->[SF_RESOLVER] = (
    $default_resolver ||= POE::Component::Resolver->new()
  );

  my $session = POE::Session->create(
    object_states => [
      $self => {
        _start                 => "_ka_initialize",
        _stop                  => "_ka_stopped",
        ka_add_to_queue        => "_ka_add_to_queue",
        ka_cancel_dns_response => "_ka_cancel_dns_response",
        ka_conn_failure        => "_ka_conn_failure",
        ka_conn_success        => "_ka_conn_success",
        ka_deallocate          => "_ka_deallocate",
        ka_dns_response        => "_ka_dns_response",
        ka_keepalive_timeout   => "_ka_keepalive_timeout",
        ka_reclaim_socket      => "_ka_reclaim_socket",
        ka_relinquish_socket   => "_ka_relinquish_socket",
        ka_request_timeout     => "_ka_request_timeout",
        ka_resolve_request     => "_ka_resolve_request",
        ka_set_timeout         => "_ka_set_timeout",
        ka_shutdown            => "_ka_shutdown",
        ka_socket_activity     => "_ka_socket_activity",
        ka_wake_up             => "_ka_wake_up",
      },
    ],
  );

  $self->[SF_ALIAS] = ref($self) . "::" . $session->ID();

  return $self;
}

# Initialize the hidden session behind this component.
# Rendezvous with the object via a mutually agreed upon alias.

sub _ka_initialize {
  my ($object, $kernel, $heap) = @_[OBJECT, KERNEL, HEAP];
  $instances++;
  $heap->{dns_requests} = { };
  $kernel->alias_set(ref($object) . "::" . $_[SESSION]->ID());
}

# When programs crash, the session may stop in a non-shutdown state.
# _ka_stopped and DESTROY catch this either way the death occurs.

sub _ka_stopped {
  $_[OBJECT][SF_SHUTDOWN] = 1;
}

sub DESTROY {
  $_[0]->shutdown();
}

# Request to wake up.  This should only happen during the edge
# condition where the component's request queue goes from empty to
# having one item.
#
# It also happens during free(), to see if there are more sockets to
# deal with.
#
# TODO - Make the _ka_wake_up stuff smart enough not to post duplicate
# messages to the queue.

sub _ka_wake_up {
  my ($self, $kernel) = @_[OBJECT, KERNEL];

  # Scan the list of requests, until we find one that can be met.
  # Fire off POE::Wheel::SocketFactory to begin the connection
  # process.

  my $request_index  = 0;
  my $currently_open = keys(%{$self->[SF_USED]}) + keys(%{$self->[SF_SOCKETS]});
  my @splice_list;

  QUEUED:
  foreach my $request (@{$self->[SF_QUEUE]}) {
    DEBUG and warn "WAKEUP: checking for $request->[RQ_CONN_KEY]";

    # Sweep away inactive requests.

    unless ($request->[RQ_ACTIVE]) {
      push @splice_list, $request_index;
      next;
    }

    # Skip this request if its scheme/address/port triple is maxed
    # out.

    my $req_key = $request->[RQ_CONN_KEY];
    next if (
      ($self->[SF_USED_EACH]{$req_key} || 0) >= $self->[SF_MAX_HOST]
    );

    # Honor the request from the free pool, if possible.  The
    # currently open socket count does not increase.

    my $existing_connection = $self->_check_free_pool($req_key);
    if ($existing_connection) {
      push @splice_list, $request_index;

      _respond(
        $request, {
          connection => $existing_connection,
          from_cache => "deferred",
        }
      );

      # Remove the wheel-to-request index.
      delete $self->[SF_REQ_INDEX]{$request->[RQ_ID]};
      _free_req_id($request->[RQ_ID]);

      next;
    }

    # we can't easily take this out of the outer loop since _check_free_pool
    # can change it from under us
    my @free_sockets   = keys(%{$self->[SF_SOCKETS]});

    # Try to free over-committed (but unused) sockets until we're back
    # under SF_MAX_OPEN sockets.  Bail out if we can't free enough.
    # TODO - Consider removing @free_sockets in least- to
    # most-recently used order.
    while ($currently_open >= $self->[SF_MAX_OPEN]) {
      last QUEUED unless @free_sockets;
      my $next_to_go = $free_sockets[rand(@free_sockets)];
      $self->_remove_socket_from_pool($next_to_go);
      $currently_open--;
    }

    # Start the request.  Create a wheel to begin the connection.
    # Move the wheel and its request into SF_WHEELS.
    DEBUG and warn "WAKEUP: creating wheel for $req_key";

    my $addr = ($request->[RQ_IP] or $request->[RQ_ADDRESS]);
    my $wheel = POE::Wheel::SocketFactory->new(
      (
        defined($self->[SF_BIND_ADDR])
        ? (BindAddress => $self->[SF_BIND_ADDR])
        : ()
      ),
      RemoteAddress => $addr,
      RemotePort    => $request->[RQ_PORT],
      SuccessEvent  => "ka_conn_success",
      FailureEvent  => "ka_conn_failure",
      SocketDomain  => $request->[RQ_ADDR_FAM],
    );

    $self->[SF_WHEELS]{$wheel->ID} = [
      $wheel,     # WHEEL_WHEEL
      $request,   # WHEEL_REQUEST
    ];

    # store the wheel's ID in the request object
    $request->[RQ_WHEEL_ID] = $wheel->ID;

    # Count it as used, so we don't over commit file handles.
    $currently_open++;
    $self->[SF_USED_EACH]{$req_key}++;

    # Temporarily store the SF_USED record under the wheel ID.  It
    # will be moved to the socket when the wheel responds.
    $self->[SF_USED]{$wheel->ID} = [
      undef,     # USED_SOCKET
      time(),    # USED_TIME
      $req_key,  # USED_KEY
    ];

    # Mark the request index as one to splice out.

    push @splice_list, $request_index;
  }
  continue {
    $request_index++;
  }

  # The @splice_list is a list of element indices that need to be
  # spliced out of the request queue.  We scan in backwards, from
  # highest index to lowest, so that each splice does not affect the
  # indices of the other.
  #
  # This removes the request from the queue.  It's vastly important
  # that the request be entered into SF_WHEELS before now.

  my $splice_index = @splice_list;
  while ($splice_index--) {
    splice @{$self->[SF_QUEUE]}, $splice_list[$splice_index], 1;
  }
}

sub allocate {
  my $self = shift;
  croak "allocate() needs an even number of parameters" if @_ % 2;
  my %args = @_;

  # TODO - Validate arguments.

  my $scheme  = delete $args{scheme};
  croak "allocate() needs a 'scheme'"  unless $scheme;
  my $address = delete $args{addr};
  croak "allocate() needs an 'addr'"   unless $address;
  my $port    = delete $args{port};
  croak "allocate() needs a 'port'"    unless $port;
  my $event   = delete $args{event};
  croak "allocate() needs an 'event'"  unless $event;
  my $context = delete $args{context};
  croak "allocate() needs a 'context'" unless $context;
  my $timeout = delete $args{timeout};
  $timeout    = $self->[SF_TIMEOUT]    unless $timeout;

  my $for_scheme  = delete($args{for_scheme}) || $scheme;
  my $for_address = delete($args{for_addr}) || $address;
  my $for_port    = delete($args{for_port}) || $port;

  croak "allocate() on shut-down connection manager" if $self->[SF_SHUTDOWN];

  my @unknown = sort keys %args;
  if (@unknown) {
    croak "allocate() doesn't accept: @unknown";
  }

  my $conn_key = (
    "$scheme $address $port for $for_scheme $for_address $for_port"
  );

  # If we have a connection pool for the scheme/address/port triple,
  # then we can maybe post an available connection right away.

  my $existing_connection = $self->_check_free_pool($conn_key);
  if (defined $existing_connection) {
    $poe_kernel->post(
      $poe_kernel->get_active_session,
      $event => {
        addr       => $address,
        context    => $context,
        port       => $port,
        scheme     => $scheme,
        connection => $existing_connection,
        from_cache => "immediate",
      }
    );
    return;
  }

  # We can't honor the request immediately, so it's put into a queue.
  DEBUG and warn "ALLOCATE: enqueuing request for $conn_key";

  my $request = [
    $poe_kernel->get_active_session(),  # RQ_SESSION
    $event,       # RQ_EVENT
    $scheme,      # RQ_SCHEME
    $address,     # RQ_ADDRESS
    undef,        # RQ_IP
    $port,        # RQ_PORT
    $conn_key,    # RQ_CONN_KEY
    $context,     # RQ_CONTEXT
    $timeout,     # RQ_TIMEOUT
    time(),       # RQ_START
    undef,        # RQ_TIMER_ID
    undef,        # RQ_WHEEL_ID
    1,            # RQ_ACTIVE
    _allocate_req_id(), # RQ_ID
    undef,        # RQ_ADDR_FAM
    $for_scheme,  # RQ_FOR_SCHEME
    $for_address, # RQ_FOR_ADDRESS
    $for_port,    # RQ_FOR_PORT
    undef,        # RQ_RESOLVER_ID
  ];

  $self->[SF_REQ_INDEX]{$request->[RQ_ID]} = $request;

  $poe_kernel->refcount_increment(
    $request->[RQ_SESSION]->ID(),
    "poco-client-keepalive"
  );

  $poe_kernel->call($self->[SF_ALIAS], ka_set_timeout     => $request);
  $poe_kernel->call($self->[SF_ALIAS], ka_resolve_request => $request);

  return $request->[RQ_ID];
}

sub deallocate {
  my ($self, $req_id) = @_;

  croak "deallocate() requires a request ID" unless(
    defined($req_id) and exists($active_req_ids{$req_id})
  );

  my $request = delete $self->[SF_REQ_INDEX]{$req_id};
  unless (defined $request) {
    DEBUG_DEALLOCATE and warn "deallocate could not find request $req_id";
    return;
  }
  _free_req_id($request->[RQ_ID]);

  # Now pass the vetted request & its ID into our manager session.
  $poe_kernel->call($self->[SF_ALIAS], "ka_deallocate", $request, $req_id);
}

sub _ka_deallocate {
  my ($self, $heap, $request, $req_id) = @_[OBJECT, HEAP, ARG0, ARG1];

  my $conn_key = $request->[RQ_CONN_KEY];
  my $existing_connection = $self->_check_free_pool($conn_key);

  # Existing connection.  Remove it from the pool, and delete the socket.
  if (defined $existing_connection) {
    $self->_remove_socket_from_pool($existing_connection->{socket});
    DEBUG_DEALLOCATE and warn(
      "deallocate called, deleted already-connected socket"
    );
    return;
  }

  # No connection yet.  Cancel the request.
  DEBUG_DEALLOCATE and warn(
    "deallocate called without an existing connection.  ",
    "cancelling connection request"
  );

  unless (exists $heap->{dns_requests}{$request->[RQ_ADDRESS]}) {
    DEBUG_DEALLOCATE and warn(
      "deallocate cannot cancel dns -- no pending request"
    );
    return;
  }

  $poe_kernel->call( $self->[SF_ALIAS], ka_cancel_dns_response => $request );
  return;
}

sub _ka_cancel_dns_response {
  my ($self, $kernel, $heap, $request) = @_[OBJECT, KERNEL, HEAP, ARG0];

  my $address = $request->[RQ_ADDRESS];
  DEBUG_DNS and warn "DNS: canceling request for $address\n";

  my $requests = $heap->{dns_requests}{$address};

  # Remove the resolver request for the address of this connection
  # request

  my $req_index = @$requests;
  while ($req_index--) {
    next unless $requests->[$req_index] == $request;
    splice(@$requests, $req_index, 1);
    last;
  }

  # Clean up the structure for the address if there are no more
  # requests to resolve that address.

  unless (@$requests) {
    DEBUG_DNS and warn "DNS: canceled all requests for $address";
    $self->[SF_RESOLVER]->cancel( $request->[RQ_RESOLVER_ID] );
    delete $heap->{dns_requests}{$address};
  }

  # cancel our attempt to connect
  $poe_kernel->alarm_remove( $request->[RQ_TIMER_ID] );
  $poe_kernel->refcount_decrement(
    $request->[RQ_SESSION]->ID(), "poco-client-keepalive"
  );
}

# Set the request's timeout, in the component's context.

sub _ka_set_timeout {
  my ($kernel, $request) = @_[KERNEL, ARG0];
  $request->[RQ_TIMER_ID] = $kernel->delay_set(
    ka_request_timeout => $request->[RQ_TIMEOUT], $request
  );
}

# The request has timed out.  Mark it as defunct, and respond with an
# ETIMEDOUT error.

sub _ka_request_timeout {
  my ($self, $kernel, $request) = @_[OBJECT, KERNEL, ARG0];

  DEBUG and warn(
    "CON: request from session ", $request->[RQ_SESSION]->ID,
    " for address ", $request->[RQ_ADDRESS], " timed out"
  );
  $! = ETIMEDOUT;

  # The easiest way to do this?  Simulate an error from the wheel
  # itself.

  if (defined $request->[RQ_WHEEL_ID]) {
    @_[ARG0..ARG3] = ("connect", $!+0, "$!", $request->[RQ_WHEEL_ID]);
    goto &_ka_conn_failure;
  }

  my ($errnum, $errstr) = ($!+0, "$!");

  # No wheel yet.  It must have timed out in connect.
  if ($request->[RQ_RESOLVER_ID]) {
    $self->[SF_RESOLVER]->cancel( $request->[RQ_RESOLVER_ID] );
    $request->[RQ_RESOLVER_ID] = undef;
  }

  # But what if there is no wheel?
  _respond_with_error($request, "connect", $errnum, $errstr),
}

# Connection failed.  Remove the SF_WHEELS record corresponding to the
# request.  Remove the SF_USED placeholder record so it won't count
# anymore.  Send a failure notice to the requester.

sub _ka_conn_failure {
  my ($self, $func, $errnum, $errstr, $wheel_id) = @_[OBJECT, ARG0..ARG3];

  DEBUG and warn "CON: sending $errstr for function $func";
  # Remove the SF_WHEELS record.
  my $wheel_rec = delete $self->[SF_WHEELS]{$wheel_id};
  my $request   = $wheel_rec->[WHEEL_REQUEST];

  # Remove the SF_USED placeholder.
  delete $self->[SF_USED]{$wheel_id};

  # remove the wheel-to-request index
  delete $self->[SF_REQ_INDEX]{$request->[RQ_ID]};
  _free_req_id($request->[RQ_ID]);

  # Discount the use by request key, removing the SF_USED record
  # entirely if it's now moot.
  my $request_key = $request->[RQ_CONN_KEY];
  $self->_decrement_used_each($request_key);

  # Tell the requester about the failure.
  _respond_with_error($request, $func, $errnum, $errstr),

  $self->_ka_wake_up($_[KERNEL]);
}

# Connection succeeded.  Remove the SF_WHEELS record corresponding to
# the request.  Flesh out the placeholder SF_USED record so it counts.

sub _ka_conn_success {
  my ($self, $socket, $wheel_id) = @_[OBJECT, ARG0, ARG3];

  # Remove the SF_WHEELS record.
  my $wheel_rec = delete $self->[SF_WHEELS]{$wheel_id};
  my $request   = $wheel_rec->[WHEEL_REQUEST];

  # remove the wheel-to-request index
  delete $self->[SF_REQ_INDEX]{$request->[RQ_ID]};
  _free_req_id($request->[RQ_ID]);

  # Remove the SF_USED placeholder, add in the socket, and store it
  # properly.
  my $used = delete $self->[SF_USED]{$wheel_id};

  unless ($request->[RQ_SCHEME] eq 'https') {
    $self->_store_socket($used, $socket);
    $self->_send_back_socket($request, $socket);
    return;
  }

  # HTTPS here.
  # Really applies to all SSL schemes.

  unless ($ssl_available) {
    die "There is no SSL support, please install POE::Component::SSLify";
  }

  eval {
    $socket = POE::Component::SSLify::Client_SSLify(
      $socket,

      # TODO - To make non-blocking sslify work, I need to somehow
      # defer the response until the following callback says it's
      # fine.  Or if the callback says there's an error, it needs to
      # be propagated out.
      #
      # Problem is, just setting the callback doesn't seem to get the
      # connection to complete (successfully or otherwise).  There
      # needs to be something more going on... but what?

#      sub {
#        my ($socket, $status, $errval) = @_;
#        $errval = 'undef' unless defined $errval;
#
#        warn "socket($socket) status($status) errval($errval)";
#
#        # Connected okay.
#        if ($status == 1) {
#          $self->_send_back_socket($request, $socket);
#          $self = $request = undef;
#          return;
#        }
#
#        # Didn't connect okay, or hasn't so far.
#        # Report the error.
#        if ($errval == 1) {
#
#          # Get all known errors, but only retain the most recent one.
#          # I'm not sure this is needed, but the API mentions an error
#          # queue, which implies that it could contain stale errors.
#
#          my $errnum;
#          while (my $new_errnum = Net::SSLeay::ERR_get_error()) {
#            $errnum = $new_errnum;
#          }
#
#          my $errstr = Net::SSLeay::ERR_error_string($errnum);
#          warn "   ssl_error($errnum) string($errstr)";
#          _respond_with_error($request, "sslify", undef, $errstr);
#
#          # TODO - May the circle be broken.
#          $self = $request = undef;
#          return;
#        }
#      }
    );
  };

  if ($@) {
    _respond_with_error($request, "sslify", undef, "$@");
    return;
  }

  # TODO - I think for SSL we just need to _store_socket().  The call
  # to _send_back_socket() should be inside the SSL callback.
  #
  # Also, I think the callback might leak.  $request and $self may
  # need to be weakened.

  $self->_store_socket($used, $socket);
  $self->_send_back_socket($request, $socket);
}

sub _store_socket {
  my ($self, $used, $socket) = @_;
  $used->[USED_SOCKET]      = $socket;
  $self->[SF_USED]{$socket} = $used;
}

sub _send_back_socket {
  my ($self, $request, $socket) = @_;

  DEBUG and warn(
    "CON: posting... to $request->[RQ_SESSION] . $request->[RQ_EVENT]"
  );

  # Build a connection object around the socket.
  my $connection = POE::Component::Connection::Keepalive->new(
    socket  => $socket,
    manager => $self,
  );

  # Give the socket to the requester.
  _respond(
    $request, {
      connection => $connection,
    }
  );
}

# The user is done with a socket.  Make it available for reuse.

sub free {
  my ($self, $socket) = @_;

  return if $self->[SF_SHUTDOWN];
  DEBUG and warn "FREE: freeing socket";

  # Remove the accompanying SF_USED record.
  croak "can't free() undefined socket" unless defined $socket;
  my $used = delete $self->[SF_USED]{$socket};
  croak "can't free() unallocated socket" unless defined $used;

  # Reclaim the socket.
  $poe_kernel->call($self->[SF_ALIAS], "ka_reclaim_socket", $used);

  # Avoid returning things by mistake.
  return;
}

# A sink for deliberately unhandled events.

sub _ka_ignore_this_event {
  # Do nothing.
}

# An internal method to fetch a socket from the free pool, if one
# exists.

sub _check_free_pool {
  my ($self, $conn_key) = @_;

  return unless exists $self->[SF_POOL]{$conn_key};

  my $free = $self->[SF_POOL]{$conn_key};

  DEBUG and warn "CHECK: reusing $conn_key";

  my $next_socket = (values %$free)[0];
  delete $free->{$next_socket};
  unless (keys %$free) {
    delete $self->[SF_POOL]{$conn_key};
  }

  # _check_free_pool() may be operating in another session, so we call
  # the correct one here.
  $poe_kernel->call($self->[SF_ALIAS], "ka_relinquish_socket", $next_socket);

  $self->[SF_USED]{$next_socket} = [
    $next_socket,  # USED_SOCKET
    time(),        # USED_TIME
    $conn_key,     # USED_KEY
  ];

  delete $self->[SF_SOCKETS]{$next_socket};

  $self->[SF_USED_EACH]{$conn_key}++;

    # Build a connection object around the socket.
    my $connection = POE::Component::Connection::Keepalive->new(
      socket  => $next_socket,
      manager => $self,
    );

  return $connection;
}

sub _decrement_used_each {
  my ($self, $request_key) = @_;
  unless (--$self->[SF_USED_EACH]{$request_key}) {
    delete $self->[SF_USED_EACH]{$request_key};
  }
}

# Reclaim a socket.  Put it in the free socket pool, and wrap it with
# select_read() to discard any data and detect when it's closed.

sub _ka_reclaim_socket {
  my ($self, $kernel, $used) = @_[OBJECT, KERNEL, ARG0];

  my $socket = $used->[USED_SOCKET];

  # Decrement the usage counter for the given connection key.
  my $request_key = $used->[USED_KEY];
  $self->_decrement_used_each($request_key);

  # Socket is closed.  We can't reuse it.
  unless (defined fileno $socket) {
    DEBUG and warn "RECLAIM: freed socket has previously been closed";
    goto &_ka_wake_up;
  }

  # Socket is still open.  Check for lingering data.
  DEBUG and warn "RECLAIM: checking if socket still works";

  # Check for data on the socket, which implies that the server
  # doesn't know we're done.  That leads to desynchroniziation on the
  # protocol level, which strongly implies that we can't reuse the
  # socket.  In this case, we'll make a quick attempt at fetching all
  # the data, then close the socket.

  my $rin = '';
  vec($rin, fileno($socket), 1) = 1;
  my ($rout, $eout);
  my $socket_is_active = select ($rout=$rin, undef, $eout=$rin, 0);

  if ($socket_is_active) {
    DEBUG and warn "RECLAIM: socket is still active; trying to drain";
    use bytes;

    my $socket_had_data = sysread($socket, my $buf = "", 65536) || 0;
    DEBUG and warn "RECLAIM: socket had $socket_had_data bytes. 0 means EOF";
    DEBUG and warn "RECLAIM: Giving up on socket.";

    # Avoid common FIN_WAIT_2 issues, but only for valid sockets.
    #if ($socket_had_data and fileno($socket)) {
    if ($socket_had_data) {
      my $opt_result = setsockopt(
        $socket, SOL_SOCKET, SO_LINGER, pack("sll",1,0,0)
      );
      die "setsockopt: " . ($!+0) . " $!" if (not $opt_result and $!  != EBADF);
    }

    goto &_ka_wake_up;
  }

  # Socket is alive and has no data, so it's in a quiet, theoretically
  # reclaimable state.

  DEBUG and warn "RECLAIM: reclaiming socket";

  # Watch the socket, and set a keep-alive timeout.
  $kernel->select_read($socket, "ka_socket_activity");
  my $timer_id = $kernel->delay_set(
    ka_keepalive_timeout => $self->[SF_KEEPALIVE], $socket
  );

  # Record the socket as free to be used.
  $self->[SF_POOL]{$request_key}{$socket} = $socket;
  $self->[SF_SOCKETS]{$socket} = [
    $request_key,       # SK_KEY
    $timer_id,          # SK_TIMER
  ];

  goto &_ka_wake_up;
}

# Socket timed out.  Discard it.

sub _ka_keepalive_timeout {
  my ($self, $socket) = @_[OBJECT, ARG0];
  $self->_remove_socket_from_pool($socket);
}

# Relinquish a socket.  Stop selecting on it.

sub _ka_relinquish_socket {
  my ($kernel, $socket) = @_[KERNEL, ARG0];
  $kernel->alarm_remove($_[OBJECT][SF_SOCKETS]{$socket}[SK_TIMER]);
  $kernel->select_read($socket, undef);
}

# Shut down the component.  Release any sockets we're currently
# holding onto.  Clean up any timers.  Remove the alias it's known by.

sub shutdown {
  my $self = shift;
  return if $self->[SF_SHUTDOWN];
  $poe_kernel->call($self->[SF_ALIAS], "ka_shutdown");
}

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

  return if $self->[SF_SHUTDOWN];

  $instances--;

  # Clean out the request queue.
  foreach my $request (@{$self->[SF_QUEUE]}) {
    $self->_shutdown_request($kernel, $request);
  }
  $self->[SF_QUEUE] = [ ];

  # Clean out the socket pool.
  foreach my $sockets (values %{$self->[SF_POOL]}) {
    foreach my $socket (values %$sockets) {
      $kernel->alarm_remove($self->[SF_SOCKETS]{$socket}[SK_TIMER]);
      $kernel->select_read($socket, undef);
    }
  }

  # Stop any pending resolver requests.
  foreach my $host (keys %{$heap->{dns_requests}}) {
    DEBUG and warn "SHT: Shutting down resolver requests for $host";

    foreach my $request (@{$heap->{dns_requests}{$host}}) {
      $self->_shutdown_request($kernel, $request);
    }

    # Technically not needed since the resolver shutdown should do it.
    # They all share the same host, so canceling the first should get
    # them all.
    $self->[SF_RESOLVER]->cancel(
      $heap->{dns_requests}{$host}[0][RQ_RESOLVER_ID]
    );
  }

  $heap->{dns_requests} = { };

  # Shut down the resolver.
  DEBUG and warn "SHT: Shutting down resolver";
  if ( $self->[SF_RESOLVER] != $default_resolver ) {
    $self->[SF_RESOLVER]->shutdown();
  }
  $self->[SF_RESOLVER] = undef;

  if ( $default_resolver and !$instances ) {
    $default_resolver->shutdown();
    $default_resolver = undef;
  }

  # Finish keepalive's shutdown.
  $kernel->alias_remove($self->[SF_ALIAS]);
  $self->[SF_SHUTDOWN] = 1;

  return;
}

sub _shutdown_request {
  my ($self, $kernel, $request) = @_;

  if (defined $request->[RQ_TIMER_ID]) {
    DEBUG and warn "SHT: Shutting down resolver timer $request->[RQ_TIMER_ID]";
    $kernel->alarm_remove($request->[RQ_TIMER_ID]);
  }

  if (defined $request->[RQ_WHEEL_ID]) {
    DEBUG and warn "SHT: Shutting down resolver wheel $request->[RQ_TIMER_ID]";
    delete $self->[SF_WHEELS]{$request->[RQ_WHEEL_ID]};

    # remove the wheel-to-request index
    delete $self->[SF_REQ_INDEX]{$request->[RQ_ID]};
    _free_req_id($request->[RQ_ID]);
  }

  if (defined $request->[RQ_SESSION]) {
    my $session_id = $request->[RQ_SESSION]->ID;
    DEBUG and warn "SHT: Releasing session $session_id";
    $kernel->refcount_decrement($session_id, "poco-client-keepalive");
  }
}

# A socket in the free pool has activity.  Read from it and discard
# the output.  Discard the socket on error or remote closure.

sub _ka_socket_activity {
  my ($self, $kernel, $socket) = @_[OBJECT, KERNEL, ARG0];

  if (DEBUG) {
    my $socket_rec = $self->[SF_SOCKETS]{$socket};
    my $key = $socket_rec->[SK_KEY];
    warn "CON: Got activity on socket for $key";
  }

  # Any socket activity on a kept-alive socket implies that the socket
  # is no longer reusable.

  use bytes;
  my $socket_had_data = sysread($socket, my $buf = "", 65536) || 0;
  DEBUG and warn "CON: socket had $socket_had_data bytes. 0 means EOF";
  DEBUG and warn "CON: Removing socket from the pool";

  $self->_remove_socket_from_pool($socket);
}

sub _ka_resolve_request {
  my ($self, $kernel, $heap, $request) = @_[OBJECT, KERNEL, HEAP, ARG0];

  my $host = $request->[RQ_ADDRESS];

  # Skip DNS resolution if it's already a dotted quad.
  # ip_is_ipv4() doesn't require quads, so we count the dots.
  #
  # TODO - Do the same for IPv6 addresses containing colons?
  # TODO - Would require AF_INET6 support around the SocketFactory.
  if ((($host =~ tr[.][.]) == 3) and ip_is_ipv4($host)) {
    DEBUG_DNS and warn "DNS: $host is a dotted quad; skipping lookup";
    $kernel->call($self->[SF_ALIAS], ka_add_to_queue => $request);
    return;
  }

  # It's already pending DNS resolution.  Combine this with previous.
  if (exists $heap->{dns_requests}{$host}) {
    DEBUG_DNS and warn "DNS: $host is piggybacking on a pending lookup.\n";

    # All requests for the same host share the same resolver ID.
    # TODO - Although it should probably be keyed on host:port.
    $request->[RQ_RESOLVER_ID] = $heap->{dns_requests}{$host}[0][RQ_RESOLVER_ID];

    push @{$heap->{dns_requests}{$host}}, $request;
    return;
  }

  # New request.  Start lookup.
  $heap->{dns_requests}{$host} = [ $request ];

  $request->[RQ_RESOLVER_ID] = $self->[SF_RESOLVER]->resolve(
    event   => 'ka_dns_response',
    host    => $host,
    service => $request->[RQ_PORT],
    hints   => { protocol => TCP_PROTO },
  );

  DEBUG_DNS and warn "DNS: looking up $host in the background.\n";
}

sub _ka_dns_response {
  my ($self, $kernel, $heap, $response_error, $addresses, $request) = @_[
    OBJECT, KERNEL, HEAP, ARG0..ARG2
  ];

  # We've shut down.  Nothing to do here.
  return if $self->[SF_SHUTDOWN];

  my $request_address = $request->{host};
  my $requests = delete $heap->{dns_requests}{$request_address};

  DEBUG_DNS and warn "DNS: got response for request address $request_address";

  # Requests on record.
  if (defined $requests) {
    # We can receive responses for canceled requests.  Ignore them: we
    # cannot cancel PoCo::Client::DNS requests, so this is how we reap
    # them when they're canceled.
    if ($requests eq 'cancelled') {
      DEBUG_DNS and warn "DNS: reaping cancelled request for $request_address";
      return;
    }
    unless (ref $requests eq 'ARRAY') {
      die "DNS: got an unknown requests for $request_address: $requests";
    }
  }
  else {
    die "DNS: Unexpectedly undefined requests for $request_address";
  }

  # This is an error.  Cancel all requests for the address.
  # Tell everybody that their requests failed.
  if ($response_error) {
    DEBUG_DNS and warn "DNS: resolver error = $response_error";
    foreach my $request (@$requests) {
      _respond_with_error($request, "resolve", undef, $response_error),
    }
    return;
  }

  DEBUG_DNS and warn "DNS: got a response";

  # A response!
  foreach my $address_rec (@$addresses) {
    my $numeric = $self->[SF_RESOLVER]->unpack_addr($address_rec);

    DEBUG_DNS and warn "DNS: $request_address resolves to $numeric";

    foreach my $request (@$requests) {
      # Don't bother continuing inactive requests.
      next unless $request->[RQ_ACTIVE];
      $request->[RQ_IP] = $numeric;
      $request->[RQ_ADDR_FAM] = $address_rec->{family};
      $kernel->yield(ka_add_to_queue => $request);
    }

    # Return after the first good answer.
    return;
  }

  # Didn't return here.  No address record for the host?
  foreach my $request (@$requests) {
    DEBUG_DNS and warn "DNS: $request_address does not resolve";
    _respond_with_error($request, "resolve", undef, "Host has no address."),
  }
}


sub _ka_add_to_queue {
  my ($self, $kernel, $request) = @_[OBJECT, KERNEL, ARG0];

  push @{ $self->[SF_QUEUE] }, $request;

  # If the queue has more than one request in it, then it already has
  # a wakeup event pending.  We don't need to send another one.

  return if @{$self->[SF_QUEUE]} > 1;

  # If the component's allocated socket count is maxed out, then it
  # will check the queue when an existing socket is released.  We
  # don't need to wake it up here.

  return if keys(%{$self->[SF_USED]}) >= $self->[SF_MAX_OPEN];

  # Likewise, we shouldn't awaken the session if there are no
  # available slots for the given scheme/address/port triple.  "|| 0"
  # to avoid an undef error.

  my $conn_key = $request->[RQ_CONN_KEY];
  return if (
    ($self->[SF_USED_EACH]{$conn_key} || 0) >= $self->[SF_MAX_HOST]
  );

  # Wake the session up, and return nothing, signifying sound and fury
  # yet to come.
  DEBUG and warn "posting wakeup for $conn_key";
  $poe_kernel->post($self->[SF_ALIAS], "ka_wake_up");
  return;
}

# Remove a socket from the free pool, by the socket handle itself.

sub _remove_socket_from_pool {
  my ($self, $socket) = @_;

  my $socket_rec = delete $self->[SF_SOCKETS]{$socket};
  my $key = $socket_rec->[SK_KEY];

  # Get the blessed version.
  DEBUG and warn "removing socket for $key";
  $socket = delete $self->[SF_POOL]{$key}{$socket};

  unless (keys %{$self->[SF_POOL]{$key}}) {
    delete $self->[SF_POOL]{$key};
  }

  $poe_kernel->alarm_remove($socket_rec->[SK_TIMER]);
  $poe_kernel->select_read($socket, undef);

  # Avoid common FIN_WAIT_2 issues.
  # Commented out because fileno() will return true for closed
  # sockets, which makes setsockopt() highly unhappy.  Also, SO_LINGER
  # will cause te socket closure to block, which is less than ideal.
  # We need to revisit this another way, or just let sockets enter
  # FIN_WAIT_2.

#  if (fileno $socket) {
#    setsockopt($socket, SOL_SOCKET, SO_LINGER, pack("sll",1,0,0)) or die(
#      "setsockopt: $!"
#    );
#  }
}

# Internal function.  NOT AN EVENT HANDLER.

sub _respond_with_error {
  my ($request, $func, $num, $string) = @_;
  _respond(
    $request,
    {
      connection => undef,
      function   => $func,
      error_num  => $num,
      error_str  => $string,
    }
  );
}

sub _respond {
  my ($request, $fields) = @_;

  # Bail out early if the request isn't active.
  return unless $request->[RQ_ACTIVE] and $request->[RQ_SESSION];

  $poe_kernel->post(
    $request->[RQ_SESSION],
    $request->[RQ_EVENT],
    {
      addr        => $request->[RQ_ADDRESS],
      context     => $request->[RQ_CONTEXT],
      port        => $request->[RQ_PORT],
      scheme      => $request->[RQ_SCHEME],
      for_addr    => $request->[RQ_FOR_ADDRESS],
      for_scheme  => $request->[RQ_FOR_SCHEME],
      for_port    => $request->[RQ_FOR_PORT],
      %$fields,
    }
  );

  # Drop the extra refcount.
  $poe_kernel->refcount_decrement(
    $request->[RQ_SESSION]->ID(),
    "poco-client-keepalive"
  );

  # Remove associated timer.
  if ($request->[RQ_TIMER_ID]) {
    $poe_kernel->alarm_remove($request->[RQ_TIMER_ID]);
    $request->[RQ_TIMER_ID] = undef;
  }

  # Deactivate the request.
  $request->[RQ_ACTIVE] = undef;
}

1;

__END__