Pub/Sub Architecture

This document describes how Dragonfly implements the Publish-Subscribe (Pub/Sub) messaging paradigm within its shared-nothing, multi-threaded architecture. It covers the global subscription registry backed by a ShardedHashMap with fine-grained per-shard locking, the RCU-style pointer swap and per-shard shared locking used for concurrent reads on the publish path, the asynchronous message delivery pipeline, and the backpressure system that protects the server from slow-subscriber OOM.

Overview

In a shared-nothing architecture, handling PUBLISH and SUBSCRIBE commands presents a unique challenge: subscriptions must be globally addressable across all threads, but taking a global lock on every PUBLISH would create a severe bottleneck. A single popular channel with thousands of subscribers could serialize all publish operations onto one shard thread.

Dragonfly solves this with a ChannelStore backed by ShardedHashMap — a hash map split into 16 independently-locked shards:

• Reads (PUBLISH / SPUBLISH) acquire a per-shard shared read lock (read_mu_) and read the SubscribeMap pointer via an atomic acquire load (UpdatablePointer::Get()). Multiple readers on the same shard proceed concurrently.
• Writes (SUBSCRIBE / UNSUBSCRIBE / PSUBSCRIBE / PUNSUBSCRIBE) acquire the shard's exclusive write lock (write_mu_), copy the SubscribeMap, apply the mutation, and atomically swap the pointer — all without blocking readers. Concurrent writes to different shards are fully independent.

This design avoids contention on a single shard thread for heavy throughput on a single channel and seamlessly scales across multiple threads even with a small number of channels. Publish latency is low because no inter-thread hop is required to look up subscribers — any thread can read the global ChannelStore directly.

Dragonfly supports three flavors of Pub/Sub:

Primary Data Structures

Data Flow Overview

Subscription Management (Shard-Locked ChannelStore)

Data Structure Layout

The ChannelStore holds two ChannelMap instances — one for exact-channel subscriptions and one for pattern subscriptions. Each ChannelMap is a ShardedHashMap<string, UpdatablePointer, 16> backed by 16 independently-locked shards:

Each shard carries two fiber-aware locks:

UpdatablePointer wraps a std::atomic<SubscribeMap*> with memory_order_acquire on read and memory_order_release on write. This ensures that when a thread reads the pointer, it also sees the fully constructed SubscribeMap that the writer published.

Two Granularities of Update

The ChannelStoreUpdater groups pending operations by shard index and processes each shard in a single Mutate() call, minimizing lock acquisitions:

1. RCU pointer swap (existing channel) — when a subscriber is added/removed from a channel that already exists in the map. The SubscribeMap is copied, the mutation is applied, and the UpdatablePointer is atomically swapped via Set(). This happens under write_mu_ only — readers are NOT blocked. The old SubscribeMap is placed in a freelist_ and deleted after acquiring read_mu_ exclusively (draining in-flight readers).

2. Structural map change (new channel / last subscriber leaves) — when a channel slot must be inserted (first subscriber) or erased (last subscriber leaves). Inside the Mutate() callback, the AcquireReaderExclusiveLock callable is invoked, which acquires read_mu_ exclusively, blocking all readers on that shard while the key is inserted or erased. Writers on other shards are unaffected.

Apply() Flow

The ChannelStoreUpdater::Apply() method iterates over each shard that has pending operations and calls map.Mutate(ShardId{sid}, ...) to acquire write_mu_ once per shard. Inside the callback:

1. Phase 1 — RCU pointer swaps (under write_mu_ only): For each key where the channel already exists, the SubscribeMap is copied, the subscriber is added/removed, and the UpdatablePointer is atomically swapped. Old SubscribeMap pointers are saved in a per-shard freelist_. Readers continue concurrently on the shared read_mu_.

2. Phase 2 — Structural map changes (under read_mu_ exclusive): If any keys require inserting a new slot or erasing the last subscriber's slot, AcquireReaderExclusiveLock() is called. This acquires read_mu_ exclusively, draining any in-flight readers on this shard. The insert/erase is then performed on the mutable map reference.

3. Phase 3 — Freelist cleanup: After Mutate() returns (releasing write_mu_), WithReadExclusiveLock(ShardId{sid}, ...) is called to acquire read_mu_ exclusively once more, ensuring all readers that may have loaded old SubscribeMap pointers have completed. The old SubscribeMap pointers in the freelist are then safely deleted.

Per-Key Mutation Logic

