Skip to content

Commit 81f50d7

Browse files
rojiroji
committed
Add option to disable concurrency detection
1 parent ede9db2 commit 81f50d7

File tree

40 files changed

+1016
-529
lines changed

40 files changed

+1016
-529
lines changed

‎src/EFCore.Cosmos/Query/Internal/CosmosShapedQueryCompilingExpressionVisitor.QueryingEnumerable.cs

Lines changed: 60 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ private sealed class QueryingEnumerable<T> : IEnumerable<T>, IAsyncEnumerable<T>
3535
private readonly string _partitionKey;
3636
private readonly IDiagnosticsLogger<DbLoggerCategory.Query> _queryLogger;
3737
private readonly bool _standAloneStateManager;
38+
private readonly bool _concurrencyDetectionEnabled;
3839

3940
public QueryingEnumerable(
4041
CosmosQueryContext cosmosQueryContext,
@@ -44,7 +45,8 @@ public QueryingEnumerable(
4445
Func<CosmosQueryContext, JObject, T> shaper,
4546
Type contextType,
4647
string partitionKeyFromExtension,
47-
bool standAloneStateManager)
48+
bool standAloneStateManager,
49+
bool concurrencyDetectionEnabled)
4850
{
4951
_cosmosQueryContext = cosmosQueryContext;
5052
_sqlExpressionFactory = sqlExpressionFactory;
@@ -54,6 +56,7 @@ public QueryingEnumerable(
5456
_contextType = contextType;
5557
_queryLogger = cosmosQueryContext.QueryLogger;
5658
_standAloneStateManager = standAloneStateManager;
59+
_concurrencyDetectionEnabled = concurrencyDetectionEnabled;
5760

5861
var partitionKey = selectExpression.GetPartitionKey(cosmosQueryContext.ParameterValues);
5962
if (partitionKey != null && partitionKeyFromExtension != null && partitionKeyFromExtension != partitionKey)
@@ -113,6 +116,7 @@ private sealed class Enumerator : IEnumerator<T>
113116
private readonly string _partitionKey;
114117
private readonly IDiagnosticsLogger<DbLoggerCategory.Query> _queryLogger;
115118
private readonly bool _standAloneStateManager;
119+
private readonly IConcurrencyDetector _concurrencyDetector;
116120

117121
private IEnumerator<JObject> _enumerator;
118122

@@ -126,6 +130,10 @@ public Enumerator(QueryingEnumerable<T> queryingEnumerable)
126130
_partitionKey = queryingEnumerable._partitionKey;
127131
_queryLogger = queryingEnumerable._queryLogger;
128132
_standAloneStateManager = queryingEnumerable._standAloneStateManager;
133+
134+
_concurrencyDetector = queryingEnumerable._concurrencyDetectionEnabled
135+
? _cosmosQueryContext.ConcurrencyDetector
136+
: null;
129137
}
130138

131139
public T Current { get; private set; }
@@ -135,41 +143,44 @@ object IEnumerator.Current
135143

136144
public bool MoveNext()
137145
{
146+
_concurrencyDetector?.EnterCriticalSection();
147+
138148
try
139149
{
140-
using (_cosmosQueryContext.ConcurrencyDetector.EnterCriticalSection())
150+
if (_enumerator == null)
141151
{
142-
if (_enumerator == null)
143-
{
144-
var sqlQuery = _queryingEnumerable.GenerateQuery();
152+
var sqlQuery = _queryingEnumerable.GenerateQuery();
145153

146-
EntityFrameworkEventSource.Log.QueryExecuting();
154+
EntityFrameworkEventSource.Log.QueryExecuting();
147155

148-
_enumerator = _cosmosQueryContext.CosmosClient
149-
.ExecuteSqlQuery(
150-
_selectExpression.Container,
151-
_partitionKey,
152-
sqlQuery)
153-
.GetEnumerator();
154-
_cosmosQueryContext.InitializeStateManager(_standAloneStateManager);
155-
}
156+
_enumerator = _cosmosQueryContext.CosmosClient
157+
.ExecuteSqlQuery(
158+
_selectExpression.Container,
159+
_partitionKey,
160+
sqlQuery)
161+
.GetEnumerator();
162+
_cosmosQueryContext.InitializeStateManager(_standAloneStateManager);
163+
}
156164

157-
var hasNext = _enumerator.MoveNext();
165+
var hasNext = _enumerator.MoveNext();
158166

159-
Current
160-
= hasNext
161-
? _shaper(_cosmosQueryContext, _enumerator.Current)
162-
: default;
167+
Current
168+
= hasNext
169+
? _shaper(_cosmosQueryContext, _enumerator.Current)
170+
: default;
163171

164-
return hasNext;
165-
}
172+
return hasNext;
166173
}
167174
catch (Exception exception)
168175
{
169176
_queryLogger.QueryIterationFailed(_contextType, exception);
170177

171178
throw;
172179
}
180+
finally
181+
{
182+
_concurrencyDetector?.ExitCriticalSection();
183+
}
173184
}
174185

