Skip to content

Commit 2ed833e

Browse files
JamesNKJamesNK
committed
Change subchannel BalancerAddress when attributes change
1 parent 5a8e2ba commit 2ed833e

File tree

10 files changed

+195
-56
lines changed

10 files changed

+195
-56
lines changed

‎src/Grpc.Net.Client/Balancer/BalancerAddress.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
#region Copyright notice and license
1+
#region Copyright notice and license
22

33
// Copyright 2019 The gRPC Authors
44
//
@@ -30,7 +30,7 @@ namespace Grpc.Net.Client.Balancer;
3030
/// </summary>
3131
public sealed class BalancerAddress
3232
{
33-
private BalancerAttributes? _attributes;
33+
internal BalancerAttributes? _attributes;
3434

3535
/// <summary>
3636
/// Initializes a new instance of the <see cref="BalancerAddress"/> class with the specified <see cref="DnsEndPoint"/>.

‎src/Grpc.Net.Client/Balancer/BalancerAttributes.cs‎

Lines changed: 45 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -38,20 +38,22 @@ public sealed class BalancerAttributes : IDictionary<string, object?>, IReadOnly
3838
/// <summary>
3939
/// Gets a read-only collection of metadata attributes.
4040
/// </summary>
41-
public static readonly BalancerAttributes Empty = new BalancerAttributes(new ReadOnlyDictionary<string, object?>(new Dictionary<string, object?>()));
41+
public static readonly BalancerAttributes Empty = new BalancerAttributes(new Dictionary<string, object?>(), readOnly: true);
4242

43-
private readonly IDictionary<string, object?> _attributes;
43+
internal readonly Dictionary<string, object?> _attributes;
44+
private readonly bool _readOnly;
4445

4546
/// <summary>
4647
/// Initializes a new instance of the <see cref="BalancerAttributes"/> class.
4748
/// </summary>
48-
public BalancerAttributes() : this(new Dictionary<string, object?>())
49+
public BalancerAttributes() : this(new Dictionary<string, object?>(), readOnly: false)
4950
{
5051
}
5152

52-
private BalancerAttributes(IDictionary<string, object?> attributes)
53+
private BalancerAttributes(Dictionary<string, object?> attributes, bool readOnly)
5354
{
5455
_attributes = attributes;
56+
_readOnly = readOnly;
5557
}
5658

5759
object? IDictionary<string, object?>.this[string key]
@@ -62,28 +64,49 @@ private BalancerAttributes(IDictionary<string, object?> attributes)
6264
}
6365
set
6466
{
67+
ValidateReadOnly();
6568
_attributes[key] = value;
6669
}
6770
}
6871

