Control.Concurrent

Авторское право	(c) The University of Glasgow 2001
Лицензия	В стиле BSD (см. файл libraries/base/LICENSE)
Сопровождающий	libraries@haskell.org
Стабильность	стабильный
Переносимость	непереносимый (параллелизм)
Безопасный Haskell	Заслуживающий доверия
Язык	Haskell2010

Параллельный Haskell

Расширение параллелизма для Haskell описано в статье Concurrent Haskell http://www.haskell.org/ghc/docs/papers/concurrent-haskell.ps.gz.

Параллелизм является «лёгковесным»: создание потоков и переключение контекста имеют крайне низкие накладные расходы. Планирование потоков Haskell выполняется внутри среды выполнения Haskell и не использует предоставляемые операционной системой пакеты потоков.

Однако, если вам нужно взаимодействовать со сторонней библиотекой, которая требует от программы использования предоставляемого операционной системой пакета потоков, это можно сделать, используя forkOS вместо forkIO.

Потоки Haskell могут обмениваться данными через MVar — разновидность синхронизированной изменяемой переменной (см. Control.Concurrent.MVar). На основе MVar можно создать несколько распространённых абстракций параллелизма; они предоставляются модулем Control.Concurrent. В GHC потоки также могут обмениваться данными с помощью исключений.

Основные операции параллелизма

data ThreadId Источник

ThreadId — абстрактный тип, представляющий дескриптор потока. ThreadId является экземпляром Eq, Ord и Show; экземпляр Ord реализует произвольный полный порядок для ThreadId. Экземпляр Show позволяет преобразовать ThreadId с произвольным значением в строковое представление; вывод значения ThreadId иногда полезен при отладке или диагностике поведения параллельной программы.

Примечание: в GHC наличие ThreadId фактически означает, что у вас есть указатель непосредственно на поток. Это означает, что сам поток нельзя собрать сборщиком мусора, пока вы не освободите ThreadId. Эту нежелательную особенность было бы сложно устранить, сохранив поддержку threadStatus.

myThreadId :: IO ThreadId Источник

Возвращает ThreadId вызывающего потока (только GHC).

forkIO :: IO () -> IO ThreadId Источник

Создаёт новый поток для выполнения вычисления IO, переданного первым аргументом, и возвращает ThreadId созданного потока.

Новый поток будет лёгковесным, непривязанным потоком. Вызовы внешних функций из этого потока не гарантируется выполнять в каком-либо определённом потоке ОС; если внешние вызовы должны выполняться в определённом потоке ОС, используйте forkOS.

Новый поток наследует маскированное состояние родительского потока (см. mask).

Для вновь созданного потока устанавливается обработчик исключений, который отбрасывает исключения BlockedIndefinitelyOnMVar, BlockedIndefinitelyOnSTM и ThreadKilled, а все остальные исключения передаёт обработчику неперехваченных исключений.

ПРЕДУПРЕЖДЕНИЕ: исключения в новом потоке не будут повторно возбуждены в потоке, который его создал. Это означает, что вы можете совершенно не узнать о проблеме, если она возникнет. Вместо этого можно использовать библиотеку async.

forkFinally :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId Источник

Создаёт поток и вызывает переданную функцию непосредственно перед его завершением, передавая ей исключение или возвращённое значение. Функция вызывается с замаскированными асинхронными исключениями.

forkFinally action and_then =
  mask $ \restore ->
    forkIO $ try (restore action) >>= and_then

Эта функция полезна, например, чтобы сообщить родительскому потоку о завершении дочернего.

Начиная с: base-4.6.0.0

