Haskell concurrent

Haskell concurrent amplia Haskell98 amb concurrència explícita.

Els dos conceptes principals en què es basa Haskell concurrent són les Mutable variables MVar α i la possibilitat d'engegar un nou fil d'execució via forkIO.

Concurrència

forkIO

engega un fil d'execució lleuger del planificador del Run Time System.

accés sincronitzat amb variables MVar (Mutable variable)

La consulta d'una variable MVar (takeMVar) en buida el contingut causant que altres fils d'execució que hi accedeixin posteriorment quedin blocats en espera que se'n reposi el valor actualitzat (amb putMVar).

• "takeMVar mvar" bloca el fil d'exec. si mvar estava buida
• "putMVar mvar valor" bloca si mvar estava plena

Generadors:

do 
  mvarPlenaInicialment <- newMVar contingut     -- newMVar :: t -> IO (MVar t)

  mvarBuidaInicialment <- newEmptyMVar :: IO (MVar T)   -- mvar per a un contingut de tipus T

Una MVar té tres facetes:

• Variable amb accés sincronitzat
• Bústia de comunicació d'un sol element (takeMVar com a receive, putMVar com a send)
• Semàfor binari (takeMVar com a wait, putMVar com a signal)

mvarSemàfor <- newEmptyMVar :: IO (MVar Bool)

-- engega fil d'exec. i en acabar 
--        desperta el primer dels fils suspesos pendents de la MVar
threadId <- forkIO procés `finally` putMVar mvarSemàfor True
...
-- suspèn en espera que el fil d'exec. de ''procés'' acabi
takeMVar mvarSemàfor

Darrerament s'hi han afegit crides no blocants (tryTakeMVar, tryPutMVar).

També tenim modifyMVar_ (composició de takeMVar i putMVar que retorna IO ()) i altres novetats.

L'accés a una variable MVar és d'avaluació tardana, per tant el contingut serà avaluat en el consumidor i no en el productor!!

Els fils blocats es desperten per ordre de suspensió (FIFO). "putMVar mvar" desperta el primer dels fils que ha cridat takeMVar amb la mateixa mvar, que s'emporta el valor i la deixa buida.

bústies de comunicació amb MVar, Chan, BoundedChan

Canals amb sincronització per baldes (ang: locks).

• MVar: bústies d'un sol element (de Control.Concurrent)
• Chan: bústies amb cua il·limitada (de Control.Concurrent.Chan)
• BoundedChan: bústies limitades (del paquet BoundedChan)

Exemples:

• #Concurrència simple amb MVars - Productor-consumidor
• i afegint-hi un canal asíncron a #Client-servidor - Cues d'entrada (Chan)

fils d'execució amb crides externes o amb allotjament local al fil. Bound threads

GHC implementa multitasca cooperativa, assignant els fils d'exec. lleugers llançats amb forkIO en relació N-M amb els fils del sistema (anomenats capability) un per cada processador elemental. No s'ha implementat el de multitasca arrabassadora, ang:preemptive.

forkOS: llança un fil d'execució lligat als del sistema (Bound threads), per fer ús de crides externes (FFI: Foreign Function Interface) en un fil, o l'ús d'allotjament lligat al fil d'execució. Biblioteques com OpenGL requereixen aquest ús.

Paral·lelisme

Paral·lelisme de tasques - Compilació per a processadors multicor

Vegeu

par

par:: a -> b -> b -- activa el càlcul del primer operand en paral·lel (que s'encua en espera d'una CPU disponible) mentre que el segon s'executa al fil d'exec. actual, retornant el resultat d'aquest darrer.

pseq

pseq:: a -> b -> b -- avalua el primer operand en el fil d'exec. actual, de manera primerenca (estricta) i avalua el segon de manera tardana (lazy) quin resultat retorna. Vegeu "seq vs. pseq"

opció multiprocessador

L'opció -threaded de "ghc --make" relliga el programa amb la biblio. del Run Time System multiprocessador, emprant diversos fils d'execució del sistema per a possibilitar el paral·lelisme, altrament el relliga amb la de l' RTS uniprocessador.

