Skip to content

S
StackExchange.Redis
Commit c6af6cf0 authored by Marc Gravell's avatar Marc Gravell
Browse files
Options

fully working now; renamed to ChannelMessageQueue

parent 1c2f2834
Hide whitespace changes
Inline Side-by-side
Showing with 116 additions and 175 deletions
StackExchange.Redis.Tests/BasicOps.cs
View file @ c6af6cf0
......@@ -301,22 +301,19 @@ public void TestQuit()
Assert.Throws<RedisConnectionException>(() => db.Ping());
watch.Stop();
Output.WriteLine("Time to notice quit: {0}ms ({1})", watch.ElapsedMilliseconds,
preserveOrder ? "preserve order" : "any order");
"any order");
Thread.Sleep(20);
Debug.WriteLine("Pinging...");
Assert.Equal(key, (string)db.StringGet(key));
}
}
[Theory]
[InlineData(true)]
[InlineData(false)]
public async Task TestSevered(bool preserveOrder)
[Fact]
public async Task TestSevered()
{
SetExpectedAmbientFailureCount(2);
using (var muxer = Create(allowAdmin: true))
{
muxer.PreserveAsyncOrder = preserveOrder;
var db = muxer.GetDatabase();
string key = Guid.NewGuid().ToString();
db.KeyDelete(key, CommandFlags.FireAndForget);
......@@ -326,7 +323,7 @@ public async Task TestSevered(bool preserveOrder)
db.Ping();
watch.Stop();
Output.WriteLine("Time to re-establish: {0}ms ({1})", watch.ElapsedMilliseconds,
preserveOrder ? "preserve order" : "any order");
"any order");
await Task.Delay(2000).ForAwait();
Debug.WriteLine("Pinging...");
Assert.Equal(key, db.StringGet(key));
......
StackExchange.Redis.Tests/Booksleeve/PubSub.cs
View file @ c6af6cf0
......@@ -135,8 +135,7 @@ public async Task PubSubGetAllCorrectOrder()
const int count = 500000;
var syncLock = new object();
var dataList = new List<int>(count);
var dataHash = new HashSet<int>();
var data = new List<int>(count);
var subChannel = await sub.SubscribeAsync(channel);
await sub.PingAsync();
......@@ -147,12 +146,11 @@ async Task RunLoop()
{
var work = await subChannel.ReadAsync();
int i = int.Parse(Encoding.UTF8.GetString(work.Value));
lock (dataList)
lock (data)
{
dataList.Add(i);
dataHash.Add(i);
if (dataList.Count == count) break;
if ((dataList.Count % 10) == 99) Output.WriteLine(dataList.Count.ToString());
data.Add(i);
if (data.Count == count) break;
if ((data.Count % 10) == 99) Output.WriteLine(data.Count.ToString());
}
}
lock (syncLock)
......@@ -172,12 +170,11 @@ async Task RunLoop()
// subChannel.Unsubscribe();
if (!Monitor.Wait(syncLock, 20000))
{
throw new TimeoutException("Items: " + dataList.Count);
throw new TimeoutException("Items: " + data.Count);
}
for (int i = 0; i < count; i++)
{
Assert.Contains(i, dataHash);
// Assert.Equal(i, data[i]);
Assert.Equal(i, data[i]);
}
}
}
......
StackExchange.Redis.Tests/TaskTests.cs deleted 100644 → 0
View file @ 1c2f2834
using System;
using System.Threading;
using System.Threading.Tasks;
using Xunit;
namespace StackExchange.Redis.Tests
{
public class TaskTests
{
#if DEBUG
[Theory]
[InlineData(SourceOrign.NewTCS)]
[InlineData(SourceOrign.Create)]
public void VerifyIsSyncSafe(SourceOrign origin)
{
var source = Create<int>(origin);
// Yes this looks stupid, but it's the proper pattern for how we statically init now
// ...and if we're dropping NET45 support, we can just nuke it all.
#if NET462
Assert.True(TaskSource.IsSyncSafe(source.Task));
#elif NETCOREAPP2_0
Assert.True(TaskSource.IsSyncSafe(source.Task));
#endif
}
private static TaskCompletionSource<T> Create<T>(SourceOrign origin)
{
switch (origin)
{
case SourceOrign.NewTCS: return new TaskCompletionSource<T>();
case SourceOrign.Create: return TaskSource.Create<T>(null);
default: throw new ArgumentOutOfRangeException(nameof(origin));
}
}
[Theory]
// regular framework behaviour: 2 out of 3 cause hijack
[InlineData(SourceOrign.NewTCS, AttachMode.ContinueWith, true)]
[InlineData(SourceOrign.NewTCS, AttachMode.ContinueWithExecSync, false)]
[InlineData(SourceOrign.NewTCS, AttachMode.Await, true)]
// Create is just a wrapper of ^^^; expect the same
[InlineData(SourceOrign.Create, AttachMode.ContinueWith, true)]
[InlineData(SourceOrign.Create, AttachMode.ContinueWithExecSync, false)]
[InlineData(SourceOrign.Create, AttachMode.Await, true)]
public void TestContinuationHijacking(SourceOrign origin, AttachMode attachMode, bool expectHijack)
{
TaskCompletionSource<int> source = Create<int>(origin);
int settingThread = Environment.CurrentManagedThreadId;
var state = new AwaitState();
state.Attach(source.Task, attachMode);
source.TrySetResult(123);
state.Wait(); // waits for the continuation to run
int from = state.Thread;
Assert.NotEqual(-1, from); // not set
if (expectHijack)
{
Assert.True(settingThread != from, $"expected hijack; didn't happen, Origin={settingThread}, Final={from}");
}
else
{
Assert.True(settingThread == from, $"setter was hijacked, Origin={settingThread}, Final={from}");
}
}
public enum SourceOrign
{
NewTCS,
Create
}
public enum AttachMode
{
ContinueWith,
ContinueWithExecSync,
Await
}
private class AwaitState
{
public int Thread => continuationThread;
private volatile int continuationThread = -1;
private readonly ManualResetEventSlim evt = new ManualResetEventSlim();
public void Wait()
{
if (!evt.Wait(5000)) throw new TimeoutException();
}
public void Attach(Task task, AttachMode attachMode)
{
switch (attachMode)
{
case AttachMode.ContinueWith:
task.ContinueWith(Continue);
break;
case AttachMode.ContinueWithExecSync:
task.ContinueWith(Continue, TaskContinuationOptions.ExecuteSynchronously);
break;
case AttachMode.Await:
DoAwait(task);
break;
default:
throw new ArgumentOutOfRangeException(nameof(attachMode));
}
}
private void Continue(Task task)
{
continuationThread = Environment.CurrentManagedThreadId;
evt.Set();
}
private async void DoAwait(Task task)
{
await task.ConfigureAwait(false);
continuationThread = Environment.CurrentManagedThreadId;
evt.Set();
}
}
#endif
}
}
StackExchange.Redis/StackExchange/Redis/ChannelMessage.cs → StackExchange.Redis/StackExchange/Redis/ChannelMessageQueue.cs
View file @ c6af6cf0
......@@ -30,7 +30,7 @@ internal ChannelMessage(RedisChannel channel, RedisValue value)
/// <summary>
/// Represents a message queue of pub/sub notifications
/// </summary>
public sealed class ChannelMessageChannel
public sealed class ChannelMessageQueue
{
private readonly Channel<ChannelMessage> _channel;
private readonly RedisChannel _redisChannel;
......@@ -41,19 +41,36 @@ public sealed class ChannelMessageChannel
/// </summary>
public bool IsComplete { get; private set; }
internal ChannelMessageChannel(RedisChannel redisChannel, ISubscriber parent)
internal ChannelMessageQueue(RedisChannel redisChannel, ISubscriber parent)
{
_redisChannel = redisChannel;
_parent = parent;
_channel = Channel.CreateUnbounded<ChannelMessage>();
_channel = Channel.CreateUnbounded<ChannelMessage>(s_ChannelOptions);
_channel.Reader.Completion.ContinueWith(
(t, state) => ((ChannelMessageChannel)state).IsComplete = true, this, TaskContinuationOptions.ExecuteSynchronously);
(t, state) => ((ChannelMessageQueue)state).IsComplete = true, this, TaskContinuationOptions.ExecuteSynchronously);
}
static readonly UnboundedChannelOptions s_ChannelOptions = new UnboundedChannelOptions
{
SingleWriter = true,
SingleReader = false,
AllowSynchronousContinuations = false,
};
internal void Subscribe(CommandFlags flags) => _parent.Subscribe(_redisChannel, HandleMessage, flags);
internal Task SubscribeAsync(CommandFlags flags) => _parent.SubscribeAsync(_redisChannel, HandleMessage, flags);
private void HandleMessage(RedisChannel channel, RedisValue value)
=> _channel.Writer.TryWrite(new ChannelMessage(channel, value));
{
var writer = _channel.Writer;
if (channel.IsNull && value.IsNull) // see ForSyncShutdown
{
writer.TryComplete();
}
else
{
writer.TryWrite(new ChannelMessage(channel, value));
}
}
/// <summary>
/// Consume a message from the channel
......@@ -82,6 +99,18 @@ internal async Task UnsubscribeAsyncImpl(Exception error = null, CommandFlags fl
}
}
internal static bool IsOneOf(Action<RedisChannel, RedisValue> handler)
{
try
{
return handler != null && handler.Target is ChannelMessageQueue
&& handler.Method.Name == nameof(HandleMessage);
} catch
{
return false;
}
}
/// <summary>
/// Stop receiving messages on this channel
/// </summary>
......
StackExchange.Redis/StackExchange/Redis/Interfaces/ISubscriber.cs
View file @ c6af6cf0
......@@ -73,7 +73,7 @@ public interface ISubscriber : IRedis
/// <returns>A channel that represents this source</returns>
/// <remarks>https://redis.io/commands/subscribe</remarks>
/// <remarks>https://redis.io/commands/psubscribe</remarks>
ChannelMessageChannel Subscribe(RedisChannel channel, CommandFlags flags = CommandFlags.None);
ChannelMessageQueue Subscribe(RedisChannel channel, CommandFlags flags = CommandFlags.None);
/// <summary>
/// Subscribe to perform some operation when a change to the preferred/active node is broadcast.
......@@ -93,7 +93,7 @@ public interface ISubscriber : IRedis
/// <returns>A channel that represents this source</returns>
/// <remarks>https://redis.io/commands/subscribe</remarks>
/// <remarks>https://redis.io/commands/psubscribe</remarks>
Task<ChannelMessageChannel> SubscribeAsync(RedisChannel channel, CommandFlags flags = CommandFlags.None);
Task<ChannelMessageQueue> SubscribeAsync(RedisChannel channel, CommandFlags flags = CommandFlags.None);
/// <summary>
/// Inidicate to which redis server we are actively subscribed for a given channel; returns null if
......
StackExchange.Redis/StackExchange/Redis/MessageCompletable.cs
View file @ c6af6cf0
......@@ -7,35 +7,50 @@ internal sealed class MessageCompletable : ICompletable
{
private readonly RedisChannel channel;
private readonly Action<RedisChannel, RedisValue> handler;
private readonly Action<RedisChannel, RedisValue> syncHandler, asyncHandler;
private readonly RedisValue message;
public MessageCompletable(RedisChannel channel, RedisValue message, Action<RedisChannel, RedisValue> handler)
public MessageCompletable(RedisChannel channel, RedisValue message, Action<RedisChannel, RedisValue> syncHandler, Action<RedisChannel, RedisValue> asyncHandler)
{
this.channel = channel;
this.message = message;
this.handler = handler;
this.syncHandler = syncHandler;
this.asyncHandler = asyncHandler;
}
public override string ToString() => (string)channel;
public bool TryComplete(bool isAsync)
{
if (handler == null) return true;
if (isAsync)
{
ConnectionMultiplexer.TraceWithoutContext("Invoking...: " + (string)channel, "Subscription");
foreach(Action<RedisChannel, RedisValue> sub in handler.GetInvocationList())
if (asyncHandler != null)
{
try { sub.Invoke(channel, message); }
catch { }
ConnectionMultiplexer.TraceWithoutContext("Invoking (async)...: " + (string)channel, "Subscription");
foreach (Action<RedisChannel, RedisValue> sub in asyncHandler.GetInvocationList())
{
try { sub.Invoke(channel, message); }
catch { }
}
ConnectionMultiplexer.TraceWithoutContext("Invoke complete (async)", "Subscription");
}
ConnectionMultiplexer.TraceWithoutContext("Invoke complete", "Subscription");
return true;
}
// needs to be called async (unless there is nothing to do!)
return false;
else
{
if (syncHandler != null)
{
ConnectionMultiplexer.TraceWithoutContext("Invoking (sync)...: " + (string)channel, "Subscription");
foreach (Action<RedisChannel, RedisValue> sub in syncHandler.GetInvocationList())
{
try { sub.Invoke(channel, message); }
catch { }
}
ConnectionMultiplexer.TraceWithoutContext("Invoke complete (sync)", "Subscription");
}
return asyncHandler != null; // anything async to do?
}
}
void ICompletable.AppendStormLog(StringBuilder sb)
......
StackExchange.Redis/StackExchange/Redis/RedisSubscriber.cs
View file @ c6af6cf0
......@@ -31,15 +31,16 @@ internal Task AddSubscription(RedisChannel channel, Action<RedisChannel, RedisVa
{
if (handler != null)
{
bool asAsync = !ChannelMessageQueue.IsOneOf(handler);
lock (subscriptions)
{
if (subscriptions.TryGetValue(channel, out Subscription sub))
{
sub.Add(handler);
sub.Add(asAsync, handler);
}
else
{
sub = new Subscription(handler);
sub = new Subscription(asAsync, handler);
subscriptions.Add(channel, sub);
var task = sub.SubscribeToServer(this, channel, flags, asyncState, false);
if (task != null) return task;
......@@ -84,7 +85,14 @@ internal Task RemoveAllSubscriptions(CommandFlags flags, object asyncState)
{
foreach (var pair in subscriptions)
{
pair.Value.Remove(null); // always wipes
var msg = pair.Value.ForSyncShutdown();
if(msg != null)
{
// TODO: execute
}
pair.Value.Remove(true, null);
pair.Value.Remove(false, null);
var task = pair.Value.UnsubscribeFromServer(pair.Key, flags, asyncState, false);
if (task != null) last = task;
}
......@@ -97,7 +105,8 @@ internal Task RemoveSubscription(RedisChannel channel, Action<RedisChannel, Redi
{
lock (subscriptions)
{
if (subscriptions.TryGetValue(channel, out Subscription sub) && sub.Remove(handler))
bool asAsync = ChannelMessageQueue.IsOneOf(handler);
if (subscriptions.TryGetValue(channel, out Subscription sub) && sub.Remove(asAsync, handler))
{
subscriptions.Remove(channel);
var task = sub.UnsubscribeFromServer(channel, flags, asyncState, false);
......@@ -143,30 +152,46 @@ internal long ValidateSubscriptions()
private sealed class Subscription
{
private Action<RedisChannel, RedisValue> handler;
private Action<RedisChannel, RedisValue> _asyncHandler, _syncHandler;
private ServerEndPoint owner;
public Subscription(Action<RedisChannel, RedisValue> value) => handler = value;
public Subscription(bool asAsync, Action<RedisChannel, RedisValue> value)
{
if (asAsync) _asyncHandler = value;
else _syncHandler = value;
}
public void Add(Action<RedisChannel, RedisValue> value) => handler += value;
public void Add(bool asAsync, Action<RedisChannel, RedisValue> value)
{
if (asAsync) _asyncHandler += value;
else _syncHandler += value;
}
public ICompletable ForSyncShutdown()
{
var syncHandler = _syncHandler;
return syncHandler == null ? null : new MessageCompletable(default, default, syncHandler, null);
}
public ICompletable ForInvoke(RedisChannel channel, RedisValue message)
{
var tmp = handler;
return tmp == null ? null : new MessageCompletable(channel, message, tmp);
var syncHandler = _syncHandler;
var asyncHandler = _asyncHandler;
return (syncHandler == null && asyncHandler == null) ? null : new MessageCompletable(channel, message, syncHandler, asyncHandler);
}
public bool Remove(Action<RedisChannel, RedisValue> value)
public bool Remove(bool asAsync, Action<RedisChannel, RedisValue> value)
{
if (value == null)
{ // treat as blanket wipe
handler = null;
return true;
if (asAsync) _asyncHandler = null;
else _syncHandler = null;
}
else
{
return (handler -= value) == null;
if (asAsync) _asyncHandler -= value;
else _syncHandler -= value;
}
return _syncHandler == null && _asyncHandler == null;
}
public Task SubscribeToServer(ConnectionMultiplexer multiplexer, RedisChannel channel, CommandFlags flags, object asyncState, bool internalCall)
......@@ -287,9 +312,9 @@ public void Subscribe(RedisChannel channel, Action<RedisChannel, RedisValue> han
if ((flags & CommandFlags.FireAndForget) == 0) Wait(task);
}
public ChannelMessageChannel Subscribe(RedisChannel channel, CommandFlags flags = CommandFlags.None)
public ChannelMessageQueue Subscribe(RedisChannel channel, CommandFlags flags = CommandFlags.None)
{
var c = new ChannelMessageChannel(channel, this);
var c = new ChannelMessageQueue(channel, this);
c.Subscribe(flags);
return c;
}
......@@ -300,9 +325,9 @@ public Task SubscribeAsync(RedisChannel channel, Action<RedisChannel, RedisValue
return multiplexer.AddSubscription(channel, handler, flags, asyncState);
}
public async Task<ChannelMessageChannel> SubscribeAsync(RedisChannel channel, CommandFlags flags = CommandFlags.None)
public async Task<ChannelMessageQueue> SubscribeAsync(RedisChannel channel, CommandFlags flags = CommandFlags.None)
{
var c = new ChannelMessageChannel(channel, this);
var c = new ChannelMessageQueue(channel, this);
await c.SubscribeAsync(flags);
return c;
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment