É possível criar uma tarefa sem iniciá-la, para que eu possa limitar a simultaneidade máxima?

Você imaginou uma solução errada para o problema. Esta abordagem não fará o que você pensa:

Parallel.ForEach(allTasks, new ParallelOptions {MaxDegreeOfParallelism = 10},
aTask =>
{
     aTask.Start();
});

Isso limitará a simultaneidade do início das tarefas somente. Iniciar uma tarefa com Starté praticamente instantâneo, já que tudo o que ele faz é agendar a execução da tarefa para o padrão TaskScheduler, que é o ThreadPool. Então o SomeAsyncFunctionnão será limitado. A solução correta é usar o Parallel.ForEach, ou melhor ainda o Parallel.ForEachAsync, para executar o SomeAsyncFunctiondiretamente, sem gerenciamento explícito de tarefas. A Parallelclasse usa tarefas internamente, para que você não tenha que fazer isso sozinho (é por isso que a tecnologia geral é chamada de Task Parallel Library ).

public async Task DoAllTheWork(IEnumerable<IGrouping<string, string>> groups)
{
    IEnumerable<string> firstValues = groups.SelectMany(g => g.SkipLast(1));
    IEnumerable<string> lastValues = groups.Select(g => g.Last());

    ParallelOptions options = new()
    {
        MaxDegreeOfParallelism = 10
    };

    await Parallel.ForEachAsync(firstValues, options, (value, _) =>
    {
        await SomeAsyncFunction(value);
    }).ConfigureAwait(false);

    await Parallel.ForEachAsync(lastValues, options, (value, _) =>
    {
        await SomeAsyncFunction(value);
    }).ConfigureAwait(false);
}

Esta solução não é perfeita porque você perderá alguma simultaneidade enquanto o firstValuesestiver terminando e antes de o lastValuesestiver iniciado, mas o efeito desta imperfeição deve ser pequeno. Implementar uma solução perfeita, onde o MaxDegreeOfParallelismé imposto corretamente do início ao fim, não é trivial.

Eu sugeriria usar um semáforo como mecanismo de sincronização - por exemplo, SemaphoreSlim.

Primeiro, defina um método conveniente para lidar com a aquisição e liberação do semáforo:

static async Task SomeAsyncFunctionSynchronized(string value, SemaphoreSlim semaphore)
{
    try
    {
        await semaphore.WaitAsync();
        await SomeAsyncFunction(value);
    }
    finally
    {
        semaphore.Release();
    }
}

Então você pode usar esse método na sua lógica.

Abaixo está um exemplo simplificado e completo, incorporando esta abordagem:

var items = new[]
{
    ("1", "1"),
    ("1", "2"),
    ("1", "3"),
    ("2", "4"),
    ("2", "5"),
    ("2", "6"),
    ("3", "7"),
    ("3", "8"),
    ("3", "9"),
};

await DoAllTheWork(
    items.GroupBy(i => i.Item1, i => i.Item2).ToArray(),
    10);

static async Task DoAllTheWork(IEnumerable<IGrouping<string, string>> groups, int maxConcurrency)
{
    var semaphore = new SemaphoreSlim(maxConcurrency);
    var allTasks = new List<Task>();

    foreach (var group in groups)
    {
        var values = group.ToList();
        var tasksInGroup = new List<Task>();

        // Start tasks for all values except the last
        foreach (var value in values.SkipLast(1))
        {
            var task = SomeAsyncFunctionSynchronized(value, semaphore);
            tasksInGroup.Add(task);
            allTasks.Add(task);
        }

        // Ensure the last item in the group runs only after the others
        var lastTask = async () =>
        {
            await Task.WhenAll(tasksInGroup);
            await SomeAsyncFunctionSynchronized(values.Last(), semaphore);
        };

        allTasks.Add(lastTask());
    }

    await Task.WhenAll(allTasks);
}

static async Task SomeAsyncFunctionSynchronized(string value, SemaphoreSlim semaphore)
{
    try
    {
        await semaphore.WaitAsync();
        await SomeAsyncFunction(value);
    }
    finally
    {
        semaphore.Release();
    }
}

static async Task SomeAsyncFunction(string value)
{
    Console.WriteLine($"{DateTime.Now} Starting {nameof(SomeAsyncFunction)} with value {value}");
    await Task.Delay(2000);
    Console.WriteLine($"{DateTime.Now} Finishing {nameof(SomeAsyncFunction)} with value {value}");
}

Referência

• Classe SemaphoreSlim

Você pode usar minha biblioteca PowerThreadPool para atingir seu objetivo.

1. Você pode definir MaxThreadscomo 10 usando o PowerPoolOptionpara controlar o número máximo de threads simultâneos. ( wiki )

2. Você pode definir StartSuspendedpara true. Isso permite que você enfileire tarefas sem iniciar o pool de threads imediatamente. ( wiki )

3. Você pode usar a WorkOption.Dependentspropriedade para garantir que o último valor de cada grupo só comece depois que as tarefas anteriores do grupo forem concluídas. ( wiki )

4. Depois que todas as tarefas e dependências forem adicionadas, chame Start()para iniciar a execução.

Exemplo:

var powerPool = new PowerPool(new PowerPoolOption
{
    MaxThreads = 10,
    StartSuspended = true
});

// Add tasks and set dependencies here
// ...

powerPool.Start(); // Begin execution

Espero que ajude você.

Aqui estão algumas respostas interessantes. Eu ofereço a você a maneira como eu faria isso:

async Task DoAllTheWorkAsync(IEnumerable<IGrouping<string, string>> groups)
{
    IEnumerable<Task> allTasks = groups.Select(ProcessGroupAsync);

    await Task.WhenAll(allTasks);
}

static Task ProcessGroupAsync(IGrouping<string, string> group)
{
    IEnumerable<Task> groupTasksWithoutLast = group
        .SkipLast(1)
        .Select(SomeFunctionAsync);

    return Task.WhenAll(groupTasksWithoutLast)
        .ContinueWith(t =>
        {
            if (t.IsFaulted)
            {
                // Handle the error
                Console.WriteLine("An error occurred in the group tasks.");
            }

            // This will run after all tasks in the group are completed
            return SomeFunctionAsync(group.Last());
        });
}

static async Task SomeFunctionAsync(string value)
{
    // Simulate some async work
    await Task.Delay(1000);
    Console.WriteLine($"Processed: {value}");
}

Se o seu "trabalho" não for um cálculo complexo, eu não recomendaria usar Parallel.ForEach, especialmente se for algum tipo de operação de multiplicação de E/S.