GHC conté un planificador de fils d'execució lleugers, llançats amb forkIO, que reparteix l'execució als nuclis de CPU als quals assigna internament un fil d'execució del sistema llançats internament amb forkOS. la funció forkOnIO permet designar el nucli on desitgem que s'executi.

 -- fibo paral·lel

 import Control.Parallel (par, pseq)

 parfib :: Int -> Integer
 parfib 0 = 0
 parfib 1 = 1
 parfib n | n > 1 = nf2 `par` (nf1 `pseq` (nf1+nf2)) -- calc nf2 en un fil en paral·lel i nf1 al fil principal
                                            --   i seqüencialment, al fil principal,
                                            --   en acabar en retorna la suma
           where nf1 = parfib (n-1)
                 nf2 = parfib (n-2)

 main = print $ parfib 10

-- compilació amb -threaded per fer servir la biblio. "Run Time System multi-processador"
-- afegir -rtsopts per poder afegir paràmetres al llançador per a l'R.T.S.

ghc --make -threaded -rtsopts parfib.hs

-- execució mostrant estadístiques  "+RTS -s". Afegirem -Nx per a un nombre x de processadors elementals.

-- "si el user time (temps en mode usuari) és major que l' elapsed time (temps transcorregut)
--     és que s'ha emprat més d'un processador
-- proveu-ho també sense -Nx

-- per distingir les opcions d'execució específiques per al Haskell Run Time System de les del programa, 
--    cal escriure els params. per al Haskell després de +RTS 
--                        i tancar amb -RTS si volem afegir params. per al programa.
./parfib +RTS -s -N2

Estratègies

Vegeu ref.

import Control.Parallel.Strategies (parMap, rpar)

llista = [2..11]
càlcul x = 2 * sqrt x

mapejaEnParallel = parMap rpar

resultatsDelCàlculEnParallel = mapejaEnParallel càlcul llista

main = print resultatsDelCàlculEnParallel

compilació i exec.

ghc --make -threaded -rtsopts prova.hs

#             opcions: estadístiques: -s, utilitza tots els nuclis de CPU: -N
./prova +RTS -s -N

la mònada Par

Permet combinar tasques en paral·lel encadenant resultats com a paràmetres de paral·lelitzacions subseqüents. Vegeu refs. Similar a la clàusula Async del llenguatge F#.

Exemples de concurrència

Concurrència simple amb MVars - Productor-consumidor

Amb variables de sincronització per baldes (blocants) MVar.

forkIO

Engega fil d'execució lleuger del planificador del Haskell en multiprocés cooperatiu.

putMVar mvar valor

bloca si la variable MVar és plena (ocupada) fins que estigui disponible (buida) i llavors l'omple amb el valor.

takeMVar mvar

bloca el fil d'execució si la variable MVar és buida fins que la li omplin i en retorna el valor buidant-la.

module Main( main ) where

import Control.Concurrent (forkIO, threadDelay, MVar, newEmptyMVar, putMVar, takeMVar)
import Control.Exception (finally)
import qualified Control.Monad as Monad
import System.IO (stdout, hFlush)
import Text.Printf (printf)
import Data.Time (FormatTime, formatTime, getCurrentTime, utcToLocalZonedTime)
import System.Locale (defaultTimeLocale)

obtenir_hora :: IO String
obtenir_hora = do
    local_t <- (getCurrentTime >>= utcToLocalZonedTime)   -- els parèntesis hi són per legibilitat 

    return $ formatTime defaultTimeLocale "%T" local_t

productor :: MVar Int -> IO ()
productor mv_bústia = do
  Monad.forM_ [3,2..0] $ \compte_enrere -> do  -- per als valors de la llista
    threadDelay 1000000            -- espera microsegons
    putMVar mv_bústia compte_enrere    -- posa valor a la MVar

consumidor :: MVar Int -> IO ()
consumidor mv_bústia = do
  catch
    (Monad.forever $ do        -- repeteix seqüencialment
      x <- takeMVar mv_bústia  -- bloca mentre no li omplin la MVar
      h <- obtenir_hora
      printf "%s - consumidor: recollit %d\n" h x
      hFlush stdout
      if x == 0 then ioError (userError "s'ha acabat") -- excepció per sortir del "forever"
         else return ()
    )
    ( \ _excep -> return ()                -- recull l'excepció
    )

