Skip to content
This repository was archived by the owner on Dec 24, 2022. It is now read-only.

Commit e177a48

Browse files
mythzmythz
committed
Implement RedisConfig.AssertAccessOnlyOnSameThread
Throw when client is accessed from a different thread than the thread used to resolve the client from the pool from.
1 parent a11d835 commit e177a48

6 files changed

Lines changed: 165 additions & 9 deletions

File tree

‎src/ServiceStack.Redis/PooledRedisClientManager.cs‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ public partial class PooledRedisClientManager
4141
public int? SocketSendTimeout { get; set; }
4242
public int? SocketReceiveTimeout { get; set; }
4343
public int? IdleTimeOutSecs { get; set; }
44+
public bool AssertAccessOnlyOnSameThread { get; set; }
4445

4546
/// <summary>
4647
/// Gets or sets object key prefix.
@@ -140,6 +141,8 @@ public PooledRedisClientManager(
140141
? poolTimeOutSeconds * 1000
141142
: 2000; //Default Timeout
142143

144+
this.AssertAccessOnlyOnSameThread = RedisConfig.AssertAccessOnlyOnSameThread;
145+
143146
JsConfig.InitStatics();
144147

145148
if (this.Config.AutoStart)
@@ -244,7 +247,9 @@ public IRedisClient GetClient()
244247

245248
InitClient(inActiveClient);
246249

247-
return inActiveClient;
250+
return !AssertAccessOnlyOnSameThread
251+
? inActiveClient
252+
: inActiveClient.LimitAccessToThread(Thread.CurrentThread.ManagedThreadId, Environment.StackTrace);
248253
}
249254
}
250255

@@ -277,7 +282,10 @@ public IRedisClient GetClient()
277282

278283
WritePoolIndex++;
279284
writeClients[inactivePoolIndex] = newClient;
280-
return newClient;
285+
286+
return !AssertAccessOnlyOnSameThread
287+
? newClient
288+
: newClient.LimitAccessToThread(Thread.CurrentThread.ManagedThreadId, Environment.StackTrace);
281289
}
282290
}
283291
catch
@@ -525,6 +533,7 @@ public void DisposeClient(RedisNativeClient client)
525533
}
526534
else
527535
{
536+
client.TrackThread = null;
528537
client.Active = false;
529538
}
530539

@@ -545,6 +554,7 @@ public void DisposeClient(RedisNativeClient client)
545554
}
546555
else
547556
{
557+
client.TrackThread = null;
548558
client.Active = false;
549559
}
550560

‎src/ServiceStack.Redis/RedisClient.cs‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
using System.Collections.Generic;
1717
using System.Linq;
1818
using System.Text;
19+
using System.Threading;
1920
using ServiceStack.Redis.Generic;
2021
using ServiceStack.Redis.Pipeline;
2122
using ServiceStack.Text;
@@ -1087,7 +1088,30 @@ private static RedisServerRole ToServerRole(string roleName)
10871088
return RedisServerRole.Unknown;
10881089
}
10891090
}
1091+
1092+
internal RedisClient LimitAccessToThread(int originalThreadId, string originalStackTrace)
1093+
{
1094+
TrackThread = new TrackThread(originalThreadId, originalStackTrace);
1095+
return this;
1096+
}
1097+
}
1098+
1099+
internal struct TrackThread
1100+
{
1101+
public readonly int ThreadId;
1102+
public readonly string StackTrace;
10901103

1104+
public TrackThread(int threadId, string stackTrace)
1105+
{
1106+
ThreadId = threadId;
1107+
StackTrace = stackTrace;
1108+
}
1109+
}
1110+
1111+
public class InvalidAccessException : RedisException
1112+
{
1113+
public InvalidAccessException(int threadId, string stackTrace)
1114+
: base($"The Current Thread #{Thread.CurrentThread.ManagedThreadId} is different to the original Thread #{threadId} that resolved this pooled client at: \n{stackTrace}") { }
10911115
}
10921116

10931117
}

‎src/ServiceStack.Redis/RedisConfig.cs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,12 @@ public class RedisConfig
9797
public static LocalCertificateSelectionCallback CertificateSelectionCallback { get; set; }
9898
public static RemoteCertificateValidationCallback CertificateValidationCallback { get; set; }
9999

