Программирование на языке Ruby - Хэл Фултон

Тривиальный случай возникновения такой ситуации вряд ли возможен. В самом деле, кто станет писать такой код:

@mutex = Mutex.new

@mutex.synchronize do

 @mutex.synchronize do

  #...

 end

end

Но нечто подобное может произойти в сложной программе (или при рекурсивном вызове метода). Какова бы ни была причина, последствием будет тупиковая ситуация. Уход от нее — одно из достоинств модуля-примеси Monitor.

@mutex = Mutex.new

def some_method

 @mutex.synchronize do

  #...

  some_other_method # Тупиковая ситуация!

 end

end

def some_other_method

 @mutex.synchronize do

  #...

 end

end

Модуль-примесь Monitor обычно применяется для расширения объекта. Для создания условной переменной предназначен метод new_cond.

Класс ConditionVariable в библиотеке monitor.rb дополнен по сравнению с определением в библиотеке thread. У него есть методы wait_until и wait_while, которые блокируют поток в ожидании выполнения условия. Кроме того, возможен тайм-аут при ожидании, поскольку у метода wait имеется параметр timeout, равный количеству секунд (по умолчанию nil).

Поскольку примеры работы с потоками у нас кончаются, то в листинге 13.5 мы предлагаем реализацию классов Queue и SizedQueue с помощью монитора. Код приводится с разрешения автора, Шуго Маэда (Shugo Maeda).

# Автор: Shugo Maeda

require 'monitor'

class Queue

 def initialize

  @que = []

  @monitor = Monitor.new

  @empty_cond = @monitor.new_cond

 end

 def enq(obj)

  @monitor.synchronize do

   @que.push(obj)

   @empty_cond.signal

  end

 end

 def deq

  @monitor.synchronize do

   while @que.empty?

    @empty_cond.wait

   end

   return @que.shift

  end

 end

end

class SizedQueue < Queue

 attr :max

 def initialize(max)

  super()

  @max = max

  @full_cond = @monitor.new_cond

 end

 def enq(obj)

  @monitor.synchronize do

   while @que.length >= @max

    @full_cond.wait

   end

   super(obj)

  end

 end

 def deq

  @monitor.synchronize do

   obj = super

   if @que.length < @max

    @full_cond.signal

   end

   return obj

  end

 end

 def max=(max)

  @monitor.synchronize do

   @max = max

   @full_cond.broadcast

  end

 end

end

Еще один вариант синхронизации (двузначную блокировку со счетчиком) предлагает библиотека sync.rb. В ней определен модуль Sync_m, который можно применять вместе с ключевыми словами include и extend (как и Mutex_m). Этот модуль содержит методы locked?, shared?, exclusive?, lock, unlock и try_lock.

13.2.6. Тайм-аут при выполнении операций

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

Библиотека timeout.rb предлагает решение этой проблемы на основе потоков (см. листинг 13.6). С методом timeout ассоциирован выполняемый блок. Если истечет заданное число секунд, метод возбуждает исключение TimeoutError, которое можно перехватить с помощью rescue.

require 'timeout.rb'

flag = false

answer = nil

begin

 timeout(5) do

  puts "Хочу печенье!"

  answer = gets.chomp

  flag = true

 end

 rescue TimeoutError

 flag = false

end

if flag

 if answer == "cookie"

  puts "Спасибо! Хрум, хрум..."

 else

  puts "Это же не печенье!"

  exit

 end

else

 puts "Эй, слишком медленно!"

 exit

end

puts "До встречи..."

13.2.7. Ожидание события

Часто один или несколько потоков следят за «внешним миром», а остальные выполняют полезную работу. Все примеры в этом разделе надуманные, но общий принцип они все же иллюстрируют.

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

$flag = false

work1 = Thread.new { job1() }

work2 = Thread.new { job2() }

work3 = Thread.new { job3() }

thread4 = Thread.new { Thread.stop; job4() }

thread5 = Thread.new { Thread.stop; job5() }

watcher = Thread.new do

 loop do

  sleep 5

  if $flag

   thread4.wakeup

   thread5.wakeup

   Thread.exit

  end

 end

end

Если в какой-то момент выполнения метода job, переменная $flag станет равной true, то в течение пяти секунд после этого потоки thread4 и thread5 гарантированно запустятся. После этого поток watcher завершается.

В следующем примере мы ждем создания файла. Каждые 30 секунд проверяется его существование, и как только файл появится, мы запускаем новый поток. Тем временем остальные потоки занимаются своим делом. На самом деле ниже мы наблюдаем за тремя разными файлами.

def waitfor(filename)

 loop do

  if File.exist? filename

   file_processor = Thread.new { process_file(filename) }

   Thread.exit

  else

   sleep 30

  end

 end

end

waiter1 = Thread.new { waitfor("Godot") }

sleep 10

waiter2 = Thread.new { waitfor("Guffman") }

sleep 10

headwaiter = Thread.new { waitfor("head") }

# Основной поток занимается другими делами...

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

13.2.8. Продолжение обработки во время ввода/вывода

Часто приложению приходится выполнять одну или более длительных операций ввода/вывода. Прежде всего, речь идет о вводе данных с клавиатуры, поскольку человек печатает куда медленнее, чем вращается диск. Это время можно употребить на пользу с помощью потоков.

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

Предположим, что итератор predict_move генерирует вероятные ходы человека (и ответные ходы программы). Тогда в момент, когда человек сделает ход, не исключено, что у компьютера уже будет готов ответ.

scenario = {} # Хэш ход-ответ.

humans_turn = true

thinking_ahead = Thread.new(board) do

 predict_move do |m|

  scenario[m] = my_response(board,m)

  Thread.exit if humans_turn == false

 end

end

human_move = get_human_move(board)

humans_turn = false # Остановить поток.

# Теперь можно посмотреть, нет ли в хэше scenario хода,

# сделанного пользователем...

Конечно, настоящие шахматные программы работают не так.

13.2.9. Реализация параллельных итераторов

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

Рассмотрим следующий пример. Пусть compose — имя магического метода, который выполняет композицию итераторов. Допустим еще, что у каждого объекта есть стандартный итератор each и что каждый объект возвращает по одному элементу на каждой итерации.

arr1 = [1, 2, 3, 4]

arr2 = [5, 10, 15, 20]

compose(arr1, arr2) {|a,b| puts "#{а} и #{b}" }

# Должно быть напечатано:

# 1 и 5

# 2 и 10

# 3 и 15

# 4 и 20

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