From 14989a64f5c911d8acd778f921443b21c221f960 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Thu, 14 May 2026 13:51:08 -0300 Subject: [PATCH 01/22] Micro-optimization A bunch of low-hanging, low-impact micro-optimizations. Mostly common subexpression extraction to local variables when beneficial. Barely noticeable benchmark impact but technically an improvement without readability impact IMO. --- benchmarks/command_packer_benchmark.py | 2 + redis/_parsers/hiredis.py | 22 ++++---- redis/cluster.py | 71 ++++++++++++++------------ redis/commands/policies.py | 21 +++++--- 4 files changed, 65 insertions(+), 51 deletions(-) diff --git a/benchmarks/command_packer_benchmark.py b/benchmarks/command_packer_benchmark.py index 4fb7196422..4a2cf9d4e5 100644 --- a/benchmarks/command_packer_benchmark.py +++ b/benchmarks/command_packer_benchmark.py @@ -9,6 +9,8 @@ def send_packed_command(self, command, check_health=True): if not self._sock: self.connect() try: + if isinstance(command, list): + command = SYM_EMPTY.join(command) self._sock.sendall(command) except OSError as e: self.disconnect() diff --git a/redis/_parsers/hiredis.py b/redis/_parsers/hiredis.py index fb2ea7709c..3fdcfe2d4d 100644 --- a/redis/_parsers/hiredis.py +++ b/redis/_parsers/hiredis.py @@ -108,7 +108,7 @@ def read_from_socket(self, timeout=SENTINEL, raise_on_timeout=True): try: if custom_timeout: sock.settimeout(timeout) - bufflen = self._sock.recv_into(self._buffer) + bufflen = sock.recv_into(self._buffer) if bufflen == 0: raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) self._reader.feed(self._buffer, 0, bufflen) @@ -138,7 +138,8 @@ def read_response( push_request=False, timeout: Union[float, object] = SENTINEL, ): - if not self._reader: + reader = self._reader + if not reader: raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) # _next_response might be cached from a can_read() call @@ -162,16 +163,16 @@ def read_response( return response if disable_decoding: - response = self._reader.gets(False) + response = reader.gets(False) else: - response = self._reader.gets() + response = reader.gets() while response is NOT_ENOUGH_DATA: self.read_from_socket(timeout=timeout) if disable_decoding: - response = self._reader.gets(False) + response = reader.gets(False) else: - response = self._reader.gets() + response = reader.gets() # if the response is a ConnectionError or the response is a list and # the first item is a ConnectionError, raise it as something bad # happened @@ -272,17 +273,18 @@ async def read_response( if not self._connected: raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None + reader = self._reader if disable_decoding: - response = self._reader.gets(False) + response = reader.gets(False) else: - response = self._reader.gets() + response = reader.gets() while response is NOT_ENOUGH_DATA: await self.read_from_socket() if disable_decoding: - response = self._reader.gets(False) + response = reader.gets(False) else: - response = self._reader.gets() + response = reader.gets() # if the response is a ConnectionError or the response is a list and # the first item is a ConnectionError, raise it as something bad diff --git a/redis/cluster.py b/redis/cluster.py index 4e5088e8dd..0aa23c5f6b 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -4044,14 +4044,23 @@ def _send_cluster_commands( nodes: dict[str, NodeCommands] = {} nodes_written = 0 nodes_read = 0 + pipe = self._pipe + + # commonly used policies for reuse + default_keyless = CommandPolicies() + default_keyed = CommandPolicies( + request_policy=RequestPolicy.DEFAULT_KEYED, + response_policy=ResponsePolicy.DEFAULT_KEYED, + ) try: # as we move through each command that still needs to be processed, # we figure out the slot number that command maps to, then from # the slot determine the node. for c in attempt: - command_policies = self._pipe._policy_resolver.resolve( - c.args[0].lower() + args = c.args + command_policies = pipe._policy_resolver.resolve( + args[0].lower() ) # refer to our internal node -> slot table that # tells us where a given command should route to. @@ -4062,60 +4071,56 @@ def _send_cluster_commands( target_nodes = self._parse_target_nodes(passed_targets) if not command_policies: - command_policies = CommandPolicies() + command_policies = default_keyless else: if not command_policies: - command = c.args[0].upper() - if ( - len(c.args) >= 2 - and f"{c.args[0]} {c.args[1]}".upper() - in self._pipe.command_flags - ): - command = f"{c.args[0]} {c.args[1]}".upper() + if len(args) >= 2: + command = f"{args[0]} {args[1]}".upper() + if command not in pipe.command_flags: + command = args[0].upper() + else: + command = args[0].upper() # We only could resolve key properties if command is not # in a list of pre-defined request policies command_flag = self.command_flags.get(command) if not command_flag: # Fallback to default policy - if not self._pipe.get_default_node(): + if not pipe.get_default_node(): keys = None else: - keys = self._pipe._get_command_keys(*c.args) + keys = pipe._get_command_keys(*args) if not keys or len(keys) == 0: - command_policies = CommandPolicies() + command_policies = default_keyless else: - command_policies = CommandPolicies( - request_policy=RequestPolicy.DEFAULT_KEYED, - response_policy=ResponsePolicy.DEFAULT_KEYED, - ) + command_policies = default_keyed else: - if command_flag in self._pipe._command_flags_mapping: + if command_flag in pipe._command_flags_mapping: command_policies = CommandPolicies( - request_policy=self._pipe._command_flags_mapping[ + request_policy=pipe._command_flags_mapping[ command_flag ] ) else: - command_policies = CommandPolicies() + command_policies = default_keyless target_nodes = self._determine_nodes( - *c.args, + *args, request_policy=command_policies.request_policy, node_flag=passed_targets, ) if not target_nodes: raise RedisClusterException( - f"No targets were found to execute {c.args} command on" + f"No targets were found to execute {args} command on" ) c.command_policies = command_policies if len(target_nodes) > 1: raise RedisClusterException( - f"Too many targets for command {c.args}" + f"Too many targets for command {args}" ) node = target_nodes[0] - if node == self._pipe.get_default_node(): + if node == pipe.get_default_node(): is_default_node = True # now that we know the name of the node @@ -4123,7 +4128,7 @@ def _send_cluster_commands( # we can build a list of commands for each node. node_name = node.name if node_name not in nodes: - redis_node = self._pipe.get_redis_connection(node) + redis_node = pipe.get_redis_connection(node) try: connection = get_connection(redis_node) except (ConnectionError, TimeoutError): @@ -4134,7 +4139,7 @@ def _send_cluster_commands( # Retry object. Reinitialize the node -> slot table. self._nodes_manager.initialize() if is_default_node: - self._pipe.replace_default_node() + pipe.replace_default_node() nodes = {} raise nodes[node_name] = NodeCommands( @@ -4232,16 +4237,16 @@ def _send_cluster_commands( # If a lot of commands have failed, we'll be setting the # flag to rebuild the slots table from scratch. # So MOVED errors should correct themselves fairly quickly. - self._pipe.reinitialize_counter += 1 - if self._pipe._should_reinitialized(): + pipe.reinitialize_counter += 1 + if pipe._should_reinitialized(): self._nodes_manager.initialize() if is_default_node: - self._pipe.replace_default_node() + pipe.replace_default_node() for c in attempt: try: # send each command individually like we # do in the main client. - c.result = self._pipe.parent_execute_command(*c.args, **c.options) + c.result = pipe.parent_execute_command(*c.args, **c.options) except RedisError as e: c.result = e @@ -4249,13 +4254,13 @@ def _send_cluster_commands( # to the sequence of commands issued in the stack in pipeline.execute() response = [] for c in sorted(stack, key=lambda x: x.position): - if c.args[0] in self._pipe.cluster_response_callbacks: + if c.args[0] in pipe.cluster_response_callbacks: # Remove keys entry, it needs only for cache. c.options.pop("keys", None) - c.result = self._pipe._policies_callback_mapping[ + c.result = pipe._policies_callback_mapping[ c.command_policies.response_policy ]( - self._pipe.cluster_response_callbacks[c.args[0]]( + pipe.cluster_response_callbacks[c.args[0]]( c.result, **c.options ) ) diff --git a/redis/commands/policies.py b/redis/commands/policies.py index c0c98d37f1..4a4309a39b 100644 --- a/redis/commands/policies.py +++ b/redis/commands/policies.py @@ -189,26 +189,29 @@ def __init__( self._fallback = fallback def resolve(self, command_name: str) -> Optional[CommandPolicies]: - parts = command_name.split(".") + parts = command_name.split(".", 1) if len(parts) > 2: raise ValueError(f"Wrong command or module name: {command_name}") module, command = parts if len(parts) == 2 else ("core", parts[0]) - if self._policies.get(module, None) is None: + module_policies = self._policies.get(module, None) + + if module_policies is None: if self._fallback is not None: return self._fallback.resolve(command_name) else: return None - if self._policies.get(module).get(command, None) is None: + command_policy = module_policies.get(command, None) + if command_policy is None: if self._fallback is not None: return self._fallback.resolve(command_name) else: return None - return self._policies.get(module).get(command) + return command_policy @abstractmethod def with_fallback(self, fallback: "PolicyResolver") -> "PolicyResolver": @@ -227,26 +230,28 @@ def __init__( self._fallback = fallback async def resolve(self, command_name: str) -> Optional[CommandPolicies]: - parts = command_name.split(".") + parts = command_name.split(".", 1) if len(parts) > 2: raise ValueError(f"Wrong command or module name: {command_name}") module, command = parts if len(parts) == 2 else ("core", parts[0]) - if self._policies.get(module, None) is None: + module_policies = self._policies.get(module, None) + if module_policies is None: if self._fallback is not None: return await self._fallback.resolve(command_name) else: return None - if self._policies.get(module).get(command, None) is None: + command_policy = module_policies.get(command, None) + if command_policy is None: if self._fallback is not None: return await self._fallback.resolve(command_name) else: return None - return self._policies.get(module).get(command) + return command_policy @abstractmethod def with_fallback(self, fallback: "AsyncPolicyResolver") -> "AsyncPolicyResolver": From 797a3a8a174b45170d0f4a109e128ebd59d374e9 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Fri, 15 May 2026 00:28:06 -0300 Subject: [PATCH 02/22] More micro-optimizations Concentrate around key extraction and node determination. Optimizes the most common path for key extraction which is that of single-key commands. --- redis/_parsers/commands.py | 97 ++++++++++++++++++++---------- redis/cluster.py | 120 ++++++++++++++++++------------------- 2 files changed, 127 insertions(+), 90 deletions(-) diff --git a/redis/_parsers/commands.py b/redis/_parsers/commands.py index 0d9dc40bd1..450c201b87 100644 --- a/redis/_parsers/commands.py +++ b/redis/_parsers/commands.py @@ -114,6 +114,18 @@ def initialize(self, r): for cmd in uppercase_commands: commands[cmd.lower()] = commands.pop(cmd) self.commands = commands + for command in self.commands.values(): + first_key_pos = command["first_key_pos"] + last_key_pos = command["last_key_pos"] + flags = command["flags"] + if ( + first_key_pos > 0 + and first_key_pos == last_key_pos + and "movablekeys" not in flags + and "pubsub" not in flags + and command["name"] != "pubsub" + ): + command["_single_key_pos"] = first_key_pos # As soon as this PR is merged into Redis, we should reimplement # our logic to use COMMAND INFO changes to determine the key positions @@ -134,33 +146,44 @@ def get_keys(self, redis_conn, *args): return None cmd_name = args[0].lower() - if cmd_name not in self.commands: + commands = self.commands + command = commands.get(cmd_name) + if command is None: # try to split the command name and to take only the main command, # e.g. 'memory' for 'memory usage' cmd_name_split = cmd_name.split() cmd_name = cmd_name_split[0] - if cmd_name in self.commands: + if cmd_name in commands: # save the split command to args args = cmd_name_split + list(args[1:]) else: # We'll try to reinitialize the commands cache, if the engine # version has changed, the commands may not be current self.initialize(redis_conn) - if cmd_name not in self.commands: + commands = self.commands + if cmd_name not in commands: raise RedisError( f"{cmd_name.upper()} command doesn't exist in Redis commands" ) + command = commands.get(cmd_name) + + single_pos = command.get("_single_key_pos") + if single_pos is not None: + return [args[single_pos]] - command = self.commands.get(cmd_name) - if "movablekeys" in command["flags"]: + flags = command["flags"] + if "movablekeys" in flags: keys = self._get_moveable_keys(redis_conn, *args) - elif "pubsub" in command["flags"] or command["name"] == "pubsub": + elif "pubsub" in flags or command["name"] == "pubsub": keys = self._get_pubsub_keys(*args) else: + step_count = command["step_count"] + first_key_pos = command["first_key_pos"] + last_key_pos = command["last_key_pos"] if ( - command["step_count"] == 0 - and command["first_key_pos"] == 0 - and command["last_key_pos"] == 0 + step_count == 0 + and first_key_pos == 0 + and last_key_pos == 0 ): is_subcmd = False if "subcommands" in command: @@ -169,19 +192,22 @@ def get_keys(self, redis_conn, *args): if str_if_bytes(subcmd[0]) == subcmd_name: command = self.parse_subcommand(subcmd) - if command["first_key_pos"] > 0: + step_count = command["step_count"] + first_key_pos = command["first_key_pos"] + last_key_pos = command["last_key_pos"] + + if first_key_pos > 0: is_subcmd = True # The command doesn't have keys in it if not is_subcmd: return None - last_key_pos = command["last_key_pos"] if last_key_pos < 0: - last_key_pos = len(args) - abs(last_key_pos) - keys_pos = list( - range(command["first_key_pos"], last_key_pos + 1, command["step_count"]) - ) - keys = [args[pos] for pos in keys_pos] + last_key_pos += len(args) + keys = list(map( + args.__getitem__, + range(first_key_pos, last_key_pos + 1, step_count) + )) return keys @@ -454,21 +480,30 @@ async def get_keys(self, *args: Any) -> Optional[Tuple[str, ...]]: # We'll try to reinitialize the commands cache, if the engine # version has changed, the commands may not be current await self.initialize() - if cmd_name not in self.commands: + commands = self.commands + if cmd_name not in commands: raise RedisError( f"{cmd_name.upper()} command doesn't exist in Redis commands" ) - command = self.commands.get(cmd_name) - if "movablekeys" in command["flags"]: + single_pos = command.get("_single_key_pos") + if single_pos is not None: + return [args[single_pos]] + + command = commands.get(cmd_name) + flags = command["flags"] + if "movablekeys" in flags: keys = await self._get_moveable_keys(*args) - elif "pubsub" in command["flags"] or command["name"] == "pubsub": + elif "pubsub" in flags or command["name"] == "pubsub": keys = self._get_pubsub_keys(*args) else: + step_count = command["step_count"] + first_key_pos = command["first_key_pos"] + last_key_pos = command["last_key_pos"] if ( - command["step_count"] == 0 - and command["first_key_pos"] == 0 - and command["last_key_pos"] == 0 + step_count == 0 + and first_key_pos == 0 + and last_key_pos == 0 ): is_subcmd = False if "subcommands" in command: @@ -477,19 +512,21 @@ async def get_keys(self, *args: Any) -> Optional[Tuple[str, ...]]: if str_if_bytes(subcmd[0]) == subcmd_name: command = self.parse_subcommand(subcmd) - if command["first_key_pos"] > 0: + first_key_pos = command["first_key_pos"] + last_key_pos = command["last_key_pos"] + step_count = command["step_count"] + if first_key_pos > 0: is_subcmd = True # The command doesn't have keys in it if not is_subcmd: return None - last_key_pos = command["last_key_pos"] if last_key_pos < 0: - last_key_pos = len(args) - abs(last_key_pos) - keys_pos = list( - range(command["first_key_pos"], last_key_pos + 1, command["step_count"]) - ) - keys = [args[pos] for pos in keys_pos] + last_key_pos += len(args) + keys = [ + args[pos] + for pos in range(first_key_pos, last_key_pos + 1, step_count) + ] return keys diff --git a/redis/cluster.py b/redis/cluster.py index 0aa23c5f6b..a4bd650498 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -935,18 +935,23 @@ def __init__( self._policies_callback_mapping: dict[ Union[RequestPolicy, ResponsePolicy], Callable ] = { - RequestPolicy.DEFAULT_KEYLESS: lambda command_name: [ + RequestPolicy.DEFAULT_KEYLESS: lambda self, command_name, *args, **kwargs: [ self.get_random_primary_or_all_nodes(command_name) ], - RequestPolicy.DEFAULT_KEYED: lambda command, - *args: self.get_nodes_from_slot(command, *args), - RequestPolicy.DEFAULT_NODE: lambda: [self.get_default_node()], - RequestPolicy.ALL_SHARDS: self.get_primaries, - RequestPolicy.ALL_NODES: self.get_nodes, - RequestPolicy.ALL_REPLICAS: self.get_replicas, - RequestPolicy.MULTI_SHARD: lambda *args, - **kwargs: self._split_multi_shard_command(*args, **kwargs), - RequestPolicy.SPECIAL: self.get_special_nodes, + RequestPolicy.DEFAULT_KEYED: lambda self, command_name, *args, **kwargs: + self.get_nodes_from_slot(command, *args), + RequestPolicy.DEFAULT_NODE: lambda self, command_name, *args, **kwargs: + [self.get_default_node()], + RequestPolicy.ALL_SHARDS: lambda self, command_name, *args, **kwargs: + self.get_primaries(), + RequestPolicy.ALL_NODES: lambda self, command_name, *args, **kwargs: + self.get_nodes(), + RequestPolicy.ALL_REPLICAS: lambda self, command_name, *args, **kwargs: + self.get_replicas(), + RequestPolicy.MULTI_SHARD: lambda self, command_name, *args, **kwargs: + self._split_multi_shard_command(*args, **kwargs), + RequestPolicy.SPECIAL: lambda self, command_name, *args, **kwargs: + self.get_special_nodes(), ResponsePolicy.DEFAULT_KEYLESS: lambda res: res, ResponsePolicy.DEFAULT_KEYED: lambda res: res, } @@ -1315,9 +1320,10 @@ def _determine_nodes( """ Determines a nodes the command should be executed on. """ - command = args[0].upper() - if len(args) >= 2 and f"{args[0]} {args[1]}".upper() in self.command_flags: - command = f"{args[0]} {args[1]}".upper() + arg0 = args[0] + command = arg0.upper() + if len(args) >= 2 and f"{arg0} {args[1]}".upper() in self.command_flags: + command = f"{arg0} {args[1]}".upper() nodes_flag = kwargs.pop("nodes_flag", None) if nodes_flag is not None: @@ -1327,21 +1333,14 @@ def _determine_nodes( # get the nodes group for this command if it was predefined command_flag = self.command_flags.get(command) - if command_flag in self._command_flags_mapping: - request_policy = self._command_flags_mapping[command_flag] + request_policy = self._command_flags_mapping.get( + command_flag, request_policy) - policy_callback = self._policies_callback_mapping[request_policy] - - if request_policy == RequestPolicy.DEFAULT_KEYED: - nodes = policy_callback(command, *args) - elif request_policy == RequestPolicy.MULTI_SHARD: - nodes = policy_callback(*args, **kwargs) - elif request_policy == RequestPolicy.DEFAULT_KEYLESS: - nodes = policy_callback(args[0]) - else: - nodes = policy_callback() + nodes = self._policies_callback_mapping[request_policy]( + self, command, *args, **kwargs, + ) - if args[0].lower() == "ft.aggregate": + if arg0.lower() == "ft.aggregate": self._aggregate_nodes = nodes return nodes @@ -1397,14 +1396,15 @@ def determine_slot(self, *args) -> Optional[int]: # CLIENT TRACKING is a special case. # It doesn't have any keys, it needs to be sent to the provided nodes # By default it will be sent to all nodes. - if command.upper() == "CLIENT TRACKING": + commandu = command.upper() + if commandu == "CLIENT TRACKING": return None # EVAL and EVALSHA are common enough that it's wasteful to go to the # redis server to parse the keys. Besides, there is a bug in redis<7.0 # where `self._get_command_keys()` fails anyway. So, we special case # EVAL/EVALSHA. - if command.upper() in ("EVAL", "EVALSHA"): + if commandu in ("EVAL", "EVALSHA"): # command syntax: EVAL "script body" num_keys ... if len(args) <= 2: raise RedisClusterException(f"Invalid args in command: {args}") @@ -1417,10 +1417,10 @@ def determine_slot(self, *args) -> Optional[int]: keys = eval_keys else: keys = self._get_command_keys(*args) - if keys is None or len(keys) == 0: + if not keys: # FCALL can call a function with 0 keys, that means the function # can be run on any node so we can just return a random slot - if command.upper() in ("FCALL", "FCALL_RO"): + if commandu in ("FCALL", "FCALL_RO"): return random.randrange(0, REDIS_CLUSTER_HASH_SLOTS) raise RedisClusterException( "No way to dispatch this command to Redis Cluster. " @@ -1429,12 +1429,13 @@ def determine_slot(self, *args) -> Optional[int]: ) # single key command + keyslot = self.keyslot if len(keys) == 1: - return self.keyslot(keys[0]) + return keyslot(keys[0]) # multi-key command; we need to make sure all keys are mapped to # the same slot - slots = {self.keyslot(key) for key in keys} + slots = {keyslot(key) for key in keys} if len(slots) != 1: raise RedisClusterException( f"{command} - all keys must map to the same key slot" @@ -3435,18 +3436,23 @@ def __init__( self._policies_callback_mapping: dict[ Union[RequestPolicy, ResponsePolicy], Callable ] = { - RequestPolicy.DEFAULT_KEYLESS: lambda command_name: [ + RequestPolicy.DEFAULT_KEYLESS: lambda self, command_name, *args, **kwargs: [ self.get_random_primary_or_all_nodes(command_name) ], - RequestPolicy.DEFAULT_KEYED: lambda command, - *args: self.get_nodes_from_slot(command, *args), - RequestPolicy.DEFAULT_NODE: lambda: [self.get_default_node()], - RequestPolicy.ALL_SHARDS: self.get_primaries, - RequestPolicy.ALL_NODES: self.get_nodes, - RequestPolicy.ALL_REPLICAS: self.get_replicas, - RequestPolicy.MULTI_SHARD: lambda *args, - **kwargs: self._split_multi_shard_command(*args, **kwargs), - RequestPolicy.SPECIAL: self.get_special_nodes, + RequestPolicy.DEFAULT_KEYED: lambda self, command_name, *args, **kwargs: + self.get_nodes_from_slot(command, *args), + RequestPolicy.DEFAULT_NODE: lambda self, command_name, *args, **kwargs: + [self.get_default_node()], + RequestPolicy.ALL_SHARDS: lambda self, command_name, *args, **kwargs: + self.get_primaries(), + RequestPolicy.ALL_NODES: lambda self, command_name, *args, **kwargs: + self.get_nodes(), + RequestPolicy.ALL_REPLICAS: lambda self, command_name, *args, **kwargs: + self.get_replicas(), + RequestPolicy.MULTI_SHARD: lambda self, command_name, *args, **kwargs: + self._split_multi_shard_command(*args, **kwargs), + RequestPolicy.SPECIAL: lambda self, command_name, *args, **kwargs: + self.get_special_nodes(), ResponsePolicy.DEFAULT_KEYLESS: lambda res: res, ResponsePolicy.DEFAULT_KEYED: lambda res: res, } @@ -4299,12 +4305,14 @@ def _determine_nodes( ) -> List["ClusterNode"]: # Determine which nodes should be executed the command on. # Returns a list of target nodes. - command = args[0].upper() + pipe = self._pipe + arg0 = args[0] + command = arg0.upper() if ( len(args) >= 2 - and f"{args[0]} {args[1]}".upper() in self._pipe.command_flags + and f"{arg0} {args[1]}".upper() in pipe.command_flags ): - command = f"{args[0]} {args[1]}".upper() + command = f"{arg0} {args[1]}".upper() nodes_flag = kwargs.pop("nodes_flag", None) if nodes_flag is not None: @@ -4312,23 +4320,15 @@ def _determine_nodes( command_flag = nodes_flag else: # get the nodes group for this command if it was predefined - command_flag = self._pipe.command_flags.get(command) + command_flag = pipe.command_flags.get(command) - if command_flag in self._pipe._command_flags_mapping: - request_policy = self._pipe._command_flags_mapping[command_flag] - - policy_callback = self._pipe._policies_callback_mapping[request_policy] - - if request_policy == RequestPolicy.DEFAULT_KEYED: - nodes = policy_callback(command, *args) - elif request_policy == RequestPolicy.MULTI_SHARD: - nodes = policy_callback(*args, **kwargs) - elif request_policy == RequestPolicy.DEFAULT_KEYLESS: - nodes = policy_callback(args[0]) - else: - nodes = policy_callback() + request_policy = pipe._command_flags_mapping.get( + command_flag, request_policy) + nodes = pipe._policies_callback_mapping[request_policy]( + pipe, command, *args, **kwargs + ) - if args[0].lower() == "ft.aggregate": + if arg0.lower() == "ft.aggregate": self._aggregate_nodes = nodes return nodes From 144ee44f103798e5ba72f3d5c356cd7af11e6350 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Fri, 15 May 2026 02:20:38 -0300 Subject: [PATCH 03/22] More micro-optimization Common subexpression factorization --- redis/cluster.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/redis/cluster.py b/redis/cluster.py index a4bd650498..21595d49d9 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -4058,6 +4058,9 @@ def _send_cluster_commands( request_policy=RequestPolicy.DEFAULT_KEYED, response_policy=ResponsePolicy.DEFAULT_KEYED, ) + policy_resolver = pipe._policy_resolver + pipe_command_flags = pipe.command_flags + command_flags = self.command_flags try: # as we move through each command that still needs to be processed, @@ -4065,8 +4068,9 @@ def _send_cluster_commands( # the slot determine the node. for c in attempt: args = c.args - command_policies = pipe._policy_resolver.resolve( - args[0].lower() + arg0 = args[0] + command_policies = policy_resolver.resolve( + arg0.lower() ) # refer to our internal node -> slot table that # tells us where a given command should route to. @@ -4081,15 +4085,15 @@ def _send_cluster_commands( else: if not command_policies: if len(args) >= 2: - command = f"{args[0]} {args[1]}".upper() - if command not in pipe.command_flags: - command = args[0].upper() + command = f"{arg0} {args[1]}".upper() + if command not in pipe_command_flags: + command = arg0.upper() else: - command = args[0].upper() + command = arg0.upper() # We only could resolve key properties if command is not # in a list of pre-defined request policies - command_flag = self.command_flags.get(command) + command_flag = command_flags.get(command) if not command_flag: # Fallback to default policy if not pipe.get_default_node(): From 0c9df7f804173fa824d605b680fe0c829d721137 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Fri, 15 May 2026 18:59:51 -0300 Subject: [PATCH 04/22] Fix typo --- redis/cluster.py | 36 ++++++++++++++++++------------------ 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/redis/cluster.py b/redis/cluster.py index 21595d49d9..c1029359c5 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -935,22 +935,22 @@ def __init__( self._policies_callback_mapping: dict[ Union[RequestPolicy, ResponsePolicy], Callable ] = { - RequestPolicy.DEFAULT_KEYLESS: lambda self, command_name, *args, **kwargs: [ - self.get_random_primary_or_all_nodes(command_name) + RequestPolicy.DEFAULT_KEYLESS: lambda self, command, *args, **kwargs: [ + self.get_random_primary_or_all_nodes(command) ], - RequestPolicy.DEFAULT_KEYED: lambda self, command_name, *args, **kwargs: + RequestPolicy.DEFAULT_KEYED: lambda self, command, *args, **kwargs: self.get_nodes_from_slot(command, *args), - RequestPolicy.DEFAULT_NODE: lambda self, command_name, *args, **kwargs: + RequestPolicy.DEFAULT_NODE: lambda self, command, *args, **kwargs: [self.get_default_node()], - RequestPolicy.ALL_SHARDS: lambda self, command_name, *args, **kwargs: + RequestPolicy.ALL_SHARDS: lambda self, command, *args, **kwargs: self.get_primaries(), - RequestPolicy.ALL_NODES: lambda self, command_name, *args, **kwargs: + RequestPolicy.ALL_NODES: lambda self, command, *args, **kwargs: self.get_nodes(), - RequestPolicy.ALL_REPLICAS: lambda self, command_name, *args, **kwargs: + RequestPolicy.ALL_REPLICAS: lambda self, command, *args, **kwargs: self.get_replicas(), - RequestPolicy.MULTI_SHARD: lambda self, command_name, *args, **kwargs: + RequestPolicy.MULTI_SHARD: lambda self, command, *args, **kwargs: self._split_multi_shard_command(*args, **kwargs), - RequestPolicy.SPECIAL: lambda self, command_name, *args, **kwargs: + RequestPolicy.SPECIAL: lambda self, command, *args, **kwargs: self.get_special_nodes(), ResponsePolicy.DEFAULT_KEYLESS: lambda res: res, ResponsePolicy.DEFAULT_KEYED: lambda res: res, @@ -3436,22 +3436,22 @@ def __init__( self._policies_callback_mapping: dict[ Union[RequestPolicy, ResponsePolicy], Callable ] = { - RequestPolicy.DEFAULT_KEYLESS: lambda self, command_name, *args, **kwargs: [ - self.get_random_primary_or_all_nodes(command_name) + RequestPolicy.DEFAULT_KEYLESS: lambda self, command, *args, **kwargs: [ + self.get_random_primary_or_all_nodes(command) ], - RequestPolicy.DEFAULT_KEYED: lambda self, command_name, *args, **kwargs: + RequestPolicy.DEFAULT_KEYED: lambda self, command, *args, **kwargs: self.get_nodes_from_slot(command, *args), - RequestPolicy.DEFAULT_NODE: lambda self, command_name, *args, **kwargs: + RequestPolicy.DEFAULT_NODE: lambda self, command, *args, **kwargs: [self.get_default_node()], - RequestPolicy.ALL_SHARDS: lambda self, command_name, *args, **kwargs: + RequestPolicy.ALL_SHARDS: lambda self, command, *args, **kwargs: self.get_primaries(), - RequestPolicy.ALL_NODES: lambda self, command_name, *args, **kwargs: + RequestPolicy.ALL_NODES: lambda self, command, *args, **kwargs: self.get_nodes(), - RequestPolicy.ALL_REPLICAS: lambda self, command_name, *args, **kwargs: + RequestPolicy.ALL_REPLICAS: lambda self, command, *args, **kwargs: self.get_replicas(), - RequestPolicy.MULTI_SHARD: lambda self, command_name, *args, **kwargs: + RequestPolicy.MULTI_SHARD: lambda self, command, *args, **kwargs: self._split_multi_shard_command(*args, **kwargs), - RequestPolicy.SPECIAL: lambda self, command_name, *args, **kwargs: + RequestPolicy.SPECIAL: lambda self, command, *args, **kwargs: self.get_special_nodes(), ResponsePolicy.DEFAULT_KEYLESS: lambda res: res, ResponsePolicy.DEFAULT_KEYED: lambda res: res, From 658573461b68c5dd59664b240f7433d6bc1598ac Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Mon, 18 May 2026 15:20:13 -0300 Subject: [PATCH 05/22] Fix tests --- redis/_parsers/commands.py | 8 +++++--- redis/commands/policies.py | 2 +- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/redis/_parsers/commands.py b/redis/_parsers/commands.py index 450c201b87..5270ca23b4 100644 --- a/redis/_parsers/commands.py +++ b/redis/_parsers/commands.py @@ -468,12 +468,14 @@ async def get_keys(self, *args: Any) -> Optional[Tuple[str, ...]]: return None cmd_name = args[0].lower() - if cmd_name not in self.commands: + commands = self.commands + command = commands.get(cmd_name) + if command is None: # try to split the command name and to take only the main command, # e.g. 'memory' for 'memory usage' cmd_name_split = cmd_name.split() cmd_name = cmd_name_split[0] - if cmd_name in self.commands: + if cmd_name in commands: # save the split command to args args = cmd_name_split + list(args[1:]) else: @@ -485,12 +487,12 @@ async def get_keys(self, *args: Any) -> Optional[Tuple[str, ...]]: raise RedisError( f"{cmd_name.upper()} command doesn't exist in Redis commands" ) + command = commands.get(cmd_name) single_pos = command.get("_single_key_pos") if single_pos is not None: return [args[single_pos]] - command = commands.get(cmd_name) flags = command["flags"] if "movablekeys" in flags: keys = await self._get_moveable_keys(*args) diff --git a/redis/commands/policies.py b/redis/commands/policies.py index 4a4309a39b..4063b15ac9 100644 --- a/redis/commands/policies.py +++ b/redis/commands/policies.py @@ -189,7 +189,7 @@ def __init__( self._fallback = fallback def resolve(self, command_name: str) -> Optional[CommandPolicies]: - parts = command_name.split(".", 1) + parts = command_name.split(".", 2) if len(parts) > 2: raise ValueError(f"Wrong command or module name: {command_name}") From 12cd751fabd5b6938c2dbef8f46d3de0a427d8f7 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Mon, 18 May 2026 17:39:28 -0300 Subject: [PATCH 06/22] Fix tests --- redis/commands/policies.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/redis/commands/policies.py b/redis/commands/policies.py index 4063b15ac9..77febba13d 100644 --- a/redis/commands/policies.py +++ b/redis/commands/policies.py @@ -230,7 +230,7 @@ def __init__( self._fallback = fallback async def resolve(self, command_name: str) -> Optional[CommandPolicies]: - parts = command_name.split(".", 1) + parts = command_name.split(".", 2) if len(parts) > 2: raise ValueError(f"Wrong command or module name: {command_name}") From 3c3b0d354ef7e57e22fba440715333600400a19a Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 20 May 2026 09:53:19 -0300 Subject: [PATCH 07/22] Fast-path for policy callback search Getting the policy callback out of all the dicts is very indirect and wasteful for simple commands. Cache simple commands so the next time it only takes a single lookup. About 3.7% improvement on overall benchmark. --- redis/_parsers/encoders.py | 9 ++-- redis/cluster.py | 93 +++++++++++++++++++++++--------------- 2 files changed, 61 insertions(+), 41 deletions(-) diff --git a/redis/_parsers/encoders.py b/redis/_parsers/encoders.py index 6fdf0ad882..761be0191a 100644 --- a/redis/_parsers/encoders.py +++ b/redis/_parsers/encoders.py @@ -15,6 +15,8 @@ def encode(self, value): "Return a bytestring or bytes-like representation of the value" if isinstance(value, (bytes, memoryview)): return value + elif isinstance(value, str): + return value.encode(self.encoding, self.encoding_errors) elif isinstance(value, bool): # special case bool since it is a subclass of int raise DataError( @@ -22,17 +24,14 @@ def encode(self, value): "bytes, string, int or float first." ) elif isinstance(value, (int, float)): - value = repr(value).encode() - elif not isinstance(value, str): + return repr(value).encode() + else: # a value we don't know how to deal with. throw an error typename = type(value).__name__ raise DataError( f"Invalid input of type: '{typename}'. " f"Convert to a bytes, string, int or float first." ) - if isinstance(value, str): - value = value.encode(self.encoding, self.encoding_errors) - return value def decode(self, value, force=False): "Return a unicode string from the bytes-like representation" diff --git a/redis/cluster.py b/redis/cluster.py index c1029359c5..1949def264 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -957,6 +957,7 @@ def __init__( } self._policy_resolver = policy_resolver + self._policy_cb_cache = {} self.commands_parser = CommandsParser(self) # Node where FT.AGGREGATE command is executed. @@ -1070,11 +1071,12 @@ def get_nodes_from_slot(self, command: str, *args): Returns a list of nodes that hold the specified keys' slots. """ # get the node that holds the key's slot + is_read = command in READ_COMMANDS slot = self.determine_slot(*args) node = self.nodes_manager.get_node_from_slot( slot, - self.read_from_replicas and command in READ_COMMANDS, - self.load_balancing_strategy if command in READ_COMMANDS else None, + self.read_from_replicas and is_read, + self.load_balancing_strategy if is_read else None, ) return [node] @@ -1321,24 +1323,33 @@ def _determine_nodes( Determines a nodes the command should be executed on. """ arg0 = args[0] - command = arg0.upper() - if len(args) >= 2 and f"{arg0} {args[1]}".upper() in self.command_flags: - command = f"{arg0} {args[1]}".upper() - - nodes_flag = kwargs.pop("nodes_flag", None) - if nodes_flag is not None: - # nodes flag passed by the user - command_flag = nodes_flag + policy_cb = self._policy_cb_cache.get(arg0) + if policy_cb is None: + command = arg0.upper() + if len(args) >= 2 and f"{arg0} {args[1]}".upper() in self.command_flags: + command = f"{arg0} {args[1]}".upper() + + nodes_flag = kwargs.pop("nodes_flag", None) + if nodes_flag is not None: + # nodes flag passed by the user + command_flag = nodes_flag + else: + # get the nodes group for this command if it was predefined + command_flag = self.command_flags.get(command) + + request_policy = self._command_flags_mapping.get( + command_flag, request_policy) + + policy_cb = self._policies_callback_mapping[request_policy] + if command == arg0: + if len(self._policy_cb_cache) > 5000: + # Prevent unbounded memory leak on abnormal use + self._policy_cb_cache.clear() + self._policy_cb_cache[arg0] = policy_cb else: - # get the nodes group for this command if it was predefined - command_flag = self.command_flags.get(command) + command = arg0 - request_policy = self._command_flags_mapping.get( - command_flag, request_policy) - - nodes = self._policies_callback_mapping[request_policy]( - self, command, *args, **kwargs, - ) + nodes = policy_cb(self, command, *args, **kwargs) if arg0.lower() == "ft.aggregate": self._aggregate_nodes = nodes @@ -3458,6 +3469,7 @@ def __init__( } self._policy_resolver = policy_resolver + self._policy_cb_cache = {} if event_dispatcher is None: self._event_dispatcher = EventDispatcher() @@ -4311,26 +4323,35 @@ def _determine_nodes( # Returns a list of target nodes. pipe = self._pipe arg0 = args[0] - command = arg0.upper() - if ( - len(args) >= 2 - and f"{arg0} {args[1]}".upper() in pipe.command_flags - ): - command = f"{arg0} {args[1]}".upper() + policy_cb = pipe._policy_cb_cache.get(arg0) + if policy_cb is None: + command = arg0.upper() + if ( + len(args) >= 2 + and f"{arg0} {args[1]}".upper() in pipe.command_flags + ): + command = f"{arg0} {args[1]}".upper() - nodes_flag = kwargs.pop("nodes_flag", None) - if nodes_flag is not None: - # nodes flag passed by the user - command_flag = nodes_flag + nodes_flag = kwargs.pop("nodes_flag", None) + if nodes_flag is not None: + # nodes flag passed by the user + command_flag = nodes_flag + else: + # get the nodes group for this command if it was predefined + command_flag = pipe.command_flags.get(command) + + request_policy = pipe._command_flags_mapping.get( + command_flag, request_policy) + policy_cb = pipe._policies_callback_mapping[request_policy] + + if command == arg0: + if len(pipe._policy_cb_cache) > 5000: + # Prevent unbounded memory leak on abnormal use + pipe._policy_cb_cache.clear() + pipe._policy_cb_cache[arg0] = policy_cb else: - # get the nodes group for this command if it was predefined - command_flag = pipe.command_flags.get(command) - - request_policy = pipe._command_flags_mapping.get( - command_flag, request_policy) - nodes = pipe._policies_callback_mapping[request_policy]( - pipe, command, *args, **kwargs - ) + command = arg0 + nodes = policy_cb(pipe, command, *args, **kwargs) if arg0.lower() == "ft.aggregate": self._aggregate_nodes = nodes From 918115160f64aac4d972794a8053fd93f94acb03 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 20 May 2026 10:29:09 -0300 Subject: [PATCH 08/22] Local cache of command policies Avoid resolving the same command policies over and over. All resolvers are "static" in the short term, during a single pipeline execution, so cache their results for the duration to reduce the amount of work done. --- redis/cluster.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/redis/cluster.py b/redis/cluster.py index 1949def264..95e89e294d 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -4074,6 +4074,9 @@ def _send_cluster_commands( pipe_command_flags = pipe.command_flags command_flags = self.command_flags + policy_cache = {} + SENTINEL = object() + try: # as we move through each command that still needs to be processed, # we figure out the slot number that command maps to, then from @@ -4081,9 +4084,14 @@ def _send_cluster_commands( for c in attempt: args = c.args arg0 = args[0] - command_policies = policy_resolver.resolve( - arg0.lower() - ) + + command_policies = policy_cache.get(arg0, SENTINEL) + if command_policies is SENTINEL: + command_policies = policy_resolver.resolve( + arg0.lower() + ) + policy_cache[arg0] = command_policies + # refer to our internal node -> slot table that # tells us where a given command should route to. # (it might be possible we have a cached node that no longer From c311f8979e37059b991b8238a091858990f27652 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 20 May 2026 14:59:47 -0300 Subject: [PATCH 09/22] Revert mis-optimization --- redis/cluster.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/redis/cluster.py b/redis/cluster.py index 95e89e294d..adc072ce35 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -1440,13 +1440,12 @@ def determine_slot(self, *args) -> Optional[int]: ) # single key command - keyslot = self.keyslot if len(keys) == 1: - return keyslot(keys[0]) + return self.keyslot(keys[0]) # multi-key command; we need to make sure all keys are mapped to # the same slot - slots = {keyslot(key) for key in keys} + slots = {self.keyslot(key) for key in keys} if len(slots) != 1: raise RedisClusterException( f"{command} - all keys must map to the same key slot" From e19de3c5f56e7ff565d12e06cb77880982b6ae28 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 20 May 2026 15:36:15 -0300 Subject: [PATCH 10/22] Fast-path for parse_response Simple commands have simple responses, so a fast path for those improves response parsing speed in the average --- redis/client.py | 32 ++++++++++++++++++-------------- 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/redis/client.py b/redis/client.py index 1d2cb5f5ff..8558cc6371 100755 --- a/redis/client.py +++ b/redis/client.py @@ -820,22 +820,26 @@ def failure_callback(error, failure_count): def parse_response(self, connection, command_name, **options): """Parses a response from the Redis server""" - try: - if NEVER_DECODE in options: - response = connection.read_response(disable_decoding=True) - options.pop(NEVER_DECODE) - else: - response = connection.read_response() - except ResponseError: - if EMPTY_RESPONSE in options: - return options[EMPTY_RESPONSE] - raise + if not options: + # Fast-path for the common case of no options + response = connection.read_response() + else: + try: + if NEVER_DECODE in options: + response = connection.read_response(disable_decoding=True) + options.pop(NEVER_DECODE) + else: + response = connection.read_response() + except ResponseError: + if EMPTY_RESPONSE in options: + return options[EMPTY_RESPONSE] + raise - if EMPTY_RESPONSE in options: - options.pop(EMPTY_RESPONSE) + if EMPTY_RESPONSE in options: + options.pop(EMPTY_RESPONSE) - # Remove keys entry, it needs only for cache. - options.pop("keys", None) + # Remove keys entry, it needs only for cache. + options.pop("keys", None) if command_name in self.response_callbacks: return self.response_callbacks[command_name](response, **options) From c8d510486aedcb722a0193157b62b0c57400e879 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 20 May 2026 22:36:31 -0300 Subject: [PATCH 11/22] Fix tests --- redis/cluster.py | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/redis/cluster.py b/redis/cluster.py index adc072ce35..8b1c55d9d2 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -1323,13 +1323,16 @@ def _determine_nodes( Determines a nodes the command should be executed on. """ arg0 = args[0] - policy_cb = self._policy_cb_cache.get(arg0) + nodes_flag = kwargs.pop("nodes_flag", None) + if nodes_flag is None: + policy_cb = self._policy_cb_cache.get(arg0) + else: + policy_cb = None if policy_cb is None: command = arg0.upper() if len(args) >= 2 and f"{arg0} {args[1]}".upper() in self.command_flags: command = f"{arg0} {args[1]}".upper() - nodes_flag = kwargs.pop("nodes_flag", None) if nodes_flag is not None: # nodes flag passed by the user command_flag = nodes_flag @@ -1341,7 +1344,7 @@ def _determine_nodes( command_flag, request_policy) policy_cb = self._policies_callback_mapping[request_policy] - if command == arg0: + if nodes_flag is None and command == arg0: if len(self._policy_cb_cache) > 5000: # Prevent unbounded memory leak on abnormal use self._policy_cb_cache.clear() @@ -4330,7 +4333,11 @@ def _determine_nodes( # Returns a list of target nodes. pipe = self._pipe arg0 = args[0] - policy_cb = pipe._policy_cb_cache.get(arg0) + nodes_flag = kwargs.pop("nodes_flag", None) + if nodes_flag is None: + policy_cb = pipe._policy_cb_cache.get(arg0) + else: + policy_cb = None if policy_cb is None: command = arg0.upper() if ( @@ -4339,7 +4346,6 @@ def _determine_nodes( ): command = f"{arg0} {args[1]}".upper() - nodes_flag = kwargs.pop("nodes_flag", None) if nodes_flag is not None: # nodes flag passed by the user command_flag = nodes_flag @@ -4351,7 +4357,7 @@ def _determine_nodes( command_flag, request_policy) policy_cb = pipe._policies_callback_mapping[request_policy] - if command == arg0: + if nodes_flag is None and command == arg0: if len(pipe._policy_cb_cache) > 5000: # Prevent unbounded memory leak on abnormal use pipe._policy_cb_cache.clear() From f569f1c8f839bed5ccda19564e3343041a588540 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Fri, 22 May 2026 02:09:58 -0300 Subject: [PATCH 12/22] More policy resolution caching Another big speedup in command policy resolution --- redis/_parsers/commands.py | 19 +++++++++++++++++++ redis/cluster.py | 6 +++++- 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/redis/_parsers/commands.py b/redis/_parsers/commands.py index 5270ca23b4..f05966f091 100644 --- a/redis/_parsers/commands.py +++ b/redis/_parsers/commands.py @@ -127,6 +127,25 @@ def initialize(self, r): ): command["_single_key_pos"] = first_key_pos + def _is_keyed_command(self, *args): + """ + Determines whether the command is always keyed, never keyless. + + Must have been initialized, won't automatically do so. + """ + if len(args) < 2: + # The command has no keys in it + return False + + cmd_name = args[0].lower() + commands = self.commands + command = commands.get(cmd_name) + if command is None: + return False + + single_pos = command.get("_single_key_pos") + return single_pos is not None + # As soon as this PR is merged into Redis, we should reimplement # our logic to use COMMAND INFO changes to determine the key positions # https://github.com/redis/redis/pull/8324 diff --git a/redis/cluster.py b/redis/cluster.py index 8b1c55d9d2..cf1d50b11a 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -4075,6 +4075,7 @@ def _send_cluster_commands( policy_resolver = pipe._policy_resolver pipe_command_flags = pipe.command_flags command_flags = self.command_flags + no_default_node = not pipe.get_default_node() policy_cache = {} SENTINEL = object() @@ -4118,7 +4119,7 @@ def _send_cluster_commands( command_flag = command_flags.get(command) if not command_flag: # Fallback to default policy - if not pipe.get_default_node(): + if no_default_node: keys = None else: keys = pipe._get_command_keys(*args) @@ -4126,6 +4127,9 @@ def _send_cluster_commands( command_policies = default_keyless else: command_policies = default_keyed + if command == arg0 and pipe.commands_parser._is_keyed_command(command): + # safe to cache + policy_cache[arg0] = command_policies else: if command_flag in pipe._command_flags_mapping: command_policies = CommandPolicies( From d84702ce41fb3122bb667399bd489bae2b75de32 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Fri, 22 May 2026 02:46:20 -0300 Subject: [PATCH 13/22] Fix optimization --- redis/cluster.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/redis/cluster.py b/redis/cluster.py index cf1d50b11a..215c5955ca 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -4127,7 +4127,7 @@ def _send_cluster_commands( command_policies = default_keyless else: command_policies = default_keyed - if command == arg0 and pipe.commands_parser._is_keyed_command(command): + if command == arg0 and pipe.commands_parser._is_keyed_command(*args): # safe to cache policy_cache[arg0] = command_policies else: From 7f26b3ffd215c188b12f4caffb15b91a6b18e02e Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 27 May 2026 10:09:29 -0300 Subject: [PATCH 14/22] Run CI --- .github/workflows/integration.yaml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml index ec2e87a6c3..982bff778b 100644 --- a/.github/workflows/integration.yaml +++ b/.github/workflows/integration.yaml @@ -13,6 +13,7 @@ on: branches: - master - '[0-9].[0-9]' + - micro_opt schedule: - cron: '0 1 * * *' # nightly build From cbd75435050309218994aff337351b0e8f5642d6 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 27 May 2026 10:13:16 -0300 Subject: [PATCH 15/22] Add more branches --- .github/workflows/integration.yaml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml index 982bff778b..b6100547db 100644 --- a/.github/workflows/integration.yaml +++ b/.github/workflows/integration.yaml @@ -13,6 +13,8 @@ on: branches: - master - '[0-9].[0-9]' + - '[0-9].[0-9]-j' + - '[0-9].[0-9]-j-next' - micro_opt schedule: - cron: '0 1 * * *' # nightly build From 19fad84c53b8bea3b7bcba3f482b070d9144d0c9 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 27 May 2026 11:34:27 -0300 Subject: [PATCH 16/22] Lint fix --- redis/cluster.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/redis/cluster.py b/redis/cluster.py index 215c5955ca..cc09af2f04 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -4078,7 +4078,7 @@ def _send_cluster_commands( no_default_node = not pipe.get_default_node() policy_cache = {} - SENTINEL = object() + sentinel = object() try: # as we move through each command that still needs to be processed, @@ -4088,8 +4088,8 @@ def _send_cluster_commands( args = c.args arg0 = args[0] - command_policies = policy_cache.get(arg0, SENTINEL) - if command_policies is SENTINEL: + command_policies = policy_cache.get(arg0, sentinel) + if command_policies is sentinel: command_policies = policy_resolver.resolve( arg0.lower() ) From 271846b5804146ffa33c92dfc2647145575ddee3 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 27 May 2026 13:37:36 -0300 Subject: [PATCH 17/22] Small opt in response parsing --- redis/_parsers/helpers.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/redis/_parsers/helpers.py b/redis/_parsers/helpers.py index 6900936a14..2034941ef8 100644 --- a/redis/_parsers/helpers.py +++ b/redis/_parsers/helpers.py @@ -1,4 +1,4 @@ -import datetime +from datetime import datetime from redis.utils import str_if_bytes @@ -11,7 +11,7 @@ def timestamp_to_datetime(response): response = int(response) except ValueError: return None - return datetime.datetime.fromtimestamp(response) + return datetime.fromtimestamp(response) def parse_debug_object(response): @@ -869,7 +869,7 @@ def float_or_none(response): def bool_ok(response, **options): - return str_if_bytes(response) == "OK" + return response in (b"OK", "OK") def parse_zadd(response, **options): @@ -1253,7 +1253,7 @@ def parse_pubsub_numsub(response, **options): def parse_client_kill(response, **options): if isinstance(response, int): return response - return str_if_bytes(response) == "OK" + return bool_ok(response) def parse_acl_getuser(response, **options): @@ -1445,7 +1445,7 @@ def parse_set_result(response, **options): # Redis will return a getCommand result. # See `setGenericCommand` in t_string.c return response - return response and str_if_bytes(response) == "OK" + return response and bool_ok(response) def parse_function_list_unified(response, **options): From e34e06fcad7f9f73bf5927b0731bf0caa522a980 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 27 May 2026 14:01:01 -0300 Subject: [PATCH 18/22] Lint fixes --- redis/_parsers/commands.py | 24 +++------ redis/cluster.py | 100 ++++++++++++++++++++----------------- 2 files changed, 62 insertions(+), 62 deletions(-) diff --git a/redis/_parsers/commands.py b/redis/_parsers/commands.py index f05966f091..b44e8fe2d7 100644 --- a/redis/_parsers/commands.py +++ b/redis/_parsers/commands.py @@ -199,11 +199,7 @@ def get_keys(self, redis_conn, *args): step_count = command["step_count"] first_key_pos = command["first_key_pos"] last_key_pos = command["last_key_pos"] - if ( - step_count == 0 - and first_key_pos == 0 - and last_key_pos == 0 - ): + if step_count == 0 and first_key_pos == 0 and last_key_pos == 0: is_subcmd = False if "subcommands" in command: subcmd_name = f"{cmd_name}|{args[1].lower()}" @@ -223,10 +219,11 @@ def get_keys(self, redis_conn, *args): return None if last_key_pos < 0: last_key_pos += len(args) - keys = list(map( - args.__getitem__, - range(first_key_pos, last_key_pos + 1, step_count) - )) + keys = list( + map( + args.__getitem__, range(first_key_pos, last_key_pos + 1, step_count) + ) + ) return keys @@ -521,11 +518,7 @@ async def get_keys(self, *args: Any) -> Optional[Tuple[str, ...]]: step_count = command["step_count"] first_key_pos = command["first_key_pos"] last_key_pos = command["last_key_pos"] - if ( - step_count == 0 - and first_key_pos == 0 - and last_key_pos == 0 - ): + if step_count == 0 and first_key_pos == 0 and last_key_pos == 0: is_subcmd = False if "subcommands" in command: subcmd_name = f"{cmd_name}|{args[1].lower()}" @@ -545,8 +538,7 @@ async def get_keys(self, *args: Any) -> Optional[Tuple[str, ...]]: if last_key_pos < 0: last_key_pos += len(args) keys = [ - args[pos] - for pos in range(first_key_pos, last_key_pos + 1, step_count) + args[pos] for pos in range(first_key_pos, last_key_pos + 1, step_count) ] return keys diff --git a/redis/cluster.py b/redis/cluster.py index cc09af2f04..25e4d750a5 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -938,20 +938,27 @@ def __init__( RequestPolicy.DEFAULT_KEYLESS: lambda self, command, *args, **kwargs: [ self.get_random_primary_or_all_nodes(command) ], - RequestPolicy.DEFAULT_KEYED: lambda self, command, *args, **kwargs: - self.get_nodes_from_slot(command, *args), - RequestPolicy.DEFAULT_NODE: lambda self, command, *args, **kwargs: - [self.get_default_node()], - RequestPolicy.ALL_SHARDS: lambda self, command, *args, **kwargs: - self.get_primaries(), - RequestPolicy.ALL_NODES: lambda self, command, *args, **kwargs: - self.get_nodes(), - RequestPolicy.ALL_REPLICAS: lambda self, command, *args, **kwargs: - self.get_replicas(), - RequestPolicy.MULTI_SHARD: lambda self, command, *args, **kwargs: - self._split_multi_shard_command(*args, **kwargs), - RequestPolicy.SPECIAL: lambda self, command, *args, **kwargs: - self.get_special_nodes(), + RequestPolicy.DEFAULT_KEYED: lambda self, command, *args, **kwargs: ( + self.get_nodes_from_slot(command, *args) + ), + RequestPolicy.DEFAULT_NODE: lambda self, command, *args, **kwargs: [ + self.get_default_node() + ], + RequestPolicy.ALL_SHARDS: lambda self, command, *args, **kwargs: ( + self.get_primaries() + ), + RequestPolicy.ALL_NODES: lambda self, command, *args, **kwargs: ( + self.get_nodes() + ), + RequestPolicy.ALL_REPLICAS: lambda self, command, *args, **kwargs: ( + self.get_replicas() + ), + RequestPolicy.MULTI_SHARD: lambda self, command, *args, **kwargs: ( + self._split_multi_shard_command(*args, **kwargs) + ), + RequestPolicy.SPECIAL: lambda self, command, *args, **kwargs: ( + self.get_special_nodes() + ), ResponsePolicy.DEFAULT_KEYLESS: lambda res: res, ResponsePolicy.DEFAULT_KEYED: lambda res: res, } @@ -1341,7 +1348,8 @@ def _determine_nodes( command_flag = self.command_flags.get(command) request_policy = self._command_flags_mapping.get( - command_flag, request_policy) + command_flag, request_policy + ) policy_cb = self._policies_callback_mapping[request_policy] if nodes_flag is None and command == arg0: @@ -3452,20 +3460,27 @@ def __init__( RequestPolicy.DEFAULT_KEYLESS: lambda self, command, *args, **kwargs: [ self.get_random_primary_or_all_nodes(command) ], - RequestPolicy.DEFAULT_KEYED: lambda self, command, *args, **kwargs: - self.get_nodes_from_slot(command, *args), - RequestPolicy.DEFAULT_NODE: lambda self, command, *args, **kwargs: - [self.get_default_node()], - RequestPolicy.ALL_SHARDS: lambda self, command, *args, **kwargs: - self.get_primaries(), - RequestPolicy.ALL_NODES: lambda self, command, *args, **kwargs: - self.get_nodes(), - RequestPolicy.ALL_REPLICAS: lambda self, command, *args, **kwargs: - self.get_replicas(), - RequestPolicy.MULTI_SHARD: lambda self, command, *args, **kwargs: - self._split_multi_shard_command(*args, **kwargs), - RequestPolicy.SPECIAL: lambda self, command, *args, **kwargs: - self.get_special_nodes(), + RequestPolicy.DEFAULT_KEYED: lambda self, command, *args, **kwargs: ( + self.get_nodes_from_slot(command, *args) + ), + RequestPolicy.DEFAULT_NODE: lambda self, command, *args, **kwargs: [ + self.get_default_node() + ], + RequestPolicy.ALL_SHARDS: lambda self, command, *args, **kwargs: ( + self.get_primaries() + ), + RequestPolicy.ALL_NODES: lambda self, command, *args, **kwargs: ( + self.get_nodes() + ), + RequestPolicy.ALL_REPLICAS: lambda self, command, *args, **kwargs: ( + self.get_replicas() + ), + RequestPolicy.MULTI_SHARD: lambda self, command, *args, **kwargs: ( + self._split_multi_shard_command(*args, **kwargs) + ), + RequestPolicy.SPECIAL: lambda self, command, *args, **kwargs: ( + self.get_special_nodes() + ), ResponsePolicy.DEFAULT_KEYLESS: lambda res: res, ResponsePolicy.DEFAULT_KEYED: lambda res: res, } @@ -4090,9 +4105,7 @@ def _send_cluster_commands( command_policies = policy_cache.get(arg0, sentinel) if command_policies is sentinel: - command_policies = policy_resolver.resolve( - arg0.lower() - ) + command_policies = policy_resolver.resolve(arg0.lower()) policy_cache[arg0] = command_policies # refer to our internal node -> slot table that @@ -4127,7 +4140,10 @@ def _send_cluster_commands( command_policies = default_keyless else: command_policies = default_keyed - if command == arg0 and pipe.commands_parser._is_keyed_command(*args): + if ( + command == arg0 + and pipe.commands_parser._is_keyed_command(*args) + ): # safe to cache policy_cache[arg0] = command_policies else: @@ -4151,9 +4167,7 @@ def _send_cluster_commands( ) c.command_policies = command_policies if len(target_nodes) > 1: - raise RedisClusterException( - f"Too many targets for command {args}" - ) + raise RedisClusterException(f"Too many targets for command {args}") node = target_nodes[0] if node == pipe.get_default_node(): @@ -4295,11 +4309,7 @@ def _send_cluster_commands( c.options.pop("keys", None) c.result = pipe._policies_callback_mapping[ c.command_policies.response_policy - ]( - pipe.cluster_response_callbacks[c.args[0]]( - c.result, **c.options - ) - ) + ](pipe.cluster_response_callbacks[c.args[0]](}c.result, **c.options)) response.append(c.result) if raise_on_error: @@ -4344,10 +4354,7 @@ def _determine_nodes( policy_cb = None if policy_cb is None: command = arg0.upper() - if ( - len(args) >= 2 - and f"{arg0} {args[1]}".upper() in pipe.command_flags - ): + if len(args) >= 2 and f"{arg0} {args[1]}".upper() in pipe.command_flags: command = f"{arg0} {args[1]}".upper() if nodes_flag is not None: @@ -4358,7 +4365,8 @@ def _determine_nodes( command_flag = pipe.command_flags.get(command) request_policy = pipe._command_flags_mapping.get( - command_flag, request_policy) + command_flag, request_policy + ) policy_cb = pipe._policies_callback_mapping[request_policy] if nodes_flag is None and command == arg0: From 81762f8f45d2f24464a359049e4d4d5acfc1539b Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 27 May 2026 14:12:46 -0300 Subject: [PATCH 19/22] Fix typo --- redis/cluster.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/redis/cluster.py b/redis/cluster.py index 25e4d750a5..ed76244935 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -4309,7 +4309,7 @@ def _send_cluster_commands( c.options.pop("keys", None) c.result = pipe._policies_callback_mapping[ c.command_policies.response_policy - ](pipe.cluster_response_callbacks[c.args[0]](}c.result, **c.options)) + ](pipe.cluster_response_callbacks[c.args[0]](c.result, **c.options)) response.append(c.result) if raise_on_error: From 0169d2adb2881591fe577b0669c374e4dbd0c475 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Wed, 27 May 2026 15:35:18 -0300 Subject: [PATCH 20/22] Lint fixes --- redis/cluster.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/redis/cluster.py b/redis/cluster.py index ed76244935..e9d0e8b825 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -3463,7 +3463,7 @@ def __init__( RequestPolicy.DEFAULT_KEYED: lambda self, command, *args, **kwargs: ( self.get_nodes_from_slot(command, *args) ), - RequestPolicy.DEFAULT_NODE: lambda self, command, *args, **kwargs: [ + RequestPolicy.DEFAULT_NODE: lambda self, command, *args, **kwargs: [ self.get_default_node() ], RequestPolicy.ALL_SHARDS: lambda self, command, *args, **kwargs: ( From 99df02060368c7d29a6c6299d44b0cf6dca2d658 Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Mon, 1 Jun 2026 09:54:20 -0300 Subject: [PATCH 21/22] Revert CI changes Were meant for pre-PR testing only --- .github/workflows/integration.yaml | 3 --- 1 file changed, 3 deletions(-) diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml index 8b69b57d9d..d4ab1439a5 100644 --- a/.github/workflows/integration.yaml +++ b/.github/workflows/integration.yaml @@ -13,9 +13,6 @@ on: branches: - master - '[0-9].[0-9]' - - '[0-9].[0-9]-j' - - '[0-9].[0-9]-j-next' - - micro_opt schedule: - cron: '0 1 * * *' # nightly build From 7f517750f5bc038274534751850c88befdf1046b Mon Sep 17 00:00:00 2001 From: Claudio Freire Date: Mon, 1 Jun 2026 10:18:12 -0300 Subject: [PATCH 22/22] Initialize _single_key_pos in async commands parser --- redis/_parsers/commands.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/redis/_parsers/commands.py b/redis/_parsers/commands.py index b44e8fe2d7..f7ad4ef9b6 100644 --- a/redis/_parsers/commands.py +++ b/redis/_parsers/commands.py @@ -463,7 +463,22 @@ async def initialize(self, node: Optional["ClusterNode"] = None) -> None: self.node = node commands = await self.node.execute_command("COMMAND") - self.commands = {cmd.lower(): command for cmd, command in commands.items()} + commands = {cmd.lower(): command for cmd, command in commands.items()} + + for command in commands.values(): + first_key_pos = command["first_key_pos"] + last_key_pos = command["last_key_pos"] + flags = command["flags"] + if ( + first_key_pos > 0 + and first_key_pos == last_key_pos + and "movablekeys" not in flags + and "pubsub" not in flags + and command["name"] != "pubsub" + ): + command["_single_key_pos"] = first_key_pos + + self.commands = commands # As soon as this PR is merged into Redis, we should reimplement # our logic to use COMMAND INFO changes to determine the key positions