Z-IO-0.6.4.0: Simple and high performance IO toolkit for Haskell
Copyright(c) Dong Han 2017-2020
LicenseBSD
Maintainerwinterland1989@gmail.com
Stabilityexperimental
Portabilitynon-portable
Safe HaskellNone
LanguageHaskell2010

Z.IO.BIO.Concurrent

Description

This module provides some concurrent BIO node, to ease the implementation of producer-consumer model. All sources and sinks return by this module are safe to be used in multiple threads.

  • Use newTQueueNode for common cases.
  • Use newTBQueueNode if you have a fast producer and you don't want input get piled up in memory.
  • Use newBroadcastTChanNode if you want messages get broadcasted, i.e. every message written by producers will be received by every consumers.

It's important to correctly set the numebr of producers, internally it keeps a counter on how many producers reached their ends, and send EOF to all consumers when last producer ends. So it's a good idea to catch exceptions and pull the sink(which indicate EOF) on producer side.

(sink, src) <- newTQueueNode 2  -- it's important to correctly set the numebr of producers

--------------------------------------------------------------------------------
-- producers

forkIO $ do
    ...
    push x sink             -- producer using push
    ...
    pull sink               -- when EOF is reached, manually pull, you may consider put it in a bracket.

forkIO $ do
    ...
    (runBIO $ ... >|> sink) -- producer using BIO
        onException (pull sink)

--------------------------------------------------------------------------------
-- consumers

forkIO $ do
    ...
    r <- pull src           -- consumer using pull
    case r of Just r' -> ...
              _ -> ...      -- Nothing indicate all producers reached EOF

forkIO $ do
    ...
    runBIO $ src >|> ...    -- consumer using BIO
Synopsis

Documentation

newTQueueNode Source #

Arguments

:: Int

number of producers

-> IO (Sink a, Source a) 

Make an unbounded queue and a pair of sink and souce connected to it.

newTBQueueNode Source #

Arguments

:: Int

number of producers

-> Natural

queue buffer bound

-> IO (Sink a, Source a) 

Make an bounded queue and a pair of sink and souce connected to it.

newBroadcastTChanNode Source #

Arguments

:: Int

number of producers

-> IO (Sink a, IO (Source a))

(Sink, IO Source)

Make a broadcast chan and a sink connected to it, and a function return sources to receive broadcast message.