MongoDB async IIdGenerator for InsertOneAsync operation

user_15551492 0
Viewed 235

We have exposed a JobService (which is Dotnet Core service using kestrel) which does MongoDB updates for our object(JobDocument) and are seeing that in case of Post call to createJob which actually triggers MongoDB.Driver.MongoCollectionBase`1.InsertOneAsync there is a performance hit ie. around 200 concurrent requests make the service unresponsive. After debugging and using some tools, found that issue is in the api call InsertOneAsync. This is using our custom implementation for GenerateId, which uses sync function call FindOneAndUpdate.Is there an async way to GenerateId. How to accomplish this? For reference attaching the stack using tool Ben Blocking detector which points to same:

{"Timestamp":"2021-04-03T15:22:38.5854816+05:30","Level":"Warning","MessageTemplate":"Blocking method has been invoked and blocked, this can lead to threadpool starvation.\r\n{stackTrace}","Properties":{"stackTrace":"   

at System.Threading.Tasks.TplEventSource.TaskWaitBegin(Int32 OriginatingTaskSchedulerID, Int32 OriginatingTaskID, Int32 TaskID, TaskWaitBehavior Behavior, Int32 ContinueWithTaskID)\r\n   
at System.Threading.Tasks.Task.InternalWaitCore(Int32 millisecondsTimeout, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.Connections.TcpStreamFactory.CreateStream(EndPoint endPoint, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.Connections.BinaryConnection.OpenHelper(CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.Connections.BinaryConnection.Open(CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.ConnectionPools.ExclusiveConnectionPool.PooledConnection.Open(CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.ConnectionPools.ExclusiveConnectionPool.AcquiredConnection.Open(CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.Servers.Server.GetChannel(CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.Bindings.ServerChannelSource.GetChannel(CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.Bindings.ChannelSourceHandle.GetChannel(CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.Operations.RetryableWriteContext.Initialize(CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.Operations.RetryableWriteContext.Create(IWriteBinding binding, Boolean retryRequested, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.Operations.RetryableWriteOperationExecutor.Execute[TResult](IRetryableWriteOperation`1 operation, IWriteBinding binding, Boolean retryRequested, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.Core.Operations.FindAndModifyOperationBase`1.Execute(IWriteBinding binding, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.OperationExecutor.ExecuteWriteOperation[TResult](IWriteBinding binding, IWriteOperation`1 operation, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.ExecuteWriteOperation[TResult](IClientSessionHandle session, IWriteOperation`1 operation, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.FindOneAndUpdate[TProjection](IClientSessionHandle session, FilterDefinition`1 filter, UpdateDefinition`1 update, FindOneAndUpdateOptions`2 options, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.<>c__DisplayClass53_0`1.<FindOneAndUpdate>b__0(IClientSessionHandle session)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.UsingImplicitSession[TResult](Func`2 func, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.FindOneAndUpdate[TProjection](FilterDefinition`1 filter, UpdateDefinition`1 update, FindOneAndUpdateOptions`2 options, CancellationToken cancellationToken)\r\n   
at JobManager.Data.MongoDB.IdGenerator.JobIdGenerator.GenerateId(Object container, Object document)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.AssignId(TDocument document)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.ConvertWriteModelToWriteRequest(WriteModel`1 model, Int32 index)\r\n   
at System.Linq.Enumerable.SelectIterator[TSource,TResult](IEnumerable`1 source, Func`3 selector)+MoveNext()\r\n   
at System.Collections.Generic.List`1..ctor(IEnumerable`1 collection)\r\n   
at System.Linq.Enumerable.ToList[TSource](IEnumerable`1 source)\r\n
at MongoDB.Driver.Core.Operations.BulkMixedWriteOperation..ctor(CollectionNamespace collectionNamespace, IEnumerable`1 requests, MessageEncoderSettings messageEncoderSettings)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.CreateBulkWriteOperation(IClientSessionHandle session, IEnumerable`1 requests, BulkWriteOptions options)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.BulkWriteAsync(IClientSessionHandle session, IEnumerable`1 requests, BulkWriteOptions options, CancellationToken cancellationToken)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.BulkWriteAsync(IClientSessionHandle session, IEnumerable`1 requests, BulkWriteOptions options, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.<>c__DisplayClass25_0.<BulkWriteAsync>b__0(IClientSessionHandle session)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.UsingImplicitSessionAsync[TResult](Func`2 funcAsync, CancellationToken cancellationToken)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.UsingImplicitSessionAsync[TResult](Func`2 funcAsync, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.MongoCollectionImpl`1.BulkWriteAsync(IEnumerable`1 requests, BulkWriteOptions options, CancellationToken cancellationToken)\r\n   
at MongoDB.Driver.MongoCollectionBase`1.<>c__DisplayClass68_0.<InsertOneAsync>b__0(IEnumerable`1 requests, BulkWriteOptions bulkWriteOptions)\r\n   
at MongoDB.Driver.MongoCollectionBase`1.InsertOneAsync(TDocument document, InsertOneOptions options, Func`3 bulkWriteAsync)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at MongoDB.Driver.MongoCollectionBase`1.InsertOneAsync(TDocument document, InsertOneOptions options, Func`3 bulkWriteAsync)\r\n   
at MongoDB.Driver.MongoCollectionBase`1.InsertOneAsync(TDocument document, InsertOneOptions options, CancellationToken cancellationToken)\r\n   
at Code.DataProtection.JobManager.Data.MongoDB.Repositories.MongoJobRepository.Code.DataProtection.JobManager.Data.Contract.IJobRepository.CreateJob(Job job)\r\n 
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Code.DataProtection.JobManager.Data.MongoDB.Repositories.MongoJobRepository.Code.DataProtection.JobManager.Data.Contract.IJobRepository.CreateJob(Job job)\r\n   
at Code.DataProtection.JobManager.Services.JobService.CreateJob(Job job)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Code.DataProtection.JobManager.Services.JobService.CreateJob(Job job)\r\n   
at Code.DataProtection.JobManager.Services.JobService.Code.DataProtection.JobManager.Contract.Interfaces.IJobService.CreateJobWithWorkflow(Job job)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Code.DataProtection.JobManager.Services.JobService.Code.DataProtection.JobManager.Contract.Interfaces.IJobService.CreateJobWithWorkflow(Job job)\r\n   
at Code.DataProtection.JobManager.Api.Controllers.v2.JobsController.CreateJob(Job job)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Code.DataProtection.JobManager.Api.Controllers.v2.JobsController.CreateJob(Job job)\r\n   at lambda_method(Closure , Object , Object[] )\r\n   
at Microsoft.Extensions.Internal.ObjectMethodExecutor.Execute(Object target, Object[] parameters)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ActionMethodExecutor.TaskOfIActionResultExecutor.Execute(IActionResultTypeMapper mapper, ObjectMethodExecutor executor, Object controller, Object[] arguments)\r\n   at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ActionMethodExecutor.TaskOfIActionResultExecutor.Execute(IActionResultTypeMapper mapper, ObjectMethodExecutor executor, Object controller, Object[] arguments)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeActionMethodAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.Next(State& next, Scope& scope, Object& state, Boolean& isCompleted)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeNextActionFilterAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeNextActionFilterAwaitedAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Filters.ActionFilterAttribute.OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Microsoft.AspNetCore.Mvc.Filters.ActionFilterAttribute.OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.Next(State& next, Scope& scope, Object& state, Boolean& isCompleted)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeNextActionFilterAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeNextActionFilterAwaitedAsync()\r\n   
at Code.DataProtection.Common.Utils.Filters.ContextFilter.OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Code.DataProtection.Common.Utils.Filters.ContextFilter.OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.Next(State& next, Scope& scope, Object& state, Boolean& isCompleted)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeNextActionFilterAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.Next(State& next, Scope& scope, Object& state, Boolean& isCompleted)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeNextActionFilterAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.Next(State& next, Scope& scope, Object& state, Boolean& isCompleted)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeNextActionFilterAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeNextActionFilterAwaitedAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Controller.OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next)\r\n   
at Microsoft.AspNetCore.Mvc.Filters.ControllerActionFilter.OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next)\r\n  
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.Next(State& next, Scope& scope, Object& state, Boolean& isCompleted)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeInnerFilterAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.Next(State& next, Scope& scope, Object& state, Boolean& isCompleted)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.InvokeNextResourceFilter()\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.Next(State& next, Scope& scope, Object& state, Boolean& isCompleted)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.InvokeFilterPipelineAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.<InvokeAsync>g__Logged|17_1(ResourceInvoker invoker)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.<InvokeAsync>g__Logged|17_1(ResourceInvoker invoker)\r\n   
at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.InvokeAsync()\r\n   
at Microsoft.AspNetCore.Mvc.Routing.ActionEndpointFactory.<>c__DisplayClass7_0.<CreateRequestDelegate>b__0(HttpContext context)\r\n   
at Microsoft.AspNetCore.Routing.EndpointMiddleware.Invoke(HttpContext httpContext)\r\n   
at Microsoft.AspNetCore.StaticFiles.StaticFileMiddleware.Invoke(HttpContext context)\r\n   
at Swashbuckle.AspNetCore.SwaggerUI.SwaggerUIMiddleware.Invoke(HttpContext httpContext)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Swashbuckle.AspNetCore.SwaggerUI.SwaggerUIMiddleware.Invoke(HttpContext httpContext)\r\n   
at Swashbuckle.AspNetCore.Swagger.SwaggerMiddleware.Invoke(HttpContext httpContext, ISwaggerProvider swaggerProvider)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Swashbuckle.AspNetCore.Swagger.SwaggerMiddleware.Invoke(HttpContext httpContext, ISwaggerProvider swaggerProvider)\r\n   
at Microsoft.AspNetCore.Builder.UseMiddlewareExtensions.<>c__DisplayClass4_1.<UseMiddleware>b__2(HttpContext context)\r\n   
at Microsoft.AspNetCore.Routing.EndpointRoutingMiddleware.Invoke(HttpContext httpContext)\r\n   
at CorrelationId.CorrelationIdMiddleware.Invoke(HttpContext context, ICorrelationContextFactory correlationContextFactory)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at CorrelationId.CorrelationIdMiddleware.Invoke(HttpContext context, ICorrelationContextFactory correlationContextFactory)\r\n   
at Microsoft.AspNetCore.Builder.UseMiddlewareExtensions.<>c__DisplayClass4_1.<UseMiddleware>b__2(HttpContext context)\r\n   
at Code.DataProtection.Common.Utils.ExceptionHandler.ExceptionHandlerMiddleware.Invoke(HttpContext context)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Code.DataProtection.Common.Utils.ExceptionHandler.ExceptionHandlerMiddleware.Invoke(HttpContext context)\r\n   
at Ben.Diagnostics.BlockingDetectionMiddleware.Invoke(HttpContext httpContext)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Ben.Diagnostics.BlockingDetectionMiddleware.Invoke(HttpContext httpContext)\r\n   
at Microsoft.AspNetCore.Mvc.Versioning.ApiVersioningMiddleware.InvokeAsync(HttpContext context)\r\n   
at Microsoft.AspNetCore.HostFiltering.HostFilteringMiddleware.Invoke(HttpContext context)\r\n   
at Microsoft.AspNetCore.Hosting.HostingApplication.ProcessRequestAsync(Context context)\r\n   
at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.HttpProtocol.ProcessRequests[TContext](IHttpApplication`1 application)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.HttpProtocol.ProcessRequests[TContext](IHttpApplication`1 application)\r\n   
at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.HttpProtocol.ProcessRequestsAsync[TContext](IHttpApplication`1 application)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.HttpProtocol.ProcessRequestsAsync[TContext](IHttpApplication`1 application)\r\n   
at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.HttpConnection.ProcessRequestsAsync[TContext](IHttpApplication`1 httpApplication)\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.HttpConnection.ProcessRequestsAsync[TContext](IHttpApplication`1 httpApplication)\r\n   
at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.HttpConnectionMiddleware`1.OnConnectionAsync(ConnectionContext connectionContext)\r\n   
at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure.KestrelConnection.ExecuteAsync()\r\n   
at System.Runtime.CompilerServices.AsyncMethodBuilderCore.Start[TStateMachine](TStateMachine& stateMachine)\r\n   
at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure.KestrelConnection.ExecuteAsync()\r\n   
at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure.KestrelConnection.System.Threading.IThreadPoolWorkItem.Execute()\r\n   
at System.Threading.ThreadPoolWorkQueue.Dispatch()\r\n   
at System.Threading._ThreadPoolWaitCallback.PerformWaitCallback()\r\n","EventId":{"Id":6,"Name":"BlockingMethodCalled"},"SourceContext":"Ben.Diagnostics.BlockingMonitor","ActionId":"25c60a2a-f366-415b-8f16-66e10f134608","ActionName":"Code.DataProtection.JobManager.Api.Controllers.v2.JobsController.CreateJob (JobManager.Api)","RequestId":"0HM7MGOIK401C:00000001","RequestPath":"/api/jobmanager/v2.0/jobs","SpanId":"|11136163-4c80e7d34af376d0.","TraceId":"11136163-4c80e7d34af376d0","ParentId":"","ConnectionId":"0HM7MGOIK401C"}}
kestrel mongodb-.net-driver mongodb .net-core c#
1 Answers

i believe your performance issue stems from the technique used for generating incremental ids using mongodb server and not due to not being able to use async/await.

this is a well known scaling/performance limitation of sequential id generation due to mongodb having to lock the counter document in order to increment the value.

for ex: say a single FindOneAndUpdate operation takes 50ms to update the counter. now to do that from 200 threads simultaneously would take roughly 200 x 50ms = 10secs or more. i.e. only 1 thread/operation is allowed to write to the counter document at any given time and an exclusive lock is obtained per opertation/thread.

you have a few options here:

  • give up using incremental ids and use ObjectId instead
  • create a background service that pre-generates batches of sequential ids if the current stock is running low. then your JobIdGenerator will 'FindOneAndUpdate' the next unused id as IsUsed:true instead of trying to increment a single counter doc.

also since there's no IIdGenerator.GenerateIdAsync method, you might wanna take things in to your own hands and set the values from you repositories/whatever instead of using IIdGenerator when saving/creating new entities.

user_43684850
Related