view release on metacpan or  search on metacpan

lib/Async/Redis/Pool.pm  view on Meta::CPAN

        $self->{_active}{refaddr($conn)} = $conn;
        return $conn;
    }

    # At max capacity - wait for release
    my $waiter = Future->new;
    push @{$self->{_waiters}}, $waiter;

    my $timeout_future = Future::IO->sleep($self->{acquire_timeout})->then(sub {
        Future->fail(Async::Redis::Error::Timeout->new(
            message => "Acquire timed out after $self->{acquire_timeout}s",
            timeout => $self->{acquire_timeout},
        ));
    });

    my $wait_f = Future->wait_any($waiter, $timeout_future);

    my $result;
    eval {
        $result = await $wait_f;
    };
    my $error = $@;

    # If waiter was cancelled by timeout, remove from queue
    if (!$waiter->is_done) {
        @{$self->{_waiters}} = grep { $_ != $waiter } @{$self->{_waiters}};
    }

    die $error if $error;
    return $result;
}

# Release a connection back to the pool
sub release {
    my ($self, $conn) = @_;

    return unless defined $conn;

    # Check for fork - if forked, don't return to pool
    if ($self->_check_fork) {
        # Pool was cleared, just drop this connection
        return;
    }

    my $id = refaddr($conn);
    unless (exists $self->{_active}{$id}) {
        warn "Pool: release called on unknown or already-released connection";
        return;
    }
    delete $self->{_active}{$id};

    # After shutdown, destroy instead of pooling
    if ($self->{_shutdown}) {
        $self->_destroy_connection($conn);
        return;
    }

    # Check if connection is dirty
    if ($conn->is_dirty) {
        if ($self->{on_dirty} eq 'cleanup' && $self->_can_attempt_cleanup($conn)) {
            # Attempt cleanup asynchronously
            $self->_track_pending(
                $self->_cleanup_connection($conn)->on_done(sub {
                    $self->_return_to_pool($conn);
                })->on_fail(sub {
                    $self->_destroy_connection($conn);
                    $self->_maybe_create_replacement;
                })
            );
        }
        else {
            # Default: destroy and potentially replace
            $self->_destroy_connection($conn);
            $self->_maybe_create_replacement;
        }
        return;
    }

    # Clean connection - return to pool or give to waiter
    $self->_return_to_pool($conn);
}

sub _return_to_pool {
    my ($self, $conn) = @_;

    # Give to waiting acquirer if any
    if (@{$self->{_waiters}}) {
        my $waiter = shift @{$self->{_waiters}};
        $self->{_active}{refaddr($conn)} = $conn;
        $waiter->done($conn);
        return;
    }

    # Return to idle pool
    push @{$self->{_idle}}, $conn;
}

# Create a new connection
async sub _create_connection {
    my ($self) = @_;

    my $conn = Async::Redis->new(%{$self->{_conn_args}});
    await $conn->connect;

    $self->{_total_created}++;

    return $conn;
}

# Destroy a connection
sub _destroy_connection {
    my ($self, $conn) = @_;

    eval { $conn->disconnect };
    $self->{_total_destroyed}++;
}

# Maybe create a replacement connection to maintain min
sub _maybe_create_replacement {
    my ($self) = @_;

    my $current_total = (scalar keys %{$self->{_active}})
                      + (scalar @{$self->{_idle}})
                      + $self->{_creating};

    if ($current_total < $self->{min}) {
        # Create replacement asynchronously
        $self->{_creating}++;
        $self->_track_pending(
            $self->_create_connection->on_done(sub {
                my ($conn) = @_;
                $self->{_creating}--;
                $self->_return_to_pool($conn);
            })->on_fail(sub {
                $self->{_creating}--;
                # Failed to create replacement - log and continue
                warn "Failed to create replacement connection: @_";
            })
        );
    }
}

# Track a pending background future
sub _track_pending {
    my ($self, $f) = @_;

    push @{$self->{_pending}}, $f;

    # Clean up completed futures
    $f->on_ready(sub {
        @{$self->{_pending}} = grep { !$_->is_ready } @{$self->{_pending}};
    });

    return $f;
}

