poolboy (empty) → 0.1.0.0
raw patch · 4 files changed
+324/−0 lines, 4 filesdep +basedep +hspecdep +hspec-core
Dependencies added: base, hspec, hspec-core, poolboy, safe-exceptions, stm
Files
- LICENSE +15/−0
- poolboy.cabal +86/−0
- src/Data/Poolboy.hs +182/−0
- test/Spec.hs +41/−0
+ LICENSE view
@@ -0,0 +1,15 @@+ISC License++Copyright (c) 2023 Gautier DI FOLCO++Permission to use, copy, modify, and/or distribute this software for any+purpose with or without fee is hereby granted, provided that the above+copyright notice and this permission notice appear in all copies.++THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH+REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY+AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,+INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM+LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR+OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR+PERFORMANCE OF THIS SOFTWARE.
+ poolboy.cabal view
@@ -0,0 +1,86 @@+cabal-version: 3.0+name: poolboy+version: 0.1.0.0+author: Gautier DI FOLCO+maintainer: gautier.difolco@gmail.com+category: Data+build-type: Simple+license: ISC+license-file: LICENSE+synopsis: Simple work queue for bounded concurrency+description: In-memory work queue with helping with load management.+Homepage: https://github.com/blackheaven/poolboy+tested-with: GHC==9.2.4++library+ default-language: Haskell2010+ build-depends:+ base == 4.*+ , safe-exceptions == 0.1.*+ , stm == 2.*+ hs-source-dirs: src+ exposed-modules:+ Data.Poolboy+ other-modules:+ Paths_poolboy+ autogen-modules:+ Paths_poolboy+ default-extensions:+ DataKinds+ DefaultSignatures+ DeriveAnyClass+ DeriveGeneric+ DerivingStrategies+ DerivingVia+ DuplicateRecordFields+ FlexibleContexts+ GADTs+ GeneralizedNewtypeDeriving+ KindSignatures+ LambdaCase+ OverloadedStrings+ OverloadedRecordDot+ RankNTypes+ RecordWildCards+ ScopedTypeVariables+ TypeApplications+ TypeFamilies+ TypeOperators+ ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints++test-suite poolboy-test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ other-modules:+ Paths_poolboy+ autogen-modules:+ Paths_poolboy+ default-extensions:+ DataKinds+ DefaultSignatures+ DeriveAnyClass+ DeriveGeneric+ DerivingStrategies+ DerivingVia+ DuplicateRecordFields+ FlexibleContexts+ GADTs+ GeneralizedNewtypeDeriving+ KindSignatures+ LambdaCase+ OverloadedStrings+ OverloadedRecordDot+ RankNTypes+ RecordWildCards+ ScopedTypeVariables+ TypeApplications+ TypeFamilies+ TypeOperators+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints+ build-depends:+ base+ , poolboy+ , hspec+ , hspec-core+ default-language: Haskell2010
+ src/Data/Poolboy.hs view
@@ -0,0 +1,182 @@+module Data.Poolboy+ ( -- * Configuration+ PoolboySettings (..),+ WorkersCountSettings (..),+ defaultPoolboySettings,+ poolboySettingsWith,+ simpleSerializedLogger,++ -- * Running+ withPoolboy,+ newPoolboy,+ stopWorkQueue,+ isStopWorkQueue,++ -- * Driving+ changeDesiredWorkersCount,+ waitReadyQueue,++ -- * Enqueueing+ enqueue,+ enqueueAfter,+ )+where++import Control.Concurrent+import Control.Concurrent.STM.TQueue+import Control.Exception.Safe (bracket, tryAny)+import Control.Monad+import Control.Monad.STM+import Data.IORef++-- | Initial settings+data PoolboySettings = PoolboySettings+ { workersCount :: WorkersCountSettings,+ log :: String -> IO ()+ }++-- | Initial number of threads+data WorkersCountSettings+ = -- | 'getNumCapabilities' based number+ CapabilitiesWCS+ | FixedWCS Int -- arbitrary number+ deriving stock (Eq, Show)++-- | Usual configuration 'CapabilitiesWCS' and no log+defaultPoolboySettings :: PoolboySettings+defaultPoolboySettings =+ PoolboySettings+ { workersCount = CapabilitiesWCS,+ log = \_ -> return ()+ }++-- | Arbitrary-numbered settings+poolboySettingsWith :: Int -> PoolboySettings+poolboySettingsWith c = defaultPoolboySettings {workersCount = FixedWCS c}++-- | Simple (but not particularly performant) serialized logger+simpleSerializedLogger :: IO (String -> IO ())+simpleSerializedLogger = do+ logLock <- newMVar ()+ return $ \x ->+ withMVar logLock $ \() -> do+ putStrLn x+ return ()++-- | 'backet'-based usage (recommended)+withPoolboy :: PoolboySettings -> (WorkQueue -> IO a) -> IO a+withPoolboy settings = bracket (newPoolboy settings) (\wq -> stopWorkQueue wq >> waitStopWorkQueue wq)++-- | Standalone/manual usage+newPoolboy :: PoolboySettings -> IO WorkQueue+newPoolboy settings = do+ wq <-+ WorkQueue+ <$> newTQueueIO+ <*> newTQueueIO+ <*> newIORef 0+ <*> newEmptyMVar+ <*> return settings.log++ count <-+ case settings.workersCount of+ CapabilitiesWCS -> getNumCapabilities+ FixedWCS x -> return x++ changeDesiredWorkersCount wq count+ void $ forkIO $ controller wq++ return wq++-- | Request a worker number adjustment+changeDesiredWorkersCount :: WorkQueue -> Int -> IO ()+changeDesiredWorkersCount wq =+ atomically . writeTQueue wq.commands . ChangeDesiredWorkersCount++-- | Request stopping wrokers+stopWorkQueue :: WorkQueue -> IO ()+stopWorkQueue wq =+ atomically $ writeTQueue wq.commands Stop++-- | Non-blocking check of the work queue's running status+isStopWorkQueue :: WorkQueue -> IO Bool+isStopWorkQueue wq =+ not <$> isEmptyMVar wq.stopped++-- | Block until the queue is totally stopped (no more running worker)+waitStopWorkQueue :: WorkQueue -> IO ()+waitStopWorkQueue wq =+ void $ tryAny $ readMVar wq.stopped++-- | Enqueue one action in the work queue (non-blocking)+enqueue :: WorkQueue -> IO () -> IO ()+enqueue wq =+ atomically . writeTQueue wq.queue . Right++-- | Block until one worker is available+waitReadyQueue :: WorkQueue -> IO ()+waitReadyQueue wq = do+ ready <- newEmptyMVar+ enqueue wq $ putMVar ready ()+ readMVar ready++-- | Enqueue action and some actions to be run after it+enqueueAfter :: Foldable f => WorkQueue -> IO () -> f (IO ()) -> IO ()+enqueueAfter wq x xs =+ enqueue wq $ do+ x+ forM_ xs $ enqueue wq++-- Support (internal)+data WorkQueue = WorkQueue+ { commands :: TQueue Commands,+ queue :: TQueue (Either () (IO ())),+ currentWorkersCount :: IORef Int,+ stopped :: MVar (),+ log :: String -> IO ()+ }++data Commands+ = ChangeDesiredWorkersCount Int+ | Stop+ deriving stock (Show)++controller :: WorkQueue -> IO ()+controller wq = do+ command <- atomically $ readTQueue wq.commands+ let stopOneWorker = atomically $ writeTQueue wq.queue $ Left ()+ wq.log $ "Command: " <> show command+ case command of+ ChangeDesiredWorkersCount n -> do+ currentCount <- readIORef wq.currentWorkersCount+ let diff = currentCount - n+ if diff > 0+ then replicateM_ diff stopOneWorker+ else replicateM_ (abs diff) $ do+ wq.log "Pre-fork"+ forkIO $ worker wq+ controller wq+ Stop -> do+ currentCount <- readIORef wq.currentWorkersCount+ wq.log $ "Stopping " <> show currentCount <> " workers"+ replicateM_ currentCount stopOneWorker++worker :: WorkQueue -> IO ()+worker wq = do+ wq.log "New worker"+ newCount <- atomicModifyIORef' wq.currentWorkersCount $ \n -> (n + 1, n + 1)+ wq.log $ "New worker count " <> show newCount+ let loop = do+ command <- atomically $ readTQueue wq.queue+ case command of+ Left () -> do+ wq.log "Stopping"+ remaining <-+ atomicModifyIORef' wq.currentWorkersCount $ \n ->+ let count = max 0 (n - 1) in (count, count)+ wq.log $ "Remaining: " <> show remaining+ when (remaining == 0) $+ void $+ tryPutMVar wq.stopped ()+ Right act -> wq.log "pop" >> void (tryAny act) >> wq.log "poped" >> loop+ loop
+ test/Spec.hs view
@@ -0,0 +1,41 @@+module Main (main) where++import Control.Concurrent+import Control.Exception (Exception (..), throw)+import Control.Monad+import Data.IORef+import Data.Maybe+import Data.Poolboy+import System.Timeout+import Test.Hspec++main :: IO ()+main = hspec spec++spec :: Spec+spec =+ describe "Poolboy" $ do+ it "threadDelay should be absorbed in mulitple threads" $ do+ computations <-+ timeout 2500 $+ withPoolboy (poolboySettingsWith 100) $ \wq -> do+ replicateM_ 100 $ enqueue wq $ threadDelay 1000+ waitReadyQueue wq+ threadDelay 1000+ computations `shouldSatisfy` isJust+ replicateM_ 1 $+ it "should be resilient to errors and Exceptions" $ do+ witness <- newIORef False+ computations <-+ timeout 10000 $+ withPoolboy (poolboySettingsWith 5) $ \wq -> do+ mapM_ (enqueue wq) [error "an error", throw RandomException, writeIORef witness True]+ waitReadyQueue wq+ threadDelay 100+ computations `shouldSatisfy` isJust+ readIORef witness `shouldReturn` True++data RandomException = RandomException+ deriving (Show)++instance Exception RandomException