Многопоточность в Ruby

Каждая программа, выполняющаяся в системе, — это процесс. Каждый процесс содержит один или несколько потоков.

Поток — это единый последовательный поток управления в программе. Одновременное выполнение нескольких потоков в одной программе для выполнения разных задач называется многопоточностью.

В Ruby мы можем создать многопоточность с помощью класса Thread. Потоки Ruby легковесны и позволяют эффективно реализовать параллельный код.

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

Чтобы запустить новый поток, просто вызовите Thread.new:

#Раздел кода потока #1 Thread.new { #Поток #2 выполняет код } #Поток #1 выполняет код

Пример

В следующем примере показано, как использовать многопоточность в программе на Ruby:

Пример

#!/usr/bin/ruby def func1 i=0 while i<=2 puts "func1 at: #{Time.now}" sleep(2) i=i+1 end end def func2 j=0 while j<=2 puts "func2 в: #{Time.now}" sleep(1) j=j+1 end end puts "Начато в #{Time.now}" t1=Thread.new{func1()} t2=Thread.new{func2()} t1.join t2.join puts "Завершено в #{Time.now}"

Попробуйте »

Результат выполнения приведённого выше кода:

Started At Wed May 14 08:21:54 -0700 2014
func1 at: Wed May 14 08:21:54 -0700 2014
func2 at: Wed May 14 08:21:54 -0700 2014
func2 at: Wed May 14 08:21:55 -0700 2014
func1 at: Wed May 14 08:21:56 -0700 2014
func2 at: Wed May 14 08:21:56 -0700 2014
func1 at: Wed May 14 08:21:58 -0700 2014
End at Wed May 14 08:22:00 -0700 2014

Жизненный цикл потока

1. Thread.new можно использовать для создания потока. Аналогично Thread.start или Thread.fork можно использовать с тем же синтаксисом для создания потоков.

2. После создания потока его не нужно запускать; поток выполнится автоматически.

3. Класс Thread определяет некоторые методы для управления потоками. Поток выполняет блок кода в Thread.new.

4. Последний оператор в блоке кода потока — это значение потока, которое можно получить с помощью методов потока. Если поток завершил выполнение, он возвращает значение потока; в противном случае значение не возвращается, пока поток не завершится.

5. Метод Thread.current возвращает объект, представляющий текущий поток. Метод Thread.main возвращает главный поток.

6. Используйте метод Thread.join для выполнения потоков. Этот метод приостанавливает главный поток, пока текущий поток не завершит выполнение.

Состояния потоков

У потоков есть 5 состояний:

Состояние потокаВозвращаемое значение
Runnablerun
СпящийСпящий
Выходаварийное завершение
Нормальное завершениеfalse
Завершён из-за исключенияnil

Потоки и исключения

Когда в потоке возникает исключение и оно не перехватывается rescue, поток обычно завершается без предупреждения. Однако если другие потоки ожидают этот поток с помощью Thread#join, в ожидающих потоках также будет возбуждено то же исключение.

begin t = Thread.new do Thread.pass #Главный поток действительно ожидает join raise "необработанное исключение" end t.join rescue p $! # => "необработанное исключение" end

Используя следующие 3 метода, можно заставить интерпретатор прервать выполнение, когда поток завершается из-за исключения.

  • Укажите при запуске скрипта-dопцию и запустите в режиме отладки.
  • использоватьThread.abort_on_exceptionУстановите флаг.
  • ИспользуйтеThread#abort_on_exceptionУстановите флаг для указанного потока.

После использования одного из трёх вышеописанных методов весь интерпретатор будет прерван.

t = Thread.new { ... } t.abort_on_exception = true

Управление синхронизацией потоков

В Ruby есть три способа достижения синхронизации, а именно:

1. Реализация синхронизации потоков через класс Mutex

2. Реализация синхронизации потоков через класс Queue для передачи данных

3. Использование ConditionVariable для управления синхронизацией

Реализация синхронизации потоков с помощью класса Mutex

Чтобы реализовать управление синхронизацией потоков с помощью класса Mutex, если нескольким потокам одновременно нужна переменная программы, можно заблокировать эту переменную с помощью lock. Код выглядит следующим образом:

Пример

#!/usr/bin/ruby require "thread" puts "Синхронизировать поток" @num=200 @mutex=Mutex.new def buyTicket(num) @mutex.lock if @num>=num @num=@num-num puts "вы успешно купили #{num} билетов" else puts "извините, недостаточно билетов" end @mutex.unlock end ticket1=Thread.new 10 do 10.times do |value| ticketNum=15 buyTicket(ticketNum) sleep 0.01 end end ticket2=Thread.new 10 do 10.times do |value| ticketNum=20 buyTicket(ticketNum) sleep 0.01 end end sleep 1 ticket1.join ticket2.join

Попробуйте »

Результат выполнения приведённого выше кода:

Synchronize Thread
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets

