From ec9f64a02e112d4c1c45487da9c970a0d3733485 Mon Sep 17 00:00:00 2001 From: "ugo.lattanzi" Date: Mon, 3 Aug 2026 16:22:32 +0200 Subject: [PATCH] Optimize memory and CPU usage across pool, core, serializers, and compressors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Connection pool: - LeastLoaded no longer re-evaluates TotalOutstanding() per comparison (each call allocates a ServerCounters snapshot in SE.Redis) - Debug log argument evaluation guarded by logger.IsEnabled - RoundRobin uses a lock-free Interlocked counter instead of a CSPRNG (and new Random() per call on netstandard2.1) - Removed pointless static lock held during blocking pool creation - Removed custom MinBy fallback in favor of an allocation-free loop Core: - RedisDatabase instances cached per db number in RedisClient - HashGetAsync(string[]) issues one HMGET instead of N parallel HGETs - HashGetAllAsyncAtOneTimeAsync replaced per-call Lua script with HMGET, fixing KeyPrefix misapplication and a Lua injection vector - UpdateExpiryAsync drops the redundant EXISTS pre-check - AddAllAsync materializes entries in one pre-sized pass (no LINQ pipeline) - Lazy Select results materialized eagerly (SetPop, SortedSetRangeByScore, HashKeys, SearchKeys) to avoid re-deserialization on re-enumeration - Tag writes serialize the key once instead of once per tag - Pub/Sub handler drops the per-message ContinueWith allocation - ToFastArray gains array/List fast paths (CollectionsMarshal on net8+) - Dead code removed: FastIteration, SpanExtensions.Any, duplicated null check in ListGetFromRightAsync Serializers: - Newtonsoft: cached JsonSerializer, stream-based paths with pooled char buffers — no more intermediate UTF-16 string per payload - ServiceStack: stream-based serialization (no intermediate string) - MsgPack: PackSingleObject/UnpackSingleObject, no wrapping MemoryStream - Protobuf: IBufferWriter serialize and ReadOnlyMemory deserialize - Utf8Json: null-guard on Deserialize consistent with other serializers Compressors: - Snappier: removed 100% redundant output copy in Decompress; rented worst-case buffer in Compress - Brotli: one-shot BrotliEncoder.TryCompress instead of stream pipeline - ZstdSharp: per-thread reused compression contexts, exact-size decompress - GZip: pre-sized MemoryStreams ASP.NET Core: - IDistributedCache: static field arrays (no per-Get allocation), batched HSET+EXPIRE in a single network write on SetAsync Version bumped to 13.0.1. Claude-Session: https://claude.ai/code/session_01WfvJwzENxWcSD79kau7rwn --- Directory.Build.props | 16 +- .../Caching/RedisDistributedCache.cs | 29 +++- .../BrotliCompressor.cs | 35 +++- .../GZipCompressor.cs | 7 +- .../SnappierCompressor.cs | 20 ++- .../ZstdSharpCompressor.cs | 38 ++++- .../Extensions/SpanExtensions.cs | 11 -- .../Extensions/ValueLengthExtensions.cs | 51 ++---- .../Helpers/ExceptionThrowHelper.cs | 9 +- .../Helpers/GenericsExtensions.cs | 54 +++--- .../Implementations/RedisClient.cs | 35 +++- .../RedisConnectionPoolManager.cs | 50 ++++-- .../Implementations/RedisDatabase.Hash.cs | 68 ++------ .../Implementations/RedisDatabase.List.cs | 3 - .../Implementations/RedisDatabase.PubSub.cs | 53 +++--- .../Implementations/RedisDatabase.Sort.cs | 3 +- .../Implementations/RedisDatabase.Tags.cs | 13 +- .../Implementations/RedisDatabase.cs | 58 +++---- .../ServerIteration/ServerIteratorFactory.cs | 12 +- .../MsgPackObjectSerializer.cs | 141 ++++++++-------- .../NewtonsoftSerializer.cs | 154 +++++++++++------- .../ProtobufSerializer.cs | 15 +- .../ServiceStackJsonSerializer.cs | 12 +- .../Utf8JsonSerializer.cs | 4 + 24 files changed, 511 insertions(+), 380 deletions(-) diff --git a/Directory.Build.props b/Directory.Build.props index a63e691e..598c3f06 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -3,7 +3,7 @@ Ugo Lattanzi - 13.0.0 + 13.0.1 @@ -50,6 +50,20 @@ Features: Serializer packages (pick one): System.Text.Json, Newtonsoft, MemoryPack, MsgPack, Protobuf, ServiceStack, Utf8Json. +v13.0.1: +- Performance: connection pool selection no longer allocates multiple ServerCounters snapshots per operation (LeastLoaded) and uses a true lock-free round-robin instead of a cryptographic RNG (RoundRobin) +- Performance: RedisDatabase instances are cached per database number instead of being allocated on every Db0..Db16 access +- Performance: HashGetAsync multi-key now issues a single HMGET instead of N parallel HGET round-trips +- Performance: UpdateExpiryAsync no longer issues a redundant EXISTS before EXPIRE (half the round-trips) +- Performance: AddAllAsync builds its payload without LINQ intermediate materializations +- Performance: eager materialization of previously lazy IEnumerable results (SetPopAsync, SortedSetRangeByScoreAsync, HashKeysAsync, SearchKeysAsync) — no more re-deserialization on repeated enumeration +- Performance: serializers (Newtonsoft, ServiceStack, MsgPack, Protobuf) no longer materialize intermediate UTF-16 strings or wrapping MemoryStreams +- Performance: compressors use pooled buffers (Snappier, Brotli, ZstdSharp) and reused Zstd contexts; Snappier.Decompress no longer duplicates the output array +- Performance: IDistributedCache adapter uses cached field arrays and batches HSET+EXPIRE in a single network write +- Fixed HashGetAllAsyncAtOneTimeAsync applying KeyPrefix to hash fields instead of the hash key, and removed a potential Lua injection through the hash key +- Fixed Utf8JsonSerializer.Deserialize throwing on null input instead of returning default +- Fixed ListGetFromRightAsync duplicated null check (dead code) + v13.0.0: - Added HyperLogLog API: HyperLogLogAddAsync, HyperLogLogLengthAsync, HyperLogLogMergeAsync (#637) - Added Distributed Lock API: LockTakeAsync, LockReleaseAsync, LockExtendAsync, LockQueryAsync, LockAcquireAsync with IAsyncDisposable auto-release (#638) diff --git a/src/aspnet/StackExchange.Redis.Extensions.AspNetCore/Caching/RedisDistributedCache.cs b/src/aspnet/StackExchange.Redis.Extensions.AspNetCore/Caching/RedisDistributedCache.cs index 9223d15f..a9e8ba05 100644 --- a/src/aspnet/StackExchange.Redis.Extensions.AspNetCore/Caching/RedisDistributedCache.cs +++ b/src/aspnet/StackExchange.Redis.Extensions.AspNetCore/Caching/RedisDistributedCache.cs @@ -22,6 +22,11 @@ internal sealed class RedisDistributedCache : IDistributedCache private static readonly RedisValue AbsoluteExpirationField = "absexp"; private static readonly RedisValue SlidingExpirationField = "sldexp"; + // Cached field arrays: a collection expression at the call-site would allocate a new array on every Get/Refresh. + // SE.Redis never mutates the array (same pattern used by Microsoft.Extensions.Caching.StackExchangeRedis). + private static readonly RedisValue[] AllFields = [DataField, AbsoluteExpirationField, SlidingExpirationField]; + private static readonly RedisValue[] MetadataFields = [AbsoluteExpirationField, SlidingExpirationField]; + private readonly IDatabase db; /// @@ -110,12 +115,22 @@ public async Task SetAsync(string key, byte[] value, DistributedCacheEntryOption new(SlidingExpirationField, slidingTicks), }; - await db.HashSetAsync(key, fields).ConfigureAwait(false); - var expiry = GetExpirationTimeout(absoluteExpiration, options.SlidingExpiration); if (expiry.HasValue) - await db.KeyExpireAsync(key, expiry.Value).ConfigureAwait(false); + { + // A batch flushes HSET + EXPIRE in a single network write instead of two sequential round-trips. + var batch = db.CreateBatch(); + var setTask = batch.HashSetAsync(key, fields); + var expireTask = batch.KeyExpireAsync(key, expiry.Value); + batch.Execute(); + + await Task.WhenAll(setTask, expireTask).ConfigureAwait(false); + } + else + { + await db.HashSetAsync(key, fields).ConfigureAwait(false); + } } /// @@ -163,9 +178,9 @@ public async Task RefreshAsync(string key, CancellationToken token = default) RedisValue[] results; if (getData) - results = db.HashGet(key, [DataField, AbsoluteExpirationField, SlidingExpirationField]); + results = db.HashGet(key, AllFields); else - results = db.HashGet(key, [AbsoluteExpirationField, SlidingExpirationField]); + results = db.HashGet(key, MetadataFields); if (results[0].IsNull) return null; @@ -181,9 +196,9 @@ public async Task RefreshAsync(string key, CancellationToken token = default) RedisValue[] results; if (getData) - results = await db.HashGetAsync(key, [DataField, AbsoluteExpirationField, SlidingExpirationField]).ConfigureAwait(false); + results = await db.HashGetAsync(key, AllFields).ConfigureAwait(false); else - results = await db.HashGetAsync(key, [AbsoluteExpirationField, SlidingExpirationField]).ConfigureAwait(false); + results = await db.HashGetAsync(key, MetadataFields).ConfigureAwait(false); if (results[0].IsNull) return null; diff --git a/src/compressors/StackExchange.Redis.Extensions.Compression.Brotli/BrotliCompressor.cs b/src/compressors/StackExchange.Redis.Extensions.Compression.Brotli/BrotliCompressor.cs index e2a2b9da..cf096df8 100644 --- a/src/compressors/StackExchange.Redis.Extensions.Compression.Brotli/BrotliCompressor.cs +++ b/src/compressors/StackExchange.Redis.Extensions.Compression.Brotli/BrotliCompressor.cs @@ -1,5 +1,7 @@ // Copyright (c) Ugo Lattanzi. All Rights Reserved. Licensed under the MIT license. See License.txt in the project root for license information. +using System; +using System.Buffers; using System.IO; using System.IO.Compression; @@ -11,6 +13,8 @@ namespace StackExchange.Redis.Extensions.Core; /// public class BrotliCompressor : ICompressor { + private const int BrotliDefaultWindow = 22; + private readonly CompressionLevel compressionLevel; /// @@ -25,12 +29,22 @@ public BrotliCompressor(CompressionLevel compressionLevel = CompressionLevel.Fas /// public byte[] Compress(byte[] data) { - using var output = new MemoryStream(); + // The one-shot encoder avoids the BrotliStream state machine and the MemoryStream growth copies: + // a single rented worst-case buffer plus the exact-size result array. + var maxLength = BrotliEncoder.GetMaxCompressedLength(data.Length); + var buffer = ArrayPool.Shared.Rent(maxLength); - using (var brotli = new BrotliStream(output, compressionLevel)) - brotli.Write(data, 0, data.Length); + try + { + if (!BrotliEncoder.TryCompress(data, buffer, out var written, GetQuality(compressionLevel), BrotliDefaultWindow)) + throw new InvalidOperationException("Brotli compression failed."); - return output.ToArray(); + return buffer.AsSpan(0, written).ToArray(); + } + finally + { + ArrayPool.Shared.Return(buffer); + } } /// @@ -38,10 +52,21 @@ public byte[] Decompress(byte[] compressedData) { using var input = new MemoryStream(compressedData); using var brotli = new BrotliStream(input, CompressionMode.Decompress); - using var output = new MemoryStream(); + + // Pre-sized with a typical-ratio heuristic to avoid the growth copies of an empty MemoryStream. + using var output = new MemoryStream(compressedData.Length * 4); brotli.CopyTo(output); return output.ToArray(); } + + // Same CompressionLevel-to-quality mapping the runtime uses internally for BrotliStream. + private static int GetQuality(CompressionLevel level) => level switch + { + CompressionLevel.NoCompression => 0, + CompressionLevel.Fastest => 1, + CompressionLevel.SmallestSize => 11, + _ => 4, + }; } diff --git a/src/compressors/StackExchange.Redis.Extensions.Compression.GZip/GZipCompressor.cs b/src/compressors/StackExchange.Redis.Extensions.Compression.GZip/GZipCompressor.cs index 4d83c578..ee9191c3 100644 --- a/src/compressors/StackExchange.Redis.Extensions.Compression.GZip/GZipCompressor.cs +++ b/src/compressors/StackExchange.Redis.Extensions.Compression.GZip/GZipCompressor.cs @@ -25,7 +25,8 @@ public GZipCompressor(CompressionLevel compressionLevel = CompressionLevel.Faste /// public byte[] Compress(byte[] data) { - using var output = new MemoryStream(); + // Pre-sized to avoid the growth copies of an empty MemoryStream (the +64 covers the GZip header on tiny payloads). + using var output = new MemoryStream((data.Length / 2) + 64); using (var gzip = new GZipStream(output, compressionLevel)) gzip.Write(data, 0, data.Length); @@ -38,7 +39,9 @@ public byte[] Decompress(byte[] compressedData) { using var input = new MemoryStream(compressedData); using var gzip = new GZipStream(input, CompressionMode.Decompress); - using var output = new MemoryStream(); + + // Pre-sized with a typical-ratio heuristic to avoid the growth copies of an empty MemoryStream. + using var output = new MemoryStream(compressedData.Length * 3); gzip.CopyTo(output); diff --git a/src/compressors/StackExchange.Redis.Extensions.Compression.Snappier/SnappierCompressor.cs b/src/compressors/StackExchange.Redis.Extensions.Compression.Snappier/SnappierCompressor.cs index 92f07425..79e28f1e 100644 --- a/src/compressors/StackExchange.Redis.Extensions.Compression.Snappier/SnappierCompressor.cs +++ b/src/compressors/StackExchange.Redis.Extensions.Compression.Snappier/SnappierCompressor.cs @@ -1,6 +1,7 @@ // Copyright (c) Ugo Lattanzi. All Rights Reserved. Licensed under the MIT license. See License.txt in the project root for license information. using System; +using System.Buffers; using Snappier; @@ -15,11 +16,20 @@ public class SnappierCompressor : ICompressor /// public byte[] Compress(byte[] data) { + // The worst-case buffer (~1.17x input) is only transient: renting it avoids a heap (or LOH) allocation per call. var maxLength = Snappy.GetMaxCompressedLength(data.Length); - var buffer = new byte[maxLength]; - var compressedLength = Snappy.Compress(data, buffer); + var buffer = ArrayPool.Shared.Rent(maxLength); - return buffer.AsSpan(0, compressedLength).ToArray(); + try + { + var compressedLength = Snappy.Compress(data, buffer); + + return buffer.AsSpan(0, compressedLength).ToArray(); + } + finally + { + ArrayPool.Shared.Return(buffer); + } } /// @@ -29,6 +39,8 @@ public byte[] Decompress(byte[] compressedData) var buffer = new byte[decompressedLength]; var actualLength = Snappy.Decompress(compressedData, buffer); - return buffer.AsSpan(0, actualLength).ToArray(); + // The header-declared length matches the actual output, so the buffer can be returned as-is + // instead of being copied a second time. + return actualLength == decompressedLength ? buffer : buffer.AsSpan(0, actualLength).ToArray(); } } diff --git a/src/compressors/StackExchange.Redis.Extensions.Compression.ZstdSharp/ZstdSharpCompressor.cs b/src/compressors/StackExchange.Redis.Extensions.Compression.ZstdSharp/ZstdSharpCompressor.cs index 5865131f..56252ed2 100644 --- a/src/compressors/StackExchange.Redis.Extensions.Compression.ZstdSharp/ZstdSharpCompressor.cs +++ b/src/compressors/StackExchange.Redis.Extensions.Compression.ZstdSharp/ZstdSharpCompressor.cs @@ -1,5 +1,9 @@ // Copyright (c) Ugo Lattanzi. All Rights Reserved. Licensed under the MIT license. See License.txt in the project root for license information. +using System; +using System.Buffers; +using System.Threading; + using ZstdSharp; namespace StackExchange.Redis.Extensions.Core; @@ -10,7 +14,10 @@ namespace StackExchange.Redis.Extensions.Core; /// public class ZstdSharpCompressor : ICompressor { - private readonly int compressionLevel; + // Zstd contexts are designed for reuse and are expensive to initialize, but they are not thread-safe: + // one context per thread replaces the previous new-context-per-call pattern. + private readonly ThreadLocal compressor; + private readonly ThreadLocal decompressor = new(static () => new Decompressor()); /// /// Initializes a new instance of the class. @@ -18,22 +25,39 @@ public class ZstdSharpCompressor : ICompressor /// The Zstd compression level (1-22). Defaults to 3 (fast). public ZstdSharpCompressor(int compressionLevel = 3) { - this.compressionLevel = compressionLevel; + compressor = new(() => new Compressor(compressionLevel)); } /// public byte[] Compress(byte[] data) { - using var compressor = new Compressor(compressionLevel); - - return compressor.Wrap(data).ToArray(); + var bound = Compressor.GetCompressBound(data.Length); + var buffer = ArrayPool.Shared.Rent(bound); + + try + { + var written = compressor.Value!.Wrap(data, buffer); + + return buffer.AsSpan(0, written).ToArray(); + } + finally + { + ArrayPool.Shared.Return(buffer); + } } /// public byte[] Decompress(byte[] compressedData) { - using var decompressor = new Decompressor(); + var size = Decompressor.GetDecompressedSize(compressedData); + + // Frames produced by Wrap always carry the content size; the fallback covers foreign frames without it. + if (size == 0 || size > int.MaxValue) + return decompressor.Value!.Unwrap(compressedData).ToArray(); + + var result = new byte[(int)size]; + var written = decompressor.Value!.Unwrap(compressedData, result); - return decompressor.Unwrap(compressedData).ToArray(); + return written == result.Length ? result : result.AsSpan(0, written).ToArray(); } } diff --git a/src/core/StackExchange.Redis.Extensions.Core/Extensions/SpanExtensions.cs b/src/core/StackExchange.Redis.Extensions.Core/Extensions/SpanExtensions.cs index 4b5d5c8f..d83f1137 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Extensions/SpanExtensions.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Extensions/SpanExtensions.cs @@ -48,17 +48,6 @@ public static void EnumerateLines(this ReadOnlySpan span, ref List(this ReadOnlySpan span, Predicate condition) - { - for (var i = 0; i < span.Length; i++) - { - if (condition(span[i])) - return true; - } - - return false; - } - public static TResult[] ToFastArray(this ReadOnlySpan span, Func action) { if (span.IsEmpty) diff --git a/src/core/StackExchange.Redis.Extensions.Core/Extensions/ValueLengthExtensions.cs b/src/core/StackExchange.Redis.Extensions.Core/Extensions/ValueLengthExtensions.cs index 9a6628ea..e96967da 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Extensions/ValueLengthExtensions.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Extensions/ValueLengthExtensions.cs @@ -7,19 +7,23 @@ namespace StackExchange.Redis.Extensions.Core.Extensions; internal static class ValueLengthExtensions { - public static IEnumerable> OfValueInListSize(this IEnumerable> items, ISerializer serializer, uint maxValueLength) + public static KeyValuePair[] ToRedisEntries(this Tuple[] items, ISerializer serializer, uint maxValueLength) { - using var iterator = items.GetEnumerator(); + var result = new KeyValuePair[items.Length]; + var count = 0; - while (iterator.MoveNext()) + foreach (var item in items) { - if (iterator.Current != null) - { - yield return new( - iterator.Current.Item1, - iterator.Current.Item2.SerializeItem(serializer).CheckLength(maxValueLength, iterator.Current.Item1)); - } + if (item == null) + continue; + + result[count++] = new(item.Item1, item.Item2.SerializeItem(serializer).CheckLength(maxValueLength, item.Item1)); } + + if (count != result.Length) + Array.Resize(ref result, count); + + return result; } public static byte[] OfValueSize(this T? value, ISerializer serializer, uint maxValueLength, string key) @@ -43,33 +47,4 @@ private static byte[] CheckLength(this byte[] byteArray, uint maxValueLength, st return byteArray; } - - public static TSource MinBy( - this IEnumerable source, - Func selector, - IComparer? comparer = null) - { - comparer ??= Comparer.Default; - - using var sourceIterator = source.GetEnumerator(); - - if (!sourceIterator.MoveNext()) - throw new InvalidOperationException("Sequence contains no elements"); - - var min = sourceIterator.Current; - var minKey = selector(min); - - while (sourceIterator.MoveNext()) - { - var candidate = sourceIterator.Current; - var candidateProjected = selector(candidate); - if (comparer.Compare(candidateProjected, minKey) < 0) - { - min = candidate; - minKey = candidateProjected; - } - } - - return min; - } } diff --git a/src/core/StackExchange.Redis.Extensions.Core/Helpers/ExceptionThrowHelper.cs b/src/core/StackExchange.Redis.Extensions.Core/Helpers/ExceptionThrowHelper.cs index 39af4617..25ec6243 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Helpers/ExceptionThrowHelper.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Helpers/ExceptionThrowHelper.cs @@ -1,15 +1,16 @@ using System; using System.Diagnostics.CodeAnalysis; -using StackExchange.Redis.Extensions.Core.Extensions; - namespace StackExchange.Redis.Extensions.Core.Helpers; internal static class ExceptionThrowHelper { public static void ThrowIfExistsNullElement(ReadOnlySpan argument, string paramName) { - if (argument.Any(x => x is null)) - ThrowNullElementException(paramName); + foreach (var item in argument) + { + if (item is null) + ThrowNullElementException(paramName); + } } [DoesNotReturn] diff --git a/src/core/StackExchange.Redis.Extensions.Core/Helpers/GenericsExtensions.cs b/src/core/StackExchange.Redis.Extensions.Core/Helpers/GenericsExtensions.cs index 253bea84..f3bea6aa 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Helpers/GenericsExtensions.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Helpers/GenericsExtensions.cs @@ -1,36 +1,13 @@ using System; using System.Collections.Generic; -using System.Runtime.CompilerServices; +#if NET8_0_OR_GREATER using System.Runtime.InteropServices; +#endif namespace StackExchange.Redis.Extensions.Core.Helpers; internal static class GenericsExtensions { - public static void FastIteration(this ICollection? request, Action action) - { - if (request == null) - return; - - if (request is TSource[] sourceArray) - { - ref var searchSpace = ref MemoryMarshal.GetReference(sourceArray.AsSpan()); - - for (var i = 0; i < sourceArray.Length; i++) - { - ref var r = ref Unsafe.Add(ref searchSpace, i); - - action.Invoke(r, i); - } - } - else - { - var i = 0; - foreach (var r in request) - action.Invoke(r, i++); - } - } - public static TResult[] ToFastArray(this TSource[]? source, Func action) { if (source is not { Length: > 0 }) @@ -48,16 +25,37 @@ public static TResult[] ToFastArray(this ICollection? if (source is null) return []; + if (source is TSource[] sourceArray) + return sourceArray.ToFastArray(action); + +#if NET8_0_OR_GREATER + // Iterating a List through ICollection would box its struct enumerator; the span avoids both + // the boxing and the per-item interface dispatch. + if (source is List sourceList) + { + var span = CollectionsMarshal.AsSpan(sourceList); + + if (span.Length == 0) + return []; + + var listResult = new TResult[span.Length]; + for (var i = 0; i < span.Length; i++) + listResult[i] = action.Invoke(span[i]); + + return listResult; + } +#endif + var srcCnt = source.Count; if (srcCnt == 0) return []; var result = new TResult[srcCnt]; - var i = 0; + var i2 = 0; foreach (var item in source) { - result[i] = action.Invoke(item); - i++; + result[i2] = action.Invoke(item); + i2++; } return result; diff --git a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisClient.cs b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisClient.cs index 270aae03..8194595a 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisClient.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisClient.cs @@ -1,5 +1,8 @@ // Copyright (c) Ugo Lattanzi. All Rights Reserved. Licensed under the MIT license. See License.txt in the project root for license information. +using System; +using System.Threading; + using Microsoft.Extensions.Logging; using StackExchange.Redis.Extensions.Core.Abstractions; @@ -13,6 +16,11 @@ public class RedisClient : IRedisClient private readonly RedisConfiguration redisConfiguration; private readonly ILogger? databaseLogger; + // RedisDatabase is immutable and stateless (the pooled connection is resolved per operation), + // so instances can be cached per database number instead of being allocated on every Db0..Db16 access. + // The prefix is stored alongside because RedisConfiguration.KeyPrefix is mutable at runtime. + private readonly CachedDatabase?[] databaseCache = new CachedDatabase?[17]; + /// /// Initializes a new instance of the class. /// @@ -90,7 +98,25 @@ public IRedisDatabase GetDb(int dbNumber, string? keyPrefix = null) if (string.IsNullOrEmpty(keyPrefix)) keyPrefix = redisConfiguration.KeyPrefix; - return new RedisDatabase( + if ((uint)dbNumber >= (uint)databaseCache.Length) + return CreateDatabase(dbNumber, keyPrefix); + + var cached = Volatile.Read(ref databaseCache[dbNumber]); + + if (cached != null && string.Equals(cached.KeyPrefix, keyPrefix, StringComparison.Ordinal)) + return cached.Database; + + var created = CreateDatabase(dbNumber, keyPrefix); + + // A benign race may create an extra instance; that matches the previous allocate-per-call behavior. + Volatile.Write(ref databaseCache[dbNumber], new CachedDatabase(created, keyPrefix)); + + return created; + } + + private RedisDatabase CreateDatabase(int dbNumber, string? keyPrefix) + { + return new( ConnectionPoolManager, Serializer, redisConfiguration.ServerEnumerationStrategy, @@ -111,4 +137,11 @@ public IRedisDatabase GetDefaultDatabase() /// public string Name => redisConfiguration.Name ?? IRedisClient.DefaultName; + + private sealed class CachedDatabase(IRedisDatabase database, string? keyPrefix) + { + public IRedisDatabase Database { get; } = database; + + public string? KeyPrefix { get; } = keyPrefix; + } } diff --git a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisConnectionPoolManager.cs b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisConnectionPoolManager.cs index 200ac750..d6bbbc8d 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisConnectionPoolManager.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisConnectionPoolManager.cs @@ -5,7 +5,7 @@ using System.ComponentModel; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; -using System.Security.Cryptography; +using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; @@ -24,10 +24,10 @@ namespace StackExchange.Redis.Extensions.Core.Implementations; /// public sealed partial class RedisConnectionPoolManager : IRedisConnectionPoolManager { - private static readonly object @lock = new(); private readonly IStateAwareConnection[] connections; private readonly RedisConfiguration redisConfiguration; private readonly ILogger logger; + private int roundRobinIndex = -1; private bool isDisposed; /// @@ -41,14 +41,11 @@ public RedisConnectionPoolManager(RedisConfiguration redisConfiguration, ILogger logger ??= redisConfiguration.LoggerFactory?.CreateLogger(); this.logger = logger ?? NullLogger.Instance; - lock (@lock) - { - connections = new IStateAwareConnection[redisConfiguration.PoolSize]; + connections = new IStateAwareConnection[redisConfiguration.PoolSize]; #pragma warning disable VSTHRD002 // Synchronous wait is required here because constructors cannot be async - EmitConnectionsAsync().GetAwaiter().GetResult(); + EmitConnectionsAsync().GetAwaiter().GetResult(); #pragma warning restore VSTHRD002 - } } /// @@ -82,12 +79,8 @@ public IConnectionMultiplexer GetConnection() switch (redisConfiguration.ConnectionSelectionStrategy) { case ConnectionSelectionStrategy.RoundRobin: - var nextIdx -#if NET6_0_OR_GREATER - = RandomNumberGenerator.GetInt32(0, redisConfiguration.PoolSize); -#else - = new Random().Next(0, redisConfiguration.PoolSize); -#endif + // Casting to uint handles the int wraparound: the modulo stays valid across overflow. + var nextIdx = (int)((uint)Interlocked.Increment(ref roundRobinIndex) % (uint)connections.Length); connection = connections[nextIdx]; if (!connection.IsConnected()) @@ -106,19 +99,40 @@ var nextIdx break; case ConnectionSelectionStrategy.LeastLoaded: - // Prefer connected connections; fall back to any if all are disconnected + // Prefer connected connections; fall back to any if all are disconnected. + // TotalOutstanding() allocates a ServerCounters snapshot in SE.Redis, so it must be called at most once per connection. IStateAwareConnection? candidate = null; + var candidateOutstanding = long.MaxValue; for (var i = 0; i < connections.Length; i++) { if (!connections[i].IsConnected()) continue; - if (candidate == null || connections[i].TotalOutstanding() < candidate.TotalOutstanding()) + var outstanding = connections[i].TotalOutstanding(); + + if (outstanding < candidateOutstanding) + { candidate = connections[i]; + candidateOutstanding = outstanding; + } + } + + if (candidate == null) + { + for (var i = 0; i < connections.Length; i++) + { + var outstanding = connections[i].TotalOutstanding(); + + if (outstanding < candidateOutstanding) + { + candidate = connections[i]; + candidateOutstanding = outstanding; + } + } } - connection = candidate ?? connections.MinBy(x => x.TotalOutstanding()); + connection = candidate!; break; default: @@ -128,7 +142,9 @@ var nextIdx if (!connection.IsConnected()) LogMessages.AllConnectionsDisconnected(logger, connection.Connection.GetHashCode()); - LogMessages.ConnectionSelected(logger, connection.Connection.GetHashCode(), connection.TotalOutstanding()); + // Guarded because TotalOutstanding() allocates: log arguments are evaluated at the call-site even when the level is disabled. + if (logger.IsEnabled(LogLevel.Debug)) + LogMessages.ConnectionSelected(logger, connection.Connection.GetHashCode(), connection.TotalOutstanding()); return connection.Connection; } diff --git a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Hash.cs b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Hash.cs index c194e594..b57007b9 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Hash.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Hash.cs @@ -3,8 +3,6 @@ using System; using System.Collections.Generic; using System.Linq; -using System.Runtime.CompilerServices; -using System.Runtime.InteropServices; using System.Threading.Tasks; using StackExchange.Redis.Extensions.Core.Helpers; @@ -47,73 +45,42 @@ public Task HashExistsAsync(string hashKey, string key, CommandFlags flag /// public async Task> HashGetAsync(string hashKey, string[] keys, CommandFlags flag = CommandFlags.None) { -#if NET6_0_OR_GREATER - var concurrent = new System.Collections.Concurrent.ConcurrentDictionary(); + // A single HMGET replaces N parallel HGET round-trips. + var fields = keys.ToFastArray(key => (RedisValue)key); - await Parallel.ForEachAsync(keys, async (key, _) => - { - var result = await HashGetAsync(hashKey, key, flag); - concurrent.TryAdd(key, result); - }) - .ConfigureAwait(false); + var values = await Database.HashGetAsync(hashKey, fields, flag).ConfigureAwait(false); - return concurrent; -#else - var tasks = new Task[keys.Length]; + var result = new Dictionary(keys.Length, StringComparer.Ordinal); for (var i = 0; i < keys.Length; i++) - tasks[i] = HashGetAsync(hashKey, keys[i], flag); - - await Task.WhenAll(tasks).ConfigureAwait(false); - - var result = new Dictionary(); - - ref var searchSpace = ref MemoryMarshal.GetReference(tasks.AsSpan()); - - for (var i = 0; i < tasks.Length; i++) { - ref var task = ref Unsafe.Add(ref searchSpace, i); - result.Add(keys[i], task.Result); + var value = values[i]; + result[keys[i]] = value.HasValue ? Serializer.Deserialize(value) : default; } return result; -#endif } /// public async Task> HashGetAllAsyncAtOneTimeAsync(string hashKey, string[] keys, CommandFlags flag = CommandFlags.None) { - var luascript = "local results = {};local insert = table.insert;local rcall = redis.call;for i=1,table.getn(KEYS) do local value = rcall('HGET','" + hashKey + "', KEYS[i]) if value then insert(results, KEYS[i]) insert(results, value) end end; return results;"; - - var redisKeys = keys.ToFastArray(key => new RedisKey(key)); - - var data = await Database.ScriptEvaluateAsync(luascript, redisKeys, flags: flag).ConfigureAwait(false); + // A single HMGET replaces the previous per-call Lua script. Going through HashGetAsync also means the + // KeyPrefix is applied to the hash key (the script embedded it unprefixed, and wrongly prefixed the + // hash fields passed as KEYS) and the hash key is no longer exposed to Lua injection. + var fields = keys.ToFastArray(key => (RedisValue)key); - var dictionary = new Dictionary(); + var values = await Database.HashGetAsync(hashKey, fields, flag).ConfigureAwait(false); - var redisValues = (RedisValue[]?)data; + var dictionary = new Dictionary(keys.Length, StringComparer.Ordinal); - ref var searchSpaceRedisValue = ref MemoryMarshal.GetReference(redisValues.AsSpan()); - - if (redisValues is not { Length: > 0 }) - return dictionary; - - for (var i = 0; i < redisValues.Length; i += 2) + for (var i = 0; i < keys.Length; i++) { - ref var key = ref Unsafe.Add(ref searchSpaceRedisValue, i); - - if (!key.HasValue) - continue; - - var redisValue = redisValues[i + 1]; + var value = values[i]; - if (!redisValue.HasValue) + if (!value.HasValue) continue; -#pragma warning disable CS8604 // Possible null reference argument. - var value = Serializer.Deserialize(redisValue); - dictionary.Add(key, value); -#pragma warning restore CS8604 // Possible null reference argument. + dictionary.Add(keys[i], Serializer.Deserialize(value)); } return dictionary; @@ -144,7 +111,8 @@ public Task HashIncrementByAsync(string hashKey, string key, double valu /// public async Task> HashKeysAsync(string hashKey, CommandFlags flag = CommandFlags.None) { - return (await Database.HashKeysAsync(hashKey, flag).ConfigureAwait(false)).Select(x => x.ToString()); + // Materialized eagerly: a lazy Select would re-allocate every string on each enumeration. + return (await Database.HashKeysAsync(hashKey, flag).ConfigureAwait(false)).ToFastArray(x => x.ToString()); } /// diff --git a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.List.cs b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.List.cs index dfab39d6..7bfc0c17 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.List.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.List.cs @@ -60,9 +60,6 @@ public Task ListAddToLeftAsync(string key, T[] items, CommandFlags flag var item = await Database.ListRightPopAsync(key, flag).ConfigureAwait(false); - if (item == RedisValue.Null) - return default; - return item == RedisValue.Null ? default : Serializer.Deserialize(item); diff --git a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.PubSub.cs b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.PubSub.cs index 8bd7f0b3..68cf1bf3 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.PubSub.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.PubSub.cs @@ -2,7 +2,6 @@ using System; using System.Collections.Generic; -using System.Threading; using System.Threading.Tasks; using StackExchange.Redis.Extensions.Core.Helpers; @@ -35,17 +34,22 @@ public Task SubscribeAsync(RedisChannel channel, Func handler, Comm void Handler(RedisChannel redisChannel, RedisValue value) { - var task = Task.Run(async () => + // Task.Run keeps user handlers off the SE.Redis callback thread; the try/catch replaces a + // ContinueWith continuation that was allocated per message even on the success path. + _ = Task.Run(async () => { - var deserialized = Serializer.Deserialize(value); - await handler(deserialized).ConfigureAwait(false); + try + { + var deserialized = Serializer.Deserialize(value); + await handler(deserialized).ConfigureAwait(false); + } +#pragma warning disable CA1031 // User handlers can throw anything; a failed message must never tear down the subscription. + catch (Exception ex) +#pragma warning restore CA1031 + { + LogMessages.SubscriptionHandlerError(logger, ex, (string?)redisChannel ?? "unknown"); + } }); - - task.ContinueWith( - t => LogMessages.SubscriptionHandlerError(logger, t.Exception, (string?)redisChannel ?? "unknown"), - CancellationToken.None, - TaskContinuationOptions.OnlyOnFaulted, - TaskScheduler.Default); } } @@ -79,33 +83,34 @@ public async Task UnsubscribeAllAsync(CommandFlags flag = CommandFlags.None) } /// - public async Task UpdateExpiryAsync(string key, DateTimeOffset expiresAt, CommandFlags flag = CommandFlags.None) + public Task UpdateExpiryAsync(string key, DateTimeOffset expiresAt, CommandFlags flag = CommandFlags.None) { - if (await Database.KeyExistsAsync(key).ConfigureAwait(false)) - return await Database.KeyExpireAsync(key, expiresAt.UtcDateTime.Subtract(DateTime.UtcNow), flag).ConfigureAwait(false); - - return false; + // EXPIRE already returns false on missing keys: the previous EXISTS pre-check doubled the round-trips + // without adding correctness (the two commands were not atomic anyway). + return Database.KeyExpireAsync(key, expiresAt.UtcDateTime.Subtract(DateTime.UtcNow), flag); } /// - public async Task UpdateExpiryAsync(string key, TimeSpan expiresIn, CommandFlags flag = CommandFlags.None) + public Task UpdateExpiryAsync(string key, TimeSpan expiresIn, CommandFlags flag = CommandFlags.None) { - if (await Database.KeyExistsAsync(key).ConfigureAwait(false)) - return await Database.KeyExpireAsync(key, expiresIn, flag).ConfigureAwait(false); - - return false; + return Database.KeyExpireAsync(key, expiresIn, flag); } /// public async Task> UpdateExpiryAllAsync(HashSet keys, DateTimeOffset expiresAt, CommandFlags flag = CommandFlags.None) { - var tasks = keys.ToFastArray(key => UpdateExpiryAsync(key, expiresAt.UtcDateTime, flag)); + // Computed once: the previous per-key computation re-read DateTime.UtcNow for every key, drifting the TTL. + var expiresIn = expiresAt.UtcDateTime.Subtract(DateTime.UtcNow); + + var tasks = keys.ToFastArray(key => UpdateExpiryAsync(key, expiresIn, flag)); await Task.WhenAll(tasks).ConfigureAwait(false); var results = new Dictionary(keys.Count, StringComparer.Ordinal); + var i = 0; - keys.FastIteration((key, i) => results.Add(key, tasks[i].Result)); + foreach (var key in keys) + results.Add(key, tasks[i++].Result); return results; } @@ -118,8 +123,10 @@ public async Task> UpdateExpiryAllAsync(HashSet(keys.Count, StringComparer.Ordinal); + var i = 0; - keys.FastIteration((key, i) => results.Add(key, tasks[i].Result)); + foreach (var key in keys) + results.Add(key, tasks[i++].Result); return results; } diff --git a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Sort.cs b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Sort.cs index 574a39d3..a0d1e0d8 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Sort.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Sort.cs @@ -47,7 +47,8 @@ public Task SortedSetRemoveAsync( { var result = await Database.SortedSetRangeByScoreAsync(key, start, stop, exclude, order, skip, take, flag).ConfigureAwait(false); - return result.Select(m => m == RedisValue.Null ? default : Serializer.Deserialize(m)); + // Materialized eagerly: a lazy Select would re-deserialize every element on each enumeration. + return result.ToFastArray(m => m == RedisValue.Null ? default : Serializer.Deserialize(m)); } /// diff --git a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Tags.cs b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Tags.cs index f6ef3f64..1fe9c65e 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Tags.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.Tags.cs @@ -2,7 +2,6 @@ using System; using System.Collections.Generic; -using System.Linq; using System.Threading.Tasks; using StackExchange.Redis.Extensions.Core.Extensions; @@ -22,9 +21,10 @@ public partial class RedisDatabase if (keys.Length == 0) return []; - var hashKeys = new HashSet(); + var hashKeys = new HashSet(keys.Length, StringComparer.Ordinal); - keys.FastIteration((key, _) => hashKeys.Add(key)); + foreach (var key in keys) + hashKeys.Add(key!); var result = await GetAllAsync(hashKeys, flag).ConfigureAwait(false); @@ -56,8 +56,11 @@ private Task ExecuteAddWithTagsAsync( TryAddCondition(transaction, when, key); - foreach (var tagKey in tags.Select(TagHelper.GenerateTagKey)) - transaction.SetAddAsync(tagKey, key.OfValueSize(Serializer, maxValueLength, tagKey), commandFlags); + // Serialized once: the previous per-tag call produced N identical byte arrays for the same key. + var serializedKey = key.OfValueSize(Serializer, maxValueLength, key); + + foreach (var tag in tags) + transaction.SetAddAsync(TagHelper.GenerateTagKey(tag), serializedKey, commandFlags); action(transaction); diff --git a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.cs b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.cs index cc081999..ecc7b42d 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/Implementations/RedisDatabase.cs @@ -109,10 +109,13 @@ public Task RemoveAllAsync(string[] keys, CommandFlags flag = CommandFlags /// public async Task GetAsync(string key, DateTimeOffset expiresAt, CommandFlags flag = CommandFlags.None) { - var result = await GetAsync(key, flag).ConfigureAwait(false); + // Database resolves a pooled connection on every access: read it once for the two commands. + var db = Database; + var valueBytes = await db.StringGetAsync(key, flag).ConfigureAwait(false); + var result = !valueBytes.HasValue ? default : Serializer.Deserialize(valueBytes); if (!EqualityComparer.Default.Equals(result, default)) - await Database.KeyExpireAsync(key, expiresAt.UtcDateTime.Subtract(DateTime.UtcNow)).ConfigureAwait(false); + await db.KeyExpireAsync(key, expiresAt.UtcDateTime.Subtract(DateTime.UtcNow)).ConfigureAwait(false); return result; } @@ -120,10 +123,12 @@ public Task RemoveAllAsync(string[] keys, CommandFlags flag = CommandFlags /// public async Task GetAsync(string key, TimeSpan expiresIn, CommandFlags flag = CommandFlags.None) { - var result = await GetAsync(key, flag).ConfigureAwait(false); + var db = Database; + var valueBytes = await db.StringGetAsync(key, flag).ConfigureAwait(false); + var result = !valueBytes.HasValue ? default : Serializer.Deserialize(valueBytes); if (!EqualityComparer.Default.Equals(result, default)) - await Database.KeyExpireAsync(key, expiresIn).ConfigureAwait(false); + await db.KeyExpireAsync(key, expiresIn).ConfigureAwait(false); return result; } @@ -226,10 +231,7 @@ public Task ReplaceAsync(string key, T value, TimeSpan expiresIn, When /// public Task AddAllAsync(Tuple[] items, When when = When.Always, CommandFlags flag = CommandFlags.None) { - var values = items - .OfValueInListSize(Serializer, maxValueLength) - .Select(x => new KeyValuePair(x.Key, x.Value)) - .ToArray(); + var values = items.ToRedisEntries(Serializer, maxValueLength); return Database.StringSetAsync(values, when, flag); } @@ -237,10 +239,7 @@ public Task AddAllAsync(Tuple[] items, When when = When.Alwa /// public async Task AddAllAsync(Tuple[] items, DateTimeOffset expiresAt, When when = When.Always, CommandFlags flag = CommandFlags.None) { - var values = items - .OfValueInListSize(Serializer, maxValueLength) - .Select(x => new KeyValuePair(x.Key, x.Value)) - .ToArray(); + var values = items.ToRedisEntries(Serializer, maxValueLength); if (values.Length == 0) return false; @@ -265,10 +264,7 @@ public async Task AddAllAsync(Tuple[] items, DateTimeOffset /// public async Task AddAllAsync(Tuple[] items, TimeSpan expiresAt, When when = When.Always, CommandFlags flag = CommandFlags.None) { - var values = items - .OfValueInListSize(Serializer, maxValueLength) - .Select(x => new KeyValuePair(x.Key, x.Value)) - .ToArray(); + var values = items.ToRedisEntries(Serializer, maxValueLength); if (values.Length == 0) return false; @@ -333,7 +329,8 @@ public Task SetAddAsync(string key, T item, CommandFlags flag = Command var items = await Database.SetPopAsync(key, count, flag).ConfigureAwait(false); - return items.Select(item => item == RedisValue.Null ? default : Serializer.Deserialize(item)); + // Materialized eagerly: a lazy Select would re-deserialize every element on each enumeration. + return items.ToFastArray(item => item == RedisValue.Null ? default : Serializer.Deserialize(item)); } /// @@ -511,22 +508,25 @@ public async Task> SearchKeysAsync(string pattern) { pattern = $"{keyPrefix}{pattern}"; var keys = new HashSet(); + var hasPrefix = !string.IsNullOrEmpty(keyPrefix); foreach (var server in ServerIteratorFactory.GetServers(connectionPoolManager.GetConnection(), serverEnumerationStrategy)) { + // The prefix is stripped while filling: a lazy Select would re-allocate every substring on each enumeration. await foreach (var key in server.KeysAsync(dbNumber, pattern, 1000).ConfigureAwait(false)) - keys.Add(key!); + keys.Add(hasPrefix ? ((string)key!)[keyPrefix.Length..] : key!); } - return !string.IsNullOrEmpty(keyPrefix) - ? keys.Select(k => k.ToString()[keyPrefix.Length..]) - : keys.Select(k => k.ToString()); + return keys; } /// public Task FlushDbAsync() { - var endPoints = Database.Multiplexer.GetEndPoints(); + // Database resolves a pooled connection and allocates the key-prefixed wrapper on every access: read it once. + var db = Database; + var multiplexer = db.Multiplexer; + var endPoints = multiplexer.GetEndPoints(); var tasks = new List(endPoints.Length); @@ -536,10 +536,10 @@ public Task FlushDbAsync() { ref var endpoint = ref Unsafe.Add(ref searchSpace, i); - var server = Database.Multiplexer.GetServer(endpoint); + var server = multiplexer.GetServer(endpoint); if (!server.IsReplica) - tasks.Add(server.FlushDatabaseAsync(Database.Database)); + tasks.Add(server.FlushDatabaseAsync(db.Database)); } return Task.WhenAll(tasks); @@ -548,9 +548,10 @@ public Task FlushDbAsync() /// public Task SaveAsync(SaveType saveType, CommandFlags flag = CommandFlags.None) { - var endPoints = Database.Multiplexer.GetEndPoints(); + var multiplexer = Database.Multiplexer; + var endPoints = multiplexer.GetEndPoints(); - var tasks = endPoints.ToFastArray(endpoint => Database.Multiplexer.GetServer(endpoint).SaveAsync(saveType, flag)); + var tasks = endPoints.ToFastArray(endpoint => multiplexer.GetServer(endpoint).SaveAsync(saveType, flag)); return Task.WhenAll(tasks); } @@ -621,9 +622,10 @@ private static Dictionary ParseInfo(string info) // Return a dictionary of the Info Key and Info value - var result = new Dictionary(); + var result = new Dictionary(data.Length); - data.FastIteration((x, _) => result.TryAdd(x.Key, x.InfoValue)); + foreach (var detail in data) + result.TryAdd(detail.Key, detail.InfoValue); return result; } diff --git a/src/core/StackExchange.Redis.Extensions.Core/ServerIteration/ServerIteratorFactory.cs b/src/core/StackExchange.Redis.Extensions.Core/ServerIteration/ServerIteratorFactory.cs index 31e727a5..0f1b0bdf 100644 --- a/src/core/StackExchange.Redis.Extensions.Core/ServerIteration/ServerIteratorFactory.cs +++ b/src/core/StackExchange.Redis.Extensions.Core/ServerIteration/ServerIteratorFactory.cs @@ -2,7 +2,6 @@ using System; using System.Collections.Generic; -using System.Linq; using StackExchange.Redis.Extensions.Core.Configuration; @@ -37,10 +36,19 @@ public static IEnumerable GetServers( serverEnumerationStrategy.TargetRole, serverEnumerationStrategy.UnreachableServerAction); - return serversSingle.Take(1); + return TakeFirst(serversSingle); default: throw new NotImplementedException(); } } + + private static IEnumerable TakeFirst(ServerEnumerable servers) + { + foreach (var server in servers) + { + yield return server; + yield break; + } + } } diff --git a/src/serializers/StackExchange.Redis.Extensions.MsgPack/MsgPackObjectSerializer.cs b/src/serializers/StackExchange.Redis.Extensions.MsgPack/MsgPackObjectSerializer.cs index a44f885d..eb29eb48 100644 --- a/src/serializers/StackExchange.Redis.Extensions.MsgPack/MsgPackObjectSerializer.cs +++ b/src/serializers/StackExchange.Redis.Extensions.MsgPack/MsgPackObjectSerializer.cs @@ -1,73 +1,68 @@ -// Copyright (c) Ugo Lattanzi. All Rights Reserved. Licensed under the MIT license. See License.txt in the project root for license information. - -using System; -using System.Globalization; -using System.IO; -using System.Text; - -using MsgPack.Serialization; - -using StackExchange.Redis.Extensions.Core; - -namespace StackExchange.Redis.Extensions.MsgPack; - -/// -/// MsgPac implementation of -/// -public class MsgPackObjectSerializer : ISerializer -{ - private readonly Encoding encoding; - - /// - /// Initializes a new instance of the class. - /// - public MsgPackObjectSerializer() - : this(null) - { - } - - /// - /// Initializes a new instance of the class. - /// - public MsgPackObjectSerializer(Action? customSerializerRegistrar = null, Encoding? encoding = null) - { - customSerializerRegistrar?.Invoke(SerializationContext.Default.Serializers); - - encoding ??= Encoding.UTF8; - - this.encoding = encoding; - } - - /// - public T? Deserialize(byte[]? serializedObject) - { - if (serializedObject == null) - return default; - - if (typeof(T) == typeof(string)) - return (T)Convert.ChangeType(encoding.GetString(serializedObject), typeof(T), CultureInfo.InvariantCulture); - - var serializer = MessagePackSerializer.Get(); - - using var byteStream = new MemoryStream(serializedObject); - - return serializer.Unpack(byteStream); - } - - /// - public byte[] Serialize(T? item) - { - if (item is null) - return []; - - if (item is string str) - return encoding.GetBytes(str); - - var serializer = MessagePackSerializer.Get(item.GetType()); - - using var byteStream = new MemoryStream(); - serializer.Pack(byteStream, item); - - return byteStream.ToArray(); - } -} +// Copyright (c) Ugo Lattanzi. All Rights Reserved. Licensed under the MIT license. See License.txt in the project root for license information. + +using System; +using System.Text; + +using MsgPack.Serialization; + +using StackExchange.Redis.Extensions.Core; + +namespace StackExchange.Redis.Extensions.MsgPack; + +/// +/// MsgPac implementation of +/// +public class MsgPackObjectSerializer : ISerializer +{ + private readonly Encoding encoding; + + /// + /// Initializes a new instance of the class. + /// + public MsgPackObjectSerializer() + : this(null) + { + } + + /// + /// Initializes a new instance of the class. + /// + public MsgPackObjectSerializer(Action? customSerializerRegistrar = null, Encoding? encoding = null) + { + customSerializerRegistrar?.Invoke(SerializationContext.Default.Serializers); + + encoding ??= Encoding.UTF8; + + this.encoding = encoding; + } + + /// + public T? Deserialize(byte[]? serializedObject) + { + if (serializedObject == null) + return default; + + if (typeof(T) == typeof(string)) + return (T)(object)encoding.GetString(serializedObject); + + var serializer = MessagePackSerializer.Get(); + + // UnpackSingleObject reads the array directly, without a wrapping MemoryStream. + return serializer.UnpackSingleObject(serializedObject); + } + + /// + public byte[] Serialize(T? item) + { + if (item is null) + return []; + + if (item is string str) + return encoding.GetBytes(str); + + var serializer = MessagePackSerializer.Get(item.GetType()); + + // PackSingleObject returns the buffer directly, without MemoryStream growth plus final copy. + return serializer.PackSingleObject(item); + } +} diff --git a/src/serializers/StackExchange.Redis.Extensions.Newtonsoft/NewtonsoftSerializer.cs b/src/serializers/StackExchange.Redis.Extensions.Newtonsoft/NewtonsoftSerializer.cs index 44dab7fd..cfc4a6d0 100644 --- a/src/serializers/StackExchange.Redis.Extensions.Newtonsoft/NewtonsoftSerializer.cs +++ b/src/serializers/StackExchange.Redis.Extensions.Newtonsoft/NewtonsoftSerializer.cs @@ -1,59 +1,95 @@ -// Copyright (c) Ugo Lattanzi. All Rights Reserved. Licensed under the MIT license. See License.txt in the project root for license information. - -using System.Text; - -using Newtonsoft.Json; - -using StackExchange.Redis.Extensions.Core; - -namespace StackExchange.Redis.Extensions.Newtonsoft; - -/// -/// JSon.Net implementation of -/// -/// -/// Initializes a new instance of the class. -/// -/// The settings. -public class NewtonsoftSerializer(JsonSerializerSettings? settings) : ISerializer -{ - /// - /// Encoding to use to convert string to byte[] and the other way around. - /// - /// - /// StackExchange.Redis uses Encoding.UTF8 to convert strings to bytes, - /// hence we do same here. - /// - private static readonly Encoding encoding = Encoding.UTF8; - - private readonly JsonSerializerSettings settings = settings ?? new(); - - /// - /// Initializes a new instance of the class. - /// - public NewtonsoftSerializer() - : this(null) - { - } - - /// - public byte[] Serialize(T? item) - { - if (item == null) - return []; - - var type = item.GetType(); - var jsonString = JsonConvert.SerializeObject(item, type, settings); - return encoding.GetBytes(jsonString); - } - - /// - public T? Deserialize(byte[]? serializedObject) - { - if (serializedObject == null) - return default; - - var jsonString = encoding.GetString(serializedObject); - return JsonConvert.DeserializeObject(jsonString, settings); - } -} +// Copyright (c) Ugo Lattanzi. All Rights Reserved. Licensed under the MIT license. See License.txt in the project root for license information. + +using System.Buffers; +using System.IO; +using System.Text; + +using Newtonsoft.Json; + +using StackExchange.Redis.Extensions.Core; + +namespace StackExchange.Redis.Extensions.Newtonsoft; + +/// +/// JSon.Net implementation of +/// +/// +/// Initializes a new instance of the class. +/// +/// The settings. +public class NewtonsoftSerializer(JsonSerializerSettings? settings) : ISerializer +{ + /// + /// Encoding to use to convert string to byte[] and the other way around. + /// + /// + /// StackExchange.Redis uses Encoding.UTF8 to convert strings to bytes, + /// hence we do same here. The BOM is suppressed to keep the on-wire bytes + /// identical to the previous Encoding.UTF8.GetBytes-based implementation. + /// + private static readonly UTF8Encoding encoding = new(false); + + // Writing straight to a stream avoids materializing every payload as an intermediate UTF-16 string, + // and the cached serializers avoid the JsonSerializer.CreateDefault call JsonConvert performs per invocation. + private readonly JsonSerializer writeSerializer = JsonSerializer.CreateDefault(settings); + + // JsonConvert.DeserializeObject enables CheckAdditionalContent unless explicitly configured: + // a dedicated instance preserves that behavior on the read path. + private readonly JsonSerializer readSerializer = CreateReadSerializer(settings); + + /// + /// Initializes a new instance of the class. + /// + public NewtonsoftSerializer() + : this(null) + { + } + + /// + public byte[] Serialize(T? item) + { + if (item == null) + return []; + + using var ms = new MemoryStream(256); + + using (var streamWriter = new StreamWriter(ms, encoding, 1024, leaveOpen: true)) + using (var jsonWriter = new JsonTextWriter(streamWriter) { ArrayPool = JsonCharArrayPool.Instance }) + writeSerializer.Serialize(jsonWriter, item, item.GetType()); + + return ms.ToArray(); + } + + /// + public T? Deserialize(byte[]? serializedObject) + { + if (serializedObject == null) + return default; + + using var ms = new MemoryStream(serializedObject, writable: false); + using var streamReader = new StreamReader(ms, encoding); + using var jsonReader = new JsonTextReader(streamReader) { ArrayPool = JsonCharArrayPool.Instance }; + + return readSerializer.Deserialize(jsonReader); + } + + private static JsonSerializer CreateReadSerializer(JsonSerializerSettings? settings) + { + var serializer = JsonSerializer.CreateDefault(settings); + serializer.CheckAdditionalContent = true; + return serializer; + } + + private sealed class JsonCharArrayPool : IArrayPool + { + public static readonly JsonCharArrayPool Instance = new(); + + public char[] Rent(int minimumLength) => ArrayPool.Shared.Rent(minimumLength); + + public void Return(char[]? array) + { + if (array != null) + ArrayPool.Shared.Return(array); + } + } +} diff --git a/src/serializers/StackExchange.Redis.Extensions.Protobuf/ProtobufSerializer.cs b/src/serializers/StackExchange.Redis.Extensions.Protobuf/ProtobufSerializer.cs index 78aa0dca..38ac66b4 100644 --- a/src/serializers/StackExchange.Redis.Extensions.Protobuf/ProtobufSerializer.cs +++ b/src/serializers/StackExchange.Redis.Extensions.Protobuf/ProtobufSerializer.cs @@ -1,6 +1,7 @@ // Copyright (c) Ugo Lattanzi. All Rights Reserved. Licensed under the MIT license. See License.txt in the project root for license information. -using System.IO; +using System; +using System.Buffers; using ProtoBuf; @@ -19,11 +20,12 @@ public byte[] Serialize(T? item) if (item == null) return []; - using var ms = new MemoryStream(); + // IBufferWriter avoids the MemoryStream layer; the final ToArray is the only full copy. + var buffer = new ArrayBufferWriter(256); - Serializer.Serialize(ms, item); + Serializer.Serialize(buffer, item); - return ms.ToArray(); + return buffer.WrittenSpan.ToArray(); } /// @@ -32,8 +34,7 @@ public byte[] Serialize(T? item) if (serializedObject == null) return default; - using var ms = new MemoryStream(serializedObject); - - return Serializer.Deserialize(ms); + // The span-based overload reads the array directly, without a wrapping MemoryStream. + return Serializer.Deserialize((ReadOnlyMemory)serializedObject); } } diff --git a/src/serializers/StackExchange.Redis.Extensions.ServiceStack/ServiceStackJsonSerializer.cs b/src/serializers/StackExchange.Redis.Extensions.ServiceStack/ServiceStackJsonSerializer.cs index 96ec3751..b8aa806d 100644 --- a/src/serializers/StackExchange.Redis.Extensions.ServiceStack/ServiceStackJsonSerializer.cs +++ b/src/serializers/StackExchange.Redis.Extensions.ServiceStack/ServiceStackJsonSerializer.cs @@ -1,5 +1,7 @@ // Copyright (c) Ugo Lattanzi. All Rights Reserved. Licensed under the MIT license. See License.txt in the project root for license information. +using System.IO; + using ServiceStack.Text; using StackExchange.Redis.Extensions.Core; @@ -34,8 +36,9 @@ public ServiceStackJsonSerializer() if (serializedObject == null) return default; - var json = JsConfig.UTF8Encoding.GetString(serializedObject); - return JsonSerializer.DeserializeFromString(json); + // The stream API avoids materializing the payload as an intermediate UTF-16 string. + using var ms = new MemoryStream(serializedObject, writable: false); + return JsonSerializer.DeserializeFromStream(ms); } /// @@ -44,7 +47,8 @@ public byte[] Serialize(T? item) if (item == null) return []; - var json = JsonSerializer.SerializeToString(item); - return JsConfig.UTF8Encoding.GetBytes(json); + using var ms = new MemoryStream(256); + JsonSerializer.SerializeToStream(item, ms); + return ms.ToArray(); } } diff --git a/src/serializers/StackExchange.Redis.Extensions.Utf8Json/Utf8JsonSerializer.cs b/src/serializers/StackExchange.Redis.Extensions.Utf8Json/Utf8JsonSerializer.cs index 676761bb..c9eeccc9 100644 --- a/src/serializers/StackExchange.Redis.Extensions.Utf8Json/Utf8JsonSerializer.cs +++ b/src/serializers/StackExchange.Redis.Extensions.Utf8Json/Utf8JsonSerializer.cs @@ -22,6 +22,10 @@ public byte[] Serialize(T? item) /// public T? Deserialize(byte[]? serializedObject) { + // Null-guard consistent with every other serializer: Utf8Json would throw on a null buffer. + if (serializedObject == null) + return default; + return JsonSerializer.Deserialize(serializedObject); } }