Skip to content

Commit 33e14d0

Browse files
adelinowonaadelinowona
authored
CSHARP-3757: Redirect read/write retries to other mongos if possible (#1304)
1 parent 61d2c77 commit 33e14d0

File tree

19 files changed

+844
-31
lines changed

19 files changed

+844
-31
lines changed

‎src/MongoDB.Driver.Core/Core/Bindings/ChannelReadBinding.cs‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,10 +14,9 @@
1414
*/
1515

1616
using System;
17+
using System.Collections.Generic;
1718
using System.Threading;
1819
using System.Threading.Tasks;
19-
using MongoDB.Driver.Core.Clusters;
20-
using MongoDB.Driver.Core.Connections;
2120
using MongoDB.Driver.Core.Misc;
2221
using MongoDB.Driver.Core.Servers;
2322

@@ -51,7 +50,7 @@ public ChannelReadBinding(IServer server, IChannelHandle channel, ReadPreference
5150
_session = Ensure.IsNotNull(session, nameof(session));
5251
}
5352

54-
// properties
53+
// properties
5554
/// <inheritdoc/>
5655
public ReadPreference ReadPreference
5756
{
@@ -90,6 +89,18 @@ public Task<IChannelSourceHandle> GetReadChannelSourceAsync(CancellationToken ca
9089
return Task.FromResult<IChannelSourceHandle>(GetReadChannelSourceHelper());
9190
}
9291

92+
/// <inheritdoc />
93+
public IChannelSourceHandle GetReadChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
94+
{
95+
return GetReadChannelSource(cancellationToken);
96+
}
97+
98+
/// <inheritdoc />
99+
public Task<IChannelSourceHandle> GetReadChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
100+
{
101+
return GetReadChannelSourceAsync(cancellationToken);
102+
}
103+
93104
private IChannelSourceHandle GetReadChannelSourceHelper()
94105
{
95106
return new ChannelSourceHandle(new ChannelChannelSource(_server, _channel.Fork(), _session.Fork()));

‎src/MongoDB.Driver.Core/Core/Bindings/ChannelReadWriteBinding.cs‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
*/
1515

1616
using System;
17+
using System.Collections.Generic;
1718
using System.Threading;
1819
using System.Threading.Tasks;
1920
using MongoDB.Driver.Core.Misc;
@@ -85,32 +86,68 @@ public Task<IChannelSourceHandle> GetReadChannelSourceAsync(CancellationToken ca
8586
return Task.FromResult(GetChannelSourceHelper());
8687
}
8788

89+
/// <inheritdoc />
90+
public IChannelSourceHandle GetReadChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
91+
{
92+
return GetReadChannelSource(cancellationToken);
93+
}
94+
95+
/// <inheritdoc />
96+
public Task<IChannelSourceHandle> GetReadChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
97+
{
98+
return GetReadChannelSourceAsync(cancellationToken);
99+
}
100+
88101
/// <inheritdoc/>
89102
public IChannelSourceHandle GetWriteChannelSource(CancellationToken cancellationToken)
90103
{
91104
ThrowIfDisposed();
92105
return GetChannelSourceHelper();
93106
}
94107

108+
/// <inheritdoc />
109+
public IChannelSourceHandle GetWriteChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
110+
{
111+
return GetWriteChannelSource(cancellationToken);
112+
}
113+
95114
/// <inheritdoc/>
96115
public IChannelSourceHandle GetWriteChannelSource(IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken)
97116
{
98117
return GetWriteChannelSource(cancellationToken); // ignore mayUseSecondary
99118
}
100119

120+
/// <inheritdoc />
121+
public IChannelSourceHandle GetWriteChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken)
122+
{
123+
return GetWriteChannelSource(mayUseSecondary, cancellationToken);
124+
}
125+
101126
/// <inheritdoc/>
102127
public Task<IChannelSourceHandle> GetWriteChannelSourceAsync(CancellationToken cancellationToken)
103128
{
104129
ThrowIfDisposed();
105130
return Task.FromResult(GetChannelSourceHelper());
106131
}
107132

133+
/// <inheritdoc />
134+
public Task<IChannelSourceHandle> GetWriteChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
135+
{
136+
return GetWriteChannelSourceAsync(cancellationToken);
137+
}
138+
108139
/// <inheritdoc/>
109140
public Task<IChannelSourceHandle> GetWriteChannelSourceAsync(IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken)
110141
{
111142
return GetWriteChannelSourceAsync(cancellationToken); // ignore mayUseSecondary
112143
}
113144

145+
/// <inheritdoc />
146+
public Task<IChannelSourceHandle> GetWriteChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken)
147+
{
148+
return GetWriteChannelSourceAsync(mayUseSecondary, cancellationToken);
149+
}
150+
114151
// private methods
115152
private IChannelSourceHandle GetChannelSourceHelper()
116153
{

‎src/MongoDB.Driver.Core/Core/Bindings/ChannelSourceReadWriteBinding.cs‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,11 @@
1414
*/
1515

1616
using System;
17+
using System.Collections.Generic;
1718
using System.Threading;
1819
using System.Threading.Tasks;
1920
using MongoDB.Driver.Core.Misc;
21+
using MongoDB.Driver.Core.Servers;
2022

2123
namespace MongoDB.Driver.Core.Bindings
2224
{
@@ -73,32 +75,68 @@ public Task<IChannelSourceHandle> GetReadChannelSourceAsync(CancellationToken ca
7375
return Task.FromResult(GetChannelSourceHelper());
7476
}
7577

78+
/// <inheritdoc />
79+
public IChannelSourceHandle GetReadChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
80+
{
81+
return GetReadChannelSource(cancellationToken);
82+
}
83+
84+
/// <inheritdoc />
85+
public Task<IChannelSourceHandle> GetReadChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
86+
{
87+
return GetReadChannelSourceAsync(cancellationToken);
88+
}
89+
7690
/// <inheritdoc/>
7791
public IChannelSourceHandle GetWriteChannelSource(CancellationToken cancellationToken)
7892
{
7993
ThrowIfDisposed();
8094
return GetChannelSourceHelper();
8195
}
8296

97+
/// <inheritdoc />
98+
public IChannelSourceHandle GetWriteChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
99+
{
100+
return GetWriteChannelSource(cancellationToken);
101+
}
102+
83103
/// <inheritdoc/>
84104
public IChannelSourceHandle GetWriteChannelSource(IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken)
85105
{
86106
return GetWriteChannelSource(cancellationToken); // ignore mayUseSecondary
87107
}
88108

109+
/// <inheritdoc />
110+
public IChannelSourceHandle GetWriteChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken)
111+
{
112+
return GetWriteChannelSource(mayUseSecondary, cancellationToken);
113+
}
114+
89115
/// <inheritdoc/>
90116
public Task<IChannelSourceHandle> GetWriteChannelSourceAsync(CancellationToken cancellationToken)
91117
{
92118
ThrowIfDisposed();
93119
return Task.FromResult(GetChannelSourceHelper());
94120
}
95121

122+
/// <inheritdoc />
123+
public Task<IChannelSourceHandle> GetWriteChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
124+
{
125+
return GetWriteChannelSourceAsync(cancellationToken);
126+
}
127+
96128
/// <inheritdoc/>
97129
public Task<IChannelSourceHandle> GetWriteChannelSourceAsync(IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken)
98130
{
99131
return GetWriteChannelSourceAsync(cancellationToken); // ignore mayUseSecondary
100132
}
101133

134+
/// <inheritdoc />
135+
public Task<IChannelSourceHandle> GetWriteChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken)
136+
{
137+
return GetWriteChannelSourceAsync(mayUseSecondary, cancellationToken);
138+
}
139+
102140
/// <inheritdoc/>
103141
public void Dispose()
104142
{

‎src/MongoDB.Driver.Core/Core/Bindings/IBinding.cs‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
*/
1515

1616
using System;
17+
using System.Collections.Generic;
1718
using System.Threading;
1819
using System.Threading.Tasks;
1920
using MongoDB.Driver.Core.Servers;
@@ -61,6 +62,22 @@ public interface IReadBinding : IBinding
6162
/// <param name="cancellationToken">The cancellation token.</param>
6263
/// <returns>A channel source.</returns>
6364
Task<IChannelSourceHandle> GetReadChannelSourceAsync(CancellationToken cancellationToken);
65+
66+
/// <summary>
67+
/// Gets a channel source for read operations while deprioritizing servers in the provided collection.
68+
/// </summary>
69+
/// <param name="deprioritizedServers">The deprioritized servers.</param>
70+
/// <param name="cancellationToken">The cancellation token.</param>
71+
/// <returns>A channel source.</returns>
72+
IChannelSourceHandle GetReadChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken);
73+
74+
/// <summary>
75+
/// Gets a channel source for read operations while deprioritizing servers in the provided collection.
76+
/// </summary>
77+
/// <param name="deprioritizedServers">The deprioritized servers.</param>
78+
/// <param name="cancellationToken">The cancellation token.</param>
79+
/// <returns>A channel source.</returns>
80+
Task<IChannelSourceHandle> GetReadChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken);
6481
}
6582

