Потоки и параллельные вычисления

Свойства потоков

Рассмотрим некоторые свойства объектов класса Thread.

Свойство Name позволяет узнать имя потока или задать потоку собственное имя, отличное от служебного имени, получаемого потоком в момент создания. Именование потоков облегчает процесс отладки многопоточного приложения.

Свойство Priority позволяет задавать и получать приоритет потока. Возможные 5 значений этого свойства задаются перечислением ThreadPriority. По умолчанию поток получает приоритет, заданный значением Normal. Можно задать для приоритета два значения ниже приоритета Normal - BelowNormal и Lowest. Можно задать для приоритета два значения выше приоритета Normal - AboveNormal и Highest.

Среди других свойств отметим группу свойств Current, позволяющих получить текущий поток, текущий контекст потока, текущую культуру, используемую потоком. Другая группа свойств Is позволяет выяснить различные характеристики потока - его состояние, является ли он фоновым или членом группы потоков.

Завершение потока

Помимо уже упомянутых методов Start, Join, Sleep в классе Thread есть еще несколько десятков методов, как статических, так и динамических, вызываемых объектами - экземплярами класса Thread. Некоторые из этих методов появятся в примерах.

Сейчас же рассмотрим два метода, позволяющие завершить поток - Interrupt и Abort.

Когда поток вызывает метод Interrupt, то все зависит от того, в каком состоянии находится поток. Если поток находится в заблокированном состоянии или через некоторое время перейдет в заблокированное состояние, например, "уснет", то поток выводится из этого состояния. Вот как это происходит. Для заблокированного потока возникает исключительная ситуация ThreadInterruptedException. Управление будет передано обработчику этой ситуации и по его завершению, завершит свою жизнь и поток. Понятно, что вызов Interrupt предполагает, что предусмотрен обработчик такой ситуации. В противном случае работа приложения будет прервана, что конечно является печальным фактом.

Когда поток вызывает метод Abort, то независимо от того, в каком состоянии находится поток, он обычно завершается. При этом также возникает исключительная ситуация ThreadAbortException, для которой можно предусмотреть обработчик события. Но даже если этого обработчика нет, поток нормально завершит свою работу. В отдельных случаях при обработке исключения можно отменить уничтожение потока, вызвав метод ResetAbort.

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

Создание и завершение потока

Рассмотрим пример, иллюстрирующий создание потоков и их принудительное завершение с использованием методов Abort и Interrupt.

Создадим консольный проект с именем ConsoleCounter. Как и положено, для методов, выполняемых в потоках, создадим класс, который назовем Counts. Класс устроен просто, у него одно закрытое поле, играющее роль счетчика, и константа, задающая максимальное значение счетчика:

/// <summary>
    /// Счетчики
    /// </summary>
    class Counts
    {
        long sum = 0;
        const long LIMIT = 100000000;
        public long Sum
        {
            get { return sum; }
        }
    }

Добавим в класс метод Count, увеличивающий значение счетчика:

