Skip to content

Spike: render request keys and values into one pooled buffer up front #3211

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
mgravell wants to merge 5 commits into
base: main
Choose a base branch
from
Draft

Spike: render request keys and values into one pooled buffer up front #3211

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f06051a
Spike: render request keys and values into one pooled buffer up front
mgravell Sep 9, 2026
a536abe
Spike: add the write and slot-routing halves
mgravell Sep 9, 2026
acbfb05
Enumerator walks the rendered buffer read-only
mgravell Sep 9, 2026
cafa028
Spike: move Execute and ScriptEvaluate onto the rendered buffer
mgravell Sep 9, 2026
3e13576
Wire up RenderedArgs recycling on message completion - partial fix, k…
mgravell Sep 9, 2026
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
7 changes: 6 additions & 1 deletion src/StackExchange.Redis/Message.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -536,7 +536,12 @@ bool ICompletable.TryComplete(bool isAsync)
return true;
}

public void Complete(PhysicalConnection? connection)
// virtual so a subclass owning a rented request buffer (e.g. RenderedArgs) can recycle it here:
// every path that terminally finishes a message - normal completion, high-integrity validation,
// and the RecordConnectionFailed drain - runs through this (via SetExceptionAndComplete, for the
// last one), and PrepareToResend/caller-side timeouts never call it at all, which is exactly the
// "message is definitely never touched again" boundary the buffer needs.
public virtual void Complete(PhysicalConnection? connection)
{
// Ensure we can never call Complete on the same resultBox from two threads by grabbing it now
var currBox = Interlocked.Exchange(ref resultBox, null);
Expand Down
145 changes: 67 additions & 78 deletions src/StackExchange.Redis/RedisDatabase.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2007,7 +2007,7 @@ public Task<long> PublishAsync(RedisChannel channel, RedisValue message, Command

public RespResult ExecuteResp(string command, ReadOnlyMemory<RedisKeyOrValue> args, CommandFlags flags = CommandFlags.None)
{
var msg = new ExecMessage(multiplexer?.CommandMap, Database, flags, command, args);
var msg = new ExecMessage(multiplexer?.CommandMap, Database, flags, command, args, multiplexer?.RawConfig?.RequestBufferPool);
return ExecuteSync(msg, ResultProcessor.RespResult)!;
}

Expand All @@ -2022,7 +2022,7 @@ public RedisResult Execute(string command, ICollection<object> args, CommandFlag

public Task<RespResult> ExecuteRespAsync(string command, ReadOnlyMemory<RedisKeyOrValue> args, CommandFlags flags = CommandFlags.None)
{
var msg = new ExecMessage(multiplexer?.CommandMap, Database, flags, command, args);
var msg = new ExecMessage(multiplexer?.CommandMap, Database, flags, command, args, multiplexer?.RawConfig?.RequestBufferPool);
return ExecuteAsync(msg, ResultProcessor.RespResult, defaultValue: RespResult.NullReply);
}

Expand All @@ -2038,7 +2038,7 @@ public Task<RedisResult> ExecuteAsync(string command, ICollection<object>? args,
public RespResult ScriptEvaluateResp(string script, ReadOnlyMemory<RedisKey> keys, ReadOnlyMemory<RedisValue> values, CommandFlags flags = CommandFlags.None)
{
var command = ResultProcessor.ScriptLoadProcessor.IsSHA1(script) ? RedisCommand.EVALSHA : RedisCommand.EVAL;
var msg = new ScriptEvalMessage(Database, flags, command, script, keys, values);
var msg = new ScriptEvalMessage(Database, flags, command, script, keys, values, multiplexer?.RawConfig?.RequestBufferPool);
try
{
return ExecuteSync(msg, ResultProcessor.RespResult)!;
Expand Down Expand Up @@ -2084,7 +2084,7 @@ public RedisResult ScriptEvaluate(LoadedLuaScript script, object? parameters = n
public async Task<RespResult> ScriptEvaluateRespAsync(string script, ReadOnlyMemory<RedisKey> keys, ReadOnlyMemory<RedisValue> values, CommandFlags flags = CommandFlags.None)
{
var command = ResultProcessor.ScriptLoadProcessor.IsSHA1(script) ? RedisCommand.EVALSHA : RedisCommand.EVAL;
var msg = new ScriptEvalMessage(Database, flags, command, script, keys, values);
var msg = new ScriptEvalMessage(Database, flags, command, script, keys, values, multiplexer?.RawConfig?.RequestBufferPool);

try
{
Expand Down Expand Up @@ -2158,7 +2158,7 @@ public RespResult ScriptEvaluateReadOnlyResp(string script, ReadOnlyMemory<Redis
multiplexer.CommandMap,
ResultProcessor.ScriptLoadProcessor.IsSHA1(script) ? RedisCommand.EVALSHA_RO : RedisCommand.EVAL_RO,
ref flags);
var msg = new ScriptEvalMessage(Database, flags, command, script, keys, values);
var msg = new ScriptEvalMessage(Database, flags, command, script, keys, values, multiplexer?.RawConfig?.RequestBufferPool);
try
{
return ExecuteSync(msg, ResultProcessor.RespResult)!;
Expand Down Expand Up @@ -2201,7 +2201,7 @@ public async Task<RespResult> ScriptEvaluateReadOnlyRespAsync(string script, Rea
multiplexer.CommandMap,
ResultProcessor.ScriptLoadProcessor.IsSHA1(script) ? RedisCommand.EVALSHA_RO : RedisCommand.EVAL_RO,
ref flags);
var msg = new ScriptEvalMessage(Database, flags, command, script, keys, values);
var msg = new ScriptEvalMessage(Database, flags, command, script, keys, values, multiplexer?.RawConfig?.RequestBufferPool);
try
{
return await ExecuteAsync(msg, ResultProcessor.RespResult, defaultValue: RespResult.NullReply).ForAwait();
Expand Down Expand Up @@ -6045,7 +6045,11 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes

internal sealed class ExecMessage : Message
{
private readonly ReadOnlyMemory<RedisKeyOrValue> _args;
// not readonly: RenderedArgs.Recycle swaps the buffer out, and cannot do that through a
// defensive copy - see RenderedArgs
private RenderedArgs _args;
private readonly bool _hasSubCommand;
private readonly SubCommand _subCommand;
private string _unknownCommand;

private static int RemoveDbIfNotRequired(int suggestedDb, string adhocCommand, out RedisCommand knownCommand)
Expand All @@ -6065,7 +6069,7 @@ private static int RemoveDbIfNotRequired(int suggestedDb, string adhocCommand, o
return suggestedDb;
}

public ExecMessage(CommandMap? map, int db, CommandFlags flags, string command, ReadOnlyMemory<RedisKeyOrValue> args)
public ExecMessage(CommandMap? map, int db, CommandFlags flags, string command, ReadOnlyMemory<RedisKeyOrValue> args, MemoryPool<byte>? pool)
: base(RemoveDbIfNotRequired(db, command, out var knownCommand), flags, knownCommand)
{
if (args.Length >= MessageWriter.REDIS_MAX_ARGS) // using >= here because we will be adding 1 for the command itself (which is an arg for the purposes of the multi-bulk protocol)
Expand All @@ -6088,71 +6092,59 @@ public ExecMessage(CommandMap? map, int db, CommandFlags flags, string command,
{
throw ExceptionFactory.CommandDisabled(command);
}
_args = args;

// resolved now, while we still have the arguments as values; only the first one is a
// sub-command candidate, which is what the old write-time loop amounted to
if (args.Length != 0)
{
ref readonly var first = ref args.Span[0];
if (first.IsValue) _hasSubCommand = SubCommandMetadata.TryGetSubCommand(first.Value, out _subCommand);
}

_args = RenderedArgs.Create(args.Span, pool);
}

protected override void WriteImpl(in MessageWriter writer)
{
if (Command is RedisCommand.UNKNOWN)
{
writer.WriteHeader(_unknownCommand, _args.Length);
writer.WriteHeader(_unknownCommand, _args.Count);
}
else
{
writer.WriteHeader(Command, _args.Length);
}
foreach (ref readonly var arg in _args.Span)
{
if (arg.IsKey)
{
writer.Write(arg.Key);
}
else if (arg.IsValue)
{
writer.WriteBulkString(arg.Value);
}
else
{
Debug.Assert(arg.IsNull);
throw new InvalidOperationException("A null is not valid in this context");
}
writer.WriteHeader(Command, _args.Count);
}

// keys already carry their prefix, and a key and a value are the same thing on the wire
_args.WriteTo(writer);
}

public override string CommandString => Command is RedisCommand.UNKNOWN ? _unknownCommand : base.CommandString;
public override string CommandAndKey => CommandString;

public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy)
=> _args.GetHashSlot(serverSelectionStrategy);

public override int ArgCount => _args.Count;

protected override bool TryGetSubCommand(out SubCommand subCommand)
{
int slot = ServerSelectionStrategy.NoSlot;
foreach (ref readonly var arg in _args.Span)
{
var key = arg.Key;
if (!key.IsNull)
{
slot = serverSelectionStrategy.CombineSlot(slot, key);
}
}
return slot;
// resolved in the constructor: by now the arguments are rendered bytes, and this needs
// them as values
subCommand = _hasSubCommand ? _subCommand : SubCommand.Unknown;
return _hasSubCommand;
}
public override int ArgCount => _args.Length;

protected override bool TryGetSubCommand(out SubCommand subCommand)
public override void Complete(PhysicalConnection? connection)
{
// the sub-command (if any) is the first argument after the command itself,
// e.g. CLIENT [GETNAME]; ad-hoc Execute args are boxed objects, so normalize
// the first one to a RedisValue before probing it against the known sub-commands
foreach (ref readonly var arg in _args.Span)
{
var value = arg.Value;
if (!value.IsNull)
{
return SubCommandMetadata.TryGetSubCommand(value, out subCommand);
}
break; // only the first argument is a sub-command candidate
}
subCommand = SubCommand.Unknown;
return false;
// Status flips to Sent strictly *after* WriteImpl returns (see WriteMessageToServerInsideWriteLock),
// but a message is enqueued into _writtenAwaitingResponse - making it eligible for the async-timeout
// heartbeat's SetExceptionAndComplete - *before* that write actually happens. Recycling on that
// premature completion would race an in-flight WriteImpl reading the same buffer. If the write did
// land, the real reply's own later Complete() call recycles once Status has caught up; if it never
// gets sent at all, the buffer leaks rather than risking corruption - the lesser evil.
if (Status == CommandStatus.Sent) RenderedArgs.Recycle(ref _args);
base.Complete(connection);
}
}

Expand Down Expand Up @@ -6297,28 +6289,24 @@ private static bool IsReadOnlyScript(RedisCommand command)

private sealed class ScriptEvalMessage : Message, IMultiMessage
{
private readonly ReadOnlyMemory<RedisKey> _keys;
private readonly ReadOnlyMemory<RedisValue> _values;
// not readonly: RenderedArgs.Recycle swaps the buffer out, and cannot do that through a
// defensive copy - see RenderedArgs. The script itself stays out of the buffer: it is needed
// intact for hash lookup and SCRIPT LOAD, and being a string it carries no lifetime hazard.
private RenderedArgs _args;
private readonly int _keyCount;
private readonly string _script;
private byte[]? asciiHash;
private bool useReadOnly;
public ScriptEvalMessage(int db, CommandFlags flags, RedisCommand command, string script, ReadOnlyMemory<RedisKey> keys, ReadOnlyMemory<RedisValue> values)
public ScriptEvalMessage(int db, CommandFlags flags, RedisCommand command, string script, ReadOnlyMemory<RedisKey> keys, ReadOnlyMemory<RedisValue> values, MemoryPool<byte>? pool)
: base(db, flags, command)
{
_script = script ?? throw new ArgumentNullException(nameof(script));
_keys = keys;
_values = values;
_keyCount = keys.Length;
_args = RenderedArgs.Create(keys.Span, values.Span, pool);
}

public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy)
{
int slot = ServerSelectionStrategy.NoSlot;
foreach (ref readonly var key in _keys.Span)
{
slot = serverSelectionStrategy.CombineSlot(slot, key);
}
return slot;
}
=> _args.GetHashSlot(serverSelectionStrategy);

public IEnumerable<Message> GetMessages(PhysicalConnection connection)
{
Expand Down Expand Up @@ -6348,28 +6336,29 @@ protected override void WriteImpl(in MessageWriter writer)
{
if (asciiHash != null)
{
writer.WriteHeader(useReadOnly ? RedisCommand.EVALSHA_RO : RedisCommand.EVALSHA, 2 + _keys.Length + _values.Length);
writer.WriteHeader(useReadOnly ? RedisCommand.EVALSHA_RO : RedisCommand.EVALSHA, ArgCount);
writer.WriteBulkString(asciiHash);
}
else
{
writer.WriteHeader(useReadOnly ? RedisCommand.EVAL_RO : RedisCommand.EVAL, 2 + _keys.Length + _values.Length);
writer.WriteHeader(useReadOnly ? RedisCommand.EVAL_RO : RedisCommand.EVAL, ArgCount);
writer.WriteBulkString(_script);
}

writer.WriteBulkString(_keys.Length);
writer.WriteBulkString(_keyCount);

foreach (ref readonly var key in _keys.Span)
{
writer.Write(key);
}
// rendered keys-then-values, in the order EVAL wants them; keys already carry their prefix
_args.WriteTo(writer);
}

foreach (ref readonly var value in _values.Span)
{
writer.WriteBulkString(value);
}
public override int ArgCount => 2 + _args.Count;

public override void Complete(PhysicalConnection? connection)
{
// see ExecMessage.Complete for why this is gated on Status rather than unconditional
if (Status == CommandStatus.Sent) RenderedArgs.Recycle(ref _args);
base.Complete(connection);
}
public override int ArgCount => 2 + _keys.Length + _values.Length;
}

private sealed class ScriptEvaluateMessage : Message, IMultiMessage
Expand Down
Loading
Loading