Polly: Using Bulkhead correctly

[galmok]

I have a list of objects that I have to send individually to a worker, but I can't send all at once as this would overload the system. It seems Bulkhead policy can help here, but I am not certain how. To make it more complex, if the worker throws certain exceptions, the object causing it can be retried. And this is async.

So what I need is to limit the number of concurrent running workers, retrying the failed ones a number of times.

I keep getting this error, though:

The bulkhead semaphore and queue are full and execution was rejected.

The list of objects can be upwards of half a million or more, and I have made a policies like this:

  var policy2 = Policy.Handle<Exception>().WaitAndRetryAsync(new[] {
      TimeSpan.FromSeconds(1),
      TimeSpan.FromSeconds(2),
      TimeSpan.FromSeconds(4)
    },
    (exception, timeSpan, context) =>
    {
      // do something
      Console.WriteLine("retrying");
    });

  var bulkheadPolicy = Policy.BulkheadAsync(3,1000000000);

  var wrappedPolicy = Policy.WrapAsync(bulkheadPolicy, policy2);

I used 1000000000 to ensure every object is passed to the worker, but if this was an infinite list, I could not use this solution.

I run the execute worker like this:

    await numberList.ForEachAsync(a =>
    {
      return wrappedPolicy.ExecuteAsync(() => DoWork(a));
    });

Where the ForEachAsync extension is made like this:

public static Task ForEachAsync<T>(this IEnumerable<T> sequence, Func<T, Task> action)
{
  return Task.WhenAll(sequence.Select(action));
}

So, I am not sure I use the policies correctly. Is there a better way, one that also works with infinite lists?

[reisenberger]

Hi @galmok . My initial impression is that BulkheadPolicy (in its current form) is probably not a fit for this. From the problem description it sounds as if what you need is parallelization limiting with blocking/backpressure on excess load. BulkheadPolicy (in its current form) offers parallelization limiting with load-shedding on excess load.

I'll try to set down some further thoughts in the next 24-48 hours.

[galmok]

I have my own algorithm now. A ForEachAsync that has a limit on how many tasks in parallel it keeps going. Together with the retry from Polly, I have a decently working system. :)

[older]

Hi @reisenberger

Just wanted to ask question about this:

My initial impression is that BulkheadPolicy (in its current form) is probably not a fit for this. From the problem description it sounds as if what you need is parallelization limiting with blocking/backpressure on excess load. BulkheadPolicy (in its current form) offers parallelization limiting with load-shedding on excess load.

Has anything changed since then? Does Polly provide policy suitable for this particular scenario today?

[reisenberger]

@older

My initial impression is that BulkheadPolicy (in its current form) is probably not a fit for this. From the problem description it sounds as if what you need is parallelization limiting with blocking/backpressure on excess load. BulkheadPolicy (in its current form) offers parallelization limiting with load-shedding on excess load.

Has anything changed since then? Does Polly provide policy suitable for this particular scenario today?

Nothing has changed in bulkhead policy since then. You can set the maxQueuingActions parameter to int.MaxValue, so you shouldn't run in to BulkheadRejectedException unless you have a very large number of items to queue. For clarity: Using the policy with a large queue buffers the items to execute, but this doesn't cause backpressure (it doesn't slow down the producers of those items).