/// <summary>
        /// Счетчик, в цикле увеличивается, 
        /// пока не достигнет значения LIMIT
        /// Метод ленивый - засыпает на каждом шагу, 
        /// </summary>
        public void Count()
        {
            sum = 0;
            try
            {
                while (sum < LIMIT)
                {
                    sum++;
                    Thread.Sleep(0);
                }
            }
            catch (ThreadInterruptedException )
            {
                Console.WriteLine("Метод Count был прерван вызовом
                                             Interrupt!");
            }
            finally
            {
                Console.WriteLine("Конец - делу венец!");
            }
            
        }

Метод на каждом шагу засыпает, так что его выполнение в потоке блокируется, и он периодически будет переходить в состояние "ожидание". Метод предусматривает обработку события Interrupt.

А теперь в процедуре main нашего приложения создадим поток, выполняющий метод Count:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace ConsoleCounter
{
    class Program
    {
        static void Main(string[] args)
        {
            //Создание и запуск потока
            Thread thread_count;
            Counts my_counter = new Counts();
            thread_count = new Thread(my_counter.Count);
            thread_count.Start();
            //Засыпаем, дав потоку возможность поработать
            Thread.Sleep(1);
            //Просыпаемся и останавливаем работу потока
            thread_count.Interrupt();
            //Дожидаемся завершения потока
            thread_count.Join();
            //Печать результатов работы потока
            Console.WriteLine("count = " + my_counter.Sum);
      }
  }
}

Обратите внимание, вначале создается объект my_counter класса Counts, а затем этот объект, вызывающий метод Count передается конструктору класса Thread при создании потока thread_count. Что делается в процедуре main после создания и запуска потока? Основной поток засыпает, давая возможность поработать дочернему потоку, но, проснувшись, прерывает работу дочернего потока, который мог и не успеть закончить свою работу. Запустив этот проект на выполнение на своем одноядерном компьютере, я получил следующие результаты:

Понятно, что произошло. Поток thread_count часто "засыпал". Когда в таком состоянии основной поток вызвал метод Interrupt, то возникла исключительная ситуация - ThreadInterruptedException, которая была обработана, после чего поток выполнил действия, указанные в блоке finally и завершил свою работу, успев немного посчитать.

Что произойдет, если поток засыпать не будет? Давайте в методе Count закомментируем оператор

Thread.Sleep(0);

Поскольку до завершения своей работы поток thread_count не блокируется, то вызов Interrupt не оказывает на него никакого воздействия, и он спокойно досчитает до максимально возможного значения LIMIT.

А что произойдет, если поток блокируется, но обработчик события ThreadInterruptedException не предусмотрен? Закомментируем обработчик события в методе Count. Для потока в заблокированном состоянии возникнет исключительная ситуация и обрабатывать ее будет системный обработчик, прерывающий работу приложения. Блок finally и в этом случае будет работать.

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

Для проверки работы метода завершения потока Abort добавим в класс Counts метод Count2, похожий на метод Count, но устроенный чуть сложнее, поскольку он восстанавливает свою работу в результате обработки исключительной ситуации:

/// <summary>
        /// Счетчик, в цикле увеличивается, 
        /// пока не достигнет значения LIMIT
        /// Метод упорный - завершает счет,
        /// несмотря на попытку прерывания при вызове Abort
        /// </summary>
        public void Count2()
        {
            try
            {
                while (sum < LIMIT)
                {
                    sum += 2;                    
                }
            }           
            catch (ThreadAbortException)
            {
                Console.WriteLine("Не буду завершать Count2 при вызове Abort" +
                    "\r\n" + "Восстановлюсь и завершу свою работу!!");
                Thread.ResetAbort();
            }
            finally
            {
                Console.WriteLine("Конец - делу венец!");
            }
            while (sum < LIMIT)
            {
                sum += 2;
            }
        }

Обратите внимание, после завершения обработчика события Abort продолжается увеличение счетчика.

Добавим теперь в процедуру main создание второго потока, выполняющего метод Count2:

//Создание второго потока
            Thread thread_count2;
            Counts my_counter2 = new Counts();
            thread_count2 = new Thread(my_counter2.Count2);
            thread_count2.Start();
            Thread.Sleep(1);           
            thread_count2.Abort();
            thread_count2.Join();
            Console.WriteLine("count2 = " + my_counter2.Sum);

Заметьте, здесь для прекращения работы потока вызывается метод Abort. У потока возникает исключительная ситуация, хотя он и не блокируется, находясь в состоянии "выполнение". В данном примере у метода Count2 предусмотрен обработчик этой ситуации, который отменяет попытку прекращения работы потока, так что последний завершает выполнение своей работы. Вот как выглядят результаты работы в этой ситуации:

Что произойдет, если в обработчике события не вызывать метод ResetAbort, отменяющий прерывание работы потока? Если закомментировать этот оператор, то работа потока будет прервана.

Более того, можно вообще отключить обработчик события! И в этом случае поток прервет свою работу без всяких осложнений, никакой системный обработчик события вызываться не будет. Прерывание работы потока будет происходить независимо от того, будет ли поток блокирован или находится в состоянии "выполнение".

Потоки и параллельные алгоритмы

После рассмотрения учебных примеров давайте попробуем применить потоки к алгоритмам, которые рассматривались в главе 3. Начнем с алгоритма вычисления определенного интеграла.

Вычисление определенного интеграла

Для вычисления интеграла на интервале [a, b] для распараллеливания вычислений можно применить простой прием. Интервал [a, b] разбивается на p отрезков по числу используемых процессоров. На каждом отрезке вычисляется интеграл по обычной последовательной схеме. Это типичный прием, когда исходная задача сводится к p подзадачам меньшего размера. Объединение решений подзадач в данном случае сводится к простому суммированию полученных значений.

В главе 3 мы построили класс NewIntegral, в который включили чисто последовательный вариант вычисления интеграла - DefiniteIntegral и последовательный вариант с разделением интервала интегрирования на отрезки - SequenceIntegralWithSegments.

В методе SequenceIntegralWithSegments итерации внешнего цикла независимы и могут выполняться параллельно. Давайте посмотрим, как можно использовать потоки для решения этой задачи. Очевидное решение состоит в том, чтобы создать массив потоков по числу отрезков, на которые разбивается интервал интегрирования. Каждый из потоков должен выполнять метод DefiniteIntegral для своих входных данных и результат посылать в свою ячейку памяти. Пересечений по данным у потоков нет, и потому проблем, связанных с взаимодействием потоков, не должно возникать.

Как в данной ситуации передать потоку метод, который он должен выполнять? Кажется, что наилучшее решение дает определение в момент создания потока анонимного метода, в котором можно вызвать метод DefiniteIntegral, передав ему нужные параметры.

К сожалению, теоретически правильная конструкция на практике не всегда работает корректно. Точно описать ситуацию, когда возникает некорректность в работе, пока не могу.

Давайте рассмотрим другой более надежный вариант решения стоящей перед нами задачи.

В класс NewIntegral, введенный в главе 3, добавим метод, в котором распараллеливание вычислений ведется с использованием потоков:

/// <summary>
        /// Параллельное вычисление интеграла
        /// несколькими потоками
        /// </summary>        
        public void ParallelIntegralWithThreads()
        {
            Thread[] threads = new Thread[p];           
            for (int i = 0; i < p; i++)
            {               
                threads[i] = new Thread(ThreadTaskIntegral);                    
                threads[i].Start(i);
            }
            for (int i = 0; i < p; i++)
            {
                threads[i].Join();                
            }
            result = 0;
            for (int i = 0; i < p; i++)
            {
                result += results[i];
            }
        }

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

void ThreadTaskIntegral(object index)
        {
            int i = (int)index;
            double dx = (b - a) / p;
            double start = 0, finish = 0;
            start = a + i * dx;
            finish = start + dx;
            DefiniteIntegral(start, finish, out results[i]);
        }

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

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

Здесь A - амплитуда, а $$\omega$$ - частота колебаний. Вычисления, как уже говорилось, ведутся на 64-х битном компьютере с 6 Гб памяти, с четырьмя ядрами и восемью виртуальными процессорами. Приведу результаты экспериментов по вычислению интеграла в зависимости от числа разбиений интервала интегрирования на отрезки, а тем самым от степени параллелизма и числа создаваемых потоков:

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

Следует заметить, что результаты зависят от величины интервала интегрирования и простоты интегрируемой функции. Чем выше частота гармонических колебаний, тем лучшие результаты показывает параллельный алгоритм с потоками в сравнении с последовательным алгоритмом. Но и для простейшего случая, когда амплитуда и частота колебаний равна 1, а интегрирование ведется на отрезке от 0 до $$\pi /2$$, потоковый алгоритм показывает лучший результат, заканчиваясь практически с нулевым временем работы. На следующем рисунке показан сеанс работы алгоритма для такого случая:

Пузырьковая сортировка и потоки

Рассмотрим теперь подключение потоков к алгоритмам сортировки, рассмотренным в главе 3. Начнем с алгоритма пузырьковой сортировки. Параллельная версия этого алгоритма, которую мы написали, основана на распараллеливании по данным. Исходный массив разбивается на подмножества, для чего используется шаговый алгоритм. Каждое подмножество сортируется независимо, а затем результаты отсортированных подмножеств сливаются. Этот алгоритм показывает лучшие результаты, даже при выполнении его одним процессором. Можно ли добиться улучшения, если сортировка каждого подмножества выполняется в отдельном потоке?

Давайте напишем реализацию этого подхода. При создании потоков будем использовать не анонимные методы, а надежный подход с объектами специально созданного класса SortOne. Этот класс будет содержать метод, выполняющий сортировку, а все необходимые данные будут содержаться в полях класса. Вот как выглядит этот класс:

class SortOne
    {
        double[] mas;
        int j;
        int p;
        int n;        
        public SortOne(double[] mas, int j, int p)
        {
            this.mas = mas;
            this.j = j;
            this.p = p;
            n = mas.Length;
        }
       
        /// <summary>
        /// Сортирует пузырьком часть массива mas
        /// начиная с элемента с номером n - j -1
        /// Сортируемые элементы отстоят на расстоянии p
        /// </summary>
        public void BubbleSortPart()
        {
            int i0 = n - j - 1, m = i0 / p;
            double temp;
            //цикл по числу проходов m
            for (int k = 0; k < m; k++)
            {
                //цикл всплытия легкого элемента на k-м проходе
                for (int i = i0; i - p >= k * p; i = i - p)
                {
                    if (mas[i] < mas[i - p])
                    {//swap
                        temp = mas[i];
                        mas[i] = mas[i - p];
                        mas[i - p] = temp;
                    }
                }
            }
        }        
    }

Теперь напишем версию параллельного алгоритма пузырьковой сортировки с введением потоков:

/// <summary>
        /// Версия параллельного алгоритма
        /// пузырьковой сортировки с введением потоков
        /// для сортировки подмножеств массива
        /// </summary>
        /// <param name="mas">сортируемый массив</param>
        /// <param name="processors">число подмножеств</param>
        public void BubbleSortWithTreads(double[] mas, int processors)
        {
             Thread[] threads = new Thread[processors];
            SortOne[] sorts = new SortOne[processors];
            //Создание объектов SortOne,
            //передаваемых создаваемым потокам
            for (int i = 0; i < processors; i++)
            {
                sorts[i] = new SortOne(mas, i, processors);
                threads[i] = new Thread(sorts[i].BubbleSortPart);               
                threads[i].Start();
            }
            //Синхронизация
            for (int i = 0; i < processors; i++)
            {
                threads[i].Join();                
            }            
            //Слияние отсортированных последовательностей
            Merge(mas, processors);        
        }

Особых пояснений помимо комментариев, включенных в текст метода, видимо не требуется, поскольку подобный прием уже применялся при построении метода вычисления интеграла. Текст процедуры слияния Merge ранее приведен в главе 3. В заключение приведем результаты экспериментов по сортировке массива вещественных чисел, содержащего100 000 элементов.

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

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

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

Быстрая сортировка Хоора и потоки

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

Чтобы удовлетворить технологическим требованиям, введем структуру, описывающую параметры, требуемые методу сортировки:

struct StructOne
    {
        public double[] mas;        
        public int n;
        public int start, finish;
        public StructOne(double[] mas, int start, int finish)
        {
            this.mas = mas;
            this.start = start;
            this.finish = finish;
            n = mas.Length;
        }
    }

Рассмотрим теперь версию параллельного алгоритма, работающего с потоками:

public void QSortWithTreads(double[] mas, int p)
        {
            Thread[] threads = new Thread[p];
            StructOne[] sorts = new StructOne[p];
            int start = 0, finish = 0, n = mas.Length, m = n/p;
            for (int i = 0; i < p; i++)
            {
                start = i * m;
                finish = i != p - 1 ? start + m - 1 : n - 1;
                sorts[i] = new StructOne(mas, start, finish);
                threads[i] = new Thread(QSortStruc);
                threads[i].Start(sorts[i]);
            }
            for (int i = 0; i < p; i++)
            {
                threads[i].Join();
            }
            //Слияние отсортированных последовательностей
            MergeQ(mas, p);
        }

Как можно видеть, при создании потока ему не передается экземпляр некоторого класса. Всем потокам передается один и тот же метод QSortStruc, но параметр, передаваемый потоку, у каждого потока свой, что гарантирует корректное распараллеливание по данным.

Метод QSortStruc представляет собой слегка модифицированную версию классического алгоритма быстрой сортировки:

void QSortStruc(object param)
        {
            StructOne s = (StructOne)param; 
            double[] mas = s.mas;
            int start = s.start;
            int finish = s.finish;
            int n = s.n;
            if (finish - start > 0)
            {
                double cand = 0, temp = 0;
                int l = start, r = finish;
                cand = mas[(r + l) / 2];
                while (l <= r)
                {
                    while (mas[l] < cand) l++;
                    while (mas[r] > cand) r--;
                    if (l <= r)
                    {
                        temp = mas[l];
                        mas[l] = mas[r];
                        mas[r] = temp;
                        l++; r--;
                    }
                }
                StructOne left = new StructOne(mas, start, r);
                QSortStruc(left);
                StructOne right = new StructOne(mas, l, finish);
                QSortStruc(right);
            }
        }

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

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