view release on metacpan or  search on metacpan

lib/DBIO/Storage/PoolBase.pm  view on Meta::CPAN

package DBIO::Storage::PoolBase;
# ABSTRACT: Shared connection pool mechanics for async DBIO drivers

use strict;
use warnings;
use base 'DBIO::Storage::Pool';

use Carp 'croak';
use Scalar::Util ();
use namespace::clean;



sub new {
  my ($class, %args) = @_;
  croak('conninfo or conninfo_provider required')
    unless $args{conninfo} || $args{conninfo_provider};

  my $self = bless {
    conninfo          => $args{conninfo},
    conninfo_provider => $args{conninfo_provider},
    max_size          => $args{size} || 5,
    on_error          => $args{on_error} || sub { warn "Pool error: $_[0]\n" },
    _connections      => [],
    _idle             => [],
    _waiters          => [],
  }, $class;

  $self->{future_class} = $args{future_class} if $args{future_class};
  my $fc = $self->future_class;
  eval "require $fc" or croak "Cannot load future class $fc: $@";

  # karr #68: optional back-reference to the owning DBIO::Storage::Async, held
  # weakly (the storage owns the pool, not the other way round). When present,
  # the pool replays the storage's configured on_connect / on_disconnect actions
  # against every freshly-spawned / torn-down connection -- see the connect-action
  # hooks in L</_spawn_connection> and L</shutdown>. A pool used standalone, or an
  # async pool whose owner does not wire it, simply carries no {storage} and skips
  # the hooks entirely.
  if (defined $args{storage}) {
    $self->{storage} = $args{storage};
    Scalar::Util::weaken($self->{storage}) if ref $self->{storage};
  }

  return $self;
}


sub future_class { $_[0]->{future_class} || 'Future' }


sub acquire {
  my ($self) = @_;
  return $self->_acquire_slot->then(sub {
    my $conn = shift;
    return $self->_connection_ready_future($conn);
  });
}

# Raw pool-slot acquisition: idle reuse / capacity spawn / waiter queue, WITHOUT
# readiness gating. Returns a Future resolving to a connection object (or a
# pending waiter Future that resolves to one on release). acquire() wraps this in
# the _connection_ready_future seam so all three paths are gated identically and
# in exactly one place -- an async subclass overrides the seam, never this or
# acquire() (karr #75). Kept as a distinct method so a subclass can still reach
# the un-gated slot logic if it ever needs to.
sub _acquire_slot {
  my ($self) = @_;

  if (@{ $self->{_idle} }) {
    # FIFO: hand out the connection that has been idle longest (the one at
    # the front of the queue). Release pushes onto the tail, so shift takes
    # from the head — this cycles every conn through instead of always
    # reusing the most-recently-released one (LIFO), which caused 1 hot
    # conn and N-1 cold conns under bursty load.
    return $self->future_class->done(shift @{ $self->{_idle} });
  }

  if (@{ $self->{_connections} } < $self->{max_size}) {
    my $conn = $self->_spawn_connection;
    return $self->future_class->done($conn);
  }

  my $f = $self->future_class->new;
  push @{ $self->{_waiters} }, $f;
  return $f;
}


sub _connection_ready_future {
  my ($self, $conn) = @_;
  return $self->future_class->done($conn);
}


sub _register_connection_ready {
  my ($self, $conn, $ready) = @_;
  $self->{_ready}{ Scalar::Util::refaddr($conn) } = $ready;
  return $ready;
}