Implement query throttling #485
Conversation
| await queryPermitter.Ask<QueryStartGranted>(RequestQueryStart.Instance); | ||
| try | ||
| { | ||
| return await factory.ExecuteWithTransactionAsync(level, token, handler); | ||
| } | ||
| finally | ||
| { | ||
| queryPermitter.Tell(ReturnQueryStart.Instance); | ||
| } |
There was a problem hiding this comment.
This is the query throttling mechanism, we always ask for a permission before we execute any DB operations
| await queryPermitter.Ask<QueryStartGranted>(RequestQueryStart.Instance); | ||
| try | ||
| { | ||
| return await factory.ConnectionFactory.ExecuteWithTransactionAsync(state, factory.IsolationLevel, factory.ShutdownToken, handler); | ||
| } | ||
| finally | ||
| { | ||
| queryPermitter.Tell(ReturnQueryStart.Instance); | ||
| } |
There was a problem hiding this comment.
This is the query throttling mechanism, we always ask for a permission before we execute any DB operations
| static async input => | ||
| { | ||
| return await input._dbStateHolder.ExecuteWithTransactionAsync( | ||
| return await input._dbStateHolder.ExecuteQueryWithTransactionAsync( |
There was a problem hiding this comment.
Replace all ExecuteWithTransactionAsync() method call inside the read journal with ExecuteQueryWithTransactionAsync() to make sure that all DB queries are throttled
| @@ -142,7 +145,8 @@ public static AkkaConfigurationBuilder WithSqlPersistence( | |||
| DatabaseMapping? databaseMapping = null, | |||
| TagMode? tagStorageMode = null, | |||
| bool? deleteCompatibilityMode = null, | |||
| bool? useWriterUuidColumn = null) | |||
| bool? useWriterUuidColumn = null, | |||
| int? maxConcurrentQueries = null) | |||
There was a problem hiding this comment.
Technically a breaking change without a method overload, but I don't think there's any other meta-packages built on top of Akka.Persistence.Sql.Hosting so it's probably fine as it's source-compatible
| CancellationToken token, | ||
| Func<AkkaDataConnection, CancellationToken, Task<T>> handler) | ||
| { | ||
| await queryPermitter.Ask<QueryStartGranted>(RequestQueryStart.Instance); |
There was a problem hiding this comment.
This should use a timeout or a CTS
| TState state, | ||
| Func<AkkaDataConnection, CancellationToken, TState, Task<T>> handler) | ||
| { | ||
| return factory.ConnectionFactory.ExecuteWithTransactionAsync(state, factory.IsolationLevel, factory.ShutdownToken, handler); | ||
| await queryPermitter.Ask<QueryStartGranted>(RequestQueryStart.Instance); |
There was a problem hiding this comment.
Should use a timeout or a CTS
Implement query throttling scheme, similar to akkadotnet/akka.net#6436
Changes
akka.persistence.query.journal.sql.max-concurrent-queriesHOCON settingQueryThrottlerclass