Блог #threading

Давайте напишем счетчик, значение которого мы хотим увеличивать из разных потоков, а после того как все потоки закончат работу, мы выводим это число.

public class Counter {
     public int value;
     public void Increment() => Interlocked.Increment(ref this.value);
}

Вот вроде бы и все, но на самом деле - можно быстрее. Каким образом?

public class Counter {
     public int[] values;
     public int Count {
         get {
             var count = 0;
             for (int i = 0; i < this.values.Length; ++i) count += this.values[i];
             return count;
         }
     }
     public void Increment(int threadIndex) => ++this.values[threadIndex];
} 

Т.е. мы должны знать количество потоков и порядковый номер потока, в котором работаем (В Unity Jobs есть JobsUtility.ThreadIndex и JobsUtility.ThreadIndexCount).

Т.е. мы создаем Counter с массивом по количеству потоков и при каждой операции Increment мы передаем номер текущего потока. Тогда этот счетчик будет работать без оверхеда на добавление совсем. А когда операции закончились - мы суммируем все счетчики и возрващаем значение.

CC (`Concurrent Collections`) коллекции - это набор коллекций данных, разработанных для работы в многопоточной среде. Одной из особенностей CC коллекций является их lock-free (без блокировок) реализация, которая позволяет не блокировать весь многопоточный поток при обращении к коллекции.

Все CC коллекции стараются обходиться без lock, т.е. в нормальном режиме работы - либо вообще без lock, либо в редких исключениях его использование. 

Давайте разберем простой пример, чтобы было понятно как именно работают такие коллекции.

Допустим, что нам нужно написать коллекцию Stack<> (возьмем самую простую). В однопоточной реализации мы используем массив элементов + индекс, который говорит нам где мы находимся в данный момент. При Push мы просто кладем элемент по индексу и увеличиваем индекс, а при Pop просто уменьшаем индекс. Ну еще при Push нам нужно проверить размер массива и сделать новый, если это нужно. 

А теперь в многопоточность.

Как реализовать такую коллекцию? Давайте не будем вообще создавать никаких массивов, а будем использовать односвязный список из нод. Node - это объект, который имеет указатель на предыдущий элемент и данные внутри себя. 

Коллекция же имеет только ссылку на head-ноду. При добавлении элемента нам нужно создать ноду и каким-то образом ее запихнуть к последней, используем Interlocked.CompareExchange и заменяем head на наш элемент. При Pop делаем обратную операцию. 

false sharing - это ситуация, когда несколько потоков одновременно обращаются к разным переменным, которые находятся в одной кеш-линии. Кеш-линия - это минимальная единица данных, которая копируется из оперативной памяти в кеш процессора.

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

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

public static void Lock(ref int lockIndex) {
  for (;;) {
    if (System.Threading.Interlocked.Exchange(ref lockIndex, 1) == 0) {
      break;
    }
  }
}

public static void Unlock(ref int lockIndex) {
  System.Threading.Interlocked.Exchange(ref lockIndex, 0);
}

Мы заводим int поле и используем его в качестве идентификатора для операции блокирования. Другими словами, пока не будет вызван Unlock, второй поток не пройдет через Lock. 

Из минусов такого подхода - если ваш код между этими вызовами упадет по исключению, то все ожидающие потоки повиснут. Для этого я написал дополнение, которое выходит из цикла с ошибкой, если мы ждем слишком долго. 

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

интервал я = 123; // всегда атомарно
длинный j = 234L; // атомарно на x64, но кому сейчас надо x32?
я++; // никогда не атомарно, т.к. мы читаем данные, увеличиваем, а потом записываем

Чтобы атомарно изменить значение переменной, можно использовать lock , но это один из самых долгих способов. Гораздо быстрее использовать взаимосвязанные методы.

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