Skip to content

Commit 0b5c43a

Browse files
marcusbooyahmarcusbooyah
committed
Updated resource management with ConcurrentDictionary
Updated the `ResourceManager.cs` file to improve resource management. Removed the `System.Net` namespace and added the `System.Collections.Concurrent` namespace. Introduced a new `ConcurrentDictionary` named `reconcileTokens` to keep track of the `Status` property of resources. Updated the `WatchAsync`, `ReconcileAsync`, `DeletedAsync`, and `StatusModifiedAsync` methods to add or update the `reconcileTokens` dictionary with a `CancellationTokenSource` linked to the `cancellationToken` for each resource, handle `OperationCanceledException`, and clear the `reconcileTokens` dictionary in the `finally` block. The `DeletedAsync` method now cancels the token from the `reconcileTokens` dictionary for the deleted resource.
1 parent bd8ec13 commit 0b5c43a

File tree

1 file changed

+82
-14
lines changed

1 file changed

+82
-14
lines changed

‎src/Neon.Operator/ResourceManager/ResourceManager.cs

Lines changed: 82 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -16,11 +16,11 @@
1616
// limitations under the License.
1717

1818
using System;
19+
using System.Collections.Concurrent;
1920
using System.Collections.Generic;
2021
using System.Diagnostics;
2122
using System.Diagnostics.Contracts;
2223
using System.Linq;
23-
using System.Net;
2424
using System.Reflection;
2525
using System.Text.Json;
2626
using System.Text.RegularExpressions;
@@ -199,6 +199,7 @@ static ResourceManager()
199199
private Task watcherTask;
200200
private CancellationTokenSource watcherTcs;
201201
private EventQueue<TEntity, TController> eventQueue;
202+
private ConcurrentDictionary<string, CancellationTokenSource> reconcileTokens;
202203

203204
/// <summary>
204205
/// Default constructor.
@@ -684,16 +685,6 @@ private async Task WatchAsync(CancellationToken cancellationToken)
684685
{
685686
await SyncContext.Clear;
686687

687-
//-----------------------------------------------------------------
688-
// We're going to use this dictionary to keep track of the [Status]
689-
// property of the resources we're watching so we can distinguish
690-
// between changes to the status vs. changes to anything else in
691-
// the resource.
692-
//
693-
// The dictionary simply holds the status property serialized to
694-
// JSON, with these keyed by resource name. Note that the resource
695-
// entities might not have a [Status] property.
696-
697688
var entityType = typeof(TEntity);
698689
var statusGetter = entityType.GetProperty("Status")?.GetMethod;
699690

@@ -747,9 +738,24 @@ private async Task WatchAsync(CancellationToken cancellationToken)
747738
await finalizerManager.RegisterAllFinalizersAsync(resource, cancellationToken: cancellationToken);
748739
}
749740

741+
var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
742+
743+
reconcileTokens.AddOrUpdate(
744+
key: resource.Uid(),
745+
addValueFactory: (id) =>
746+
{
747+
return cts;
748+
749+
},
750+
updateValueFactory: (id, token) =>
751+
{
752+
return cts;
753+
754+
});
755+
750756
using (metrics.ReconcileTimeSeconds.NewTimer())
751757
{
752-
result = await CreateController(scope.ServiceProvider).ReconcileAsync(resource, cancellationToken: cancellationToken);
758+
result = await CreateController(scope.ServiceProvider).ReconcileAsync(resource, cancellationToken: cts.Token);
753759
}
754760
}
755761
catch (RequeueException e)
@@ -782,6 +788,13 @@ await eventQueue.RequeueAsync(
782788
return;
783789

784790
}
791+
catch (OperationCanceledException e)
792+
{
793+
logger?.LogErrorEx(e);
794+
metrics.ReconcileErrorsTotal.Inc();
795+
796+
return;
797+
}
785798
catch (Exception e)
786799
{
787800
metrics.ReconcileErrorsTotal.Inc();
@@ -804,6 +817,10 @@ await eventQueue.RequeueAsync(
804817
return;
805818
}
806819
}
820+
finally
821+
{
822+
reconcileTokens.TryRemove(resource.Uid(), out _);
823+
}
807824

808825
break;
809826

@@ -813,6 +830,11 @@ await eventQueue.RequeueAsync(
813830
{
814831
metrics.DeleteEventsTotal?.Inc();
815832

833+
if (reconcileTokens.TryGetValue(resource.Uid(), out var token))
834+
{
835+
await token.CancelAsync();
836+
}
837+
816838
using (metrics.DeleteTimeSeconds.NewTimer())
817839
{
818840
await CreateController(scope.ServiceProvider).DeletedAsync(resource, cancellationToken: cancellationToken);
@@ -838,9 +860,24 @@ await eventQueue.RequeueAsync(
838860
{
839861
metrics.ReconcileEventsTotal?.Inc();
840862

863+
var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
864+
865+
reconcileTokens.AddOrUpdate(
866+
key: resource.Uid(),
867+
addValueFactory: (id) =>
868+
{
869+
return cts;
870+
871+
},
872+
updateValueFactory: (id, token) =>
873+
{
874+
return cts;
875+
876+
});
877+
841878
using (metrics.ReconcileTimeSeconds.NewTimer())
842879
{
843-
result = await CreateController(scope.ServiceProvider).ReconcileAsync(resource, cancellationToken: cancellationToken);
880+
result = await CreateController(scope.ServiceProvider).ReconcileAsync(resource, cancellationToken: cts.Token);
844881
}
845882
}
846883
catch (RequeueException e)
@@ -872,6 +909,13 @@ await eventQueue.RequeueAsync(
872909

873910
return;
874911
}
912+
catch (OperationCanceledException e)
913+
{
914+
logger?.LogErrorEx(e);
915+
metrics.ReconcileErrorsTotal.Inc();
916+
917+
return;
918+
}
875919
catch (Exception e)
876920
{
877921
metrics.ReconcileErrorsTotal?.Inc();
@@ -894,6 +938,11 @@ await eventQueue.RequeueAsync(
894938
return;
895939
}
896940
}
941+
finally
942+
{
943+
reconcileTokens.TryRemove(resource.Uid(), out _);
944+
}
945+
897946
break;
898947

899948
case ModifiedEventType.Finalizing:
@@ -961,16 +1010,35 @@ await eventQueue.RequeueAsync(
9611010
{
9621011
metrics.StatusModifiedTotal?.Inc();
9631012

1013+
var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
1014+
1015+
reconcileTokens.AddOrUpdate(
1016+
key: resource.Uid(),
1017+
addValueFactory: (id) =>
1018+
{
1019+
return cts;
1020+
1021+
},
1022+
updateValueFactory: (id, token) =>
1023+
{
1024+
return cts;
1025+
1026+
});
1027+
9641028
using (metrics.StatusModifiedTimeSeconds.NewTimer())
9651029
{
966-
await CreateController(scope.ServiceProvider).StatusModifiedAsync(resource, cancellationToken: cancellationToken);
1030+
await CreateController(scope.ServiceProvider).StatusModifiedAsync(resource, cancellationToken: cts.Token);
9671031
}
9681032
}
9691033
catch (Exception e)
9701034
{
9711035
metrics.StatusModifiedErrorsTotal?.Inc();
9721036
logger?.LogErrorEx(e);
9731037
}
1038+
finally
1039+
{
1040+
reconcileTokens.TryRemove(resource.Uid(), out _);
1041+
}
9741042
}
9751043
else
9761044
{

0 commit comments

Comments
 (0)