main = do
 -- nova MVar per la sicronització productor / consumidor
 mv_bústia <- newEmptyMVar :: IO (MVar Int)

 -- nova MVar per la sincronització de finalització de fil d'exec.
 mv_fi_prod <- newEmptyMVar :: IO (MVar Bool)
 mv_fi_consum <- newEmptyMVar :: IO (MVar Bool)

 -- forkIO: engega fil d'execució
 consumidor_id <- forkIO $ consumidor mv_bústia `finally`
                             putMVar mv_fi_consum True  -- assenyala l'acabament a la MVar
                                  -- despertant el primer dels fils blocats per la mateixa

 productor_id <- forkIO $ productor mv_bústia `finally`
                            putMVar mv_fi_prod True

 -- emulem amb MVar's la feina de ''pthread_join()'' de l'Unix
 --  per esperar la finalització dels fils d'exec. creats

 takeMVar mv_fi_prod       -- bloca fins a la fi del productor
 takeMVar mv_fi_consum      -- bloca fins a la fi del consumidor
 putStrLn "fi del programa"

produeix la sortida següent:

11:32:03 - consumidor: recollit 3
11:32:04 - consumidor: recollit 2
11:32:05 - consumidor: recollit 1
11:32:06 - consumidor: recollit 0
fi del programa

Client-servidor - Cues d'entrada (Chan)

Client-servidor, canalitzant la impressió

Canals no acotats (en la dimensió de la cua) (Control.Concurrent.Chan) "de primera classe" (és a dir, que es pot passar com a paràmetre)

• forkIO: Engega fil d'execució lleuger del planificador del Haskell.
• forkOS: engega un fil lligat a un del sistema operatiu.
• comunic. per cues il·limitades

writeChan canal

afegeix a la cua il·limitada i retorna tot seguit (sense blocar)

readChan canal

bloca si la cua del canal és buida

• resposta per MVars

putMVar mvar valor

bloca si la variable MVar és plena (ocupada) fins que estigui disponible (buida) i llavors l'omple amb el valor.

takeMVar mvar

bloca el fil d'execució si la variable MVar és buida fins que la li omplin i en retorna el valor buidant-la.

• A l'exemple el client encua el parell (comanda, ref. resposta (mv_resposta)), i queda a l'espera de la resposta.

module Main( main ) where

import Control.Concurrent (forkIO, forkOS, threadDelay, 
                           MVar, newEmptyMVar, putMVar, takeMVar, isEmptyMVar)
import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
import Control.Exception (finally)
import Data.IORef (IORef, newIORef, readIORef, writeIORef, modifyIORef)
import qualified Control.Monad as Monad
import System.IO (stdout, hFlush)
import Text.Printf (printf)
import Data.Time (FormatTime, formatTime, getCurrentTime, utcToLocalZonedTime)
import System.Locale (defaultTimeLocale)

data TInfo = InfoDelClient Int | InfoDelServidor String Int | InfoPlega  -- missatges a l'informador

type Canal_Comanda = Chan (Int, MVar_Resposta)
type MVar_Resposta = MVar Int
type Canal_Info = Chan TInfo


obtenir_hora :: IO String
obtenir_hora = do
        local_t <- (getCurrentTime >>= utcToLocalZonedTime)  -- els parèntesis hi són per legibilitat però, de fet, no calen
        return $ formatTime defaultTimeLocale "%T" local_t

-- equivalent
-- obtenir_hora = getCurrentTime >>= utcToLocalZonedTime >>= (return . formatTime defaultTimeLocale "%T")

