Skip to content

S
StackExchange.Redis
Commit cf91c206 authored by Jon Cole's avatar Jon Cole
Browse files
Options

Merge branch 'master' of https://github.com/StackExchange/StackExchange.Redis

parents 0cd9148c 2bc2fb8f
Hide whitespace changes
Inline Side-by-side
Showing with 206 additions and 100 deletions
StackExchange.Redis.Tests/TestBase.cs
View file @ cf91c206
...@@ -243,7 +243,7 @@ private static RedisConnection GetOldStyleConnection(string host, int port, bool ...@@ -243,7 +243,7 @@ private static RedisConnection GetOldStyleConnection(string host, int port, bool
protected static TimeSpan RunConcurrent(Action work, int threads, int timeout = 10000, [CallerMemberName] string caller = null) protected static TimeSpan RunConcurrent(Action work, int threads, int timeout = 10000, [CallerMemberName] string caller = null)
{ {
if (work == null) throw new ArgumentNullException("work"); if (work == null) throw new ArgumentNullException("work");
if (threads < 1) throw new ArgumentOutOfRangeException("theads"); if (threads < 1) throw new ArgumentOutOfRangeException("threads");
if(string.IsNullOrWhiteSpace(caller)) caller = Me(); if(string.IsNullOrWhiteSpace(caller)) caller = Me();
Stopwatch watch = null; Stopwatch watch = null;
ManualResetEvent allDone = new ManualResetEvent(false); ManualResetEvent allDone = new ManualResetEvent(false);
......
StackExchange.Redis/StackExchange/Redis/ClusterConfiguration.cs
View file @ cf91c206
...@@ -172,6 +172,7 @@ public sealed class ClusterConfiguration ...@@ -172,6 +172,7 @@ public sealed class ClusterConfiguration
private readonly ServerSelectionStrategy serverSelectionStrategy; private readonly ServerSelectionStrategy serverSelectionStrategy;
internal ClusterConfiguration(ServerSelectionStrategy serverSelectionStrategy, string nodes, EndPoint origin) internal ClusterConfiguration(ServerSelectionStrategy serverSelectionStrategy, string nodes, EndPoint origin)
{ {
// Beware: Any exception thrown here will wreak silent havoc like inability to connect to cluster nodes or non returning calls
this.serverSelectionStrategy = serverSelectionStrategy; this.serverSelectionStrategy = serverSelectionStrategy;
this.origin = origin; this.origin = origin;
using (var reader = new StringReader(nodes)) using (var reader = new StringReader(nodes))
...@@ -180,9 +181,36 @@ internal ClusterConfiguration(ServerSelectionStrategy serverSelectionStrategy, s ...@@ -180,9 +181,36 @@ internal ClusterConfiguration(ServerSelectionStrategy serverSelectionStrategy, s
while ((line = reader.ReadLine()) != null) while ((line = reader.ReadLine()) != null)
{ {
if (string.IsNullOrWhiteSpace(line)) continue; if (string.IsNullOrWhiteSpace(line)) continue;
var node = new ClusterNode(this, line, origin); var node = new ClusterNode(this, line, origin);
nodeLookup.Add(node.EndPoint, node); // Be resilient to ":0 {master,slave},fail,noaddr" nodes
if (node.IsNoAddr)
continue;
if (nodeLookup.ContainsKey(node.EndPoint))
{
// Deal with conflicting node entries for the same endpoint
// This can happen in dynamic environments when a node goes down and a new one is created
// to replace it.
if (!node.IsConnected)
{
// The node we're trying to add is probably about to become stale. Ignore it.
continue;
}
else if (!nodeLookup[node.EndPoint].IsConnected)
{
// The node we registered previously is probably stale. Replace it with a known good node.
nodeLookup[node.EndPoint] = node;
}
else
{
// We have conflicting connected nodes. There's nothing much we can do other than
// wait for the cluster state to converge and refresh on the next pass.
// The same is true if we have multiple disconnected nodes.
}
}
else
{
nodeLookup.Add(node.EndPoint, node);
}
} }
} }
} }
...@@ -262,6 +290,10 @@ public sealed class ClusterNode : IEquatable<ClusterNode>, IComparable<ClusterN ...@@ -262,6 +290,10 @@ public sealed class ClusterNode : IEquatable<ClusterNode>, IComparable<ClusterN
private readonly bool isSlave; private readonly bool isSlave;
private readonly bool isNoAddr;
private readonly bool isConnected;
private readonly string nodeId, parentNodeId, raw; private readonly string nodeId, parentNodeId, raw;
private readonly IList<SlotRange> slots; private readonly IList<SlotRange> slots;
...@@ -275,22 +307,18 @@ public sealed class ClusterNode : IEquatable<ClusterNode>, IComparable<ClusterN ...@@ -275,22 +307,18 @@ public sealed class ClusterNode : IEquatable<ClusterNode>, IComparable<ClusterN
internal ClusterNode() { } internal ClusterNode() { }
internal ClusterNode(ClusterConfiguration configuration, string raw, EndPoint origin) internal ClusterNode(ClusterConfiguration configuration, string raw, EndPoint origin)
{ {
// http://redis.io/commands/cluster-nodes
this.configuration = configuration; this.configuration = configuration;
this.raw = raw; this.raw = raw;
var parts = raw.Split(StringSplits.Space); var parts = raw.Split(StringSplits.Space);
var flags = parts[2].Split(StringSplits.Comma); var flags = parts[2].Split(StringSplits.Comma);
if (flags.Contains("myself")) endpoint = Format.TryParseEndPoint(parts[1]);
{
endpoint = origin;
}
else
{
endpoint = Format.TryParseEndPoint(parts[1]);
}
nodeId = parts[0]; nodeId = parts[0];
isSlave = flags.Contains("slave"); isSlave = flags.Contains("slave");
isNoAddr = flags.Contains("noaddr");
parentNodeId = string.IsNullOrWhiteSpace(parts[3]) ? null : parts[3]; parentNodeId = string.IsNullOrWhiteSpace(parts[3]) ? null : parts[3];
List<SlotRange> slots = null; List<SlotRange> slots = null;
...@@ -305,6 +333,7 @@ internal ClusterNode(ClusterConfiguration configuration, string raw, EndPoint or ...@@ -305,6 +333,7 @@ internal ClusterNode(ClusterConfiguration configuration, string raw, EndPoint or
} }
} }
this.slots = slots == null ? NoSlots : slots.AsReadOnly(); this.slots = slots == null ? NoSlots : slots.AsReadOnly();
this.isConnected = parts[7] == "connected"; // Can be "connected" or "disconnected"
} }
/// <summary> /// <summary>
/// Gets all child nodes of the current node /// Gets all child nodes of the current node
...@@ -339,6 +368,16 @@ public IList<ClusterNode> Children ...@@ -339,6 +368,16 @@ public IList<ClusterNode> Children
/// </summary> /// </summary>
public bool IsSlave { get { return isSlave; } } public bool IsSlave { get { return isSlave; } }
/// <summary>
/// Gets whether this node is flagged as noaddr
/// </summary>
public bool IsNoAddr { get { return isNoAddr; } }
/// <summary>
/// Gets the node's connection status
/// </summary>
public bool IsConnected { get { return isConnected; } }
/// <summary> /// <summary>
/// Gets the unique node-id of the current node /// Gets the unique node-id of the current node
/// </summary> /// </summary>
......
StackExchange.Redis/StackExchange/Redis/CommandMap.cs
View file @ cf91c206
...@@ -33,7 +33,7 @@ private static readonly CommandMap ...@@ -33,7 +33,7 @@ private static readonly CommandMap
RedisCommand.SCRIPT, RedisCommand.SCRIPT,
RedisCommand.AUTH, RedisCommand.ECHO, RedisCommand.PING, RedisCommand.QUIT, RedisCommand.SELECT, RedisCommand.ECHO, RedisCommand.PING, RedisCommand.QUIT, RedisCommand.SELECT,
RedisCommand.BGREWRITEAOF, RedisCommand.BGSAVE, RedisCommand.CLIENT, RedisCommand.CLUSTER, RedisCommand.CONFIG, RedisCommand.DBSIZE, RedisCommand.BGREWRITEAOF, RedisCommand.BGSAVE, RedisCommand.CLIENT, RedisCommand.CLUSTER, RedisCommand.CONFIG, RedisCommand.DBSIZE,
RedisCommand.DEBUG, RedisCommand.FLUSHALL, RedisCommand.FLUSHDB, RedisCommand.INFO, RedisCommand.LASTSAVE, RedisCommand.MONITOR, RedisCommand.SAVE, RedisCommand.DEBUG, RedisCommand.FLUSHALL, RedisCommand.FLUSHDB, RedisCommand.INFO, RedisCommand.LASTSAVE, RedisCommand.MONITOR, RedisCommand.SAVE,
......
StackExchange.Redis/StackExchange/Redis/ConnectionMultiplexer.cs
View file @ cf91c206
...@@ -1203,114 +1203,155 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text ...@@ -1203,114 +1203,155 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text
} }
const CommandFlags flags = CommandFlags.NoRedirect | CommandFlags.HighPriority; const CommandFlags flags = CommandFlags.NoRedirect | CommandFlags.HighPriority;
var available = new Task<bool>[endpoints.Count]; List<ServerEndPoint> masters = new List<ServerEndPoint>(endpoints.Count);
var servers = new ServerEndPoint[available.Length];
bool useTieBreakers = !string.IsNullOrWhiteSpace(configuration.TieBreaker); bool useTieBreakers = !string.IsNullOrWhiteSpace(configuration.TieBreaker);
var tieBreakers = useTieBreakers ? new Task<string>[endpoints.Count] : null;
RedisKey tieBreakerKey = useTieBreakers ? (RedisKey)configuration.TieBreaker : default(RedisKey); ServerEndPoint[] servers = null;
for (int i = 0; i < available.Length; i++) Task<string>[] tieBreakers = null;
bool encounteredConnectedClusterServer = false;
Stopwatch watch = null;
int iterCount = first ? 2 : 1;
// this is fix for https://github.com/StackExchange/StackExchange.Redis/issues/300
// auto discoverability of cluster nodes is made synchronous.
// we try to connect to endpoints specified inside the user provided configuration
// and when we encounter one such endpoint to which we are able to successfully connect,
// we get the list of cluster nodes from this endpoint and try to proactively connect
// to these nodes instead of relying on auto configure
for (int iter = 0; iter < iterCount; ++iter)
{ {
Trace("Testing: " + Format.ToString(endpoints[i])); if (endpoints == null) break;
var server = GetServerEndPoint(endpoints[i]);
//server.ReportNextFailure(); var available = new Task<bool>[endpoints.Count];
servers[i] = server; tieBreakers = useTieBreakers ? new Task<string>[endpoints.Count] : null;
if (reconfigureAll && server.IsConnected) servers = new ServerEndPoint[available.Length];
RedisKey tieBreakerKey = useTieBreakers ? (RedisKey)configuration.TieBreaker : default(RedisKey);
for (int i = 0; i < available.Length; i++)
{ {
LogLocked(log, "Refreshing {0}...", Format.ToString(server.EndPoint)); Trace("Testing: " + Format.ToString(endpoints[i]));
// note that these will be processed synchronously *BEFORE* the tracer is processed, var server = GetServerEndPoint(endpoints[i]);
// so we know that the configuration will be up to date if we see the tracer //server.ReportNextFailure();
server.AutoConfigure(null); servers[i] = server;
} if (reconfigureAll && server.IsConnected)
available[i] = server.SendTracer(log); {
if (useTieBreakers) LogLocked(log, "Refreshing {0}...", Format.ToString(server.EndPoint));
{ // note that these will be processed synchronously *BEFORE* the tracer is processed,
LogLocked(log, "Requesting tie-break from {0} > {1}...", Format.ToString(server.EndPoint), configuration.TieBreaker); // so we know that the configuration will be up to date if we see the tracer
Message msg = Message.Create(0, flags, RedisCommand.GET, tieBreakerKey); server.AutoConfigure(null);
msg.SetInternalCall(); }
msg = LoggingMessage.Create(log, msg); available[i] = server.SendTracer(log);
tieBreakers[i] = server.QueueDirectAsync(msg, ResultProcessor.String); if (useTieBreakers)
{
LogLocked(log, "Requesting tie-break from {0} > {1}...", Format.ToString(server.EndPoint), configuration.TieBreaker);
Message msg = Message.Create(0, flags, RedisCommand.GET, tieBreakerKey);
msg.SetInternalCall();
msg = LoggingMessage.Create(log, msg);
tieBreakers[i] = server.QueueDirectAsync(msg, ResultProcessor.String);
}
} }
}
LogLocked(log, "Allowing endpoints {0} to respond...", TimeSpan.FromMilliseconds(configuration.ConnectTimeout)); watch = watch ?? Stopwatch.StartNew();
Trace("Allowing endpoints " + TimeSpan.FromMilliseconds(configuration.ConnectTimeout) + " to respond..."); var remaining = configuration.ConnectTimeout - checked((int)watch.ElapsedMilliseconds);
await WaitAllIgnoreErrorsAsync(available, configuration.ConnectTimeout, log).ForAwait(); LogLocked(log, "Allowing endpoints {0} to respond...", TimeSpan.FromMilliseconds(remaining));
List<ServerEndPoint> masters = new List<ServerEndPoint>(available.Length); Trace("Allowing endpoints " + TimeSpan.FromMilliseconds(remaining) + " to respond...");
await WaitAllIgnoreErrorsAsync(available, remaining, log).ForAwait();
for (int i = 0; i < available.Length; i++) EndPointCollection updatedClusterEndpointCollection = null;
{ for (int i = 0; i < available.Length; i++)
var task = available[i];
Trace(Format.ToString(endpoints[i]) + ": " + task.Status);
if (task.IsFaulted)
{ {
servers[i].SetUnselectable(UnselectableFlags.DidNotRespond); var task = available[i];
var aex = task.Exception; Trace(Format.ToString(endpoints[i]) + ": " + task.Status);
foreach (var ex in aex.InnerExceptions) if (task.IsFaulted)
{
servers[i].SetUnselectable(UnselectableFlags.DidNotRespond);
var aex = task.Exception;
foreach (var ex in aex.InnerExceptions)
{
LogLocked(log, "{0} faulted: {1}", Format.ToString(endpoints[i]), ex.Message);
failureMessage = ex.Message;
}
}
else if (task.IsCanceled)
{ {
LogLocked(log, "{0} faulted: {1}", Format.ToString(endpoints[i]), ex.Message); servers[i].SetUnselectable(UnselectableFlags.DidNotRespond);
failureMessage = ex.Message; LogLocked(log, "{0} was canceled", Format.ToString(endpoints[i]));
} }
} else if (task.IsCompleted)
else if (task.IsCanceled)
{
servers[i].SetUnselectable(UnselectableFlags.DidNotRespond);
LogLocked(log, "{0} was canceled", Format.ToString(endpoints[i]));
}
else if (task.IsCompleted)
{
var server = servers[i];
if (task.Result)
{ {
servers[i].ClearUnselectable(UnselectableFlags.DidNotRespond); var server = servers[i];
LogLocked(log, "{0} returned with success", Format.ToString(endpoints[i])); if (task.Result)
// count the server types
switch (server.ServerType)
{ {
case ServerType.Twemproxy: servers[i].ClearUnselectable(UnselectableFlags.DidNotRespond);
case ServerType.Standalone: LogLocked(log, "{0} returned with success", Format.ToString(endpoints[i]));
standaloneCount++;
break; // count the server types
case ServerType.Sentinel: switch (server.ServerType)
sentinelCount++; {
break; case ServerType.Twemproxy:
case ServerType.Cluster: case ServerType.Standalone:
clusterCount++; standaloneCount++;
break; break;
} case ServerType.Sentinel:
sentinelCount++;
break;
case ServerType.Cluster:
clusterCount++;
break;
}
// set the server UnselectableFlags and update masters list if (clusterCount > 0 && !encounteredConnectedClusterServer)
switch (server.ServerType) {
// we have encountered a connected server with clustertype for the first time.
// so we will get list of other nodes from this server using "CLUSTER NODES" command
// and try to connect to these other nodes in the next iteration
encounteredConnectedClusterServer = true;
updatedClusterEndpointCollection = GetEndpointsFromClusterNodes(server, log);
}
// set the server UnselectableFlags and update masters list
switch (server.ServerType)
{
case ServerType.Twemproxy:
case ServerType.Sentinel:
case ServerType.Standalone:
case ServerType.Cluster:
servers[i].ClearUnselectable(UnselectableFlags.ServerType);
if (server.IsSlave)
{
servers[i].ClearUnselectable(UnselectableFlags.RedundantMaster);
}
else
{
masters.Add(server);
}
break;
default:
servers[i].SetUnselectable(UnselectableFlags.ServerType);
break;
}
}
else
{ {
case ServerType.Twemproxy: servers[i].SetUnselectable(UnselectableFlags.DidNotRespond);
case ServerType.Sentinel: LogLocked(log, "{0} returned, but incorrectly", Format.ToString(endpoints[i]));
case ServerType.Standalone:
case ServerType.Cluster:
servers[i].ClearUnselectable(UnselectableFlags.ServerType);
if (server.IsSlave)
{
servers[i].ClearUnselectable(UnselectableFlags.RedundantMaster);
}
else
{
masters.Add(server);
}
break;
default:
servers[i].SetUnselectable(UnselectableFlags.ServerType);
break;
} }
} }
else else
{ {
servers[i].SetUnselectable(UnselectableFlags.DidNotRespond); servers[i].SetUnselectable(UnselectableFlags.DidNotRespond);
LogLocked(log, "{0} returned, but incorrectly", Format.ToString(endpoints[i])); LogLocked(log, "{0} did not respond", Format.ToString(endpoints[i]));
} }
} }
if (encounteredConnectedClusterServer)
{
endpoints = updatedClusterEndpointCollection;
}
else else
{ {
servers[i].SetUnselectable(UnselectableFlags.DidNotRespond); break; // we do not want to repeat the second iteration
LogLocked(log, "{0} did not respond", Format.ToString(endpoints[i]));
} }
} }
...@@ -1419,6 +1460,23 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text ...@@ -1419,6 +1460,23 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text
} }
} }
private EndPointCollection GetEndpointsFromClusterNodes(ServerEndPoint server, TextWriter log)
{
var message = Message.Create(-1, CommandFlags.None, RedisCommand.CLUSTER, RedisLiterals.NODES);
ClusterConfiguration clusterConfig = null;
try
{
clusterConfig = this.ExecuteSyncImpl(message, ResultProcessor.ClusterNodes, server);
return new EndPointCollection(clusterConfig.Nodes.Select(node => node.EndPoint).ToList());
}
catch (Exception ex)
{
LogLocked(log, "Encountered error while updating cluster config: " + ex.Message);
return null;
}
}
private void ResetAllNonConnected() private void ResetAllNonConnected()
{ {
var snapshot = serverSnapshot; var snapshot = serverSnapshot;
......
StackExchange.Redis/StackExchange/Redis/EndPointCollection.cs
View file @ cf91c206
using System; using System;
using System.Collections.Generic;
using System.Collections.ObjectModel; using System.Collections.ObjectModel;
using System.Net; using System.Net;
...@@ -9,6 +10,14 @@ namespace StackExchange.Redis ...@@ -9,6 +10,14 @@ namespace StackExchange.Redis
/// </summary> /// </summary>
public sealed class EndPointCollection : Collection<EndPoint> public sealed class EndPointCollection : Collection<EndPoint>
{ {
public EndPointCollection() : base()
{
}
public EndPointCollection(IList<EndPoint> endpoints) : base(endpoints)
{
}
/// <summary> /// <summary>
/// Format an endpoint /// Format an endpoint
/// </summary> /// </summary>
......
StackExchange.Redis/StackExchange/Redis/Format.cs
View file @ cf91c206
...@@ -133,7 +133,7 @@ internal static bool TryGetHostPort(EndPoint endpoint, out string host, out int ...@@ -133,7 +133,7 @@ internal static bool TryGetHostPort(EndPoint endpoint, out string host, out int
internal static bool TryParseDouble(string s, out double value) internal static bool TryParseDouble(string s, out double value)
{ {
if(s == null || s.Length == 0) if(string.IsNullOrEmpty(s))
{ {
value = 0; value = 0;
return false; return false;
......
StackExchange.Redis/StackExchange/Redis/RedisDatabase.cs
View file @ cf91c206
...@@ -484,7 +484,7 @@ sealed class KeyMigrateCommandMessage : Message.CommandKeyBase // MIGRATE is aty ...@@ -484,7 +484,7 @@ sealed class KeyMigrateCommandMessage : Message.CommandKeyBase // MIGRATE is aty
public KeyMigrateCommandMessage(int db, RedisKey key, EndPoint toServer, int toDatabase, int timeoutMilliseconds, MigrateOptions migrateOptions, CommandFlags flags) public KeyMigrateCommandMessage(int db, RedisKey key, EndPoint toServer, int toDatabase, int timeoutMilliseconds, MigrateOptions migrateOptions, CommandFlags flags)
: base(db, flags, RedisCommand.MIGRATE, key) : base(db, flags, RedisCommand.MIGRATE, key)
{ {
if (toServer == null) throw new ArgumentNullException("server"); if (toServer == null) throw new ArgumentNullException("toServer");
string toHost; string toHost;
int toPort; int toPort;
if (!Format.TryGetHostPort(toServer, out toHost, out toPort)) throw new ArgumentException("toServer"); if (!Format.TryGetHostPort(toServer, out toHost, out toPort)) throw new ArgumentException("toServer");
......
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