175186
public void Dispose()
@@ -193,6 +204,7 @@ private sealed class AsyncEnumerator : IAsyncEnumerator<T>
193204
private readonly IDiagnosticsLogger<DbLoggerCategory.Query> _queryLogger;
194205
private readonly bool _standAloneStateManager;
195206
private readonly CancellationToken _cancellationToken;
207+
private readonly IConcurrencyDetector _concurrencyDetector;
196208

197209
private IAsyncEnumerator<JObject> _enumerator;
198210

@@ -207,47 +219,54 @@ public AsyncEnumerator(QueryingEnumerable<T> queryingEnumerable, CancellationTok
207219
_queryLogger = queryingEnumerable._queryLogger;
208220
_standAloneStateManager = queryingEnumerable._standAloneStateManager;
209221
_cancellationToken = cancellationToken;
222+
223+
_concurrencyDetector = queryingEnumerable._concurrencyDetectionEnabled
224+
? _cosmosQueryContext.ConcurrencyDetector
225+
: null;
210226
}
211227

212228
public T Current { get; private set; }
213229

214230
public async ValueTask<bool> MoveNextAsync()
215231
{
232+
_concurrencyDetector?.EnterCriticalSection();
233+
216234
try
217235
{
218-
using (_cosmosQueryContext.ConcurrencyDetector.EnterCriticalSection())
236+
if (_enumerator == null)
219237
{
220-
if (_enumerator == null)
221-
{
222-
var sqlQuery = _queryingEnumerable.GenerateQuery();
238+
var sqlQuery = _queryingEnumerable.GenerateQuery();
223239

224-
EntityFrameworkEventSource.Log.QueryExecuting();
240+
EntityFrameworkEventSource.Log.QueryExecuting();
225241

226-
_enumerator = _cosmosQueryContext.CosmosClient
227-
.ExecuteSqlQueryAsync(
228-
_selectExpression.Container,
229-
_partitionKey,
230-
sqlQuery)
231-
.GetAsyncEnumerator(_cancellationToken);
232-
_cosmosQueryContext.InitializeStateManager(_standAloneStateManager);
233-
}
242+
_enumerator = _cosmosQueryContext.CosmosClient
243+
.ExecuteSqlQueryAsync(
244+
_selectExpression.Container,
245+
_partitionKey,
246+
sqlQuery)
247+
.GetAsyncEnumerator(_cancellationToken);
248+
_cosmosQueryContext.InitializeStateManager(_standAloneStateManager);
249+
}
234250

235-
var hasNext = await _enumerator.MoveNextAsync().ConfigureAwait(false);
251+
var hasNext = await _enumerator.MoveNextAsync().ConfigureAwait(false);
236252

237-
Current
238-
= hasNext
239-
? _shaper(_cosmosQueryContext, _enumerator.Current)
240-
: default;
253+
Current
254+
= hasNext
255+
? _shaper(_cosmosQueryContext, _enumerator.Current)
256+
: default;
241257

242-
return hasNext;
243-
}
258+
return hasNext;
244259
}
245260
catch (Exception exception)
246261
{
247262
_queryLogger.QueryIterationFailed(_contextType, exception);
248263

249264
throw;
250265
}
266+
finally
267+
{
268+
_concurrencyDetector?.ExitCriticalSection();
269+
}
251270
}
252271