6683
/// <summary>
@@ -75,6 +92,14 @@ public interface IWriteBinding : IBinding
7592
/// <returns>A channel source.</returns>
7693
IChannelSourceHandle GetWriteChannelSource(CancellationToken cancellationToken);
7794

95+
/// <summary>
96+
/// Gets a channel source for write operations while deprioritizing servers in the provided collection.
97+
/// </summary>
98+
/// <param name="deprioritizedServers">The deprioritized servers.</param>
99+
/// <param name="cancellationToken">The cancellation token.</param>
100+
/// <returns>A channel source.</returns>
101+
IChannelSourceHandle GetWriteChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken);
102+
78103
/// <summary>
79104
/// Gets a channel source for write operations that may use a secondary.
80105
/// </summary>
@@ -83,20 +108,46 @@ public interface IWriteBinding : IBinding
83108
/// <returns>A channel source.</returns>
84109
IChannelSourceHandle GetWriteChannelSource(IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken);
85110

111+
/// <summary>
112+
/// Gets a channel source for write operations that may use a secondary and deprioritizes servers in the provided collection.
113+
/// </summary>
114+
/// <param name="deprioritizedServers">The deprioritized servers.</param>
115+
/// <param name="mayUseSecondary">The may use secondary criteria.</param>
116+
/// <param name="cancellationToken">The cancellation token.</param>
117+
/// <returns>A channel source.</returns>
118+
IChannelSourceHandle GetWriteChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken);
119+
86120
/// <summary>
87121
/// Gets a channel source for write operations.
88122
/// </summary>
89123
/// <param name="cancellationToken">The cancellation token.</param>
90124
/// <returns>A channel source.</returns>
91125
Task<IChannelSourceHandle> GetWriteChannelSourceAsync(CancellationToken cancellationToken);
92126