Inside the Mutate() callback, each key in the shard's pending operations is processed:

Phase 1 — RCU swaps (write_mu_ held, read_mu_ NOT held):

  For each key in shard_keys:
    it = m.find(key)

    Case 1: Adding, key exists (existing channel)
      → old_sm = it->second.Get()
      → new_sm = new SubscribeMap{*old_sm}
      → new_sm->emplace(cntx_, thread_id_)
      → it->second.Set(new_sm)            // atomic release-store
      → freelist_.push_back(old_sm)        // defer deletion

    Case 2: Removing, key exists, >1 subscriber
      → old_sm = it->second.Get()
      → new_sm = new SubscribeMap{*old_sm}
      → new_sm->erase(cntx_)
      → it->second.Set(new_sm)            // atomic release-store
      → freelist_.push_back(old_sm)        // defer deletion

    Case 3: Adding, key NOT in map (new channel)
      → mark needs_map_change[i] = true   // deferred to Phase 2

    Case 4: Removing, last subscriber (channel disappears)
      → mark needs_map_change[i] = true   // deferred to Phase 2

Phase 2 — Structural changes (read_mu_ acquired exclusive):

  If any needs_map_change:
    locked_map = AcquireReaderExclusiveLock()

    Case 3: → locked_map.map.emplace(key, new SubscribeMap{{cntx_, thread_id_}})
    Case 4: → delete it->second.Get()
             → locked_map.map.erase(it)

Old SubscribeMap pointers from Phase 1 are not immediately deleted because concurrent readers on other threads may still hold references obtained via UpdatablePointer::Get(). They are placed in a freelist_ and deleted only after WithReadExclusiveLock completes — at which point every reader on that shard has released read_mu_ (shared) and no reader can hold a reference to the old maps.

Connection-Level Subscription State

Each connection tracks its own subscriptions in ConnectionState::SubscribeInfo:

struct SubscribeInfo {
  absl::flat_hash_set<std::string> channels;
  absl::flat_hash_set<std::string> patterns;

  bool IsEmpty() const;
  unsigned SubscriptionCount() const;
};

• Created lazily: subscribe_info is allocated on the first SUBSCRIBE or PSUBSCRIBE. A subscriptions counter on ConnectionContext is incremented.
• Destroyed on empty: when all channels and patterns are removed, subscribe_info is reset and subscriptions is decremented.
• Gates pub/sub mode: AsyncOperations::operator()(PubMessage) checks cntx()->subscriptions == 0 and discards stale messages that arrive after full unsubscription (possible due to inter-thread dispatch delays).
• Cleaned up on disconnect: Service::OnConnectionClose calls UnsubscribeAll() and PUnsubscribeAll() if subscribe_info is non-null, ensuring the ChannelStore is updated even if the client disconnects abruptly.

Message Publishing and Dispatch

ChannelStore::SendMessages

When a client issues PUBLISH channel message (or SPUBLISH):

SendMessages(channel, messages, sharded)
  1. subscribers = FetchSubscribers(channel)
     → exact match: channels_.FindIf(channel, ...) (shared read lock on shard)
     → pattern match: patterns_.ForEachShared(...)
         for each (pat, subs): if GlobMatcher{pat}.Matches(channel): Fill(subs, pat, &result)
     → sort result by thread_id  (enables efficient per-thread dispatch)

  2. If subscribers empty → return 0

  3. Backpressure gate (per destination thread):
     For each unique subscriber thread:
       Connection::EnsureMemoryBudget(sub_thread)
       → blocks fiber if that thread's subscriber_bytes > hard limit
         (publish_buffer_limit * kPubSubHardMultiplier)

  4. Build message payload:
     BuildSender copies channel + message into a single shared_ptr<char[]>
     Creates string_view references into the shared buffer

  5. Cross-thread dispatch:
     shard_set->pool()->DispatchBrief(cb)
       → cb runs on EVERY I/O thread
       → uses lower_bound on sorted subscribers to find this thread's connections
       → for each local subscriber: conn->SendPubMessageAsync(PubMessage{...})

  6. Return subscriber count

BuildSender — Shared Buffer Optimization

The BuildSender helper (anonymous namespace in channel_store.cc) creates a functor that captures the message payload in a single allocation:

auto buf = shared_ptr<char[]>{new char[channel.size() + messages_size]};
memcpy(buf.get(), channel.data(), channel.size());
// ... copy each message contiguously after the channel

