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

Question

В этом примере я ищу синхронизацию двух puts, так что выход будет ababab..., без каких-либо двойных a s или b s на выходе.

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

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

Пример очереди: Работа.

def a 
    Thread.new do 
    $queue.pop 
    puts "a" 
    b 
    end 
end 

def b 
    Thread.new do 
    sleep(rand) 
    puts "b" 
    $queue << true 
    end 
end 

$queue = Queue.new 
$queue << true 
loop{a; sleep(rand)} 

Файл примера Mutex: не работает.

def a 
    Thread.new do 
    $mutex.flock(File::LOCK_EX) 
    puts "a" 
    b 
    end 
end 

def b 
    Thread.new do 
    sleep(rand) 
    puts "b" 
    $mutex.flock(File::LOCK_UN) 
    end 
end 

MUTEX_FILE_PATH = '/tmp/mutex' 
File.open(MUTEX_FILE_PATH, "w") unless File.exists?(MUTEX_FILE_PATH) 
$mutex = File.new(MUTEX_FILE_PATH,"r+") 
loop{a; sleep(rand)} 

Mutex variable example: Не работает.

def a 
    Thread.new do 
    $mutex.lock 
    puts "a" 
    b 
    end 
end 

def b 
    Thread.new do 
    sleep(rand) 
    puts "b" 
    $mutex.unlock 
    end 
end 

$mutex = Mutex.new 
loop{a; sleep(rand)} 

Answer 1

Короткий ответ
Неправильное использование мьютекса. С Queue вы можете заполнить одним потоком, а затем pop с другим, но вы не можете заблокировать Mutex одним потоком, а затем разблокировать с другим.

Как объяснил @matt, есть несколько тонких вещей, которые происходят как мьютексы, автоматически разблокированные и молчащие исключения, которые вы не видите.

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

Например:

1. thread_a и thread_b могут быть синхронизированы с помощью общей логической переменной, такой как true_a_false_b.
2. При каждом использовании вы должны получить доступ, протестировать и переключить эту логическую переменную - многоэтапный процесс.
3. Необходимо обеспечить, чтобы этот многоступенчатый процесс происходил атомарно, то есть не прерывался. Это когда вы будете использовать мьютекс. Упростил пример следующим образом:

 

require 'thread' 
Thread.abort_on_exception = true 
true_a_false_b = true 
mutex = Mutex.new 

thread_a = Thread.new do 
    loop do 
    mutex.lock 
    if true_a_false_b 
     puts "a" 
     true_a_false_b = false 
    end 
    mutex.unlock 
    end 
end 

thread_b = Thread.new do 
    loop do 
    mutex.lock 
    if !true_a_false_b 
     puts "b" 
     true_a_false_b = true 
    end 
    mutex.unlock 
    end 

sleep(1) # if in irb/console, yield the "current" thread to thread_a and thread_b 

Answer 2

В вашем примере мьютекса, поток создается в методе b спит на некоторое время, печатает b затем пытается разблокировать мьютекс. Это не является законным, поток не может разблокировать мьютекс, если он уже не считает, что замок и поднимает ThreadError, если вы попробуете:

m = Mutex.new 
m.unlock 

результаты:

release.rb:2:in `unlock': Attempt to unlock a mutex which is not locked (ThreadError) 
     from release.rb:2:in `<main>' 

Вы не увидите в вашем примере, потому что by default Ruby silently ignores exceptions raised in threads other than the main thread. Вы можете изменить это с помощью Thread::abort_on_exception= - если добавить

Thread.abort_on_exception = true 

в верхней части файла, вы увидите что-то вроде:

a 
b 
with-mutex.rb:15:in `unlock': Attempt to unlock a mutex which is not locked (ThreadError) 
     from with-mutex.rb:15:in `block in b' 

(вы можете увидеть больше, чем один a, но там будет только один b).

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

В целом замок не имеет большого эффекта. Он не предотвращает запуск потоков b, и хотя он предотвращает два потока a, работающих одновременно, он освобождается, как только поток, удерживающий его, выходит.

Думаю, вы могли бы подумать о semaphores, а в то время как документы Ruby “Mutex implements a simple semaphore” - это not quite the same.

Ruby не предоставляет семафоров в стандартной библиотеке, но предоставляет condition variables. (Эта ссылка относится к старым документам 2.0.0. Стандартная библиотека thread по умолчанию требуется в Ruby 2.1+, и, похоже, это привело к тому, что текущие документы не были доступны.Также имейте в виду, что Ruby также имеет отдельную библиотеку monitor, которая (я думаю) добавляет те же функции (мьютексы и переменные условия) более объектно-ориентированным образом.)

Используя переменные условий и мьютексы, вы можете контролировать координацию между потоки. Uri Agassi’s answer показывает один из возможных способов сделать это (хотя я думаю, что есть состояние гонки с тем, как его решение начинается).

Если вы посмотрите на source for Queue (опять же это ссылка на 2.0.0 - библиотека потоков была преобразована в C в последних версиях, а в Ruby-версии проще), вы можете увидеть, что она реализована с помощью Mutexes и ConditionVariables. Когда вы вызываете $queue.pop в потоке a в очереди, вы в конечном итоге вызываете wait on the mutex так же, как ответ Ури Агасси вызывает $cv.wait($mutex) по его методу a. Аналогичным образом, когда вы вызываете $queue << true в свою b, вы заканчиваете calling signal on the condition variable так же, как звонки Ури Агасси в его b нить.

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

Answer 3

Проблемы с версией файла на основе не разобрались должным образом. Причина, по которой она не работает, заключается в том, что f.flock(File::LOCK_EX) не блокирует, если вызывается в том же файле f несколько раз. Это можно проверить с помощью этой простой последовательной программы:

require 'thread' 

MUTEX_FILE_PATH = '/tmp/mutex' 
$fone= File.new(MUTEX_FILE_PATH, "w") 
$ftwo= File.open(MUTEX_FILE_PATH) 

puts "start" 
$fone.flock(File::LOCK_EX) 
puts "locked" 
$fone.flock(File::LOCK_EX) 
puts "so what" 
$ftwo.flock(File::LOCK_EX) 
puts "dontcare" 

, который печатает все, кроме dontcare.

Так что программа на основе файлов не работает, потому что

$mutex.flock(File::LOCK_EX) 

никогда блоков.