Parallel ForEach использует очень мало вычислительной мощности с течением времени

Question

У меня работает следующий код, и с течением времени (час или два) я замечаю, что итерации по элементам занимают все больше времени Есть ли что-то, что я делаю, что вызывает это? Если так, как я могу это исправить?

        int totalProcessed = 0;
        int totalRecords = MyList.Count();

        Parallel.ForEach(Partitioner.Create(0, totalRecords), (range, loopState) =>
        {
            for (int index = range.Item1; index < range.Item2; index++)
            {
                DoStuff(MyList.ElementAt(index));
                Interlocked.Increment(ref totalImported);
                if (totalImported % 1000 == 0)
                    Log(String.Format("Processed {0} of {1} records",totalProcessed, totalRecords));
            }
        });

         public void DoStuff(IEntity entity)
         {
              foreach (var client in Clients)
              {
                  // Add entity to a db using EF
                  client.Add(entity);
              }
          }

Спасибо за любую помощь

Answer 1

ElementAt - очень медленный метод расширения со следующей реализацией:

public static void T ElementAt(this IEnumerable<T> collection, int index) 
{
    int i = 0;
    foreach(T e in collection)
    {
        if(i == index)
        {
            return e;
        }
        i++;
    }
    throw new IndexOutOfRangeException();
}

Очевидно, что он работает дольше, когда индекс больше. Вы должны использовать индексатор MyList[index] вместо ElementAt.

Answer 2

Как указал @mace, использование ElementAt имеет проблемы с производительностью. Каждый раз, когда вы вызываете это, итератор начинается с начала MyList и пропускает n элементов, пока не достигнет нужного индекса. Это становится все хуже, так как позиция индекса становится выше.

Если вам все еще требуется потоковый доступ к MyList, вы можете уменьшить проблему с производительностью, используя Skip и Take. При поиске позиции в MyList все равно будет некоторое влияние на производительность, но Take будет гарантировать, что вы получите пакет элементов, как только попадете туда, вместо того, чтобы делать это для каждого элемента.

Я также заметил, что вы используете стиль раздела foreach, но вы делаете это для всего диапазона. В примере ниже я реализовал стиль разбиения с помощью пакетной обработки.

int totalRecords = MyList.Count();
int batchSize = 250;

Parallel.ForEach(Partitioner.Create(0, totalRecords, batchSize), range =>
{
    foreach (var thing in MyList.Skip(range.Item1).Take(batchSize))
    {
        DoStuff(thing);

        //logging and stuff...           
    }
});

Обновление

После прочтения вопроса у вас также могут возникнуть проблемы с использованием слишком большого количества потоков, что, вероятно, является проблемой, связанной с вводом-выводом, т. Е. С сетью, а затем с DB \ disk. Я говорю это так же, как вы говорите, что загрузка процессора незначительна, и это заставляет меня думать, что вы заблокированы на IO, и это становится все хуже.

Если бы оно было чисто до ElementAt, вы бы все равно увидели высокую загрузку ЦП.

Сконфигурируйте MaxDegreeOfParallelism, чтобы настроить максимальное количество используемых потоков:

const int BatchSize = 250;

int totalRecords = MyList.Count();
var partitioner = Partitioner.Create(0, totalRecords, BatchSize);
var options = new ParallelOptions { MaxDegreeOfParallelism = 2 };

Parallel.ForEach(partitioner, options, range =>
{
    foreach (int thing in MyList.Skip(range.Item1).Take(BatchSize))
    {
        DoStuff(thing);

        //logging and stuff...           
    }
});