# Health check
async sub _health_check {
    my ($self, $conn) = @_;

    # Can't PING a pubsub connection
    if ($conn->in_pubsub) {
        return 0;
    }

    # Quick PING with 1 second timeout
    my $ok = 0;
    eval {
        my $ping_f = $conn->ping;
        my $timeout_f = Future::IO->sleep(1)->then(sub { Future->fail('health_timeout') });
        my $result = await Future->wait_any($ping_f, $timeout_f);
        $ok = 1 if defined $result && $result eq 'PONG';
    };

    return $ok;
}

# Check if cleanup can be attempted
sub _can_attempt_cleanup {
    my ($self, $conn) = @_;

    # NEVER attempt cleanup for these states:

    # PubSub mode - UNSUBSCRIBE returns confirmation frames that
    # must be correctly drained in modal pubsub mode. Too risky.
    return 0 if $conn->in_pubsub;

lib/Async/Redis/Pool.pm  view on Meta::CPAN

async sub _cleanup_connection {
    my ($self, $conn) = @_;

    # Note: Only called for in_multi or watching states
    # PubSub and inflight connections are always destroyed

    eval {
        # Reset transaction state
        if ($conn->in_multi) {
            my $discard_f = $conn->command('DISCARD');
            my $timeout_f = Future::IO->sleep($self->{cleanup_timeout})->then(sub {
                Future->fail('cleanup_timeout');
            });
            await Future->wait_any($discard_f, $timeout_f);
            $conn->{in_multi} = 0;
        }

        if ($conn->watching) {
            my $unwatch_f = $conn->command('UNWATCH');
            my $timeout_f = Future::IO->sleep($self->{cleanup_timeout})->then(sub {
                Future->fail('cleanup_timeout');
            });
            await Future->wait_any($unwatch_f, $timeout_f);
            $conn->{watching} = 0;
        }
    };

    if ($@) {
        die "Cleanup failed: $@";
    }

    # Verify connection is now clean
    if ($conn->is_dirty) {
        die "Connection still dirty after cleanup";
    }

    return $conn;
}

# The recommended pattern
async sub with {
    my ($self, $code) = @_;

    my $conn = await $self->acquire;
    my $result;
    my $error;

    eval {
        $result = await $code->($conn);
    };
    $error = $@;

    # Always release, even on exception
    # release() handles dirty detection
    $self->release($conn);

    die $error if $error;
    return $result;
}

# Shutdown the pool — synchronous. Blocks new acquires, fails waiters,
# closes idle connections. Active connections are destroyed when released.
sub shutdown {
    my ($self) = @_;
    return if $self->{_shutdown};
    $self->{_shutdown} = 1;

    # Close idle connections
    for my $conn (@{$self->{_idle}}) {
        $self->_destroy_connection($conn);
    }
    $self->{_idle} = [];

    # Fail all pending acquire waiters
    for my $waiter (@{$self->{_waiters}}) {
        $waiter->fail(Async::Redis::Error::Disconnected->new(
            message => "Pool is shutting down",
        )) unless $waiter->is_ready;
    }
    $self->{_waiters} = [];
}

1;

__END__

=head1 NAME

Async::Redis::Pool - Connection pool for Async::Redis

=head1 SYNOPSIS

    my $pool = Async::Redis::Pool->new(
        host => 'localhost',
        min  => 2,
        max  => 10,
    );

    # Recommended: scoped pattern
    my $result = await $pool->with(async sub {
        my ($redis) = @_;
        await $redis->incr('counter');
    });

    # Manual acquire/release (be careful!)
    my $redis = await $pool->acquire;
    await $redis->set('key', 'value');
    $pool->release($redis);

=head1 DESCRIPTION

Manages a pool of Redis connections with automatic dirty detection. Pool-specific
options are consumed by C<Async::Redis::Pool>; all other constructor arguments
are passed through to C<< Async::Redis->new >>.

=head1 CONSTRUCTOR

=head2 new

    my $pool = Async::Redis::Pool->new(
        host            => 'localhost',