forkIOWithUnmask :: ((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId Источник

Как forkIO, но дочернему потоку передаётся функция, с помощью которой можно снять маску с асинхронных исключений. Обычно эту функцию используют следующим образом:

 ... mask_ $ forkIOWithUnmask $ \unmask ->
                catch (unmask ...) handler

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

Обратите внимание: переданную дочернему потоку функцию снятия маски следует использовать только в этом потоке; её вызов в другом потоке приводит к неопределённому поведению.

Начиная с: base-4.4.0.0

killThread :: ThreadId -> IO () Источник

killThread возбуждает исключение ThreadKilled в указанном потоке (только GHC).

killThread tid = throwTo tid ThreadKilled

throwTo :: Exception e => ThreadId -> e -> IO () Источник

throwTo возбуждает произвольное исключение в целевом потоке (только GHC).

Доставка исключения синхронизирована между исходным и целевым потоками: throwTo не возвращает управление, пока исключение не будет возбуждено в целевом потоке. Поэтому вызывающий поток может быть уверен, что целевой поток получил исключение. Доставка исключения также атомарна относительно других исключений. Атомарность полезна при работе с состояниями гонки: например, если два потока могут завершить друг друга, гарантируется, что завершить другой поток сможет только один из них.

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

Если целевой поток в данный момент выполняет внешний вызов, исключение не будет возбуждено (и, следовательно, throwTo не вернёт управление), пока вызов не завершится. Это верно независимо от того, находится ли вызов внутри mask. Однако в GHC внешний вызов можно пометить атрибутом interruptible; в этом случае throwTo заставит RTS попытаться прервать вызов и вернуть управление. Дополнительные сведения см. в документации GHC.

Важное примечание: поведение throwTo отличается от описанного в статье «Asynchronous exceptions in Haskell» (http://research.microsoft.com/~simonpj/Papers/asynch-exns.htm). В статье throwTo не блокирует выполнение, однако реализация библиотеки использует более синхронный подход, при котором throwTo не возвращает управление, пока целевой поток не получит исключение. Компромисс между этими подходами обсуждается в разделе 9 статьи. Поэтому, как и любая блокирующая операция, throwTo допускает прерывание (см. раздел 5.3 статьи). Однако, в отличие от других прерываемых операций, throwTo всегда допускает прерывание, даже если фактически не блокирует выполнение.

Нет гарантии, что исключение будет доставлено быстро, хотя среда выполнения постарается избежать произвольных задержек. В GHC исключение может быть возбуждено только в безопасной точке: безопасной считается точка, в которой происходит выделение памяти. Некоторые циклы не выделяют память внутри себя, поэтому их нельзя прервать с помощью throwTo.

Если целью throwTo является вызывающий поток, поведение совпадает с throwIO, за исключением того, что исключение выбрасывается как асинхронное. Это означает, что при наличии охватывающего чистого вычисления, что возможно, если текущая операция IO находится внутри unsafePerformIO или unsafeInterleaveIO, это вычисление не заменяется исключением окончательно, а приостанавливается, как если бы оно получило асинхронное исключение.

Обратите внимание: если throwTo вызвана с текущим потоком в качестве целевого, исключение будет выброшено, даже если поток в данный момент находится внутри mask или uninterruptibleMask.

Потоки с привязкой

forkOn :: Int -> IO () -> IO ThreadId Источник

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

Аргумент Int задаёт номер способности (см. getNumCapabilities). Обычно способности соответствуют физическим процессорам, но точное поведение зависит от реализации. Значение, переданное в forkOn, интерпретируется по модулю общего числа способностей, возвращаемого getNumCapabilities.

Примечание для GHC: число способностей задаётся параметром +RTS -N при запуске программы. Если базовая операционная система поддерживает эту возможность, способности можно привязать к реальным ядрам процессора с помощью +RTS -qa, однако на практике это обычно не требуется (а в некоторых случаях может даже снизить производительность — рекомендуется экспериментировать).

Начиная с: base-4.4.0.0

forkOnWithUnmask :: Int -> ((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId Источник

Как forkIOWithUnmask, но дочерний поток закрепляется за указанным ЦП, как и при использовании forkOn.

Начиная с: base-4.4.0.0

getNumCapabilities :: IO Int Источник

Возвращает число потоков Haskell, которые в любой момент времени могут выполняться действительно одновременно (на отдельных физических процессорах). Чтобы изменить это значение, используйте setNumCapabilities.

Начиная с: base-4.4.0.0

setNumCapabilities :: Int -> IO () Источник

Устанавливает число потоков Haskell, которые в любой момент времени могут выполняться действительно одновременно (на отдельных физических процессорах). Число, переданное в forkOn, интерпретируется по модулю этого значения. Начальное значение задаётся параметром среды выполнения +RTS -N.

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

Начиная с: base-4.5.0.0

threadCapability :: ThreadId -> IO (Int, Bool) Источник

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

Начиная с: base-4.4.0.0

Планирование

Планирование может быть вытесняющим или кооперативным — это зависит от реализации параллельного Haskell (сведения о конкретных компиляторах приведены ниже). В кооперативной системе переключение контекста происходит только при использовании одной из примитивных операций, определённых в этом модуле. Это означает, что такие программы, как:

  main = forkIO (write 'a') >> write 'b'
    where write c = putChar c >> write c

выведут либо aaaaaaaaaaaaaa..., либо bbbbbbbbbbbb..., а не произвольное чередование a и b. На практике кооперативной многозадачности достаточно для создания простых графических пользовательских интерфейсов.

yield :: IO () Источник

Действие yield позволяет (а в реализации с кооперативной многозадачностью — принудительно вызывает) переключение контекста на любой другой поток, готовый к выполнению (если таковой имеется), и иногда полезно при реализации абстракций параллелизма.

Блокировка

Разные реализации Haskell по-разному ведут себя в отношении операций, блокирующих все потоки.

При использовании GHC без параметра -threaded все внешние вызовы блокируют все остальные потоки Haskell в системе, но операции ввода-вывода этого не делают. С параметром -threaded все остальные потоки блокируются только внешними вызовами с атрибутом unsafe.

Ожидание

threadDelay :: Int -> IO () Источник

Приостанавливает текущий поток на указанное число микросекунд (только GHC).

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

Не превышайте maxBound :: Int: на 32-разрядных машинах это всего 2147483647 мкс, то есть меньше 36 минут. Рассмотрите возможность использования Control.Concurrent.Thread.Delay.delay из пакета unbounded-delays.

threadWaitRead :: Fd -> IO () Источник

Блокирует текущий поток до появления данных для чтения из указанного файлового дескриптора (только GHC).

Если файловый дескриптор был закрыт, пока поток находился в заблокированном состоянии, будет выброшено исключение IOError. Чтобы безопасно закрыть файловый дескриптор, использовавшийся с threadWaitRead, используйте closeFdWith.

threadWaitWrite :: Fd -> IO () Источник

Блокирует текущий поток до появления возможности записать данные в указанный файловый дескриптор (только GHC).

Если файловый дескриптор был закрыт, пока поток находился в заблокированном состоянии, будет выброшено исключение IOError. Чтобы безопасно закрыть файловый дескриптор, использовавшийся с threadWaitWrite, используйте closeFdWith.

threadWaitReadSTM :: Fd -> IO (STM (), IO ()) Источник

Возвращает действие STM, которое можно использовать для ожидания данных для чтения из файлового дескриптора. Второе возвращаемое значение — действие IO, с помощью которого можно отменить регистрацию интереса к файловому дескриптору.

Начиная с: base-4.7.0.0

threadWaitWriteSTM :: Fd -> IO (STM (), IO ()) Источник

Возвращает действие STM, которое можно использовать для ожидания возможности записать данные в файловый дескриптор. Второе возвращаемое значение — действие IO, с помощью которого можно отменить регистрацию интереса к файловому дескриптору.

Начиная с: base-4.7.0.0

Абстракции для обмена данными

module GHC.Internal.Control.Concurrent.MVar

module Control.Concurrent.Chan

module Control.Concurrent.QSem

module Control.Concurrent.QSemN

Привязанные потоки

Поддержка нескольких потоков операционной системы и описанных ниже привязанных потоков в настоящее время доступна только в среде выполнения GHC при компоновке с параметром -threaded.

Другие системы Haskell в настоящее время не поддерживают несколько потоков операционной системы.

Привязанный поток — это поток Haskell, привязанный к потоку операционной системы. Хотя планированием привязанного потока по-прежнему занимается среда выполнения Haskell, все внешние вызовы, выполняемые этим потоком, обслуживает поток операционной системы.

Для внешней библиотеки привязанный поток будет выглядеть в точности как обычный поток операционной системы, созданный с помощью таких функций ОС, как pthread_create или CreateThread.

Привязанные потоки можно создавать с помощью приведённой ниже функции forkOS. Все экспортированные во внешний мир функции выполняются в привязанном потоке (привязанном к потоку ОС, вызвавшему функцию). Кроме того, действие main каждой программы на Haskell выполняется в привязанном потоке.

Зачем это нужно? Если внешняя библиотека вызывается из потока, созданного с помощью forkIO, у неё не будет доступа к состоянию, локальному для потока — переменным состояния, значения которых различаются для каждого потока ОС (см. pthread_key_create в POSIX или TlsAlloc в Win32). Поэтому некоторые библиотеки (например, OpenGL) не будут работать из потока, созданного с помощью forkIO. Они нормально работают в потоках, созданных с помощью forkOS, или при вызове из main либо из foreign export.

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

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

rtsSupportsBoundThreads :: Bool Источник

True, если поддерживаются привязанные потоки. Если rtsSupportsBoundThreads равно False, isCurrentThreadBound всегда будет возвращать False, а и forkOS, и runInBoundThread завершатся с ошибкой.

forkOS :: IO () -> IO ThreadId Источник

Как и forkIO, эта функция запускает новый поток для выполнения вычисления IO, переданного первым аргументом, и возвращает ThreadId вновь созданного потока.

Однако forkOS создаёт привязанный поток, необходимый, если требуется вызывать внешние (не относящиеся к Haskell) библиотеки, использующие состояние, локальное для потока, например OpenGL (см. Control.Concurrent).

Использование forkOS вместо forkIO совершенно не влияет на планирование в системе выполнения Haskell. Распространено заблуждение, будто для предотвращения блокировки всех потоков Haskell при выполнении внешнего вызова нужно использовать forkOS вместо forkIO; это не так. Чтобы внешние вызовы не блокировали все потоки Haskell (в GHC), достаточно при компоновке программы использовать параметр -threaded и убедиться, что внешний импорт не помечен как unsafe.

forkOSWithUnmask :: ((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId Источник

Как и forkIOWithUnmask, но дочерний поток является привязанным, как и в случае с forkOS.

isCurrentThreadBound :: IO Bool Источник

Возвращает True, если вызывающий поток является привязанным, то есть если из него безопасно использовать внешние библиотеки, зависящие от состояния, локального для потока.

runInBoundThread :: IO a -> IO a Источник

Выполняет вычисление IO, переданное первым аргументом. Если вызывающий поток не является привязанным, временно создаётся привязанный поток. runInBoundThread не завершается, пока не завершится вычисление IO.

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

runInUnboundThread :: IO a -> IO a Источник

Выполняет вычисление IO, переданное первым аргументом. Если вызывающий поток является привязанным, с помощью forkIO временно создаётся непривязанный поток. runInBoundThread не завершается, пока не завершится вычисление IO.

Используйте эту функцию только в редких случаях, когда вы действительно наблюдали снижение производительности из-за использования привязанных потоков. Если главному потоку программы не требуется быть привязанным, а параллельные вычисления используются интенсивно (например, в веб-сервере), может иметь смысл обернуть действие main в runInUnboundThread.

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

Слабые ссылки на ThreadId

mkWeakThreadId :: ThreadId -> IO (Weak ThreadId) Источник

Создаёт слабый указатель на ThreadId. Это может быть важно, если требуется хранить ссылку на ThreadId и при этом позволить потоку получать семейство исключений BlockedIndefinitely (например, BlockedIndefinitelyOnMVar). Обычная сильная ссылка ThreadId помешает доставке исключений BlockedIndefinitely, поскольку в любой момент эта ссылка может быть использована как цель для throwTo, что разблокирует поток.

С другой стороны, хранение Weak ThreadId не помешает потоку получать исключения BlockedIndefinitely. Исключение по-прежнему можно отправить Weak ThreadId, но вызывающая сторона должна сначала использовать deRefWeak, чтобы определить, существует ли ещё поток.

Начиная с: base-4.6.0.0

Реализация параллелизма в GHC

В этом разделе описаны возможности, специфичные для реализации Concurrent Haskell в GHC.

Потоки Haskell и потоки операционной системы

В GHC потоки, создаваемые с помощью forkIO, являются лёгковесными и полностью управляются средой выполнения GHC. Как правило, потоки Haskell на порядок или два эффективнее потоков операционной системы как по времени, так и по объёму используемой памяти.

Недостаток лёгковесных потоков состоит в том, что одновременно может выполняться только один из них. Поэтому, например, если один поток блокируется при внешнем вызове, остальные потоки не могут продолжать работу. При необходимости среда выполнения GHC обходит это ограничение, используя полноценные потоки ОС. Если программа собрана с параметром -threaded (для компоновки с многопоточной версией среды выполнения), поток, выполняющий внешний вызов safe, не блокирует остальные потоки системы: другой поток ОС продолжит выполнять потоки Haskell до возврата исходного вызова. Среда выполнения поддерживает пул таких рабочих потоков, поэтому несколько потоков Haskell могут одновременно выполнять внешние вызовы.

Модуль System.IO выполняет мультиплексирование по-своему. В системах Windows он использует внешние вызовы safe, чтобы потоки, выполняющие операции ввода-вывода, не блокировали всю среду выполнения. В системах Unix все текущие заблокированные запросы ввода-вывода обрабатываются одним потоком (потоком диспетчера ввода-вывода) с помощью такого механизма, как epoll или kqueue, в зависимости от возможностей операционной системы.

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

Если не использовать параметр -threaded, среда выполнения не использует несколько потоков ОС. Внешние вызовы блокируют все остальные выполняющиеся потоки Haskell до своего завершения. Модуль System.IO по-прежнему выполняет мультиплексирование, поэтому несколько потоков могут одновременно выполнять ввод-вывод; внутри среды выполнения это обрабатывается с помощью select.

Завершение программы

В автономной программе GHC для завершения процесса достаточно завершения главного потока. Поэтому все остальные порождённые потоки просто завершаются одновременно с главным потоком (такое поведение называют «демоническими потоками»).

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

  myForkIO :: IO () -> IO (MVar ())
  myForkIO io = do
    mvar <- newEmptyMVar
    forkFinally io (\_ -> putMVar mvar ())
    return mvar

Обратите внимание: мы используем forkFinally, чтобы гарантировать запись MVar, даже если поток по какой-либо причине завершится или будет принудительно остановлен.

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

   children :: MVar [MVar ()]
   children = unsafePerformIO (newMVar [])

   waitForChildren :: IO ()
   waitForChildren = do
     cs <- takeMVar children
     case cs of
       []   -> return ()
       m:ms -> do
          putMVar children ms
          takeMVar m
          waitForChildren

   forkChild :: IO () -> IO ThreadId
   forkChild io = do
       mvar <- newEmptyMVar
       childs <- takeMVar children
       putMVar children (mvar:childs)
       forkFinally io (\_ -> putMVar mvar ())

    main =
      later waitForChildren $
      ...

Принцип главного потока также применим к вызовам Haskell извне с помощью foreign export. При вызове функции foreign exported запускается новый главный поток, и функция возвращает управление, когда этот главный поток завершается. Если вызов порождает новые потоки, они могут остаться в системе после возврата функции foreign exported.

Вытеснение

GHC реализует вытесняющую многозадачность: выполнение потоков случайным образом чередуется. Точнее, поток может быть вытеснен при каждом выделении памяти. К сожалению, это означает, что плотные циклы, не выделяющие память, обычно блокируют остальные потоки (однако это, по-видимому, происходит только в патологических примерах кода, похожих на тесты производительности).

По умолчанию таймер перепланирования срабатывает с периодичностью 20 мс, но это значение можно изменить с помощью параметра RTS -i<n>. После «тика» перепланирования выполняющийся поток вытесняется при первой возможности.

И последнее замечание: пример с aaaa bbbb может работать в GHC не очень хорошо (см. раздел «Планирование» выше) из-за блокировки на Handle. Одновременно блокировку на Handle может удерживать только один поток, поэтому, если во время удержания блокировки происходит перепланирование, другой поток не сможет выполняться. В результате переключение с aaaa на bbbbb происходит редко. Ситуацию можно улучшить, уменьшив период тика перепланирования. У нас также есть исправление, которое вызывает перепланирование при пробуждении потока, ожидающего блокировку, но мы не нашли ему применения, кроме этого примера :-)

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

GHC пытается обнаруживать взаимную блокировку потоков с помощью сборщика мусора. Поток, который недостижим (его нельзя найти, проследовав по указателям от живых объектов), должен быть заблокирован; в этом случае ему отправляется исключение. Это исключение может быть BlockedIndefinitelyOnMVar, BlockedIndefinitelyOnSTM, NonTermination или Deadlock — в зависимости от причины взаимной блокировки потока.

Обратите внимание: эта возможность предназначена для отладки, и на неё не следует полагаться для правильной работы программы. Нет гарантии, что сборщик мусора сможет достаточно точно обнаружить взаимную блокировку или что он запустится достаточно своевременно. По сути, к обнаружению взаимных блокировок применимы те же оговорки, что и к финализаторам.

Между обнаружением взаимных блокировок и финализаторами (созданными с помощью newForeignPtr или функций из System.Mem.Weak) существует тонкое взаимодействие: если поток заблокирован в ожидании запуска финализатора, он будет считаться заблокированным и получит исключение. Поэтому лучше не делать так, но, если альтернативы нет, можно предотвратить признание потока заблокированным, создав StablePtr, указывающий на него. Не забудьте позднее освободить StablePtr с помощью freeStablePtr.