[Question]: In what scenario should I use TryAddBuilder() instead of GetOrAddPipeline() for a registry?

[stambunan]

What are you wanting to achieve?

I'm trying to figure out what the primary difference between TryAddBuilder() and GetOrAddPipeline() when used with the registry.

During most of my testing, using both methods will allow me to update the pipeline whenever the cancellation token is cancelled.

The one thing I've noticed is that GetOrAddPipeline() will always return a single pipeline instance when called with _registry.GetPipeline("A");

With TryAddBuilder(), I'm thinking that it should also return a single cached pipeline instance when _registry.GetPipeline("A") is called.

1. If so, which should we use?

The primary difference I've noticed so far is that TryAddBuilder() can be registered once but when _registry.GetPipeline("A") is called by multiple threads at once, it seems that our _valueChecker.OnUpdate will contain more invocations than expected.

I think it might be because

Polly/src/Polly.Core/Registry/ResiliencePipelineRegistry.cs

Line 83 in 2d02db1

pipeline = GetOrAddPipeline(key, configure);

is called when we use TryAddBuilder() and

Polly/src/Polly.Core/Registry/ResiliencePipelineRegistry.cs

Line 125 in 2d02db1

return _pipelines.GetOrAdd(key, k =>

is called to create the pipeline. However, if multiple threads are running and calling _registry.GetPipeline() when the pipeline isn't cached and a builder is used, configure can be executed multiple times as a concurrent dictionary will only save one value. This causes issues with my event handlers as there are now more registered than I expected.

2. I'm not sure if I'm doing this incorrectly or whether I should just use GetOrAddPipeline() for this scenario?

What code or approach do you have so far?

var registry = new ResiliencePipelineRegistry<string>();

// The registration is only done once during startup which might be why GetOrAddPipeline caches a single pipeline instance

// Register builder for pipeline "A"
registry.TryAddBuilder("A", (builder, context) =>
{
    // Add the reload token. Tokens that are already canceled are ignored.
    var cancellationTokenSource = new CancellationTokenSource();
    
    // Register the callback to reload the builder 
    _valueChecker.OnUpdate += cancellationTokenSource.cancel;
    context.AddReloadToken(cancellationTokenSource.Token);

    // Define your pipeline
    builder.AddRetry(new CircuitBreakerStrategyOptions() {   
                    // _valueChecker is used
   });
});

// Option 2 using GetOrAddPipeline
registry.GetOrAddPipeline("A", (builder, context) =>
{
    // Add the reload token. Tokens that are already canceled are ignored.
    var cancellationTokenSource = new CancellationTokenSource();
    
    // Register the callback to reload the builder 
    _valueChecker.OnUpdate += cancellationTokenSource.cancel;

    context.AddReloadToken(cancellationTokenSource.Token);

    // Define your pipeline
    builder.AddRetry(new CircuitBreakerStrategyOptions() {   
                    // _valueChecker is used
   });
});

Additional context

No response

[stambunan]

It doesn't look like TryAddBuilder is thread safe as calling GetPipeline() with multiple threads can cause issues with dynamic reloading. At best, it seems to reload more often than necessary.

class Program
{
    private static ResiliencePipelineRegistry<string>? _registry;
    private static ResiliencePipelineRegistry<string>? _registryByAddingEndpoint;
    private static event Action OnCancellation;
    private static event Action OnCancellationByAddingEndpoint;
    
    async static Task Main(string[] args)
    {
        Console.WriteLine("Polly testing!");
        
        CreateRegistryUsingTryAddBuilder();
        CreateRegistryByAddingEndpoint();

        Parallel.For(0, 10, (index) =>
        {
            _registry!.GetPipeline("AddBuilder");
            _registryByAddingEndpoint!.GetPipeline("GetOrAddPipeline");
        });

        // Test using tasks
        // var tasks = new List<Task>();
        // for (int i = 0; i < 10; i++)
        // {
        //     tasks.Add(Task.Run(() => _registry!.GetPipeline("AddBuilder")));
        //     tasks.Add(Task.Run(() => _registryByAddingEndpoint!.GetPipeline("GetOrAddPipeline")));
        // }
        //
        // Task.WaitAll(tasks.ToArray());
        
        Console.WriteLine($"Number of OnCancellation invocations: {OnCancellation.GetInvocationList().Length}");
        Console.WriteLine($"Number of OnCancellationByAddingEndpoint invocations: {OnCancellationByAddingEndpoint.GetInvocationList().Length}");
    }

    private static void CreateRegistryByAddingEndpoint()
    {
        _registryByAddingEndpoint = new ResiliencePipelineRegistry<string>();

        _registryByAddingEndpoint.GetOrAddPipeline("GetOrAddPipeline", (builder, context) =>
        {
            var cancellationToken = new CancellationTokenSource();
            
            OnCancellationByAddingEndpoint += cancellationToken.Cancel;
            
            context.AddReloadToken(cancellationToken.Token);
            builder.AddRetry(new RetryStrategyOptions());
        });
    }

    private static void CreateRegistryUsingTryAddBuilder()
    {
        _registry = new ResiliencePipelineRegistry<string>();

        _registry.TryAddBuilder("AddBuilder", (builder, context) =>
        {
            var cancellationToken = new CancellationTokenSource();
            
            OnCancellation += cancellationToken.Cancel;
            
            context.AddReloadToken(cancellationToken.Token);
            builder.AddRetry(new RetryStrategyOptions());
        });
    }
}

With the above code, the number of registered OnCancellation will be more than one since multiple threads are creating the pipeline and trying to add it to the concurrent dictionary.

[martincostello]

when the pipeline isn't cached and a builder is used, configure can be executed multiple times as a concurrent dictionary will only save one value. This causes issues with my event handlers as there are now more registered than I expected.

This is typical with the usage of ConcurrentDictionary<,>.GetOrAdd - there's no guarantee that only one value will be created when the cache is cold.

Since a key/value can be inserted by another thread while valueFactory is generating a value, you cannot trust that just because valueFactory executed, its produced value will be inserted into the dictionary and returned. If you call GetOrAdd simultaneously on different threads, valueFactory may be called multiple times, but only one key/value pair will be added to the dictionary.

You're not using it wrong, and the Polly code is safe to be used concurrently, but the code of yours that it will call during the process of creation needs to be safe to use concurrently too. You'll need to adjust your application as appropriate to be safe in this way if it's accessed this way.

I can't give a specific recommendation of how to do that, as it depends entirely on what your application as actually does.

[stambunan]

We've ended up restructuring our code to create the pipeline during startup, using GetOrAdd to avoid any race conditions.

My confusion was that we had originally set up the registry once and added a TryAddBuilder to create the builder safely and I assumed GetPipeline() can be called by multiple threads afterwards. However, this caused the above issues and only after looking at the source code did I realize there was a concurrent dictionary used to create the pipelines. Most of the time that should be fine, but it caused issues with dynamic reloads since we ended up reloading the pipeline more times than necessary, when the valueFactory was executed.

This is typical with the usage of ConcurrentDictionary<,>.GetOrAdd - there's no guarantee that only one value will be created when the cache is cold.

Perhaps a lock should be introduced so we can call GetPipeline() safely the first time?

If not, then can we update the documentation that we should be careful on how GetPipeline is used with TryAddBuilder() as this can subtle issues with dynamic reloading?

[martincostello]

I think we'd prefer resolving this with documentation rather than a lock.

GetPipeline() would typically be used on the hot path, so a lock is not going to be good for overall application performance.

[stambunan]

Thanks. Documentation would definitely help.

I was thinking that adding a lock here,

Polly/src/Polly.Core/Registry/ResiliencePipelineRegistry.cs

Line 124 in 2d02db1

, just right before the getOrAdd is called would be helpful.

Since the line before returns the pipeline if found, the lock should only affect calls that are creating the pipeline the first time?

Another option is perhaps using Lazy here, https://andrewlock.net/making-getoradd-on-concurrentdictionary-thread-safe-using-lazy/, but that might be a bigger change to the dictionary.