The shared_ptr is captured by value in the dispatch functor, so all subscribers share the same underlying buffer. Each PubMessage holds a copy of the shared_ptr (incrementing the reference count) along with string_views pointing into it. This avoids per-subscriber string allocations.

FetchSubscribers — Routing and Pattern Matching

FetchSubscribers(channel)
  1. Exact match: channels_.FindIf(channel, ...)
     → acquires read_mu_ shared on the channel's shard
     → if found, Fill() creates Subscriber entries from the SubscribeMap

  2. Pattern match: patterns_.ForEachShared(...)
     → iterates ALL pattern shards (each read_mu_ shared independently)
     → for each (pat, subs): GlobMatcher{pat, case_sensitive=true}.Matches(channel)
     → matching subscribers are added with their pattern string

  3. Sort by Subscriber::ByThread (thread_id ordering)
     → enables O(log n) per-thread lookup during dispatch

Note: FetchSubscribers is not a global snapshot — each shard is locked independently, so the result may not reflect a fully consistent state. This trade-off is acceptable for pub/sub use cases.

The Fill helper reads the SubscribeMap (via UpdatablePointer::Get() — acquire load) and creates Subscriber structs that hold a ConnectionRef (weak reference) obtained via conn->Borrow().

Delivery: I/O Loop and Dispatch

Once a message lands on the destination thread via DispatchBrief, the callback calls conn->SendPubMessageAsync(PubMessage{...}). This wraps the message in a MessageHandle and pushes it onto the connection's dispatch_q_ via SendAsync().

How that queue is drained depends on which I/O loop the connection uses. Dragonfly has two I/O loop implementations, selected at connection setup time:

v1: IoLoop + AsyncFiber (Redis connections)

This is the primary path for all Redis Pub/Sub traffic. Each connection manages two streams:

IoLoop runs in the connection's main fiber: it calls recv() (blocking the fiber, not the thread), parses commands into the parsed_head_ linked list, and notifies the consumer.

AsyncFiber runs as a separate dedicated fiber (Connection::AsyncFiber()) that consumes both streams in a prioritized loop:

Key insight: The dispatch queue (Control Path) takes default priority over the pipeline (Data Path). The pipeline is processed only when dispatch_q_ is empty or when the async_dispatch_quota (default: 100) is reached. The quota prevents Control Path items from starving the Data Path: after processing async_dispatch_quota dispatch queue items without interleaving pipeline work, the fiber yields to let IoLoop parse pending socket data, then switches to pipeline processing.

v2: IoLoopV2 (Memcache and optional RESP connections)

IoLoopV2 is an event-driven, single-fiber loop that uses socket->RegisterOnRecv() instead of blocking on recv(). There is no separate AsyncFiber — the dispatch queue is drained inline at the top of each iteration (strict priority over command parsing), using the same AsyncOperations handler as v1. Key differences: no async_dispatch_quota interleaving, no pipeline squashing, and backpressure uses notifyAll() instead of per-publisher notify().

Note: By default, Pub/Sub traffic flows through the v1 AsyncFiber path because --enable_resp_io_loop_v2 defaults to false. When --enable_resp_io_loop_v2 is enabled, non-TLS RESP connections (including Pub/Sub subscribers) use IoLoopV2 instead.

PubMessage Processing (shared by v1 and v2)

Both loops dispatch PubMessage through the same handler: AsyncOperations::operator()(const PubMessage& pub_msg).

operator()(const PubMessage& pub_msg)
  // Discard stale messages after client fully unsubscribed
  if cntx()->subscriptions == 0:
    return  // silently drop

  // Cluster migration: force-unsubscribe the client
  if pub_msg.force_unsubscribe:
    → send PUSH ["sunsubscribe", channel, 0]
    → cntx()->Unsubscribe(channel)
    return

  // Format RESP push message
  if pattern empty:
    type = is_sharded ? "smessage" : "message"
    → send PUSH [type, channel, message]
  else:
    → send PUSH ["pmessage", pattern, channel, message]

Messages are sent as RESP3 Push types (CollectionType::PUSH) via RedisReplyBuilder::SendBulkStrArr.

Backpressure (RESP V1)

This section describes the current publisher back-pressure behavior for Redis/RESP connections using Dragonfly's V1 connection loop (IoLoop plus AsyncFiber). It focuses on a fast PUBLISH/SPUBLISH producer and a subscriber that does not read from its socket. The V2 connection loop is intentionally out of scope.