127+
/// <summary>
128+
/// Gets a channel source for write operations while deprioritizing servers in the provided collection.
129+
/// </summary>
130+
/// <param name="deprioritizedServers">The deprioritized servers.</param>
131+
/// <param name="cancellationToken">The cancellation token.</param>
132+
/// <returns>A channel source.</returns>
133+
Task<IChannelSourceHandle> GetWriteChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken);
134+
93135
/// <summary>
94136
/// Gets a channel source for write operations that may use a secondary.
95137
/// </summary>
96138
/// <param name="mayUseSecondary">The may use secondary criteria.</param>
97139
/// <param name="cancellationToken">The cancellation token.</param>
98140
/// <returns>A channel source.</returns>
99141
Task<IChannelSourceHandle> GetWriteChannelSourceAsync(IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken);
142+
143+
/// <summary>
144+
/// Gets a channel source for write operations that may use a secondary and deprioritizes servers in the provided collection.
145+
/// </summary>
146+
/// <param name="deprioritizedServers">The deprioritized servers.</param>
147+
/// <param name="mayUseSecondary">The may use secondary criteria.</param>
148+
/// <param name="cancellationToken">The cancellation token.</param>
149+
/// <returns>A channel source.</returns>
150+
Task<IChannelSourceHandle> GetWriteChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, IMayUseSecondaryCriteria mayUseSecondary, CancellationToken cancellationToken);
100151
}
101152