253272
public ValueTask DisposeAsync()

‎src/EFCore.Cosmos/Query/Internal/CosmosShapedQueryCompilingExpressionVisitor.ReadItemQueryingEnumerable.cs

Lines changed: 66 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -35,20 +35,23 @@ private sealed class ReadItemQueryingEnumerable<T> : IEnumerable<T>, IAsyncEnume
3535
private readonly Type _contextType;
3636
private readonly IDiagnosticsLogger<DbLoggerCategory.Query> _queryLogger;
3737
private readonly bool _standAloneStateManager;
38+
private readonly bool _concurrencyDetectionEnabled;
3839

3940
public ReadItemQueryingEnumerable(
4041
CosmosQueryContext cosmosQueryContext,
4142
ReadItemExpression readItemExpression,
4243
Func<CosmosQueryContext, JObject, T> shaper,
4344
Type contextType,
44-
bool standAloneStateManager)
45+
bool standAloneStateManager,
46+
bool concurrencyDetectionEnabled)
4547
{
4648
_cosmosQueryContext = cosmosQueryContext;
4749
_readItemExpression = readItemExpression;
4850
_shaper = shaper;
4951
_contextType = contextType;
5052
_queryLogger = _cosmosQueryContext.QueryLogger;
5153
_standAloneStateManager = standAloneStateManager;
54+
_concurrencyDetectionEnabled = concurrencyDetectionEnabled;
5255
}
5356

5457
public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
@@ -173,6 +176,7 @@ private sealed class Enumerator : IEnumerator<T>, IAsyncEnumerator<T>
173176
private readonly Type _contextType;
174177
private readonly IDiagnosticsLogger<DbLoggerCategory.Query> _queryLogger;
175178
private readonly bool _standAloneStateManager;
179+
private readonly bool _concurrencyDetectionEnabled;
176180
private readonly ReadItemQueryingEnumerable<T> _readItemEnumerable;
177181
private readonly CancellationToken _cancellationToken;
178182

@@ -187,6 +191,7 @@ public Enumerator(ReadItemQueryingEnumerable<T> readItemEnumerable, Cancellation
187191
_contextType = readItemEnumerable._contextType;
188192
_queryLogger = readItemEnumerable._queryLogger;
189193
_standAloneStateManager = readItemEnumerable._standAloneStateManager;
194+
_concurrencyDetectionEnabled = readItemEnumerable._concurrencyDetectionEnabled;
190195
_readItemEnumerable = readItemEnumerable;
191196
_cancellationToken = cancellationToken;
192197
}
@@ -198,82 +203,100 @@ object IEnumerator.Current
198203