Dragonfly does not block a publisher on an individual subscriber socket. Instead, published messages are placed on each subscriber connection's control-path dispatch queue, and the queued Pub/Sub memory is accounted per I/O thread in QueueBackpressure::subscriber_bytes. Two thresholds derive from publish_buffer_limit:

• Soft limit — publish_buffer_limit. Crossing it starts a back-pressure episode and enables slow-subscriber protection (see below), but does not park publishers.
• Hard limit — publish_buffer_limit * kPubSubHardMultiplier (hardcoded to 4). Publishers targeting the thread park their fibers only once subscriber_bytes reaches this red line, until queued subscriber messages are processed or the connections are closed.

The hard limit is a throttling point, not a strict cap. A publish checks the budget before dispatching and does not reserve space, so concurrent publishers can add messages after the check and temporarily push the total above the configured value.

IoLoopV1 delivery path

For PUBLISH and the non-cluster SPUBLISH path, the delivery sequence is:

1. ChannelStore::SendMessages finds exact and pattern subscribers and sorts them by their owning I/O thread.
2. Before dispatching, it calls Connection::EnsureMemoryBudget(thread_id) once per destination thread. QueueBackpressure::EnsureBelowLimit parks the publishing fiber only while subscriber_bytes exceeds the hard limit (the equality case is allowed); being above the soft limit alone does not park publishers.
3. BuildSender creates one shared payload containing the channel and message arguments. DispatchBrief runs on each I/O thread and calls Connection::SendPubMessageAsync for local subscribers.
4. SendPubMessageAsync wraps the PubMessage in a MessageHandle and calls SendAsync. The message is appended to the connection's dispatch_q_, the V1 AsyncFiber is started if necessary, and the fiber is notified.
5. AsyncFiber consumes the control path. ProcessAdminMessage invokes the Pub/Sub handler, which formats a RESP push (message, smessage, or pmessage) into the subscriber's reply builder.
6. After the handler returns, the message's queue accounting is removed. The reply builder flushes according to the normal V1 connection-loop rules. A slow or non-reading socket therefore leaves messages in dispatch_q_, keeping the subscriber memory accounted and eventually throttling publishers for that I/O thread.

The budget is shared by all subscriber connections on one I/O thread. A single stuck subscriber can therefore throttle publishers sending to other subscribers assigned to the same thread, but it does not directly consume the budget of other I/O threads.

Slow-subscriber protection

To keep one stuck subscriber from throttling publishers up to the hard limit, Dragonfly can close it once it is clearly the culprit. ProcessAdminMessage records async_op_start_cycle_ around the std::visit that runs the Pub/Sub send, so the connection knows how long its current send has been blocked.