6972
ICollection<string> IDictionary<string, object?>.Keys => _attributes.Keys;
7073
ICollection<object?> IDictionary<string, object?>.Values => _attributes.Values;
7174
int ICollection<KeyValuePair<string, object?>>.Count => _attributes.Count;
72-
bool ICollection<KeyValuePair<string, object?>>.IsReadOnly => _attributes.IsReadOnly;
75+
bool ICollection<KeyValuePair<string, object?>>.IsReadOnly => _readOnly ? true : ((ICollection<KeyValuePair<string, object?>>)_attributes).IsReadOnly;
7376
IEnumerable<string> IReadOnlyDictionary<string, object?>.Keys => _attributes.Keys;
7477
IEnumerable<object?> IReadOnlyDictionary<string, object?>.Values => _attributes.Values;
7578
int IReadOnlyCollection<KeyValuePair<string, object?>>.Count => _attributes.Count;
7679
object? IReadOnlyDictionary<string, object?>.this[string key] => _attributes[key];
77-
void IDictionary<string, object?>.Add(string key, object? value) => _attributes.Add(key, value);
78-
void ICollection<KeyValuePair<string, object?>>.Add(KeyValuePair<string, object?> item) => _attributes.Add(item);
79-
void ICollection<KeyValuePair<string, object?>>.Clear() => _attributes.Clear();
80+
void IDictionary<string, object?>.Add(string key, object? value)
81+
{
82+
ValidateReadOnly();
83+
_attributes.Add(key, value);
84+
}
85+
void ICollection<KeyValuePair<string, object?>>.Add(KeyValuePair<string, object?> item)
86+
{
87+
ValidateReadOnly();
88+
((ICollection<KeyValuePair<string, object?>>)_attributes).Add(item);
89+
}
90+
void ICollection<KeyValuePair<string, object?>>.Clear()
91+
{
92+
ValidateReadOnly();
93+
_attributes.Clear();
94+
}
8095
bool ICollection<KeyValuePair<string, object?>>.Contains(KeyValuePair<string, object?> item) => _attributes.Contains(item);
8196
bool IDictionary<string, object?>.ContainsKey(string key) => _attributes.ContainsKey(key);
82-
void ICollection<KeyValuePair<string, object?>>.CopyTo(KeyValuePair<string, object?>[] array, int arrayIndex) => _attributes.CopyTo(array, arrayIndex);
97+
void ICollection<KeyValuePair<string, object?>>.CopyTo(KeyValuePair<string, object?>[] array, int arrayIndex) => ((ICollection<KeyValuePair<string, object?>>)_attributes).CopyTo(array, arrayIndex);
8398
IEnumerator<KeyValuePair<string, object?>> IEnumerable<KeyValuePair<string, object?>>.GetEnumerator() => _attributes.GetEnumerator();
8499
IEnumerator System.Collections.IEnumerable.GetEnumerator() => ((System.Collections.IEnumerable)_attributes).GetEnumerator();
85-
bool IDictionary<string, object?>.Remove(string key) => _attributes.Remove(key);
86-
bool ICollection<KeyValuePair<string, object?>>.Remove(KeyValuePair<string, object?> item) => _attributes.Remove(item);
100+
bool IDictionary<string, object?>.Remove(string key)
101+
{
102+
ValidateReadOnly();
103+
return _attributes.Remove(key);
104+
}
105+
bool ICollection<KeyValuePair<string, object?>>.Remove(KeyValuePair<string, object?> item)
106+
{
107+
ValidateReadOnly();
108+
return ((ICollection<KeyValuePair<string, object?>>)_attributes).Remove(item);
109+
}
87110
bool IDictionary<string, object?>.TryGetValue(string key, out object? value) => _attributes.TryGetValue(key, out value);
88111
bool IReadOnlyDictionary<string, object?>.ContainsKey(string key) => _attributes.ContainsKey(key);
89112
bool IReadOnlyDictionary<string, object?>.TryGetValue(string key, out object? value) => _attributes.TryGetValue(key, out value);
@@ -121,6 +144,7 @@ public bool TryGetValue<TValue>(BalancerAttributesKey<TValue> key, [MaybeNullWhe
121144
/// <param name="value">The value.</param>
122145
public void Set<TValue>(BalancerAttributesKey<TValue> key, TValue value)
123146
{
147+
ValidateReadOnly();
124148
_attributes[key.Key] = value;
125149
}
126150

@@ -135,10 +159,19 @@ public void Set<TValue>(BalancerAttributesKey<TValue> key, TValue value)
135159
/// </returns>
136160
public bool Remove<TValue>(BalancerAttributesKey<TValue> key)
137161
{
162+
ValidateReadOnly();
138163
return _attributes.Remove(key.Key);
139164
}
140165

141-
internal string DebuggerToString()
166+
private void ValidateReadOnly()
167+
{
168+
if (_readOnly)
169+
{
170+
throw new NotSupportedException("Collection is read-only.");
171+
}
172+
}
173+
174+
private string DebuggerToString()
142175
{
143176
return $"Count = {_attributes.Count}";
144177
}

‎src/Grpc.Net.Client/Balancer/Internal/BalancerAddressEqualityComparer.cs‎

Lines changed: 41 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
#region Copyright notice and license
1+
#region Copyright notice and license
22

33
// Copyright 2019 The gRPC Authors
44
//
@@ -44,6 +44,46 @@ public bool Equals(BalancerAddress? x, BalancerAddress? y)
4444
return false;
4545
}
4646

47+
var xAttributes = x._attributes?._attributes;
48+
var yAttributes = y._attributes?._attributes;
49+
if (!AttributesEqual(xAttributes, yAttributes))
50+
{
51+
return false;
52+
}
53+
54+
return true;
55+
}
56+
57+
private bool AttributesEqual(Dictionary<string, object?>? x, Dictionary<string, object?>? y)
58+
{
59+
if (x == null && y == null)
60+
{
61+
return true;
62+
}
63+
64+
if (x == null || y == null)
65+
{
66+
return false;
67+
}
68+
69+
if (x.Count != y.Count)
70+
{
71+
return false;
72+
}
73+
74+
foreach (var kvp in x)
75+
{
76+
if (!y.TryGetValue(kvp.Key, out var value))
77+
{
78+
return false;
79+
}
80+
81+
if (!Equals(kvp.Value, value))
82+
{
83+
return false;
84+
}
85+
}
86+
4787
return true;
4888
}
4989