100+
/// <summary>
101+
/// Assert all access using pooled RedisClient instance should be limited to same thread.
102+
/// Captures StackTrace so is very slow, use only for debugging connection issues.
103+
/// </summary>
104+
public static bool AssertAccessOnlyOnSameThread = false;
105+
100106
/// <summary>
101107
/// Resets Redis Config and Redis Stats back to default values
102108
/// </summary>
@@ -119,6 +125,7 @@ public static void Reset()
119125
DisableVerboseLogging = false;
120126
CertificateSelectionCallback = null;
121127
CertificateValidationCallback = null;
128+
AssertAccessOnlyOnSameThread = false;
122129
}
123130
}
124131
}

‎src/ServiceStack.Redis/RedisManagerPool.cs‎

Lines changed: 15 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,8 @@ public partial class RedisManagerPool
6161

6262
public int MaxPoolSize { get; private set; }
6363

64+
public bool AssertAccessOnlyOnSameThread { get; set; }
65+
6466
public RedisManagerPool() : this(RedisConfig.DefaultHost) { }
6567
public RedisManagerPool(string host) : this(new[] { host }) { }
6668
public RedisManagerPool(string host, RedisPoolConfig config) : this(new[] { host }, config) { }
@@ -69,7 +71,7 @@ public RedisManagerPool(IEnumerable<string> hosts) : this(hosts, null) { }
6971
public RedisManagerPool(IEnumerable<string> hosts, RedisPoolConfig config)
7072
{
7173
if (hosts == null)
72-
throw new ArgumentNullException("hosts");
74+
throw new ArgumentNullException(nameof(hosts));
7375

7476
RedisResolver = new RedisResolver(hosts, null);
7577

@@ -83,6 +85,8 @@ public RedisManagerPool(IEnumerable<string> hosts, RedisPoolConfig config)
8385
clients = new RedisClient[MaxPoolSize];
8486
poolIndex = 0;
8587

88+
this.AssertAccessOnlyOnSameThread = RedisConfig.AssertAccessOnlyOnSameThread;
89+
8690
JsConfig.InitStatics();
8791
}
8892

@@ -123,7 +127,7 @@ public void FailoverTo(IEnumerable<string> readWriteHosts, IEnumerable<string> r
123127
{
124128
FailoverTo(readWriteHosts.ToArray()); //only use readWriteHosts
125129
}
126-
130+
127131
/// <summary>
128132
/// Returns a Read/Write client (The default) using the hosts defined in ReadWriteHosts
129133
/// </summary>
@@ -137,17 +141,18 @@ public IRedisClient GetClient()
137141
{
138142
AssertValidPool();
139143

140-
RedisClient inActiveClient;
141144
//-1 when no available clients otherwise index of reservedSlot or existing Client
142-
inactivePoolIndex = GetInActiveClient(out inActiveClient);
145+
inactivePoolIndex = GetInActiveClient(out var inActiveClient);
143146

144147
//inActiveClient != null only for Valid InActive Clients
145148
if (inActiveClient != null)
146149
{
147150
poolIndex++;
148151
inActiveClient.Active = true;
149152

150-
return inActiveClient;
153+
return !AssertAccessOnlyOnSameThread
154+
? inActiveClient
155+
: inActiveClient.LimitAccessToThread(Thread.CurrentThread.ManagedThreadId, Environment.StackTrace);
151156
}
152157
}
153158

@@ -188,7 +193,10 @@ public IRedisClient GetClient()
188193

189194
poolIndex++;
190195
clients[inactivePoolIndex] = newClient;
191-
return newClient;
196+
197+
return !AssertAccessOnlyOnSameThread
198+
? newClient
199+
: newClient.LimitAccessToThread(Thread.CurrentThread.ManagedThreadId, Environment.StackTrace);
192200
}
193201
}
194202
catch
@@ -291,6 +299,7 @@ public void DisposeClient(RedisNativeClient client)
291299
}
292300
else
293301
{
302+
client.TrackThread = null;
294303
client.Active = false;
295304
}
296305

