rio-process-pool (empty) → 1.0.0
raw patch · 12 files changed
+2795/−0 lines, 12 filesdep +HUnitdep +QuickCheckdep +async
Dependencies added: HUnit, QuickCheck, async, atomic-primops, base, containers, criterion, data-default, hashable, mtl, rio, rio-process-pool, tasty, tasty-html, tasty-hunit, tasty-quickcheck, text, unliftio, unliftio-messagebox
Files
- LICENSE +9/−0
- README.md +66/−0
- rio-process-pool.cabal +180/−0
- src-benchmark/Main.hs +172/−0
- src-pool-memleak-test/Main.hs +198/−0
- src-test/BrokerTest.hs +1038/−0
- src-test/Main.hs +27/−0
- src-test/PoolTest.hs +366/−0
- src-test/Utils.hs +95/−0
- src/RIO/ProcessPool.hs +53/−0
- src/RIO/ProcessPool/Broker.hs +390/−0
- src/RIO/ProcessPool/Pool.hs +201/−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,66 @@+# Concurrent Process Pools++**NOTE:** To be able to fully view this README, use the [GitHub Pages Version](https://sheyll.github.io/rio-process-pool/).++* [API docs](./generated-reports/haddock-report/rio-process-pool)+* [API docs on Hackage](http://hackage.haskell.org/package/rio-process-pool)++A process pool processes incoming messages and dispatches them to worker threads.++Worker threads are started and stopped and each has their own message box, that decouples+from the processes for asynchronous processing.++The main architectural advantages are the usage of multiple CPU cores and the seperation of+address spaces of each worker.++This is clearly more low-level than STM in that regard, that STM offers +shared memory concurrency.++This is also just a small library built on top of [unliftio-messagebox](https://sheyll.github.io/unliftio-messagebox/).++## Module Structure++The library is contained in modules with names starting with +**RIO.ProcessPool**.++++Also the module +`RIO.ProcessPool` [(API docs)](./generated-reports/haddock-report/rio-process-pool/rio-process-pool.html)+[(Hackage)](http://hackage.haskell.org/package/rio-process-pool/docs/rio-process-pool.html)+exposes the API, and can be used to import everything.++The full documentation is either [on this page](./generated-reports/haddock-report/rio-process-pool/index.html)[(build log)](./generated-reports/haddock-report/build.log)+or on [Hackage](http://hackage.haskell.org/package/rio-process-pool).++## 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/rio-process-pool-test.prof)++## Memory Leak Tests++Benchmark memory usage of a very simple `Pool` example.++A single dispatcher process sends a `Start`, some `Work` and a+`Stop` message to a `Pool` that spawns and dispatches the message++Like the previous benchmark this is a rather long running test +executed with capped memory so that when space is leaked, it +will crash the benchmark.++++The output is printed into [this log file](./generated-reports/pool-memleak-test-report/test.log).
+ rio-process-pool.cabal view
@@ -0,0 +1,180 @@+cabal-version: 3.0+name: rio-process-pool+version: 1.0.0+description: Please see the README on GitHub at <https://github.com/sheyll/rio-process-pool#readme>+synopsis: A library for process pools coupled with asynchronous message queues+homepage: https://github.com/sheyll/rio-process-pool#readme+bug-reports: https://github.com/sheyll/rio-process-pool/issues+author: Sven Heyll+maintainer: sven.heyll@gmail.com+category: Concurrency+tested-with: GHC==8.10.4,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/rio-process-pool++flag development+ description: + Configure a development version, i.e. -fprof-auto and -Werror+ default: False+ manual: True+ +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+ -fno-full-laziness+ if flag(development)+ ghc-options:+ -Werror+ ghc-prof-options:+ -fprof-auto + 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, + NoImplicitPrelude,+ 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 executable-flags+ ghc-options:+ -threaded+ -rtsopts+ "-with-rtsopts=-T -N -qa"+ if flag(development)+ ghc-prof-options: + "-with-rtsopts=-T -xt -xc -Pa -hc -L256"+ +common deps+ build-depends:+ async >= 2 && <3,+ base >= 4.14 && <5,+ containers >=0.5.8 && <0.7,+ data-default >= 0.7 && < 0.8,+ hashable,+ mtl,+ QuickCheck,+ rio,+ text,+ unliftio,+ unliftio-messagebox >= 2.0.0 && < 3++library+ import: deps, compiler-flags+ default-language: Haskell2010+ hs-source-dirs:+ src+ exposed-modules:+ RIO.ProcessPool,+ RIO.ProcessPool.Broker,+ RIO.ProcessPool.Pool+ +executable rio-process-pool-memleak-test+ import: deps, compiler-flags, executable-flags+ default-language: Haskell2010+ main-is: Main.hs+ hs-source-dirs: src-pool-memleak-test + ghc-prof-options: + "-with-rtsopts=-T -hy -pa -L256 -xt -N"+ build-depends:+ rio-process-pool+ +test-suite rio-process-pool-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:+ BrokerTest,+ PoolTest,+ Utils+ build-depends:+ base,+ tasty,+ tasty-html,+ tasty-hunit,+ tasty-quickcheck,+ rio-process-pool,+ atomic-primops,+ HUnit+ +benchmark rio-process-pool-bench+ import: deps, compiler-flags, executable-flags+ type: exitcode-stdio-1.0+ default-language: Haskell2010+ main-is: Main.hs+ hs-source-dirs: src-benchmark+ build-depends:+ rio-process-pool,+ criterion+
+ src-benchmark/Main.hs view
@@ -0,0 +1,172 @@+{-# 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 OverloadedStrings #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++module Main (main) where++import Criterion.Main (defaultMain)+import Criterion.Types+ ( bench,+ bgroup,+ nfAppIO,+ )+import RIO+import RIO.ProcessPool++main =+ defaultMain+ [ bgroup+ "Pool"+ [ bench+ ( "Sending "+ <> show noMessages+ <> " msgs for "+ <> show receiverNo+ <> " receivers"+ )+ ( nfAppIO+ (runSimpleApp . unidirectionalMessagePassing BlockingUnlimited BlockingUnlimited)+ (noMessages, receiverNo)+ )+ | noMessages <- [10,50],+ receiverNo <- [500, 1000]+ ]+ ]++mkTestMessage :: Int -> TestMessage+mkTestMessage !i =+ MkTestMessage+ ( i,+ ( "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 "+ <> utf8BuilderToText (displayShow (23 * i)),+ "large",+ "meeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeessssssssssssssssssssssssssssssssss"+ <> utf8BuilderToText (displayShow i),+ ( "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",+ "ggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee",+ even i,+ 123423421111111111111111111123234 * toInteger i+ )+ )+ )++newtype TestMessage = MkTestMessage (Int, (Text, Text, Text, (Text, Text, Bool, Integer)))+ deriving newtype (Show)++instance Display TestMessage where+ display (MkTestMessage (x, (a, b, c, (d, e, f, g)))) =+ display x <> display ':'+ <> display a+ <> display ' '+ <> display b+ <> display ' '+ <> display c+ <> display ' '+ <> display d+ <> display ' '+ <> display e+ <> display ' '+ <> displayShow f+ <> display ' '+ <> display g+ <> display '.'++unidirectionalMessagePassing ::+ (IsMessageBoxArg boxPool, IsMessageBoxArg boxWorker) =>+ boxPool ->+ boxWorker ->+ (Int, Int) ->+ RIO SimpleApp ()+unidirectionalMessagePassing !boxPool !boxWorker (!nM, !nC) = do+ allStopped <- newEmptyMVar+ logInfo "~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ BEGIN ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~"+ spawnPool+ boxPool+ boxWorker+ ( MkPoolWorkerCallback+ ( \k b ->+ ( do+ consume k b+ putMVar allStopped ()+ )+ `withException` ( \(e :: SomeException) ->+ logError+ ( "consume threw exception: "+ <> display e+ <> " for worker: "+ <> display k+ )+ )+ )+ )+ >>= \case+ Left err -> error (show err)+ Right pool -> do+ forM_ [0..1 :: Int] $ \ !runIndex -> do+ logInfo " ~~~ NEXT ROUND ~~~"+ producer runIndex pool+ logInfo " ~~ WAITING FOR RESULTS ~~"+ awaitAllStopped allStopped + -- sendPoison runIndex pool++ cancel (poolAsync pool)+ where+ producer !runIndex !pool = do+ mapM_+ (deliverOrLog pool)+ ((`Initialize` Nothing) . (+ (nC * runIndex)) <$> [0 .. nC - 1])+ mapM_+ (deliverOrLog pool)+ (Dispatch . (+ (nC * runIndex)) <$> [0 .. nC - 1] <*> (Just . mkTestMessage <$> [0 .. nM - 1]))+ deliverOrLog pool !msg = do+ !ok <- deliver (poolInput pool) msg+ unless+ ok+ ( error+ ( "producer failed to deliver: "+ <> show+ ( case msg of+ Dispatch k _ -> k+ Initialize k _ -> k+ )+ )+ )+ -- sendPoison !runIndex !pool = + -- mapM_+ -- (deliverOrLog pool)+ -- ((`Dispatch` Nothing) . (+ (nC * runIndex)) <$> [0 .. nC - 1])++ consume k inBox = do+ receive inBox+ >>= maybe+ (logError ("consumer: " <> display k <> " failed to receive a message"))+ ( \msg@(MkTestMessage (!x, !_)) -> do+ logDebug ("consumer: " <> display k <> " received: " <> display msg)+ unless+ (x == nM - 1)+ (consume k inBox)+ )++ awaitAllStopped allStopped =+ replicateM_ nC (void $ takeMVar allStopped)+ `withException` ( \(e :: SomeException) ->+ logError ("exception in awaitAllStopped: " <> display e)+ )
+ src-pool-memleak-test/Main.hs view
@@ -0,0 +1,198 @@+module Main (main) where++import RIO+import RIO.Partial+import RIO.ProcessPool+import System.Environment (getArgs)++main :: IO ()+main = do+ args <- getArgs+ runSimpleApp $ do+ let (rounds, repetitions, nWorkMessages) =+ case args of+ [n', r', w'] -> (read n', read r', read w')+ [n', r'] -> (read n', read r', 10)+ [n'] -> (read n', 1, 10)+ _ -> (1, 1, 10)+ logInfo+ ( "Benchmark: Running "+ <> display repetitions+ <> " times a benchmark with "+ <> display rounds+ <> " active processes, each processing "+ <> display nWorkMessages+ <> " work messages."+ )+ bench repetitions rounds nWorkMessages++newtype Key = Key Int deriving newtype (Eq, Ord, Show, Display)++data Msg = Start | Work | SyncWork (MVar Int) | Stop++{-# NOINLINE bench #-}++-- | Start @bSize@ clients that work 10000 work units.+bench :: Int -> Int -> Int -> RIO SimpleApp ()+bench repetitions bSize nWorkMessages = do+ resultBox <- newMessageBox BlockingUnlimited+ resultBoxIn <- newInput resultBox+ -- start a pool+ x <-+ spawnPool+ BlockingUnlimited+ BlockingUnlimited+ (MkPoolWorkerCallback (enterWorkLoop resultBoxIn))+ case x of+ Left err ->+ logError ("Error: " <> display err)+ Right pool -> do+ forM_ [1 .. repetitions] $ \ !rep -> do+ logInfo+ ( "================= BEGIN (rep: "+ <> display rep+ <> ") ================"+ )+ do+ let groupSize = 1000+ mapConcurrently_+ (traverse_ (sendWork nWorkMessages (poolInput pool) . Key))+ ( [groupSize * ((bSize - 1) `div` groupSize) .. bSize - 1] :+ [ [groupSize * x0 .. groupSize * x0 + (groupSize - 1)]+ | x0 <- [0 .. ((bSize - 1) `div` groupSize) - 1]+ ]+ )++ logInfo "Waiting for results"+ printResults bSize resultBox++ logInfo ("================= DONE (rep: " <> display rep <> ") ================")+ cancel (poolAsync pool)++{-# NOINLINE printResults #-}+printResults ::+ (IsMessageBox box) =>+ Int ->+ box (Key, Int) ->+ RIO SimpleApp ()+printResults bSize box+ | bSize <= 0 = return ()+ | otherwise =+ receive box+ >>= maybe+ (logInfo "done!")+ ( \(Key k, v) -> do+ when+ (k `mod` 1000 == 0)+ (logInfo ("Result of " <> display k <> " is: " <> display v))+ printResults (bSize - 1) box+ )++{-# NOINLINE sendWork #-}+sendWork ::+ (IsInput input) =>+ Int ->+ input (Multiplexed Key (Maybe Msg)) ->+ Key ->+ RIO SimpleApp ()+sendWork nWorkMessages poolBoxIn k@(Key x) = do+ when+ (x `mod` 1000 == 0)+ (logInfo ("Delivering Messages for: " <> display k))+ void $ deliver poolBoxIn (Initialize k (Just (Just Start)))+ when+ (x `mod` 1000 == 0)+ (logInfo ("Delivering Sync Work Messages for: " <> display k))+ do+ resRef <- newEmptyMVar+ replicateM_+ nWorkMessages+ ( do+ deliver_ poolBoxIn (Dispatch k (Just (SyncWork resRef)))+ res <- takeMVar resRef+ when+ (x `mod` 1000 == 0 && res `mod` 1000 == 0)+ ( logInfo+ ( "Current counter of: "+ <> display k+ <> " is: "+ <> display res+ )+ )+ )+ when+ (x `mod` 1000 == 0)+ (logInfo ("Delivering Async Work Messages for: " <> display k))+ replicateM_+ nWorkMessages+ ( deliver_ poolBoxIn (Dispatch k (Just Work))+ )++ -- replicateM_ nWorkMessages (deliver poolBoxIn (Dispatch k (Just Work)))+ void $ deliver poolBoxIn (Dispatch k (Just Stop))+ when+ (x `mod` 1000 == 0)+ (logInfo ("Delivered all Messages for: " <> display k))++{-# NOINLINE enterWorkLoop #-}+enterWorkLoop ::+ (IsMessageBox box, IsInput input) =>+ input (Key, Int) ->+ Key ->+ box Msg ->+ RIO SimpleApp ()+enterWorkLoop resultBoxIn (Key k) box =+ tryAny (receive box)+ >>= ( \case+ Left ex ->+ logError+ ( "Receive threw exception for worker: "+ <> display k+ <> " "+ <> display ex+ )+ Right Nothing ->+ logError ("Receive failed for worker: " <> display k)+ Right (Just Start) -> do+ when+ (k `mod` 1000 == 0)+ (logInfo ("Started: " <> display k))+ workLoop 0+ Right (Just Work) ->+ logWarn ("Got unexpected Work: " <> display k)+ Right (Just (SyncWork ref)) -> do+ logWarn ("Got unexpected SyncWork: " <> display k)+ putMVar ref 0+ Right (Just Stop) ->+ logWarn ("Got unexpected Stop: " <> display k)+ )+ where+ workLoop !counter =+ tryAny (receive box)+ >>= \case+ Left ex ->+ logError+ ( "Receive threw exception for worker: "+ <> display k+ <> " "+ <> display ex+ )+ Right Nothing ->+ logWarn ("Receive failed for worker: " <> display k)+ Right (Just Start) -> do+ logWarn ("Re-Start: " <> display k)+ workLoop 0+ Right (Just Work) ->+ workLoop (counter + 1)+ Right (Just (SyncWork ref)) -> do+ putMVar ref counter+ -- timeout 5000000 (putMVar ref counter)+ -- >>= maybe+ -- (liftIO (putStrLn "SyncWork timeout putting the result"))+ -- (const (return ()))+ workLoop (counter + 1)+ Right (Just Stop) -> do+ void (deliver resultBoxIn (Key k, counter))+ when+ (k `mod` 1000 == 0)+ (logInfo ("Stopped: " <> display k))
+ src-test/BrokerTest.hs view
@@ -0,0 +1,1038 @@+module BrokerTest (test) where++import Control.Exception (throw)+import Data.List (sort)+import RIO+import RIO.ProcessPool.Broker+import Test.Tasty+import Test.Tasty.HUnit+import UnliftIO.MessageBox.Class+import UnliftIO.MessageBox.Unlimited+import Utils+ ( MockBox (MkMockBox),+ MockBoxInit (MkMockBoxInit),+ NoOpArg (..),+ NoOpBox,+ NoOpInput (..),+ runTestApp,+ )++noBrokerConfig :: BrokerConfig k w' w a m+noBrokerConfig =+ MkBrokerConfig+ { demultiplexer = const $ error "unexpected invokation: demultiplexer",+ messageDispatcher = const $ error "unexpected invokation: messageDispatcher",+ resourceCreator = const $ error "unexpected invokation: resourceCreator",+ resourceCleaner = const $ error "unexpected invokation: resourceCleaner"+ }++expectedException :: StringException+expectedException = stringException "Test"++test :: HasCallStack => TestTree+test =+ testGroup+ "BrokerTests"+ [ testCase+ "the show instance of BrokerResult makes sense"+ (assertEqual "" "MkBrokerResult" (show MkBrokerResult)),+ testCase+ "when a broker message box creation throws an exception, \+ \ the exception is returned in a Left..."+ $ do+ Just (Left a) <-+ runTestApp $+ timeout 1000000 $+ spawnBroker @_ @Int @() @() @NoOpArg+ ( MkMockBoxInit @NoOpBox+ (throwIO expectedException)+ Nothing+ )+ noBrokerConfig+ assertEqual+ "exception expected"+ (show (SomeException expectedException))+ (show a),+ testCase+ "when a broker message box input creation throws an exception,\+ \ the exception is returned in a Left..."+ $ do+ Just (Left a) <-+ runTestApp $+ timeout 1000000 $+ spawnBroker @_ @Int @() @() @NoOpArg+ ( MkMockBoxInit+ ( return+ ( MkMockBox @NoOpInput+ (throwIO expectedException)+ (error "unexpected invokation: receive")+ (error "unexpected invokation: tryReceive")+ )+ )+ Nothing+ )+ noBrokerConfig+ assertEqual+ "exception expected"+ (show (SomeException expectedException))+ (show a),+ testCase+ "when the receive function throws a synchronous exception,\+ \ then waiting on the broker will return the exception"+ $ do+ Just (Right (_, a)) <-+ runTestApp $+ timeout 1000000 $+ spawnBroker @_ @Int @() @() @NoOpArg+ ( MkMockBoxInit+ ( return+ ( MkMockBox+ ( return+ (OnDeliver (error "unexpected invokation: OnDeliver"))+ )+ (throwIO expectedException)+ (error "unexpected invokation: tryReceive")+ )+ )+ Nothing+ )+ noBrokerConfig+ r <- runTestApp $ waitCatch a+ assertEqual+ "exception expected"+ ( show+ ( Left (SomeException expectedException) ::+ Either SomeException ()+ )+ )+ (show r),+ testCase+ "when evaluation of an incoming message causes an exception,\+ \ then the broker ignores the error and continues"+ $ runTestApp $ do+ spRes <-+ spawnBroker+ ( MkMockBoxInit+ ( return+ ( MkMockBox+ ( return+ (OnDeliver (const (pure True)))+ )+ (return (Just (throw expectedException)))+ (error "unexpected invokation: tryReceive")+ )+ )+ Nothing+ )+ ( MkBrokerConfig+ { demultiplexer = Dispatch (777 :: Int),+ messageDispatcher =+ const (error "unexpected invokation: messageDispatcher"),+ resourceCreator =+ const (error "unexpected invokation: resourceCreator"),+ resourceCleaner =+ const (error "unexpected invokation: resourceCleaner")+ }+ )+ case spRes of+ Left err -> error (show err)+ Right (brokerIn, brokerA) -> do+ deliver_ brokerIn ()+ cancel brokerA+ r <- waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "Left AsyncCancelled"+ (show r),+ testCase+ "when evaluation of the first incoming message causes an async\+ \ exception, then the broker exits with that exception"+ $ runTestApp $ do+ spawnBroker+ ( MkMockBoxInit+ ( return+ ( MkMockBox+ ( return+ (OnDeliver (const (pure True)))+ )+ (return (Just (throw (AsyncExceptionWrapper expectedException))))+ (error "unexpected invokation: tryReceive")+ )+ )+ Nothing+ )+ ( MkBrokerConfig+ { demultiplexer = Dispatch (777 :: Int),+ messageDispatcher =+ const (error "unexpected invokation: messageDispatcher"),+ resourceCreator =+ const (error "unexpected invokation: resourceCreator"),+ resourceCleaner =+ const (error "unexpected invokation: resourceCleaner")+ }+ )+ >>= \case+ Left err -> error (show err)+ Right (brokerIn, brokerA) -> do+ deliver_ brokerIn ()+ r <- waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ ( show+ ( Left (SomeException (AsyncExceptionWrapper expectedException)) ::+ Either SomeException ()+ )+ )+ (show r),+ testCase+ "when a broker is cancelled while waiting for the first message,\+ \ then the broker exits with AsyncCancelled"+ $ runTestApp $ do+ goOn <- newEmptyMVar+ spawnBroker @_ @Int @() @()+ ( MkMockBoxInit+ ( return+ ( MkMockBox+ ( return+ (OnDeliver (const (pure True)))+ )+ ( do+ putMVar goOn ()+ threadDelay 1_000_000+ return (Just (error "unexpected evaluation"))+ )+ (error "unexpected invokation: tryReceive")+ )+ )+ Nothing+ )+ noBrokerConfig+ >>= \case+ Left err -> error (show err)+ Right (_brokerIn, brokerA) -> do+ takeMVar goOn+ cancel brokerA+ r <- waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "Left AsyncCancelled"+ (show r),+ testCase+ "when a broker receives a message for a missing resource,\+ \ it silently drops the message"+ $ runTestApp $ do+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer = Dispatch (777 :: Int),+ messageDispatcher =+ const (error "unexpected invokation: messageDispatcher"),+ resourceCreator =+ const (error "unexpected invokation: resourceCreator"),+ resourceCleaner =+ const (error "unexpected invokation: resourceCleaner")+ }+ spawnBroker BlockingUnlimited brokerCfg+ >>= \case+ Left err -> error (show err)+ Right (brokerIn, brokerA) -> do+ deliver_ brokerIn ()+ cancel brokerA+ r <- waitCatch brokerA+ liftIO $+ assertEqual+ "success expected"+ "Left AsyncCancelled"+ (show r),+ testCase+ "when an empty broker receives a start message without payload\+ \ and the creator callback throws an exception,\+ \ a normal message for that key will be ignored,\+ \ no cleanup is performed, and the broker lives on"+ $ runTestApp $ do+ workerInitialized <- newEmptyMVar+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer =+ \m ->+ if m+ then Initialize (777 :: Int) Nothing+ else Dispatch (777 :: Int) (),+ messageDispatcher =+ const (error "unexpected invokation: messageDispatcher"),+ resourceCreator = \_k _mw -> do+ putMVar workerInitialized ()+ throwIO expectedException,+ resourceCleaner =+ const (error "unexpected invokation: resourceCleaner")+ }+ spawnBroker BlockingUnlimited brokerCfg+ >>= \case+ Left err -> error (show err)+ Right (brokerIn, brokerA) -> do+ deliver_ brokerIn True+ timeout 1000000 (takeMVar workerInitialized)+ >>= liftIO+ . assertEqual+ "resourceCreator wasn't executed"+ (Just ())+ deliver_ brokerIn False+ cancel brokerA+ r <- waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "Left AsyncCancelled"+ (show r),+ testCase+ "when an empty broker receives a start message with a payload\+ \ and when the MessageHandler callback throws an exception when\+ \ applied to that payload, cleanup is performed once, and\+ \ incoming messages for that key will be ignored,\+ \ and the broker lives on"+ $ runTestApp $ do+ cleanupCalls <- newEmptyMVar+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer =+ \isInitPayload ->+ if isInitPayload+ then Initialize (777 :: Int) (Just True)+ else Dispatch (777 :: Int) False,+ messageDispatcher =+ \_k isInitPayload _ ->+ if isInitPayload+ then throwIO expectedException+ else error "unexpected invokation: messageDispatcher",+ resourceCreator = \_k _mw -> do+ putMVar cleanupCalls (0 :: Int)+ return (),+ resourceCleaner =+ \_k () ->+ modifyMVar+ cleanupCalls+ (\cnt -> return (cnt + 1, ()))+ }+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn True+ deliver_ brokerIn False+ threadDelay 10_000+ cancel brokerA+ r <- waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "Left AsyncCancelled"+ (show r)+ takeMVar cleanupCalls+ >>= liftIO . assertEqual "resourceCleaner wasn't executed" 1+ -- prevent GC of msg box input:+ deliver_ brokerIn False,+ testCase+ "when 3 resources are initialized and then the broker is cancelled,\+ \ cleanup is performed foreach resource."+ $ runTestApp $ do+ resourceCreated <- newEmptyMVar+ cleanupCalled <- newTVarIO []+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer = \k -> Initialize k (Just k),+ messageDispatcher = \k _w _a -> do+ putMVar resourceCreated k+ threadDelay 10_000 -- delay here so we make sure to+ -- be cancelled before we leave this+ -- function.+ -- Now if the implementation isn't+ -- handling async exceptions well,+ -- the resource isn't in the resource+ -- map when cleanup is called,+ -- and hence won't be properly+ -- cleaned up!+ return KeepResource,+ resourceCreator =+ \k _mw -> return k,+ resourceCleaner =+ \k a ->+ atomically (modifyTVar cleanupCalled ((k, a) :))+ }+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn (1 :: Int)+ deliver_ brokerIn 2+ deliver_ brokerIn 3+ takeMVar resourceCreated+ >>= liftIO . assertEqual "invalid resource created" 1+ takeMVar resourceCreated+ >>= liftIO . assertEqual "invalid resource created" 2+ takeMVar resourceCreated+ >>= liftIO . assertEqual "invalid resource created" 3+ cancel brokerA+ r <- waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "Left AsyncCancelled"+ (show r)+ readTVarIO cleanupCalled+ >>= liftIO+ . assertEqual+ "resourceCleaner wasn't executed"+ [(1, 1), (2, 2), (3, 3)]+ . sort+ -- prevent GC of msg box input:+ deliver_ brokerIn 666,+ testCase+ "when 3 resources are initialized and then the broker is cancelled,\+ \ cleanup is performed foreach resource, even if exceptions are thrown\+ \ from the cleanup callbacks"+ $ runTestApp $ do+ resourceCreated <- newEmptyMVar+ cleanupCalled <- newTVarIO []+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer = \k -> Initialize k (Just k),+ messageDispatcher = \k _w _a -> do+ putMVar resourceCreated k+ threadDelay 10_000 -- delay here so we make sure to+ -- be cancelled before we leave this+ -- function.+ -- Now if the implementation isn't+ -- handling async exceptions well,+ -- the resource isn't in the resource+ -- map when cleanup is called,+ -- and hence won't be properly+ -- cleaned up!+ return KeepResource,+ resourceCreator =+ \k _mw -> return k,+ resourceCleaner =+ \k a -> do+ atomically (modifyTVar cleanupCalled ((k, a) :))+ throwIO expectedException+ }+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn (1 :: Int)+ deliver_ brokerIn 2+ deliver_ brokerIn 3+ takeMVar resourceCreated+ >>= liftIO . assertEqual "invalid resource created" 1+ takeMVar resourceCreated+ >>= liftIO . assertEqual "invalid resource created" 2+ takeMVar resourceCreated+ >>= liftIO . assertEqual "invalid resource created" 3+ cancel brokerA+ r <- waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "Left AsyncCancelled"+ (show r)+ readTVarIO cleanupCalled+ >>= liftIO+ . assertEqual+ "resourceCleaner wasn't executed"+ [(1, 1), (2, 2), (3, 3)]+ . sort+ -- prevent GC of msg box input:+ deliver_ brokerIn 666,+ testCase+ "when 2 resources are added and\+ \ while adding a 3rd an async exception is thrown\+ \ when handling the initial message,\+ \ then the broker cleans up the 3 resources and exists"+ $ runTestApp $ do+ resourceCreated <- newEmptyMVar+ cleanupCalled <- newTVarIO []+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer = \k -> Initialize k (Just k),+ messageDispatcher = \k _w _a -> do+ putMVar resourceCreated k+ return KeepResource,+ resourceCreator =+ \k _mw -> return k,+ resourceCleaner =+ \k a -> do+ atomically (modifyTVar cleanupCalled ((k, a) :))+ }+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn (1 :: Int)+ deliver_ brokerIn 2+ deliver_ brokerIn 3+ takeMVar resourceCreated+ >>= liftIO . assertEqual "invalid resource created" 1+ takeMVar resourceCreated+ >>= liftIO . assertEqual "invalid resource created" 2+ takeMVar resourceCreated+ >>= liftIO . assertEqual "invalid resource created" 3+ throwTo (asyncThreadId brokerA) expectedException+ r <- either id (error . show) <$> waitCatch brokerA+ readTVarIO cleanupCalled+ >>= liftIO+ . assertEqual+ "resourceCleaner wasn't executed"+ [(1, 1), (2, 2), (3, 3)]+ . sort+ liftIO $+ assertEqual+ "exception expected"+ (show expectedException)+ (show r)+ -- prevent GC of msg box input:+ deliver_ brokerIn 666,+ testCase+ "when adding a new resource with an extra initial message,\+ \ when the messageHandler returns KeepResource,\+ \ then the resource returned from the create callback is passed to\+ \ the cleanup function"+ $ runTestApp $ do+ resourceCreated <- newEmptyMVar+ cleanupCalled <- newEmptyTMVarIO+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer = \_k -> Initialize (777 :: Int) (Just ()),+ messageDispatcher = \_k _w a -> do+ putMVar resourceCreated a+ return KeepResource,+ resourceCreator =+ \_k _mw -> return initialResource,+ resourceCleaner =+ \_k a -> do+ atomically (putTMVar cleanupCalled a)+ }+ initialResource :: Int+ initialResource = 123+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn (1 :: Int)+ takeMVar resourceCreated+ >>= liftIO+ . assertEqual "invalid resource created" initialResource+ cancel brokerA+ r <- either id (error . show) <$> waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "AsyncCancelled"+ (show r)+ atomically (takeTMVar cleanupCalled)+ >>= liftIO+ . assertEqual+ "resourceCleaner wasn't executed"+ initialResource+ -- prevent GC of msg box input:+ deliver_ brokerIn 666,+ testCase+ "when adding a new resource with an extra initial message,\+ \ when the messageHandler returns (UpdateResource x),\+ \ then x is passed to the cleanup function"+ $ runTestApp $ do+ resourceCreated <- newEmptyMVar+ cleanupCalled <- newEmptyTMVarIO+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer = \_k -> Initialize (787 :: Int) (Just ()),+ messageDispatcher = \_k _w a -> do+ void $ async (threadDelay 100000 >> putMVar resourceCreated a)+ return (UpdateResource x),+ resourceCreator =+ \_k _mw -> return initialResource,+ resourceCleaner =+ \_k -> atomically . putTMVar cleanupCalled+ }+ initialResource :: Int+ initialResource = 123+ x :: Int+ x = 234+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn (1 :: Int)+ takeMVar resourceCreated+ >>= liftIO+ . assertEqual+ "invalid resource created"+ initialResource+ cancel brokerA+ r <- either id (error . show) <$> waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "AsyncCancelled"+ (show r)+ atomically (takeTMVar cleanupCalled)+ >>= liftIO+ . assertEqual+ "resourceCleaner wasn't executed"+ x+ -- prevent GC of msg box input:+ deliver_ brokerIn 666,+ testCase+ "when adding a new resource with an extra initial message,\+ \ when the messageHandler returns (RemoveResource Nothing),\+ \ then the initial resource is passed to the cleanup function"+ $ runTestApp $ do+ resourceCreated <- newEmptyMVar+ readyForCleanup <- newEmptyMVar+ cleanupCalled <- newEmptyTMVarIO+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer = \_k -> Initialize (787 :: Int) (Just ()),+ messageDispatcher = \_k _w a -> do+ void . async $ do+ () <- takeMVar readyForCleanup+ putMVar resourceCreated a+ return (RemoveResource Nothing),+ resourceCreator =+ \_k _mw -> return initialResource,+ resourceCleaner =+ \_k a -> do+ putMVar readyForCleanup ()+ atomically (putTMVar cleanupCalled a)+ }+ initialResource :: Int+ initialResource = 123+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn (1 :: Int)+ takeMVar resourceCreated+ >>= liftIO+ . assertEqual+ "invalid resource created"+ initialResource+ cancel brokerA+ r <- either id (error . show) <$> waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "AsyncCancelled"+ (show r)+ atomically (takeTMVar cleanupCalled)+ >>= liftIO+ . assertEqual+ "resourceCleaner wasn't executed"+ initialResource+ -- prevent GC of msg box input:+ deliver_ brokerIn 666,+ testCase+ "when adding a new resource with an extra initial message,\+ \ when the messageHandler returns (RemoveResource (Just x)),\+ \ then x is passed to the cleanup function"+ $ runTestApp $ do+ resourceCreated <- newEmptyMVar+ readyForCleanup <- newEmptyMVar+ cleanupCalled <- newEmptyTMVarIO+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer = \_k -> Initialize (787 :: Int) (Just ()),+ messageDispatcher = \_k _w a -> do+ void . async $ do+ () <- takeMVar readyForCleanup+ putMVar resourceCreated a+ return (RemoveResource (Just x)),+ resourceCreator =+ \_k _mw -> return initialResource,+ resourceCleaner =+ \_k a -> do+ putMVar readyForCleanup ()+ atomically (putTMVar cleanupCalled a)+ }+ initialResource :: Int+ initialResource = 123+ x :: Int+ x = 234+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn (1 :: Int)+ takeMVar resourceCreated+ >>= liftIO+ . assertEqual+ "invalid resource created"+ initialResource+ cancel brokerA+ r <- either id (error . show) <$> waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "AsyncCancelled"+ (show r)+ atomically (takeTMVar cleanupCalled)+ >>= liftIO+ . assertEqual+ "resourceCleaner wasn't executed"+ x+ -- prevent GC of msg box input:+ deliver_ brokerIn 666,+ testCase+ "when adding a new resource without an extra initial message,\+ \ then the initial resource is passed to the cleanup function"+ $ runTestApp $ do+ resourceCreated <- newEmptyMVar+ cleanupCalled <- newEmptyTMVarIO+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer = \_k -> Initialize (787 :: Int) (Just ()),+ messageDispatcher = \_k _w a -> do+ void $ async (threadDelay 100000 >> putMVar resourceCreated a)+ return (UpdateResource x),+ resourceCreator =+ \_k _mw -> return initialResource,+ resourceCleaner =+ \_k -> atomically . putTMVar cleanupCalled+ }+ initialResource :: Int+ initialResource = 123+ x :: Int+ x = 234+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn (1 :: Int)+ takeMVar resourceCreated+ >>= liftIO+ . assertEqual+ "invalid resource created"+ initialResource+ cancel brokerA+ r <- either id (error . show) <$> waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "AsyncCancelled"+ (show r)+ atomically (takeTMVar cleanupCalled)+ >>= liftIO+ . assertEqual+ "resourceCleaner wasn't executed"+ x+ -- prevent GC of msg box input:+ deliver_ brokerIn 666,+ testCase+ "on a broker with two resources a and b with keys 1 and 2,\+ \ for an incoming message with key=1 the MessageDispatcher is\+ \ called with resource a and for key=2 with resource b"+ $ runTestApp $ do+ messageDispatched <- newEmptyMVar+ cleanupCalled <- newTVarIO []+ let brokerCfg =+ MkBrokerConfig+ { demultiplexer = \(k, isInit) ->+ if isInit+ then Initialize k Nothing+ else Dispatch k k,+ resourceCreator =+ \k _mw -> return k,+ messageDispatcher = \k _w a -> do+ putMVar messageDispatched (k, a)+ return KeepResource,+ resourceCleaner =+ \k a -> do+ atomically (modifyTVar cleanupCalled ((k, a) :))+ }+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn (1 :: Int, True)+ deliver_ brokerIn (2, True)+ deliver_ brokerIn (1, False)+ deliver_ brokerIn (2, False)+ takeMVar messageDispatched+ >>= liftIO+ . assertEqual "invalid resource" (1, 1)+ takeMVar messageDispatched+ >>= liftIO+ . assertEqual "invalid resource" (2, 2)+ throwTo (asyncThreadId brokerA) expectedException+ r <- either id (error . show) <$> waitCatch brokerA+ readTVarIO cleanupCalled+ >>= liftIO+ . assertEqual+ "resourceCleaner wasn't executed"+ [(1, 1), (2, 2)]+ . sort+ liftIO $+ assertEqual+ "exception expected"+ (show expectedException)+ (show r)+ -- prevent GC of msg box input:+ deliver_ brokerIn (666, False),+ testCase+ "When the MessageDispatcher is called with resource x and \+ \ returns (UpdateResource y), then next time it is called with\+ \ resource y"+ $ runTestApp $ do+ messageDispatched <- newEmptyMVar+ let x :: Int+ x = 42+ y = x + 1295+ brokerCfg =+ MkBrokerConfig+ { demultiplexer = \isInit ->+ if isInit+ then Initialize (787 :: Int) Nothing+ else Dispatch (787 :: Int) (),+ resourceCreator = \_k _mw ->+ return x,+ messageDispatcher = \_k _w a -> do+ putMVar messageDispatched a+ return (UpdateResource y),+ resourceCleaner = \_k _a ->+ return ()+ }+ (brokerIn, brokerA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited brokerCfg+ deliver_ brokerIn True+ deliver_ brokerIn False+ deliver_ brokerIn False+ takeMVar messageDispatched+ >>= liftIO . assertEqual "invalid resource" x+ takeMVar messageDispatched+ >>= liftIO . assertEqual "invalid resource" y+ cancel brokerA+ r <- either id (error . show) <$> waitCatch brokerA+ liftIO $+ assertEqual+ "exception expected"+ "AsyncCancelled"+ (show r)+ -- prevent GC of msg box input:+ deliver_ brokerIn False,+ testCase+ "when the receive function throws a synchronous exception x,\+ \ then all resources will be cleaned up and\+ \ the broker will re-throw x"+ $ runTestApp $ do+ cr <- newTVarIO []+ cl <- newTVarIO []+ let bCfg =+ MkBrokerConfig+ { demultiplexer = (`Initialize` Nothing),+ resourceCreator = \k _mw -> do+ atomically $+ modifyTVar cr (k :)+ return (),+ messageDispatcher = \_k _w _a ->+ return KeepResource,+ resourceCleaner = \k _a ->+ atomically $ modifyTVar cl (k :)+ }+ inter =+ MkInterceptingBoxArg+ ( MkInterceptor $ do+ rs <- readTVarIO cr+ if rs == [3, 2, 1]+ then throwIO expectedException+ else return Nothing+ )+ (MkInterceptor (return Nothing))+ BlockingUnlimited+ (b, bA) <-+ either (error . show) id+ <$> spawnBroker inter bCfg+ deliver_ b (1 :: Int)+ deliver_ b (2 :: Int)+ deliver_ b (3 :: Int)+ deliver_ b (4 :: Int)+ ex <- either id (error . show) <$> waitCatch bA+ liftIO $+ assertEqual+ "exception expected"+ (show expectedException)+ (show ex)+ readTVarIO cl+ >>= liftIO+ . assertEqual+ "all resources must be cleaned"+ [3, 2, 1]+ -- prevent GC of msg box input:+ deliver_ b 666,+ testCase+ "when the receive function returns Nothing,\+ \ then all resources will be cleaned up and\+ \ the broker will exit."+ $ runTestApp $ do+ cr <- newTVarIO []+ cl <- newTVarIO []+ let bCfg =+ MkBrokerConfig+ { demultiplexer = (`Initialize` Nothing),+ resourceCreator = \k _mw -> do+ atomically $+ modifyTVar cr (k :)+ return (),+ messageDispatcher = \_k _w _a ->+ return KeepResource,+ resourceCleaner = \k _a ->+ atomically $ modifyTVar cl (k :)+ }+ inter =+ MkInterceptingBoxArg+ ( MkInterceptor $ do+ rs <- readTVarIO cr+ if rs == [3, 2, 1]+ then return (Just False)+ else return Nothing+ )+ (MkInterceptor (return Nothing))+ BlockingUnlimited+ (b, bA) <-+ either (error . show) id+ <$> spawnBroker inter bCfg+ deliver_ b (1 :: Int)+ deliver_ b (2 :: Int)+ deliver_ b (3 :: Int)+ deliver_ b (4 :: Int)+ wait bA+ >>= liftIO+ . assertEqual "exit expected" MkBrokerResult+ readTVarIO cl+ >>= liftIO+ . assertEqual "all resources must be cleaned" [3, 2, 1]+ -- prevent GC of msg box input:+ deliver_ b 666,+ testCase+ "when an initialisation message without payload for an existant resource\+ \ is received, the message is ignored"+ $ runTestApp $ do+ cr <- newTVarIO []+ cl <- newTVarIO []+ let bCfg =+ MkBrokerConfig+ { demultiplexer = (`Initialize` Nothing),+ resourceCreator = \k _mw ->+ atomically $ modifyTVar cr (k :),+ messageDispatcher = \_k _w _a ->+ error "unexpected call to messageDispatcher",+ resourceCleaner = \k _a ->+ atomically $ modifyTVar cl (k :)+ }+ (b, bA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited bCfg+ deliver_ b (1 :: Int)+ deliver_ b (1 :: Int)+ deliver_ b (2 :: Int)+ atomically $ do+ crs <- readTVar cr+ checkSTM (crs == [2, 1])+ cancel bA+ waitCatch bA+ >>= liftIO+ . assertEqual+ "exit expected"+ "Left AsyncCancelled"+ . show+ readTVarIO cl+ >>= liftIO+ . assertEqual+ "all resources must be cleaned"+ [2, 1]+ -- prevent GC of msg box input:+ deliver_ b 666,+ testCase+ "when an initialisation message with payload, \+ \ then after the resource creation is done, \+ \ the payload is dispatched to the messageDispatcher"+ $ runTestApp $ do+ done <- newEmptyMVar+ let bCfg =+ MkBrokerConfig+ { demultiplexer = (`Initialize` Just ()),+ resourceCreator = \_k mw ->+ case mw of+ Just () ->+ return ()+ Nothing ->+ error "now initial message",+ messageDispatcher = \_k _w _a -> do+ putMVar done ()+ return KeepResource,+ resourceCleaner = \_k _a -> return ()+ }+ (b, bA) <-+ either (error . show) id+ <$> spawnBroker BlockingUnlimited bCfg+ deliver_ b (1 :: Int)+ deliver_ b (1 :: Int)+ deliver_ b (2 :: Int)+ takeMVar done+ cancel bA+ waitCatch bA+ >>= liftIO+ . assertEqual "exit expected" "Left AsyncCancelled"+ . show+ -- prevent GC of msg box input:+ deliver_ b 666+ ]++newtype Interceptor = MkInterceptor+ { runInterceptor ::+ forall m.+ MonadUnliftIO m =>+ m (Maybe Bool)+ }++data InterceptingBoxArg b+ = MkInterceptingBoxArg+ Interceptor -- receive interceptor+ Interceptor -- deliver interceptor+ b++data InterceptingBox b a+ = MkInterceptingBox+ Interceptor -- receive interceptor+ Interceptor -- deliver interceptor+ (b a)++data InterceptingBoxIn i a+ = MkInterceptingBoxIn+ Interceptor+ (i a)++instance IsMessageBoxArg b => IsMessageBoxArg (InterceptingBoxArg b) where+ type MessageBox (InterceptingBoxArg b) = InterceptingBox (MessageBox b)+ newMessageBox (MkInterceptingBoxArg r i a) =+ MkInterceptingBox r i <$> newMessageBox a+ getConfiguredMessageLimit (MkInterceptingBoxArg _ _ a) =+ getConfiguredMessageLimit a++instance IsMessageBox b => IsMessageBox (InterceptingBox b) where+ type Input (InterceptingBox b) = InterceptingBoxIn (Input b)+ newInput (MkInterceptingBox _ i b) =+ MkInterceptingBoxIn i <$> newInput b+ receive (MkInterceptingBox r _ b) = do+ mi <- runInterceptor r+ case mi of+ Nothing ->+ receive b+ Just True ->+ receive b+ Just False ->+ return Nothing+ tryReceive (MkInterceptingBox r _ b) = do+ -- oh my ... dunno what to do in this case...+ _mi <- runInterceptor r+ tryReceive b++instance IsInput i => IsInput (InterceptingBoxIn i) where+ deliver (MkInterceptingBoxIn interceptIn i) x = do+ mi <- runInterceptor interceptIn+ case mi of+ Nothing ->+ deliver i x+ Just o ->+ return o
+ src-test/Main.hs view
@@ -0,0 +1,27 @@+module Main (main) where++import RIO+import qualified BrokerTest+import qualified PoolTest+import System.Environment (setEnv)+import Test.Tasty+ ( TestTree,+ defaultIngredients,+ defaultMainWithIngredients,+ testGroup,+ )+import Test.Tasty.Runners.Html (htmlRunner)++main :: IO ()+main =+ do+ setEnv "TASTY_NUM_THREADS" "1"+ defaultMainWithIngredients (htmlRunner : defaultIngredients) test++test :: TestTree+test =+ testGroup+ "Tests"+ [ BrokerTest.test,+ PoolTest.test+ ]
+ src-test/PoolTest.hs view
@@ -0,0 +1,366 @@+module PoolTest (test) where++import qualified Data.Atomics.Counter as Atomic+import RIO+import RIO.ProcessPool+import Test.Tasty+import Test.Tasty.HUnit+import Utils++test :: TestTree+test =+ let expectedException = stringException "expected exception"+ in testGroup+ "Pool"+ [ testGroup+ "empty pool"+ [ testCase+ "when broker creation throws an exception, the process doesn't\+ \ block and returns an error"+ $ runTestApp $ do+ let badPoolBox =+ MkMockBoxInit+ ( return+ ( MkMockBox+ (throwIO expectedException)+ (error "unexpected invokation: receive")+ (error "unexpected invokation: tryReceive")+ )+ )+ Nothing+ e <-+ fromLeft (error "error expected")+ <$> spawnPool @Int @() @(MockBoxInit (MockBox NoOpInput))+ badPoolBox+ BlockingUnlimited+ MkPoolWorkerCallback+ { runPoolWorkerCallback = const (const (return ()))+ }+ liftIO $+ assertEqual "error expected" (show expectedException) (show e),+ testCase+ "when receiving from the pool input throws an exception,\+ \ the pool exits with that exception"+ $ runTestApp $ do+ let badPoolBox =+ MkMockBoxInit+ ( return+ ( MkMockBox+ ( return+ (OnDeliver (error "unexpected invokation: OnDeliver"))+ )+ (throwIO expectedException)+ (error "unexpected invokation: tryReceive")+ )+ )+ Nothing+ pool <-+ fromRight (error "failed to start pool")+ <$> spawnPool @Int @()+ badPoolBox+ BlockingUnlimited+ MkPoolWorkerCallback+ { runPoolWorkerCallback = const (const (return ()))+ }+ e <-+ fromLeft (error "error expected")+ <$> waitCatch (poolAsync pool)+ liftIO $+ assertEqual+ "error expected"+ (show expectedException)+ (show e)+ ],+ testGroup+ "non-empty pool"+ [ testCase+ "when delivering an Initialize,\+ \ the worker callback is executed in a new process"+ $ runTestApp $ do+ workerIdRef <- newEmptyMVar+ let cb = MkPoolWorkerCallback $ \_k _box ->+ myThreadId >>= putMVar workerIdRef+ pool <-+ either (error . show) id+ <$> spawnPool @Int BlockingUnlimited BlockingUnlimited cb+ deliver_ (poolInput pool) (Initialize 777 (Just (Just ())))+ workerId <- takeMVar workerIdRef+ cancel (poolAsync pool)+ liftIO $+ assertBool+ "worker and broker threadIds must be different"+ (asyncThreadId (poolAsync pool) /= workerId),+ testCase+ "when delivering an Initialize and the worker message box creation fails,\+ \ the worker will be cleaned up and not be in the pool"+ $ runTestApp $ do+ workerIdRef <- newEmptyMVar+ let badWorkerBox =+ MkMockBoxInit+ ( do+ myThreadId >>= putMVar workerIdRef+ throwIO expectedException+ )+ Nothing+ let cb = MkPoolWorkerCallback $ \_k _box ->+ error "unexpected invokation: callback"+ pool <-+ fromRight (error "failed to start pool")+ <$> spawnPool @Int @_ @_ @(MockBoxInit (MockBox NoOpInput))+ BlockingUnlimited+ badWorkerBox+ cb+ deliver_ (poolInput pool) (Initialize 777 (Just (Just ())))+ workerId1 <- takeMVar workerIdRef+ deliver_ (poolInput pool) (Initialize 777 (Just (Just ())))+ workerId2 <- takeMVar workerIdRef+ cancel (poolAsync pool)+ liftIO $+ assertBool+ "the broken worker must be removed from the pool"+ (workerId1 /= workerId2),+ testCase+ "when delivering an Initialize and the worker message box input creation fails,\+ \ the worker will be cleaned up and not be in the pool"+ $ runTestApp $ do+ workerIdRef <- newEmptyMVar+ let badWorkerBox =+ MkMockBoxInit+ ( return+ ( MkMockBox+ ( do+ myThreadId >>= putMVar workerIdRef+ throwIO expectedException+ )+ (error "unexpected invokation: receive")+ (error "unexpected invokation: tryReceive")+ )+ )+ Nothing+ let cb = MkPoolWorkerCallback $ \_k _box ->+ error "unexpected invokation: callback"+ pool <-+ fromRight (error "failed to start pool")+ <$> spawnPool @Int @_ @_ @(MockBoxInit (MockBox NoOpInput))+ BlockingUnlimited+ badWorkerBox+ cb+ deliver_ (poolInput pool) (Initialize 777 (Just (Just ())))+ workerId1 <- takeMVar workerIdRef+ deliver_ (poolInput pool) (Initialize 777 (Just (Just ())))+ workerId2 <- takeMVar workerIdRef+ cancel (poolAsync pool)+ liftIO $+ assertBool+ "the broken worker must be removed from the pool"+ (workerId1 /= workerId2),+ testCase+ "when delivering an 'Initialize (Just Nothing)',\+ \ the worker will be cleaned up and not be in the pool"+ $ runTestApp $ do+ counter <- liftIO $ Atomic.newCounter 0+ let cb = MkPoolWorkerCallback $ \_k box -> do+ liftIO $ Atomic.incrCounter_ 1 counter+ receive box+ >>= maybe+ (return ())+ (either id (`putMVar` ()))+ pool <-+ fromRight (error "failed to start pool")+ <$> spawnPool @Int+ BlockingUnlimited+ BlockingUnlimited+ cb+ deliver_ (poolInput pool) (Initialize 777 (Just Nothing))+ threadDelay 1000+ -- must not happen:+ failed <- newEmptyMVar+ deliver_ (poolInput pool) (Dispatch 777 (Just (Left (putMVar failed ()))))+ timeout 10000 (takeMVar failed)+ >>= liftIO+ . assertBool "the worker should not be running"+ . isNothing+ deliver_ (poolInput pool) (Initialize 777 Nothing)+ done <- newEmptyMVar+ deliver_ (poolInput pool) (Dispatch 777 (Just (Right done)))+ takeMVar done+ liftIO+ ( Atomic.readCounter counter+ >>= assertEqual "invalid number of pool worker callback invokations" 2+ )+ cancel (poolAsync pool),+ testCase+ "when several workers are initialized with different keys,\+ \ all are created and available in the pool."+ $ runTestApp $ do+ startedRef <- newEmptyMVar+ let cb = MkPoolWorkerCallback $ \k box -> do+ putMVar startedRef k+ fix $ \again -> do+ m <- receive box+ case m of+ Nothing -> return ()+ Just ref -> do+ putMVar ref k+ again++ pool <-+ fromRight (error "failed to start pool")+ <$> spawnPool @Int+ BlockingUnlimited+ BlockingUnlimited+ cb++ liftIO $ do+ deliver_ (poolInput pool) (Initialize 0 Nothing)+ takeMVar startedRef >>= assertEqual "wrong startedRef value" 0+ deliver_ (poolInput pool) (Initialize 1 Nothing)+ takeMVar startedRef >>= assertEqual "wrong startedRef value" 1+ deliver_ (poolInput pool) (Initialize 2 Nothing)+ takeMVar startedRef >>= assertEqual "wrong startedRef value" 2+ deliver_ (poolInput pool) (Initialize 3 Nothing)+ takeMVar startedRef >>= assertEqual "wrong startedRef value" 3++ workRef <- newEmptyMVar+ deliver_ (poolInput pool) (Dispatch 0 (Just workRef))+ takeMVar workRef >>= assertEqual "wrong workRef value" 0+ deliver_ (poolInput pool) (Dispatch 1 (Just workRef))+ takeMVar workRef >>= assertEqual "wrong workRef value" 1+ deliver_ (poolInput pool) (Dispatch 2 (Just workRef))+ takeMVar workRef >>= assertEqual "wrong workRef value" 2+ deliver_ (poolInput pool) (Dispatch 3 (Just workRef))+ takeMVar workRef >>= assertEqual "wrong workRef value" 3++ cancel (poolAsync pool),+ testCase+ "when 'Dispatch k (Just x)' is delivered, and a worker k exists,\+ \ the worker will receive the message x, otherwise it will silently be dropped."+ $ runTestApp $ do+ startedRef <- newEmptyMVar+ let cb = MkPoolWorkerCallback $ \k box -> do+ putMVar startedRef k+ fix $ \again -> do+ m <- receive box+ case m of+ Nothing -> return ()+ Just ref -> do+ putMVar ref k+ again++ pool <-+ fromRight (error "failed to start pool")+ <$> spawnPool @Int+ BlockingUnlimited+ BlockingUnlimited+ cb+ liftIO $ do+ deliver_ (poolInput pool) (Initialize 0 Nothing)+ takeMVar startedRef >>= assertEqual "wrong startedRef value" 0+ deliver_ (poolInput pool) (Initialize 1 Nothing)+ takeMVar startedRef >>= assertEqual "wrong startedRef value" 1++ workRef <- newEmptyMVar+ deliver_ (poolInput pool) (Dispatch 0 (Just workRef))+ takeMVar workRef >>= assertEqual "wrong workRef value" 0++ deliver_ (poolInput pool) (Dispatch 666 (Just workRef))+ timeout 10000 (takeMVar workRef)+ >>= assertEqual "wrong workRef value" Nothing++ cancel (poolAsync pool),+ testCase+ "when Dispatch k Nothing is delivered, and a worker k exists,\+ \ the worker will be cleaned up and removed from the pool"+ $ runTestApp $ do+ startedRef <- newEmptyMVar+ let cb = MkPoolWorkerCallback $ \k box -> do+ putMVar startedRef k+ fix $ \again -> do+ m <- receive box+ case m of+ Nothing -> return ()+ Just ref -> do+ putMVar ref k+ again++ pool <-+ fromRight (error "failed to start pool")+ <$> spawnPool @Int+ BlockingUnlimited+ BlockingUnlimited+ cb+ liftIO $ do+ deliver_ (poolInput pool) (Initialize 0 Nothing)+ takeMVar startedRef >>= assertEqual "wrong startedRef value" 0+ deliver_ (poolInput pool) (Initialize 1 Nothing)+ takeMVar startedRef >>= assertEqual "wrong startedRef value" 1++ workRef <- newEmptyMVar+ deliver_ (poolInput pool) (Dispatch 0 (Just workRef))+ takeMVar workRef >>= assertEqual "wrong workRef value" 0++ deliver_ (poolInput pool) (Dispatch 1 Nothing)++ deliver_ (poolInput pool) (Dispatch 1 (Just workRef))+ timeout 10000 (takeMVar workRef)+ >>= assertEqual "wrong workRef value" Nothing++ deliver_ (poolInput pool) (Dispatch 0 (Just workRef))+ takeMVar workRef >>= assertEqual "wrong workRef value" 0++ cancel (poolAsync pool),+ testCase+ "when 'deliver' to a worker message box returns False,\+ \ the worker is removed (and cancelled)"+ $ runTestApp $ do+ startedRef <- newEmptyMVar+ let cb = MkPoolWorkerCallback $ \k box -> do+ putMVar startedRef k+ fix $ \again -> do+ m <- receive box+ case m of+ Nothing -> return ()+ Just (Left ref) -> do+ putMVar ref k+ again+ Just (Right ()) -> do+ threadDelay 100000000++ pool <-+ fromRight (error "failed to start pool")+ <$> spawnPool @Int+ BlockingUnlimited+ (NonBlockingBoxLimit MessageLimit_2)+ cb++ liftIO $ do+ deliver_ (poolInput pool) (Initialize 0 Nothing)+ takeMVar startedRef >>= assertEqual "wrong startedRef value" 0++ deliver_ (poolInput pool) (Initialize 1 Nothing)+ takeMVar startedRef >>= assertEqual "wrong startedRef value" 1++ workRef <- newEmptyMVar++ deliver_ (poolInput pool) (Dispatch 0 (Just (Left workRef)))+ takeMVar workRef >>= assertEqual "wrong workRef value" 0++ -- Now overflow the input by first sending a Right message and then+ -- so many messages, that the message limit will cause 'deliver'+ -- to return 'False':+ replicateM_+ 16+ (deliver_ (poolInput pool) (Dispatch 0 (Just (Right ()))))++ -- Now the process should be dead:+ deliver_ (poolInput pool) (Dispatch 0 (Just (Left workRef)))+ timeout 10000 (takeMVar workRef)+ >>= assertEqual "wrong workRef value" Nothing++ -- ... while the other process should be alive:+ deliver_ (poolInput pool) (Dispatch 1 (Just (Left workRef)))+ takeMVar workRef >>= assertEqual "wrong workRef value" 1++ cancel (poolAsync pool)+ ]+ ]
+ src-test/Utils.hs view
@@ -0,0 +1,95 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}++module Utils+ ( MockBoxInit (..),+ MockBox (..),+ NoOpArg (..),+ NoOpBox (..),+ NoOpInput (..),+ runTestApp,+ )+where++import RIO+import UnliftIO.MessageBox.Class+ ( IsInput (..),+ IsMessageBox (Input, newInput, receive, tryReceive),+ IsMessageBoxArg (..),+ )+import UnliftIO.MessageBox.Util.Future (Future (Future))++runTestApp :: MonadUnliftIO m => RIO LogFunc b -> m b+runTestApp x = do+ let isVerbose = True+ logToStdErr <- logOptionsHandle stdout isVerbose+ withLogFunc logToStdErr $ flip runRIO x++-- message box implementation+-- NOTE: Because of parametricity and the existential quantification+-- of the message payload, the receive and deliver methods+-- are only capable of throwing exceptions or bottoming out++data MockBoxInit msgBox = MkMockBoxInit+ { mockBoxNew :: forall m a. MonadUnliftIO m => m (msgBox a),+ mockBoxLimit :: !(Maybe Int)+ }++instance IsMessageBox msgBox => IsMessageBoxArg (MockBoxInit msgBox) where+ type MessageBox (MockBoxInit msgBox) = msgBox+ getConfiguredMessageLimit = mockBoxLimit+ newMessageBox b = mockBoxNew b++data MockBox input a = MkMockBox+ { mockBoxNewInput :: forall m. MonadUnliftIO m => m (input a),+ mockBoxReceive :: forall m. MonadUnliftIO m => m (Maybe a),+ mockBoxTryReceive :: forall m. MonadUnliftIO m => m (Future a)+ }++instance IsInput input => IsMessageBox (MockBox input) where+ type Input (MockBox input) = input+ newInput m = mockBoxNewInput m+ receive m = mockBoxReceive m+ tryReceive m = mockBoxTryReceive m++data NoOpArg+ = NoOpArg+ 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 IsMessageBoxArg NoOpArg where+ type MessageBox NoOpArg = NoOpBox+ newMessageBox NoOpArg = 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/RIO/ProcessPool.hs view
@@ -0,0 +1,53 @@+-- | Launch- and Dispatch messages to processes.+--+-- A pool has an 'Input' for 'Multiplexed' messages,+-- and dispatches incoming messges to concurrent+-- processes using user defined @'MessageBox'es@.+--+-- The pool starts and stops the processes and+-- creates the message boxes.+--+-- The user supplied 'PoolWorkerCallback'+-- usually runs a loop that @'receive's@ messages+-- from the 'MessageBox' created by the pool for that worker.+--+-- When a worker process dies, e.g. because the+-- 'PoolWorkerCallback' returns, the pool+-- process will also 'cancel' the process (just to make sure...)+-- and cleanup the internal 'Broker'.+module RIO.ProcessPool+ ( -- | A process that receives messages and dispatches them to + -- a callback. + -- Each message must contain a /key/ that identifies a resource.+ -- That resource is created and cleaned by user supplied + -- callback functions. + module RIO.ProcessPool.Broker,+ -- | A process that receives messages and dispatches them to + -- other processes.+ -- Building directly on "RIO.ProcessPool.Broker", it provides + -- a central message box 'Input', from which messages are + -- are delivered to the corresponding message box 'Input's.+ module RIO.ProcessPool.Pool,+ -- | Re-export.+ module UnliftIO.MessageBox,+ )+where++import RIO.ProcessPool.Broker+ ( BrokerConfig (..),+ BrokerResult (..),+ Demultiplexer,+ MessageHandler,+ Multiplexed (..),+ ResourceCleaner,+ ResourceCreator,+ ResourceUpdate (..),+ spawnBroker,+ )+import RIO.ProcessPool.Pool+ ( Pool (..),+ PoolWorkerCallback (..),+ removePoolWorkerMessage,+ spawnPool,+ )+import UnliftIO.MessageBox
+ src/RIO/ProcessPool/Broker.hs view
@@ -0,0 +1,390 @@+{-# LANGUAGE Strict #-}++-- | A broker extracts a /key/ value from incoming messages+-- and creates, keeps and destroys a /resource/ for each key.+--+-- The demultiplexed messages and their resources are passed to+-- a custom 'MessageHandler'/+--+-- The user provides a 'Demultiplexer' is a pure function that+-- returns a key for the resource associated+-- to the message and potientially changes the+-- message.+--+-- The demultiplexer may also return a value indicating that+-- a new resource must be created, or that a message+-- shall be ignored.+--+-- The broker is run in a seperate process using 'async'.+-- The usual way to stop a broker is to 'cancel' it.+--+-- When cancelling a broker, the resource cleanup+-- actions for all resources will be called with+-- async exceptions masked.+--+-- In order to prevent the resource map filling up with+-- /dead/ resources, the user of this module has to ensure+-- that whenever a resource is not required anymore, a message+-- will be sent to the broker, that will cause the 'MessageHandler'+-- to be executed for the resource, which will in turn return,+-- return 'RemoveResource'.+module RIO.ProcessPool.Broker+ ( spawnBroker,+ BrokerConfig (..),+ BrokerResult (..),+ ResourceCreator,+ Demultiplexer,+ ResourceCleaner,+ MessageHandler,+ Multiplexed (..),+ ResourceUpdate (..),+ )+where++import qualified Data.Map.Strict as Map+import RIO+import UnliftIO.MessageBox+ ( IsMessageBox (Input, newInput, receive),+ IsMessageBoxArg (MessageBox, newMessageBox),+ )+import Control.Concurrent.Async(AsyncCancelled) ++-- | Spawn a broker with a new 'MessageBox',+-- and return its message 'Input' channel as well as+-- the 'Async' handle of the spawned process, needed to+-- stop the broker process.+--+-- * @k@ is the /key/ for the resource associated to an incoming+-- message+-- * @w'@ is the type of incoming messages.+-- * @w@ is the type of the demultiplexed messages.+-- * @a@ specifies the resource type.+-- * @m@ is the base monad+spawnBroker ::+ forall brokerBoxArg k w' w a m.+ ( HasLogFunc m,+ Ord k,+ Display k,+ IsMessageBoxArg brokerBoxArg+ ) =>+ brokerBoxArg ->+ BrokerConfig k w' w a m ->+ RIO+ m+ ( Either+ SomeException+ ( Input (MessageBox brokerBoxArg) w',+ Async BrokerResult+ )+ )+spawnBroker brokerBoxArg config = do+ brokerA <- async $ do+ mBrokerBox <-+ tryAny+ ( do+ b <- newMessageBox brokerBoxArg+ i <- newInput b+ return (b, i)+ )+ case mBrokerBox of+ Left er -> return (Left er)+ Right (brokerBox, brokerInp) -> do+ aInner <- mask_ $+ asyncWithUnmask $ \unmaskInner ->+ brokerLoop unmaskInner brokerBox config Map.empty+ return (Right (brokerInp, aInner))+ join <$> waitCatch brokerA++-- | This is just what the 'Async' returned from+-- 'spawnBroker' returns, it's current purpose is to+-- make code easier to read.+--+-- Instead of some @Async ()@ that could be anything,+-- there is @Async BrokerResult@.+data BrokerResult = MkBrokerResult+ deriving stock (Show, Eq)++-- | The broker configuration, used by 'spawnBroker'.+--+-- * @k@ is the /key/ for the resource associated to an incoming+-- message+-- * @w'@ is the type of incoming messages.+-- * @w@ is the type of the demultiplexed messages.+-- * @a@ specifies the resource type.+-- * @m@ is the base monad+data BrokerConfig k w' w a m = MkBrokerConfig+ { demultiplexer :: !(Demultiplexer w' k w),+ messageDispatcher :: !(MessageHandler k w a m),+ resourceCreator :: !(ResourceCreator k w a m),+ resourceCleaner :: !(ResourceCleaner k a m)+ }++-- | User supplied callback to extract the key and the 'Multiplexed'+-- from a message.+-- (Sync-) Exceptions thrown from this function are caught and lead+-- to dropping of the incoming message, while the broker continues.+--+-- * @k@ is the /key/ for the resource associated to an incoming+-- message+-- * @w'@ is the type of incoming messages.+-- * @w@ is the type of the demultiplexed messages.+type Demultiplexer w' k w = w' -> Multiplexed k w++-- | User supplied callback to use the 'Multiplexed' message and+-- the associated resource.+-- (Sync-) Exceptions thrown from this function are caught and lead+-- to immediate cleanup of the resource but the broker continues.+--+-- * Type @k@ is the /key/ for the resource associated to an incoming+-- message+-- * Type @w@ is the type of incoming, demultiplexed, messages.+-- * Type @a@ specifies the resource type.+-- * Type @m@ is the base monad+type MessageHandler k w a m = k -> w -> a -> RIO m (ResourceUpdate a)++-- | This value indicates in what state a worker is in after the+-- 'MessageHandler' action was executed.+data ResourceUpdate a+ = -- | The resources is still required.+ KeepResource+ | -- | The resource is still required but must be updated.+ UpdateResource a+ | -- | The resource is obsolete and can+ -- be removed from the broker.+ -- The broker will call 'ResourceCleaner' either+ -- on the current, or an updated resource value.+ RemoveResource !(Maybe a)++-- | The action that the broker has to take for in incoming message.+--+-- * @k@ is the /key/ for the resource associated to an incoming+-- message+-- * @w@ is the type of the demultiplexed messages.+data Multiplexed k w+ = -- | The message is an initialization message, that requires the+ -- creation of a new resouce for the given key.+ -- When the resource is created, then /maybe/ additionally+ -- a message will also be dispatched.+ Initialize k !(Maybe w)+ | -- | Dispatch a message using an existing resource.+ -- Silently ignore if no resource for the key exists.+ Dispatch k w++-- deriving stock (Show)++-- | User supplied callback to create and initialize a resource.+-- (Sync-) Exceptions thrown from this function are caught,+-- and the broker continues.+--+-- * @k@ is the /key/ for the resource associated to an incoming+-- message+-- * @w@ is the type of the demultiplexed messages.+-- * @a@ specifies the resource type.+-- * @m@ is the monad of the returned action.+type ResourceCreator k w a m = k -> Maybe w -> RIO m a++-- | User supplied callback called _with exceptions masked_+-- when the 'MessageHandler' returns 'RemoveResource'+-- (Sync-) Exceptions thrown from this function are caught,+-- and do not prevent the removal of the resource, also the+-- broker continues.+--+-- * @k@ is the /key/ for the resource associated to an incoming+-- message+-- * @a@ specifies the resource type.+-- * @m@ is the monad of the returned action.+type ResourceCleaner k a m = k -> a -> RIO m ()++type BrokerState k a = Map k a++{-# NOINLINE brokerLoop #-}+brokerLoop ::+ ( HasLogFunc m,+ Ord k,+ Display k,+ IsMessageBox msgBox+ ) =>+ (forall x. RIO m x -> RIO m x) ->+ msgBox w' ->+ BrokerConfig k w' w a m ->+ BrokerState k a ->+ RIO m BrokerResult+brokerLoop unmask brokerBox config brokerState =+ withException+ ( unmask (receive brokerBox)+ >>= traverse (tryAny . onIncoming unmask config brokerState)+ )+ ( \(ex :: SomeException) -> do + case fromException ex of+ Just (_cancelled :: AsyncCancelled) ->+ logDebug "broker loop: cancelled"+ _ ->+ logError+ ( "broker loop: exception while \+ \receiving and dispatching messages: "+ <> display ex+ )+ cleanupAllResources config brokerState+ )+ >>= maybe+ ( do+ logError "broker loop: failed to receive next message"+ cleanupAllResources config brokerState+ return MkBrokerResult+ )+ ( \res -> do+ next <-+ either+ ( \err -> do+ logWarn+ ( "broker loop: Handling the last message\+ \ caused an exception:"+ <> display err+ )+ return brokerState+ )+ return+ res+ brokerLoop unmask brokerBox config next+ )++{-# NOINLINE onIncoming #-}+onIncoming ::+ (Ord k, HasLogFunc m, Display k) =>+ (forall x. RIO m x -> RIO m x) ->+ BrokerConfig k w' w a m ->+ BrokerState k a ->+ w' ->+ RIO m (BrokerState k a)+onIncoming unmask config brokerState w' =+ case demultiplexer config w' of+ Initialize k mw ->+ onInitialize unmask k config brokerState mw+ Dispatch k w ->+ onDispatch unmask k w config brokerState++onInitialize ::+ (Ord k, HasLogFunc m, Display k) =>+ (forall x. RIO m x -> RIO m x) ->+ k ->+ BrokerConfig k w' w a m ->+ BrokerState k a ->+ Maybe w ->+ RIO m (BrokerState k a)+onInitialize unmask k config brokerState mw =+ case Map.lookup k brokerState of+ Just _ -> do+ logError+ ( "cannot initialize a new worker, a worker with that ID exists: "+ <> display k+ )+ return brokerState+ Nothing ->+ tryAny (unmask (resourceCreator config k mw))+ >>= either+ ( \err -> do+ logError+ ( "the resource creator for worker "+ <> display k+ <> " threw an exception: "+ <> display err+ )+ return brokerState+ )+ ( \res ->+ let brokerState1 = Map.insert k res brokerState+ in case mw of+ Nothing ->+ return brokerState1+ Just w ->+ onException+ (onDispatch unmask k w config brokerState1)+ ( do+ logError+ ( "exception while dispatching the "+ <> "post-initialization message for worker: "+ <> display k+ )+ resourceCleaner config k res+ )+ )++onDispatch ::+ (Ord k, HasLogFunc m, Display k) =>+ (forall x. RIO m x -> RIO m x) ->+ k ->+ w ->+ BrokerConfig k w' w a m ->+ BrokerState k a ->+ RIO m (BrokerState k a)+onDispatch unmask k w config brokerState =+ maybe notFound dispatch (Map.lookup k brokerState)+ where+ notFound = do+ logWarn+ ( "cannot dispatch message, worker not found: "+ <> display k+ )+ return brokerState+ dispatch res =+ tryAny (unmask (messageDispatcher config k w res))+ >>= either+ ( \err -> do++ logError+ ( "the message dispatcher callback for worker "+ <> display k+ <> " threw: "+ <> display err+ )+ cleanupResource+ k+ config+ brokerState+ )+ ( \case+ KeepResource ->+ return brokerState+ UpdateResource newRes ->+ return (Map.insert k newRes brokerState)+ RemoveResource mNewRes ->+ cleanupResource+ k+ config+ ( maybe+ brokerState+ ( \newRes ->+ Map.insert k newRes brokerState+ )+ mNewRes+ )+ )++cleanupAllResources ::+ BrokerConfig k w' w a m ->+ BrokerState k a ->+ RIO m ()+cleanupAllResources config brokerState =+ traverse_+ ( uncurry+ (tryResourceCleaner config)+ )+ (Map.assocs brokerState)++cleanupResource ::+ (Ord k) =>+ k ->+ BrokerConfig k w' w a m ->+ Map k a ->+ RIO m (Map k a)+cleanupResource k config brokerState = do+ traverse_ (tryResourceCleaner config k) (Map.lookup k brokerState)+ return (Map.delete k brokerState)++tryResourceCleaner ::+ BrokerConfig k w' w a m ->+ k ->+ a ->+ RIO m ()+tryResourceCleaner config k res = do+ void $ tryAny (resourceCleaner config k res)
+ src/RIO/ProcessPool/Pool.hs view
@@ -0,0 +1,201 @@+-- | Launch- and Dispatch messages to processes.+--+-- A pool has an 'Input' for 'Multiplexed' messages,+-- and dispatches incoming messges to concurrent+-- processes using user defined @'MessageBox'es@.+--+-- The pool starts and stops the processes and+-- creates the message boxes.+--+-- The user supplied 'PoolWorkerCallback' +-- usually runs a loop that @'receive's@ messages+-- from the 'MessageBox' created by the pool for that worker.+--+-- When a worker process dies, e.g. because the +-- 'PoolWorkerCallback' returns, the pool+-- process will also 'cancel' the process (just to make sure...)+-- and cleanup the internal 'Broker'.+module RIO.ProcessPool.Pool+ ( Pool (..),+ spawnPool,+ PoolWorkerCallback (..),+ removePoolWorkerMessage,+ )+where++import RIO+import RIO.ProcessPool.Broker+ ( BrokerConfig (MkBrokerConfig),+ BrokerResult,+ Multiplexed (Dispatch),+ ResourceUpdate (KeepResource, RemoveResource),+ spawnBroker,+ )+import UnliftIO.MessageBox.Class+ ( IsInput (deliver),+ IsMessageBox (Input, newInput),+ IsMessageBoxArg (MessageBox, newMessageBox),+ )++-- | Start a 'Pool'.+--+-- Start a process that receives messages sent to the+-- 'poolInput' and dispatches them to the 'Input' of+-- __pool member__ processes. If necessary the+-- pool worker processes are started.+--+-- Each pool worker process is started using 'async' and+-- executes the 'PoolWorkerCallback'.+--+-- When the callback returns, the process will exit.+--+-- Internally the pool uses the 'async' function to wrap+-- the callback.+--+-- When a 'Multiplixed' 'Dispatch' message is received with+-- a @Nothing@ then the worker is @'cancel'led@ and the+-- worker is removed from the map.+--+-- Such a message is automatically sent after the 'PoolWorkerCallback'+-- has returned, even when an exception was thrown. See+-- 'finally'.+spawnPool ::+ forall k w poolBox workerBox m.+ ( IsMessageBoxArg poolBox,+ IsMessageBoxArg workerBox,+ Ord k,+ Display k,+ HasLogFunc m+ ) =>+ poolBox ->+ workerBox ->+ PoolWorkerCallback workerBox w k m ->+ RIO+ m+ ( Either+ SomeException+ (Pool poolBox k w)+ )+spawnPool poolBox workerBoxArg poolMemberImpl = do+ brInRef <- newEmptyMVar+ let brCfg =+ MkBrokerConfig+ id+ dispatchToWorker+ (spawnWorker workerBoxArg brInRef poolMemberImpl)+ removeWorker+ spawnBroker poolBox brCfg+ >>= traverse+ ( \(brIn, brA) -> do+ putMVar brInRef brIn+ return MkPool {poolInput = brIn, poolAsync = brA}+ )++-- | This message will 'cancel' the worker+-- with the given key.+-- If the 'PoolWorkerCallback' wants to do cleanup+-- it should use 'finally' or 'onException'.+removePoolWorkerMessage :: k -> Multiplexed k (Maybe w)+removePoolWorkerMessage !k = Dispatch k Nothing++-- | The function that processes a+-- 'MessageBox' of a worker for a specific /key/.+newtype PoolWorkerCallback workerBox w k m = MkPoolWorkerCallback+ { runPoolWorkerCallback :: k -> MessageBox workerBox w -> RIO m ()+ }++-- | A record containing the message box 'Input' of the+-- 'Broker' and the 'Async' value required to 'cancel'+-- the pools broker process.+data Pool poolBox k w = MkPool+ { -- | Message sent to this input are dispatched to workers.+ -- If the message is an 'Initialize' message, a new 'async'+ -- process will be started.+ -- If the message value is 'Nothing', the processes is killed.+ poolInput :: !(Input (MessageBox poolBox) (Multiplexed k (Maybe w))),+ -- | The async of the internal 'Broker'.+ poolAsync :: !(Async BrokerResult)+ }++-- | Internal data structure containing a workers+-- message 'Input' and 'Async' value for cancellation.+data PoolWorker workerBox w = MkPoolWorker+ { poolWorkerIn :: !(Input (MessageBox workerBox) w),+ poolWorkerAsync :: !(Async ())+ }++dispatchToWorker ::+ (HasLogFunc m, IsInput (Input (MessageBox b)), Display k) =>+ k ->+ Maybe w ->+ PoolWorker b w ->+ RIO m (ResourceUpdate (PoolWorker b w))+dispatchToWorker k pMsg pm =+ case pMsg of+ Just w -> helper w+ Nothing -> return (RemoveResource Nothing)+ where+ helper msg = do+ ok <- deliver (poolWorkerIn pm) msg+ if not ok+ then do+ logError ("failed to deliver message to pool worker: " <> display k)+ return (RemoveResource Nothing)+ else return KeepResource++spawnWorker ::+ forall k w poolBoxIn workerBox m.+ ( IsMessageBoxArg workerBox,+ HasLogFunc m,+ IsInput poolBoxIn,+ Display k+ ) =>+ workerBox ->+ MVar (poolBoxIn (Multiplexed k (Maybe w))) ->+ PoolWorkerCallback workerBox w k m ->+ k ->+ Maybe (Maybe w) ->+ RIO m (PoolWorker workerBox w)+spawnWorker workerBox brInRef pmCb this _mw = do+ inputRef <- newEmptyMVar+ a <- async (go inputRef `finally` enqueueCleanup)+ boxInM <- takeMVar inputRef+ case boxInM of+ Nothing -> do+ cancel a+ throwIO (stringException "failed to spawnWorker")+ Just boxIn ->+ return MkPoolWorker {poolWorkerIn = boxIn, poolWorkerAsync = a}+ where+ go inputRef = do+ (b, boxIn) <-+ withException+ ( do+ b <- newMessageBox workerBox+ boxIn <- newInput b+ return (b, boxIn)+ )+ (\(ex :: SomeException) -> do+ logError+ ( "failed to create the message box for the new pool worker: "+ <> display this+ <> " exception caught: "+ <> display ex+ )+ putMVar inputRef Nothing+ )+ putMVar inputRef (Just boxIn)+ runPoolWorkerCallback pmCb this b+ enqueueCleanup =+ tryReadMVar brInRef+ >>= traverse_+ ( \brIn ->+ void (deliver brIn (removePoolWorkerMessage this))+ )++removeWorker :: + k ->+ PoolWorker workerBox w ->+ RIO m ()+removeWorker _k =+ void . cancel . poolWorkerAsync