Pusher.hs

module Main (main) where

import           Control.Concurrent.Async
import qualified Control.Exception        as Exception
import           Control.Monad            as Monad
import qualified Data.ByteString          as ByteString
import qualified Data.Map.Strict          as Map
import           System.Clock
import           System.IO
import           System.Posix.Process
import qualified System.Process           as P
import Text.Printf

maxPause = 10 * 1000 * 1000

main :: IO ()
main = do
    (hOutr, hOutw) <- P.createPipe
    hSetBuffering hOutr NoBuffering

    let pinger = forever $ do
          t <- getTime Monotonic
          hPrint hOutw t
    putStrLn "Starting"


    let ponger t0 = do
          t1 <- read <$> hGetLine hOutr
          let t10 = toNanoSecs $ diffTimeSpec t1 t0
          when (t10 > maxPause) $
              printf "Detected pause of %0.3f milliseconds\n" (fromIntegral t10 / 1e6 :: Double)
          ponger t1

    _ <- forkProcess $ do
      ponger =<< getTime Monotonic

    race mainPusher pinger
    hClose hOutw

    putStrLn "Done"

------------------------------------------------------------------------
-- Pusher model

data Msg = Msg !Int !ByteString.ByteString
type Chan = Map.Map Int ByteString.ByteString

message :: Int -> Msg
message n = Msg n (ByteString.replicate 1024 (fromIntegral n))

pushMsg :: Chan -> Msg -> IO Chan
pushMsg chan (Msg msgId msgContent) =
  Exception.evaluate $
    let
      inserted = Map.insert msgId msgContent chan
    in
      if 500000 < Map.size inserted
      then Map.deleteMin inserted
      else inserted

mainPusher :: IO ()
mainPusher = Monad.foldM_ pushMsg Map.empty (map message [1..2000000])