Resolving read race condition between pubsub's get_message() and execute_command()

.github/workflows/install_and_test.sh

@@ -38,4 +38,8 @@ cd ${TESTDIR}
 
 # install, run tests
 pip install ${PKG}
-pytest
+# Redis tests
+pytest -m 'not onlycluster'
+# RedisCluster tests
+CLUSTER_URL="redis://localhost:16379/0"
+pytest -m 'not onlynoncluster and not redismod' --redis-url=${CLUSTER_URL}

CONTRIBUTING.md

@@ -68,6 +68,14 @@ configuration](https://redis.io/topics/sentinel).
 
 ## Testing
 
+Call `invoke tests` to run all tests, or `invoke all-tests` to run linters 
+tests as well. With the 'tests' and 'all-tests' targets, all Redis and 
+RedisCluster tests will be run. 
+
+It is possible to run only Redis client tests (with cluster mode disabled) by 
+using `invoke redis-tests`; similarly, RedisCluster tests can be run by using 
+`invoke cluster-tests`.
+
 Each run of tox starts and stops the various dockers required. Sometimes
 things get stuck, an `invoke clean` can help.
 

README.md

@@ -948,8 +948,272 @@ C 3
 
 ### Cluster Mode
 
-redis-py does not currently support [Cluster
-Mode](https://redis.io/topics/cluster-tutorial).
+redis-py is now supports cluster mode and provides a client for
+[Redis Cluster](<https://redis.io/topics/cluster-tutorial>).
+
+The cluster client is based on Grokzen's 
+[redis-py-cluster](https://github.com/Grokzen/redis-py-cluster), has added bug 
+fixes, and now supersedes that library. Support for these changes is thanks to 
+his contributions.
+
+
+**Create RedisCluster:**
+
+Connecting redis-py to a Redis Cluster instance(s) requires at a minimum a 
+single node for cluster discovery. There are multiple ways in which a cluster 
+instance can be created:
+
+- Using 'host' and 'port' arguments:
+
+``` pycon
+    >>> from redis.cluster import RedisCluster as Redis
+    >>> rc = Redis(host='localhost', port=6379)
+    >>> print(rc.get_nodes())
+    [[host=127.0.0.1,port=6379,name=127.0.0.1:6379,server_type=primary,redis_connection=Redis<ConnectionPool<Connection<host=127.0.0.1,port=6379,db=0>>>], [host=127.0.0.1,port=6378,name=127.0.0.1:6378,server_type=primary,redis_connection=Redis<ConnectionPool<Connection<host=127.0.0.1,port=6378,db=0>>>], [host=127.0.0.1,port=6377,name=127.0.0.1:6377,server_type=replica,redis_connection=Redis<ConnectionPool<Connection<host=127.0.0.1,port=6377,db=0>>>]]
+```
+- Using the Redis URL specification:
+
+``` pycon
+    >>> from redis.cluster import RedisCluster as Redis
+    >>> rc = Redis.from_url("redis://localhost:6379/0")
+```
+
+- Directly, via the ClusterNode class:
+
+``` pycon
+    >>> from redis.cluster import RedisCluster as Redis
+    >>> from redis.cluster import ClusterNode
+    >>> nodes = [ClusterNode('localhost', 6379), ClusterNode('localhost', 6378)]
+    >>> rc = Redis(startup_nodes=nodes)
+```
+
+When a RedisCluster instance is being created it first attempts to establish a
+connection to one of the provided startup nodes. If none of the startup nodes
+are reachable, a 'RedisClusterException' will be thrown.
+After a connection to the one of the cluster's nodes is established, the
+RedisCluster instance will be initialized with 3 caches:
+a slots cache which maps each of the 16384 slots to the node/s handling them,
+a nodes cache that contains ClusterNode objects (name, host, port, redis connection)
+for all of the cluster's nodes, and a commands cache contains all the server
+supported commands that were retrieved using the Redis 'COMMAND' output.
+
+RedisCluster instance can be directly used to execute Redis commands. When a
+command is being executed through the cluster instance, the target node(s) will
+be internally determined. When using a key-based command, the target node will
+be the node that holds the key's slot.
+Cluster management commands and other commands that are not key-based have a 
+parameter called 'target_nodes' where you can specify which nodes to execute 
+the command on. In the absence of target_nodes, the command will be executed 
+on the default cluster node. As part of cluster instance initialization, the 
+cluster's default node is randomly selected from the cluster's primaries, and 
+will be updated upon reinitialization. Using r.get_default_node(), you can 
+get the cluster's default node, or you can change it using the 
+'set_default_node' method.
+
+The 'target_nodes' parameter is explained in the following section, 
+'Specifying Target Nodes'.
+
+``` pycon
+    >>> # target-nodes: the node that holds 'foo1's key slot
+    >>> rc.set('foo1', 'bar1')
+    >>> # target-nodes: the node that holds 'foo2's key slot
+    >>> rc.set('foo2', 'bar2')
+    >>> # target-nodes: the node that holds 'foo1's key slot
+    >>> print(rc.get('foo1'))
+    b'bar'
+    >>> # target-node: default-node
+    >>> print(rc.keys())
+    [b'foo1']
+    >>> # target-node: default-node
+    >>> rc.ping()
+```
+
+**Specifying Target Nodes:**
+
+As mentioned above, all non key-based RedisCluster commands accept the kwarg 
+parameter 'target_nodes' that specifies the node/nodes that the command should 
+be executed on.
+The best practice is to specify target nodes using RedisCluster class's node
+flags: PRIMARIES, REPLICAS, ALL_NODES, RANDOM. When a nodes flag is passed
+along with a command, it will be internally resolved to the relevant node/s.
+If the nodes topology of the cluster changes during the execution of a command,
+the client will be able to resolve the nodes flag again with the new topology
+and attempt to retry executing the command.
+
+``` pycon
+    >>> from redis.cluster import RedisCluster as Redis
+    >>> # run cluster-meet command on all of the cluster's nodes
+    >>> rc.cluster_meet('127.0.0.1', 6379, target_nodes=Redis.ALL_NODES)
+    >>> # ping all replicas
+    >>> rc.ping(target_nodes=Redis.REPLICAS)
+    >>> # ping a specific node
+    >>> rc.ping(target_nodes=Redis.RANDOM)
+    >>> # get the keys from all cluster nodes
+    >>> rc.keys(target_nodes=Redis.ALL_NODES)
+    [b'foo1', b'foo2']
+    >>> # execute bgsave in all primaries
+    >>> rc.bgsave(Redis.PRIMARIES)
+```
+
+You could also pass ClusterNodes directly if you want to execute a command on a
+specific node / node group that isn't addressed by the nodes flag. However, if
+the command execution fails due to cluster topology changes, a retry attempt
+will not be made, since the passed target node/s may no longer be valid, and
+the relevant cluster or connection error will be returned.
+
+``` pycon
+    >>> node = rc.get_node('localhost', 6379)
+    >>> # Get the keys only for that specific node
+    >>> rc.keys(target_nodes=node)
+    >>> # get Redis info from a subset of primaries
+    >>> subset_primaries = [node for node in rc.get_primaries() if node.port > 6378]
+    >>> rc.info(target_nodes=subset_primaries)
+```
+
+In addition, the RedisCluster instance can query the Redis instance of a 
+specific node and execute commands on that node directly. The Redis client,
+however, does not handle cluster failures and retries.
+
+``` pycon
+    >>> cluster_node = rc.get_node(host='localhost', port=6379)
+    >>> print(cluster_node)
+    [host=127.0.0.1,port=6379,name=127.0.0.1:6379,server_type=primary,redis_connection=Redis<ConnectionPool<Connection<host=127.0.0.1,port=6379,db=0>>>]
+    >>> r = cluster_node.redis_connection
+    >>> r.client_list()
+    [{'id': '276', 'addr': '127.0.0.1:64108', 'fd': '16', 'name': '', 'age': '0', 'idle': '0', 'flags': 'N', 'db': '0', 'sub': '0', 'psub': '0', 'multi': '-1', 'qbuf': '26', 'qbuf-free': '32742', 'argv-mem': '10', 'obl': '0', 'oll': '0', 'omem': '0', 'tot-mem': '54298', 'events': 'r', 'cmd': 'client', 'user': 'default'}]
+    >>> # Get the keys only for that specific node
+    >>> r.keys()
+    [b'foo1']
+```
+
+**Multi-key commands:**
+
+Redis supports multi-key commands in Cluster Mode, such as Set type unions or
+intersections, mset and mget, as long as the keys all hash to the same slot.
+By using RedisCluster client, you can use the known functions (e.g. mget, mset)
+to perform an atomic multi-key operation. However, you must ensure all keys are
+mapped to the same slot, otherwise a RedisClusterException will be thrown.
+Redis Cluster implements a concept called hash tags that can be used in order
+to force certain keys to be stored in the same hash slot, see 
+[Keys hash tag](https://redis.io/topics/cluster-spec#keys-hash-tags).
+You can also use nonatomic for some of the multikey operations, and pass keys
+that aren't mapped to the same slot. The client will then map the keys to the
+relevant slots, sending the commands to the slots' node owners. Non-atomic
+operations batch the keys according to their hash value, and then each batch is
+sent separately to the slot's owner.
+
+``` pycon
+    #  Atomic operations can be used when all keys are mapped to the same slot
+    >>> rc.mset({'{foo}1': 'bar1', '{foo}2': 'bar2'})
+    >>> rc.mget('{foo}1', '{foo}2')
+    [b'bar1', b'bar2']
+    # Non-atomic multi-key operations splits the keys into different slots
+    >>> rc.mset_nonatomic({'foo': 'value1', 'bar': 'value2', 'zzz': 'value3')
+    >>> rc.mget_nonatomic('foo', 'bar', 'zzz')
+    [b'value1', b'value2', b'value3']
+```
+
+**Cluster PubSub:**
+
+When a ClusterPubSub instance is created without specifying a node, a single
+node will be transparently chosen for the pubsub connection on the
+first command execution. The node will be determined by:
+ 1. Hashing the channel name in the request to find its keyslot
+ 2. Selecting a node that handles the keyslot: If read_from_replicas is
+    set to true, a replica can be selected.
+
+*Known limitations with pubsub:*
+
+Pattern subscribe and publish do not currently work properly due to key slots. 
+If we hash a pattern like fo* we will receive a keyslot for that string but 
+there are endless possibilities for channel names based on this pattern - 
+unknowable in advance. This feature is not disabled but the commands are not 
+currently recommended for use.
+See [redis-py-cluster documentation](https://redis-py-cluster.readthedocs.io/en/stable/pubsub.html) 
+ for more.
+
+``` pycon
+    >>> p1 = rc.pubsub()
+    # p1 connection will be set to the node that holds 'foo' keyslot
+    >>> p1.subscribe('foo')
+    # p2 connection will be set to node 'localhost:6379'
+    >>> p2 = rc.pubsub(rc.get_node('localhost', 6379))
+```
+
+**Read Only Mode**
+
+By default, Redis Cluster always returns MOVE redirection response on accessing 
+a replica node. You can overcome this limitation and scale read commands by 
+triggering READONLY mode.
+
+To enable READONLY mode pass read_from_replicas=True to RedisCluster 
+constructor. When set to true, read commands will be assigned between the
+primary and its replications in a Round-Robin manner. 
+
+READONLY mode can be set at runtime by calling the readonly() method with 
+target_nodes='replicas', and read-write access can be restored by calling the 
+readwrite() method.
+
+``` pycon
+    >>> from cluster import RedisCluster as Redis
+    # Use 'debug' log level to print the node that the command is executed on
+    >>> rc_readonly = Redis(startup_nodes=startup_nodes, 
+                    read_from_replicas=True, debug=True)
+    >>> rc_readonly.set('{foo}1', 'bar1')
+    >>> for i in range(0, 4):
+            # Assigns read command to the slot's hosts in a Round-Robin manner
+    >>>     rc_readonly.get('{foo}1')
+    # set command would be directed only to the slot's primary node
+    >>> rc_readonly.set('{foo}2', 'bar2')
+    # reset READONLY flag
+    >>> rc_readonly.readwrite(target_nodes='replicas')
+    # now the get command would be directed only to the slot's primary node
+    >>> rc_readonly.get('{foo}1')
+```
+
+**Cluster Pipeline**
+
+ClusterPipeline is a subclass of RedisCluster that provides support for Redis 
+pipelines in cluster mode. 
+When calling the execute() command, all the commands are grouped by the node 
+on which they will be executed, and are then executed by the respective nodes 
+in parallel. The pipeline instance will wait for all the nodes to respond 
+before returning the result to the caller. Command responses are returned as a 
+list sorted in the same order in which they were sent.
+Pipelines can be used to dramatically increase the throughput of Redis Cluster 
+by significantly reducing the the number of network round trips between the 
+client and the server.
+
+``` pycon
+    >>> with rc.pipeline() as pipe:
+    >>>     pipe.set('foo', 'value1')
+    >>>     pipe.set('bar', 'value2')
+    >>>     pipe.get('foo')
+    >>>     pipe.get('bar')
+    >>>     print(pipe.execute())
+    [True, True, b'value1', b'value2']
+    >>>     pipe.set('foo1', 'bar1').get('foo1').execute()
+    [True, b'bar1']
+```
+Please note:
+- RedisCluster pipelines currently only support key-based commands.
+- The pipeline gets its 'read_from_replicas' value from the cluster's parameter.
+Thus, if read from replications is enabled in the cluster instance, the pipeline 
+will also direct read commands to replicas.
+- The 'transcation' option is NOT supported in cluster-mode. In non-cluster mode, 
+the 'transaction' option is available when executing pipelines. This wraps the 
+pipeline commands with MULTI/EXEC commands, and effectively turns the pipeline 
+commands into a single transaction block. This means that all commands are 
+executed sequentially without any interruptions from other clients. However, 
+in cluster-mode this is not possible, because commands are partitioned 
+according to their respective destination nodes. This means that we can not 
+turn the pipeline commands into one transaction block, because in most cases 
+they are split up into several smaller pipelines.
+
+
+See [Redis Cluster tutorial](https://redis.io/topics/cluster-tutorial) and
+[Redis Cluster specifications](https://redis.io/topics/cluster-spec)
+to learn more about Redis Cluster.
 
 ### Author
 

docker/base/Dockerfile.cluster

@@ -0,0 +1,8 @@
+FROM redis:6.2.6-buster
+
+COPY create_cluster.sh /create_cluster.sh
+RUN chmod +x /create_cluster.sh
+
+EXPOSE 16379 16380 16381 16382 16383 16384
+
+CMD [ "/create_cluster.sh"]

docker/base/create_cluster.sh

@@ -0,0 +1,26 @@
+#! /bin/bash
+mkdir -p /nodes
+touch /nodes/nodemap
+for PORT in $(seq 16379 16384); do
+  mkdir -p /nodes/$PORT
+  if [[ -e /redis.conf ]]; then
+    cp /redis.conf /nodes/$PORT/redis.conf
+  else
+    touch /nodes/$PORT/redis.conf
+  fi
+  cat << EOF >> /nodes/$PORT/redis.conf
+port ${PORT}
+cluster-enabled yes
+daemonize yes
+logfile /redis.log
+dir /nodes/$PORT
+EOF
+  redis-server /nodes/$PORT/redis.conf
+  if [ $? -ne 0 ]; then
+    echo "Redis failed to start, exiting."
+    exit 3
+  fi
+  echo 127.0.0.1:$PORT >> /nodes/nodemap
+done
+echo yes | redis-cli --cluster create $(seq -f 127.0.0.1:%g 16379 16384) --cluster-replicas 1
+tail -f /redis.log

docker/cluster/redis.conf

@@ -0,0 +1,3 @@
+# Redis Cluster config file will be shared across all nodes.
+# Do not change the following configurations that are already set:
+# port, cluster-enabled, daemonize, logfile, dir

redis/__init__.py

@@ -1,4 +1,5 @@
 from redis.client import Redis, StrictRedis
+from redis.cluster import RedisCluster
 from redis.connection import (
     BlockingConnectionPool,
     ConnectionPool,
@@ -57,6 +58,7 @@ def int_or_str(value):
     'PubSubError',
     'ReadOnlyError',
     'Redis',
+    'RedisCluster',
     'RedisError',
     'ResponseError',
     'Sentinel',