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

Прочитать значение x (124)

Увеличить значение на 1 (125)

Записать результат в x

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

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

В качестве примера рассмотрим задачу индексирования Web-сайтов. Мы извлекаем слова из многочисленных страниц в Сети и сохраняем их в хэше. Ключом является само слово, а значением — строка, идентифицирующая документ и номер строки в этом документе.

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

• будем представлять удаленные документы в виде строк;

• ограничимся всего тремя строками (они будут «зашиты» в код);

• сетевые задержки будем моделировать «засыпанием» на случайный промежуток времени.

Взгляните на программу в листинге 13.1. Она даже не печатает получаемые данные целиком, а выводит лишь счетчик слов (не уникальный). Каждый раз при чтении или обновлении хэша мы вызываем метод hesitate, который приостанавливает поток на случайное время. Тем самым поведение программы становится недетерминированным и приближенным к реальности.

@list = []

@list[0]="shoes ships\nsealing-wax"

@list[1]="cabbages kings"

@list[2]="quarks\nships\ncabbages"

def hesitate

sleep rand(0)

end

@hash = {}

def process_list(listnum)

lnum = 0

@list[listnum].each do |line|

words = line.chomp.split

words.each do |w|

hesitate

if @hash[w]

hesitate

@hash[w] += ["#{listnum}:#{lnum}"]

else

hesitate

@hash[w] = ["#{listnum}:#{lnum}"]

end

end

lnum += 1

end

end

t1 = Thread.new(0) {|num| process_list(num) }

t2 = Thread.new(1) {|num| process_list(num) }

t3 = Thread.new(2) {|num| process_list(num) }

t1.join

t2.join

t3.join

count = 0

@hash.values.each {|v| count += v.size }

puts "Всего слов: #{count} " # Может быть напечатано 7 или 8!

Здесь имеется проблема. Если ваша система ведет себя примерно так же, как наша, то программа может напечатать одно из двух значений! В наших тестах с одинаковой вероятностью печаталось 7 или 8. Если слов и списков больше, то и разброс окажется более широким.

Попробуем исправить положение с помощью мьютекса , который будет контролировать доступ к разделяемому ресурсу. (Слово «mutex» — это сокращение от mutual exclusion, «взаимная блокировка».)

Обратимся к листингу 13.2. Библиотека Mutexпозволяет создавать мьютексы и манипулировать ими. Мы можем захватить (lock) мьютекс перед доступом к хэшу и освободить (unlock) его по завершении операции.

require 'thread.rb'

@list = []

@list[0]="shoes ships\nsealing-wax"

@list[1]="cabbages kings"

@list[2]="quarks\nships\ncabbages"

def hesitate

sleep rand(0)

end

@hash = {}

@mutex = Mutex.new

def process_list(listnum)

lnum = 0

@list[listnum].each do |line|

words = line.chomp.split

words.each do |w|

hesitate

@mutex.lock

if @hash[w]

hesitate

@hash[w] += ["#{listnum}:#{lnum}"]

else

hesitate

@hash[w] = ["#{listnum}:#{lnum}"]

end

@mutex.unlock

end

lnum += 1

end

end

t1 = Thread.new(0) {|num| process_list(num) }

t2 = Thread.new(1) {|num| process_list(num) }

t3 = Thread.new(2) {|num| process_list(num) }

t1.join

t2.join

t3.join

count = 0

@hash.values.each {|v| count += v.size }

puts "Всего слов: #{count} " # Всегда печатается 8!

Отметим, что помимо метода lockв классе Mutexесть также метод try_lock. Он отличается от lockтем, что если мьютекс уже захвачен другим потоком, то он не дожидается освобождения, а сразу возвращает false.

require 'thread'

mutex = Mutex.new

t1 = Thread.new { mutex.lock; sleep 30 }

sleep 1

t2 = Thread.new do

if mutex.try_lock

puts "Захватил"

else

puts "He сумел захватить" # Печатается немедленно.

end

end

sleep 2

Эта возможность полезна, если поток не хочет приостанавливать выполнение. Есть также метод synchronize, который захватывает мьютекс, а потом автоматически освобождает его.

mutex = Mutex.new

mutex.synchronize do

# Любой код, нуждающийся в защите...

end

Существует еще библиотека mutex_m, где определен модуль Mutex_m, который можно подмешивать к классу (или использовать для расширения объекта). У такого расширенного объекта будут все методы мьютекса, так что он сам может выступать в роли мьютекса.

require 'mutex_m'

class MyClass

include Mutex_m

# Теперь любой объект класса MyClass может вызывать

# методы lock, unlock, synchronize...

# Внешние объекты также могут вызывать эти

# методы для объекта MyClass.

end

В библиотеке thread.rbесть пара классов, которые иногда бывают полезны. Класс Queueреализует безопасную относительно потоков очередь, доступ к обоим концам которой синхронизирован. Это означает, что разные потоки могут, ничего не опасаясь, работать с такой очередью. Класс SizedQueueотличается от предыдущего тем, что позволяет ограничить размер очереди (число элементов в ней).

Оба класса имеют практически один и тот же набор методов, поскольку SizedQueueнаследует Queue. Правда, в подклассе определен еще акцессор max, позволяющий получить и установить максимальный размер очереди.

buff = SizedQueue.new(25)

upper1 = buff.max #25

# Увеличить размер очереди...

buff.max = 50

upper2 = buff.max # 50

В листинге 13.3 приведено решение задачи о производителе и потребителе. Для производителя задержка (аргумент sleep) чуть больше, чем для потребителя, чтобы единицы продукции «накапливались».

require 'thread'

buffer = SizedQueue.new(2)

producer = Thread.new do

item = 0

loop do

sleep rand 0

puts "Производитель произвел #{item}"

buffer.enq item

item += 1

end

end

consumer = Thread.new do

loop do

sleep (rand 0)+0.9

item = buffer.deq

puts "Потребитель потребил #{item}"

puts " ожидает = #{buffer.num_waiting}"

end

end

sleep 60 # Работать одну минуту, потом завершить оба потока.