Async-Redis
view release on metacpan or search on metacpan
0.001002 2026-01-17
- Bug Fix: Concurrent command response matching
- Fixed race condition where multiple async commands on a single
connection could receive mismatched responses
- Implemented Response Queue pattern with FIFO ordering
- Commands now register in inflight queue before sending
- Single reader coroutine processes responses in order
- New Features:
- Added inflight_count() method to check pending commands
- Added _wait_for_inflight_drain() for pipeline/PubSub synchronization
- Documentation:
- Added CONCURRENT COMMANDS section to POD
- Documented Response Queue pattern and best practices
- Testing:
- Added t/92-concurrency/response-ordering.t test suite
- Tests for concurrent SET, GET, mixed command types
- Stress test with 100 concurrent commands
- Inflight tracking verification
0.001001 2026-01-03
examples/stress/lib/Stress/Workload.pm view on Meta::CPAN
my $seq = 0;
my $period = 1.0 / $rate_hz;
while (!$stop->is_ready) {
$seq++;
my $job = "job_${seq}";
# Pre-increment pushed BEFORE the await. Otherwise the Perl event
# loop can fire the consumer's BLPOP-response continuation before
# the driver's LPUSH-response continuation, creating a transient
# popped > pushed state even though Redis itself never popped a
# phantom message. By bumping pushed synchronously, any BLPOP
# wakeup necessarily sees pushed >= corresponding popped.
#
# We do NOT decrement on LPUSH failure: under chaos, an await can
# fail after the bytes reached Redis (response lost on disconnect),
# so we can't reliably know whether the push actually happened.
# Treating pushed as ATTEMPTS — not successes — is conservative:
# pushed never falls below actual pushes, so the invariant
# "popped > pushed" remains a true bug indicator.
$integrity->note_queue_pushed;
my $t0 = time;
lib/Async/Redis.pm view on Meta::CPAN
# 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;
lib/Async/Redis.pm view on Meta::CPAN
$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;
}
lib/Async/Redis.pm view on Meta::CPAN
}
})->()->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.)
lib/Async/Redis.pm view on Meta::CPAN
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.
lib/Async/Redis/Pool.pm view on Meta::CPAN
# 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 {
lib/Async/Redis/Pool.pm view on Meta::CPAN
# 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: @_";
lib/Async/Redis/Pool.pm view on Meta::CPAN
$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);
}
lib/Async/Redis/Subscription.pm view on Meta::CPAN
use Carp ();
use Future;
use Future::AsyncAwait;
use Future::IO;
use Scalar::Util qw(blessed refaddr weaken);
# Threshold for periodic event-loop yield inside the callback driver
# loop. Prevents stack growth when many messages are pre-queued and
# await on an already-ready Future returns synchronously.
use constant MAX_SYNC_DEPTH => 32;
sub new {
my ($class, %args) = @_;
return bless {
redis => $args{redis},
channels => {}, # channel => 1 (for regular subscribe)
patterns => {}, # pattern => 1 (for psubscribe)
sharded_channels => {}, # channel => 1 (for ssubscribe)
lib/Async/Redis/Subscription.pm view on Meta::CPAN
# user's _on_message callback, and awaits its returned Future if any
# for consumer-opted backpressure.
#
# Exits cleanly when _dequeue returns undef (subscription closed or
# paused for reconnect). Dies with the typed error if _dequeue dies
# (fatal); _run_driver's Future failure is visible through the
# client's Future::Selector to any caller using run_until_ready.
#
# Periodic sleep(0) yield every MAX_SYNC_DEPTH iterations prevents
# stack growth when messages are pre-queued and await returns
# synchronously from an already-ready Future.
async sub _run_driver {
my ($self) = @_;
my $iter = 0;
while (!$self->{_closed} && !$self->{_paused}) {
my $msg;
my $deq_ok = eval { $msg = await $self->_dequeue(1); 1 };
unless ($deq_ok) {
my $err = $@;
# _fail_fatal already set _closed and fired on_error; don't
# double-fire. Any other propagation path routes through
lib/Async/Redis/Subscription.pm view on Meta::CPAN
my $err = $@;
return if $self->{_closed} || $self->{_paused};
$self->_handle_fatal_error(
"on_message callback Future failed: $err"
);
return;
}
}
# Periodic yield prevents stack blowup when pre-queued messages
# resolve await synchronously.
await Future::IO->sleep(0) if ++$iter % MAX_SYNC_DEPTH == 0;
}
}
# Start the driver if not already running. Idempotent.
# Only starts when _on_message is set (callback mode). Iterator mode
# consumers call next() directly — no driver loop needed.
#
# Ownership: the driver Future is added to the client's Future::Selector
# ($redis->{_tasks}) and stored in $self->{_driver_step}. The selector
lib/Async/Redis/Subscription.pm view on Meta::CPAN
while your callback runs, the driver doesn't read the next frame, so
TCP fills, Redis's output buffer grows. But Redis enforces
C<client-output-buffer-limit pubsub> (defaulting to S<32mb 8mb 60>
in recent versions) — if your subscriber cannot keep up for sustained
periods, B<Redis will disconnect you>. There is no amount of
client-side buffering that changes this: the limit is on the server.
If your processing is genuinely slow, return a Future from your
callback (enabling opt-in backpressure above) AND consider moving the
expensive work to a worker pool so the callback can return quickly.
Long synchronous processing in pub/sub callbacks is an anti-pattern at
scale regardless of client.
=head1 INTERNAL LIFECYCLE METHODS
The following methods are used by L<Async::Redis> to manage subscription
state. They are not part of the public API for end consumers, but are
documented here for maintainers.
=head2 _close
script/commands.json view on Meta::CPAN
"command_flags": [
"noscript",
"loading",
"stale",
"fast",
"no_auth",
"allow_busy"
]
},
"BGREWRITEAOF": {
"summary": "Asynchronously rewrites the append-only file to disk.",
"since": "1.0.0",
"group": "server",
"complexity": "O(1)",
"acl_categories": [
"@admin",
"@slow",
"@dangerous"
],
"arity": 1,
"command_flags": [
"admin",
"noscript",
"no_async_loading"
]
},
"BGSAVE": {
"summary": "Asynchronously saves the database(s) to disk.",
"since": "1.0.0",
"group": "server",
"complexity": "O(1)",
"history": [
[
"3.2.2",
"Added the `SCHEDULE` option."
]
],
"acl_categories": [
script/commands.json view on Meta::CPAN
"display_text": "key",
"key_spec_index": 0
}
],
"command_flags": [
"readonly",
"fast"
]
},
"UNLINK": {
"summary": "Asynchronously deletes one or more keys.",
"since": "4.0.0",
"group": "generic",
"complexity": "O(1) for each key removed regardless of its size. Then the command does O(N) work in a different thread in order to reclaim memory, where N is the number of allocations the deleted objects where composed of.",
"acl_categories": [
"@keyspace",
"@write",
"@fast"
],
"arity": -2,
"key_specs": [
script/commands.json view on Meta::CPAN
"arity": 1,
"command_flags": [
"noscript",
"loading",
"stale",
"fast",
"allow_busy"
]
},
"WAIT": {
"summary": "Blocks until the asynchronous replication of all preceding write commands sent by the connection is completed.",
"since": "3.0.0",
"group": "generic",
"complexity": "O(1)",
"acl_categories": [
"@slow",
"@connection"
],
"arity": 3,
"arguments": [
{
t/03-pubsub.t view on Meta::CPAN
use Test2::V0;
use Time::HiRes qw(time);
use Future::IO;
use lib 'lib';
use Async::Redis;
# Each subtest body runs inside a single async sub and ->get is called
# exactly once at the end. This avoids the Future::AsyncAwait
# "lost its returning future" pitfall that arose when the body mixed
# fire-and-forget `(async sub { ... })->()` with synchronous `->get`
# on a separate signal Future (the caller would unwind through the
# subtest's closing brace before F::AA finished bookkeeping on the
# background sub, destroying its returning Future mid-suspension).
# ============================================================================
# Test: Pub/Sub basic flow
# ============================================================================
subtest 'publish and subscribe' => sub {
(async sub {
t/50-pubsub/on-message.t view on Meta::CPAN
$sub->on_error($cb);
is($sub->on_error, $cb, 'accessor returns the set callback');
};
subtest 'next() croaks once on_message is set (sticky mode)' => sub {
my $redis = Async::Redis->new(host => 'localhost');
my $sub = Async::Redis::Subscription->new(redis => $redis);
$sub->on_message(sub { });
# `async sub` traps exceptions onto the returned Future; ->get
# re-throws them synchronously so we can assert on $@.
my $err;
eval { $sub->next->get; };
$err = $@;
ok($err, 'next() throws');
like($err, qr/callback-driven/i, 'error mentions callback-driven');
};
subtest '_invoke_user_callback returns callback result for sync callback' => sub {
my $redis = Async::Redis->new(host => 'localhost');
my $sub = Async::Redis::Subscription->new(redis => $redis);
t/50-pubsub/on-message.t view on Meta::CPAN
$sub->on_message(sub { 1 }); # set callback mode
my $frame = [ 'message', 'chan', 'payload' ];
my $result = $sub->_dispatch_frame($frame);
# Message must be in the queue for the driver to dequeue.
is(scalar @{$sub->{_pending_messages}}, 1, 'message buffered in queue');
is($sub->{_pending_messages}[0]{type}, 'message', 'correct type');
is($sub->{_pending_messages}[0]{channel}, 'chan', 'correct channel');
is($sub->{_pending_messages}[0]{data}, 'payload', 'correct data');
# _dispatch_frame returns undef (no Future) when queued synchronously.
is($result, undef, 'dispatch returns undef (queued synchronously)');
};
subtest '_dispatch_frame: depth backpressure applies in callback mode too' => sub {
# Backpressure Future is returned when queue is at depth — same path
# in both modes since _dispatch_frame is now mode-agnostic.
my $redis = bless { message_queue_depth => 1 }, 'Async::Redis';
my $sub = Async::Redis::Subscription->new(redis => $redis);
$sub->on_message(sub { 1 }); # set callback mode
t/50-pubsub/on-message.t view on Meta::CPAN
# Give deferred GC/event-loop callbacks a chance to fire.
Future::IO->sleep(0.1)->get;
my @faa_warnings = grep { /lost.+returning future/i } @warnings;
is(scalar @faa_warnings, 0,
'no "lost its returning future" warnings from on_message path')
or note("warnings captured: @warnings");
};
# Full end-to-end backpressure timing is flaky in this test
# harness because the synchronous-callback path is extremely
# tight — messages are dispatched as soon as frames arrive,
# which can race the publisher. The backpressure LOGIC is
# covered by the unit test below (`_dispatch_frame returns
# Future when callback does`) and the failed-Future
# integration test. The end-to-end timing test is documented
# as a known gap pending a deterministic sync primitive.
subtest 'callback returning a failed Future routes to on_error' => sub {
# Fresh publisher too — the shared one can linger in a
# state that interacts oddly with the fatal-close sequence.
t/50-pubsub/unified-reader.t view on Meta::CPAN
my $redis_mock = bless { message_queue_depth => 2 }, 'Async::Redis';
my $sub = Async::Redis::Subscription->new(redis => $redis_mock);
# Prime channels so _start_driver guard passes (not needed here, just state)
$sub->{channels}{'test-ch'} = 1;
my $frame1 = ['message', 'test-ch', 'v1'];
my $frame2 = ['message', 'test-ch', 'v2'];
my $frame3 = ['message', 'test-ch', 'v3'];
# Dispatch first two — should queue synchronously (depth not exceeded yet)
my $r1 = $sub->_dispatch_frame($frame1);
my $r2 = $sub->_dispatch_frame($frame2);
ok !defined($r1) || !ref($r1), 'first dispatch returns undef (synced)';
ok !defined($r2) || !ref($r2), 'second dispatch returns undef (synced)';
is scalar(@{$sub->{_pending_messages}}), 2, '2 messages queued';
# Third dispatch at depth=2 should return a Future
my $r3 = $sub->_dispatch_frame($frame3);
ok ref($r3) && $r3->isa('Future'), 'third dispatch returns Future (queue full)';
is scalar(@{$sub->{_pending_messages}}), 2, 'queue still at depth (third not yet added)';
t/92-concurrency/reader-invariants.t view on Meta::CPAN
}
return 0;
}
subtest 'synthetic EOF mid-pipeline fails all pipeline entries with typed error' => sub {
(async sub {
my $r = new_redis();
await $r->connect;
# Schedule the pipeline but do NOT await yet — inject EOF
# synchronously before yielding to the event loop so the reader
# sees a closed socket before any response arrives.
my $pipe_f = $r->_execute_pipeline([
['SET', 'eof-k1', '1'],
['SET', 'eof-k2', '2'],
['GET', 'eof-k1'],
]);
inject_eof($r);
my $ok = eval { await $pipe_f; 1 };
ok !$ok || _array_has_error(($pipe_f->is_done ? [$pipe_f->get] : [])),
( run in 1.803 second using v1.01-cache-2.11-cpan-9581c071862 )