Commit 1e7703c0 authored by Marc Gravell's avatar Marc Gravell

Merge branch 'master' of github.com:StackExchange/StackExchange.Redis into jeremymeng-netcore

parents af0b2223 51704fd7
...@@ -281,14 +281,8 @@ internal ClusterNode(ClusterConfiguration configuration, string raw, EndPoint or ...@@ -281,14 +281,8 @@ internal ClusterNode(ClusterConfiguration configuration, string raw, EndPoint or
var flags = parts[2].Split(StringSplits.Comma); var flags = parts[2].Split(StringSplits.Comma);
if (flags.Contains("myself"))
{
endpoint = origin;
}
else
{
endpoint = Format.TryParseEndPoint(parts[1]); endpoint = Format.TryParseEndPoint(parts[1]);
}
nodeId = parts[0]; nodeId = parts[0];
isSlave = flags.Contains("slave"); isSlave = flags.Contains("slave");
parentNodeId = string.IsNullOrWhiteSpace(parts[3]) ? null : parts[3]; parentNodeId = string.IsNullOrWhiteSpace(parts[3]) ? null : parts[3];
......
...@@ -1216,11 +1216,31 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text ...@@ -1216,11 +1216,31 @@ 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;
ServerEndPoint[] servers = null;
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)
{
if (endpoints == null) break;
var available = new Task<bool>[endpoints.Count];
tieBreakers = useTieBreakers ? new Task<string>[endpoints.Count] : null;
servers = new ServerEndPoint[available.Length];
RedisKey tieBreakerKey = useTieBreakers ? (RedisKey)configuration.TieBreaker : default(RedisKey); RedisKey tieBreakerKey = useTieBreakers ? (RedisKey)configuration.TieBreaker : default(RedisKey);
for (int i = 0; i < available.Length; i++) for (int i = 0; i < available.Length; i++)
{ {
Trace("Testing: " + Format.ToString(endpoints[i])); Trace("Testing: " + Format.ToString(endpoints[i]));
...@@ -1245,11 +1265,13 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text ...@@ -1245,11 +1265,13 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text
} }
} }
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();
EndPointCollection updatedClusterEndpointCollection = null;
for (int i = 0; i < available.Length; i++) for (int i = 0; i < available.Length; i++)
{ {
var task = available[i]; var task = available[i];
...@@ -1292,6 +1314,15 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text ...@@ -1292,6 +1314,15 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text
break; break;
} }
if (clusterCount > 0 && !encounteredConnectedClusterServer)
{
// 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 // set the server UnselectableFlags and update masters list
switch (server.ServerType) switch (server.ServerType)
{ {
...@@ -1327,6 +1358,16 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text ...@@ -1327,6 +1358,16 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text
} }
} }
if (encounteredConnectedClusterServer)
{
endpoints = updatedClusterEndpointCollection;
}
else
{
break; // we do not want to repeat the second iteration
}
}
if (clusterCount == 0) if (clusterCount == 0)
{ {
// set the serverSelectionStrategy // set the serverSelectionStrategy
...@@ -1432,6 +1473,23 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text ...@@ -1432,6 +1473,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;
......
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>
......
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