client :: Canal_Comanda -> MVar_Resposta -> Canal_Info -> IO ()
client chan_torn mv_resposta chan_info = do
  Monad.forM_ [3,2..0] $ \cnt -> do    -- llista de valors a passar, finalitzant en zero
    threadDelay 1000000                -- espera microsegons
    writeChan chan_info (InfoDelClient cnt)  -- no bloca (asíncron, encua al canal i continua)

    -- assegura que la mvar de resposta sigui buida
    mv_resp_esBuida <- isEmptyMVar mv_resposta
    Monad.when (not mv_resp_esBuida) (takeMVar mv_resposta >> return ())

    -- encua la comanda passant la ref. de la mvar de resposta.
    writeChan chan_torn (cnt, mv_resposta)   
    takeMVar mv_resposta                     -- bloca (espera resposta per continuar)

obtenir_resposta :: Int -> IORef Int -> IO Int
obtenir_resposta x ref_estat = readIORef ref_estat >>= (return . (+x))  -- la que vulgueu

servidor :: Canal_Comanda -> Canal_Info -> IORef Int -> IO ()
servidor chan_torn chan_info ref_estat = do
  catch
    (Monad.forever $ do
        (x, mv_resposta) <- readChan chan_torn               -- bloca fins obtenir comanda al chan_torn
        h <- obtenir_hora
        resp <- obtenir_resposta x ref_estat

        mv_resp_esBuida <- isEmptyMVar mv_resposta
        Monad.when mv_resp_esBuida $ putMVar mv_resposta resp    -- respon si és possible

        writeChan chan_info (InfoDelServidor h x)   -- no bloca (asíncron, encua al canal i continua)
        if x == 0 then ioError $ userError "s'ha acabat"  -- excepció per sortir del ''forever''
                  else return ()
     )
     (\ _excep -> return ()
     )

informador :: Canal_Info -> IO ()
informador chan_info =
  catch (
    Monad.forever $ do
        info <- readChan chan_info           -- bloca si la cua és buida
        case info of
          InfoDelClient intValor -> do
                printf "client: comanda %d\n" intValor
                hFlush stdout

          InfoDelServidor strHora intValor -> do
                printf "%s - servidor: recollit %d\n" strHora intValor
                hFlush stdout

          InfoPlega -> ioError $ userError "s'ha acabat"  -- excepció per sortir del ''forever''
    )
    (\ _excep -> return ()
    )

main = do
  ref_estat <- newIORef 0 :: IO (IORef Int) -- ref. no sincronitzada (la manipula un sol fil)
  chan_torn <- newChan :: IO (Canal_Comanda)  -- cua de comandes al servidor
  mv_resposta <- newEmptyMVar :: IO (MVar_Resposta) -- ref. sincronitzada

  chan_informacio <- newChan :: IO (Canal_Info)  -- cua d'impressió

  -- semàfors per a l'espera d'acabament dels fils d'execució
  mv_fi_client <- newEmptyMVar :: IO (MVar Bool)
  mv_fi_servidor <- newEmptyMVar :: IO (MVar Bool)
  mv_fi_info <- newEmptyMVar :: IO (MVar Bool)

  -- malgrat que la gestió de ''stdout'' pel fil d'exec. de l'informador 
  --   no requereix ''bound threads'', li poso el forkOS per trencar el tabú.

  informador_id <- forkOS {- per les crides externes -} $ informador chan_informacio
                                 `finally` putMVar mv_fi_info True
  servidor_id <- forkIO $ servidor chan_torn chan_informacio ref_estat
                                 `finally` putMVar mv_fi_servidor True
  client_id <- forkIO $ client chan_torn mv_resposta chan_informacio
                                 `finally` putMVar mv_fi_client True

  takeMVar mv_fi_client              -- espera fi client
  takeMVar mv_fi_servidor            -- espera fi servidor

  writeChan chan_informacio InfoPlega
  takeMVar mv_fi_info        -- espera fi informador

  putStrLn "fi del programa"

dóna la següent sortida:

client: comanda 3
15:18:55 - servidor: recollit 3
client: comanda 2
15:18:56 - servidor: recollit 2
client: comanda 1
15:18:57 - servidor: recollit 1
client: comanda 0
15:18:58 - servidor: recollit 0
fi del programa

Concurrència condicionada amb TVars - Mònada STM - Memòria transaccional

Només al compilador GHC. Les transaccions de memòria eviten blocar els fils d'execució, descartant canvis a les variables transaccionals si no es completa, excepte en cas que hi posem condicions forçant el reintent amb la clàusula "retry".