SendPubMessageAsync (which runs on the subscriber's thread while that send is parked) closes the subscriber when its active send has been blocked continuously for at least --pubsub_slow_subscriber_timeout_ms and either of these conditions holds:

1. this connection's queued Pub/Sub data has reached one-sixteenth of publish_buffer_limit, or
2. the thread is in a soft-limit episode (subscriber_bytes above the soft limit).

Thus, a slow connection cannot consume the entire per-thread budget before the policy can evict it, while a full thread can evict any subscriber with a timed-out send. A slow send below the per-connection threshold is left alone until the thread reaches its soft limit, and a full budget with a still-progressing send is also left alone. When the policy triggers, RequestPubsubClose discards the new message, evicts the connection's queued PubMessage items (immediately releasing their subscriber accounting and waking parked publishers), and calls the non-blocking MarkForClose — it never waits, joins, or interrupts the in-flight write. ProcessAdminMessage observes request_shutdown_ once the parked send returns and lets the normal shutdown path close the connection; further pub messages are dropped in the meantime.

--pubsub_slow_subscriber_timeout_ms is startup-only (like publish_buffer_limit) and defaults to 0, which disables the protection.

Limits, cleanup, and observability

Queued Pub/Sub messages are accounted both per connection and in the owning I/O thread's QueueBackpressure::subscriber_bytes. When a message is processed, or when a connection closes and its queue is drained, the bytes are released. Once the thread falls back below the hard limit, V1 wakes waiting publishers. Messages are normally retained until they are processed or the subscriber disconnects; stale messages are dropped after the client leaves Pub/Sub mode.

The main configuration knobs are publish_buffer_limit (196 MB per I/O thread, startup-only) and pubsub_slow_subscriber_timeout_ms (startup-only, 0 disables). proactor_threads changes how subscribers and independent budgets are distributed. Other V1 pipeline and socket settings can affect processing or output latency, but do not change Pub/Sub admission directly.

For diagnosis, use:

• dispatch_queue_subscriber_bytes: queued Pub/Sub bytes;
• pubsub_backpressure_events_total{event=...}: back-pressure and slow-subscriber events by type — soft_limit (soft-limit crossings), hard_limit (publisher throttle episodes), forced_disconnect (subscribers closed by the policy), and messages_discarded;
• send_delay_ms / send_delay_seconds: age of the oldest pending send;
• total_net_output_bytes and total_writes_processed: output progress; and
• cmdstat_publish: indirect evidence of publisher waiting.

Aside from pubsub_backpressure_events_total, these signals are aggregated and do not identify the specific subscriber holding the budget; the policy also emits a rate-limited structured log naming the connection it closes. pipeline_queue_bytes and pipeline_throttle_total can indicate competing data-path pressure. Coverage is in tests/dragonfly/connection_test.py: test_publish_stuck (publishers resume after the stuck subscriber disconnects), test_pubsub_slow_subscriber_closed (the policy force-closes a stuck subscriber), and test_pubsub_slow_subscriber_disabled (a 0 timeout never closes).

Cluster Mode Integration

Standard Pub/Sub — Blocked

In cluster mode, standard PUBLISH, SUBSCRIBE, PSUBSCRIBE, UNSUBSCRIBE, and PUNSUBSCRIBE commands return an error:

(error) PUBLISH is not supported in cluster mode yet

This is enforced in Service::Publish, Service::Subscribe, Service::Unsubscribe, Service::PSubscribe, and Service::PUnsubscribe by checking IsClusterEnabled().

Sharded Pub/Sub — Supported

Sharded Pub/Sub (SPUBLISH, SSUBSCRIBE, SUNSUBSCRIBE) works in cluster mode. The channel name is treated as a key for slot routing purposes:

// In Service::FindKeys:
if (cid->PubSubKind() == CO::PubSubKind::SHARDED) {
  if (cid->name() == "SPUBLISH")
    return KeyIndex(0, 1);      // channel is the key
  return {KeyIndex(0, args.size())};  // all channels are keys
}

This ensures that SPUBLISH and SSUBSCRIBE for the same channel are routed to the same slot, and cluster slot ownership checks apply.

Slot Migration — Forced Unsubscription

When cluster slots migrate away from a node, ChannelStore::UnsubscribeAfterClusterSlotMigration is called:

UnsubscribeAfterClusterSlotMigration(deleted_slots)
  1. Collect: ForEachShared over channels_, collect subscribers for channels
     whose slot is in deleted_slots
  2. Remove: RemoveAllSubscribers(channel) for each matched channel
     → uses Mutate() to erase under exclusive read_mu_
  3. Notify: AwaitFiberOnAll dispatches force-unsubscribe messages
     to affected connections on their owning threads

RemoveAllSubscribers calls map.Mutate(channel, ...) which acquires write_mu_ and then read_mu_ exclusively to erase the channel entry and delete its SubscribeMap.

On each thread, UnsubscribeConnectionsFromDeletedSlots sends PubMessages with force_unsubscribe=true, which triggers sunsubscribe push messages to affected clients.

Keyspace Event Notifications

Dragonfly integrates a limited form of keyspace notifications through the Pub/Sub system. Currently, only expired-key events (Ex) are supported, controlled by the --notify_keyspace_events flag.

When enabled:

1. DbSlice sets expired_keys_events_recording_ = true.
2. As keys expire (via ExpireIfNeeded), their names are appended to db->expired_keys_events_.
3. At the end of DeleteExpiredStep, batched events are published:

channel_store->SendMessages(
    absl::StrCat("__keyevent@", cntx.db_index, "__:expired"),
    events, false);
events.clear();

This uses the standard SendMessages path, so clients subscribed to __keyevent@0__:expired receive notifications through the same dispatch pipeline.

Command Registration

All Pub/Sub commands are registered in Service::Register (src/server/main_service.cc):

Notable flags:

• CO::FAST on PUBLISH/SPUBLISH — these are non-transactional, concurrent reads (shared read_mu_).
• CO::NOSCRIPT on all subscribe/unsubscribe — cannot be called from Lua scripts.
• CO::LOADING — permitted during database loading.
• None of the Pub/Sub commands are transactional (IsTransactional() == false).

Key Files Reference