Is it possible to change parallelOptions.MaxDegreeOfParallelism during execution of a Parallel.ForEach?

Below is a variant of the Parallel.ForEachAsync API that was introduced in .NET 6, that allows to configure dynamically the maximum degree of parallelism. It shares the same parameters and behavior with the native API, except from taking a derived version of the ParallelOptions as argument (DynamicParallelOptions). Changing the MaxDegreeOfParallelism results in a rapid adaptation of the currently active degree of parallelism.

This implementation is based on the idea of throttling the source of the Parallel.ForEachAsync API. The API itself is configured with maximum parallelism (Int32.MaxValue), but the actual parallelism is limited effectively by denying the loop from free access to the source elements. An element is propagated forward every time another element is processed. The throttling itself is performed with an unbounded SemaphoreSlim. Changing the maximum degree of parallelism is performed by calling the Release/WaitAsync methods of the semaphore.

/// <summary>
/// Executes a parallel for-each operation on an async-enumerable sequence,
/// enforcing a dynamic maximum degree of parallelism.
/// </summary>
public static Task DynamicParallelForEachAsync<TSource>(
    IAsyncEnumerable<TSource> source,
    DynamicParallelOptions options,
    Func<TSource, CancellationToken, ValueTask> body)
{
    if (source == null) throw new ArgumentNullException(nameof(source));
    if (options == null) throw new ArgumentNullException(nameof(options));
    if (body == null) throw new ArgumentNullException(nameof(body));

    var throttler = new SemaphoreSlim(options.MaxDegreeOfParallelism);
    options.DegreeOfParallelismChangedDelta += Options_ChangedDelta;
    void Options_ChangedDelta(object sender, int delta)
    {
        if (delta > 0)
            throttler.Release(delta);
        else
            for (int i = delta; i < 0; i++) throttler.WaitAsync();
    }

    async IAsyncEnumerable<TSource> GetThrottledSource()
    {
        await foreach (var item in source.ConfigureAwait(false))
        {
            await throttler.WaitAsync().ConfigureAwait(false);
            yield return item;
        }
    }

    return Parallel.ForEachAsync(GetThrottledSource(), options, async (item, ct) =>
    {
        try { await body(item, ct).ConfigureAwait(false); }
        finally { throttler.Release(); }
    }).ContinueWith(t =>
    {
        options.DegreeOfParallelismChangedDelta -= Options_ChangedDelta;
        return t;
    }, default, TaskContinuationOptions.DenyChildAttach, TaskScheduler.Default)
        .Unwrap();
}

/// <summary>
/// Stores options that configure the DynamicParallelForEachAsync method.
/// </summary>
public class DynamicParallelOptions : ParallelOptions
{
    private int _maxDegreeOfParallelism;

    public event EventHandler<int> DegreeOfParallelismChangedDelta;

    public DynamicParallelOptions()
    {
        // Set the base DOP to the maximum.
        // That's what the native Parallel.ForEachAsync will see.
        base.MaxDegreeOfParallelism = Int32.MaxValue;
        _maxDegreeOfParallelism = Environment.ProcessorCount;
    }

    public new int MaxDegreeOfParallelism
    {
        get { return _maxDegreeOfParallelism; }
        set
        {
            if (value < 1) throw new ArgumentOutOfRangeException();
            if (value == _maxDegreeOfParallelism) return;
            int delta = value - _maxDegreeOfParallelism;
            DegreeOfParallelismChangedDelta?.Invoke(this, delta);
            _maxDegreeOfParallelism = value;
        }
    }
}

The DynamicParallelOptions.MaxDegreeOfParallelism property is not thread-safe. It is assumed that controlling the maximum degree of parallelism will be performed by a single thread, or at least that the operations will be synchronized.

Usage example, featuring a Channel<T> as the source of the parallel loop:

var channel = Channel.CreateUnbounded<int>();
var options = new DynamicParallelOptions() { MaxDegreeOfParallelism = 2 };

await DynamicParallelForEachAsync(
    channel.Reader.ReadAllAsync(), options, async (item, ct) =>
    {
        Console.WriteLine($"Processing #{item}");
        await Task.Delay(1000, ct); // Simulate an I/O-bound operation
    });

// Push values to the channel from any thread
channel.Writer.TryWrite(1);
channel.Writer.TryWrite(2);
channel.Writer.TryWrite(3);
channel.Writer.Complete();

// Set the MaxDegreeOfParallelism to a positive value from a single thread
options.MaxDegreeOfParallelism = 5;

Some overloads with synchronous source, or synchronous body:

public static Task DynamicParallelForEachAsync<TSource>(
    IEnumerable<TSource> source,
    DynamicParallelOptions options,
    Func<TSource, CancellationToken, ValueTask> body)
{
    if (source == null) throw new ArgumentNullException(nameof(source));

    #pragma warning disable CS1998
    async IAsyncEnumerable<TSource> GetSource()
    { foreach (var item in source) yield return item; }
    #pragma warning restore CS1998

    return DynamicParallelForEachAsync(GetSource(), options, body);
}

public static void DynamicParallelForEach<TSource>(
    IEnumerable<TSource> source,
    DynamicParallelOptions options,
    Action<TSource, CancellationToken> body)
{
    if (body == null) throw new ArgumentNullException(nameof(body));
    DynamicParallelForEachAsync(source, options, (item, ct) =>
    {
        body(item, ct); return ValueTask.CompletedTask;
    }).Wait();
}