Pub/Sub Internal

Data Structures

The Pub/Sub system registers channels and patterns on the Redis server and clients.
In short, channels are stored in a dict (hash table) and patterns are stored in a linked list.

First let's look at the dict (hash table) that stores channels.

Server dict (hash table) data structure for channels

The server struct (redisServer struct) has pubsub_channels and pubsub_patterns fields.   The pubsub_channels field points to the dict struct that stores channels.   From the dict struct to dictEntry is the hash table used in Redis.
The key field of dictEntry points to the channel.   Because multiple clients can subscribe to one channel, the value field of dictEntry points to a list (linked list).   The list has multiple listNodes, and each node points to a client.
PUBLISH channel message first finds the channel in the hash table, then sends the message one by one to the clients stored in the list.   Then it sends the message to clients that registered patterns.
The linked list structure that stores patterns is as follows.

Server linked list data structure for patterns

The pubsub_patterns field of the server struct points to the list that stores patterns.   Each node of the list points to a pubsubPattern struct, and that struct holds a client and a pattern.
PUBLISH channel message first sends the message to clients by channel name, then finds patterns that match the channel and sends the message to those clients.

Client dict (hash table) data structure for channels

The client struct (redisClient struct) also has, like the server struct, pubsub_channels and pubsub_patterns fields.   The pubsub_channels field of the client struct also points to a dict struct that stores channels.
From the dict struct to dictEntry it is the same as the server struct, and the key field of dictEntry pointing to the channel is also the same as on the server.   But the client does not have a value.
When SUBSCRIBE is executed, the channel is stored in the client struct, and it is also stored in the server struct.   When PUBLISH is executed, finding the channel and sending the message to clients uses the server struct.
The client struct is used so the client registers the channel, and so the client is switched to pubsub mode.   If the client is in normal mode, the client buffer is unlimited, but in pubsub mode the hard limit is 32mb and the soft limit is 8mb.   When UNSUBSCRIBE or PUNSUBSCRIBE is executed and the client struct has no channels or patterns left, the client is switched from pubsub mode to normal mode.

Client linked list data structure for patterns

The pubsub_patterns field of the client struct points to the list that stores patterns.   Unlike the server, each node of the list points directly to a pattern struct.
When PSUBSCRIBE is executed, the pattern is stored in the client struct list, and it is also stored in the server struct, and the client is switched to pubsub mode.

Now that the data structures have been explained, the next section explains the functions.

FUNCTIONS

Redis Pub/Sub consists of six commands.
SUBSCRIBE and PSUBSCRIBE receive messages, and UNSUBSCRIBE and PUNSUBSCRIBE stop receiving them.   PUBLISH sends messages, and PUBSUB looks up the registered channel list and counts, and the number of patterns.
Let's look at the internals one by one.

SUBSCRIBE

This command registers a channel and receives messages.
When SUBSCRIBE is executed, the channel is registered in the server struct (redisServer struct) and the client struct (redisClient struct).
Some function argument names in this explanation were changed to struct names to make them easier to understand, unlike the original source.   Functions shown in the figure above are in bold.

1. First run dictAdd(client->pubsub_channels,channel,NULL) to register the channel name in the dict of redisClient.pubsub_channels.   The third argument is the value; passing NULL stores NULL as the value.   See Figure 1-2.
2. From here, this is the process of registering on the server struct.
   Run dictFind(server.pubsub_channels,channel); if the channel does not exist, run list = listCreate() to create a list.   If the channel is already registered, run list = dictGetVal(dictEntry) to get the list, and go to step 4.
3. Run dictAdd(server.pubsub_channels,channel,list) to register the channel and list in the dict.
4. Run listAddNodeTail(list,client) to register the client in the list.

It helps to look at this together with Figure 1-1.

PSUBSCRIBE

This command registers a pattern and receives messages.
When PSUBSCRIBE is executed, the pattern is registered in the linked lists of the server and the client.

1. listSearchKey(client->pubsub_patterns,pattern) checks whether the pattern already exists.
   If not, run listAddNodeTail(client->pubsub_patterns,pattern) to register the pattern in the client list.   See Figure 1-4.
2. listAddNodeTail(server.pubsub_patterns,pat) registers the pattern on the server. The client stores only the pattern, while the server stores both client and pattern in the pubsubPattern struct.   That is so the message can be sent to the client that registered the pattern.   See Figure 1-2.

UNSUBSCRIBE

Removes registered channels so messages are no longer received.
If no channel is given as an argument, it calls pubsubUnsubscribeAllChannels() and removes all channels registered on the client.   pubsubUnsubscribeAllChannels() loops with a while loop, calling pubsubUnsubscribeChannel() and using dictNext() to delete channels one by one.
If channels are given, it loops as many times as the number of channels, calling pubsubUnsubscribeChannel() to delete them one by one.

1. Run dictDelete(client->pubsub_channels,channel) to delete the channel from the client.
2. Get the dictEntry from the server with dictEntry = dictFind(server.pubsub_channels,channel).
3. Get the list stored in value with list = dictGetVal(dictEntry).
4. Run listNode = listSearchKey(list,client) to get the list node.
5. Run listDelNode(list,listNode) to delete the node from the list.
6. If the list has no nodes left, run dictDelete(server.pubsub_channels,channel) to delete the list.

PUNSUBSCRIBE

Removes registered patterns so messages are no longer received.
If no pattern is given as an argument, it calls pubsubUnsubscribeAllPatterns() and removes all patterns registered on the client.   pubsubUnsubscribeAllPatterns() loops with a while loop, calling pubsubUnsubscribePattern() and using listDelNode() to delete them one by one.
If channels are given, it loops as many times as the number of channels, calling pubsubUnsubscribePattern() to delete them one by one.

1. First run listSearchKey(client->pubsub_patterns,pattern) to find the pattern on the client.
2. Run listDelNode(client->pubsub_patterns,listNode) to delete the pattern.
3. Then run listSearchKey(server->pubsub_patterns,pubsubPattern) to find the pattern on the server.
4. Run listDelNode(server->pubsub_patterns,listNode) to delete the pattern.

PUBLISH

Of the four commands above, the first two receive messages and the last two stop receiving messages.
PUBLISH is the command that sends a message.
PUBLISH has two main parts.   One sends the message to clients within the server, and the other, if the server is in cluster mode, forwards the message to other servers so each server can send it to the clients connected to it.
In short, Pub/Sub supports cluster mode.

This is the in-server behavior.

1. First find the channel on the server. dictFind(server.pubsub_channels,channel)
2. Loop with listNext(listNode) until there are no more nodes and send the message to registered clients with addReplyBulk(client,message).
3. Next, check with listLength(server.pubsub_patterns); if there are patterns registered on the server
4. Loop with listNext(listNode) until there are no more nodes and compare with stringmatchlen(); if it matches, send the message to the client with addReplyBulk(pubsubPattern->client,message).

This is the process of delivering to cluster nodes.
From clusterPropagatePublish() onward, the code is in cluster.c.
In short, information about cluster servers is stored in a dict on the server, and dictNext() is used to get server info and send the channel and message.   Then that server receives the channel and message, goes through the same process, and sends the message to its clients.

PUBSUB

• If the subcommand is channels, it walks server.pubsub_channels with dictNext() and shows the channels. If a pattern was given, it compares with stringmatchlen() and shows only matching channels.
• If the subcommand is numsub, it gets the number of clients registered on that channel with listLength(list) and shows it.
• If the subcommand is numpat, it gets the number of patterns registered in the list with listLength(list) and shows it.