unliftio-messagebox (empty) → 1.0.0
raw patch · 30 files changed
+5800/−0 lines, 30 filesdep +HUnitdep +QuickCheckdep +atomic-primops
Dependencies added: HUnit, QuickCheck, atomic-primops, base, containers, criterion, data-default, hashable, mtl, tasty, tasty-html, tasty-hunit, tasty-quickcheck, text, unagi-chan, unliftio, unliftio-messagebox
Files
- LICENSE +9/−0
- README.md +73/−0
- src-benchmark/BookStoreBenchmark.hs +264/−0
- src-benchmark/CommandBenchmark.hs +59/−0
- src-benchmark/Main.hs +154/−0
- src-benchmark/MediaBenchmark.hs +618/−0
- src-memleak-test/Main.hs +39/−0
- src-memleak-test/MediaBenchmark.hs +646/−0
- src-test/CallIdTest.hs +54/−0
- src-test/CatchAllTest.hs +58/−0
- src-test/CommandTest.hs +862/−0
- src-test/CornerCaseTests.hs +296/−0
- src-test/FreshTest.hs +9/−0
- src-test/LimitedMessageBoxTest.hs +486/−0
- src-test/Main.hs +40/−0
- src-test/MessageBoxClassTest.hs +330/−0
- src-test/MessageBoxCommon.hs +129/−0
- src-test/QCOrphans.hs +30/−0
- src-test/UnlimitedMessageBoxTest.hs +21/−0
- src-test/Utils.hs +181/−0
- src/UnliftIO/MessageBox.hs +91/−0
- src/UnliftIO/MessageBox/CatchAll.hs +95/−0
- src/UnliftIO/MessageBox/Class.hs +119/−0
- src/UnliftIO/MessageBox/Command.hs +327/−0
- src/UnliftIO/MessageBox/Limited.hs +339/−0
- src/UnliftIO/MessageBox/Unlimited.hs +117/−0
- src/UnliftIO/MessageBox/Util/CallId.hs +41/−0
- src/UnliftIO/MessageBox/Util/Fresh.hs +69/−0
- src/UnliftIO/MessageBox/Util/Future.hs +32/−0
- unliftio-messagebox.cabal +212/−0
+ LICENSE view
@@ -0,0 +1,9 @@+Copyright 2020 Sven Heyll <sven.heyll@gmail.com>++Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:++1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.++2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,73 @@+# Fast and Robust Message Queues for Concurrent Processes++**NOTE:** To be able to fully view this README, use the [web page](https://sheyll.github.io/unliftio-messagebox/).++A thin wrapper around a subset of `unagi-chan` based on `unliftio`.++The unit tests and benchmarks in this project excercise a subset +of `unagi-chan`, so hopefully the user of this library doesn't+have to worry about this.++This project also aims to implicitly verify parts of `unagi-chan`.++The overall goal is to reduce the risk of live and dead locks and +thread starvation, as well as acceptable performance +even in massively concurrent programs.++Additionally in +## Module Structure++The library is contained in modules with names starting with +**UnliftIO.MessageBox**.++++Also the module +`UnliftIO.MessageBox` [(API docs)](./generated-reports/haddock-report/unliftio-messagebox/UnliftIO-MessageBox.html)+[(Hackage)](http://hackage.haskell.org/package/unliftio-messagebox/docs/UnliftIO-MessageBox.html)+exposes the API, and can be used to import everything.++The full documentation is either [on this page](./generated-reports/haddock-report/unliftio-messagebox/index.html)[(build log)](./generated-reports/haddock-report/build.log)+or on [Hackage](http://hackage.haskell.org/package/unliftio-messagebox).++## Benchmarks++* [Single Core Results](./generated-reports/benchmark-report/benchmark-1-CORES.html)+* [Multi Core Results](./generated-reports/benchmark-report/benchmark-ALL-CORES.html)++## Unit Tests++* [Test Results](./generated-reports/test-profiling-report/test-result.html)+* [Test Coverage](./generated-reports/test-coverage-report/hpc_index.html)+### Heap Profiling++++### Time Profiling++[Test Profiling Report](./generated-reports/test-profiling-report/unliftio-messagebox-test.prof)++## Memory Leak Test ++This is a small application with a 1002 processes, each performing a fix amount of +work.++When the work is done, all processes synchronise before starting a new iteration.+After each iteration, the memory usage is queried from the GHC runtime +statistics.+There are 100 iterations like that. ++After that all processes are shutdown, and the process+starts all over again, 30 times.++In total 3000 iterations.++If memory is leaked it should become visible.++The test program is executated with the `+RTS -M400m` option that instructs+the runtime to limit the available heap to 400MB, so when there is a memory+leak, the program would at some point crash with a heap exhaustion error.++++The output is printed into [this log file](./generated-reports/memleak-test-report/test.log).
+ src-benchmark/BookStoreBenchmark.hs view
@@ -0,0 +1,264 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE EmptyDataDecls #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++-- | Basic benchmark for the Command api.+-- In this benchmark n producers `Donate` books to the book store,+-- which are processed by m consumers+module BookStoreBenchmark (benchmark) where++import Control.Monad (unless, replicateM)+import Criterion.Types+ (Benchmark, bench,+ bgroup,+ nfAppIO,+ )+import Data.Semigroup (Semigroup (stimes))+import UnliftIO.MessageBox.Command as Command+ ( Command,+ Message (Blocking, NonBlocking),+ ReturnType (FireAndForget, Return),+ call,+ cast,+ replyTo,+ )+import UnliftIO.MessageBox.Util.CallId+ ( CallId,+ HasCallIdCounter (getCallIdCounter),+ )+import UnliftIO.MessageBox.Util.Fresh+ ( CounterVar,+ newCounterVar,+ )+import UnliftIO.MessageBox.Class+ ( IsMessageBox(newInput, receive),+ IsMessageBoxFactory(newMessageBox),+ handleMessage )+import UnliftIO+ ( MonadIO (liftIO),+ MonadUnliftIO,+ conc,+ runConc,+ )+import Control.Monad.Reader (ReaderT(runReaderT))+import Data.Foldable (traverse_)++mkExampleBook :: Int -> Book+mkExampleBook !i =+ MkBook+ ( "The not so very very very very very very very very very very very very very very very very very very very very very very very very " ++ show i,+ "large",+ "Laoreet non curabitur gravida arcu ac tortor dignissim convallis aenean. Placerat in egestas erat imperdiet sed euismod nisi porta. Id consectetur purus ut faucibus pulvinar. Nulla porttitor massa id neque aliquam vestibulum morbi blandit. Risus nullam eget felis eget nunc lobortis. Et malesuada fames ac turpis. Pellentesque nec nam aliquam sem. Tellus rutrum tellus pellentesque eu tincidunt tortor aliquam nulla facilisi. Aliquam id diam maecenas ultricies mi. Eu lobortis elementum nibh tellus molestie nunc non." ++ show i,+ ( "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.",+ "Dolor sit amet consectetur adipiscing elit ut aliquam. Integer eget aliquet nibh praesent tristique magna. Nulla facilisi morbi tempus iaculis. Cursus mattis molestie a iaculis at erat pellentesque adipiscing. Posuere sollicitudin aliquam ultrices sagittis orci a scelerisque purus semper. Amet venenatis urna cursus eget nunc scelerisque viverra. Fermentum odio eu feugiat pretium nibh ipsum consequat nisl. Vitae auctor eu augue ut lectus arcu bibendum at varius. Quis commodo odio aenean sed adipiscing diam donec adipiscing tristique. Dictumst quisque sagittis purus sit amet volutpat consequat mauris nunc. Integer vitae justo eget magna fermentum iaculis eu non diam. Et egestas quis ipsum suspendisse ultrices gravida dictum fusce ut. Senectus et netus et malesuada. In arcu cursus euismod quis viverra. Fames ac turpis egestas integer. Tortor condimentum lacinia quis vel eros donec ac odio. Interdum velit laoreet id donec ultrices tincidunt arcu non. Aenean et tortor at risus viverra adipiscing at.",+ even i,+ 123423421111111111111111111123234 * toInteger i+ )+ )++newtype Book = MkBook ([Char], [Char], [Char], ([Char], [Char], Bool, Integer))+ deriving stock (Show, Eq, Ord)++data BookStore++data instance Command BookStore _ where+ Donate :: Book -> Command BookStore 'FireAndForget+ GetBooks :: Command BookStore ( 'Return [Book])++deriving stock instance Eq (Command BookStore 'FireAndForget)++deriving stock instance Show (Command BookStore 'FireAndForget)++deriving stock instance Eq (Command BookStore ( 'Return [Book]))++deriving stock instance Show (Command BookStore ( 'Return [Book]))++newtype BookStoreEnv = MkBookStoreEnv+ {_fresh :: CounterVar CallId}++instance HasCallIdCounter BookStoreEnv where+ getCallIdCounter MkBookStoreEnv {_fresh} = _fresh++onlyCasts ::+ (MonadUnliftIO m, IsMessageBoxFactory cfg) =>+ (Int -> Book) ->+ cfg ->+ (Int, Int, Int) ->+ m ()+onlyCasts !msgGen !impl (!nP, !nMTotal, !nC) = do+ freshCounter <- newCounterVar+ flip runReaderT (MkBookStoreEnv {_fresh = freshCounter}) $ do+ bookStoreOutput <- newMessageBox impl+ bookStoreInput <- newInput bookStoreOutput+ let producer 0 = pure ()+ producer workLeft = do+ ok <- cast bookStoreInput (Donate $ msgGen 1)+ unless ok (error "cast failed!")+ producer (workLeft - 1)+ let consumer 0 = pure ()+ consumer workLeft =+ let handler =+ receive bookStoreOutput+ >>= maybe+ (return Nothing)+ ( \case+ NonBlocking _actual -> do+ return (Just Nothing)+ a -> do+ liftIO $ putStrLn "blocking case called"+ pure (Just (Just ("did not expect message: " <> show a)))+ )+ in handler+ >>= maybe+ (error "HandleMessage failed!")+ ( maybe+ (consumer (workLeft - 1))+ error+ )+ let perProducerWork = nMTotal `div` nP -- books to donate per producer+ perConsumerWork = nMTotal `div` nC -- books to receive per consumer+ let consumers = stimes nC (conc $ consumer perConsumerWork)+ let producers = stimes nP (conc $ producer perProducerWork)+ runConc (producers <> consumers)++castsAndCalls ::+ (MonadUnliftIO m, IsMessageBoxFactory cfg) =>+ (Int -> Book) ->+ cfg ->+ ((Int, Int), Int, (Int, Int)) ->+ m ()+castsAndCalls+ !msgGen+ !impl+ ( (!nDonors, !nDonationsPerStore),+ !nStores,+ (!nCustomers, !nRequestsPerStore)+ ) = do+ freshCounter <- newCounterVar+ flip runReaderT (MkBookStoreEnv {_fresh = freshCounter}) $ do+ let -- donate nDonationsPerStore books to all bookStores+ donor !bookStores !producerId =+ let books !storeId =+ msgGen+ . (storeId +)+ . (nStores *)+ . ((nDonationsPerStore * producerId) +)+ <$> [0 .. nDonationsPerStore - 1]+ donateIt (!storeId, !bookStoreInput) =+ traverse_+ ( \ !b ->+ do+ ok <- cast bookStoreInput (Donate b)+ unless ok (error "cast failed!")+ )+ (books storeId)+ in conc (traverse_ donateIt (zip [0 ..] bookStores))+ let -- ask all bookstores nRequestsPerStore times for their books+ customer !stores = conc (go nRequestsPerStore)+ where+ go 0 = return ()+ go !workLeft =+ let getBooks !store =+ call store GetBooks 5_000_000+ >>= either+ (error . ("get books failed: " ++) . show)+ (const (pure ()))+ in traverse_ getBooks stores+ >> go (workLeft - 1)++ let -- handle nMessagesToHandle requests+ bookStore nMessagesToHandle = do+ bIn <- newMessageBox impl+ bOut <- newInput bIn+ return (bOut, conc (go bIn nMessagesToHandle []))+ where+ go _bIn 0 _myBooks = pure ()+ go !inBox !workLeft !myBooks =+ let handler =+ handleMessage inBox $+ \case+ NonBlocking (Donate !b) ->+ pure (workLeft - 1, b : myBooks)+ Blocking GetBooks replyBox -> do+ replyTo replyBox myBooks+ pure (workLeft - 1, myBooks)+ in handler+ >>= maybe+ (error "HandleMessage failed!")+ (uncurry (go inBox))+ let nMessagesPerStore =+ nDonationsPerStore * nDonors+ + nRequestsPerStore * nCustomers+ (bookStoresInputes, bookStoresConc) <-+ unzip <$> replicateM nStores (bookStore nMessagesPerStore)+ let customers = stimes nCustomers (customer bookStoresInputes)+ let donors = mconcat (donor bookStoresInputes <$> [0 .. nDonors - 1])+ runConc (donors <> customers <> mconcat bookStoresConc)++benchmark :: IsMessageBoxFactory cfg => cfg -> Benchmark+benchmark cfg =+ bgroup+ "BookStore"+ [ bgroup+ "cast"+ [ bench+ ( show noMessages+ <> " "+ <> show senderNo+ <> " >>= "+ <> show receiverNo+ )+ ( nfAppIO+ (onlyCasts mkExampleBook cfg)+ (senderNo, noMessages, receiverNo)+ )+ | noMessages <- [100_000],+ (senderNo, receiverNo) <-+ [ -- (1, 1000),+ (1, 1),+ (10, 10),+ (1000, 1)+ ]+ ],+ bgroup+ "call"+ [ bench+ ( " donors: "+ <> show nDonors+ <> " stores: "+ <> show nStores+ <> " customers: "+ <> show nCustomers+ <> " total donations: "+ <> show (nDonors * nDonationsPerStore * nStores)+ <> " total GetBooks: "+ <> show (nCustomers * nGetBooksPerStore * nStores)+ )+ ( nfAppIO+ (castsAndCalls mkExampleBook cfg)+ ((nDonors, nDonationsPerStore), nStores, (nCustomers, nGetBooksPerStore))+ )+ | ((nDonors, nDonationsPerStore), nStores, (nCustomers, nGetBooksPerStore)) <-+ [ -- ((1, 1), 1, (1, 1)),+ ((500, 10), 10, (500, 10))+ ]+ ]+ ]
+ src-benchmark/CommandBenchmark.hs view
@@ -0,0 +1,59 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE EmptyDataDecls #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++module CommandBenchmark (benchmark) where++import qualified BookStoreBenchmark+import Criterion.Types+ ( Benchmark,+ bgroup,+ )+import qualified MediaBenchmark+import UnliftIO.MessageBox.CatchAll+ ( CatchAllFactory (CatchAllFactory),+ )+import UnliftIO.MessageBox.Class+ ( IsMessageBoxFactory (..),+ )+import qualified UnliftIO.MessageBox.Limited as L+import qualified UnliftIO.MessageBox.Unlimited as U++benchmark =+ bgroup+ "Command"+ ( foldMap+ go+ [ SomeBench MediaBenchmark.benchmark,+ SomeBench BookStoreBenchmark.benchmark+ ]+ )+ where+ go :: SomeBench -> [Benchmark]+ go (SomeBench b) =+ [ (\x -> bgroup "Unlimited" [b x]) U.BlockingUnlimited,+ (\x -> bgroup "CatchUnlimited" [b x]) (CatchAllFactory U.BlockingUnlimited),+ -- (\x -> bgroup (show x) [b x]) (L.BlockingBoxLimit L.MessageLimit_256),+ -- (\x -> bgroup (show x) [b x]) (L.WaitingBoxLimit Nothing 5_000_000 L.MessageLimit_256),+ (\x -> bgroup "Waiting256" [b x]) (L.WaitingBoxLimit (Just 60_000_000) 5_000_000 L.MessageLimit_256)+ ]++newtype SomeBench = SomeBench+ {_fromSomeBench :: forall cfg. (Show cfg, IsMessageBoxFactory cfg) => (cfg -> Benchmark)}
+ src-benchmark/Main.hs view
@@ -0,0 +1,154 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE EmptyDataDecls #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++module Main (main) where++import qualified CommandBenchmark+import Control.Monad (replicateM, unless)+import Criterion.Main (defaultMain)+import Criterion.Types+ ( bench,+ bgroup,+ nfAppIO,+ )+import Data.Semigroup (Semigroup (stimes))+import UnliftIO.MessageBox.CatchAll+ ( CatchAllFactory (..),+ )+import UnliftIO.MessageBox.Class+ ( IsInput (..),+ IsMessageBox (..),+ IsMessageBoxFactory (..),+ deliver,+ newInput,+ receive,+ )+import qualified UnliftIO.MessageBox.Limited as L+import qualified UnliftIO.MessageBox.Unlimited as U+import UnliftIO (MonadUnliftIO, conc, runConc)++main =+ defaultMain+ [ CommandBenchmark.benchmark,+ bgroup+ "Messaging"+ [ bench+ ( mboxImplTitle <> " "+ <> show noMessages+ <> " "+ <> show senderNo+ <> " : "+ <> show receiverNo+ )+ ( nfAppIO+ impl+ (senderNo, noMessages, receiverNo)+ )+ | noMessages <- [100_000],+ (isNonBlocking, mboxImplTitle, impl) <-+ [ let x = U.BlockingUnlimited+ in (False, "Unlimited", unidirectionalMessagePassing mkTestMessage x),+ let x = CatchAllFactory U.BlockingUnlimited+ in (False, "CatchUnlimited", unidirectionalMessagePassing mkTestMessage x),+ -- let x = L.BlockingBoxLimit L.MessageLimit_1+ -- in (False, show x, unidirectionalMessagePassing mkTestMessage x),+ -- let x = L.BlockingBoxLimit L.MessageLimit_16+ -- in (False, show x, unidirectionalMessagePassing mkTestMessage x),+ -- let x = L.BlockingBoxLimit L.MessageLimit_32+ -- in (False, show x, unidirectionalMessagePassing mkTestMessage x),+ -- let x = L.BlockingBoxLimit L.MessageLimit_64+ -- in (False, show x, unidirectionalMessagePassing mkTestMessage x),+ -- let x = L.BlockingBoxLimit L.MessageLimit_128+ -- in (False, show x, unidirectionalMessagePassing mkTestMessage x),+ let x = L.BlockingBoxLimit L.MessageLimit_256+ in (False, "Blocking256", unidirectionalMessagePassing mkTestMessage x),+ -- ,+ -- let x = L.BlockingBoxLimit L.MessageLimit_512+ -- in (False, show x, unidirectionalMessagePassing mkTestMessage x),+ -- let x = L.BlockingBoxLimit L.MessageLimit_4096+ -- in (False, show x, unidirectionalMessagePassing mkTestMessage x),+ let x = L.NonBlockingBoxLimit L.MessageLimit_128+ in (True, show x, unidirectionalMessagePassing mkTestMessage x),+ -- let x = L.WaitingBoxLimit Nothing 5_000_000 L.MessageLimit_128+ -- in (False, show x, unidirectionalMessagePassing mkTestMessage x),+ let x = L.WaitingBoxLimit (Just 60_000_000) 5_000_000 L.MessageLimit_256+ in (True, "Waiting256", unidirectionalMessagePassing mkTestMessage x)+ -- let x = CatchAllFactory (L.BlockingBoxLimit L.MessageLimit_128)+ -- in (False, show x, unidirectionalMessagePassing mkTestMessage x)+ ],+ (senderNo, receiverNo) <-+ [ -- (1, 1000),+ (10, 100),+ (1, 1),+ (1000, 1)+ ],+ not isNonBlocking || senderNo == 1 && receiverNo > 1+ ]+ ]++mkTestMessage :: Int -> TestMessage+mkTestMessage !i =+ MkTestMessage+ ( "The not so very very very very very very very very very very very very very very very very very very very very very very very very " ++ show i,+ "large",+ "meeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeessssssssssssssssssssssssssssssssss" ++ show i,+ ( "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",+ "ggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee",+ even i,+ 123423421111111111111111111123234 * toInteger i+ )+ )++newtype TestMessage = MkTestMessage (String, String, String, (String, String, Bool, Integer))+ deriving newtype (Show)++unidirectionalMessagePassing ::+ (MonadUnliftIO m, IsMessageBoxFactory cfg) =>+ (Int -> TestMessage) ->+ cfg ->+ (Int, Int, Int) ->+ m ()+unidirectionalMessagePassing !msgGen !impl (!nP, !nM, !nC) = do+ (ccs, cs) <- consumers+ let ps = producers cs+ runConc (ps <> ccs)+ where+ producers !cs = stimes nP (conc producer)+ where+ producer =+ mapM_+ ( \(!msg, !cons) -> do+ !ok <- deliver cons msg+ unless ok (error ("producer failed to deliver: " <> show msg))+ )+ ((,) <$> (msgGen <$> [0 .. (nM `div` (nC * nP)) - 1]) <*> cs)+ consumers = do+ cis <- replicateM nC (newMessageBox impl)+ let ccs = foldMap (conc . consume (nM `div` nC)) cis+ cs <- traverse newInput cis+ return (ccs, cs)+ where+ consume 0 _inBox = return ()+ consume workLeft inBox = do+ receive inBox+ >>= maybe+ (error "consumer failed to receive")+ (const (consume (workLeft - 1) inBox))
+ src-benchmark/MediaBenchmark.hs view
@@ -0,0 +1,618 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE EmptyDataDecls #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE Strict #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++-- | More complex benchmarks, mimiking real world applications+-- from the media domain.+module MediaBenchmark (benchmark) where++import Control.Monad.Reader+ ( MonadIO (liftIO),+ ReaderT (runReaderT),+ asks,+ fix,+ forM,+ forM_,+ replicateM_,+ unless,+ void,+ )+import Criterion.Types+ ( Benchmark,+ bench,+ bgroup,+ nfAppIO,+ )+import Data.Map (Map)+import qualified Data.Map.Strict as Map+import Data.Set (Set)+import qualified Data.Set as Set+import GHC.Stack (HasCallStack)+import UnliftIO.MessageBox.Command as Command+ ( Command,+ Message (..),+ ReturnType (FireAndForget, Return),+ call,+ cast,+ delegateCall,+ replyTo,+ )+import UnliftIO.MessageBox.Util.CallId as CallId+ ( CallId,+ HasCallIdCounter (..),+ )+import UnliftIO.MessageBox.Util.Fresh+ ( CounterVar,+ HasCounterVar (getCounterVar),+ newCounterVar,+ )+import UnliftIO.MessageBox.Class+ ( IsInput (..),+ IsMessageBox (..),+ IsMessageBoxFactory (..),+ handleMessage,+ newInput,+ receiveAfter,+ )+import UnliftIO+ ( Conc,+ SomeException,+ conc,+ runConc,+ try,+ )+import UnliftIO.Concurrent (forkIO)++benchmark :: (IsMessageBoxFactory cfg) => cfg -> Benchmark+benchmark cfg =+ bgroup+ "Media"+ $ [ bench ("Mixing/" ++ show p) (nfAppIO (mediaAppBenchmark cfg) p)+ | p <-+ [ -- Param 100 500 4,+ Param 100 1000 4+ ]+ ]++-- | A benchmark of a fictional media mixing application.+--+-- The applicaton has these layers:+--+-- 1. Media resources (DSPs) for mixing media streams (aka members)+-- * One MediaApi process with @i@ Dsps+-- * Exits after a stop message is received+-- and all media resources are removed.+--+-- 2. Logical Mixing Groups+-- * one process per group+-- * use the media resources to mix groups+-- spanning several DSPs+-- * create bridges between low-level mixers+--+-- 3. Mixing Group Broker+-- * Starts/Stops the processes for the+-- logical mixing groups.+-- * Delegates the MixingApi to the+-- logical mixing group processes+-- * When all groups are destroyed:+-- - Send the media resources process a stop message+-- - exit+--+-- 4. Mixing Group Applications (aka clients)+-- * uses the Mixing Group Broker+-- * corresponds to a single exclusive mixing group+-- * Follows this procedure:+-- - create the mixing group+-- - add members+-- - remove members+-- - destroy the group+-- - exit+--+-- This whole process is repeated @m@ times.+--+-- When all clients have finished, send a shutdown message to the media process,+-- and wait for every process to finish.+mediaAppBenchmark ::+ forall cfg.+ (HasCallStack, IsMessageBoxFactory cfg) =>+ cfg ->+ Param ->+ IO ()+mediaAppBenchmark cfg param = do+ (mixingOut, c1) <- do+ (mediaInput, mediaConc) <- spawnMediaApi+ (mixingOut, mixingConc) <- spawnMixingBroker mediaInput+ return (mixingOut, mediaConc <> mixingConc)+ appCounters <- AppCounters <$> newCounterVar <*> newCounterVar+ let clients = spawnMixingApps mixingOut+ runReaderT (runConc (c1 <> clients)) appCounters+ where+ spawnMediaApi = do+ mediaOutput <- newMessageBox cfg+ mediaInput <- newInput mediaOutput+ let startMediaServer =+ void $ mediaServerLoop (mkMediaSimSt (toDspConfig param))+ mediaServerLoop st = do+ let !isFinished =+ shuttingDown st && Map.null (allMixers st)+ unless isFinished $+ try+ (handleMessage mediaOutput (handleMediaApi st))+ >>= either+ ( liftIO+ . putStrLn+ . ("media server failed to receive next message: " ++)+ . (show :: SomeException -> String)+ )+ (maybe (error "media server loop premature exit") mediaServerLoop)+ return (mediaInput, conc startMediaServer)++ spawnMixingBroker ::+ Input (MessageBox cfg) (Message MediaApi) ->+ IO (Input (MessageBox cfg) (Message MixingApi), Conc (ReaderT AppCounters IO) ())+ spawnMixingBroker mediaBoxOut = do+ mixingOutput <- newMessageBox cfg+ mixingInput <- newInput mixingOutput+ let startMixingServer =+ let groupMap :: Map MixingGroupId (Input (MessageBox cfg) (Message MixingApi))+ groupMap = Map.empty+ in mixingServerLoop (0, groupMap)+ mixingServerLoop !groupMap =+ try+ ( handleMessage+ mixingOutput+ (dispatchMixingApi groupMap)+ )+ >>= either+ ( liftIO+ . putStrLn+ . ("mixingServerLoop failed: " ++)+ . (show :: SomeException -> String)+ )+ ( maybe+ (error "mixing server loop premature exit")+ ( \(!nDestroyed', !groupMap') ->+ if Map.null groupMap' && nDestroyed' == nGroups param+ then void $ cast mediaBoxOut MediaShutdown+ else mixingServerLoop (nDestroyed', groupMap')+ )+ )+ dispatchMixingApi ::+ (Int, Map MixingGroupId (Input (MessageBox cfg) (Message MixingApi))) ->+ Message MixingApi ->+ ReaderT+ AppCounters+ IO+ (Int, Map MixingGroupId (Input (MessageBox cfg) (Message MixingApi)))+ dispatchMixingApi (!nDestroyed, !st) =+ \case+ Blocking cm@(CreateMixingGroup !mgId) !r -> do+ unless+ (Map.notMember mgId st)+ (error ("Mixing group ID conflict: " ++ show mgId))+ !mgInput <- spawnMixingGroup mediaBoxOut+ !ok <- delegateCall mgInput cm r+ unless ok (error ("delegation failed: " ++ show cm))+ return (nDestroyed, Map.insert mgId mgInput st)+ Blocking (DestroyMixingGroup !mgId) !r ->+ case Map.lookup mgId st of+ Nothing ->+ error ("DestroyMixingGroup: Mixing group doesn't exist: " ++ show mgId)+ Just !mgInput -> do+ !ok <- delegateCall mgInput (DestroyMixingGroup mgId) r+ unless ok (error ("delegation failed: " ++ show (DestroyMixingGroup mgId)))+ return (nDestroyed + 1, Map.delete mgId st)+ NonBlocking m@(Join !mgId _ _) ->+ case Map.lookup mgId st of+ Nothing ->+ error ("Mixing group doesn't exist: " ++ show mgId ++ " in: " ++ show m)+ Just !mgInput -> do+ !ok <- cast mgInput m+ unless ok (error ("delegation failed: " ++ show m))+ return (nDestroyed, st)+ NonBlocking m@(UnJoin !mgId _ _) ->+ case Map.lookup mgId st of+ Nothing ->+ error ("Mixing group doesn't exist: " ++ show mgId ++ " in: " ++ show m)+ Just !mgInput -> do+ !ok <- cast mgInput m+ unless ok (error ("delegation failed: " ++ show m))+ return (nDestroyed, st)+ return (mixingInput, conc startMixingServer)+ spawnMixingGroup ::+ Input (MessageBox cfg) (Message MediaApi) ->+ ReaderT AppCounters IO (Input (MessageBox cfg) (Message MixingApi))+ spawnMixingGroup !mediaInput = do+ !mgOutput <- newMessageBox cfg+ !mgInput <- newInput mgOutput+ let mgLoop (!mgId, !groupMap) =+ try+ ( handleMessage+ mgOutput+ (handleCmd (mgId, groupMap))+ )+ >>= either+ ( liftIO+ . putStrLn+ . (("mixingGroup " ++ show mgId ++ " exception: ") ++)+ . (show :: SomeException -> String)+ )+ ( maybe+ (error ("mixingGroup loop " ++ show mgId ++ " premature exit"))+ (maybe (return ()) mgLoop)+ )+ handleCmd ::+ (MixingGroupId, Map DspId (MixerId, Set MemberId)) ->+ Message MixingApi ->+ ReaderT AppCounters IO (Maybe (MixingGroupId, Map DspId (MixerId, Set MemberId)))+ handleCmd (!mgId, !st) =+ \case+ Blocking (CreateMixingGroup !mgId') !r -> do+ replyTo r ()+ return (Just (mgId', st))+ Blocking (DestroyMixingGroup _) !r -> do+ replyTo r ()+ return Nothing -- exit+ NonBlocking (Join !_mgId !memberId !callBack) ->+ call mediaInput FetchDsps 500_000+ >>= either+ ( \ !mErr -> do+ void $ deliver callBack (MemberUnJoined mgId memberId)+ error (show mErr)+ )+ ( \ !dsps ->+ let selectedDspId =+ let !ks = Set.toList dsps+ !nDsps = length ks+ !ki = memberId `rem` nDsps+ in ks !! ki+ doAdd !theMixerId !theMembers =+ let !m = AddToMixer theMixerId memberId+ in call mediaInput m 500_000+ >>= \case+ Left !err ->+ error (show m ++ " failed: " ++ show err)+ Right False -> do+ void $ deliver callBack (MemberUnJoined mgId memberId)+ error (show m ++ " did not succeed")+ Right True -> do+ let st' =+ Map.insert+ selectedDspId+ (theMixerId, Set.insert memberId theMembers)+ st+ void $ deliver callBack (MemberJoined mgId memberId)+ return (Just (mgId, st'))+ in if Set.null dsps+ then error "Not enough DSP capacity"+ else case Map.lookup selectedDspId st of+ Nothing ->+ call mediaInput (CreateMixer selectedDspId) 500_000+ >>= \case+ Left !err ->+ error (show err)+ Right Nothing ->+ error ("create mixer failed on: " ++ show selectedDspId)+ Right (Just !theMixerId) ->+ doAdd theMixerId Set.empty+ Just (!mixerId, !members) -> doAdd mixerId members+ )+ NonBlocking (UnJoin _ !memberId !callBack) ->+ case Map.toList (Map.filter (Set.member memberId . snd) st) of+ ((!theDspId, (!theMixerId, !theMembers)) : _) -> do+ call mediaInput (RemoveFromMixer theMixerId memberId) 500_000+ >>= either (error . show) (const (return ()))+ let theMembers' = Set.delete memberId theMembers+ if Set.null theMembers'+ then do+ !ok <- cast mediaInput (DestroyMixer theMixerId)+ unless ok (error (show (DestroyMixer theMixerId) ++ " failed!"))+ void $ deliver callBack (MemberUnJoined mgId memberId)+ return (Just (mgId, Map.delete theDspId st))+ else do+ void $ deliver callBack (MemberUnJoined mgId memberId)+ return (Just (mgId, Map.insert theDspId (theMixerId, theMembers') st))+ [] ->+ return (Just (mgId, st))+ void $ forkIO (mgLoop (-1, Map.empty))+ return mgInput++ spawnMixingApps mixingInput =+ let !clients = foldMap spawnClient [0 .. nGroups param - 1]++ spawnClient !mixingGroupId = conc $ do+ eventsIn <- newMessageBox cfg+ eventsOut <- newInput eventsIn+ -- create conference,+ call mixingInput (CreateMixingGroup mixingGroupId) 50_000_000+ >>= either+ (error . ((show (CreateMixingGroup mixingGroupId) ++ " failed: ") ++) . show)+ (const (return ()))+ -- add participants and wait for the joined event+ !members <- forM [0 .. nMembers param - 1] $ \i -> do+ let !memberId = nMembers param * mixingGroupId + i+ !castSuccessful <- cast mixingInput (Join mixingGroupId memberId eventsOut)+ unless+ castSuccessful+ (error ("Failed to cast: " ++ show (Join mixingGroupId memberId eventsOut)))+ return memberId+ replicateM_+ (nMembers param)+ ( fix $ \ ~again ->+ receive eventsIn+ >>= \case+ Just (MemberJoined _ _) ->+ return ()+ Just unexpected ->+ error ("Unexpected mixing group event: " ++ show unexpected ++ " expected MemberJoined")+ Nothing ->+ again+ )+ -- remove participants and wait for unjoined+ forM_ members $ \ !memberId -> do+ !castSuccessful <- cast mixingInput (UnJoin mixingGroupId memberId eventsOut)+ unless+ castSuccessful+ (error ("Failed to cast: " ++ show (UnJoin mixingGroupId memberId eventsOut)))+ replicateM_+ (nMembers param)+ ( fix $ \ ~again ->+ receiveAfter eventsIn 500_000+ >>= \case+ Just (MemberUnJoined _ _) ->+ return ()+ Just unexpected ->+ error ("Unexpected mixing group event: " ++ show unexpected ++ " expected MemberUnJoined")+ Nothing ->+ again+ )+ -- destroy the conference,+ call mixingInput (DestroyMixingGroup mixingGroupId) 500_000+ >>= either+ (error . ((show (DestroyMixingGroup mixingGroupId) ++ " failed: ") ++) . show)+ (const (return ()))+ in clients++data AppCounters = AppCounters+ { callIdCounter :: !(CounterVar CallId),+ idCounter :: !(CounterVar Int)+ }++instance CallId.HasCallIdCounter AppCounters where+ getCallIdCounter = asks callIdCounter++instance HasCounterVar Int AppCounters where+ getCounterVar = asks idCounter++data Param = Param+ { nDsps :: !Int,+ nGroups :: !Int,+ nMembers :: !Int+ }++instance Show Param where+ showsPrec _ Param {nDsps, nGroups, nMembers} =+ shows nDsps . showString "DSPs/"+ . shows nGroups+ . showString "GRPs/"+ . shows nMembers+ . showString "MEMBERS"++toDspConfig :: Param -> Map DspId Capacity+toDspConfig p =+ let perDspCapacity =+ max 2 (2 * requiredTotal) -- (2 * (ceiling (fromIntegral requiredTotal / fromIntegral (nDsps p) :: Double)))+ requiredTotal = nGroups p + nGroups p * nMembers p+ in Map.fromList ([0 ..] `zip` replicate (nDsps p) perDspCapacity)++-- * Types for the domain of the benchmarks in this module++data MediaApi++data MixingApi++type DspId = Int++type MixerId = Int++type MemberId = Int++type MixingGroupId = Int++data instance Command MediaApi _ where+ FetchDsps :: Command MediaApi ('Return (Set DspId))+ CreateMixer :: DspId -> Command MediaApi ('Return (Maybe MixerId))+ DestroyMixer :: MixerId -> Command MediaApi 'FireAndForget+ AddToMixer :: MixerId -> MemberId -> Command MediaApi ('Return Bool)+ RemoveFromMixer :: MixerId -> MemberId -> Command MediaApi ('Return ())+ MediaShutdown :: Command MediaApi 'FireAndForget++deriving stock instance Show (Command MediaApi ('Return (Set DspId)))++deriving stock instance Show (Command MediaApi ('Return (Maybe MixerId)))++deriving stock instance Show (Command MediaApi ('Return Bool))++deriving stock instance Show (Command MediaApi ('Return ()))++deriving stock instance Show (Command MediaApi 'FireAndForget)++data MixingGroupEvent where+ MemberJoined :: MixingGroupId -> MemberId -> MixingGroupEvent+ MemberUnJoined :: MixingGroupId -> MemberId -> MixingGroupEvent+ deriving stock (Show, Eq)++data instance Command MixingApi _ where+ CreateMixingGroup ::+ MixingGroupId ->+ Command MixingApi ('Return ())+ DestroyMixingGroup ::+ MixingGroupId ->+ Command MixingApi ('Return ())+ Join ::+ IsInput outBox =>+ MixingGroupId ->+ MemberId ->+ outBox MixingGroupEvent ->+ Command MixingApi 'FireAndForget+ UnJoin ::+ IsInput outBox =>+ MixingGroupId ->+ MemberId ->+ outBox MixingGroupEvent ->+ Command MixingApi 'FireAndForget++instance Show (Command MixingApi ('Return ())) where+ showsPrec d (CreateMixingGroup i) =+ showParen (d >= 9) (showString "CreateMixingGroup " . shows i)+ showsPrec d (DestroyMixingGroup i) =+ showParen (d >= 9) (showString "DestroyMixingGroup " . shows i)++instance Show (Command MixingApi 'FireAndForget) where+ showsPrec d (Join i j _) =+ showParen (d >= 9) (showString "Join " . shows i . showChar ' ' . shows j)+ showsPrec d (UnJoin i j _) =+ showParen (d >= 9) (showString "UnJoin " . shows i . showChar ' ' . shows j)++type Capacity = Int++-- | Mix media streams using 'MediaApi'++-- | Media server simulation+--+-- Handle 'MediaApi' requests and manage a map of+-- dsps and mixers.+handleMediaApi ::+ MediaSimSt ->+ Message MediaApi ->+ ReaderT env IO MediaSimSt+handleMediaApi !st =+ \case+ Blocking FetchDsps replyBox -> do+ let !goodDsps = Map.keysSet (Map.filter (> 1) (allDsps st))+ replyTo replyBox goodDsps+ return st+ Blocking (CreateMixer dspId) replyBox ->+ case Map.lookup dspId (allDsps st) of+ Just capacity | capacity > 0 -> do+ let theNewMixerId = nextMixerId st+ replyTo replyBox (Just theNewMixerId)+ return+ ( st+ { allDsps =+ Map.insert+ dspId+ (capacity - 1)+ (allDsps st),+ allMixers =+ Map.insert+ theNewMixerId+ (dspId, Set.empty)+ (allMixers st),+ nextMixerId = theNewMixerId + 1+ }+ )+ _ -> do+ replyTo replyBox Nothing+ return st+ Blocking (AddToMixer theMixer newMember) replyBox ->+ case Map.lookup theMixer (allMixers st) of+ Nothing ->+ replyTo replyBox False >> return st+ Just (!dsp, !streams) ->+ if Set.member newMember streams+ then replyTo replyBox True >> return st+ else case Map.lookup dsp (allDsps st) of+ Just capacity+ | capacity >= 1 ->+ replyTo replyBox True+ >> return+ ( st+ { allMixers =+ Map.insert+ theMixer+ (dsp, Set.insert newMember streams)+ (allMixers st),+ allDsps =+ Map.insert+ dsp+ (capacity - 1)+ (allDsps st)+ }+ )+ _ ->+ replyTo replyBox False >> return st+ Blocking (RemoveFromMixer theMixer theMember) replyBox -> do+ replyTo replyBox ()+ case Map.lookup theMixer (allMixers st) of+ Just (theDsp, theMembers)+ | Set.member theMember theMembers ->+ return+ st+ { allDsps =+ Map.update+ (Just . (+ 1))+ theDsp+ (allDsps st),+ allMixers =+ Map.insert+ theMixer+ (theDsp, Set.delete theMember theMembers)+ (allMixers st)+ }+ _ ->+ return st+ NonBlocking (DestroyMixer theMixerId) ->+ let foundMixer (mixersDsp, mediaStreams) =+ st+ { allMixers =+ Map.delete theMixerId (allMixers st),+ allDsps =+ Map.update+ (Just . (+ (1 + Set.size mediaStreams)))+ mixersDsp+ (allDsps st)+ }+ in pure $+ maybe+ st+ foundMixer+ (Map.lookup theMixerId (allMixers st))+ NonBlocking MediaShutdown ->+ pure st {shuttingDown = True}++data MediaSimSt = MediaSimSt+ { allDsps :: !(Map DspId Capacity),+ nextMixerId :: !MixerId,+ allMixers :: !(Map MixerId (DspId, Set MemberId)),+ shuttingDown :: !Bool+ }++mkMediaSimSt :: Map DspId Capacity -> MediaSimSt+mkMediaSimSt x =+ MediaSimSt+ { allDsps = x,+ nextMixerId = 0,+ allMixers = Map.empty,+ shuttingDown = False+ }
+ src-memleak-test/Main.hs view
@@ -0,0 +1,39 @@+module Main (main) where++import Control.Monad (forM_, when)+import MediaBenchmark+ ( Param(Param, nDsps, nGroups, nMembers, nRounds),+ mediaAppBenchmark )+import System.Environment (getArgs)+import UnliftIO.MessageBox+ ( BlockingUnlimited (BlockingUnlimited),+ )++main :: IO ()+main = do+ args <- getArgs+ let (rounds, repetitions) =+ case args of+ [n', r'] -> (read n', read r')+ [n'] -> (read n', 1)+ _ -> (1, 1)+ forM_ [1 .. repetitions :: Int] $ \rep -> do+ putStrLn ""+ putStrLn ("================= BEGIN (rep: " ++ show rep ++ ") ================")+ putStrLn ""+ let p =+ Param+ { nDsps = 100,+ nGroups = 1000,+ nMembers = 4,+ nRounds = rounds+ }+ print p+ putStrLn ""+ mediaAppBenchmark BlockingUnlimited p+ -- mediaAppBenchmark (WaitingBoxLimit (Just (24 * 3600 * 1000000)) 10000000 MessageLimit_256) p+ putStrLn ""+ putStrLn ("================= DONE (rep: " ++ show rep ++ ") ================")+ when (rep < repetitions) $ do+ putStrLn ""+ putStrLn "(Sleeping...)"
+ src-memleak-test/MediaBenchmark.hs view
@@ -0,0 +1,646 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE EmptyDataDecls #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE Strict #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++-- | More complex benchmarks, mimiking real world applications+-- from the media domain.+module MediaBenchmark (mediaAppBenchmark, Param (..)) where++import Control.Monad (when)+import Control.Monad.Reader+ ( MonadIO (liftIO),+ ReaderT (runReaderT),+ asks,+ fix,+ forM,+ forM_,+ replicateM_,+ unless,+ void,+ )+import Data.Map (Map)+import qualified Data.Map.Strict as Map+import Data.Set (Set)+import qualified Data.Set as Set+import GHC.Stack (HasCallStack)+import GHC.Stats+import UnliftIO+ ( Conc,+ SomeException,+ TVar,+ atomically,+ checkSTM,+ conc,+ newTVarIO,+ readTVar,+ runConc,+ try,+ writeTVar,+ )+import UnliftIO.Concurrent (forkIO, threadDelay)+import UnliftIO.MessageBox++-- | A benchmark of a fictional media mixing application.+--+-- The applicaton has these layers:+--+-- 1. Media resources (DSPs) for mixing media streams (aka members)+-- * One MediaApi process with @i@ Dsps+-- * Exits after a stop message is received+-- and all media resources are removed.+--+-- 2. Logical Mixing Groups+-- * one process per group+-- * use the media resources to mix groups+-- spanning several DSPs+-- * create bridges between low-level mixers+--+-- 3. Mixing Group Broker+-- * Starts/Stops the processes for the+-- logical mixing groups.+-- * Delegates the MixingApi to the+-- logical mixing group processes+-- * When all groups are destroyed:+-- - Send the media resources process a stop message+-- - exit+--+-- 4. Mixing Group Applications (aka clients)+-- * uses the Mixing Group Broker+-- * corresponds to a single exclusive mixing group+-- * Follows this procedure:+-- - create the mixing group+-- - add members+-- - remove members+-- - destroy the group+-- - exit+--+-- This whole process is repeated @m@ times.+--+-- When all clients have finished, send a shutdown message to the media process,+-- and wait for every process to finish.+mediaAppBenchmark ::+ forall cfg.+ (HasCallStack, IsMessageBoxFactory cfg) =>+ cfg ->+ Param ->+ IO ()+mediaAppBenchmark cfg param = do+ currentIterationVar <- newTVarIO 0+ (mixingOut, c1) <- do+ (mediaInput, mediaConc) <- spawnMediaApi+ (mixingOut, mixingConc) <- spawnMixingBroker currentIterationVar mediaInput+ return (mixingOut, mediaConc <> mixingConc)+ appCounters <- AppCounters <$> newCounterVar <*> newCounterVar+ let clients = spawnMixingApps currentIterationVar mixingOut+ runReaderT (runConc (c1 <> clients)) appCounters+ where+ spawnMediaApi = do+ mediaOutput <- newMessageBox cfg+ mediaInput <- newInput mediaOutput+ let startMediaServer = do+ liftIO $ putStrLn "===== BEGIN: Media Server Loop"+ forM_ [1 .. nRounds param] $ \iteration -> do+ liftIO $ putStrLn ("===== ITERATION: " ++ show iteration ++ ": Media Server Loop ")+ mediaServerLoop (mkMediaSimSt (toDspConfig param))+ liftIO $ do+ hasStats <- getRTSStatsEnabled+ when hasStats $ do+ rtsStats <- getRTSStats+ putStrLn+ ( "Total memory in use: "+ ++ show (gcdetails_mem_in_use_bytes (gc rtsStats) `div` (1024 * 1024))+ ++ "m"+ )+ putStrLn ("===== FINISHED: " ++ show iteration ++ ": Media Server Loop")+ liftIO $ putStrLn "===== END: Media Server Loop"+ mediaServerLoop st = do+ let !isFinished =+ shuttingDown st && Map.null (allMixers st)+ unless isFinished $+ try+ (handleMessage mediaOutput (handleMediaApi st))+ >>= either+ ( liftIO+ . putStrLn+ . ("media server failed to receive next message: " ++)+ . (show :: SomeException -> String)+ )+ (maybe (error "media server loop premature exit") mediaServerLoop)+ return (mediaInput, conc startMediaServer)++ spawnMixingBroker ::+ TVar Int ->+ Input (MessageBox cfg) (Message MediaApi) ->+ IO (Input (MessageBox cfg) (Message MixingApi), Conc (ReaderT AppCounters IO) ())+ spawnMixingBroker currentIterationVar mediaBoxOut = do+ mixingOutput <- newMessageBox cfg+ mixingInput <- newInput mixingOutput+ let startMixingServer = do+ let groupMap :: Map MixingGroupId (Input (MessageBox cfg) (Message MixingApi))+ groupMap = Map.empty+ liftIO $ putStrLn "===== BEGIN: Mixing Server Loop"+ forM_ [1 .. nRounds param] $ \iteration -> do+ liftIO $ putStrLn ("===== ITERATION: " ++ show iteration ++ ": Mixing Server Loop ")+ atomically $ writeTVar currentIterationVar iteration+ mixingServerLoop iteration (0, groupMap)+ liftIO $ putStrLn ("===== FINISHED: " ++ show iteration ++ ": Mixing Server Loop")+ when (iteration < nRounds param) (threadDelay 500_000)+ liftIO $ putStrLn "===== END: Mixing Server Loop"++ mixingServerLoop !iteration !groupMap =+ try+ ( handleMessage+ mixingOutput+ (dispatchMixingApi groupMap)+ )+ >>= either+ ( liftIO+ . putStrLn+ . ("mixingServerLoop failed: " ++)+ . (show :: SomeException -> String)+ )+ ( maybe+ (error "mixing server loop premature exit")+ ( \(!nDestroyed', !groupMap') ->+ if Map.null groupMap' && nDestroyed' == nGroups param+ then void $ cast mediaBoxOut MediaShutdown+ else mixingServerLoop iteration (nDestroyed', groupMap')+ )+ )+ dispatchMixingApi ::+ (Int, Map MixingGroupId (Input (MessageBox cfg) (Message MixingApi))) ->+ Message MixingApi ->+ ReaderT+ AppCounters+ IO+ (Int, Map MixingGroupId (Input (MessageBox cfg) (Message MixingApi)))+ dispatchMixingApi (!nDestroyed, !st) =+ \case+ Blocking cm@(CreateMixingGroup !mgId) !r -> do+ unless+ (Map.notMember mgId st)+ (error ("Mixing group ID conflict: " ++ show mgId))+ !mgInput <- spawnMixingGroup mediaBoxOut+ !ok <- delegateCall mgInput cm r+ unless ok (error ("delegation failed: " ++ show cm))+ return (nDestroyed, Map.insert mgId mgInput st)+ Blocking (DestroyMixingGroup !mgId) !r ->+ case Map.lookup mgId st of+ Nothing ->+ error ("DestroyMixingGroup: Mixing group doesn't exist: " ++ show mgId)+ Just !mgInput -> do+ !ok <- delegateCall mgInput (DestroyMixingGroup mgId) r+ unless ok (error ("delegation failed: " ++ show (DestroyMixingGroup mgId)))+ return (nDestroyed + 1, Map.delete mgId st)+ NonBlocking m@(Join !mgId _ _) ->+ case Map.lookup mgId st of+ Nothing ->+ error ("Mixing group doesn't exist: " ++ show mgId ++ " in: " ++ show m)+ Just !mgInput -> do+ !ok <- cast mgInput m+ unless ok (error ("delegation failed: " ++ show m))+ return (nDestroyed, st)+ NonBlocking m@(UnJoin !mgId _ _) ->+ case Map.lookup mgId st of+ Nothing ->+ error ("Mixing group doesn't exist: " ++ show mgId ++ " in: " ++ show m)+ Just !mgInput -> do+ !ok <- cast mgInput m+ unless ok (error ("delegation failed: " ++ show m))+ return (nDestroyed, st)+ return (mixingInput, conc startMixingServer)+ spawnMixingGroup ::+ Input (MessageBox cfg) (Message MediaApi) ->+ ReaderT AppCounters IO (Input (MessageBox cfg) (Message MixingApi))+ spawnMixingGroup !mediaInput = do+ !mgOutput <- newMessageBox cfg+ !mgInput <- newInput mgOutput+ let mgLoop (!mgId, !groupMap) =+ try+ ( handleMessage+ mgOutput+ (handleCmd (mgId, groupMap))+ )+ >>= either+ ( liftIO+ . putStrLn+ . (("mixingGroup " ++ show mgId ++ " exception: ") ++)+ . (show :: SomeException -> String)+ )+ ( maybe+ (error ("mixingGroup loop " ++ show mgId ++ " premature exit"))+ (maybe (return ()) mgLoop)+ )+ handleCmd ::+ (MixingGroupId, Map DspId (MixerId, Set MemberId)) ->+ Message MixingApi ->+ ReaderT AppCounters IO (Maybe (MixingGroupId, Map DspId (MixerId, Set MemberId)))+ handleCmd (!mgId, !st) =+ \case+ Blocking (CreateMixingGroup !mgId') !r -> do+ replyTo r ()+ return (Just (mgId', st))+ Blocking (DestroyMixingGroup _) !r -> do+ replyTo r ()+ return Nothing -- exit+ NonBlocking (Join !_mgId !memberId !callBack) ->+ call mediaInput FetchDsps 500_000+ >>= either+ ( \ !mErr -> do+ void $ deliver callBack (MemberUnJoined mgId memberId)+ error (show mErr)+ )+ ( \ !dsps ->+ let selectedDspId =+ let !ks = Set.toList dsps+ !nDsps = length ks+ !ki = memberId `rem` nDsps+ in ks !! ki+ doAdd !theMixerId !theMembers =+ let !m = AddToMixer theMixerId memberId+ in call mediaInput m 500_000+ >>= \case+ Left !err ->+ error (show m ++ " failed: " ++ show err)+ Right False -> do+ void $ deliver callBack (MemberUnJoined mgId memberId)+ error (show m ++ " did not succeed")+ Right True -> do+ let st' =+ Map.insert+ selectedDspId+ (theMixerId, Set.insert memberId theMembers)+ st+ void $ deliver callBack (MemberJoined mgId memberId)+ return (Just (mgId, st'))+ in if Set.null dsps+ then error "Not enough DSP capacity"+ else case Map.lookup selectedDspId st of+ Nothing ->+ call mediaInput (CreateMixer selectedDspId) 500_000+ >>= \case+ Left !err ->+ error (show err)+ Right Nothing ->+ error ("create mixer failed on: " ++ show selectedDspId)+ Right (Just !theMixerId) ->+ doAdd theMixerId Set.empty+ Just (!mixerId, !members) -> doAdd mixerId members+ )+ NonBlocking (UnJoin _ !memberId !callBack) ->+ case Map.toList (Map.filter (Set.member memberId . snd) st) of+ ((!theDspId, (!theMixerId, !theMembers)) : _) -> do+ call mediaInput (RemoveFromMixer theMixerId memberId) 500_000+ >>= either (error . show) (const (return ()))+ let theMembers' = Set.delete memberId theMembers+ if Set.null theMembers'+ then do+ !ok <- cast mediaInput (DestroyMixer theMixerId)+ unless ok (error (show (DestroyMixer theMixerId) ++ " failed!"))+ void $ deliver callBack (MemberUnJoined mgId memberId)+ return (Just (mgId, Map.delete theDspId st))+ else do+ void $ deliver callBack (MemberUnJoined mgId memberId)+ return (Just (mgId, Map.insert theDspId (theMixerId, theMembers') st))+ [] ->+ return (Just (mgId, st))+ void $ forkIO (mgLoop (-1, Map.empty))+ return mgInput++ spawnMixingApps currentIterationVar mixingInput =+ let !clients = foldMap spawnClient [0 .. nGroups param - 1]+ spawnClient !mixingGroupId = conc $ do+ let isLogged = mixingGroupId == nGroups param - 1+ when+ isLogged+ (liftIO $ putStrLn ("Client: " ++ show mixingGroupId ++ " started."))+ forM_ [1 .. nRounds param] $ \iteration -> do+ atomically $ readTVar currentIterationVar >>= checkSTM . (== iteration)+ when+ isLogged+ (liftIO $ putStrLn ("Client: " ++ show mixingGroupId ++ " started iteration: " ++ show iteration))+ eventsIn <- newMessageBox cfg+ eventsOut <- newInput eventsIn+ -- create conference,+ call mixingInput (CreateMixingGroup mixingGroupId) 1_000_000+ >>= either+ (error . ((show (CreateMixingGroup mixingGroupId) ++ " failed: ") ++) . show)+ (const (return ()))+ -- add participants and wait for the joined event+ !members <- forM [0 .. nMembers param - 1] $ \i -> do+ let !memberId = nMembers param * mixingGroupId + i+ !castSuccessful <- cast mixingInput (Join mixingGroupId memberId eventsOut)+ unless+ castSuccessful+ (error ("Failed to cast: " ++ show (Join mixingGroupId memberId eventsOut)))+ return memberId+ replicateM_+ (nMembers param)+ ( fix $ \ ~again ->+ receive eventsIn+ >>= \case+ Just (MemberJoined _ _) ->+ return ()+ Just unexpected ->+ error ("Unexpected mixing group event: " ++ show unexpected ++ " expected MemberJoined")+ Nothing ->+ again+ )+ when+ isLogged+ (liftIO $ putStrLn ("Client: " ++ show mixingGroupId ++ " joined participants: " ++ show iteration))+ -- remove participants and wait for unjoined+ forM_ members $ \ !memberId -> do+ !castSuccessful <- cast mixingInput (UnJoin mixingGroupId memberId eventsOut)+ unless+ castSuccessful+ (error ("Failed to cast: " ++ show (UnJoin mixingGroupId memberId eventsOut)))+ replicateM_+ (nMembers param)+ ( fix $ \ ~again ->+ receiveAfter eventsIn 500_000+ >>= \case+ Just (MemberUnJoined _ _) ->+ return ()+ Just unexpected ->+ error ("Unexpected mixing group event: " ++ show unexpected ++ " expected MemberUnJoined")+ Nothing ->+ again+ )+ -- destroy the conference,+ call mixingInput (DestroyMixingGroup mixingGroupId) 500_000+ >>= either+ (error . ((show (DestroyMixingGroup mixingGroupId) ++ " failed: ") ++) . show)+ (const (return ()))++ when+ isLogged+ ( liftIO $+ putStrLn+ ( "Client: "+ ++ show mixingGroupId+ ++ " unjoined participants: "+ ++ show iteration+ )+ )+ when+ (iteration < nRounds param)+ ( do+ when+ isLogged+ ( liftIO $+ putStrLn+ ( "Client: "+ ++ show mixingGroupId+ ++ " finished with iteration: "+ ++ show iteration+ ++ " sleeping before next iteration."+ )+ )+ )+ in clients++data AppCounters = AppCounters+ { callIdCounter :: !(CounterVar CallId),+ idCounter :: !(CounterVar Int)+ }++instance HasCallIdCounter AppCounters where+ getCallIdCounter = asks callIdCounter++instance HasCounterVar Int AppCounters where+ getCounterVar = asks idCounter++data Param = Param+ { nDsps :: !Int,+ nGroups :: !Int,+ nMembers :: !Int,+ nRounds :: !Int+ }++instance Show Param where+ showsPrec _ Param {nDsps, nGroups, nMembers} =+ shows nDsps . showString "DSPs/"+ . shows nGroups+ . showString "GRPs/"+ . shows nMembers+ . showString "MEMBERS"++toDspConfig :: Param -> Map DspId Capacity+toDspConfig p =+ let perDspCapacity =+ max 2 (2 * requiredTotal) -- (2 * (ceiling (fromIntegral requiredTotal / fromIntegral (nDsps p) :: Double)))+ requiredTotal = nGroups p + nGroups p * nMembers p+ in Map.fromList ([0 ..] `zip` replicate (nDsps p) perDspCapacity)++-- * Types for the domain of the benchmarks in this module++data MediaApi++data MixingApi++type DspId = Int++type MixerId = Int++type MemberId = Int++type MixingGroupId = Int++data instance Command MediaApi _ where+ FetchDsps :: Command MediaApi ('Return (Set DspId))+ CreateMixer :: DspId -> Command MediaApi ('Return (Maybe MixerId))+ DestroyMixer :: MixerId -> Command MediaApi 'FireAndForget+ AddToMixer :: MixerId -> MemberId -> Command MediaApi ('Return Bool)+ RemoveFromMixer :: MixerId -> MemberId -> Command MediaApi ('Return ())+ MediaShutdown :: Command MediaApi 'FireAndForget++deriving stock instance Show (Command MediaApi ('Return (Set DspId)))++deriving stock instance Show (Command MediaApi ('Return (Maybe MixerId)))++deriving stock instance Show (Command MediaApi ('Return Bool))++deriving stock instance Show (Command MediaApi ('Return ()))++deriving stock instance Show (Command MediaApi 'FireAndForget)++data MixingGroupEvent where+ MemberJoined :: MixingGroupId -> MemberId -> MixingGroupEvent+ MemberUnJoined :: MixingGroupId -> MemberId -> MixingGroupEvent+ deriving stock (Show, Eq)++data instance Command MixingApi _ where+ CreateMixingGroup ::+ MixingGroupId ->+ Command MixingApi ('Return ())+ DestroyMixingGroup ::+ MixingGroupId ->+ Command MixingApi ('Return ())+ Join ::+ IsInput outBox =>+ MixingGroupId ->+ MemberId ->+ outBox MixingGroupEvent ->+ Command MixingApi 'FireAndForget+ UnJoin ::+ IsInput outBox =>+ MixingGroupId ->+ MemberId ->+ outBox MixingGroupEvent ->+ Command MixingApi 'FireAndForget++instance Show (Command MixingApi ('Return ())) where+ showsPrec d (CreateMixingGroup i) =+ showParen (d >= 9) (showString "CreateMixingGroup " . shows i)+ showsPrec d (DestroyMixingGroup i) =+ showParen (d >= 9) (showString "DestroyMixingGroup " . shows i)++instance Show (Command MixingApi 'FireAndForget) where+ showsPrec d (Join i j _) =+ showParen (d >= 9) (showString "Join " . shows i . showChar ' ' . shows j)+ showsPrec d (UnJoin i j _) =+ showParen (d >= 9) (showString "UnJoin " . shows i . showChar ' ' . shows j)++type Capacity = Int++-- | Mix media streams using 'MediaApi'++-- | Media server simulation+--+-- Handle 'MediaApi' requests and manage a map of+-- dsps and mixers.+handleMediaApi ::+ MediaSimSt ->+ Message MediaApi ->+ ReaderT env IO MediaSimSt+handleMediaApi !st =+ \case+ Blocking FetchDsps replyBox -> do+ let !goodDsps = Map.keysSet (Map.filter (> 1) (allDsps st))+ replyTo replyBox goodDsps+ return st+ Blocking (CreateMixer dspId) replyBox ->+ case Map.lookup dspId (allDsps st) of+ Just capacity | capacity > 0 -> do+ let theNewMixerId = nextMixerId st+ replyTo replyBox (Just theNewMixerId)+ return+ ( st+ { allDsps =+ Map.insert+ dspId+ (capacity - 1)+ (allDsps st),+ allMixers =+ Map.insert+ theNewMixerId+ (dspId, Set.empty)+ (allMixers st),+ nextMixerId = theNewMixerId + 1+ }+ )+ _ -> do+ replyTo replyBox Nothing+ return st+ Blocking (AddToMixer theMixer newMember) replyBox ->+ case Map.lookup theMixer (allMixers st) of+ Nothing ->+ replyTo replyBox False >> return st+ Just (!dsp, !streams) ->+ if Set.member newMember streams+ then replyTo replyBox True >> return st+ else case Map.lookup dsp (allDsps st) of+ Just capacity+ | capacity >= 1 ->+ replyTo replyBox True+ >> return+ ( st+ { allMixers =+ Map.insert+ theMixer+ (dsp, Set.insert newMember streams)+ (allMixers st),+ allDsps =+ Map.insert+ dsp+ (capacity - 1)+ (allDsps st)+ }+ )+ _ ->+ replyTo replyBox False >> return st+ Blocking (RemoveFromMixer theMixer theMember) replyBox -> do+ replyTo replyBox ()+ case Map.lookup theMixer (allMixers st) of+ Just (theDsp, theMembers)+ | Set.member theMember theMembers ->+ return+ st+ { allDsps =+ Map.update+ (Just . (+ 1))+ theDsp+ (allDsps st),+ allMixers =+ Map.insert+ theMixer+ (theDsp, Set.delete theMember theMembers)+ (allMixers st)+ }+ _ ->+ return st+ NonBlocking (DestroyMixer theMixerId) ->+ let foundMixer (mixersDsp, mediaStreams) =+ st+ { allMixers =+ Map.delete theMixerId (allMixers st),+ allDsps =+ Map.update+ (Just . (+ (1 + Set.size mediaStreams)))+ mixersDsp+ (allDsps st)+ }+ in pure $+ maybe+ st+ foundMixer+ (Map.lookup theMixerId (allMixers st))+ NonBlocking MediaShutdown ->+ pure st {shuttingDown = True}++data MediaSimSt = MediaSimSt+ { allDsps :: !(Map DspId Capacity),+ nextMixerId :: !MixerId,+ allMixers :: !(Map MixerId (DspId, Set MemberId)),+ shuttingDown :: !Bool+ }++mkMediaSimSt :: Map DspId Capacity -> MediaSimSt+mkMediaSimSt x =+ MediaSimSt+ { allDsps = x,+ nextMixerId = 0,+ allMixers = Map.empty,+ shuttingDown = False+ }
+ src-test/CallIdTest.hs view
@@ -0,0 +1,54 @@+module CallIdTest (test) where++import Control.Monad (replicateM)+import Data.Data+import Data.List (nub)+import qualified UnliftIO.MessageBox.Util.CallId as CallId+import QCOrphans ()+import Test.QuickCheck+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit+ ( assertBool,+ assertEqual,+ testCase,+ )+import Test.Tasty.QuickCheck (testProperty)+import UnliftIO+ ( replicateConcurrently,+ )+import Utils (allEqOrdShowMethodsImplemented, withCallIds)++test :: TestTree+test =+ testGroup+ "UnliftIO.MessageBox.Util.CallId"+ [ testCase+ "Even when a CallId CounterVar is shared by multiple threads/processes, all callIds are unique"+ $ do+ callIds <- withCallIds (replicateConcurrently 1000 CallId.takeNext)+ assertEqual+ "all callids must be unique"+ (nub callIds)+ callIds,+ testCase+ "The next callId is greater and not equal to the previous"+ $ do+ [c1, c2] <- withCallIds (replicateM 2 CallId.takeNext)+ assertBool+ "next callId is not greater"+ (c2 > c1)+ assertBool+ "next callId is equal to the previous"+ (c2 /= c1),+ testProperty+ "more recent CallIds are > (greater than) all previous CallIds"+ $ \(Positive (Small n)) -> ioProperty $ do+ callIds <- withCallIds (replicateM n CallId.takeNext)+ return $+ n > 1 ==> (head callIds < last callIds),+ -- This ONLY exists because I wanted the test-code coverage to be 100% + testCase+ "CallId Show instance"+ (assertEqual "bad show result" "<123>" (show (CallId.MkCallId 123))),+ testProperty "CallId Laws" (allEqOrdShowMethodsImplemented (Proxy @CallId.CallId))+ ]
+ src-test/CatchAllTest.hs view
@@ -0,0 +1,58 @@+module CatchAllTest (test) where++import Data.Proxy+import UnliftIO.MessageBox.Util.Future+import UnliftIO.MessageBox.CatchAll+import UnliftIO.MessageBox.Class+import QCOrphans ()+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck+import UnliftIO+import Utils++test :: TestTree+test =+ testGroup+ "CatchAll"+ [ testProperty+ "Derived Classes Coverage"+ (allEqOrdShowMethodsImplemented (Proxy @(CatchAllFactory Int))),+ testCase+ "when the wrapped receive throws an exception,\+ \ CatchAlls receive returns Nothing"+ $ receive (CatchAllBox BadBox)+ >>= assertEqual "receive should return Nothing" (Nothing :: Maybe ()),+ testCase+ "when the wrapped tryReceive throws an exception,\+ \ CatchAlls tryReceive returns a Future that will always return Nothing"+ $ do+ f <- tryReceive (CatchAllBox BadBox)+ r <- tryNow f+ assertEqual "The Future should return Nothing" (Nothing :: Maybe ()) r,+ testCase+ "when the wrapped receiveAfter throws an exception,\+ \ CatchAlls receiveAfter should return Nothing"+ $ receiveAfter (CatchAllBox BadBox) 123+ >>= assertEqual+ "receiveAfter should return Nothing"+ (Nothing :: Maybe Int),+ testCase+ "when the wrapped deliver throws an exception, CatchAlls deliver returns False"+ $ deliver (CatchAllInput BadInput) ()+ >>= assertEqual "deliver should return False" False+ ]++data BadBox a = BadBox++data BadInput a = BadInput++instance IsMessageBox BadBox where+ type Input BadBox = BadInput+ receive _ = throwIO (stringException "test")+ tryReceive _ = throwIO (stringException "test")+ receiveAfter _ _ = throwIO (stringException "test")+ newInput _ = return BadInput++instance IsInput BadInput where+ deliver _ _ = throwIO (stringException "test")
+ src-test/CommandTest.hs view
@@ -0,0 +1,862 @@+{-# LANGUAGE NoOverloadedStrings #-}+module CommandTest (test) where++import Control.Monad.Reader+ ( MonadIO (liftIO),+ ReaderT (runReaderT),+ void,+ )+import Data.Functor (($>))+import Data.Maybe+ ( fromJust,+ fromMaybe,+ isJust,+ isNothing,+ )+import Data.Semigroup (All (All, getAll))+import UnliftIO.MessageBox.Command+ ( Command,+ CommandError (..),+ DuplicateReply (..),+ Message (Blocking, NonBlocking),+ ReturnType (FireAndForget, Return),+ call,+ callAsync,+ cast,+ delegateCall,+ replyTo,+ tryTakeReply,+ waitForReply,+ )+import UnliftIO.MessageBox.Util.CallId (CallId (MkCallId))+import qualified UnliftIO.MessageBox.Util.CallId as CallId+import UnliftIO.MessageBox.Util.Fresh (CounterVar, fresh, newCounterVar)+import UnliftIO.MessageBox.Class+ ( IsMessageBox (..),+ IsMessageBoxFactory (..),+ handleMessage,+ )+import UnliftIO.MessageBox.Limited+ ( BlockingBoxLimit (BlockingBoxLimit),+ MessageLimit (..),+ )+import UnliftIO.MessageBox.Unlimited (BlockingUnlimited (BlockingUnlimited))+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit+ ( assertBool,+ assertEqual,+ assertFailure,+ testCase,+ )+import Test.Tasty.QuickCheck+ ( Arbitrary (arbitrary),+ Large (getLarge),+ NonEmptyList (getNonEmpty),+ Positive (..),+ PrintableString (..),+ Property,+ ioProperty,+ testProperty,+ )+import UnliftIO+ ( conc,+ concurrently,+ newEmptyMVar,+ putMVar,+ runConc,+ takeMVar,+ throwTo,+ timeout,+ try,+ tryReadMVar,+ )+import UnliftIO.Concurrent (forkIO, threadDelay)+import Utils (NoOpInput (OnDeliver), untilJust, withCallIds)++test :: TestTree+test =+ testGroup+ "UnliftIO.MessageBox.Command"+ [ unitTests,+ integrationTests+ ]++integrationTests :: TestTree+integrationTests =+ testGroup+ "integration tests"+ [ testCase "Show instance of the Message data family works" $+ CallId.newCallIdCounter+ >>= \cv -> flip runReaderT cv $ do+ bookStoreOutput <- newMessageBox BlockingUnlimited+ bookStoreInput <- newInput bookStoreOutput+ let blockingMsg = GetBooks+ nonblockingMsg =+ Donate+ (Donor (PersonName "A" "B") 1)+ ( Book+ "Unsere Welt neu denken: Eine Einladung"+ (Author (PersonName "Maya" "Goepel") 208)+ (Publisher "Ullstein")+ (Year 2020)+ []+ )+ (_, shownBlockingMsg) <-+ concurrently+ (call bookStoreInput blockingMsg 1_000_000)+ (handleMessage bookStoreOutput (return . show))+ (_, shownNonBlockingMsg) <-+ concurrently+ (cast bookStoreInput nonblockingMsg)+ (handleMessage bookStoreOutput (return . show))+ liftIO $+ assertEqual+ "bad Show instance"+ (Just ("B: " <> showsPrec 9 blockingMsg " <1>"))+ shownBlockingMsg+ liftIO $+ assertEqual+ "bad Show instance"+ (Just ("NB: " <> showsPrec 9 nonblockingMsg ""))+ shownNonBlockingMsg,+ testProperty+ "all books that many donors concurrently donate into the book store end up in the bookstore"+ allDonatedBooksAreInTheBookStore,+ testCase "handling a cast succeeds" $ do+ bookStoreOutput <- newMessageBox BlockingUnlimited+ bookStoreInput <- newInput bookStoreOutput+ let expected =+ Donate+ (Donor (PersonName "Mueller" "Hans") 1)+ ( Book+ "Wann wenn nicht wir."+ (Author (PersonName "Schmitt" "Agent") 477)+ (Publisher "Kletten Verlag")+ (Year 2019)+ []+ )+ wasHandleMessageCalled <- newEmptyMVar+ void (forkIO (void (cast bookStoreInput expected)))+ result <- handleMessage bookStoreOutput $+ \case+ NonBlocking actual -> do+ putMVar wasHandleMessageCalled ()+ assertEqual "correct message" expected actual+ a ->+ assertFailure ("unexpected message: " <> show a)+ tryReadMVar wasHandleMessageCalled >>= assertBool "handle message must be called" . isJust+ assertBool "handling the message was successful" (isJust result),+ testCase "handling a call succeeds" $ do+ freshCounter <- newCounterVar+ flip runReaderT (MkBookStoreEnv {_fresh = freshCounter}) $ do+ let expectedBooks =+ [ Book+ "Wann wenn nicht wir."+ (Author (PersonName "Schmitt" "Agent") 477)+ (Publisher "Kletten Verlag")+ (Year 2019)+ []+ ]++ bookStoreOutput <- newMessageBox BlockingUnlimited+ bookStoreInput <- newInput bookStoreOutput+ let concurrentCallAction = call bookStoreInput GetBooks 100+ concurrentBookStore = handleMessage bookStoreOutput $ \case+ Blocking GetBooks replyBox -> do+ Nothing <$ replyTo replyBox expectedBooks+ NonBlocking a ->+ pure (Just ("unexpected message: " <> show a))+ (callResult, handleResult) <- concurrently concurrentCallAction concurrentBookStore+ liftIO $ do+ case callResult of+ Left err ->+ assertFailure $ "failed unexpectedly: " <> show err+ Right actualBooks ->+ assertEqual "call should return the right books" actualBooks expectedBooks+ case handleResult of+ Nothing ->+ assertFailure "could not receive message in handleMessage"+ Just Nothing -> return ()+ Just (Just err) ->+ assertFailure err,+ testCase "reply to dead caller does not crash" $ do+ let delayDuration = 1000+ freshCounter <- newCounterVar+ flip runReaderT (MkBookStoreEnv {_fresh = freshCounter}) $ do+ bookStoreOutput <- newMessageBox BlockingUnlimited+ bookStoreInput <- newInput bookStoreOutput+ let concurrentCallAction = do+ tId <- forkIO (void $ call bookStoreInput GetBooks (delayDuration * 3))+ threadDelay (1 * delayDuration)+ throwTo tId $ userError "uh-oh"++ concurrentBookStore = handleMessage bookStoreOutput $ \case+ Blocking GetBooks replyBox -> do+ threadDelay (2 * delayDuration)+ replyTo replyBox []+ pure True+ NonBlocking a ->+ error (show a)+ (_callResult, handleResult) <- concurrently concurrentCallAction concurrentBookStore+ liftIO $+ assertEqual "expect bookStore to survive a failed reply" (Just True) handleResult,+ testCase+ "call to died thread returns Nothing"+ $ do+ let delayDuration = 1000_000+ freshCounter <- newCounterVar+ flip runReaderT (MkBookStoreEnv {_fresh = freshCounter}) $ do+ bookStoreOutput <- newMessageBox BlockingUnlimited+ bookStoreInput <- newInput bookStoreOutput+ let concurrentCallAction = call bookStoreInput GetBooks (delayDuration * 2)+ concurrentBookStore = handleMessage bookStoreOutput $ \case+ Blocking GetBooks _replyBox -> do+ threadDelay delayDuration+ pure Nothing+ NonBlocking a ->+ pure (Just ("unexpected message: " <> show a))+ (callResult, handleResult) <- concurrently concurrentCallAction concurrentBookStore+ liftIO $ do+ callId' <- runReaderT fresh freshCounter+ assertEqual "no message other than GetBooks received" (Just Nothing) handleResult+ case callResult of+ (Left (BlockingCommandTimedOut (MkCallId c)))+ | MkCallId (c + 1) == callId' -> pure ()+ _ -> assertFailure "call result should match (Left (BlockingCommandTimedOut _))",+ testCase "replying to call within the given timeout is successful" $ do+ freshCounter <- newCounterVar+ flip runReaderT (MkBookStoreEnv {_fresh = freshCounter}) $ do+ let oneMilliSecond = 1_000 -- one milli second is 1000 micro seconds+ serverOutput <- newMessageBox BlockingUnlimited+ serverInput <- newInput serverOutput+ let bookstoreClient = call serverInput GetBooks (20 * oneMilliSecond)+ bookstoreServer = handleMessage serverOutput $ \case+ Blocking GetBooks replyBox -> do+ void (replyTo replyBox [])+ return Nothing+ NonBlocking a ->+ pure (Just ("unexpected message: " <> show a))+ (clientResult, serverResult) <- concurrently bookstoreClient bookstoreServer+ liftIO $ do+ assertEqual "unexpected message received: " (Just Nothing) serverResult+ case clientResult of+ (Right actualBooks) -> assertEqual "call should return the right books: " actualBooks []+ other -> assertFailure $ "unexpected call result: " <> show other,+ testCase+ "when the server sends a reply after the call timeout has elapsed, the call function \+ \returns 'Left BlockingCommandTimedOut', and the replyTo function returns False"+ $ do+ freshCounter <- newCounterVar+ flip runReaderT (MkBookStoreEnv {_fresh = freshCounter}) $ do+ bookStoreOutput <- newMessageBox BlockingUnlimited+ bookStoreInput <- newInput bookStoreOutput+ let client = do+ result <- call bookStoreInput GetBooks 100_000+ case result of+ Left (BlockingCommandTimedOut _) ->+ -- this is the error we expected+ return $ return ()+ unexpectedOther ->+ return $+ assertFailure+ ( "call result should match (Left (BlockingCommandTimedOut _)), unexpected: "+ <> show unexpectedOther+ )+ server =+ handleMessage bookStoreOutput $ \case+ Blocking GetBooks replyBox -> do+ threadDelay 500_000 -- artificial delay to cause the client to+ -- loose patience and timeout+ void (replyTo replyBox [])+ return (return ())+ NonBlocking a ->+ return (assertFailure ("unexpected message: " <> show a))+ (clientAssertion, serverAssertion) <-+ concurrently client server+ liftIO $ do+ clientAssertion+ fromMaybe (assertFailure "book store server died unexpectedly") serverAssertion,+ testCase+ "when the server receives the call and does not send a reply and immediately \+ \exits, the call function returns a timeout error"+ $ do+ freshCounter <- newCounterVar+ let env = MkBookStoreEnv {_fresh = freshCounter}+ (serverResult, clientResult) <- flip runReaderT env $ do+ bookStoreOutput <- newMessageBox (BlockingBoxLimit MessageLimit_16)+ bookStoreInput <- newInput bookStoreOutput+ concurrently+ ( handleMessage bookStoreOutput $ \case+ Blocking GetBooks _replyBox -> return Nothing+ u -> return (Just ("unexpected message: " <> show u))+ )+ ( timeout+ 1_000_000+ (call bookStoreInput GetBooks 100_000)+ )+ assertEqual+ "unexpected serverResult"+ (Just Nothing)+ serverResult+ assertEqual+ "unexpected clientResult"+ (Just (Left (BlockingCommandTimedOut (MkCallId 1))))+ clientResult+ ]++newtype BookStoreEnv = MkBookStoreEnv+ {_fresh :: CounterVar CallId}++instance CallId.HasCallIdCounter BookStoreEnv where+ getCallIdCounter MkBookStoreEnv {_fresh} = _fresh++allDonatedBooksAreInTheBookStore :: [(Donor, Book)] -> Property+allDonatedBooksAreInTheBookStore donorsAndBooks = ioProperty $ do+ bookStoreIn <- newMessageBox BlockingUnlimited+ bookStoreOut <- newInput bookStoreIn+ getAll+ <$> runConc+ ( foldMap+ ( \(donor, book) ->+ conc (cast bookStoreOut (Donate donor book) $> All True)+ )+ donorsAndBooks+ <> pure (All True)+ )++data BookStore++data instance Command BookStore _ where+ Donate :: Donor -> Book -> Command BookStore 'FireAndForget+ GetBooks :: Command BookStore ('Return [Book])++deriving stock instance Eq (Command BookStore 'FireAndForget)++deriving stock instance Show (Command BookStore 'FireAndForget)++deriving stock instance Eq (Command BookStore ('Return [Book]))++deriving stock instance Show (Command BookStore ('Return [Book]))++newtype Year = Year Int+ deriving newtype (Show, Eq, Ord)+ deriving (Arbitrary) via (Positive Int)++newtype Page = Page String+ deriving newtype (Show, Eq, Ord)+ deriving (Arbitrary) via PrintableString++data PersonName = PersonName+ { lastName :: String,+ surName :: String+ }+ deriving stock (Show, Eq, Ord)++instance Arbitrary PersonName where+ arbitrary =+ PersonName+ <$> (getPrintableString <$> arbitrary)+ <*> (getPrintableString <$> arbitrary)++data Donor = Donor+ { donorName :: PersonName,+ donorId :: Int+ }+ deriving stock (Show, Eq, Ord)++instance Arbitrary Donor where+ arbitrary =+ Donor+ <$> arbitrary+ <*> (getLarge . getPositive <$> arbitrary)++data Author = Author+ { authorName :: PersonName,+ authorId :: Int+ }+ deriving stock (Show, Eq, Ord)++instance Arbitrary Author where+ arbitrary =+ Author+ <$> arbitrary+ <*> (getLarge . getPositive <$> arbitrary)++newtype Publisher = Publisher String+ deriving newtype (Show, Eq, Ord)+ deriving (Arbitrary) via PrintableString++data Book = Book+ { bookTitle :: String,+ bookAuthor :: Author,+ bookPublisher :: Publisher,+ bookYear :: Year,+ bookContent :: [Page]+ }+ deriving stock (Show, Eq, Ord)++instance Arbitrary Book where+ arbitrary =+ Book+ <$> (getPrintableString <$> arbitrary)+ <*> arbitrary+ <*> arbitrary+ <*> arbitrary+ <*> (getNonEmpty <$> arbitrary)++-- --------------------------------------------------------+-- Unit tests+-- --------------------------------------------------------++unitTests :: TestTree+unitTests =+ testGroup+ "unit tests"+ [ testCase "Message Show instance" $ do+ withCallIds+ ( callAsync+ ( OnDeliver $ \m -> do+ liftIO (assertEqual "bad show result" "B: (Echo \"123\") <1>" (show m))+ return True+ )+ (Echo "123")+ )+ >>= assertBool "should not fail" . isJust+ withCallIds+ ( cast+ ( OnDeliver $ \m -> do+ liftIO (assertEqual "bad show result" "NB: (Tell 123)" (show m))+ return True+ )+ (Tell "123")+ )+ >>= assertBool "should not fail",+ testCase "CommandError Show instance" $ do+ assertEqual+ "Show instance broken"+ "CouldNotEnqueueCommand <1>"+ (show (CouldNotEnqueueCommand (MkCallId 1)))+ assertEqual+ "Show instance broken"+ "BlockingCommandFailure <1>"+ (show (BlockingCommandFailure (MkCallId 1)))+ assertEqual+ "Show instance broken"+ "BlockingCommandTimedOut <1>"+ (show (BlockingCommandTimedOut (MkCallId 1))),+ testCase+ "when deliver returns False, BlockingCommandFailed\+ \ should be returned"+ $ withCallIds+ ( call+ (OnDeliver (const (return False)))+ (Echo "123")+ 2_000_000+ )+ >>= assertEqual+ "bad call result"+ (Left (CouldNotEnqueueCommand (MkCallId 1))),+ testCase+ "when deliver blocks forever, call also blocks forever\+ \ regardles of the timeout parameter"+ $ do+ res <-+ withCallIds $+ timeout 10_000 $+ call+ ( OnDeliver+ ( const $ do+ threadDelay 3600_000_000+ return False+ )+ )+ (Echo "123")+ 1+ assertEqual "bad call result" Nothing res,+ testCase+ "when deliver succeeds, but the reply box is ignored \+ \and no replyTo invokation is done, BlockingCommandTimedOut\+ \ should be returned"+ $ do+ res <-+ withCallIds $+ call+ (OnDeliver (const (return True)))+ (Echo "123")+ 10_000+ assertEqual+ "bad call result"+ (Left (BlockingCommandTimedOut (MkCallId 1)))+ res,+ testCase+ "when the reply box is passed to a new process, but\+ \ replyTo is delayed very much,\+ \ BlockingCommandTimedOut should be returned"+ $ do+ res <-+ withCallIds $+ call+ ( OnDeliver+ ( \(Blocking (Echo x) rbox) -> do+ void $+ forkIO $ do+ threadDelay 3600_000_000+ replyTo rbox x+ return True+ )+ )+ (Echo "123")+ 10_000+ assertEqual+ "bad call result"+ (Left (BlockingCommandTimedOut (MkCallId 1)))+ res,+ testCase+ "when the reply box is passed to a new process, that calls\+ \ replyTo after x seconds, where x < rt (the receive timeou),\+ \ then before y seconds have passed, where x <= y < rt,\+ \ call returns the value passed to the reply box."+ $ do+ let x = 20_000+ rt = 1_000_000+ y = 500_000+ res <-+ withCallIds $+ timeout y $+ call+ ( OnDeliver+ ( \(Blocking (Echo value) rbox) -> do+ void $+ forkIO $ do+ threadDelay x+ replyTo rbox value+ return True+ )+ )+ (Echo "123")+ rt+ assertEqual+ "bad call result"+ (Just (Right "123"))+ res,+ testCase+ "when a call is delegated to, and replied by, another process, \+ \ call should return that reply"+ $ do+ res <-+ withCallIds $+ call+ ( OnDeliver+ ( \(Blocking (Echo value) rbox) ->+ delegateCall+ ( OnDeliver+ ( \(Blocking (Echo value') rbox') -> do+ replyTo rbox' value'+ return True+ )+ )+ (Echo value)+ rbox+ )+ )+ (Echo "123")+ 1_000+ assertEqual+ "bad call result"+ (Right "123")+ res,+ testCase+ "when deliver has a delay, callAsync \+ \ has at least the same delay"+ $ do+ res <-+ withCallIds $+ timeout+ 90_000+ ( callAsync+ ( OnDeliver+ ( \(Blocking (Echo _value) _rbox) ->+ do+ threadDelay 100_000+ return False+ )+ )+ (Echo "123")+ )+ assertBool+ "callAsync returned too soon"+ (isNothing res),+ testCase+ "when deliver fails, callAsync returns Nothing"+ $ do+ res <-+ withCallIds $+ callAsync+ ( OnDeliver+ (\(Blocking (Echo _value) _rbox) -> return False)+ )+ (Echo "123")+ assertBool+ "callAsync should fail"+ (isNothing res),+ testCase+ "when deliver succeeds, callAsync returns Just the pending reply"+ $ do+ res <-+ withCallIds $+ callAsync+ ( OnDeliver+ (\(Blocking (Echo _value) _rbox) -> return True)+ )+ (Echo "123")+ assertBool+ "callAsync should succeed"+ (isJust res),+ testCase+ "when deliver succeeds, but no reply is sent, \+ \ waitForReply returns Left BlockingCommandTimedOut"+ $ withCallIds+ ( callAsync+ ( OnDeliver+ (\(Blocking (Echo _value) _rbox) -> return True)+ )+ (Echo "123")+ >>= traverse (waitForReply 1_000)+ )+ >>= assertEqual+ "timeout error expected"+ (Just (Left (BlockingCommandTimedOut (MkCallId 1)))),+ testCase+ "when deliver succeeds, but no reply is sent, \+ \ tryTakeReply will return Nothing"+ $ withCallIds+ ( callAsync+ ( OnDeliver+ (\(Blocking (Echo _value) _rbox) -> return True)+ )+ (Echo "123")+ >>= traverse tryTakeReply+ )+ >>= assertBool "tryTakeReply should return Nothing" . isNothing . fromJust,+ testCase+ "when deliver succeeds, and a reply is sent, \+ \ waitForReply will return Right with the reply"+ $ withCallIds+ ( callAsync+ ( OnDeliver+ ( \(Blocking (Echo value) rbox) -> do+ replyTo rbox value+ return True+ )+ )+ (Echo "123")+ >>= traverse (waitForReply 1_000_000)+ )+ >>= assertEqual "waitForReply should return the reply" (Just (Right "123")),+ testCase+ "when deliver has succeeded, and a reply has been sent, \+ \ tryTakeReply will eventually return Just Right the reply"+ $ withCallIds+ ( callAsync+ ( OnDeliver+ ( \(Blocking (Echo value) rbox) -> do+ replyTo rbox value+ return True+ )+ )+ (Echo "123")+ >>= ( \x -> do+ threadDelay 10_000+ traverse tryTakeReply x+ )+ )+ >>= assertEqual+ "tryTakeReply should return the reply"+ (Just (Just (Right "123"))),+ testCase+ "after waitForReply returns Right the reply,\+ \ any further call to waitForReply will return\+ \ the same result immediately."+ $ withCallIds $+ callAsync+ ( OnDeliver+ ( \(Blocking (Echo value) rbox) -> do+ replyTo rbox value+ return True+ )+ )+ (Echo "123")+ >>= maybe+ (liftIO $ assertFailure "callAsync failed unexpectedly")+ ( \pendingReply -> do+ waitForReply 1_000_000 pendingReply+ >>= liftIO+ . assertEqual+ "waitForReply should return the reply"+ (Right "123")+ timeout 1_000 (waitForReply 1_000_000 pendingReply)+ >>= liftIO+ . maybe+ (assertFailure "the second waitForReply should return immediately")+ ( assertEqual+ "the second waitForReply should return the reply"+ (Right "123")+ )+ ),+ testCase+ "when waitForReply has returned, tryTakeReply will\+ \ also return Just the same result"+ $ withCallIds $+ callAsync+ ( OnDeliver+ ( \(Blocking (Echo value) rbox) -> do+ replyTo rbox value+ return True+ )+ )+ (Echo "123")+ >>= maybe+ (liftIO $ assertFailure "callAsync failed unexpectedly")+ ( \pendingReply -> do+ waitForReply 1_000_000 pendingReply+ >>= liftIO+ . assertEqual+ "waitForReply should return the reply"+ (Right "123")+ tryTakeReply pendingReply+ >>= liftIO+ . maybe+ (assertFailure "the second tryTakeReply should return Just (...)")+ ( assertEqual+ "tryTakeReply should return the reply"+ (Right "123")+ )+ ),+ testCase+ "when tryTakeReply has returned Just a result, waitForReply will\+ \ also return that same result"+ $ withCallIds $+ callAsync+ ( OnDeliver+ ( \(Blocking (Echo value) rbox) -> do+ replyTo rbox value+ return True+ )+ )+ (Echo "123")+ >>= maybe+ (liftIO $ assertFailure "callAsync failed unexpectedly")+ ( \pendingReply -> do+ timeout 1_000_000 (untilJust (tryTakeReply pendingReply))+ >>= maybe+ (liftIO $ assertFailure "tryTakeReply should return a result")+ ( liftIO+ . assertEqual+ "tryTakeReply should return the reply"+ (Right "123")+ )+ waitForReply 1_000 pendingReply+ >>= liftIO+ . assertEqual+ "waitForReply should return the same reply"+ (Right "123")+ ),+ testCase "DuplicateReply Eq and Show instance" $ do+ assertEqual "Eq instance" (DuplicateReply (MkCallId 1)) (DuplicateReply (MkCallId 1))+ assertEqual "Show instance" "more than one reply sent for: <1>" (show (DuplicateReply (MkCallId 1))),+ testCase+ "when replyTo is called again on the same replyBox, an error is thrown\+ \ and the waitForResult will still return the first reply"+ $ withCallIds+ ( do+ reply <-+ timeout 1_000_000 $+ try $+ callAsync+ ( OnDeliver+ ( \(Blocking (Echo value) rbox) -> do+ replyTo rbox value+ replyTo rbox value+ return True+ )+ )+ (Echo "123")++ liftIO $ case reply of+ (Just (Left (DuplicateReply cId))) ->+ assertEqual "wrong callId in DuplicateReply exception" (MkCallId 1) cId+ (Just (Right _)) ->+ assertFailure "expected an exception to be thrown in second `replyTo` call"+ Nothing ->+ assertFailure "unexpectedly indefinitely blocked on `replyTo` call"+ ),+ testCase+ "tryTakeReply should return Nothing until the reply is sent, and\+ \ from the moment the reply is available, it will always and\+ \ immediately return the reply, it will never return Nothing again"+ $ withCallIds+ ( do+ replyNow <- newEmptyMVar+ replySent <- newEmptyMVar+ mPendingReply <-+ callAsync+ ( OnDeliver+ ( \(Blocking (Echo value) rbox) -> do+ void $+ forkIO $ do+ () <- takeMVar replyNow+ replyTo rbox value+ putMVar replySent ()+ return True+ )+ )+ (Echo "")+ case mPendingReply of+ Nothing ->+ liftIO $ assertFailure "callAsync failed"+ Just pendingReply -> do+ tryTakeReply pendingReply+ >>= liftIO . assertEqual "tryTakeReply failed" Nothing+ putMVar replyNow ()+ takeMVar replySent+ tryTakeReply pendingReply+ >>= liftIO+ . assertEqual "tryTakeReply failed" (Just (Right ""))+ ),+ testCase+ "PendingResult's show instance contains the call Id and the \+ \ name of the result type"+ $ do+ res <-+ withCallIds $+ callAsync+ ( OnDeliver+ (const (return True))+ )+ (Echo (Just "123"))+ case res of+ Nothing -> assertFailure "unexpected delivery failure"+ Just pendingReply ->+ assertEqual+ "bad show instance for PendingResult"+ "AR: <1>"+ (show pendingReply)+ ]++-- Echo Protocol for callTest++data Echo++data instance Command Echo _ where+ Echo :: a -> Command Echo ('Return a)++instance (Show a) => Show (Command Echo ('Return a)) where+ showsPrec !d (Echo x) = showParen (d >= 9) (showString "Echo " . showsPrec 9 x)++data Tell++data instance Command Tell _ where+ Tell :: String -> Command Tell 'FireAndForget++instance Show (Command Tell 'FireAndForget) where+ showsPrec !d (Tell x) = showParen (d >= 9) (showString "Tell " . showString x)
+ src-test/CornerCaseTests.hs view
@@ -0,0 +1,296 @@+-- | Test race conditions+module CornerCaseTests (test) where++import Control.Monad.Reader (ReaderT (runReaderT))+import Data.Semigroup (Any (Any, getAny), Semigroup (stimes))+import GHC.Stack (HasCallStack)+import System.Mem (performMajorGC, performMinorGC)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit+ ( assertBool,+ assertEqual,+ assertFailure,+ testCase,+ )+import UnliftIO+ ( MVar,+ MonadIO (..),+ MonadUnliftIO,+ concurrently,+ newEmptyMVar,+ putMVar,+ takeMVar,+ timeout,+ )+import UnliftIO.Concurrent (forkIO, threadDelay, yield)+import UnliftIO.MessageBox.Class+ ( Input,+ IsInput (deliver),+ IsMessageBox (newInput, receive),+ IsMessageBoxFactory (..),+ handleMessage,+ receiveAfter,+ )+import UnliftIO.MessageBox.Command+ ( Command,+ CommandError (BlockingCommandTimedOut),+ Message (Blocking),+ ReturnType (Return),+ call,+ )+import qualified UnliftIO.MessageBox.Limited as B+import qualified UnliftIO.MessageBox.Unlimited as U+import UnliftIO.MessageBox.Util.CallId+ ( CallId (MkCallId),+ )+import qualified UnliftIO.MessageBox.Util.CallId as CallId++test :: TestTree+test =+ testGroup+ "CornerCaseTests"+ [ testGroup+ "waiting for messages from a dead process"+ [ testCase "When using the UnlimitedMessageBox, receive will timeout" $+ waitForMessageFromDeadProcess U.BlockingUnlimited+ >>= assertEqual "unexpected return value: " SecondReceiveReturnedNothing,+ testCase "When using the Limited Message BlockingBox, the test will timeout" $+ waitForMessageFromDeadProcess (B.BlockingBoxLimit B.MessageLimit_16)+ >>= assertEqual "unexpected return value: " SecondReceiveReturnedNothing,+ testCase "When using the Limited Waiting Box, the test will timeout" $+ waitForMessageFromDeadProcess (B.WaitingBoxLimit Nothing 1_000_000 B.MessageLimit_16)+ >>= assertEqual "unexpected return value: " SecondReceiveReturnedNothing,+ testCase "When using the Limited NonBlocking Box, the test will timeout" $+ waitForMessageFromDeadProcess (B.NonBlockingBoxLimit B.MessageLimit_16)+ >>= assertEqual "unexpected return value: " SecondReceiveReturnedNothing+ ],+ testGroup+ "sending messages to a dead process"+ [ testCase "When using the UnlimitedMessageBox, sending messages succeeds" $+ sendMessagesToDeadProcess U.BlockingUnlimited+ >>= assertEqual "unexpected result: " SomeMoreMessagesSent,+ testCase "When using the BlockingBoxLimit, sending messages eventually blocks and times out" $+ sendMessagesToDeadProcess (B.BlockingBoxLimit B.MessageLimit_16)+ >>= assertEqual "unexpected result: " SendingMoreMessagesTimedOut,+ testCase "When using the NonBlockingBoxLimit, sending messages eventually returns False" $+ sendMessagesToDeadProcess (B.NonBlockingBoxLimit B.MessageLimit_16)+ >>= assertEqual "unexpected result: " SomeMoreMessagesSent,+ testCase "When using the WaitingBoxLimit Nothing, sending messages eventually blocks and times out" $+ sendMessagesToDeadProcess (B.WaitingBoxLimit Nothing 1_000 B.MessageLimit_16)+ >>= assertEqual "unexpected result: " SomeMoreMessagesSent+ ],+ testGroup+ "Command"+ [ testGroup+ "waiting for a call reply after the server died"+ [ testCase "When using the UnlimitedMessageBox, BlockingCommandTimedOut is returned" $+ waitForCallReplyFromDeadServer U.BlockingUnlimited+ >>= assertEqual+ "unexpected result: "+ (CallFailed (BlockingCommandTimedOut (MkCallId 1))),+ testCase "When using the BlockingBoxLimit, BlockingCommandTimedOut is returned" $+ waitForCallReplyFromDeadServer (B.BlockingBoxLimit B.MessageLimit_16)+ >>= assertEqual+ "unexpected result: "+ (CallFailed (BlockingCommandTimedOut (MkCallId 1))),+ testCase "When using the NonBlockingBoxLimit, BlockingCommandTimedOut is returned" $+ waitForCallReplyFromDeadServer (B.NonBlockingBoxLimit B.MessageLimit_16)+ >>= assertEqual+ "unexpected result: "+ (CallFailed (BlockingCommandTimedOut (MkCallId 1))),+ testCase "When using WaitingBoxLimit Nothing, BlockingCommandTimedOut is returned" $+ waitForCallReplyFromDeadServer (B.WaitingBoxLimit Nothing 123_456 B.MessageLimit_16)+ >>= assertEqual+ "unexpected result: "+ (CallFailed (BlockingCommandTimedOut (MkCallId 1)))+ ]+ ]+ ]++data WaitForMessageFromDeadProcessResult+ = SecondReceiveReturnedNothing+ | SecondReceiveWasSuccessful+ deriving stock (Show, Eq)++waitForMessageFromDeadProcess ::+ (HasCallStack, IsMessageBoxFactory cfg) =>+ cfg ->+ IO WaitForMessageFromDeadProcessResult+waitForMessageFromDeadProcess outputCfg =+ do+ firstMessageSent <- newEmptyMVar+ let msg1 :: Int+ msg1 = 42+ output <- newMessageBox outputCfg+ _ <- forkIO $ do+ input <- newInput output+ deliver input msg1 >>= assertBool "first message not sent!"+ putMVar firstMessageSent ()+ takeMVar firstMessageSent+ receive output+ >>= maybe+ (assertFailure "failed to receive first message")+ (assertEqual "invalid first message received!" msg1)+ threadDelay 10_000+ liftIO performMinorGC+ liftIO performMajorGC+ threadDelay 10_000+ receiveAfter output 3_000_000+ >>= maybe+ (return SecondReceiveReturnedNothing)+ (const (return SecondReceiveWasSuccessful))++data SendMessageToDeadProcessResult+ = NoMoreMessagesSent+ | SomeMoreMessagesSent+ | SendingMoreMessagesTimedOut+ deriving stock (Show, Eq)++sendMessagesToDeadProcess ::+ (HasCallStack, IsMessageBoxFactory cfg) =>+ cfg ->+ IO SendMessageToDeadProcessResult+sendMessagesToDeadProcess outputCfg =+ do+ ready <- newEmptyMVar+ firstMessageSent <- newEmptyMVar+ done <- newEmptyMVar+ let msg1 :: Int+ msg1 = 42++ _receiver <- forkIO $ do+ output <- newMessageBox outputCfg+ input <- newInput output+ putMVar ready input+ takeMVar firstMessageSent+ receive output >>= putMVar done++ input <- takeMVar ready+ deliver input msg1+ >>= assertBool "first message not sent!"+ putMVar firstMessageSent ()+ takeMVar done+ >>= assertEqual "first message invalid" (Just msg1)+ threadDelay 10_000+ liftIO performMinorGC+ liftIO performMajorGC+ threadDelay 10_000++ timeout+ 2_000_000+ (stimes (100 :: Int) (Any <$> deliver input msg1))+ >>= maybe+ (return SendingMoreMessagesTimedOut)+ ( \r ->+ return $+ if getAny r+ then SomeMoreMessagesSent+ else NoMoreMessagesSent+ )++-- ------------------------------------------------------------++data WaitForCallReplyFromDeadServerResult+ = WaitForCallReplyFromDeadServerResult+ | CallFailed CommandError+ | WaitForCallReplyFromDeadServerNoResult+ deriving stock (Show, Eq)++waitForCallReplyFromDeadServer ::+ (IsMessageBoxFactory cfg) =>+ cfg ->+ IO WaitForCallReplyFromDeadServerResult+waitForCallReplyFromDeadServer outputCfg =+ do+ ready <- newEmptyMVar+ (cr, sr) <-+ concurrently+ (waitForCallReplyFromDeadServerClient ready)+ (waitForCallReplyFromDeadServerServer outputCfg ready)+ assertEqual "unexpected server process exit: " () sr+ return cr++waitForCallReplyFromDeadServerClient ::+ (MonadIO m, IsInput o, MonadUnliftIO m) =>+ MVar (o (Message TestServer)) ->+ m WaitForCallReplyFromDeadServerResult+waitForCallReplyFromDeadServerClient ready = do+ input <- takeMVar ready+ cic <- CallId.newCallIdCounter+ runReaderT (threadDelay 1_000 >> call input TestCall 100_000) cic+ >>= either+ (return . CallFailed)+ (const (return WaitForCallReplyFromDeadServerResult))++waitForCallReplyFromDeadServerServer ::+ (IsMessageBoxFactory cfg) =>+ cfg ->+ MVar (Input (MessageBox cfg) (Message TestServer)) ->+ IO ()+waitForCallReplyFromDeadServerServer outputCfg ready = do+ output <- newMessageBox outputCfg+ input <- newInput output+ putMVar ready input+ -- threadDelay 1_000+ handleLoop output+ where+ handleLoop ::+ (MonadUnliftIO m, IsMessageBox output) =>+ output (Message TestServer) ->+ m ()+ handleLoop output =+ handleMessage+ output+ ( \case+ Blocking TestCall _r -> do+ _ <- forkIO $ do+ threadDelay 10_000+ liftIO performMinorGC+ liftIO performMajorGC+ return ()+ )+ >>= maybe+ (yield >> handleLoop output)+ return++data TestServer++data instance Command TestServer _ where+ TestCall :: Command TestServer ('Return ())++deriving stock instance Show (Command TestServer ('Return ()))++-- Simple server loop++-- data ServerState protocol model = MkServerState+-- { state :: model,+-- self :: Input protocol+-- }++-- data ServerLoopResult model where+-- Continue :: ServerLoopResult model+-- ContinueWith :: model -> ServerLoopResult model+-- StopServerLoop :: ServerLoopResult model++-- type InitCallback protocol model err m =+-- ServerState protocol () -> m (Either err model)++-- type UpdateCallback protocol model m =+-- ServerState protocol model -> Message protocol -> m (ServerLoopResult model)++-- type CleanupCallback protocol model m =+-- ServerState protocol model -> m ()++-- forkServer ::+-- InitCallback protocol model initErrror m ->+-- UpdateCallback protocol model m ->+-- CleanupCallback protocol model m ->+-- m (Either initError (Input protocol))+-- forkServer = undefined++-- data Counter++-- data instance Command Counter t where+-- Incr :: Command Counter FireAndForget+-- Set :: Int -> Command Counter FireAndForget+-- Get :: Command Counter (Return Int)
+ src-test/FreshTest.hs view
@@ -0,0 +1,9 @@+module FreshTest (test) where++import qualified Test.Tasty as Tasty++test :: Tasty.TestTree+test =+ Tasty.testGroup+ "Protocols.Fresh"+ []
+ src-test/LimitedMessageBoxTest.hs view
@@ -0,0 +1,486 @@+{-# LANGUAGE NoOverloadedStrings #-}++module LimitedMessageBoxTest (test) where++import Control.Monad+ ( forever,+ replicateM,+ void,+ when,+ )+import Data.Function (fix)+import Data.Maybe+ ( fromMaybe,+ isNothing,+ )+import Data.Proxy (Proxy (Proxy))+import MessageBoxCommon (testContentionRobustness)+import UnliftIO.MessageBox.Util.Future (tryNow)+import UnliftIO.MessageBox.Class+ ( IsInput (deliver),+ IsMessageBox (newInput, receive, tryReceive),+ IsMessageBoxFactory (MessageBox, newMessageBox),+ receiveAfter,+ )+import UnliftIO.MessageBox.Limited+ ( BlockingBoxLimit (..),+ MessageLimit (..),+ NonBlockingBoxLimit (..),+ WaitingBoxLimit (..),+ messageLimitToInt,+ )+import QCOrphans ()+import Test.QuickCheck+ ( Positive (Positive),+ Small (Small),+ ioProperty,+ withMaxSuccess,+ (==>),+ )+import Test.Tasty as Tasty (TestTree, testGroup)+import Test.Tasty.HUnit as Tasty+ ( assertBool,+ assertEqual,+ testCase,+ (@?=),+ )+import Test.Tasty.QuickCheck as Tasty (testProperty)+import UnliftIO (concurrently, race, timeout)+import UnliftIO.Concurrent (threadDelay)+import Utils (allEnumMethodsImplemented, allEqOrdShowMethodsImplemented)++test :: Tasty.TestTree+test =+ testGroup+ "UnliftIO.MessageBox.Limited"+ [ testBlockingBox,+ testWaitingBox,+ testNonBlockingBox,+ testCommon BlockingBoxLimit,+ testCommon NonBlockingBoxLimit,+ testCommon (WaitingBoxLimit Nothing 5_000_000),+ testCommon (WaitingBoxLimit (Just (48 * 3600_000_000)) 5_000_000),+ derivedInstances+ ]++testCommon ::+ forall boxCfg cfg.+ ( IsMessageBoxFactory boxCfg,+ Show boxCfg,+ cfg ~ MessageBox boxCfg+ ) =>+ (MessageLimit -> boxCfg) ->+ Tasty.TestTree+testCommon mkCfg =+ let mkBox :: MessageLimit -> IO (cfg a)+ mkBox = newMessageBox . mkCfg+ in Tasty.testGroup+ ("Common Functionality for " <> show (mkCfg MessageLimit_64) <> " Note: Tests may vary the limit")+ [ testProperty "when createBox is applied to a limit > 0, then limit messages can be enqueued" $ \limit ->+ limit <= MessageLimit_128+ ==> ioProperty+ $ do+ i <- mkBox limit :: IO (cfg String)+ o <- newInput i+ results <- replicateM (messageLimitToInt limit) (deliver o "test message")+ pure $ last results,+ testProperty "when the message box is 'full' not more then one more messages can be enqueued" $+ \limit ->+ ioProperty $ do+ let nOfMessagesThatCanBeWritten = messageLimitToInt limit+ i <- mkBox limit+ o <- newInput i+ let countDeliveries n =+ timeout 100_000 (deliver o "some test message")+ >>= maybe (return n) (\ok -> if ok then countDeliveries (n + 1) else return n)+ nDelivered <- countDeliveries 0+ assertBool+ ( "at least "+ <> show nDelivered+ <> " messages should be delivered, in a queue of minimum size "+ <> show nOfMessagesThatCanBeWritten+ )+ (nDelivered >= nOfMessagesThatCanBeWritten)+ assertBool+ ( "at most "+ <> show nDelivered+ <> " messages may be delivered, in a queue of minimum size "+ <> show nOfMessagesThatCanBeWritten+ )+ (nDelivered <= 2 * nOfMessagesThatCanBeWritten),+ Tasty.testCase "When the message limit is 2, two messages can be enqueued" $ do+ i <- mkBox MessageLimit_2+ o <- newInput i+ r1 <- deliver o "Messge 1"+ r1 @?= True+ r2 <- deliver o "Messge 2"+ r2 @?= True,+ Tasty.testCase "when writing into an empty Input, True is returned" $ do+ i <- mkBox MessageLimit_32+ o <- newInput i+ deliver o "Stop the climate crisis" >>= (@?= True),+ Tasty.testCase "when writing into a full Input deliver returns False or blocks forever" $ do+ i <- mkBox MessageLimit_64+ o <- newInput i+ fix $ \next ->+ timeout 100_000 (deliver o "Stop the climate crisis") >>= \case+ Just False -> assertBool "expected 'False' when the queue is full" True+ Just True -> next+ Nothing -> return (),+ testProperty+ "when messages are send until the limit is reached,\+ \and when x, x < limit messages are received, then \+ \x more messages can be delivered again, and all \+ \messages are in FIFO order"+ $ \limit (Positive (Small x)) ->+ x < messageLimitToInt limit+ ==> withMaxSuccess 20+ $ ioProperty $+ do+ i <- mkBox limit+ -- write until the real limit is reached+ nFirstMessages <- do+ o <- newInput i+ let go msgId+ | msgId < messageLimitToInt limit = do+ success <-+ fromMaybe False+ <$> timeout+ 50_000+ (deliver o msgId)++ if success+ then go (msgId + 1)+ else return msgId+ | otherwise = return msgId+ go 0++ assertBool+ ( "internal error less than 'limit' messages could be delivered: "+ <> show nFirstMessages+ <> " < "+ <> show limit+ )+ (nFirstMessages >= messageLimitToInt limit)++ firstXReceived <- replicateM x (receive i)+ assertEqual "could not receive x messages" (Just <$> [0 .. x -1]) firstXReceived+ do+ nNextMessages <- do+ o <- newInput i+ maybe 0 length+ <$> timeout+ 5_000_000+ (mapM (deliver o) [nFirstMessages .. nFirstMessages + x - 1])+ assertEqual+ "unexpected number of messages delivered"+ x+ nNextMessages+ nextReceived <- replicateM nFirstMessages (receive i)+ assertEqual+ "could not receive x messages"+ (Just <$> [x .. x + nFirstMessages -1])+ nextReceived,+ let queueSize = MessageLimit_512+ in testContentionRobustness (mkBox queueSize :: IO (cfg (Int, Int))) (messageLimitToInt queueSize) 50+ ]++testBlockingBox :: TestTree+testBlockingBox =+ Tasty.testGroup+ "BlockingBox"+ [ Tasty.testCase "receive on an empty queue blocks forever" $ do+ i <- newMessageBox (BlockingBoxLimit MessageLimit_16)+ let foreverInMicros = 100_000+ receiveAfter i foreverInMicros+ >>= assertEqual+ "timeout expected"+ (Nothing :: Maybe Int),+ Tasty.testCase "tryReceive from an empty queue returns future that returns Nothing" $ do+ i <- newMessageBox (BlockingBoxLimit MessageLimit_16)+ f <- tryReceive i+ timeout 1_000_000 (tryNow f)+ >>= assertEqual "the future must not block and should return Nothing" (Just (Nothing @Int)),+ Tasty.testCase "when a message box is full, deliver will block forever" $ do+ i <- newMessageBox (BlockingBoxLimit MessageLimit_16)+ o <- newInput i+ let sendWhileNotFull = do+ success <- deliver o (666 :: Int)+ when success sendWhileNotFull+ timeout 1_000_000 sendWhileNotFull+ >>= assertBool "deliver must block" . isNothing,+ Tasty.testProperty+ "when sending faster than receiving all messages must be received"+ $ \queueSize -> withMaxSuccess 20 $+ ioProperty $+ do+ let nGood = 256+ tSend = 200+ tRecv = 5 * tSend++ receiverIn <- newMessageBox (BlockingBoxLimit queueSize)+ receiverOut <- newInput receiverIn+ let doReceive =+ replicateM+ nGood+ ( threadDelay tRecv+ *> receive receiverIn+ )++ let doSend =+ replicateM+ nGood+ ( threadDelay tSend+ *> deliver receiverOut ("Test Message" :: String)+ )++ (received, sendResults) <- concurrently doReceive doSend+ assertEqual "not all messages could be sent" (replicate nGood True) sendResults+ assertEqual "not all messages could be received" (replicate nGood (Just "Test Message")) received+ ]++testWaitingBox :: TestTree+testWaitingBox =+ testGroup+ "WaitingBox"+ [ Tasty.testCase "when the receive timeout is Nothing receive blocks forever" $ do+ i <- newMessageBox (WaitingBoxLimit Nothing 123 MessageLimit_16)+ let foreverInMicros = 100_000+ timeout foreverInMicros (receive i)+ >>= assertEqual+ "receive did not block"+ (Nothing :: Maybe (Maybe Int)),+ Tasty.testCase "when the receive timeout is Just x, receive returns Nothing after x micro seconds" $ do+ i <- newMessageBox (WaitingBoxLimit (Just 234) 123 MessageLimit_16)+ let foreverInMicros = 100_000+ timeout foreverInMicros (receive i)+ >>= assertEqual+ "receive did not timeout"+ (Just (Nothing :: Maybe Int)),+ Tasty.testCase+ "when the receive timeout is Just x, and another thread waits 2*x and delivers a message, receive returns Nothing"+ $ do+ let m1 = "Message 1" :: String+ mbox <- newMessageBox (WaitingBoxLimit (Just 150_000) 1_000 MessageLimit_16)+ input <- newInput mbox+ (senderOk, receiverOK) <-+ concurrently+ ( do+ threadDelay 300_000+ s1Ok <- deliver input m1+ return+ ( assertBool "delivering message failed" s1Ok+ )+ )+ ( do+ r <- receive mbox+ return $+ assertEqual "wrong message received!" Nothing r+ )+ senderOk+ receiverOK,+ Tasty.testCase+ "when the receive timeout is Just 150_000, and another thread waits 1000 and delivers a message, receive returns Just that message"+ $ do+ let m1 = "Message 1" :: String+ mbox <- newMessageBox (WaitingBoxLimit (Just 150_000) 1_000 MessageLimit_16)+ input <- newInput mbox+ (senderOk, receiverOK) <-+ concurrently+ ( do+ threadDelay 1000+ s1Ok <- deliver input m1+ return+ ( assertBool "delivering message failed" s1Ok+ )+ )+ ( do+ r <- receive mbox+ return $+ assertEqual "wrong message received!" (Just m1) r+ )+ senderOk+ receiverOK,+ Tasty.testCase "tryReceive from an empty queue returns a future that returns Nothing" $ do+ i <- newMessageBox (WaitingBoxLimit (Just 234) 123 MessageLimit_16)+ f <- tryReceive i+ timeout 1_000_000 (tryNow f)+ >>= assertEqual "the future must not block and should return Nothing" (Just (Nothing @Int)),+ Tasty.testCase "when a message box is full, deliver return False after the timeout elapsed" $ do+ i <- newMessageBox (WaitingBoxLimit (Just 234) 123 MessageLimit_16)+ o <- newInput i+ let sendWhileNotFull = do+ success <- deliver o (666 :: Int)+ when success sendWhileNotFull+ timeout 1_000_000 (sendWhileNotFull >> deliver o (666 :: Int))+ >>= assertEqual "deliver must return False after the timeout has elapsed" (Just False),+ Tasty.testProperty+ "when sending faster than receiving\+ \ and when the timeout is sufficiently large, all messages must be received"+ $ \limit -> withMaxSuccess 20 $+ ioProperty $+ do+ let nGood = 256+ tSend = 200+ tRecv = 5 * tSend++ receiverIn <- newMessageBox (WaitingBoxLimit Nothing 5_000_000 limit)+ receiverOut <- newInput receiverIn+ let doReceive =+ replicateM+ nGood+ ( threadDelay tRecv+ *> receive receiverIn+ )++ let doSend =+ replicateM+ nGood+ ( threadDelay tSend+ *> deliver receiverOut ("Test Message" :: String)+ )++ (received, sendResults) <- concurrently doReceive doSend+ assertEqual "not all messages could be sent" (replicate nGood True) sendResults+ assertEqual "not all messages could be received" (replicate nGood (Just "Test Message")) received,+ Tasty.testCase "When the message limit is 2, two messages can be enqueued" $ do+ i <- newMessageBox (WaitingBoxLimit Nothing 5_000_000 MessageLimit_2)+ o <- newInput i+ r1 <- deliver o "Messge 1"+ r1 @?= True+ r2 <- deliver o "Messge 2"+ r2 @?= True,+ Tasty.testCase "when writing into a full box, deliver returns False after the timeout" $ do+ i <- newMessageBox (WaitingBoxLimit Nothing 123 MessageLimit_32)+ o <- newInput i+ fix $ \next ->+ deliver o "Stop the climate crisis" >>= \case+ False -> return ()+ True -> next+ deliver o "foo bar"+ >>= assertBool "expect tryToDeliverAndWait to return False (timeout)" . not,+ testGroup+ "receive with/without timeout"+ [ Tasty.testCase "when a receive timeout is Just t, and no message is delivered, receive returns Nothing after the timeout" $ do+ i <- newMessageBox (WaitingBoxLimit (Just 10) 123 MessageLimit_8)+ receive i+ >>= assertEqual "timeout expected" (Nothing :: Maybe String),+ Tasty.testCase "when a receive timeout is Nothing, and no message is delivered, receive never returns" $ do+ i <- newMessageBox (WaitingBoxLimit Nothing 123 MessageLimit_8)+ let foreverInMicros = 100_000+ timeout foreverInMicros (receive i)+ >>= assertEqual "expected receive to block forever" (Nothing :: Maybe (Maybe String))+ ]+ ]++testNonBlockingBox :: TestTree+testNonBlockingBox =+ Tasty.testGroup+ "NonBlockingBox"+ [ Tasty.testCase+ "only delivery is non-blocking, when a message box has no inputs then receive blocks forever"+ $ do+ mbox <- newMessageBox (NonBlockingBoxLimit MessageLimit_16)+ r <- timeout 100_000 (receive mbox)+ assertEqual "receive must not return" (Nothing :: Maybe (Maybe String)) r,+ Tasty.testCase "when writing into a full Input deliver returns False " $ do+ i <- newMessageBox (NonBlockingBoxLimit MessageLimit_16)+ o <- newInput i+ fix $ \next ->+ deliver o "Stop the climate crisis" >>= \case+ False -> return ()+ True -> next+ deliver o "foo bar"+ >>= assertBool "deliver should not have succeeded" . not,+ Tasty.testCase+ "when using a limit smaller than half of the number\+ \ of messsages in the test, deliver first succeeds and then fails"+ $ do+ let lGood = MessageLimit_128+ nGood = messageLimitToInt lGood+ nAmbigous = 2 * nGood+ nBad = 123++ receiverOut <- newMessageBox (NonBlockingBoxLimit lGood) >>= newInput+ let doReceive = threadDelay 300_000_000++ let good = [("Test-Message GOOD", Just n) | n <- [1 .. nGood :: Int]]+ ambigous = [("Test-Message ambigous", Just n) | n <- [1 .. nAmbigous :: Int]]+ bad = [("Test-Message Base", Just n) | n <- [1 .. nBad :: Int]]++ let doSend =+ traverse (deliver receiverOut) good+ <> traverse (fmap (const True) . deliver receiverOut) ambigous+ <> traverse (deliver receiverOut) bad+ result <-+ race+ doReceive+ doSend++ result+ @?= Right+ ( replicate nGood True+ <> replicate nAmbigous True+ <> replicate nBad False+ ),+ Tasty.testCase+ "when sending 5 times faster then receiving, then deliver \+ \will begin to fail after some time"+ $ do+ let lGood = MessageLimit_128+ nGood = messageLimitToInt lGood+ nAmbigous = 2 * nGood + 1+ nBad = nGood+ tSend = 500+ tRecv = 5 * tSend++ receiverIn <- newMessageBox (NonBlockingBoxLimit lGood)+ receiverOut <- newInput receiverIn+ let doReceive = void (forever (receive receiverIn >> threadDelay tRecv))++ let !good = [("Test-Message GOOD", Just n) | !n <- [1 .. nGood :: Int]]+ !ambigous = [("Test-Message ambigous", Just n) | !n <- [1 .. nAmbigous :: Int]]+ !bad = [("Test-Message Base", Just n) | !n <- [1 .. nBad :: Int]]++ let doSend =+ traverse (\ !m -> threadDelay tSend >> deliver receiverOut m) good+ <> traverse (\ !m -> threadDelay tSend >> deliver receiverOut m >> return True) ambigous+ <> traverse (\ !m -> threadDelay tSend >> deliver receiverOut m) bad+ Right result <-+ race+ doReceive+ doSend+ let !resultGoodPart = take nGood result+ !resultBad = drop (nGood + nAmbigous) result++ assertBool "first messages succeed" (and resultGoodPart)+ assertBool "last messages fail" (not $ and resultBad)+ ]++-- This ONLY exists because I wanted the test-code coverage to be 100%+derivedInstances :: TestTree+derivedInstances =+ testGroup+ "Derived instances"+ [ -- this is just to get to 100% code coverage for+ -- even for derived instances+ testProperty+ "MessageLimit code coverage"+ (allEnumMethodsImplemented (Proxy @MessageLimit)),++ testProperty+ "MessageLimit code coverage"+ (allEqOrdShowMethodsImplemented (Proxy @MessageLimit)),++ testProperty+ "BlockingBoxLimit code coverage"+ (allEqOrdShowMethodsImplemented (Proxy @BlockingBoxLimit)),++ testProperty+ "NonBlockingBoxLimit code coverage"+ (allEqOrdShowMethodsImplemented (Proxy @NonBlockingBoxLimit)),++ testProperty+ "WaitingBoxLimit code coverage"+ (allEqOrdShowMethodsImplemented (Proxy @WaitingBoxLimit)) + ]
+ src-test/Main.hs view
@@ -0,0 +1,40 @@+module Main (main) where++import qualified CallIdTest+import qualified CatchAllTest+import qualified CommandTest+import qualified CornerCaseTests+import qualified FreshTest+import qualified LimitedMessageBoxTest+import qualified MessageBoxClassTest+import qualified ProtocolsTest+import System.Environment (setEnv)+import Test.Tasty+ ( TestTree,+ defaultIngredients,+ defaultMainWithIngredients,+ testGroup,+ )+import Test.Tasty.Runners.Html (htmlRunner)+import qualified UnlimitedMessageBoxTest++main :: IO ()+main =+ do+ setEnv "TASTY_NUM_THREADS" "1"+ defaultMainWithIngredients (htmlRunner : defaultIngredients) test++test :: TestTree+test =+ testGroup+ "Tests"+ [ CallIdTest.test,+ CatchAllTest.test,+ CornerCaseTests.test,+ CommandTest.test,+ MessageBoxClassTest.test,+ LimitedMessageBoxTest.test,+ UnlimitedMessageBoxTest.test,+ ProtocolsTest.test,+ FreshTest.test+ ]
+ src-test/MessageBoxClassTest.hs view
@@ -0,0 +1,330 @@+module MessageBoxClassTest (test) where++import Control.Monad (forM, replicateM, when)+import Data.List (sort)+import Test.Tasty as Tasty (TestTree, testGroup)+import Test.Tasty.HUnit+ ( assertBool,+ assertEqual,+ assertFailure,+ testCase,+ )+import qualified Test.Tasty.HUnit as Tasty+import UnliftIO+ ( conc,+ concurrently,+ runConc,+ timeout,+ )+import UnliftIO.Concurrent (threadDelay)+import UnliftIO.MessageBox.CatchAll+ ( CatchAllFactory (CatchAllFactory),+ )+import UnliftIO.MessageBox.Class+ ( IsInput (..),+ IsMessageBox (..),+ IsMessageBoxFactory (..),+ deliver,+ handleMessage,+ newInput,+ receive,+ )+import qualified UnliftIO.MessageBox.Limited as B+import qualified UnliftIO.MessageBox.Unlimited as U+import UnliftIO.MessageBox.Util.Future (tryNow)+import Utils+ ( NoOpBox (OnReceive),+ untilJust,+ untilM,+ )++test :: Tasty.TestTree+test =+ Tasty.testGroup+ "UnliftIO.MessageBox.Class"+ [ utilTest,+ testWith U.BlockingUnlimited,+ testWith $ CatchAllFactory U.BlockingUnlimited,+ testWith (B.BlockingBoxLimit B.MessageLimit_256),+ testWith $ CatchAllFactory (B.BlockingBoxLimit B.MessageLimit_256),+ testWith (B.BlockingBoxLimit B.MessageLimit_1),+ testWith (B.BlockingBoxLimit B.MessageLimit_2),+ testWith (B.NonBlockingBoxLimit B.MessageLimit_1),+ testWith (B.NonBlockingBoxLimit B.MessageLimit_64),+ testWith (B.WaitingBoxLimit Nothing 60_000_000 B.MessageLimit_1),+ testWith (B.WaitingBoxLimit Nothing 60_000_000 B.MessageLimit_64),+ testWith (B.WaitingBoxLimit (Just 60_000_000) 60_000_000 B.MessageLimit_64)+ ]++utilTest :: TestTree+utilTest =+ testGroup+ "Utilities"+ [ testCase+ "when receive returns Nothing, then handleMessage\+ \ does not execute the callback and returns Nothing"+ $ handleMessage (OnReceive Nothing Nothing) (const (return ()))+ >>= assertEqual "handleMessage should return Nothing" Nothing,+ testCase+ "when receive returns Just x, then handleMessage\+ \ applies the callback to x and returns Just the result"+ $ handleMessage (OnReceive Nothing (Just ("123" :: String))) return+ >>= assertEqual "handleMessage should return Just the result" (Just "123")+ ]++testWith ::+ (IsMessageBoxFactory cfg, Show cfg) =>+ cfg ->+ TestTree+testWith arg =+ Tasty.testGroup+ (show arg)+ [ commonFunctionality arg+ ]++-- standard tests+commonFunctionality ::+ (IsMessageBoxFactory cfg) =>+ cfg ->+ TestTree+commonFunctionality arg =+ let futureLoop f = tryNow f >>= maybe (threadDelay 10 >> futureLoop f) return+ in testGroup+ "Basics"+ [ testGroup+ "deliver and receive"+ [ Tasty.testCase "tryReceive from an empty queue, without any writer threads" $ do+ i <- newMessageBox arg+ f <- tryReceive i+ timeout 1_000_000 (tryNow f)+ >>= assertEqual+ "the future must not block and should return be emtpy"+ (Just (Nothing @Int)),+ testCase+ "when a process delivers a message into an input successfully, \+ \and then calls receiveAfter, the message is returned"+ $ do+ let m1 = "Message 1" :: String+ mbox <- newMessageBox arg+ input <- newInput mbox+ do+ s1Ok <- deliver input m1+ assertBool "delivering message 1 failed" s1Ok+ r <- receiveAfter mbox 200_000+ case r of+ Nothing -> assertFailure "No message received!"+ Just m -> assertEqual "wrong message received!" m1 m,+ testCase+ "when receiveAfter is called, and two different messages are delivered late but not too late, \+ \then the first message is returned"+ $ do+ let m1 = "Message 1" :: String+ m2 = "Message 2" :: String+ mbox <- newMessageBox arg+ input <- newInput mbox+ (senderOk, receiverOK) <-+ concurrently+ ( do+ threadDelay 10_000+ s1Ok <- deliver input m1+ s2Ok <- deliver input m2+ return+ ( assertBool "delivering message 1 failed" s1Ok+ >> assertBool "delivering message 2 failed" s2Ok+ )+ )+ ( do+ r <- receiveAfter mbox 200_000+ return $+ case r of+ Nothing -> assertFailure "No message received!"+ Just m -> assertEqual "wrong message received!" m1 m+ )+ senderOk+ receiverOK,+ testCase+ "when one process delivers a message into an input successfully, \+ \while another process calls receive, the message is returned"+ $ do+ let m1 = "Message 1" :: String+ mbox <- newMessageBox arg+ input <- newInput mbox+ (senderOk, receiverOK) <-+ concurrently+ ( do+ s1Ok <- deliver input m1+ return $ do+ assertBool "delivering message 1 failed" s1Ok+ )+ ( do+ r1 <- receive mbox+ return $ do+ case r1 of+ Nothing -> assertFailure "No message received!"+ Just m -> assertEqual "wrong message received!" m1 m+ )+ senderOk+ receiverOK,+ testCase+ "when one process delivers two messages into an input successfully, \+ \while another process calls receive twice, the messages are returned in FIFO order"+ $ when (maybe True (> 1) (getConfiguredMessageLimit arg)) $ do+ let m1 = "Message 1" :: String+ m2 = "Message 2" :: String+ mbox <- newMessageBox arg+ input <- newInput mbox+ (senderOk, receiverOK) <-+ concurrently+ ( do+ s1Ok <- deliver input m1+ s2Ok <- deliver input m2+ return $ do+ assertBool "delivering message 1 failed" s1Ok+ assertBool "delivering message 2 failed" s2Ok+ )+ ( do+ r1 <- receive mbox+ r2 <- receive mbox+ return $ do+ case r1 of+ Nothing -> assertFailure "No message received!"+ Just m -> assertEqual "wrong message received!" m1 m+ case r2 of+ Nothing -> assertFailure "No message received!"+ Just m -> assertEqual "wrong message received!" m2 m+ )+ senderOk+ receiverOK,+ testCase "all 113 messages of all 237 outBoxes are received by the messageBox" $ do+ let n = 113+ k = 237+ expected =+ [ ("test message: " :: String, i, j)+ | i <- [0 .. k - 1],+ j <- [0 .. n - 1]+ ]+ res <-+ timeout+ 60_000_000+ ( do+ messageBox <- newMessageBox arg+ runConc+ ( foldMap+ ( \i ->+ conc $ do+ input <- newInput messageBox+ forM+ [0 .. n - 1]+ ( \j ->+ untilM+ (deliver input ("test message: ", i, j) <* threadDelay 10)+ )+ )+ [0 .. k - 1]+ *> conc+ ( replicateM+ (n * k)+ ( untilJust+ (receive messageBox <* threadDelay 10)+ )+ )+ )+ )+ assertEqual "bad result" (Just expected) (fmap sort res)+ ],+ testGroup+ "tryReceive"+ [ testCase+ "when one process delivers one messages into an input successfully, \+ \while another process calls tryReceive, then the future returns the \+ \message, and a second tryReceives returns a future that returns Nothing"+ $ do+ let m1 = "Message 1" :: String+ mbox <- newMessageBox arg+ input <- newInput mbox+ (senderOk, receiverOK) <-+ concurrently+ ( do+ s1Ok <- deliver input m1+ return $ do+ assertBool "delivering message 1 failed" s1Ok+ )+ ( do+ f1 <- tryReceive mbox+ f2 <- tryReceive mbox+ rm1 <- futureLoop f1+ rm2 <- tryNow f2+ return $ do+ assertEqual "wrong message 1 received!" m1 rm1+ assertEqual "wrong message 2 received!" Nothing rm2+ )+ senderOk+ receiverOK,+ testCase+ "when one process delivers one messages into an input successfully, \+ \while another process calls tryReceive, then the future returns the \+ \message, and a second tryReceives returns a future that returns Nothing, \+ \then a second message is deliverd, and the second future returns that value,\+ \and the first future still returns the first value(immediately)."+ $ do+ let m1 = "Message 1" :: String+ m2 = "Message 2" :: String+ mbox <- newMessageBox arg+ input <- newInput mbox+ (senderOk, (f1, f2, receiverOK)) <-+ concurrently+ ( do+ s1Ok <- deliver input m1+ return $ do+ assertBool "delivering message 1 failed" s1Ok+ )+ ( do+ f1 <- tryReceive mbox+ f2 <- tryReceive mbox+ rm1 <- futureLoop f1+ rm2 <- tryNow f2+ return+ ( f1,+ f2,+ do+ assertEqual "wrong message 1 received!" m1 rm1+ assertEqual "wrong message 2 received!" Nothing rm2+ )+ )+ senderOk+ receiverOK+ deliver input m2 >>= assertBool "delivering message 2 failed"+ tryNow f1 >>= assertEqual "first future contains in invalid message" (Just m1)+ futureLoop f2 >>= assertEqual "second future contains in invalid message" m2,+ testCase+ "when one process delivers two messages (with random delays) into an input successfully, \+ \while another process calls tryReceive twice, then the futures each return the \+ \correct message in FIFO order"+ $ when (maybe True (> 1) (getConfiguredMessageLimit arg)) $ do+ let m1 = "Message 1" :: String+ m2 = "Message 2" :: String+ mbox <- newMessageBox arg+ input <- newInput mbox+ (senderOk, receiverOK) <-+ concurrently+ ( do+ threadDelay 20_000+ s1Ok <- deliver input m1+ threadDelay 20_000+ s2Ok <- deliver input m2+ return $ do+ assertBool "delivering message 1 failed" s1Ok+ assertBool "delivering message 2 failed" s2Ok+ )+ ( do+ f1 <- tryReceive mbox+ f2 <- tryReceive mbox+ (rm1, rm2) <- concurrently (futureLoop f1) (futureLoop f2)+ return $ do+ assertEqual "wrong message 1 received!" m1 rm1+ assertEqual "wrong message 2 received!" m2 rm2+ )+ senderOk+ receiverOK+ ]+ ]
+ src-test/MessageBoxCommon.hs view
@@ -0,0 +1,129 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}++module MessageBoxCommon (testContentionRobustness) where++import Control.Exception (ErrorCall (ErrorCall))+import Control.Monad (forM_, unless, void)+import Data.Foldable (traverse_)+import Data.Function (fix)+import Data.Functor ((<&>))+import qualified Data.IntSet as Set+import qualified Data.Map.Strict as Map+import Data.Maybe+ ( fromMaybe,+ )+import UnliftIO.MessageBox.Class+ ( IsInput (deliver),+ IsMessageBox (newInput, receive),+ )+import QCOrphans ()+import Test.Tasty as Tasty (TestTree, testGroup)+import Test.Tasty.HUnit as Tasty+ ( assertEqual,+ testCase,+ )+import UnliftIO (throwTo, timeout)+import UnliftIO.Concurrent (forkIO, myThreadId, threadDelay)++testContentionRobustness ::+ IsMessageBox msgBox =>+ IO (msgBox (Int, Set.Key)) ->+ Int ->+ Set.Key ->+ TestTree+testContentionRobustness !mkReceiverIn !nSenders !nMsgs =+ Tasty.testGroup+ "lock contention robustness"+ [ Tasty.testCase+ "when many concurrent senders send many messages to one receiver,\+ \ and when every failed deliver is retried indefinitely, \+ \ all messages will be received"+ $ do+ !receiverIn <- mkReceiverIn+ let allMessages =+ let msgIds = Set.fromList [0 .. nMsgs - 1]+ in Map.fromList [(sId, msgIds) | sId <- [0 .. nSenders - 1]]+ let doReceive = go Map.empty+ where+ go acc =+ ( timeout 2_000_000 (receive receiverIn)+ <&> fromMaybe Nothing+ )+ >>= maybe+ (return acc)+ ( \(sId, msg) ->+ -- liftIO (print (sId, msg)) >>+ go+ ( Map.alter+ ( Just+ . maybe+ (Set.singleton msg)+ (Set.insert msg)+ )+ sId+ acc+ )+ )++ receiverOut <- newInput receiverIn+ mainThread <- myThreadId+ forM_+ (Map.assocs allMessages)+ ( \(sId, msgs) ->+ void $+ forkIO $ do+ -- printf "Sender %i starting!\n" sId+ -- printf "Sender %i started!\n" sId+ traverse_+ ( \msgId ->+ timeout+ 30_000_000+ ( fix+ ( \again -> do+ ok <- deliver receiverOut (sId, msgId)+ if ok+ then return True+ else do+ threadDelay (msgId + sId)+ again+ )+ )+ >>= maybe+ ( throwTo+ mainThread+ ( ErrorCall+ ( "deliver for "+ <> show (sId, msgId)+ <> " timed out!"+ )+ )+ )+ ( `unless`+ throwTo+ mainThread+ ( ErrorCall+ ( "deliver for "+ <> show (sId, msgId)+ <> " failed!"+ )+ )+ )+ )+ (Set.toList msgs)+ -- printf "Sender %i done!\n" sId+ )+ receivedMessages <- doReceive+ assertEqual+ "failed to correctly receive all test messages"+ allMessages+ receivedMessages+ ]
+ src-test/QCOrphans.hs view
@@ -0,0 +1,30 @@+{-# OPTIONS_GHC -Wno-orphans #-}++-- | Orphan instances of QuickCheck Type Classes+module QCOrphans () where++import UnliftIO.MessageBox.Util.CallId+import UnliftIO.MessageBox.CatchAll (CatchAllFactory (..))+import qualified UnliftIO.MessageBox.Limited as Limited+import Test.QuickCheck (Arbitrary (arbitrary), elements)++instance Arbitrary Limited.MessageLimit where+ arbitrary = elements [minBound .. maxBound]++instance Arbitrary Limited.BlockingBoxLimit where+ arbitrary = Limited.BlockingBoxLimit <$> arbitrary++instance Arbitrary Limited.NonBlockingBoxLimit where+ arbitrary = Limited.NonBlockingBoxLimit <$> arbitrary++instance Arbitrary Limited.WaitingBoxLimit where+ arbitrary =+ Limited.WaitingBoxLimit <$> arbitrary+ <*> arbitrary+ <*> arbitrary++instance Arbitrary a => Arbitrary (CatchAllFactory a) where+ arbitrary = CatchAllFactory <$> arbitrary++instance Arbitrary CallId where+ arbitrary = MkCallId <$> arbitrary
+ src-test/UnlimitedMessageBoxTest.hs view
@@ -0,0 +1,21 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}++module UnlimitedMessageBoxTest (test) where++import MessageBoxCommon (testContentionRobustness)+import qualified UnliftIO.MessageBox.Class as Class+import qualified UnliftIO.MessageBox.Unlimited as Unlimited+import Test.Tasty as Tasty (TestTree, testGroup)++test :: Tasty.TestTree+test =+ Tasty.testGroup+ "UnliftIO.MessageBox.Unlimited"+ [ testContentionRobustness (Class.newMessageBox Unlimited.BlockingUnlimited) 512 50+ ]
+ src-test/Utils.hs view
@@ -0,0 +1,181 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}++module Utils+ ( untilJust,+ untilM,+ withCallIds,+ allEqOrdShowMethodsImplemented,+ allEnumMethodsImplemented,+ NoOpFactory (..),+ NoOpBox (..),+ NoOpInput (..),+ )+where++import Control.Monad.Reader (ReaderT (runReaderT))+import Data.Foldable (traverse_)+import Data.Maybe (fromMaybe)+import qualified UnliftIO.MessageBox.Util.CallId as CallId+import UnliftIO.MessageBox.Util.Fresh (CounterVar)+import UnliftIO.MessageBox.Util.Future (Future (Future))+import UnliftIO.MessageBox.Class+ ( IsInput (..),+ IsMessageBox (Input, newInput, receive, tryReceive),+ IsMessageBoxFactory (..),+ )+import Test.QuickCheck+ ( Arbitrary,+ Property,+ Testable (property),+ (.&&.),+ (.||.),+ (=/=),+ (===),+ (==>),+ )+import UnliftIO+ ( MonadIO,+ MonadUnliftIO,+ readTVarIO,+ registerDelay,+ )+import UnliftIO.Concurrent (threadDelay)++untilJust :: (Monad m) => m (Maybe a) -> m a+untilJust loopBody = do+ !res <- loopBody+ case res of+ Nothing ->+ untilJust loopBody+ Just r ->+ return r++untilM :: Monad m => m Bool -> m ()+untilM loopBody = do+ !isOk <- loopBody+ if isOk+ then return ()+ else untilM loopBody++withCallIds ::+ MonadIO m => ReaderT (CounterVar CallId.CallId) m b -> m b+withCallIds f =+ CallId.newCallIdCounter >>= runReaderT f++allEqOrdShowMethodsImplemented :: forall a proxy. (Arbitrary a, Ord a, Show a) => proxy a -> Property+allEqOrdShowMethodsImplemented _ = property $ \(x :: a) (y :: a) ->+ ( x < y+ ==> ( compare x y === LT+ .&&. compare y x === GT+ .&&. x /= y+ .&&. (y >= x)+ .&&. x < y+ .&&. (y > x)+ .&&. x <= y+ .&&. max x y === y+ .&&. min x y === x+ .&&. show x /= show y+ .&&. show (Just x) /= show (Just y)+ )+ )+ .&&. ( (x =/= y)+ .||. ( compare x y === EQ+ .&&. compare y x === EQ+ .&&. x == y+ .&&. y == x+ .&&. show x === show y+ .&&. show (Just x) === show (Just y)+ .&&. x == y+ .&&. x >= y+ .&&. y >= x+ .&&. x <= y+ .&&. y <= x+ .&&. x <= y+ .&&. x >= y+ .&&. max x y === x+ .&&. min x y === x+ )+ )++allEnumMethodsImplemented ::+ forall a proxy.+ (Show a, Ord a, Enum a, Bounded a, Arbitrary a) =>+ proxy a ->+ Property+allEnumMethodsImplemented _ =+ property $ \(x :: a) (y :: a) ->+ ( not (x > minBound && x < maxBound)+ .||. succ (pred x) === pred (succ x)+ )+ .&&. ( x /= y+ .||. ( fromEnum x === fromEnum y+ .&&. ( x >= maxBound+ .||. succ x === succ y+ )+ .&&. ( x <= minBound+ .||. pred x === pred y+ )+ )+ )+ .&&. toEnum (fromEnum x) === x+ .&&. head (enumFrom x) === x+ .&&. (x >= maxBound .||. enumFrom x !! 1 === succ x)+ .&&. ( x < y+ .||. fromEnum x >= fromEnum y+ )+ .&&. head (enumFromThen x y) === x+ .&&. if x > y+ then enumFromTo x y === []+ else+ head (enumFromTo x y)+ === x+ .&&. last (enumFromTo x y)+ === y+ .&&. ( x >= maxBound .||. enumFromThenTo x (succ x) y+ === enumFromTo x y+ )++-- message box dummy implementation++data NoOpFactory+ = NoOpFactory+ deriving stock (Show)++newtype NoOpInput a+ = OnDeliver (forall m. MonadUnliftIO m => a -> m Bool)++data NoOpBox a+ = OnReceive (Maybe Int) (Maybe a)+ deriving stock (Show)++instance IsMessageBoxFactory NoOpFactory where+ type MessageBox NoOpFactory = NoOpBox+ newMessageBox NoOpFactory = return (OnReceive Nothing Nothing)+ getConfiguredMessageLimit _ = Nothing++instance IsMessageBox NoOpBox where+ type Input NoOpBox = NoOpInput+ newInput _ = return $ OnDeliver (const (return False))+ receive (OnReceive t r) = do+ traverse_ threadDelay t+ return r+ tryReceive (OnReceive t r) = do+ timeoutVar <- traverse registerDelay t+ return+ ( Future+ ( do+ isOver <- fromMaybe True <$> traverse readTVarIO timeoutVar+ if isOver+ then return r+ else return Nothing+ )+ )++instance IsInput NoOpInput where+ deliver (OnDeliver react) m =+ react m
+ src/UnliftIO/MessageBox.hs view
@@ -0,0 +1,91 @@+-- | Fast and robust message queues for concurrent processes.+--+-- Processes of an application+-- can exchange message using __Message Boxes__.+--+-- This library is meant to be a wrapper around a+-- well tested and benchmarked subset of @unagi-chan@+-- for applications using @unliftio@.+--+-- In addition to the basic functionality, i.e.+-- _Message Boxes_, there is a very little bit of+-- type level magic dust in "UnliftIO.MessageBox.Command"+-- that helps to write code that sends a message and+-- expects the receiving process to send a reply.+--+-- This module re-exports most of the library.+module UnliftIO.MessageBox+ ( module UnliftIO.MessageBox.Class,+ module UnliftIO.MessageBox.Limited,+ module UnliftIO.MessageBox.Unlimited,+ module UnliftIO.MessageBox.CatchAll,+ module UnliftIO.MessageBox.Command,+ module UnliftIO.MessageBox.Util.CallId,+ module UnliftIO.MessageBox.Util.Fresh,+ module UnliftIO.MessageBox.Util.Future,+ )+where++import UnliftIO.MessageBox.CatchAll+ ( CatchAllBox (..),+ CatchAllFactory (..),+ CatchAllInput (..),+ )+import UnliftIO.MessageBox.Class+ ( IsInput (..),+ IsMessageBox (..),+ IsMessageBoxFactory (..),+ handleMessage,+ )+import UnliftIO.MessageBox.Command+ ( AsyncReply,+ Command,+ CommandError (..),+ DuplicateReply (..),+ Message (..),+ ReplyBox,+ ReturnType (..),+ call,+ callAsync,+ cast,+ delegateCall,+ replyTo,+ tryTakeReply,+ waitForReply,+ )+import UnliftIO.MessageBox.Limited+ ( BlockingBox (),+ BlockingBoxLimit (..),+ BlockingInput (),+ MessageLimit (..),+ NonBlockingBox (),+ NonBlockingBoxLimit (..),+ NonBlockingInput (..),+ WaitingBox (..),+ WaitingBoxLimit (..),+ WaitingInput (..),+ messageLimitToInt,+ )+import UnliftIO.MessageBox.Unlimited+ ( BlockingUnlimited (..),+ UnlimitedBox,+ UnlimitedBoxInput,+ )+import UnliftIO.MessageBox.Util.CallId+ ( CallId (..),+ HasCallIdCounter (..),+ newCallIdCounter,+ takeNext,+ )+import UnliftIO.MessageBox.Util.Fresh+ ( CounterVar,+ HasCounterVar (..),+ fresh,+ incrementAndGet,+ newCounterVar,+ )+import UnliftIO.MessageBox.Util.Future+ ( Future (..),+ awaitFuture,+ tryNow,+ )
+ src/UnliftIO/MessageBox/CatchAll.hs view
@@ -0,0 +1,95 @@+{-# LANGUAGE Strict #-}++-- | Utilities for exception safe message boxes.+--+-- This provides a wrapper around "UnliftIO.MessageBox.Class" instances+-- to catch 'SomeException' in all methods like 'deliver' and 'receive'.+module UnliftIO.MessageBox.CatchAll+ ( CatchAllFactory (..),+ CatchAllBox (..),+ CatchAllInput (..),+ )+where++import UnliftIO.MessageBox.Util.Future (Future (Future))+import UnliftIO.MessageBox.Class+ ( IsInput (..),+ IsMessageBox (..),+ IsMessageBoxFactory (..),+ )+import UnliftIO (SomeException, liftIO, try)+import UnliftIO.Concurrent (threadDelay)++-- | A wrapper around values that are instances+-- of 'IsMessageBoxFactory'. The factory wraps+-- the result of the delegated 'newMessageBox'+-- invocation into a 'CatchAllBox'.+newtype CatchAllFactory cfg = CatchAllFactory cfg+ deriving stock (Eq, Ord, Show)++-- | A wrapper around values that are instances+-- of 'IsMessageBox'.+--+-- The 'Input' type will be wrapped using+-- 'CatchAllInput'.+newtype CatchAllBox box a = CatchAllBox (box a)++-- | A wrapper around values that are instances+-- of 'IsInput'.+newtype CatchAllInput i a = CatchAllInput (i a)++instance IsMessageBoxFactory cfg => IsMessageBoxFactory (CatchAllFactory cfg) where+ type MessageBox (CatchAllFactory cfg) = CatchAllBox (MessageBox cfg)+ {-# INLINE newMessageBox #-}+ newMessageBox (CatchAllFactory !cfg) = CatchAllBox <$> newMessageBox cfg+ getConfiguredMessageLimit (CatchAllFactory !cfg) =+ getConfiguredMessageLimit cfg++instance IsMessageBox box => IsMessageBox (CatchAllBox box) where+ type Input (CatchAllBox box) = CatchAllInput (Input box)+ {-# INLINE newInput #-}+ newInput (CatchAllBox !b) =+ CatchAllInput <$> newInput b+ {-# INLINE receive #-}+ receive (CatchAllBox !box) =+ try @_ @SomeException+ (receive box)+ >>= \case+ Left _e -> liftIO (print _e) >> return Nothing+ Right r -> return r+ {-# INLINE receiveAfter #-}+ -- | Call the wrapped 'receiveAfter' and catch all sync exceptions.+ -- + -- When an exception is caught return 'Nothing'.+ receiveAfter (CatchAllBox !box) !t =+ try @_ @SomeException+ (receiveAfter box t)+ >>= \case+ Left _e -> liftIO (print _e) >> pure Nothing+ Right r -> return r+ {-# INLINE tryReceive #-}+ tryReceive (CatchAllBox !box) =+ try @_ @SomeException+ (tryReceive box)+ >>= \case+ Left _e ->+ liftIO (print _e)+ >> return+ ( Future+ ( do+ -- suspense...+ threadDelay 1000+ -- ... anyway, the truth is: there is no spoon.+ return Nothing+ )+ )+ Right r -> return r++instance (IsInput i) => IsInput (CatchAllInput i) where+ {-# INLINE deliver #-}+ deliver (CatchAllInput !i) !msg =+ try @_ @SomeException+ (deliver i msg)+ >>= \case+ Left _e -> liftIO (print _e) >> return False+ Right r -> return r
+ src/UnliftIO/MessageBox/Class.hs view
@@ -0,0 +1,119 @@+-- | This module contains a type class that+-- describes exchangable operations on messages+-- boxes.+module UnliftIO.MessageBox.Class+ ( IsMessageBoxFactory (..),+ IsMessageBox (..),+ IsInput (..),+ handleMessage,+ )+where++import Data.Kind (Type)+import UnliftIO.MessageBox.Util.Future (Future, awaitFuture)+import UnliftIO (MonadUnliftIO, timeout)++-- | Create 'IsMessageBox' instances from a parameter.+-- Types that determine 'MessageBox' values.+--+-- For a limited message box this might be the limit of+-- the message queue.+class+ (IsMessageBox (MessageBox cfg), IsInput (Input (MessageBox cfg))) =>+ IsMessageBoxFactory cfg+ where+ type MessageBox cfg :: Type -> Type++ -- | Return a message limit.+ --+ -- NOTE: This method was added for unit tests.+ -- Although the method is totally valid, it+ -- might not be super useful in production code.+ -- Also note that the naming follows the rule:+ -- Reserve short names for entities that are+ -- used often.+ getConfiguredMessageLimit :: cfg -> Maybe Int++ -- | Create a new @msgBox@.+ -- This is required to receive a message.+ -- NOTE: Only one process may receive on an msgBox.+ newMessageBox :: MonadUnliftIO m => cfg -> m (MessageBox cfg a)++-- | A type class for msgBox types.+-- A common interface for receiving messages.+class IsInput (Input msgBox) => IsMessageBox msgBox where+ -- | Type of the corresponding input+ type Input msgBox :: Type -> Type++ -- | Receive a message. Take whatever time it takes.+ -- Return 'Just' the value or 'Nothing' when an error+ -- occurred.+ --+ -- NOTE: Nothing may sporadically be returned, especially+ -- when there is a lot of load, so please make sure to + -- build your application in such a way, that it + -- anticipates failure.+ receive :: MonadUnliftIO m => msgBox a -> m (Maybe a)++ -- | Return a 'Future' that can be used to wait for the+ -- arrival of the next message.+ -- NOTE: Each future value represents the next slot in the queue+ -- so one future corresponds to exactly that message (should it arrive)+ -- and if that future value is dropped, that message will be lost!+ tryReceive :: MonadUnliftIO m => msgBox a -> m (Future a)++ -- | Wait for an incoming message or return Nothing.+ --+ -- The default implementation uses 'tryReceive' to get a+ -- 'Future' on which 'awaitFuture' inside a 'timeout' is called.+ --+ -- Instances might override this with more performant implementations+ -- especially non-blocking Unagi channel based implementation.+ --+ -- NOTE: Nothing may sporadically be returned, especially+ -- when there is a lot of load, so please make sure to + -- build your application in such a way, that it + -- anticipates failure.+ receiveAfter ::+ MonadUnliftIO m =>+ -- | Message box+ msgBox a ->+ -- | Time in micro seconds to wait until the+ -- action is invoked.+ Int ->+ m (Maybe a)+ receiveAfter !mbox !t =+ tryReceive mbox >>= timeout t . awaitFuture++ -- | Create a new @input@ that enqueus messages,+ -- which are received by the @msgBox@+ newInput :: MonadUnliftIO m => msgBox a -> m (Input msgBox a)++-- | A type class for input types.+-- A common interface for delivering messages.+class IsInput input where+ -- | Send a message. Take whatever time it takes.+ -- Depending on the implementation, this might+ -- be a non-blocking operation.+ -- Return if the operation was successful.+ --+ -- NOTE: @False@ may sporadically be returned, especially+ -- when there is a lot of load, so please make sure to + -- build your application in such a way, that it + -- anticipates failure.+ deliver :: MonadUnliftIO m => input a -> a -> m Bool++-- ** Utility Functions for Receiving Messages++-- | Receive a message and apply a function to it.+handleMessage ::+ (MonadUnliftIO m, IsMessageBox msgBox) =>+ msgBox message ->+ (message -> m b) ->+ m (Maybe b)+handleMessage !msgBox !onMessage = do+ !maybeMessage <- receive msgBox+ case maybeMessage of+ Nothing -> pure Nothing+ Just !message -> do+ Just <$> onMessage message
+ src/UnliftIO/MessageBox/Command.hs view
@@ -0,0 +1,327 @@+{-# LANGUAGE StrictData #-}++-- | Abstractions for the definition of+-- 'Command' 'Messages', that flow between+module UnliftIO.MessageBox.Command+ ( Message (..),+ Command,+ ReturnType (..),+ ReplyBox (),+ CommandError (..),+ DuplicateReply (..),+ cast,+ call,+ replyTo,+ callAsync,+ delegateCall,+ AsyncReply (),+ waitForReply,+ tryTakeReply,+ )+where++import Control.Applicative (Alternative ((<|>)))+import Control.Monad (unless)+import Control.Monad.Reader (MonadReader)+import Data.Kind (Type)+import UnliftIO.MessageBox.Util.CallId+ ( CallId (),+ HasCallIdCounter,+ )+import qualified UnliftIO.MessageBox.Util.CallId as CallId+import qualified UnliftIO.MessageBox.Class as MessageBox+import UnliftIO+ ( Exception,+ MonadUnliftIO,+ TMVar,+ Typeable,+ atomically,+ checkSTM,+ newEmptyTMVarIO,+ readTMVar,+ readTVar,+ registerDelay,+ takeTMVar,+ throwIO,+ tryPutTMVar,+ tryReadTMVar,+ )++-- | This family allows to encode imperative /commands/.+--+-- The clauses of a 'Command' define the commands that+-- a process should execute.+--+-- Every clause may specify an individual 'ReturnType' that+-- declares if and what response is valid for a message.+--+-- For example:+--+-- >+-- > type LampId = Int+-- >+-- > data instance Command LightControl r where+-- > GetLamps :: Command LigthControl (Return [LampId])+-- > SwitchOn :: LampId -> Command LigthControl FireAndForget+-- >+-- > data LightControl -- the phantom type+-- >+--+-- The type index of the Command family is the uninhabited+-- @LightControl@ type.+-- .+--+-- The second type parameter indicates if a message requires the+-- receiver to send a reply back to the blocked and waiting+-- sender, or if no reply is necessary.+data family Command apiTag :: ReturnType -> Type++-- | Indicates if a 'Command' requires the+-- receiver to send a reply or not.+data ReturnType where+ -- | Indicates that a 'Command' value is sent _one-way_.+ --+ -- Values of a 'Command' instance with 'FireAndForget' as second+ -- parameter indicate that the sender should not expect any direct+ -- answer from the recepient.+ FireAndForget :: ReturnType+ -- | Indicates that a 'Command' value requires the receiver+ -- to send a reply of the given type.+ --+ -- Values of a 'Command' instance with 'Return' as second parameter+ -- are received wrapped into a 'Blocking'.+ Return :: Type -> ReturnType++-- | A message valid for some user defined @apiTag@.+--+-- The @apiTag@ tag (phantom-) type defines the+-- messages allowed here, declared by the instance of+-- 'Command' for 'apiTag'.+data Message apiTag where+ -- | Wraps a 'Command' with a 'ReturnType' of 'Return' @result@.+ --+ -- Such a message can formed by using 'call'.+ --+ -- A 'Blocking' contains a 'ReplyBox' that can be+ -- used to send the reply to the other process+ -- blocking on 'call'+ Blocking ::+ Show (Command apiTag ( 'Return result)) =>+ Command apiTag ( 'Return result) ->+ ReplyBox result ->+ Message apiTag+ -- | If the 'Command' has a 'ReturnType' of 'FireAndForget'+ -- it has fire-and-forget semantics.+ --+ -- The smart constructor 'cast' can be used to+ -- this message.+ NonBlocking ::+ (Show (Command apiTag 'FireAndForget)) =>+ Command apiTag 'FireAndForget ->+ Message apiTag++instance Show (Message apiTag) where+ showsPrec d (NonBlocking !m) =+ showParen (d >= 9) (showString "NB: " . showsPrec 9 m)+ showsPrec d (Blocking !m (MkReplyBox _ !callId)) =+ showParen (d >= 9) (showString "B: " . showsPrec 9 m . showChar ' ' . shows callId)++-- | This is like 'Input', it can be used+-- by the receiver of a 'Blocking'+-- to either send a reply using 'reply'+-- or to fail/abort the request using 'sendRequestError'+data ReplyBox a+ = MkReplyBox+ !(TMVar (InternalReply a))+ !CallId++-- | This is the reply to a 'Blocking' sent through the 'ReplyBox'.+type InternalReply a = (Either CommandError a)++-- | The failures that the receiver of a 'Return' 'Command', i.e. a 'Blocking',+-- can communicate to the /caller/, in order to indicate that+-- processing a request did not or will not lead to the result the+-- caller is blocked waiting for.+data CommandError where+ -- | Failed to enqueue a 'Blocking' 'Command' 'Message' into the corresponding+ -- 'MessageBox.Input'+ CouldNotEnqueueCommand :: !CallId -> CommandError+ -- | The request has failed /for reasons/.+ BlockingCommandFailure :: !CallId -> CommandError+ -- | Timeout waiting for the result.+ BlockingCommandTimedOut :: !CallId -> CommandError+ deriving stock (Show, Eq)++-- | Enqueue a 'NonBlocking' 'Message' into an 'Input'.+-- This is just for symetry to 'call', this is+-- equivalent to: @\input -> MessageBox.tryToDeliver input . NonBlocking@+--+-- The+{-# INLINE cast #-}+cast ::+ ( MonadUnliftIO m,+ MessageBox.IsInput o,+ Show (Command apiTag 'FireAndForget)+ ) =>+ o (Message apiTag) ->+ Command apiTag 'FireAndForget ->+ m Bool+cast input !msg =+ MessageBox.deliver input (NonBlocking msg)++-- | Enqueue a 'Blocking' 'Message' into an 'MessageBox.IsInput' and wait for the+-- response.+--+-- If message 'deliver'y failed, return @Left 'CouldNotEnqueueCommand'@.+--+-- If no reply was given by the receiving process (using 'replyTo') within+-- a given duration, return @Left 'BlockingCommandTimedOut'@.+--+-- Important: The given timeout starts __after__ 'deliver' has returned,+-- if 'deliver' blocks and delays, 'call' might take longer than the+-- specified timeout.+--+-- The receiving process can either delegate the call using+-- 'delegateCall' or reply to the call by using: 'replyTo'.+call ::+ ( HasCallIdCounter env,+ MonadReader env m,+ MonadUnliftIO m,+ MessageBox.IsInput input,+ Show (Command apiTag ( 'Return result))+ ) =>+ input (Message apiTag) ->+ Command apiTag ( 'Return result) ->+ Int ->+ m (Either CommandError result)+call !input !pdu !timeoutMicroseconds = do+ !callId <- CallId.takeNext+ !resultVar <- newEmptyTMVarIO+ !sendSuccessful <- do+ let !rbox = MkReplyBox resultVar callId+ let !msg = Blocking pdu rbox+ MessageBox.deliver input msg+ if not sendSuccessful+ then return (Left (CouldNotEnqueueCommand callId))+ else do+ timedOutVar <- registerDelay timeoutMicroseconds+ atomically $+ takeTMVar resultVar+ <|> ( do+ readTVar timedOutVar >>= checkSTM+ return (Left (BlockingCommandTimedOut callId))+ )++-- | This is called from the callback contained in the 'Blocking' 'Message'.+--+-- When handling a 'Blocking' 'Message' the 'ReplyBox' contained+-- in the message contains the 'TMVar' for the result, and this+-- function puts the result into it.+{-# INLINE replyTo #-}+replyTo :: (MonadUnliftIO m) => ReplyBox a -> a -> m ()+replyTo (MkReplyBox !replyBox !callId) !message =+ atomically (tryPutTMVar replyBox (Right message))+ >>= \success -> unless success (throwIO (DuplicateReply callId))++-- | Exception thrown by 'replyTo' when 'replyTo' is call more than once.+newtype DuplicateReply = DuplicateReply CallId deriving stock (Eq)++instance Show DuplicateReply where+ showsPrec d (DuplicateReply !callId) =+ showParen (d >= 9) (showString "more than one reply sent for: " . shows callId)++instance Exception DuplicateReply++-- | Pass on the call to another process.+--+-- Used to implement dispatcher processes.+--+-- Returns 'True' if the 'MessageBox.deliver' operation was+-- successful.+{-# INLINE delegateCall #-}+delegateCall ::+ ( MonadUnliftIO m,+ MessageBox.IsInput o,+ Show (Command apiTag ( 'Return r))+ ) =>+ o (Message apiTag) ->+ Command apiTag ( 'Return r) ->+ ReplyBox r ->+ m Bool+delegateCall !o !c !r =+ MessageBox.deliver o (Blocking c r)++-- ** Non-Blocking call API++-- | Enqueue a 'Blocking' 'Message' into an 'MessageBox.IsInput'.+--+-- If the call to 'deliver' fails, return @Nothing@ otherwise+-- @Just@ the 'AsyncReply'.+--+-- The receiving process must use 'replyTo' with the 'ReplyBox'+-- received along side the 'Command' in the 'Blocking'.+callAsync ::+ ( HasCallIdCounter env,+ MonadReader env m,+ MonadUnliftIO m,+ MessageBox.IsInput o,+ Show (Command apiTag ( 'Return result))+ ) =>+ o (Message apiTag) ->+ Command apiTag ( 'Return result) ->+ m (Maybe (AsyncReply result))+callAsync !input !pdu = do+ !callId <- CallId.takeNext+ !resultVar <- newEmptyTMVarIO+ !sendSuccessful <- do+ let !rbox = MkReplyBox resultVar callId+ let !msg = Blocking pdu rbox+ MessageBox.deliver input msg+ if sendSuccessful+ then return (Just (MkAsyncReply callId resultVar))+ else return Nothing++-- | The result of 'callAsync'.+-- Use 'waitForReply' or 'tryTakeReply'.+data AsyncReply r+ = MkAsyncReply !CallId !(TMVar (InternalReply r))++instance (Typeable r) => Show (AsyncReply r) where+ showsPrec !d (MkAsyncReply !cId _) =+ showParen (d >= 9) (showString "AR: " . shows cId)++-- | Wait for the reply of a 'Blocking' 'Message'+-- sent by 'callAsync'.+{-# INLINE waitForReply #-}+waitForReply ::+ MonadUnliftIO m =>+ -- | The time in micro seconds to wait+ -- before returning 'Left' 'BlockingCommandTimedOut'+ Int ->+ AsyncReply result ->+ m (Either CommandError result)+waitForReply !t (MkAsyncReply !cId !rVar) = do+ !delay <- registerDelay t+ atomically+ ( ( do+ !hasTimedOut <- readTVar delay+ checkSTM hasTimedOut+ return (Left (BlockingCommandTimedOut cId))+ )+ <|> readTMVar rVar+ )++-- | If a reply for an 'callAsync' operation is available+-- return it, otherwise return 'Nothing'.+{-# INLINE tryTakeReply #-}+tryTakeReply ::+ MonadUnliftIO m =>+ AsyncReply result ->+ m (Maybe (Either CommandError result))+tryTakeReply (MkAsyncReply _expectedCallId !resultVar) = do+ !maybeTheResult <- atomically (tryReadTMVar resultVar)+ case maybeTheResult of+ Nothing ->+ return Nothing+ Just !result ->+ return (Just result)
+ src/UnliftIO/MessageBox/Limited.hs view
@@ -0,0 +1,339 @@+-- | Thread safe queues for uni directional message passing+-- between threads.+--+-- This message box has an upper limit, that means that+-- sometimes delivery either fails or is blocked until+-- the receiving thread has consumed more messages.+--+-- Use this module if the producer(s) outperform the consumer,+-- but you want the extra safety that the queue blocks the+-- 'Input' after a certain message limit is reached.+--+-- If you are sure that the producers fire at a slower rate+-- then the rate at which the consumer consumes messages, use this+-- module.+module UnliftIO.MessageBox.Limited+ ( MessageLimit (..),+ messageLimitToInt,+ BlockingBoxLimit (..),+ BlockingBox (),+ BlockingInput (),+ NonBlockingBoxLimit (..),+ NonBlockingBox (),+ NonBlockingInput (..),+ WaitingBoxLimit (..),+ WaitingBox (..),+ WaitingInput (..),+ )+where++import qualified Control.Concurrent.Chan.Unagi.Bounded as Unagi+import Control.Monad (unless)+import Data.Functor (($>))+import Data.Maybe (fromMaybe)+import UnliftIO.MessageBox.Util.Future (Future (..))+import qualified UnliftIO.MessageBox.Class as Class+import UnliftIO+ ( MonadIO (liftIO),+ MonadUnliftIO,+ timeout,+ )+import UnliftIO.Concurrent (threadDelay)++-- | Message Limit+--+-- The message limit must be a reasonable small positive integer+-- that is also a power of two. This stems from the fact that+-- Unagi is used under the hood.+--+-- The limit is a lower bound.+data MessageLimit+ = MessageLimit_1+ | MessageLimit_2+ | MessageLimit_4+ | MessageLimit_8+ | MessageLimit_16+ | MessageLimit_32+ | MessageLimit_64+ | MessageLimit_128+ | MessageLimit_256+ | MessageLimit_512+ | MessageLimit_1024+ | MessageLimit_2048+ | MessageLimit_4096+ deriving stock+ (Eq, Ord, Show, Bounded, Enum)++-- | Convert a 'MessageLimit' to the+-- 'Int' representation.+{-# INLINE messageLimitToInt #-}+messageLimitToInt :: MessageLimit -> Int+messageLimitToInt =+ \case+ MessageLimit_1 -> 1+ MessageLimit_2 -> 2+ MessageLimit_4 -> 4+ MessageLimit_8 -> 8+ MessageLimit_16 -> 16+ MessageLimit_32 -> 32+ MessageLimit_64 -> 64+ MessageLimit_128 -> 128+ MessageLimit_256 -> 256+ MessageLimit_512 -> 512+ MessageLimit_1024 -> 1024+ MessageLimit_2048 -> 2048+ MessageLimit_4096 -> 4096++-- * 'Class.IsMessageBoxFactory' instances++-- ** Blocking++-- | Contains the (vague) limit of messages that a 'BlockingBox'+-- can buffer, i.e. that 'deliver' can put into a 'BlockingInput'+-- of a 'BlockingBox'.+newtype BlockingBoxLimit = BlockingBoxLimit MessageLimit+ deriving stock (Eq, Ord)++instance Show BlockingBoxLimit where+ showsPrec _ (BlockingBoxLimit !l) =+ showString "Blocking" . showsPrec 9 (messageLimitToInt l)++-- | A message queue out of which messages can by 'receive'd.+--+-- This is the counter part of 'Input'. Can be used for reading+-- messages.+--+-- Messages can be received by 'receive' or 'tryReceive'.+data BlockingBox a+ = MkBlockingBox+ !(Unagi.InChan a)+ !(Unagi.OutChan a)++-- | A message queue into which messages can be enqued by,+-- e.g. 'tryToDeliver'.+-- Messages can be received from an 'BlockingBox`.+--+-- The 'Input' is the counter part of a 'BlockingBox'.+newtype BlockingInput a = MkBlockingInput (Unagi.InChan a)++instance Class.IsMessageBoxFactory BlockingBoxLimit where+ type MessageBox BlockingBoxLimit = BlockingBox+ {-# INLINE newMessageBox #-}+ newMessageBox (BlockingBoxLimit !limit) = create limit+ getConfiguredMessageLimit (BlockingBoxLimit !limit) =+ Just (messageLimitToInt limit)++-- | A blocking instance that invokes 'receive'.+instance Class.IsMessageBox BlockingBox where+ type Input BlockingBox = BlockingInput++ {-# INLINE receive #-}+ receive !i = Just <$> receive i+ {-# INLINE tryReceive #-}+ tryReceive !i = tryReceive i+ {-# INLINE newInput #-}+ newInput !i = newInput i+ receiveAfter (MkBlockingBox _ !s) !rto =+ do+ (!promise, !blocker) <- liftIO (Unagi.tryReadChan s)+ liftIO (Unagi.tryRead promise)+ >>= maybe+ (timeout rto (liftIO blocker))+ (return . Just)++-- | A blocking instance that invokes 'deliver'.+instance Class.IsInput BlockingInput where+ {-# INLINE deliver #-}+ deliver !o !a = deliver o a $> True++-- ** A wrapper around 'BlockingBox' for Non-Blocking Input (NBI)++-- | A 'BlockingBoxLimit' wrapper for non-blocking 'Class.IsMessageBoxFactory' instances.+newtype NonBlockingBoxLimit = NonBlockingBoxLimit MessageLimit+ deriving stock (Eq, Ord)++instance Show NonBlockingBoxLimit where+ showsPrec _ (NonBlockingBoxLimit !l) =+ showString "NonBlocking" . showsPrec 9 (messageLimitToInt l)++instance Class.IsMessageBoxFactory NonBlockingBoxLimit where+ type MessageBox NonBlockingBoxLimit = NonBlockingBox+ {-# INLINE newMessageBox #-}+ newMessageBox (NonBlockingBoxLimit !l) =+ NonBlockingBox <$> Class.newMessageBox (BlockingBoxLimit l)+ getConfiguredMessageLimit (NonBlockingBoxLimit !limit) =+ Just (messageLimitToInt limit)++-- | A 'BlockingBox' wrapper for non-blocking 'Class.IsMessageBox' instances.+--+-- The difference to the 'BlockingBox' instance is that 'Class.deliver'+-- immediately returns if the message box limit is surpassed.+newtype NonBlockingBox a = NonBlockingBox (BlockingBox a)++instance Class.IsMessageBox NonBlockingBox where+ type Input NonBlockingBox = NonBlockingInput+ {-# INLINE receive #-}+ receive (NonBlockingBox !i) = Just <$> receive i+ {-# INLINE tryReceive #-}+ tryReceive (NonBlockingBox !i) = tryReceive i+ {-# INLINE receiveAfter #-}+ receiveAfter (NonBlockingBox !b) !rto =+ Class.receiveAfter b rto+ {-# INLINE newInput #-}+ newInput (NonBlockingBox !i) = NonBlockingInput <$> newInput i++-- | A wrapper around 'BlockingInput' with a non-blocking 'Class.IsInput' instance.+--+-- 'deliver' will enqueue the message or return 'False' immediately,+-- if the message box already contains more messages than+-- it's limit allows.+newtype NonBlockingInput a = NonBlockingInput (BlockingInput a)++instance Class.IsInput NonBlockingInput where+ {-# INLINE deliver #-}+ deliver (NonBlockingInput !o) !a = do+ !res <- tryToDeliver o a+ unless res (threadDelay 10)+ return res++-- ** 'BlockingBox' Wrapper with Timeout++-- | A 'Class.IsMessageBoxFactory' instance wrapping the 'BlockingBox'+-- with independently configurable timeouts for 'receive' and 'deliver'.+data WaitingBoxLimit+ = WaitingBoxLimit+ !(Maybe Int)+ !Int+ !MessageLimit+ deriving stock (Eq, Ord)++instance Show WaitingBoxLimit where+ showsPrec _ (WaitingBoxLimit !t0 !t1 !l) =+ showString "Waiting_"+ . ( case t0 of+ Nothing -> id+ Just !t -> showsPrec 9 t . showChar '_'+ )+ . showsPrec 9 t1+ . showChar '_'+ . showsPrec 9 (messageLimitToInt l)++instance Class.IsMessageBoxFactory WaitingBoxLimit where+ type MessageBox WaitingBoxLimit = WaitingBox+ {-# INLINE newMessageBox #-}+ newMessageBox l@(WaitingBoxLimit _ _ !c) =+ WaitingBox l <$> Class.newMessageBox (BlockingBoxLimit c)+ getConfiguredMessageLimit (WaitingBoxLimit _ _ !limit) =+ Just (messageLimitToInt limit)++-- | A 'BlockingBox' an a 'WaitingBoxLimit' for+-- the 'Class.IsMessageBox' instance.+data WaitingBox a+ = WaitingBox WaitingBoxLimit (BlockingBox a)++instance Class.IsMessageBox WaitingBox where+ type Input WaitingBox = WaitingInput+ {-# INLINE receive #-}+ receive (WaitingBox (WaitingBoxLimit (Just !rto) _ _) (MkBlockingBox _ !s)) =+ liftIO $ do+ (!promise, !blocker) <- Unagi.tryReadChan s+ Unagi.tryRead promise+ >>= maybe+ (timeout rto blocker)+ (return . Just)+ receive (WaitingBox !_ !m) =+ Class.receive m+ {-# INLINE receiveAfter #-}+ receiveAfter (WaitingBox _ !b) !rto =+ Class.receiveAfter b rto+ {-# INLINE tryReceive #-}+ tryReceive (WaitingBox _ !m) = tryReceive m+ {-# INLINE newInput #-}+ newInput (WaitingBox (WaitingBoxLimit _ !dto _) !m) =+ WaitingInput dto <$> newInput m++-- | An input for a 'BlockingBox' that will block+-- for not much more than the given timeout when+-- the message box is full.+data WaitingInput a+ = WaitingInput+ !Int+ !(BlockingInput a)++instance Class.IsInput WaitingInput where+ {-# INLINE deliver #-}+ deliver (WaitingInput !t !o) !a = tryToDeliverAndWait t o a++-- Internal Functions++{-# INLINE create #-}+create :: MonadUnliftIO m => MessageLimit -> m (BlockingBox a)+create !limit = do+ (!inChan, !outChan) <- liftIO (Unagi.newChan (messageLimitToInt limit))+ return $! MkBlockingBox inChan outChan++{-# INLINE receive #-}+receive :: MonadUnliftIO m => BlockingBox a -> m a+receive (MkBlockingBox _ !s) =+ liftIO (Unagi.readChan s)++-- | Return a 'Future' for the next value that will be received.+{-# INLINE tryReceive #-}+tryReceive :: MonadUnliftIO m => BlockingBox a -> m (Future a)+tryReceive (MkBlockingBox _ !s) = liftIO $ do+ (!promise, _) <- Unagi.tryReadChan s+ return (Future (Unagi.tryRead promise))++{-# INLINE newInput #-}+newInput :: MonadUnliftIO m => BlockingBox a -> m (BlockingInput a)+newInput (MkBlockingBox !s _) = return $! MkBlockingInput s++{-# INLINE deliver #-}+deliver :: MonadUnliftIO m => BlockingInput a -> a -> m ()+deliver (MkBlockingInput !s) !a =+ liftIO $ Unagi.writeChan s a++-- | Try to put a message into the 'BlockingInput'+-- of a 'MessageBox', such that the process+-- reading the 'MessageBox' receives the message.+--+-- If the 'MessageBox' is full return False.+{-# INLINE tryToDeliver #-}+tryToDeliver :: MonadUnliftIO m => BlockingInput a -> a -> m Bool+tryToDeliver (MkBlockingInput !s) !a =+ liftIO $ Unagi.tryWriteChan s a++-- | Send a message by putting it into the 'BlockingInput'+-- of a 'MessageBox', such that the process+-- reading the 'MessageBox' receives the message.+--+-- Return False if the+-- 'MessageBox' has been closed or is full.+--+-- This assumes that the queue is likely empty, and+-- tries 'tryToDeliver' first before wasting any+-- precious cpu cycles entering 'timeout'.+tryToDeliverAndWait ::+ MonadUnliftIO m =>+ Int ->+ BlockingInput a ->+ a ->+ m Bool+tryToDeliverAndWait !t !o !a =+ -- Benchmarks have shown great improvements+ -- when calling tryToDeliver once before doing+ -- deliver in a System.Timeout.timeout;+ --+ -- We even tried calling 'tryToDeliver' more than once,+ -- but that did not lead to convinving improvements.+ --+ -- Benachmarks have also shown, that sending pessimistically+ -- (i.e. avoiding `tryToDeliver`) does not improve performance,+ -- even when the message queue is congested+ --+ -- See benchmark results:+ -- `benchmark-results/optimistic-vs-pessimistic.html`+ tryToDeliver o a >>= \case+ True -> return True+ False ->+ fromMaybe False <$> timeout t (deliver o a $> True)
+ src/UnliftIO/MessageBox/Unlimited.hs view
@@ -0,0 +1,117 @@+-- | Thread safe queues for message passing+-- between many concurrent processes.+--+-- This message box is __UNLIMITED__.+--+-- Good single producer/single consumer performance+--+-- If you are sure that the producer(s) send messages+-- at a lower rate than the rate at which the consumer+-- consumes messages, use this module.+--+-- Otherwise use the more conservative+-- "UnliftIO.MessageBox.Limited" module.+module UnliftIO.MessageBox.Unlimited+ ( BlockingUnlimited (..),+ UnlimitedBox (),+ UnlimitedBoxInput (),+ )+where++-- import qualified Control.Concurrent.Chan.Unagi.NoBlocking as Unagi+import qualified Control.Concurrent.Chan.Unagi as Unagi+import Data.Functor (($>))+import UnliftIO.MessageBox.Util.Future (Future (..))+import qualified UnliftIO.MessageBox.Class as Class+import UnliftIO+ ( MonadIO (liftIO),+ MonadUnliftIO,+ )++-- | A message queue out of which messages can+-- by 'receive'd.+--+-- This is the counter part of 'Input'. Can be+-- used for reading messages.+--+-- Messages can be received by 'receive' or 'tryReceive'.+data UnlimitedBox a+ = MkUnlimitedBox+ !(Unagi.InChan a)+ !(Unagi.OutChan a)++-- | A message queue into which messages can be enqued by,+-- e.g. 'deliver'.+-- Messages can be received from an 'UnlimitedBox`.+--+-- The 'UnlimitedBoxInput' is the counter part of a 'UnlimitedBox'.+newtype UnlimitedBoxInput a = MkUnlimitedBoxInput (Unagi.InChan a)++-- | The (empty) configuration for creating+-- 'UnlimitedBox'es using the 'Class.IsMessageBoxFactory' methods.+data BlockingUnlimited = BlockingUnlimited++instance Show BlockingUnlimited where+ showsPrec _ _ = showString "Unlimited"++instance Class.IsMessageBoxFactory BlockingUnlimited where+ type MessageBox BlockingUnlimited = UnlimitedBox+ {-# INLINE newMessageBox #-}+ newMessageBox BlockingUnlimited = create+ getConfiguredMessageLimit _ = Nothing ++-- | A blocking instance that invokes 'receive'.+instance Class.IsMessageBox UnlimitedBox where+ type Input UnlimitedBox = UnlimitedBoxInput+ {-# INLINE receive #-}+ receive !i = Just <$> receive i+ {-# INLINE tryReceive #-}+ tryReceive !i = tryReceive i+ {-# INLINE newInput #-}+ newInput !i = newInput i++-- | A blocking instance that invokes 'deliver'.+instance Class.IsInput UnlimitedBoxInput where+ {-# INLINE deliver #-}+ deliver !o !m = deliver o m $> True+++-- | Create a 'MessageBox'.+--+-- From a 'MessageBox' a corresponding 'Input' can+-- be made, that can be passed to some potential+-- communication partners.+{-# INLINE create #-}+create :: MonadUnliftIO m => m (UnlimitedBox a)+create = do+ (!inChan, !outChan) <- liftIO Unagi.newChan+ return $! MkUnlimitedBox inChan outChan++-- | Wait for and receive a message from a 'MessageBox'.+{-# INLINE receive #-}+receive :: MonadUnliftIO m => UnlimitedBox a -> m a+receive (MkUnlimitedBox _ !s) =+ --liftIO (Unagi.readChan IO.yield s)+ liftIO (Unagi.readChan s)++-- | Try to receive a message from a 'MessageBox',+-- return @Nothing@ if the queue is empty.+{-# INLINE tryReceive #-}+tryReceive :: MonadUnliftIO m => UnlimitedBox a -> m (Future a)+tryReceive (MkUnlimitedBox _ !s) = liftIO $ do+ (!promise, _) <- Unagi.tryReadChan s+ return (Future (Unagi.tryRead promise))++-- | Create an 'Input' to write the items+-- that the given 'MessageBox' receives.+{-# INLINE newInput #-}+newInput :: MonadUnliftIO m => UnlimitedBox a -> m (UnlimitedBoxInput a)+newInput (MkUnlimitedBox !s _) = return $! MkUnlimitedBoxInput s++-- | Put a message into the 'Input'+-- of a 'MessageBox', such that the process+-- reading the 'MessageBox' receives the message.+{-# INLINE deliver #-}+deliver :: MonadUnliftIO m => UnlimitedBoxInput a -> a -> m ()+deliver (MkUnlimitedBoxInput !s) !a =+ liftIO $ Unagi.writeChan s a
+ src/UnliftIO/MessageBox/Util/CallId.hs view
@@ -0,0 +1,41 @@+module UnliftIO.MessageBox.Util.CallId+ ( CallId (MkCallId),+ HasCallIdCounter (..),+ takeNext,+ newCallIdCounter,+ )+where++import Control.Monad.Reader (MonadReader, asks)+import UnliftIO.MessageBox.Util.Fresh+ ( CounterVar,+ incrementAndGet,+ newCounterVar,+ )+import UnliftIO (MonadIO, MonadUnliftIO)++-- | An identifier value every command send by 'call's.+newtype CallId = MkCallId Int+ deriving newtype (Eq, Ord)++instance Show CallId where+ showsPrec _ (MkCallId !i) =+ showChar '<' . shows i . showChar '>'++-- | Class of environment records containing a 'CounterVar' for 'CallId's.+class HasCallIdCounter env where+ getCallIdCounter :: env -> CounterVar CallId++instance HasCallIdCounter (CounterVar CallId) where+ {-# INLINE getCallIdCounter #-}+ getCallIdCounter = id++-- | Create a new 'CallId' 'CounterVar'.+{-# INLINE newCallIdCounter #-}+newCallIdCounter :: MonadIO m => m (CounterVar CallId)+newCallIdCounter = newCounterVar++-- | Increment and get a new 'CallId'.+{-# INLINE takeNext #-}+takeNext :: (MonadReader env m, HasCallIdCounter env, MonadUnliftIO m) => m CallId+takeNext = asks getCallIdCounter >>= incrementAndGet
+ src/UnliftIO/MessageBox/Util/Fresh.hs view
@@ -0,0 +1,69 @@+-- | Threadsafe, shared, atomic counters+--+-- This is based on "Data.Atomics.Counter".+module UnliftIO.MessageBox.Util.Fresh+ ( fresh,+ incrementAndGet,+ newCounterVar,+ HasCounterVar (getCounterVar),+ CounterVar (),+ )+where++import Control.Monad.Reader (MonadReader, asks)+import Data.Atomics.Counter+ ( AtomicCounter,+ incrCounter,+ newCounter,+ )+import Data.Coerce (Coercible, coerce)+import UnliftIO (MonadIO (..))++-- | A threadsafe atomic a++-- | Atomically increment and get the value of the 'Counter'+-- for type @a@ that must be present in the @env@.+{-# INLINE fresh #-}+fresh ::+ forall a env m.+ ( MonadReader env m,+ MonadIO m,+ HasCounterVar a env,+ Coercible a Int+ ) =>+ m a+fresh =+ asks (getCounterVar @a) >>= incrementAndGet++-- | Atomically increment and get the value of the 'Counter'+-- for type @a@ that must be present in the @env@.+{-# INLINE incrementAndGet #-}+incrementAndGet ::+ forall a m.+ ( MonadIO m,+ Coercible a Int+ ) =>+ CounterVar a -> m a+incrementAndGet (MkCounterVar !atomicCounter) =+ coerce <$> liftIO (incrCounter 1 atomicCounter)+++-- | Create a new 'CounterVar' starting at @0@.+{-# INLINE newCounterVar #-}+newCounterVar ::+ forall a m.+ MonadIO m =>+ m (CounterVar a)+newCounterVar =+ MkCounterVar <$> liftIO (newCounter 0)++-- | An 'AtomicCounter'.+newtype CounterVar a = MkCounterVar AtomicCounter++-- | A type class for @MonadReader@ based+-- applications.+class HasCounterVar a env | env -> a where+ getCounterVar :: env -> CounterVar a++instance HasCounterVar t (CounterVar t) where+ getCounterVar = id
+ src/UnliftIO/MessageBox/Util/Future.hs view
@@ -0,0 +1,32 @@+module UnliftIO.MessageBox.Util.Future+ ( Future (Future),+ tryNow,+ awaitFuture,+ )+where++import UnliftIO (MonadIO (liftIO), MonadUnliftIO)+import UnliftIO.Concurrent (threadDelay)++-- | A wrapper around an IO action that returns value+-- in the future.+newtype Future a = Future+ { -- | Return 'Just' the value or 'Nothing',+ -- when the value is not available yet.+ fromFuture :: IO (Maybe a)+ }++-- | Return 'Just' the value or 'Nothing',+-- when the value is not available yet.+--+-- Once the value is available, that value+-- will be returned everytime this function is+-- invoked.+{-# INLINE tryNow #-}+tryNow :: MonadUnliftIO m => Future a -> m (Maybe a)+tryNow = liftIO . fromFuture++-- | Poll a Future until the value is present.+awaitFuture :: MonadUnliftIO m => Future b -> m b+awaitFuture !f =+ tryNow f >>= maybe (threadDelay 10 >> awaitFuture f) return
+ unliftio-messagebox.cabal view
@@ -0,0 +1,212 @@+cabal-version: 3.0+name: unliftio-messagebox+version: 1.0.0+description: Please see the README on GitHub at <https://github.com/sheyll/unliftio-messagebox#readme>+synopsis: Fast and robust message queues for concurrent processes+homepage: https://github.com/sheyll/unliftio-messagebox#readme+bug-reports: https://github.com/sheyll/unliftio-messagebox/issues+author: Sven Heyll+maintainer: sven.heyll@gmail.com+category: Concurrency, Control, Effect+tested-with: GHC==8.10.2,GHC==8.10.3+copyright: Copyright Sven Heyll+license: BSD-2-Clause+license-file: LICENSE+build-type: Simple++extra-source-files:+ README.md++source-repository head+ type: git+ location: https://github.com/sheyll/unliftio-messagebox++common executable-flags+ ghc-options:+ -threaded+ -O2+ "-with-rtsopts=-T -N -qa"+ ghc-prof-options: + -threaded+ -O2+ -fprof-auto + "-with-rtsopts=-T -xt -xc -Pa -hc -L256"+ +common compiler-flags+ ghc-options:+ -Wall+ -Wcompat+ -Widentities+ -Wincomplete-uni-patterns+ -Wincomplete-record-updates+ -Wredundant-constraints+ -Wmissing-export-lists+ -Wpartial-fields+ -Wmissing-deriving-strategies+ -funbox-strict-fields+ -Wno-missing-signatures + -- -Werror+ default-extensions:+ AllowAmbiguousTypes,+ BangPatterns,+ BinaryLiterals,+ ConstraintKinds,+ DataKinds,+ DefaultSignatures,+ DeriveFoldable,+ DeriveFunctor,+ DeriveGeneric,+ DeriveTraversable,+ DerivingStrategies,+ DerivingVia,+ ExistentialQuantification,+ FlexibleContexts,+ FlexibleInstances,+ FunctionalDependencies,+ GADTs,+ GeneralizedNewtypeDeriving,+ InstanceSigs,+ LambdaCase,+ MultiParamTypeClasses,+ MultiWayIf,+ NamedFieldPuns, + NumericUnderscores,+ OverloadedStrings,+ PatternGuards,+ PolyKinds,+ RankNTypes,+ RecordWildCards,+ ScopedTypeVariables,+ StandaloneDeriving,+ StrictData,+ TupleSections,+ TypeApplications,+ TypeFamilies,+ TypeInType,+ TypeOperators,+ TypeSynonymInstances,+ ViewPatterns+ other-extensions:+ CPP,+ DeriveAnyClass,+ DeriveLift,+ ForeignFunctionInterface,+ ImplicitParams,+ MagicHash,+ PackageImports,+ QuasiQuotes,+ StaticPointers,+ StrictData,+ Strict,+ TemplateHaskell,+ TypeOperators,+ UnboxedTuples,+ UndecidableInstances,+ UnliftedFFITypes+ +common deps+ build-depends:+ atomic-primops,+ base >= 4.14 && <5,+ containers >=0.5.8 && <0.7,+ data-default >= 0.7 && < 0.8,+ hashable,+ mtl,+ QuickCheck,+ text,+ unagi-chan,+ unliftio++library+ import: deps, compiler-flags+ default-language: Haskell2010+ hs-source-dirs:+ src+ ghc-options:+ -Wall+ -Wcompat+ -Widentities+ -Wincomplete-uni-patterns+ -Wincomplete-record-updates+ -Wredundant-constraints+ -Wmissing-export-lists+ -Wpartial-fields+ -Wmissing-deriving-strategies+ -funbox-strict-fields+ -O2+ -- -Werror+ ghc-prof-options: + -fprof-auto + exposed-modules:+ UnliftIO.MessageBox,+ UnliftIO.MessageBox.CatchAll,+ UnliftIO.MessageBox.Class,+ UnliftIO.MessageBox.Command,+ UnliftIO.MessageBox.Limited,+ UnliftIO.MessageBox.Unlimited,+ UnliftIO.MessageBox.Util.CallId,+ UnliftIO.MessageBox.Util.Fresh,+ UnliftIO.MessageBox.Util.Future+ +executable unliftio-messagebox-memleak-test+ import: deps, compiler-flags, executable-flags+ default-language: Haskell2010+ main-is: Main.hs+ other-modules:+ MediaBenchmark+ hs-source-dirs: src-memleak-test + ghc-options:+ -O2+ -threaded+ -rtsopts+ "-with-rtsopts=-T -N"+ ghc-prof-options: + -O2+ -threaded+ -rtsopts+ -fprof-auto + "-with-rtsopts=-T -hy -pa -L256 -xt -N"+ build-depends:+ unliftio-messagebox++test-suite unliftio-messagebox-test+ import: deps, compiler-flags, executable-flags+ default-language: Haskell2010+ type: exitcode-stdio-1.0+ main-is: Main.hs+ hs-source-dirs: src-test+ other-modules:+ LimitedMessageBoxTest, + CallIdTest,+ CatchAllTest,+ CommandTest, + FreshTest,+ CornerCaseTests,+ MessageBoxClassTest,+ MessageBoxCommon,+ QCOrphans,+ UnlimitedMessageBoxTest,+ Utils+ build-depends:+ base,+ tasty,+ tasty-html,+ tasty-hunit,+ tasty-quickcheck,+ unliftio-messagebox,+ HUnit+ +benchmark unliftio-messagebox-bench+ import: deps, compiler-flags, executable-flags+ type: exitcode-stdio-1.0+ default-language: Haskell2010+ main-is: Main.hs+ other-modules:+ BookStoreBenchmark,+ CommandBenchmark,+ MediaBenchmark+ hs-source-dirs: src-benchmark+ build-depends:+ unliftio-messagebox,+ criterion+