haskell Asynchronous Exception Safety for Channels

unboundedChannel.hs

#!/usr/bin/env stack

{-
 stack
--resolver lts-17.13 runghc
-}
--
--  chapter-9, Parallel-concurrent-programming in haskell
-- chmod 755 ./unboundedChannel.hs && ./unboundedChannel.hs
--
{-# LANGUAGE OverloadedStrings #-}

import Control.Exception(mask_)
import Control.Concurrent
  ( MVar,
    newEmptyMVar,
    newMVar,
    putMVar,
    -- readMVar,
    takeMVar,
    modifyMVar,
  )

type Stream a = MVar (Item a)

data Item a = Item a (Stream a)

data Chan a = Chan (MVar (Stream a)) (MVar (Stream a))

newChan :: IO (Chan a)
newChan = do
  hole <- newEmptyMVar
  readVar <- newMVar hole
  writeVar <- newMVar hole
  return (Chan readVar writeVar)

readChan :: Chan a -> IO a
readChan (Chan readVar _) =  do
  modifyMVar readVar $ \stream -> do
    Item val tail <- readMVar stream
    pure (tail, val)

readMVar :: MVar a -> IO a
readMVar m = mask_ $ do
  a <- takeMVar m
  putMVar m a
  return a

writeChan :: Chan a -> a -> IO ()
writeChan (Chan _ writeVar) val = do
  newHole <- newEmptyMVar
  mask_ $ do
        oldHole <- takeMVar writeVar
        putMVar oldHole (Item val newHole)
        putMVar writeVar newHole

main :: IO ()
main = do
  channe1 <- newChan
  writeChan channe1 "hello-from channel-1"
  s <- readChan channe1
  putStrLn s