Control.Конкурентный

Авторские права	(c) Университет Глазго 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 Источник

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

forkIO :: IO () -> IO ThreadId Source

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

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

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

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

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

forkFinally :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId Source

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

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 Source

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

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

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

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

С момента: base-4.4.0.0

killThread :: ThreadId -> IO () Source

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

killThread tid = throwTo tid ThreadKilled

throwTo :: Exception e => ThreadId -> e -> IO () Source

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

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

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

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

Важное примечание: поведение throwTo отличается от описанного в статье «Асинхронные исключения в 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 Source

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

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

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

С момента: base-4.4.0.0

forkOnWithUnmask :: Int -> ((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId Source

Подобно forkIOWithUnmask, но дочерняя нить привязана к заданному процессору, как и в forkOn.

С момента: base-4.4.0.0

getNumCapabilities :: IO Int Source

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

С момента: base-4.4.0.0

setNumCapabilities :: Int -> IO () Source

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

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

С тех пор: base-4.5.0.0

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

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

С тех пор: base-4.4.0.0

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

Планирование может быть либо прерывистым, либо кооперативным, в зависимости от реализации Concurrent 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 для определения того, существует ли поток.

Since: 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 export запускается, она начинает новый основной поток, и возвращается, когда этот основной поток завершается. Если вызов приводит к созданию новых потоков, они могут оставаться в системе после возвращения вызываемой функции foreign export.

Прерывание

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

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

Ещё одно замечание: пример aaaa bbbb может не работать слишком хорошо в GHC (см. Планирование, выше), из-за блокировки на Handle. Только один поток может удерживать блокировку на Handle в любой момент, поэтому если происходит перепланирование, в то время как поток удерживает блокировку, другой поток не сможет запуститься. Результат заключается в том, что переключение с aaaa на bbbbb происходит редко. Это можно улучшить, уменьшив период перепланирования. У нас также есть исправление, которое вызывает перепланирование всякий раз, когда проснулся поток, ожидающий блокировки, но мы не нашли его полезным для чего-либо, кроме этого примера :-)

Тупик

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

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

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