Change the cancellation detection strategy (#203)

boma96 · web-flow · commit 54365e31aa9e · 2025-04-02T11:55:07.000+02:00
* Change the cancellation detection strategy

* Detect if task is cancelled due to background service shutdown

* Bump version
diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj
@@ -6,7 +6,7 @@
 		<Authors>Codaxy</Authors>
 		<Company>Codaxy</Company>
 		<PackageId>ConductorSharp.Client</PackageId>
-		<Version>3.4.0</Version>
+		<Version>3.4.1</Version>
 		<Description>Client library for Netflix Conductor, with some additional quality of life features.</Description>
 		<RepositoryUrl>https://github.com/codaxy/conductor-sharp</RepositoryUrl>
 		<PackageTags>netflix;conductor</PackageTags>
diff --git a/src/ConductorSharp.Engine/Behaviors/RequestResponseLoggingBehavior.cs b/src/ConductorSharp.Engine/Behaviors/RequestResponseLoggingBehavior.cs
@@ -11,14 +11,20 @@
 
 namespace ConductorSharp.Engine.Behaviors
 {
+    // TODO: Consider removing this
     public class RequestResponseLoggingBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
         where TRequest : IRequest<TResponse>
     {
         private readonly ILogger<RequestResponseLoggingBehavior<TRequest, TResponse>> _logger;
+        private readonly ConductorSharpExecutionContext _context;
 
-        public RequestResponseLoggingBehavior(ILogger<RequestResponseLoggingBehavior<TRequest, TResponse>> logger)
+        public RequestResponseLoggingBehavior(
+            ILogger<RequestResponseLoggingBehavior<TRequest, TResponse>> logger,
+            ConductorSharpExecutionContext context
+        )
         {
             _logger = logger;
+            _context = context;
         }
 
         public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TResponse> next, CancellationToken cancellationToken)
@@ -52,14 +58,9 @@ public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TRe
 
                 return response;
             }
-            catch (TaskCanceledException)
+            catch (OperationCanceledException) when (_context.TaskId != null)
             {
-                _logger.LogWarning(
-                    $"Request {{Request}} cancelled with payload {{@{requestName}}} and with id {{RequestId}}",
-                    requestName,
-                    request,
-                    requestId
-                );
+                // Simply rethrow and do not log in order for cancellation notifier to work
                 throw;
             }
             catch (Exception exc)
diff --git a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj
@@ -6,7 +6,7 @@
     <Authors>Codaxy</Authors>
 	<Company>Codaxy</Company>
 	<PackageId>ConductorSharp.Engine</PackageId>
-	<Version>3.4.0</Version>
+	<Version>3.4.1</Version>
     <Description>Client library for Netflix Conductor, with some additional quality of life features.</Description>
 	<RepositoryUrl>https://github.com/codaxy/conductor-sharp</RepositoryUrl>
 	<PackageTags>netflix;conductor</PackageTags>
diff --git a/src/ConductorSharp.Engine/ExecutionManager.cs b/src/ConductorSharp.Engine/ExecutionManager.cs
@@ -140,21 +140,7 @@ private async Task PollAndHandle(TaskToWorker scheduledWorker, CancellationToken
                 return;
             }
 
-            try
-            {
-                using var tokenHolder = _cancellationNotifier.GetCancellationToken(pollResponse.TaskId, cancellationToken);
-                await ProcessPolledTask(pollResponse, workerId, scheduledWorker, tokenHolder.CancellationToken);
-            }
-            catch (TaskCanceledException)
-            {
-                _logger.LogWarning(
-                    "Polled task {Task}(id={TaskId}) of workflow {Workflow}(id={WorkflowId}) is cancelled",
-                    pollResponse.TaskDefName,
-                    pollResponse.TaskId,
-                    pollResponse.WorkflowType,
-                    pollResponse.WorkflowInstanceId
-                );
-            }
+            await ProcessPolledTask(pollResponse, workerId, scheduledWorker, cancellationToken);
         }
 
         private async Task ProcessPolledTask(
@@ -164,6 +150,8 @@ private async Task ProcessPolledTask(
             CancellationToken cancellationToken
         )
         {
+            using var tokenHolder = _cancellationNotifier.GetCancellationToken(pollResponse.TaskId, cancellationToken);
+
             try
             {
                 if (!string.IsNullOrEmpty(pollResponse.ExternalInputPayloadStoragePath))
@@ -178,7 +166,7 @@ CancellationToken cancellationToken
                     );
 
                     // TODO: iffy
-                    var file = await _externalPayloadService.GetExternalStorageDataAsync(externalStorageLocation.Path, cancellationToken);
+                    var file = await _externalPayloadService.GetExternalStorageDataAsync(externalStorageLocation.Path, tokenHolder.CancellationToken);
 
                     using TextReader textReader = new StreamReader(file.Stream);
                     var json = await textReader.ReadToEndAsync();
@@ -211,7 +199,7 @@ CancellationToken cancellationToken
                     context.WorkerId = workerId;
                 }
 
-                var response = await mediator.Send(inputData, cancellationToken);
+                var response = await mediator.Send(inputData, tokenHolder.CancellationToken);
 
                 await _taskManager.UpdateAsync(
                     new TaskResult
@@ -221,13 +209,28 @@ await _taskManager.UpdateAsync(
                         OutputData = SerializationHelper.ObjectToDictionary(response, ConductorConstants.IoJsonSerializerSettings),
                         WorkflowInstanceId = pollResponse.WorkflowInstanceId
                     },
-                    cancellationToken
+                    tokenHolder.CancellationToken
                 );
             }
