events.hs

{-# OPTIONS_GHC -Wall #-}

-- Idea: A good runtime system makes event-based systems easier to implement.

--------------------------------------------------------------------------------

import Control.Concurrent (threadDelay)
import qualified Control.Concurrent.Async as A
import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)
import Control.Monad (unless, void)
import Data.ByteString (ByteString)
import Data.Maybe (isJust)

--------------------------------------------------------------------------------

-- | Timeout in milliseconds.
newtype Timeout = Timeout { timeoutMs :: Int }
  deriving (Bounded)

-- | Implementation 1: Blocking. This function returns after the data is
-- completely sent.
sendBlocking :: ByteString -> IO ()
sendBlocking = undefined

-- | Implementation 2: Polling. This function starts the operation and returns
-- immediately. Returned poll function can be used for checking the status.
-- polling function for checking the status.
sendNonBlocking_poll :: ByteString -> IO (Maybe Timeout -> IO Bool)
sendNonBlocking_poll = undefined

-- | Implementation 3: Callbacks. This function starts the operation and returns
-- immediately. Callback is called by another thread when operation is done.
sendNonBlocking_cb :: ByteString -> IO () -> IO ()
sendNonBlocking_cb = undefined

-- With a good language and runtime system we can easily convert each one of
-- these to other two.

--------------------------------------------------------------------------------
-- Blocking to others

sendNonBlocking_poll__blocking :: ByteString -> IO (Maybe Timeout -> IO Bool)
sendNonBlocking_poll__blocking bs =
    mkPoll <$> A.async (sendBlocking bs)
  where
    mkPoll :: A.Async () -> Maybe Timeout -> IO Bool
    mkPoll thread Nothing  = isJust <$> A.poll thread
    mkPoll thread (Just t) = do
      timer <- A.async (threadDelay (timeoutMs t * 1000))
      either (const False) (const True) <$> A.waitEither timer thread

sendNonBlocking_cb__blocking :: ByteString -> IO () -> IO ()
sendNonBlocking_cb__blocking bs cb =
    void (A.async (sendBlocking bs >> cb))

--------------------------------------------------------------------------------
-- Polling to others

sendBlocking__poll :: ByteString -> IO ()
sendBlocking__poll bs =
    sendNonBlocking_poll bs >>= loop
  where
    loop poll = do
      ret <- poll (Just maxBound)
      unless ret (loop poll)

sendNonBlocking_cb__poll :: ByteString -> IO () -> IO ()
sendNonBlocking_cb__poll bs cb =
    void (A.async (sendBlocking__poll bs >> cb))

--------------------------------------------------------------------------------
-- Callback to others

sendBlocking__cb :: ByteString -> IO ()
sendBlocking__cb bs = do
    mvar <- newEmptyMVar
    sendNonBlocking_cb bs (putMVar mvar ())
    takeMVar mvar

sendNonBlocking_poll__cb :: ByteString -> IO (Maybe Timeout -> IO Bool)
sendNonBlocking_poll__cb bs = do
    mvar <- newEmptyMVar
    sendNonBlocking_cb bs (putMVar mvar ())
    return (mkPoll mvar)
  where
    mkPoll :: MVar () -> Maybe Timeout -> IO Bool
    mkPoll mvar Nothing  = takeMVar mvar >> return True
    mkPoll mvar (Just t) = do
      timer  <- A.async (threadDelay (timeoutMs t * 1000))
      thread <- A.async (takeMVar mvar)
      either (const False) (const True) <$> A.waitEitherCancel timer thread

--------------------------------------------------------------------------------
-- What features did we use?
--
-- * From blocking to polling:  Cancellable threads, timers.
-- * From blocking to callback: Just threads.
-- * From polling to blocking:  Nothing.
-- * From polling to callback:  Just threads.
-- * From callback to blocking: Essentially a semaphore (`MVar` in Haskell).
-- * From callback to polling:  A semaphore (`MVar`), cancellable threads, timers.