L'evolució de la transacció es modela com a aplicacions en una Mònada STM, inicials de "Software Transactional Memory".

En aquest exemple les transaccions s'efectuen sobre variables transaccionals TVar (accés sincronitzat per STM) i s'encapsulen en una mònada STM.

També substituïm les les MVar (comunic. síncrona) per TMVar, i les Chan (comunic. asíncrona) per TChan, per quedar lliures de problemes de bloquejos.

atomically

admet o tot o no res dels canvis a les variables transaccionals, passa el resultat Mònada STM a Mònada IO

retry

provoca el reintent si no es donen les condicions esperades i reintenta una transacció alternativa per la branca "orElse" si existeix, i si no, bloca el fil d'execució fins que es modifiqui alguna de les variables transaccionals implicades en la transacció.

orElse

introdueix una transacció alternativa, que s'avalua si la primera fa "retry"

always

comprova invariant i si falla, genera un error "Transactional invariant violation" finalitzant el programa

catchSTM

atrapa excepcions dins la mònada STM

A partir de GHC 6.12, STM desapareix de la biblioteca pral. i, si no s'ha fet la instal·lació amb la Plataforma Haskell que l'incorpora, caldrà carregar el paquet stm del Hackage.

-- transaccions als comptes --fitxer stm_part1.hs

module Stm_part1 (aporta_quan_cal_i_obtenir_saldo, 
                  retira_fons_de_dos_comptes
                 ) where  
 
import Control.Monad.STM (STM, retry, orElse, always)
import Control.Concurrent.STM.TVar (TVar, readTVar, writeTVar)
import qualified Control.Monad as Monad
 
saldo_baix = 4
 
aporta_quan_saldo_baix :: TVar Int -> Int -> STM ()
aporta_quan_saldo_baix tv_compte aportacio = do 
 
		saldo <- readTVar tv_compte   
		Monad.when (saldo > saldo_baix) retry   -- bloca si no es dóna la condició, 
                                                        -- fins que es modifiqui alguna TVar, llavors reintenta
		let nou_saldo = saldo + aportacio
		writeTVar tv_compte nou_saldo
 
invariant :: TVar Int -> STM Bool
invariant tv_compte = do
                saldo <- readTVar tv_compte 
                return $ saldo >= 0
 
retira_fons :: TVar Int -> Int -> STM ()
retira_fons tv_compte quantitat = do
 
		saldo <- readTVar tv_compte
		Monad.when (saldo < quantitat) retry      -- si no hi ha saldo reintenta la transacció alternativa 
                                                          -- o bloca i torna a la inicial si no hi ha més alternatives
		writeTVar tv_compte $ saldo - quantitat
		always $ invariant tv_compte              -- comprova invariant de la transacció
 
retira_fons_de_dos_comptes :: TVar Int -> TVar Int -> Int -> STM Int
retira_fons_de_dos_comptes tv_compteA tv_compteB quantitat = do
 
                retira_fons tv_compteA quantitat `orElse`  -- alternativa de transacció
                       retira_fons tv_compteB quantitat

                saldoA <- readTVar tv_compteA
                saldoB <- readTVar tv_compteB
                return $ saldoA + saldoB
 
aporta_quan_cal_i_obtenir_saldo :: TVar Int -> TVar Int -> Int -> STM Int
aporta_quan_cal_i_obtenir_saldo tv_compteA tv_compteB quantitat = do
 
                aporta_quan_saldo_baix tv_compteB quantitat
                saldoA <- readTVar tv_compteA
                saldoB <- readTVar tv_compteB
                return $ saldoA + saldoB

Principal engegant fils d'execució per a creditor, deutor i informador (gestiona stdout).

module Main( main ) where  --fitxer stm_main.hs

import Stm_part1
 
import Control.Concurrent (forkIO, threadDelay, killThread
                           ,MVar, newEmptyMVar, putMVar, takeMVar)