-            catch (TaskCanceledException)
+            catch (OperationCanceledException) when (tokenHolder.IsCancellationRequestedByNotifier)
             {
-                // Propagate this exception to outer handler
-                throw;
+                _logger.LogWarning(
+                    "Polled task {Task}(id={TaskId}) of workflow {Workflow}(id={WorkflowId}) is cancelled",
+                    pollResponse.TaskDefName,
+                    pollResponse.TaskId,
+                    pollResponse.WorkflowType,
+                    pollResponse.WorkflowInstanceId
+                );
+            }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) // This is fine since we know cancellationToken comes from background service
+            {
+                _logger.LogWarning(
+                    "Cancelling task {Task}(id={TaskId}) of workflow {Workflow}(id={WorkflowId}) due to background service shutdown",
+                    pollResponse.TaskDefName,
+                    pollResponse.TaskId,
+                    pollResponse.WorkflowType,
+                    pollResponse.WorkflowInstanceId
+                );
             }
             catch (Exception exception)
             {
@@ -256,10 +259,10 @@ await Task.WhenAll(
                                 OutputData = SerializationHelper.ObjectToDictionary(errorMessage, ConductorConstants.IoJsonSerializerSettings),
                                 WorkflowInstanceId = pollResponse?.WorkflowInstanceId
                             },
-                            cancellationToken
+                            tokenHolder.CancellationToken
                         ),
-                        _taskManager.LogAsync(pollResponse.TaskId, exception.Message, cancellationToken),
-                        _taskManager.LogAsync(pollResponse.TaskId, exception.StackTrace, cancellationToken)
+                        _taskManager.LogAsync(pollResponse.TaskId, exception.Message, tokenHolder.CancellationToken),
+                        _taskManager.LogAsync(pollResponse.TaskId, exception.StackTrace, tokenHolder.CancellationToken)
                     ]
                 );
             }
diff --git a/src/ConductorSharp.Engine/Interface/ICancellationNotifier.cs b/src/ConductorSharp.Engine/Interface/ICancellationNotifier.cs
@@ -8,6 +8,7 @@ public interface ICancellationNotifier
         public interface ICancellationTokenHolder : IDisposable
         {
             CancellationToken CancellationToken { get; }
+            bool IsCancellationRequestedByNotifier { get; }
         }
 
         ICancellationTokenHolder GetCancellationToken(string taskId, CancellationToken engineCancellationToken);
