Async-Redis
view release on metacpan or search on metacpan
lib/Async/Redis.pm view on Meta::CPAN
# reconnect. Concurrent callers share it. The slot is the shared-await
# signal, NOT the ownership — ownership lives in $self->{_tasks}.
#
# Structured-concurrency: the reconnect task is added to the selector
# so any caller currently awaiting via run_until_ready sees reconnect
# failures propagated.
#
# NOTE: dedup is race-safe only when called from inside the write
# gate (which serialises callers). Outside the gate, a failed reconnect
# could be observed after on_ready clears the slot, allowing a second
# reconnect to start before state converges.
async sub _ensure_connected {
my ($self) = @_;
return if $self->{_socket_live};
if (my $f = $self->{_reconnect_future}) {
await $f;
return;
}
my $f = $self->_reconnect;
$self->{_reconnect_future} = $f;
$self->{_tasks}->add(data => 'reconnect', f => $f);
$f->on_ready(sub { $self->{_reconnect_future} = undef });
await $f;
}
# Reconnect and replay pubsub subscriptions
async sub _reconnect_pubsub {
my ($self) = @_;
my $sub = $self->{_subscription}
or die Async::Redis::Error::Disconnected->new(
message => "No subscription to replay",
);
my @replay = $sub->get_replay_commands;
# Ensure connection state is fully cleaned up before reconnecting.
# _reset_connection may have already been called by _read_response,
# but if the socket was closed externally, we need to clean up
# stale IO watchers and state here. It is safe to call twice —
# the on_disconnect callback is guarded by $was_connected.
$self->_reset_connection('pubsub_reconnect');
await $self->_reconnect;
# Re-enter pubsub mode before replaying so the unified reader
# classifies incoming message frames correctly during replay.
$self->{in_pubsub} = 1;
# Replay all subscription commands through the write gate and unified
# reader. Each channel/pattern gets its own command so confirmations
# are matched one-to-one via the inflight queue.
for my $cmd (@replay) {
my ($command, @args) = @$cmd;
for my $arg (@args) {
await $self->_pubsub_command($command, $arg);
}
}
}
# Asynchronously reconnect after a pubsub connection drop. Called by
# _reader_fatal when reconnect is enabled and a subscription is active.
# Fires _resume_after_reconnect on the subscription on success, or
# _fail_fatal on unrecoverable reconnect failure.
sub _reconnect_async {
my ($self, $sub) = @_;
# Dedup against any reconnect already in progress (from either this
# path or _ensure_connected). The slot is the shared signal.
return if $self->{_reconnect_future}
&& !$self->{_reconnect_future}->is_ready;
weaken(my $weak_self = $self);
weaken(my $weak_sub = $sub);
my $f = (async sub {
# Reconnect the socket. _reconnect handles retry/backoff and
# dies with Disconnected if reconnect_max_attempts is exhausted.
await $weak_self->_reconnect;
# Delegate the replay, on_reconnect, and driver-restart work to
# the subscription's unified resume path. _resume_after_reconnect
# handles clearing _paused, setting in_pubsub, replaying all
# tracked channels/patterns, firing on_reconnect, and starting
# the driver. Keeps the "who restarts what after reconnect"
# logic in one place.
if ($weak_sub) {
await $weak_sub->_resume_after_reconnect;
}
})->();
# Ownership: the selector owns the task; the slot is the dedup signal.
# No ->retain — the selector holds the strong reference.
$self->{_reconnect_future} = $f;
$self->{_tasks}->add(data => 'pubsub-reconnect', f => $f);
$f->on_ready(sub {
return unless $weak_self;
$weak_self->{_reconnect_future} = undef;
});
$f->on_fail(sub {
my $err = shift;
return unless $weak_sub;
$weak_sub->_fail_fatal($err);
});
return;
}
# Execute a Redis command
async sub command {
my ($self, $cmd, @args) = @_;
# Check for fork - invalidate connection if PID changed
$self->_check_fork;
# Block regular commands on pubsub connection
if ($self->{in_pubsub}) {
my $ucmd = uc($cmd // '');
unless ($ucmd =~ /^(SUBSCRIBE|UNSUBSCRIBE|PSUBSCRIBE|PUNSUBSCRIBE|SSUBSCRIBE|SUNSUBSCRIBE|PING|QUIT)$/) {
die Async::Redis::Error::Protocol->new(
lib/Async/Redis.pm view on Meta::CPAN
die Async::Redis::Error::Connection->new(
message => "Connection closed by server",
);
}
$self->{parser}->parse($buf);
if (my $msg = $self->{parser}->get_message) {
return $msg;
}
}
}
# Reset connection after timeout (stream is desynced)
sub _reset_connection {
my ($self, $reason) = @_;
$reason //= 'timeout';
my $was_connected = $self->{connected};
# Cancel any active read future BEFORE closing socket
# This ensures Future::IO unregisters its watcher while fileno is still valid
if ($self->{_current_read_future} && !$self->{_current_read_future}->is_ready) {
$self->{_current_read_future}->cancel;
$self->{_current_read_future} = undef;
}
# Cancel any pending inflight operations before closing socket
if (my $inflight = $self->{inflight}) {
for my $entry (@$inflight) {
if ($entry->{future} && !$entry->{future}->is_ready) {
$entry->{future}->cancel;
}
}
$self->{inflight} = [];
}
if ($self->{socket}) {
$self->_close_socket;
}
$self->{_socket_live} = 0;
$self->{_fatal_in_progress} = 0;
$self->{_reader_running} = 0;
$self->{_reconnect_future} = undef;
$self->{connected} = 0;
$self->{parser} = undef;
$self->{in_pubsub} = 0;
if ($was_connected && $self->{on_disconnect}) {
$self->{on_disconnect}->($self, $reason);
}
}
# Async write lock. The lock is a Future that resolves when the current
# holder releases. Waiters chain onto it; each waiter replaces the slot
# with its own Future before returning to the caller.
async sub _acquire_write_lock {
my ($self) = @_;
# Wait out any in-progress fatal. _reader_fatal is synchronous so this
# is typically immediate, but a callback inside fatal could yield.
# NOTE: this is a poll loop (sleep(0) per tick). Acceptable because
# _reader_fatal's transition is synchronous; if teardown ever becomes
# async, replace with a one-shot Future waiters can await on.
while ($self->{_fatal_in_progress}) {
await Future::IO->sleep(0);
}
# Chain onto the existing lock Future if any.
while (my $prev = $self->{_write_lock}) {
await $prev;
}
# We are the owner now. Install our own Future so the next caller
# waits on us.
$self->{_write_lock} = Future->new;
return;
}
sub _release_write_lock {
my ($self) = @_;
my $f = delete $self->{_write_lock};
$f->done if $f && !$f->is_ready;
}
# Wrap a body in gate acquire/release with guaranteed release even if the
# body dies. On body failure, calls _reader_fatal with a transport error.
async sub _with_write_gate {
my ($self, $body) = @_;
await $self->_acquire_write_lock;
my $ok = eval { await $body->(); 1 };
my $err = $@;
$self->_release_write_lock;
if (!$ok) {
# Convert to a typed transport error if not already.
my $typed = (ref $err && eval { $err->isa('Async::Redis::Error') })
? $err
: Async::Redis::Error::Connection->new(
message => "Write failed: $err",
host => $self->{host},
port => $self->{port},
);
$self->_reader_fatal($typed);
die $typed;
}
return;
}
# Is reconnect enabled for this client?
sub _reconnect_enabled {
my ($self) = @_;
return !!$self->{reconnect};
}
# Central "something went wrong with the stream" transition. Detaches
# inflight BEFORE closing the socket so the typed error is preserved
# (the old _reset_connection cancels inflight directly, which would
# overwrite $typed_error with a generic cancellation).
sub _reader_fatal {
my ($self, $typed_error) = @_;
return if $self->{_fatal_in_progress};
$self->{_fatal_in_progress} = 1;
lib/Async/Redis.pm view on Meta::CPAN
if ($self->{_telemetry}) {
my $elapsed_ms = (Time::HiRes::time() - $start_time) * 1000;
$self->{_telemetry}->record_pipeline($count, $elapsed_ms);
}
return \@results;
}
1;
__END__
=head1 NAME
Async::Redis - Async Redis client using Future::IO
=head1 SYNOPSIS
use Async::Redis;
use Future::AsyncAwait;
# Future::IO 0.23+ has a built-in poll-based impl that works
# out of the box. For IO::Async or UV, require the impl directly:
# require Future::IO::Impl::IOAsync; # if using IO::Async
# require Future::IO::Impl::UV; # if using UV
my $redis = Async::Redis->new(
host => 'localhost',
port => 6379,
);
(async sub {
await $redis->connect;
# Simple commands
await $redis->set('key', 'value');
my $value = await $redis->get('key');
# Pipelining for efficiency
my $pipeline = $redis->pipeline;
$pipeline->set('k1', 'v1');
$pipeline->set('k2', 'v2');
$pipeline->get('k1');
my $results = await $pipeline->execute;
# PubSub
my $sub = await $redis->subscribe('channel');
while (my $msg = await $sub->next_message) {
print "Received: $msg->{message}\n";
}
})->()->await;
B<Important:> If you're embedding Async::Redis in a larger application
(web framework, existing event loop, etc.), see L</EVENT LOOP CONFIGURATION>
for how to properly configure Future::IO. Libraries should never configure
the Future::IO backend - only your application's entry point should.
=head1 DESCRIPTION
Async::Redis is an asynchronous Redis client built on L<Future::IO>,
providing a modern, non-blocking interface for Redis operations.
Key features:
=over 4
=item * Full async/await support via L<Future::AsyncAwait>
=item * Event loop agnostic (IO::Async, AnyEvent, UV, etc.)
=item * Automatic reconnection with exponential backoff
=item * Connection pooling with health checks
=item * Pipelining and auto-pipelining
=item * PubSub with automatic subscription replay on reconnect
When a connection drops during pub/sub mode and C<reconnect> is enabled,
all subscriptions are automatically re-established. Use
C<< $subscription->on_reconnect(sub { ... }) >> to be notified when this
happens (e.g., to re-poll state that may have changed during the outage).
=item * Transaction support (MULTI/EXEC/WATCH)
=item * TLS/SSL connections
=item * OpenTelemetry observability integration
=item * Fork-safe for pre-fork servers (Starman, etc.)
=item * Full RESP2 protocol support
=item * Safe concurrent commands on single connection
=back
=head1 CONCURRENT COMMANDS
Async::Redis safely handles multiple concurrent commands on a single
connection using a response queue pattern. When you fire multiple async
commands without explicitly awaiting them:
my @futures = (
$redis->set('k1', 'v1'),
$redis->set('k2', 'v2'),
$redis->get('k1'),
);
my @results = await Future->needs_all(@futures);
Each command is registered in an inflight queue before being sent to Redis.
A single reader coroutine processes responses in FIFO order, matching each
response to the correct waiting future. This prevents response mismatch bugs
that can occur when multiple coroutines race to read from the socket.
For high-throughput scenarios, consider using:
=over 4
=item * B<Explicit pipelines> - C<< $redis->pipeline >> batches commands
lib/Async/Redis.pm view on Meta::CPAN
C<message_queue_depth> limits the number of queued pubsub messages.
Callback invocation is always serialized. With C<< message_queue_depth => 1 >>
(the default), one message may be queued while one callback is still
processing; the reader pauses when that queue slot is full. Higher values
allow more messages to buffer locally before the reader pauses.
=head1 TASK LIFECYCLE
Async::Redis organizes all fire-and-forget background work (the socket
reader, reconnect attempts, auto-pipeline submit batches, the pubsub
callback driver) under a single per-client L<Future::Selector> instance,
following Paul Evans's client pattern from L<Sys::Async::Virt> and
L<IPC::MicroSocket>.
Each background task is registered with the selector via
C<< $selector->add(data => $label, f => $task_future) >>. Command
execution awaits responses via
C<< $selector->run_until_ready($response_future) >>, which pumps the
selector and propagates any task failure to the awaiting caller. The
practical guarantee: if a background task dies (including from a
coding bug that escapes explicit fatal-error handling), awaiting
callers see a typed failure rather than hanging forever.
This structure provides the five structured-concurrency properties
articulated by the L<trio|https://trio.readthedocs.io/> /
L<asyncio.TaskGroup|https://docs.python.org/3/library/asyncio-task.html#task-groups>
ecosystems:
=over
=item * GC safety - every background task is held by the selector.
=item * Error propagation - any task's failure reaches callers awaiting
the selector via C<run_until_ready>.
=item * Cancellation - socket closure propagates to pending I/O, which
fails the owning task, which the selector propagates.
=item * Scope cleanup - C<disconnect> tears down state; remaining selector
tasks unwind via their existing on_fail handlers.
=item * Local reasoning - all concurrent work on one connection is owned
by one place.
=back
There is no user-facing API for the selector; it is internal
machinery. Clients should not call C<< $redis->{_tasks} >> directly.
I<Note:> the only use of C<< Future->retain >> in this codebase is
avoided in favor of selector ownership. Any patch that introduces
C<< ->retain >> on an Async::Redis-owned Future should instead add
the task to C<< $self->{_tasks} >> so failure propagation and lifetime
ownership are consistent.
=head1 KNOWN LIMITATIONS
=over
=item * Hostname resolution is synchronous. C<connect()> calls
C<inet_aton> before the async connect, which blocks during DNS lookup.
Not covered by C<connect_timeout>.
=item * IPv6 URI hosts are not yet supported.
=item * Some generated wrappers expose mode-changing commands (HELLO,
CLIENT REPLY, MONITOR, SYNC, PSYNC) that interact poorly with the
response model. Avoid them unless you understand the protocol
consequences.
=back
=head1 SEE ALSO
=over 4
=item * L<Future::IO> - The underlying async I/O abstraction
=item * L<Future::AsyncAwait> - Async/await syntax support
=item * L<Async::Redis::Pool> - Connection pooling
=item * L<Async::Redis::Subscription> - PubSub subscriptions
=item * L<Async::Redis::Cookbook> - Practical usage recipes
=item * L<Redis> - Synchronous Redis client
=item * L<Net::Async::Redis> - Another async Redis client
=back
=head1 AUTHOR
John Napiorkowski
=head1 COPYRIGHT AND LICENSE
This is free software; you can redistribute it and/or modify it under
the same terms as the Perl 5 programming language system itself.
=cut
( run in 0.895 second using v1.01-cache-2.11-cpan-9581c071862 )