Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
26 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions benchmarks/command_packer_benchmark.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
137 changes: 101 additions & 36 deletions redis/_parsers/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,37 @@ 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

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
Expand All @@ -134,54 +165,65 @@ 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)

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]]

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:
if (
command["step_count"] == 0
and command["first_key_pos"] == 0
and command["last_key_pos"] == 0
):
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:
is_subcmd = False
if "subcommands" in command:
subcmd_name = f"{cmd_name}|{args[1].lower()}"
for subcmd in command["subcommands"]:
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"])
last_key_pos += len(args)
keys = list(
map(
args.__getitem__, range(first_key_pos, last_key_pos + 1, step_count)
)
)
keys = [args[pos] for pos in keys_pos]

return keys

Expand Down Expand Up @@ -421,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
Expand All @@ -442,54 +499,62 @@ 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:
# 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 = commands.get(cmd_name)

single_pos = command.get("_single_key_pos")
if single_pos is not None:
return [args[single_pos]]
Comment thread
cursor[bot] marked this conversation as resolved.

command = self.commands.get(cmd_name)
if "movablekeys" in command["flags"]:
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:
if (
command["step_count"] == 0
and command["first_key_pos"] == 0
and command["last_key_pos"] == 0
):
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:
is_subcmd = False
if "subcommands" in command:
subcmd_name = f"{cmd_name}|{args[1].lower()}"
for subcmd in command["subcommands"]:
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

Expand Down
9 changes: 4 additions & 5 deletions redis/_parsers/encoders.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,24 +15,23 @@ def encode(self, value):
"Return a bytestring or bytes-like representation of the value"
if isinstance(value, (bytes, bytearray, memoryview)):
return value
elif isinstance(value, str):
return value.encode(self.encoding, self.encoding_errors)
Comment thread
cursor[bot] marked this conversation as resolved.
elif isinstance(value, bool):
# special case bool since it is a subclass of int
raise DataError(
"Invalid input of type: 'bool'. Convert to a "
"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"
Expand Down
10 changes: 5 additions & 5 deletions redis/_parsers/helpers.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import datetime
from datetime import datetime

from redis.utils import str_if_bytes

Expand All @@ -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):
Expand Down Expand Up @@ -875,7 +875,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):
Expand Down Expand Up @@ -1259,7 +1259,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):
Expand Down Expand Up @@ -1451,7 +1451,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):
Expand Down
9 changes: 5 additions & 4 deletions redis/_parsers/hiredis.py
Original file line number Diff line number Diff line change
Expand Up @@ -349,17 +349,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
Expand Down
32 changes: 18 additions & 14 deletions redis/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -927,22 +927,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)
Expand Down
Loading