Помимо использования lock для блокировки переменной, вы также можете использовать try_lock для блокировки переменной, а также можете использовать Mutex.synchronize для синхронизации доступа к переменной.

Реализация синхронизации потоков с помощью класса Queue для передачи данных

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

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

Классическая проблема производителя-потребителя:

Пример

#!/usr/bin/ruby require "thread" puts "SizedQuee Тест" queue = Queue.new producer = Thread.new do 10.times do |i| sleep rand(i) #Пусть поток поспит некоторое время queue << i puts "#{i} произведено" end end consumer = Thread.new do 10.times do |i| value = queue.pop sleep rand(i/2) puts "потреблено #{value}" end end consumer.join

Попробуйте »

Вывод программы:

SizedQuee Test
0 produced
1 produced
consumed 0
2 produced
consumed 1
consumed 2
3 produced
consumed 34 produced

consumed 4
5 produced
consumed 5
6 produced
consumed 6
7 produced
consumed 7
8 produced
9 produced
consumed 8
consumed 9

Переменные потока

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

Но иногда, что если локальные переменные потока должны быть доступны другим потокам или главному потоку? Ruby предоставляет способ создавать переменные потока по имени, рассматривая поток как таблицу поиска в стиле хеша. Используйте []= для записи данных и [] для чтения данных. Давайте посмотрим на следующий код:

Пример

#!/usr/bin/ruby count = 0 arr = [] 10.times do |i| arr[i] = Thread.new { sleep(rand(0)/10.0) Thread.current["mycount"] = count count += 1 } end arr.each {|t| t.join; print t["mycount"], ", " } puts "count = #{count}"

Вывод вышеприведённого кода:

8, 0, 3, 7, 2, 1, 6, 5, 4, 9, count = 10

Главный поток ожидает завершения выполнения дочерних потоков, затем выводит каждое значение отдельно.

Приоритет потока

Приоритет потока — основной фактор, влияющий на планирование потоков. Другие факторы включают продолжительность времени выполнения CPU, планирование групп потоков и так далее.

Вы можете использовать метод Thread.priority, чтобы получить приоритет потока, и метод Thread.priority=, чтобы изменить приоритет потока.

Приоритет потока по умолчанию равен 0. Потоки с более высоким приоритетом выполняются быстрее.

Поток может получить доступ ко всем данным в своей собственной области, но что, если вам нужно получить доступ к данным других потоков из определённого потока? Класс Thread предоставляет методы для доступа потоков к данным друг друга. Можно просто рассматривать поток как Hash-таблицу и использовать []= в любом потоке для записи данных, а [] — для чтения данных.

athr = Thread.new { Thread.current["name"] = "Thread A"; Thread.stop } bthr = Thread.new { Thread.current["name"] = "Thread B"; Thread.stop } cthr = Thread.new { Thread.current["name"] = "Thread C"; Thread.stop } Thread.list.each {|x| puts "#{x.inspect}: #{x["name"]}" }

Видно, что, рассматривая поток как хеш-таблицу и используя методы [] и []=, мы достигли совместного использования данных между потоками.

Взаимное исключение потоков

Мьютекс (взаимное исключение = мьютекс-блокировка) — это механизм, используемый в многопоточном программировании, чтобы не позволить двум потокам одновременно читать и записывать один и тот же общий ресурс (например, глобальную переменную).

Пример без использования мьютекса

Пример

#!/usr/bin/ruby require 'thread' count1 = count2 = 0 difference = 0 counter = Thread.new do loop do count1 += 1 count2 += 1 end end spy = Thread.new do loop do difference += (count1 - count2).abs end end sleep 1 puts "count1 : #{count1}" puts "count2 : #{count2}" puts "difference : #{difference}"

Вывод приведенного выше примера:

count1 :  9712487
count2 :  12501239
difference : 0

Пример использования мьютекса

Пример

#!/usr/bin/ruby require 'thread' mutex = Mutex.new count1 = count2 = 0 difference = 0 counter = Thread.new do loop do mutex.synchronize do count1 += 1 count2 += 1 end end end spy = Thread.new do loop do mutex.synchronize do difference += (count1 - count2).abs end end end sleep 1 mutex.lock puts "count1 : #{count1}" puts "count2 : #{count2}" puts "difference : #{difference}"

Вывод приведенного выше примера:

count1 :  1336406
count2 :  1336406
difference : 0

Взаимоблокировка

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

Например, если процесс p1 занимает монитор и должен также использовать принтер, но принтер занят процессом p2, а p2 должен использовать монитор, возникает взаимоблокировка.

При использовании объектов Mutex нужно учитывать возможность взаимоблокировок потоков.

Пример

#!/usr/bin/ruby require 'thread' mutex = Mutex.new cv = ConditionVariable.new a = Thread.new { mutex.synchronize { puts "A: У меня есть критическая секция, но я подожду cv" cv.wait(mutex) puts "A: У меня снова критическая секция! Я рулю!" } } puts "(Позже, снова на ранчо...)" b = Thread.new { mutex.synchronize { puts "B: Теперь у меня критическая секция, но я закончил с cv" cv.signal puts "B: Я всё ещё в критической секции, заканчиваю" } } a.join b.join

