无法动态修改Parallel.ForEach并行度时，逐步扩容工作线程的设计方案咨询

Great question—this is such a thoughtful approach for safety-critical systems where you need to ramp up load gradually without risking overload. Since you can't tweak MaxDegreeOfParallelism mid-run in Parallel.ForEach, let's break down a few practical, battle-tested design patterns to make this work for your aircraft testing scenario:

Option 1: Phased Parallel Execution (Simplest, Safety-First Approach) #

Your workflow is inherently sequential in terms of validation (test 1 → validate → test 2 → validate → ...), so the easiest way is to run discrete Parallel.ForEach batches at each concurrency level. After each batch passes your success check, you double the concurrency for the next round until you hit your max limit.

This approach is perfect for safety-focused systems because it isolates each validation phase—no risk of mid-execution changes causing unexpected behavior. Here's a code example tailored to your use case:

int currentConcurrency = 1;
int maxConcurrency = 50;
bool batchSucceeded = true;

// Assume you have a pre-defined list of test cases
List<AircraftTest> testCases = GetAllSimulationTestCases();

while (batchSucceeded && currentConcurrency <= maxConcurrency)
{
    Console.WriteLine($"Running test batch with {currentConcurrency} concurrent instances...");
    
    // Execute the batch with current concurrency limit
    Parallel.ForEach(testCases, 
        new ParallelOptions { MaxDegreeOfParallelism = currentConcurrency }, 
        testCase => RunAircraftSimulation(testCase));

    // Use your existing feedback mechanism to verify success
    batchSucceeded = DidBatchPassValidation();

    if (batchSucceeded)
    {
        // Double concurrency, cap at max to avoid overshooting
        currentConcurrency = Math.Min(currentConcurrency * 2, maxConcurrency);
    }
    else
    {
        Console.WriteLine($"Batch failed at {currentConcurrency} instances. Stopping ramp-up.");
    }
}

if (batchSucceeded)
{
    Console.WriteLine($"Successfully scaled up to {currentConcurrency} concurrent test instances.");
}

Option 2: Manual Concurrency Control with SemaphoreSlim #

If you want to avoid stopping and restarting batches (e.g., you need to process a continuous stream of test cases), a SemaphoreSlim lets you dynamically adjust the number of concurrent tasks on the fly.

This gives you fine-grained control while keeping the code relatively simple:

// Start with 1 concurrent slot, max out at 50
SemaphoreSlim concurrencyGate = new SemaphoreSlim(1, 50);
int currentConcurrency = 1;
int maxConcurrency = 50;
bool batchSucceeded = true;

List<Task> activeTasks = new List<Task>();

// Queue all test cases (or add them incrementally for continuous testing)
foreach (var testCase in testCases)
{
    await concurrencyGate.WaitAsync();
    activeTasks.Add(Task.Run(async () =>
    {
        try
        {
            RunAircraftSimulation(testCase);
        }
        finally
        {
            concurrencyGate.Release();
        }
    }));
}

// Your validation feedback loop
while (batchSucceeded && currentConcurrency < maxConcurrency)
{
    // Wait for the current phase of tasks to complete
    await Task.WhenAll(activeTasks.Where(t => !t.IsCompleted).ToArray());
    
    batchSucceeded = DidBatchPassValidation();
    if (batchSucceeded)
    {
        currentConcurrency = Math.Min(currentConcurrency * 2, maxConcurrency);
        // Release extra slots to allow more concurrent tasks
        concurrencyGate.Release(currentConcurrency - concurrencyGate.CurrentCount);
    }
}

// Wait for all remaining tasks to finish
await Task.WhenAll(activeTasks);

Option 3: Custom Task Scheduler (Advanced Dynamic Control) #

For the most flexibility (e.g., adjusting concurrency mid-batch without waiting for full completion), you can implement a custom task scheduler that respects a dynamically updatable concurrency limit. This is more complex but ideal if you need real-time adjustments based on live feedback.

Here's a simplified example of such a scheduler:

public class DynamicConcurrencyScheduler : TaskScheduler
{
    private readonly LinkedList<Task> _queuedTasks = new LinkedList<Task>();
    private int _currentRunningTasks;
    private int _maxConcurrency;
    private readonly object _lock = new object();

    public int MaxConcurrency
    {
        get => _maxConcurrency;
        set
        {
            lock (_lock)
            {
                _maxConcurrency = value;
                // Try to schedule more tasks if we increased the limit
                TryRunQueuedTasks();
            }
        }
    }

    public DynamicConcurrencyScheduler(int initialMaxConcurrency)
    {
        _maxConcurrency = initialMaxConcurrency;
    }

    protected override IEnumerable<Task> GetScheduledTasks()
    {
        lock (_lock) return _queuedTasks.ToArray();
    }

    protected override void QueueTask(Task task)
    {
        lock (_lock)
        {
            _queuedTasks.AddLast(task);
            TryRunQueuedTasks();
        }
    }

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        return false; // Disable inline execution for simplicity
    }

    private void TryRunQueuedTasks()
    {
        while (_currentRunningTasks < _maxConcurrency && _queuedTasks.Count > 0)
        {
            var task = _queuedTasks.First.Value;
            _queuedTasks.RemoveFirst();
            if (TryExecuteTask(task))
            {
                Interlocked.Increment(ref _currentRunningTasks);
                // Decrement running count when task completes
                task.ContinueWith(t => Interlocked.Decrement(ref _currentRunningTasks), 
                    TaskContinuationOptions.ExecuteSynchronously);
            }
        }
    }
}

You'd use it like this:

var scheduler = new DynamicConcurrencyScheduler(1);
int maxConcurrency = 50;
bool batchSucceeded = true;

// Start processing test cases with initial concurrency
var tasks = testCases.Select(testCase => Task.Factory.StartNew(
    () => RunAircraftSimulation(testCase),
    CancellationToken.None,
    TaskCreationOptions.None,
    scheduler)).ToList();

// Feedback loop to adjust concurrency
while (batchSucceeded && scheduler.MaxConcurrency < maxConcurrency)
{
    await Task.WhenAll(tasks.Where(t => !t.IsCompleted).ToArray());
    batchSucceeded = DidBatchPassValidation();
    
    if (batchSucceeded)
    {
        scheduler.MaxConcurrency = Math.Min(scheduler.MaxConcurrency * 2, maxConcurrency);
    }
}

await Task.WhenAll(tasks);

Recommendation #

For your safety-critical aircraft testing system, Option 1 (Phased Execution) is the best choice. It's straightforward, easy to debug, and ensures you fully validate each concurrency level before scaling up—no unexpected edge cases from mid-run adjustments.

If you need continuous testing without batch restarts, Option 2 (SemaphoreSlim) strikes a great balance between simplicity and dynamic control.

内容的提问来源于stack exchange，提问作者learner