diff --git a/src/ConductorSharp.Engine/Service/NoOpCancellationNotifier.cs b/src/ConductorSharp.Engine/Service/NoOpCancellationNotifier.cs
@@ -10,6 +10,8 @@ internal class PassthroughCancellationTokenHolder(CancellationToken cancellation
     {
         public CancellationToken CancellationToken { get; } = cancellationToken;
 
+        public bool IsCancellationRequestedByNotifier => false;
+
         public void Dispose() { }
     }
 
diff --git a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj
@@ -4,7 +4,7 @@
     <TargetFramework>net6.0</TargetFramework>
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
-    <Version>3.4.0</Version>
+    <Version>3.4.1</Version>
     <Authors>Codaxy</Authors>
     <Company>Codaxy</Company>
   </PropertyGroup>
diff --git a/src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaCancellationNotifier.cs b/src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaCancellationNotifier.cs
@@ -16,6 +16,8 @@ internal class CancellationTokenSourceHolder : ICancellationNotifier.ICancellati
 
             public CancellationToken CancellationToken { get; }
 
+            public bool IsCancellationRequestedByNotifier => _notifier.IsCancellationRequested(_taskId);
+
             public CancellationTokenSourceHolder(CancellationToken cancellationToken, string taskId, KafkaCancellationNotifier notifier)
             {
                 CancellationToken = cancellationToken;
@@ -26,10 +28,21 @@ public CancellationTokenSourceHolder(CancellationToken cancellationToken, string
             public void Dispose() => _notifier.ClearTaskCts(_taskId);
         }
 
+        private class TaskCancellationInfo
+        {
+            public TaskCancellationInfo(CancellationTokenSource cancellationTokenSource)
+            {
+                CancellationTokenSource = cancellationTokenSource;
+            }
+
+            public CancellationTokenSource CancellationTokenSource { get; }
+            public bool IsCancellationRequested { get; set; }
+        }
+
         private readonly HashSet<string> _tasks;
         private readonly object _lock = new();
         private readonly ILogger<KafkaCancellationNotifier> _logger;
-        private readonly Dictionary<string, CancellationTokenSource> _taskIdToCtsMap = new();
+        private readonly Dictionary<string, TaskCancellationInfo> _taskIdToInfoMap = new();
 
         public KafkaCancellationNotifier(IEnumerable<TaskToWorker> tasks, ILogger<KafkaCancellationNotifier> logger)
         {
@@ -39,8 +52,8 @@ public KafkaCancellationNotifier(IEnumerable<TaskToWorker> tasks, ILogger<KafkaC
 
         public ICancellationNotifier.ICancellationTokenHolder GetCancellationToken(string taskId, CancellationToken engineCancellationToken)
         {
-            var cts = CreateCts(taskId, engineCancellationToken);
-            return new CancellationTokenSourceHolder(cts.Token, taskId, this);
+            var token = CreateTaskCancellationInfoAndGetToken(taskId, engineCancellationToken);
+            return new CancellationTokenSourceHolder(token, taskId, this);
         }
 
         public void HandleKafkaEvent(TaskStatusModel taskStatusModel)
@@ -52,56 +65,65 @@ public void HandleKafkaEvent(TaskStatusModel taskStatusModel)
             )
                 return;
 
-            var cts = GetCts(taskStatusModel.TaskId);
-            if (cts is null)
+            TryToCancelTask(taskStatusModel);
+        }
+
+        private CancellationToken CreateTaskCancellationInfoAndGetToken(string taskId, CancellationToken engineCancellationToken = default)
+        {
+            CancellationToken token;
+
+            lock (_lock)
             {
-                _logger.LogWarning(
-                    "Unable to cancel task {TaskId} of workflow {WorkflowId}",
-                    taskStatusModel.TaskId,
-                    taskStatusModel.WorkflowInstanceId
+                var info = _taskIdToInfoMap[taskId] = new TaskCancellationInfo(
+                    CancellationTokenSource.CreateLinkedTokenSource(engineCancellationToken)
                 );
-                return;
+                token = info.CancellationTokenSource.Token;
             }
 
-            cts.Cancel();
+            return token;
         }
 
-        private CancellationTokenSource CreateCts(string taskId, CancellationToken engineCancellationToken = default)
+        private void TryToCancelTask(TaskStatusModel taskStatusModel)
         {
-            CancellationTokenSource cts;
-            var stopwatch = Stopwatch.StartNew();
+            TaskCancellationInfo? info;
 
             lock (_lock)
             {
-                cts = _taskIdToCtsMap[taskId] = CancellationTokenSource.CreateLinkedTokenSource(engineCancellationToken);
+                info = _taskIdToInfoMap.GetValueOrDefault(taskStatusModel.TaskId);
             }
-            _logger.LogDebug("CancellationTokenSource creation time {ElapsedMs}ms", stopwatch.ElapsedMilliseconds);
 
-            return cts;
+            if (info is null)
+            {
+                _logger.LogWarning(
+                    "Unable to cancel task {TaskId} of workflow {WorkflowId}",
+                    taskStatusModel.TaskId,
+                    taskStatusModel.WorkflowInstanceId
+                );
+                return;
+            }
+
+            lock (_lock)
+            {
+                info.IsCancellationRequested = true;
+                info.CancellationTokenSource.Cancel();
+            }
         }
 
-        private CancellationTokenSource? GetCts(string taskId)
+        private void ClearTaskCts(string taskId)
         {
-            CancellationTokenSource? cts;
-            var stopwatch = Stopwatch.StartNew();
-
             lock (_lock)
             {
-                cts = _taskIdToCtsMap.GetValueOrDefault(taskId);
+                _taskIdToInfoMap[taskId].CancellationTokenSource.Dispose();
+                _taskIdToInfoMap.Remove(taskId);
             }
-            _logger.LogDebug("CancellationTokenSource get time {ElapsedMs}ms", stopwatch.ElapsedMilliseconds);
-
-            return cts;
         }
 
-        private void ClearTaskCts(string taskId)
+        private bool IsCancellationRequested(string taskId)
         {
-            var stopwatch = Stopwatch.StartNew();
             lock (_lock)
             {
-                _taskIdToCtsMap.Remove(taskId);
+                return _taskIdToInfoMap[taskId].IsCancellationRequested;
             }
-            _logger.LogDebug("CancellationTokenSource removal time {ElapsedMs}ms", stopwatch.ElapsedMilliseconds);
         }
     }
 }
diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj
@@ -7,7 +7,7 @@
     <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
     <Authors>Codaxy</Authors>
     <Company>Codaxy</Company>
-    <Version>3.4.0</Version>
+    <Version>3.4.1</Version>
   </PropertyGroup>
 
 	<ItemGroup>

Original file line number	Diff line number	Diff line change
`@@ -11,14 +11,20 @@`
`11`	`11`
`12`	`12`	`namespace ConductorSharp.Engine.Behaviors`
`13`	`13`	`{`
	`14`	`+ // TODO: Consider removing this`
`14`	`15`	`public class RequestResponseLoggingBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>`
`15`	`16`	`where TRequest : IRequest<TResponse>`
`16`	`17`	`{`
`17`	`18`	`private readonly ILogger<RequestResponseLoggingBehavior<TRequest, TResponse>> _logger;`
	`19`	`+ private readonly ConductorSharpExecutionContext _context;`
`18`	`20`
`19`		`- public RequestResponseLoggingBehavior(ILogger<RequestResponseLoggingBehavior<TRequest, TResponse>> logger)`
	`21`	`+ public RequestResponseLoggingBehavior(`
	`22`	`+ ILogger<RequestResponseLoggingBehavior<TRequest, TResponse>> logger,`
	`23`	`+ ConductorSharpExecutionContext context`
	`24`	`+ )`
`20`	`25`	`{`
`21`	`26`	`_logger = logger;`
	`27`	`+ _context = context;`
`22`	`28`	`}`
`23`	`29`
`24`	`30`	`public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TResponse> next, CancellationToken cancellationToken)`
`@@ -52,14 +58,9 @@ public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TRe`
`52`	`58`
`53`	`59`	`return response;`
`54`	`60`	`}`
`55`		`- catch (TaskCanceledException)`
	`61`	`+ catch (OperationCanceledException) when (_context.TaskId != null)`
`56`	`62`	`{`
`57`		`- _logger.LogWarning(`
`58`		`- $"Request {{Request}} cancelled with payload {{@{requestName}}} and with id {{RequestId}}",`
`59`		`- requestName,`
`60`		`- request,`
`61`		`- requestId`
`62`		`- );`
	`63`	`+ // Simply rethrow and do not log in order for cancellation notifier to work`
`63`	`64`	`throw;`
`64`	`65`	`}`
`65`	`66`	`catch (Exception exc)`
Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,7 @@ public interface ICancellationNotifier`
`8`	`8`	`public interface ICancellationTokenHolder : IDisposable`
`9`	`9`	`{`
`10`	`10`	`CancellationToken CancellationToken { get; }`
	`11`	`+ bool IsCancellationRequestedByNotifier { get; }`
`11`	`12`	`}`
`12`	`13`
`13`	`14`	`ICancellationTokenHolder GetCancellationToken(string taskId, CancellationToken engineCancellationToken);`
Original file line number	Diff line number	Diff line change
`@@ -10,6 +10,8 @@ internal class PassthroughCancellationTokenHolder(CancellationToken cancellation`
`10`	`10`	`{`
`11`	`11`	`public CancellationToken CancellationToken { get; } = cancellationToken;`
`12`	`12`
	`13`	`+ public bool IsCancellationRequestedByNotifier => false;`
	`14`	`+`
`13`	`15`	`public void Dispose() { }`
`14`	`16`	`}`
`15`	`17`
Original file line number	Diff line number	Diff line change
`@@ -16,6 +16,8 @@ internal class CancellationTokenSourceHolder : ICancellationNotifier.ICancellati`
`16`	`16`
`17`	`17`	`public CancellationToken CancellationToken { get; }`
`18`	`18`
	`19`	`+ public bool IsCancellationRequestedByNotifier => _notifier.IsCancellationRequested(_taskId);`
	`20`	`+`
`19`	`21`	`public CancellationTokenSourceHolder(CancellationToken cancellationToken, string taskId, KafkaCancellationNotifier notifier)`
`20`	`22`	`{`
`21`	`23`	`CancellationToken = cancellationToken;`
`@@ -26,10 +28,21 @@ public CancellationTokenSourceHolder(CancellationToken cancellationToken, string`
`26`	`28`	`public void Dispose() => _notifier.ClearTaskCts(_taskId);`
`27`	`29`	`}`
`28`	`30`
	`31`	`+ private class TaskCancellationInfo`
	`32`	`+ {`
	`33`	`+ public TaskCancellationInfo(CancellationTokenSource cancellationTokenSource)`
	`34`	`+ {`
	`35`	`+ CancellationTokenSource = cancellationTokenSource;`
	`36`	`+ }`
	`37`	`+`
	`38`	`+ public CancellationTokenSource CancellationTokenSource { get; }`
	`39`	`+ public bool IsCancellationRequested { get; set; }`
	`40`	`+ }`
	`41`	`+`
`29`	`42`	`private readonly HashSet<string> _tasks;`
`30`	`43`	`private readonly object _lock = new();`
`31`	`44`	`private readonly ILogger<KafkaCancellationNotifier> _logger;`
`32`		`- private readonly Dictionary<string, CancellationTokenSource> _taskIdToCtsMap = new();`
	`45`	`+ private readonly Dictionary<string, TaskCancellationInfo> _taskIdToInfoMap = new();`
`33`	`46`
`34`	`47`	`public KafkaCancellationNotifier(IEnumerable<TaskToWorker> tasks, ILogger<KafkaCancellationNotifier> logger)`
`35`	`48`	`{`
`@@ -39,8 +52,8 @@ public KafkaCancellationNotifier(IEnumerable<TaskToWorker> tasks, ILogger<KafkaC`
`39`	`52`
`40`	`53`	`public ICancellationNotifier.ICancellationTokenHolder GetCancellationToken(string taskId, CancellationToken engineCancellationToken)`
`41`	`54`	`{`
`42`		`- var cts = CreateCts(taskId, engineCancellationToken);`
`43`		`- return new CancellationTokenSourceHolder(cts.Token, taskId, this);`
	`55`	`+ var token = CreateTaskCancellationInfoAndGetToken(taskId, engineCancellationToken);`
	`56`	`+ return new CancellationTokenSourceHolder(token, taskId, this);`
`44`	`57`	`}`
`45`	`58`
`46`	`59`	`public void HandleKafkaEvent(TaskStatusModel taskStatusModel)`
`@@ -52,56 +65,65 @@ public void HandleKafkaEvent(TaskStatusModel taskStatusModel)`
`52`	`65`	`)`
`53`	`66`	`return;`
`54`	`67`
`55`		`- var cts = GetCts(taskStatusModel.TaskId);`
`56`		`- if (cts is null)`
	`68`	`+ TryToCancelTask(taskStatusModel);`
	`69`	`+ }`
	`70`	`+`
	`71`	`+ private CancellationToken CreateTaskCancellationInfoAndGetToken(string taskId, CancellationToken engineCancellationToken = default)`
	`72`	`+ {`
	`73`	`+ CancellationToken token;`
	`74`	`+`
	`75`	`+ lock (_lock)`
`57`	`76`	`{`
`58`		`- _logger.LogWarning(`
`59`		`- "Unable to cancel task {TaskId} of workflow {WorkflowId}",`
`60`		`- taskStatusModel.TaskId,`
`61`		`- taskStatusModel.WorkflowInstanceId`
	`77`	`+ var info = _taskIdToInfoMap[taskId] = new TaskCancellationInfo(`
	`78`	`+ CancellationTokenSource.CreateLinkedTokenSource(engineCancellationToken)`
`62`	`79`	`);`
`63`		`- return;`
	`80`	`+ token = info.CancellationTokenSource.Token;`
`64`	`81`	`}`
`65`	`82`
`66`		`- cts.Cancel();`
	`83`	`+ return token;`
`67`	`84`	`}`
`68`	`85`
`69`		`- private CancellationTokenSource CreateCts(string taskId, CancellationToken engineCancellationToken = default)`
	`86`	`+ private void TryToCancelTask(TaskStatusModel taskStatusModel)`
`70`	`87`	`{`
`71`		`- CancellationTokenSource cts;`
`72`		`- var stopwatch = Stopwatch.StartNew();`
	`88`	`+ TaskCancellationInfo? info;`
`73`	`89`
`74`	`90`	`lock (_lock)`
`75`	`91`	`{`
`76`		`- cts = _taskIdToCtsMap[taskId] = CancellationTokenSource.CreateLinkedTokenSource(engineCancellationToken);`
	`92`	`+ info = _taskIdToInfoMap.GetValueOrDefault(taskStatusModel.TaskId);`
`77`	`93`	`}`
`78`		`- _logger.LogDebug("CancellationTokenSource creation time {ElapsedMs}ms", stopwatch.ElapsedMilliseconds);`
`79`	`94`
`80`		`- return cts;`
	`95`	`+ if (info is null)`
	`96`	`+ {`
	`97`	`+ _logger.LogWarning(`
	`98`	`+ "Unable to cancel task {TaskId} of workflow {WorkflowId}",`
	`99`	`+ taskStatusModel.TaskId,`
	`100`	`+ taskStatusModel.WorkflowInstanceId`
	`101`	`+ );`
	`102`	`+ return;`
	`103`	`+ }`
	`104`	`+`
	`105`	`+ lock (_lock)`
	`106`	`+ {`
	`107`	`+ info.IsCancellationRequested = true;`
	`108`	`+ info.CancellationTokenSource.Cancel();`
	`109`	`+ }`
`81`	`110`	`}`
`82`	`111`
`83`		`- private CancellationTokenSource? GetCts(string taskId)`
	`112`	`+ private void ClearTaskCts(string taskId)`
`84`	`113`	`{`
`85`		`- CancellationTokenSource? cts;`
`86`		`- var stopwatch = Stopwatch.StartNew();`
`87`		`-`
`88`	`114`	`lock (_lock)`
`89`	`115`	`{`
`90`		`- cts = _taskIdToCtsMap.GetValueOrDefault(taskId);`
	`116`	`+ _taskIdToInfoMap[taskId].CancellationTokenSource.Dispose();`
	`117`	`+ _taskIdToInfoMap.Remove(taskId);`
`91`	`118`	`}`
`92`		`- _logger.LogDebug("CancellationTokenSource get time {ElapsedMs}ms", stopwatch.ElapsedMilliseconds);`
`93`		`-`
`94`		`- return cts;`
`95`	`119`	`}`
`96`	`120`
`97`		`- private void ClearTaskCts(string taskId)`
	`121`	`+ private bool IsCancellationRequested(string taskId)`
`98`	`122`	`{`
`99`		`- var stopwatch = Stopwatch.StartNew();`
`100`	`123`	`lock (_lock)`
`101`	`124`	`{`
`102`		`- _taskIdToCtsMap.Remove(taskId);`
	`125`	`+ return _taskIdToInfoMap[taskId].IsCancellationRequested;`
`103`	`126`	`}`
`104`		`- _logger.LogDebug("CancellationTokenSource removal time {ElapsedMs}ms", stopwatch.ElapsedMilliseconds);`
`105`	`127`	`}`
`106`	`128`	`}`
`107`	`129`	`}`