Multi exec on cluster

.github/wordlist.txt

@@ -2,6 +2,7 @@ APM
 ARGV
 BFCommands
 CacheImpl
+CAS
 CFCommands
 CMSCommands
 ClusterNode

.gitignore

@@ -9,6 +9,7 @@ vagrant/.vagrant
 .cache
 .eggs
 .idea
+.vscode
 .coverage
 env
 venv

CHANGES

@@ -1,3 +1,4 @@
+ * Support transactions in ClusterPipeline
  * Removing support for RedisGraph module. RedisGraph support is deprecated since Redis Stack 7.2 (https://redis.com/blog/redisgraph-eol/)
  * Fix lock.extend() typedef to accept float TTL extension
  * Update URL in the readme linking to Redis University

docs/advanced_features.rst

@@ -177,20 +177,110 @@ the server.
  ... 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 ‘transaction’ 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.
+Please note:
+
+- RedisCluster pipelines currently only support key-based commands.
+- The pipeline gets its ‘load_balancing_strategy’ 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.
+
+
+Transactions in clusters
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+Transactions are supported in cluster-mode with one caveat: all keys of
+all commands issued on a transaction pipeline must reside on the
+same slot. This is similar to the limitation of multikey commands in
+cluster. The reason behind this is that the Redis engine does not offer
+a mechanism to block or exchange key data across nodes on the fly. A
+client may add some logic to abstract engine limitations when running
+on a cluster, such as the pipeline behavior explained on the previous
+block, but there is no simple way that a client can enforce atomicity
+across nodes on a distributed system.
+
+The compromise of limiting the transaction pipeline to same-slot keys
+is exactly that: a compromise. While this behavior is different from
+non-transactional cluster pipelines, it simplifies migration of clients
+from standalone to cluster under some circumstances. Note that application
+code that issues multi/exec commands on a standalone client without
+embedding them within a pipeline would eventually get ‘AttributeError’.
+With this approach, if the application uses ‘client.pipeline(transaction=True)’,
+then switching the client with a cluster-aware instance would simplify
+code changes (to some extent). This may be true for application code that
+makes use of hash keys, since its transactions may are already be
+mapping all commands to the same slot.
+
+An alternative is some kind of two-step commit solution, where a slot
+validation is run before the actual commands are run. This could work
+with controlled node maintenance but does not cover single node failures.
+
+Given the cluster limitations for transactions, by default pipeline isn't in
+transactional mode. To enable transactional context set:
+
+.. code:: python
+
+ >>> p = r.pipeline(transaction=True)
+
+After entering the transactional context you can add commands to a transactional
+context, by one of the following ways:
+
+.. code:: python
+
+ >>> p = r.pipeline(transaction=True) # Chaining commands
+ >>> p.set("key", "value")
+ >>> p.get("key")
+ >>> response = p.execute()
+
+Or
+
+.. code:: python
+
+ >>> with r.pipeline(transaction=True) as pipe: # Using context manager
+ ... pipe.set("key", "value")
+ ... pipe.get("key")
+ ... response = pipe.execute()
+
+As you see there's no need to explicitly send `MULTI/EXEC` commands to control context start/end
+`ClusterPipeline` will take care of it.
+
+To ensure that different keys will be mapped to a same hash slot on the server side
+prepend your keys with the same hash tag, the technique that allows you to control
+keys distribution.
+More information `here <https://redis.io/docs/latest/operate/oss_and_stack/reference/cluster-spec/#hash-tags>`_
+
+.. code:: python
+
+ >>> with r.pipeline(transaction=True) as pipe:
+ ... pipe.set("{tag}foo", "bar")
+ ... pipe.set("{tag}bar", "foo")
+ ... pipe.get("{tag}foo")
+ ... pipe.get("{tag}bar")
+ ... response = pipe.execute()
+
+CAS Transactions
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+If you want to apply optimistic locking for certain keys, you have to execute
+`WATCH` command in transactional context. `WATCH` command follows the same limitations
+as any other multi key command - all keys should be mapped to the same hash slot.
+
+However, the difference between CAS transaction and normal one is that you have to
+explicitly call MULTI command to indicate the start of transactional context, WATCH
+command itself and any subsequent commands before MULTI will be immediately executed
+on the server side so you can apply optimistic locking and get necessary data before
+transaction execution.
+
+.. code:: python
+
+ >>> with r.pipeline(transaction=True) as pipe:
+ ... pipe.watch("mykey") # Apply locking by immediately executing command
+ ... val = pipe.get("mykey") # Immediately retrieves value
+ ... val = val + 1 # Increment value
+ ... pipe.multi() # Starting transaction context
+ ... pipe.set("mykey", val) # Command will be pipelined
+ ... response = pipe.execute() # Returns OK or None if key was modified in the meantime
+
 
 Publish / Subscribe
 -------------------

redis/__init__.py

@@ -16,11 +16,14 @@
  BusyLoadingError,
  ChildDeadlockedError,
  ConnectionError,
+ CrossSlotTransactionError,
  DataError,
+ InvalidPipelineStack,
  InvalidResponse,
  OutOfMemoryError,
  PubSubError,
  ReadOnlyError,
+ RedisClusterException,
  RedisError,
  ResponseError,
  TimeoutError,
@@ -56,15 +59,18 @@ def int_or_str(value):
  "ConnectionError",
  "ConnectionPool",
  "CredentialProvider",
+ "CrossSlotTransactionError",
  "DataError",
  "from_url",
  "default_backoff",
+ "InvalidPipelineStack",
  "InvalidResponse",
  "OutOfMemoryError",
  "PubSubError",
  "ReadOnlyError",
  "Redis",
  "RedisCluster",
+ "RedisClusterException",
  "RedisError",
  "ResponseError",
  "Sentinel",

redis/client.py

@@ -34,6 +34,7 @@
 from redis.commands.core import Script
 from redis.connection import (
  AbstractConnection,
+ Connection,
  ConnectionPool,
  SSLConnection,
  UnixDomainSocketConnection,
@@ -1297,9 +1298,15 @@ class Pipeline(Redis):
 
  UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"}
 
- def __init__(self, connection_pool, response_callbacks, transaction, shard_hint):
+ def __init__(
+ self,
+ connection_pool: ConnectionPool,
+ response_callbacks,
+ transaction,
+ shard_hint,
+ ):
  self.connection_pool = connection_pool
- self.connection = None
+ self.connection: Optional[Connection] = None
  self.response_callbacks = response_callbacks
  self.transaction = transaction
  self.shard_hint = shard_hint
@@ -1434,7 +1441,9 @@ def pipeline_execute_command(self, *args, **options) -> "Pipeline":
  self.command_stack.append((args, options))
  return self
 
- def _execute_transaction(self, connection, commands, raise_on_error) -> List:
+ def _execute_transaction(
+ self, connection: Connection, commands, raise_on_error
+ ) -> List:
  cmds = chain([(("MULTI",), {})], commands, [(("EXEC",), {})])
  all_cmds = connection.pack_commands(
  [args for args, options in cmds if EMPTY_RESPONSE not in options]