Количество потоков в Parallel.ForEach

#c_sharp #tpl

Устанавливаю количество потоков в 10, но оно работает в три потока (трехядерный комп). 

Parallel.ForEach(list, 
    new ParallelOptions { MaxDegreeOfParallelism = 10 },
    () => Thread.CurrentThread.ManagedThreadId,
    (i, loopState, threadId) =>
    {
        var cells = i.Split(';');
        var ip = cells[0];
        var login = cells[1];
        var pass = cells[2];
        try
        {
            using (var client = new SshClient(ip, login, pass))
            {
                client.Connect();
                var port = new ForwardedPortLocal("localhost", 10000 + (uint)threadId,
"ya.ru", 80);
                port.Start();
                client.AddForwardedPort(port);
                newList.Add(ip);
                port.Stop();
                client.Disconnect();
            }
        }
        catch 
        {

        }

        return threadId;
    },
    (threadId) => { }

    

Ответы

Ответ 1

В Parallel.ForEach количество потоков задается максимальным, а реальное количество
потоков зависит от доступных ресурсов. Если нужно точное количество потоков, то можно
использовать PLINQ либо использовать собственное решение с помощью кастомного TaskScheduler,
либо явного контроля с помощью WaitHandle.

Также не нужно забывать, что работа с сетью это так называемые "I/O bound потоки",
поэтому лучше использовать такую асинхронность при условии, что используемая библиотека/класс
это позволяет


Ответ 2

Насколько помню, у Parallel.ForEach довольно сложная логика, когда он сам подбирает
наиболее оптимальное количество потоков для выполнения задачи.
А MaxDegreeOfParallelism определяет лишь максимально возможное количество потоков.
По своему опыту знаю, что количество активных потоков довольно сильно плавает, увеличиваясь
и уменьшаясь в зависимости от характеристик компьютера, решаемых параллельно задач
и того, насколько они занимают ресурсы.

Хардкорный способ запустить сразу большое количество потоков - увеличить размер пула
при старте приложения, например

ThreadPool.SetMinThreads(200, 200);
Parallel.ForEach(...


Не буду говорить, что это хороший способ. Но тем не менее, мне когда такое решение
помогло, когда надо было обрабатывать несколько сотен (или даже тысяч) параллельных
потоков, выполняющих большие операции, связанные с вводом/выводом и сетевым взаимодействием.


Ответ 3

Если c Parallel.ForEach что-то не выходит - можно попытаться использовать задачи:

Task.WaitAll(list.Select((item, index) => Task.StartNew(() => {
    /** тут код **/

    /* здесь item - элемент списка, index - номер элемента в списке */
})));


Но если и с задачами не выйдет - проверьте загруженность процессора. SSH же использует
криптографию, а криптография процессор неплохо нагружает...


Ответ 4

Как указывает название опции MaxDegreeOfParallelism -- это максимальное количество
потоков. Опция является скорее подсказкой, нежели приказом. Фактическое же количество
потоков подбирается библиотекой, исходя из количества ядер и загруженности системы.
Стоит отметить, что потоки занимают память, а на переключение между ними требуется
время. Оптимальное количество потоков в системе на самом деле равно количеству ядер
-- все потоки будут заняты работой, отсутствуют траты на переключение контекста. Понятно,
что в работающей ОС запущено множество потоков, поэтому библиотека оптимальный вариант
-- запустить столько же потоков, сколько у вас ядер.

Если вы хотите завести фиксированное количество потоков -- воспользуйтесь тасками.
Алгоритм следующий:

IEnumerable tasks = GetTasks();
int maxDegreeOfParallelism = 10;
var queue = new Queue(maxDegreeOfParallelism);

foreach (var task in tasks)
{
    queue.Enqueue(task);
    if (queue.Count >= maxDegreeOfParallelism)
    {
         await queue.Dequeue();
    }
}

while (queue.Count > 0)
{
    await queue.Dequeue();
}


Однако нужно тестировать: может случиться, что принудительный запуск 10 тасков (потоков)
будет работать медленнее, чем 3 потока. Особенно в вашем случае, когда SSH использует
процессор и сеть, а код у вас синхронный.

Если же существует асинхронная версия SshClient, и вы правильным образом напишете
весь код асинхронно, то теоретически получите выигрыш даже меньшим количеством потоков,
поскольку пока данные ходят по интернету, поток может не ждать их, а заняться другой
работой (например, обработкой уже пришедших результатов).