‎src/Grpc.Net.Client/Balancer/Internal/ISubchannelTransport.cs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
#region Copyright notice and license
1+
#region Copyright notice and license
22

33
// Copyright 2019 The gRPC Authors
44
//
@@ -17,6 +17,7 @@
1717
#endregion
1818

1919
#if SUPPORT_LOAD_BALANCING
20+
using System.Net;
2021
using Grpc.Shared;
2122

2223
namespace Grpc.Net.Client.Balancer.Internal;
@@ -28,7 +29,7 @@ namespace Grpc.Net.Client.Balancer.Internal;
2829
/// </summary>
2930
internal interface ISubchannelTransport : IDisposable
3031
{
31-
BalancerAddress? CurrentAddress { get; }
32+
DnsEndPoint? CurrentEndPoint { get; }
3233
TimeSpan? ConnectTimeout { get; }
3334

3435
#if NET5_0_OR_GREATER

‎src/Grpc.Net.Client/Balancer/Internal/PassiveSubchannelTransport.cs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -35,19 +35,19 @@ namespace Grpc.Net.Client.Balancer.Internal;
3535
internal class PassiveSubchannelTransport : ISubchannelTransport, IDisposable
3636
{
3737
private readonly Subchannel _subchannel;
38-
private BalancerAddress? _currentAddress;
38+
private DnsEndPoint? _currentEndPoint;
3939

4040
public PassiveSubchannelTransport(Subchannel subchannel)
4141
{
4242
_subchannel = subchannel;
4343
}
4444

45-
public BalancerAddress? CurrentAddress => _currentAddress;
45+
public DnsEndPoint? CurrentEndPoint => _currentEndPoint;
4646
public TimeSpan? ConnectTimeout { get; }
4747

4848
public void Disconnect()
4949
{
50-
_currentAddress = null;
50+
_currentEndPoint = null;
5151
_subchannel.UpdateConnectivityState(ConnectivityState.Idle, "Disconnected.");
5252
}
5353

@@ -60,12 +60,12 @@ public void Disconnect()
6060
TryConnectAsync(ConnectContext context)
6161
{
6262
Debug.Assert(_subchannel._addresses.Count == 1);
63-
Debug.Assert(CurrentAddress == null);
63+
Debug.Assert(CurrentEndPoint == null);
6464

6565
var currentAddress = _subchannel._addresses[0];
6666

6767
_subchannel.UpdateConnectivityState(ConnectivityState.Connecting, "Passively connecting.");
68-
_currentAddress = currentAddress;
68+
_currentEndPoint = currentAddress.EndPoint;
6969
_subchannel.UpdateConnectivityState(ConnectivityState.Ready, "Passively connected.");
7070

7171
#if !NETSTANDARD2_0 && !NET462
@@ -77,7 +77,7 @@ public void Disconnect()
7777

7878
public void Dispose()
7979
{
80-
_currentAddress = null;
80+
_currentEndPoint = null;
8181
}
8282

8383
#if NET5_0_OR_GREATER

‎src/Grpc.Net.Client/Balancer/Internal/SocketConnectivitySubchannelTransport.cs‎

Lines changed: 25 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -63,11 +63,11 @@ internal class SocketConnectivitySubchannelTransport : ISubchannelTransport, IDi
6363

6464
private int _lastEndPointIndex;
6565
internal Socket? _initialSocket;
66-
private BalancerAddress? _initialSocketAddress;
66+
private DnsEndPoint? _initialSocketEndPoint;
6767
private List<ReadOnlyMemory<byte>>? _initialSocketData;
6868
private DateTime? _initialSocketCreatedTime;
6969
private bool _disposed;
70-
private BalancerAddress? _currentAddress;
70+
private DnsEndPoint? _currentEndPoint;
7171

7272
public SocketConnectivitySubchannelTransport(
7373
Subchannel subchannel,
@@ -88,7 +88,7 @@ public SocketConnectivitySubchannelTransport(
8888
}
8989

9090
private object Lock => _subchannel.Lock;
91-
public BalancerAddress? CurrentAddress => _currentAddress;
91+
public DnsEndPoint? CurrentEndPoint => _currentEndPoint;
9292
public TimeSpan? ConnectTimeout { get; }
9393

9494
// For testing. Take a copy under lock for thread-safety.
@@ -127,16 +127,16 @@ private void DisconnectUnsynchronized()
127127

128128
_initialSocket?.Dispose();
129129
_initialSocket = null;
130-
_initialSocketAddress = null;
130+
_initialSocketEndPoint = null;
131131
_initialSocketData = null;
132132
_initialSocketCreatedTime = null;
133133
_lastEndPointIndex = 0;
134-
_currentAddress = null;
134+
_currentEndPoint = null;
135135
}
136136

137137
public async ValueTask<ConnectResult> TryConnectAsync(ConnectContext context)
138138
{
139-
Debug.Assert(CurrentAddress == null);
139+
Debug.Assert(CurrentEndPoint == null);
140140

141141
// Addresses could change while connecting. Make a copy of the subchannel's addresses.
142142
var addresses = _subchannel.GetAddresses();
@@ -162,10 +162,10 @@ public async ValueTask<ConnectResult> TryConnectAsync(ConnectContext context)
162162

163163
lock (Lock)
164164
{
165-
_currentAddress = currentAddress;
165+
_currentEndPoint = currentAddress.EndPoint;
166166
_lastEndPointIndex = currentIndex;
167167
_initialSocket = socket;
168-
_initialSocketAddress = currentAddress;
168+
_initialSocketEndPoint = currentAddress.EndPoint;
169169
_initialSocketData = null;
170170
_initialSocketCreatedTime = DateTime.UtcNow;
171171

@@ -240,20 +240,28 @@ private void OnCheckSocketConnection(object? state)
240240
try
241241
{
242242
Socket? socket;
243-
BalancerAddress? socketAddress;
243+
DnsEndPoint? socketEndpoint;
244244
var closeSocket = false;
245245
Exception? checkException = null;
246246

247247
lock (Lock)
248248
{
249249
socket = _initialSocket;
250-
socketAddress = _initialSocketAddress;
250+
socketEndpoint = _initialSocketEndPoint;
251251

252252
if (socket != null)
253253
{
254-
CompatibilityHelpers.Assert(socketAddress != null);
254+
CompatibilityHelpers.Assert(socketEndpoint != null);
255255

256-
closeSocket = ShouldCloseSocket(socket, socketAddress, ref _initialSocketData, out checkException);
256+
var address = _subchannel.GetAddress(socketEndpoint);
257+
if (address != null)
258+
{
259+
closeSocket = ShouldCloseSocket(socket, address, ref _initialSocketData, out checkException);
260+
}
261+
else
262+
{
263+
closeSocket = true;
264+
}
257265
}
258266
}
259267

@@ -296,27 +304,27 @@ public async ValueTask<Stream> GetStreamAsync(BalancerAddress address, Cancellat
296304
SocketConnectivitySubchannelTransportLog.CreatingStream(_logger, _subchannel.Id, address);
297305

298306
Socket? socket = null;
299-
BalancerAddress? socketAddress = null;
307+
DnsEndPoint? socketEndPoint = null;
300308
List<ReadOnlyMemory<byte>>? socketData = null;
301309
DateTime? socketCreatedTime = null;
302310
lock (Lock)
303311
{
304312
if (_initialSocket != null)
305313
{
306-
var socketAddressMatch = Equals(_initialSocketAddress, address);
314+
var socketEndPointMatch = Equals(_initialSocketEndPoint, address.EndPoint);
307315

308316
socket = _initialSocket;
309-
socketAddress = _initialSocketAddress;
317+
socketEndPoint = _initialSocketEndPoint;
310318
socketData = _initialSocketData;
311319
socketCreatedTime = _initialSocketCreatedTime;
312320
_initialSocket = null;
313-
_initialSocketAddress = null;
321+
_initialSocketEndPoint = null;
314322
_initialSocketData = null;
315323
_initialSocketCreatedTime = null;
316324

317325
// Double check the address matches the socket address and only use socket on match.
318326
// Not sure if this is possible in practice, but better safe than sorry.
319-
if (!socketAddressMatch)
327+
if (!socketEndPointMatch)
320328
{
321329
socket.Dispose();
322330
socket = null;

0 commit comments

Comments
 (0)