Вывод приведенного выше примера:

A: I have critical section, but will wait for cv
(Later, back at the ranch...)
B: Now I am critical, but am done with cv
B: I am still critical, finishing up
A: I have critical section again! I rule!

Методы класса Thread

Полный список методов класса Thread выглядит следующим образом:

Порядковый номерОписание метода
1Thread.abort_on_exception
Если его значение — true, то при завершении потока из-за исключения весь интерпретатор будет прерван. Значение по умолчанию — false. То есть при обычных обстоятельствах, если поток возбуждает исключение, не обнаруженное с помощью Thread#join или аналогичного механизма, поток будет завершён без предупреждения.
2Thread.abort_on_exception=
Если установлено вtrue, то при завершении потока из-за исключения весь интерпретатор будет прерван. Возвращает новое состояние.
3Thread.critical
Возвращает логическое значение.
4Thread.critical=
Когда его значение — true, переключение потоков не происходит. Если текущий поток приостановлен (stop) или прерван сигналом, его значение автоматически станет false.
5Thread.current
Возвращает текущий выполняющийся поток (текущий поток).
6Thread.exit
Завершает выполнение текущего потока. Возвращает текущий поток. Если текущий поток является единственным потоком, для завершения его выполнения будет использоваться exit(0).
7Thread.fork { block }
Создаёт поток так же, как Thread.new.
8Thread.kill( aThread )
Завершает выполнение потока.
9Thread.list
Возвращает массив живых потоков, которые находятся в состоянии выполнения или приостановки.
10Thread.main
Возвращает главный поток.
11Thread.new( [ arg ]* ) {| args | block }
Создаёт поток и начинает выполнение. Аргументы передаются блоку без изменений. Это позволяет передавать значения в локальные переменные потока при запуске потока.
12Thread.pass
Уступает выполнение другим потокам. Это не изменяет состояние выполняющегося потока, но передаёт управление другим готовым к выполнению потокам (явное планирование потоков).
13Thread.start( [ args ]* ) {| args | block }
Создаёт поток и начинает выполнение. Аргументы передаются блоку без изменений. Это позволяет передавать значения в локальные переменные потока при запуске потока.
14Thread.stop
Приостанавливает текущий поток, пока другой поток не использует метод run, чтобы снова его разбудить.

Методы экземпляра Thread

Следующий пример вызывает метод создания экземпляра потока join:

Пример

#!/usr/bin/ruby thr = Thread.new do #Создание экземпляра puts "In second thread" raise "Raise exception" end thr.join #Вызывает метод создания экземпляра join

Ниже приведён полный список методов создания экземпляров:

Нет.Описание метода
1thr[ name ]
Возвращает локальные данные потока, соответствующие name. name может быть строкой или символом. Если данных, соответствующих name, нет, возвращается nil.
2thr[ name ] =
Устанавливает значение локальных данных потока, соответствующих name. name может быть строкой или символом. Если задано nil, соответствующие данные в потоке удаляются.
3thr.abort_on_exception
Возвращает логическое значение.
4thr.abort_on_exception=
Если его значение — true, то, как только поток завершается из-за исключения, прерывается весь интерпретатор.
5thr.alive?
Возвращает true, если поток жив.
6thr.exit
Завершает выполнение потока. Возвращает self.
7thr.join
Приостанавливает текущий поток до завершения потока self. Если self завершается из-за исключения, текущий поток возбудит то же исключение.
8thr.key?
Возвращает true, если локальные для потока данные, соответствующие name, уже определены.
9thr.kill
АналогичноThread.exit 。
10thr.priority
Возвращает приоритет потока. Приоритет по умолчанию — 0. Чем больше значение, тем выше приоритет.
11thr.priority=
Задаёт приоритет потока. Также можно задать отрицательное число.
12thr.raise( anException )
Принудительно возбуждает исключение в потоке.
13thr.run
Перезапускает приостановленный (stop) поток. В отличие от wakeup, он немедленно выполняет переключение потока. Если этот метод применяется к мёртвому потоку, возникает исключение ThreadError.
14thr.safe_level
Возвращает уровень безопасности self. safe_level текущего потока совпадает с $SAFE.
15thr.status
Представляет статус живого потока с помощью строк "run", "sleep" или "aborting". Возвращает false, если поток завершился нормально. Возвращает nil, если он завершился из-за исключения.
16thr.stop?
Возвращает true, если поток мёртв или приостановлен (stop).
17thr.value
Ожидает завершения потока self (эквивалентно join), затем возвращает возвращаемое значение блока потока. Если во время выполнения потока возникло исключение, оно возбуждается снова.
18thr.wakeup
Изменяет состояние приостановленного (stop) потока на готовый к выполнению (run). Если этот метод выполняется для мёртвого потока, возникает исключение ThreadError.
Другие расширения