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

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.

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

Библиотека 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 "До встречи..."

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

В следующем примере прикладную задачу решают три потока. Четвертый поток каждые пять секунд просыпается, проверяет глобальную переменную $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") }

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

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

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

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

Предположим, что итератор 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 хода,

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

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

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

Рассмотрим следующий пример. Пусть 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.

def compose(*objects)

threads = []

for obj in objects do

threads << Thread.new(obj) do |myobj|

me = Thread.current

me[:queue] = []

myobj.each {|x| me[:queue].push(x) }

end

end

list = [0] # Фиктивное значение, отличное от nil.

while list.nitems > 0 do # Еще есть не nil.

list = []

for thr in threads

list << thr[:queue].shift # Удалить по одному из каждого.

end

yield list if list.nitems > 0 # He вызывать yield, если все равны nil.

end

end

x = [1, 2, 3, 4, 5, 6, 7, 8]

y = " первый\n второй\n третий\n четвертый\n пятый\n"

z = %w[a b с d e f]

compose(x, у, z) {|a,b,c| p [a, b, c] }

# Выводится:

# [1, " первый\n", "a"]

# [2, " второй\n", "b"]

# [3, " третий\n", "c"]