import Control.Monad.STM (STM, atomically)
import Control.Concurrent.STM.TVar (TVar, newTVar)
import Control.Concurrent.STM.TMVar (TMVar, newEmptyTMVarIO, putTMVar, takeTMVar)
import Control.Concurrent.STM.TChan (TChan, newTChan, readTChan, writeTChan)
import Control.Exception (finally, block)
import qualified Control.Monad as Monad
import System.IO (stdout, hFlush)
import Text.Printf (printf)
 
data TInfo = InfoDelCreditor Int Int | InfoDelDeutor Int | InfoPlega    -- missatges a l'informador
 
pagament = 3

-- creditor passa rebuts al cobrament de manera periòdica

creditor :: TVar Int -> TVar Int -> TChan TInfo -> IO ()
creditor tv_compteA tv_compteB tchan_informacio = do
 
  Monad.forM_ ([1,2..6]::[Int]) $ \periode -> do    -- per als periodes de la llista
    threadDelay 1000000                             -- espera microsegons
    saldo_conjunt <- atomically $ retira_fons_de_dos_comptes tv_compteA tv_compteB pagament
    atomically $ writeTChan tchan_informacio $ InfoDelCreditor periode saldo_conjunt

-- deutor aporta diners al compte, quan el saldo baixa per sota d'un valor ''saldo_baix'' 
 
deutor :: TVar Int -> TVar Int -> TChan TInfo -> IO ()
deutor  tv_compteA tv_compteB tchan_informacio = do
 
  Monad.forever $ do
     block $ do           -- block: no interrompible per excepcions asíncrones, tractar-les en completar el bloc
        saldo_conjunt <- atomically $ aporta_quan_cal_i_obtenir_saldo tv_compteA tv_compteB pagament
        atomically $ writeTChan tchan_informacio $ InfoDelDeutor saldo_conjunt

-- informador: gestiona sortides a ''stdout'' en un sol fil d'execució
-- vehicula missatges a imprimir a través del canal transaccional TChan (versió transac. de Chan)

informador :: TChan TInfo -> IO ()
informador tchan_informacio = do
  catch (
    Monad.forever $ do
        info <- atomically $ readTChan tchan_informacio               -- bloca mentre canal buit
        case info of
          InfoDelCreditor periode saldo -> do
                    printf "creditor: periode %d, saldo %d\n" periode saldo
                    hFlush stdout
          InfoDelDeutor saldo -> do
                    printf "deutor: saldo %d\n" saldo
                    hFlush stdout
          InfoPlega -> ioError $ userError "s'ha acabat"   -- excepció per sortir del "forever"
    )
    (\ _excep -> return ())
 
main = do
  tv_compteA <- atomically $ newTVar 10                -- compte A
  tv_compteB <- atomically $ newTVar 4                 -- compte B
 
  tchan_informacio <- atomically $ newTChan :: STM (TChan TInfo)   -- canal per a la informació a imprimir
 
  -- semàfors d'acabament de fils d'execució

  mv_fi_deutor <- newEmptyMVar :: IO (MVar Bool)          
  mv_fi_creditor <- newEmptyMVar :: IO (MVar Bool)
  mv_fi_informador <- newEmptyMVar :: IO (MVar Bool)
 
  informador_id <- forkIO $ informador tchan_informacio        -- forkIO: engega fil d'execució
                               `finally` (putMVar mv_fi_informador True) -- desperta els fils pausats per la MVar 
 
  deutor_id <- forkIO $ deutor tv_compteA tv_compteB tchan_informacio  
                               `finally` (putMVar mv_fi_deutor True)
 
  creditor_id <- forkIO $ creditor tv_compteA tv_compteB tchan_informacio  
                               `finally` (putMVar mv_fi_creditor True)
 
  takeMVar mv_fi_creditor            -- espera fi creditor
 
  killThread deutor_id               -- genera excepció asíncrona al deutor
  takeMVar mv_fi_deutor              -- espera fi deutor
 
  atomically $ writeTChan tchan_informacio InfoPlega    -- afegeix ordre de plegar al canal de l'informador
 
  takeMVar mv_fi_informador          -- espera fi informador
  putStrLn "fi del programa"

Compilació i exec.

ghc --make stm_part1.hs stm_main.hs -o stm_main
./stm_main