‎src/ServiceStack.Redis/RedisNativeClient_Utils.cs‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -510,12 +510,20 @@ private int SafeReadByte()
510510
{
511511
return Bstream.ReadByte();
512512
}
513-
513+
514+
internal TrackThread? TrackThread;
515+
514516
protected T SendReceive<T>(byte[][] cmdWithBinaryArgs,
515517
Func<T> fn,
516518
Action<Func<T>> completePipelineFn = null,
517519
bool sendWithoutRead = false)
518520
{
521+
if (TrackThread != null)
522+
{
523+
if (TrackThread.Value.ThreadId != Thread.CurrentThread.ManagedThreadId)
524+
throw new InvalidAccessException(TrackThread.Value.ThreadId, TrackThread.Value.StackTrace);
525+
}
526+
519527
var i = 0;
520528
Exception originalEx = null;
521529

‎tests/ServiceStack.Redis.Tests/TrackThreadTests.cs‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
using System;
2+
using System.Threading;
3+
using NUnit.Framework;
4+
5+
namespace ServiceStack.Redis.Tests
6+
{
7+
public class TrackThreadTests
8+
{
9+
[Test]
10+
public void Does_throw_when_using_same_client_on_different_threads()
11+
{
12+
RedisConfig.AssertAccessOnlyOnSameThread = true;
13+
InvalidAccessException poolEx = null;
14+
15+
var redisManager = new RedisManagerPool();
16+
17+
using (var redis = redisManager.GetClient())
18+
{
19+
var threadId = Thread.CurrentThread.ManagedThreadId.ToString();
20+
var key = $"Thread#{threadId}";
21+
redis.SetValue(key, threadId);
22+
23+
ThreadPool.QueueUserWorkItem(_ =>
24+
{
25+
using (var poolRedis = redisManager.GetClient())
26+
{
27+
var poolThreadId = Thread.CurrentThread.ManagedThreadId.ToString();
28+
var poolKey = $"Thread#{poolThreadId}";
29+
poolRedis.SetValue(poolKey , poolThreadId);
30+
31+
Console.WriteLine("From Pool: " + poolRedis.GetValue(poolKey));
32+
33+
try
34+
{
35+
Console.WriteLine("From Pool (using TEST): " + redis.GetValue(poolKey));
36+
}
37+
catch (InvalidAccessException ex)
38+
{
39+
poolEx = ex;
40+
}
41+
}
42+
});
43+
44+
Thread.Sleep(100);
45+
46+
Console.WriteLine("From Test: " + redis.GetValue(key));
47+
48+
if (poolEx == null)
49+
throw new Exception("Should throw InvalidAccessException");
50+
51+
Console.WriteLine("InvalidAccessException: " + poolEx.Message);
52+
}
53+
54+
RedisConfig.AssertAccessOnlyOnSameThread = false;
55+
}
56+
57+
[Test]
58+
public void Does_not_throw_when_using_different_clients_on_same_Thread()
59+
{
60+
RedisConfig.AssertAccessOnlyOnSameThread = true;
61+
InvalidAccessException poolEx = null;
62+
63+
var redisManager = new RedisManagerPool();
64+
65+
using (var redis = redisManager.GetClient())
66+
{
67+
var threadId = Thread.CurrentThread.ManagedThreadId.ToString();
68+
var key = $"Thread#{threadId}";
69+
redis.SetValue(key, threadId);
70+
71+
ThreadPool.QueueUserWorkItem(_ =>
72+
{
73+
try
74+
{
75+
using (var poolRedis = redisManager.GetClient())
76+
{
77+
var poolThreadId = Thread.CurrentThread.ManagedThreadId.ToString();
78+
var poolKey = $"Thread#{poolThreadId}";
79+
poolRedis.SetValue(poolKey , poolThreadId);
80+
81+
Console.WriteLine("From Pool: " + poolRedis.GetValue(poolKey ));
82+
}
83+
}
84+
catch (InvalidAccessException ex)
85+
{
86+
poolEx = ex;
87+
}
88+
});
89+
90+
Thread.Sleep(100);
91+
92+
Console.WriteLine("From Test: " + redis.GetValue(key));
93+
}
94+
95+
RedisConfig.AssertAccessOnlyOnSameThread = false;
96+
}
97+
}
98+
}

0 commit comments

Comments
 (0)