199204
public bool MoveNext()
200205
{
206+
if (_concurrencyDetectionEnabled)
207+
{
208+
_cosmosQueryContext.ConcurrencyDetector.EnterCriticalSection();
209+
}
210+
201211
try
202212
{
203-
using (_cosmosQueryContext.ConcurrencyDetector.EnterCriticalSection())
213+
if (!_hasExecuted)
204214
{
205-
if (!_hasExecuted)
215+
if (!_readItemEnumerable.TryGetResourceId(out var resourceId))
206216
{
207-
if (!_readItemEnumerable.TryGetResourceId(out var resourceId))
208-
{
209-
throw new InvalidOperationException(CosmosStrings.ResourceIdMissing);
210-
}
217+
throw new InvalidOperationException(CosmosStrings.ResourceIdMissing);
218+
}
211219

212-
if (!_readItemEnumerable.TryGetPartitionId(out var partitionKey))
213-
{
214-
throw new InvalidOperationException(CosmosStrings.PartitionKeyMissing);
215-
}
220+
if (!_readItemEnumerable.TryGetPartitionId(out var partitionKey))
221+
{
222+
throw new InvalidOperationException(CosmosStrings.PartitionKeyMissing);
223+
}
216224

217-
EntityFrameworkEventSource.Log.QueryExecuting();
225+
EntityFrameworkEventSource.Log.QueryExecuting();
218226

219-
_item = _cosmosQueryContext.CosmosClient.ExecuteReadItem(
220-
_readItemExpression.Container,
221-
partitionKey,
222-
resourceId);
227+
_item = _cosmosQueryContext.CosmosClient.ExecuteReadItem(
228+
_readItemExpression.Container,
229+
partitionKey,
230+
resourceId);
223231

224-
return ShapeResult();
225-
}
226-
227-
return false;
232+
return ShapeResult();
228233
}
234+
235+
return false;
229236
}
230237
catch (Exception exception)
231238
{
232239
_queryLogger.QueryIterationFailed(_contextType, exception);
233240

234241
throw;
235242
}
243+
finally
244+
{
245+
if (_concurrencyDetectionEnabled)
246+
{
247+
_cosmosQueryContext.ConcurrencyDetector.ExitCriticalSection();
248+
}
249+
}
236250
}
237251

238252
public async ValueTask<bool> MoveNextAsync()
239253
{
254+
if (_concurrencyDetectionEnabled)
255+
{
256+
_cosmosQueryContext.ConcurrencyDetector.EnterCriticalSection();
257+
}
258+
240259
try
241260
{
242-
using (_cosmosQueryContext.ConcurrencyDetector.EnterCriticalSection())
261+
if (!_hasExecuted)
243262
{
244-
if (!_hasExecuted)
263+
if (!_readItemEnumerable.TryGetResourceId(out var resourceId))
264+
{
265+
throw new InvalidOperationException(CosmosStrings.ResourceIdMissing);
266+
}
267+
268+
if (!_readItemEnumerable.TryGetPartitionId(out var partitionKey))
245269
{
246-
if (!_readItemEnumerable.TryGetResourceId(out var resourceId))
247-
{
248-
throw new InvalidOperationException(CosmosStrings.ResourceIdMissing);
249-
}
250-
251-
if (!_readItemEnumerable.TryGetPartitionId(out var partitionKey))
252-
{
253-
throw new InvalidOperationException(CosmosStrings.PartitionKeyMissing);
254-
}
255-
256-
EntityFrameworkEventSource.Log.QueryExecuting();
257-
258-
_item = await _cosmosQueryContext.CosmosClient.ExecuteReadItemAsync(
259-
_readItemExpression.Container,
260-
partitionKey,
261-
resourceId,
262-
_cancellationToken)
263-
.ConfigureAwait(false);
264-
265-
return ShapeResult();
270+
throw new InvalidOperationException(CosmosStrings.PartitionKeyMissing);
266271
}
267272

268-
return false;
273+
EntityFrameworkEventSource.Log.QueryExecuting();
274+
275+
_item = await _cosmosQueryContext.CosmosClient.ExecuteReadItemAsync(
276+
_readItemExpression.Container,
277+
partitionKey,
278+
resourceId,
279+
_cancellationToken)
280+
.ConfigureAwait(false);
281+
282+
return ShapeResult();
269283
}
284+
285+
return false;
270286
}
271287
catch (Exception exception)
272288
{
273289
_queryLogger.QueryIterationFailed(_contextType, exception);
274290

275291
throw;
276292
}
293+
finally
294+
{
295+
if (_concurrencyDetectionEnabled)
296+
{
297+
_cosmosQueryContext.ConcurrencyDetector.ExitCriticalSection();
298+
}
299+
}
277300
}
278301

279302
public void Dispose()

0 commit comments

Comments
 (0)