Haskell Language
Concurrencia

Observaciones

Hilos de desove con `forkIO`

Haskell es compatible con muchas formas de concurrencia y la más obvia es forking un hilo usando forkIO .

La función forkIO :: IO () -> IO ThreadId realiza una acción IO y devuelve su ThreadId , mientras que la acción se ejecutará en segundo plano.

Podemos demostrar esto bastante sucintamente usando ghci :

Prelude Control.Concurrent> forkIO $ (print . sum) [1..100000000]
ThreadId 290
Prelude Control.Concurrent> forkIO $ print "hi!"
"hi!"
-- some time later....
Prelude Control.Concurrent> 50000005000000

Ambas acciones se ejecutarán en segundo plano, y la segunda está casi garantizada para finalizar antes de la última.

Comunicando entre hilos con `MVar`

Es muy fácil pasar información entre subprocesos usando el tipo MVar a y las funciones que lo acompañan en Control.Concurrent :

• newEmptyMVar :: IO (MVar a) - crea un nuevo MVar a
• newMVar :: a -> IO (MVar a) - crea una nueva MVar con el valor dado
• takeMVar :: MVar a -> IO a - recupera el valor de MVar dado, o bloquea hasta que haya uno disponible
• putMVar :: MVar a -> a -> IO () - pone el valor dado en el MVar , o lo bloquea hasta que esté vacío

Sumemos los números de 1 a 100 millones en un hilo y esperemos el resultado:

import Control.Concurrent
main = do
  m <- newEmptyMVar
  forkIO $ putMVar m $ sum [1..10000000]
  print =<< takeMVar m  -- takeMVar will block 'til m is non-empty!

Una demostración más compleja podría ser tomar la entrada del usuario y la suma en segundo plano mientras se espera para obtener más información:

main2 = loop
  where 
    loop = do
        m <- newEmptyMVar
        n <- getLine
        putStrLn "Calculating. Please wait"
        -- In another thread, parse the user input and sum
        forkIO $ putMVar m $ sum [1..(read n :: Int)]
        -- In another thread, wait 'til the sum's complete then print it
        forkIO $ print =<< takeMVar m
        loop

Como se indicó anteriormente, si llama a takeMVar y la MVar está vacía, se bloquea hasta que otro hilo MVar algo en la MVar , lo que podría resultar en un problema con los filósofos de la cena . Lo mismo ocurre con putMVar : si está lleno, se bloqueará hasta que esté vacío.

Tome la siguiente función:

concurrent ma mb = do
  a <- takeMVar ma
  b <- takeMVar mb
  putMVar ma a
  putMVar mb b

MVar las dos funciones con algunos MVar s.

concurrent ma mb     -- new thread 1 
concurrent mb ma     -- new thread 2

Lo que podría pasar es que:

1. El hilo 1 lee ma y bloquea ma
2. El hilo 2 lee mb y por lo tanto bloquea mb

Ahora, el subproceso 1 no puede leer mb porque el subproceso 2 lo ha bloqueado, y el subproceso 2 no puede leer ma porque el subproceso 1 lo ha bloqueado. Un callejón sin salida clásico!

Bloques atómicos con software de memoria transaccional

Otra herramienta de concurrencia potente y madura en Haskell es la memoria transaccional de software, que permite que varios subprocesos escriban en una sola variable de tipo TVar a de una manera atómica.

TVar a es el tipo principal asociado con la mónada STM y representa la variable transaccional. Se usan como MVar pero dentro de la mónada STM través de las siguientes funciones:

atomically :: STM a -> IO a

Realizar una serie de acciones de STM de forma atómica.

readTVar :: TVar a -> STM a

Lea el valor de TVar , por ejemplo:

value <- readTVar t

writeTVar :: TVar a -> a -> STM ()

Escribe un valor a la TVar dada.

t <- newTVar Nothing
writeTVar t (Just "Hello")

Este ejemplo está tomado de la Wiki de Haskell:

import Control.Monad
import Control.Concurrent
import Control.Concurrent.STM
 
main = do 
  -- Initialise a new TVar
  shared <- atomically $ newTVar 0
  -- Read the value
  before <- atomRead shared
  putStrLn $ "Before: " ++ show before
  forkIO $ 25 `timesDo` (dispVar shared >> milliSleep 20)
  forkIO $ 10 `timesDo` (appV ((+) 2) shared >> milliSleep 50)
  forkIO $ 20 `timesDo` (appV pred shared >> milliSleep 25)
  milliSleep 800
  after <- atomRead shared
  putStrLn $ "After: " ++ show after
  where timesDo = replicateM_
       milliSleep = threadDelay . (*) 1000

atomRead = atomically . readTVar
dispVar x = atomRead x >>= print
appV fn x = atomically $ readTVar x >>= writeTVar x . fn