102153
/// <summary>

‎src/MongoDB.Driver.Core/Core/Bindings/ReadBindingHandle.cs‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,12 +15,10 @@
1515

1616
using System;
1717
using System.Collections.Generic;
18-
using System.Linq;
19-
using System.Text;
2018
using System.Threading;
2119
using System.Threading.Tasks;
22-
using MongoDB.Driver.Core.Clusters;
2320
using MongoDB.Driver.Core.Misc;
21+
using MongoDB.Driver.Core.Servers;
2422

2523
namespace MongoDB.Driver.Core.Bindings
2624
{
@@ -76,6 +74,20 @@ public Task<IChannelSourceHandle> GetReadChannelSourceAsync(CancellationToken ca
7674
return _reference.Instance.GetReadChannelSourceAsync(cancellationToken);
7775
}
7876

77+
/// <inheritdoc />
78+
public IChannelSourceHandle GetReadChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
79+
{
80+
ThrowIfDisposed();
81+
return _reference.Instance.GetReadChannelSource(deprioritizedServers, cancellationToken);
82+
}
83+
84+
/// <inheritdoc />
85+
public Task<IChannelSourceHandle> GetReadChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
86+
{
87+
ThrowIfDisposed();
88+
return _reference.Instance.GetReadChannelSourceAsync(deprioritizedServers, cancellationToken);
89+
}
90+
7991
/// <inheritdoc/>
8092
public void Dispose()
8193
{

‎src/MongoDB.Driver.Core/Core/Bindings/ReadPreferenceBinding.cs‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
*/
1515

1616
using System;
17+
using System.Collections.Generic;
1718
using System.Threading;
1819
using System.Threading.Tasks;
1920
using MongoDB.Driver.Core.Clusters;
@@ -66,17 +67,29 @@ public ICoreSessionHandle Session
6667
// methods
6768
/// <inheritdoc/>
6869
public IChannelSourceHandle GetReadChannelSource(CancellationToken cancellationToken)
70+
{
71+
return GetReadChannelSource(null, cancellationToken);
72+
}
73+
74+
/// <inheritdoc/>
75+
public Task<IChannelSourceHandle> GetReadChannelSourceAsync(CancellationToken cancellationToken)
76+
{
77+
return GetReadChannelSourceAsync(null, cancellationToken);
78+
}
79+
80+
/// <inheritdoc />
81+
public IChannelSourceHandle GetReadChannelSource(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
6982
{
7083
ThrowIfDisposed();
71-
var server = _cluster.SelectServerAndPinIfNeeded(_session, _serverSelector, cancellationToken);
84+
var server = _cluster.SelectServerAndPinIfNeeded(_session, _serverSelector, deprioritizedServers, cancellationToken);
7285
return GetChannelSourceHelper(server);
7386
}
7487

75-
/// <inheritdoc/>
76-
public async Task<IChannelSourceHandle> GetReadChannelSourceAsync(CancellationToken cancellationToken)
88+
/// <inheritdoc />
89+
public async Task<IChannelSourceHandle> GetReadChannelSourceAsync(IReadOnlyCollection<ServerDescription> deprioritizedServers, CancellationToken cancellationToken)
7790
{
7891
ThrowIfDisposed();
79-
var server = await _cluster.SelectServerAndPinIfNeededAsync(_session, _serverSelector, cancellationToken).ConfigureAwait(false);
92+
var server = await _cluster.SelectServerAndPinIfNeededAsync(_session, _serverSelector, deprioritizedServers, cancellationToken).ConfigureAwait(false);
8093
return GetChannelSourceHelper(server);
8194
}
8295

0 commit comments

Comments
 (0)