effectful-poolboy (empty) → 0.1.0.0
raw patch · 6 files changed
+488/−0 lines, 6 filesdep +asyncdep +basedep +effectful
Dependencies added: async, base, effectful, effectful-core, effectful-poolboy, hspec, hspec-core, poolboy, timeit
Files
- CHANGELOG.md +3/−0
- LICENSE +15/−0
- effectful-poolboy.cabal +96/−0
- src/Data/Poolboy/Effectful.hs +178/−0
- src/Data/Poolboy/Tactics/Effectful.hs +136/−0
- test/Spec.hs +60/−0
+ CHANGELOG.md view
@@ -0,0 +1,3 @@+## 0.1.0.0++* Initial version
+ LICENSE view
@@ -0,0 +1,15 @@+ISC License++Copyright (c) 2025 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.
+ effectful-poolboy.cabal view
@@ -0,0 +1,96 @@+cabal-version: 3.0+name: effectful-poolboy+version: 0.1.0.0+author: Gautier DI FOLCO+maintainer: foss@difolco.dev+category: Data+build-type: Simple+license: ISC+license-file: LICENSE+synopsis: Simple work queue for bounded concurrency (effectful wrapper)+description: In-memory work queue helping with load management (effectful wrapper).+Homepage: https://github.com/blackheaven/poolboy+tested-with: GHC==9.10.4++extra-source-files:+ CHANGELOG.md++library+ default-language: Haskell2010+ build-depends:+ base >= 4.2 && <5+ , async == 2.*+ , effectful-core >= 2.3.0.0 && < 2.7+ , poolboy >= 0.5.0.0 && <0.6+ hs-source-dirs: src+ exposed-modules:+ Data.Poolboy.Effectful+ Data.Poolboy.Tactics.Effectful+ other-modules:+ Paths_effectful_poolboy+ autogen-modules:+ Paths_effectful_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 spec+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ other-modules:+ Paths_effectful_poolboy+ autogen-modules:+ Paths_effectful_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+ , async+ , effectful+ , effectful-core+ , effectful-poolboy+ , hspec+ , hspec-core+ , poolboy+ , timeit >= 1.0.0.0 && < 3+ default-language: Haskell2010
+ src/Data/Poolboy/Effectful.hs view
@@ -0,0 +1,178 @@+--- |+-- Module : Data.Poolboy.Effectful+-- Copyright : Gautier DI FOLCO 2025+-- License : ISC+--+-- Maintainer : foos@difolco.dev+-- Stability : experimental+-- Portability : GHC+--+-- Thin Effectful wrapper around the existing poolboy library (Data.Poolboy).+--+-- @+-- withPoolboy defaultPoolboySettings waitingStopFinishWorkers $ \workQueue ->+-- mapM_ (enqueue workQueue . execQuery insertBookQuery) books+-- @+--++module Data.Poolboy.Effectful+ ( -- * Configuration+ PB.PoolboySettings (..),+ PB.WorkersCountSettings (..),+ PB.PoolboyCommand (..),+ PB.defaultPoolboySettings,+ PB.poolboySettingsWith,+ PB.poolboySettingsName,+ PB.poolboySettingsLog,++ -- * Running+ WorkQueue,+ withPoolboy,+ newPoolboy,+ PB.hoistWorkQueue,++ -- * Driving+ changeDesiredWorkersCount,+ waitReadyQueue,++ -- * Stopping+ stopWorkQueue,+ isStoppedWorkQueue,+ PB.WaitingStopStrategy,+ waitingStopTimeout,+ waitingStopFinishWorkers,++ -- * Enqueueing+ enqueue,+ enqueueTracking,+ enqueueAfter,+ enqueueAfterTracking,+ PB.WorkQueueStoppedException,+ )+where++import Control.Concurrent.Async (Async)+import qualified Data.Poolboy as PB+import Effectful++-- | Local wrapper type: hides the underlying PB.WorkQueue IO inside Eff.+newtype WorkQueue (es :: [Effect]) = WorkQueue (PB.WorkQueue IO)++--------------------------------------------------------------------------------+-- Bracket / creation+--------------------------------------------------------------------------------++-- | Bracket-like helper that runs poolboy's `withPoolboy` in IO while presenting+-- a `WorkQueue es` to the provided Eff callbacks.+withPoolboy ::+ forall es a.+ (IOE :> es) =>+ PB.PoolboySettings (Eff es) ->+ PB.WaitingStopStrategy (Eff es) ->+ (WorkQueue es -> Eff es a) ->+ Eff es a+withPoolboy settings waitStrat inner = do+ -- acquire a conversion Eff -> IO so we can run library code in IO threads+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ -- convert PoolboySettings (Eff es) -> PoolboySettings IO+ let settingsIO :: PB.PoolboySettings IO+ settingsIO = PB.hoistPoolboySettings toIO settings++ -- convert WaitingStopStrategy: PB.WorkQueue IO -> IO ()+ let waitStratIO :: PB.WorkQueue IO -> IO ()+ waitStratIO wqIO = toIO (waitStrat $ PB.hoistWorkQueue liftIO wqIO)++ -- run poolboy's withPoolboy in IO and convert the provided inner callback+ PB.withPoolboy settingsIO waitStratIO $ \wqIO -> do+ let wqEff = WorkQueue wqIO+ toIO (inner wqEff)++-- | Standalone/manual usage+newPoolboy ::+ forall es.+ (IOE :> es) =>+ PB.PoolboySettings (Eff es) ->+ Eff es (WorkQueue es)+newPoolboy settings = do+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ let settingsIO = PB.hoistPoolboySettings toIO settings+ wqIO <- liftIO $ PB.newPoolboy settingsIO+ pure (WorkQueue wqIO)++-- | Request a worker number adjustment+--+-- Warning: non-concurrent operation+changeDesiredWorkersCount :: (IOE :> es) => WorkQueue es -> Int -> Eff es ()+changeDesiredWorkersCount (WorkQueue wq) n = liftIO $ PB.changeDesiredWorkersCount wq n++-- | Request stopping wokers+stopWorkQueue :: (IOE :> es) => WorkQueue es -> Eff es ()+stopWorkQueue (WorkQueue wq) =+ -- underlying API expects m ~ IO, so just call in IO+ liftIO $ PB.stopWorkQueue wq++-- | Non-blocking check of the work queue's running status+isStoppedWorkQueue :: (IOE :> es) => WorkQueue es -> Eff es Bool+isStoppedWorkQueue (WorkQueue wq) = liftIO $ PB.isStoppedWorkQueue wq++-- | Block until the queue is totally stopped (no more running worker)+waitingStopFinishWorkers :: (IOE :> es) => PB.WaitingStopStrategy (Eff es)+waitingStopFinishWorkers wq =+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ PB.waitingStopFinishWorkers $ PB.hoistWorkQueue toIO wq++-- | Block until the queue is totally stopped or deadline (in micro seconds) is reached+waitingStopTimeout :: (IOE :> es) => Int -> PB.WaitingStopStrategy (Eff es)+waitingStopTimeout delay wq =+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ PB.waitingStopTimeout delay $ PB.hoistWorkQueue toIO wq++-- | Enqueue one action in the work queue (non-blocking)+--+-- Throws 'WorkQueueStoppedException' if the work queue is stopped+enqueue :: forall es. (IOE :> es) => WorkQueue es -> Eff es () -> Eff es ()+enqueue (WorkQueue wq) eff = do+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ -- PB.enqueue :: WorkQueue m -> m () -> m (); here m ~ IO+ liftIO $ PB.enqueue wq (toIO eff)++-- | Enqueue one action in the work queue (non-blocking)+--+-- Throws 'WorkQueueStoppedException' if the work queue is stopped+enqueueTracking :: forall es a. (IOE :> es) => WorkQueue es -> Eff es a -> Eff es (Async a)+enqueueTracking (WorkQueue wq) eff = do+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ -- PB.enqueueTracking :: WorkQueue m -> m a -> m (Async a)+ liftIO $ PB.enqueueTracking wq (toIO eff)++-- | Block until one worker is available+waitReadyQueue :: (IOE :> es) => WorkQueue es -> Eff es ()+waitReadyQueue (WorkQueue wq) = liftIO $ PB.waitReadyQueue wq++-- | Enqueue action and some actions to be run after it+--+-- Throws 'WorkQueueStoppedException' if the work queue is stopped+enqueueAfter ::+ forall t es.+ (Traversable t, IOE :> es) =>+ WorkQueue es ->+ Eff es () ->+ t (Eff es ()) ->+ Eff es ()+enqueueAfter (WorkQueue wq) before afters = do+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ liftIO $ PB.enqueueAfter wq (toIO before) (fmap toIO afters)++-- | Enqueue action and some actions to be run after it+--+-- Throws 'WorkQueueStoppedException' if the work queue is stopped+enqueueAfterTracking ::+ forall t a b es.+ (Traversable t, IOE :> es) =>+ WorkQueue es ->+ Eff es a ->+ t (Eff es b) ->+ Eff es (Async a, t (Async b))+enqueueAfterTracking (WorkQueue wq) before afters = do+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall c. Eff es c -> IO c) -> do+ liftIO $ PB.enqueueAfterTracking wq (toIO before) (fmap toIO afters)
+ src/Data/Poolboy/Tactics/Effectful.hs view
@@ -0,0 +1,136 @@+-- |+-- Module : Data.Poolboy.Tactics.Effectful+-- Copyright : Gautier DI FOLCO 2024-2025+-- License : ISC+--+-- Maintainer : foos@difolco.dev+-- Stability : experimental+-- Portability : GHC+--+-- A simple set of concurrent primitives.+module Data.Poolboy.Tactics.Effectful+ ( -- * Do not accumulate+ concurrentFoldable_,+ concurrentRecursive_,+ concurrentM_,++ -- * Accumulate+ concurrentFoldable,+ concurrentRecursive,+ concurrentRecursive',+ concurrentM,+ )+where++import qualified Data.Poolboy as PB+import qualified Data.Poolboy.Tactics as PB+import Effectful++-- | Concurrently run a set of actions.+--+-- Warning: blocking+--+-- > concurrentFoldable_ defaultPoolboySettings [sendEmail, saveOrderDB, notifyUser]+concurrentFoldable_ ::+ (Foldable f, Functor f, IOE :> es) =>+ PB.PoolboySettings (Eff es) ->+ f (Eff es a) ->+ Eff es ()+concurrentFoldable_ settings actions =+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ PB.concurrentFoldable_ (PB.hoistPoolboySettings toIO settings) (toIO <$> actions)++-- | Concurrently run a set of actions, recursively.+--+-- Warning: blocking+--+-- > concurrentRecursive_ defaultPoolboySettings listDirectories [listDirectories "."]+concurrentRecursive_ ::+ (Foldable f, Functor f, IOE :> es) =>+ PB.PoolboySettings (Eff es) ->+ (a -> f (Eff es a)) ->+ f (Eff es a) ->+ Eff es ()+concurrentRecursive_ settings recurse actions =+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ PB.concurrentRecursive_ (PB.hoistPoolboySettings toIO settings) ((toIO <$>) . recurse) (toIO <$> actions)++-- | Concurrently run a dynamic set of actions until it gets a 'Nothing'.+--+-- Warning: blocking+--+-- > concurrentM_ defaultPoolboySettings waitNextRequest+concurrentM_ ::+ (IOE :> es) =>+ PB.PoolboySettings (Eff es) ->+ Eff es (Maybe (Eff es a)) ->+ Eff es ()+concurrentM_ settings fetchNextAction =+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ PB.concurrentM_ (PB.hoistPoolboySettings toIO settings) (toIO $ fmap toIO <$> fetchNextAction)++-- | Concurrently run a set of actions, accumulating results.+--+-- Warning: blocking+--+-- Warning: results are collected in no particular order+--+-- > concurrentFoldable defaultPoolboySettings [sendEmail, saveOrderDB, notifyUser]+concurrentFoldable ::+ (Functor f, Foldable f, IOE :> es) =>+ PB.PoolboySettings (Eff es) ->+ f (Eff es a) ->+ Eff es [a]+concurrentFoldable settings actions = do+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ PB.concurrentFoldable (PB.hoistPoolboySettings toIO settings) (toIO <$> actions)++-- | Concurrently run a set of actions, recursively, accumulating results.+--+-- Warning: blocking+--+-- Warning: results are collected in no particular order+--+-- > concurrentRecursive defaultPoolboySettings listDirectories [listDirectories "."]+concurrentRecursive ::+ (Functor f, Foldable f, IOE :> es) =>+ PB.PoolboySettings (Eff es) ->+ (a -> f (Eff es a)) ->+ f (Eff es a) ->+ Eff es [a]+concurrentRecursive settings recurse actions =+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ PB.concurrentRecursive (PB.hoistPoolboySettings toIO settings) ((toIO <$>) . recurse) (toIO <$> actions)++-- | Concurrently run a set of actions, recursively, accumulating results.+--+-- Warning: blocking+--+-- Warning: results are collected in no particular order+--+-- > concurrentRecursive' defaultPoolboySettings listDirectories [listDirectories "."]+concurrentRecursive' ::+ (Functor f, Foldable f, IOE :> es) =>+ PB.PoolboySettings (Eff es) ->+ (a -> f (Eff es (a, b))) ->+ f (Eff es (a, b)) ->+ Eff es [b]+concurrentRecursive' settings recurse actions = do+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ PB.concurrentRecursive' (PB.hoistPoolboySettings toIO settings) ((toIO <$>) . recurse) (toIO <$> actions)++-- | Concurrently run a dynamic set of actions until it gets a 'Nothing', accumulating results.+--+-- Warning: blocking+--+-- Warning: results are collected in no particular order+--+-- > concurrentM defaultPoolboySettings waitNextRequest+concurrentM ::+ (IOE :> es) =>+ PB.PoolboySettings (Eff es) ->+ Eff es (Maybe (Eff es a)) ->+ Eff es [a]+concurrentM settings fetchNextAction = do+ withEffToIO (ConcUnlift Ephemeral Unlimited) $ \(toIO :: forall b. Eff es b -> IO b) -> do+ PB.concurrentM (PB.hoistPoolboySettings toIO settings) (toIO $ fmap toIO <$> fetchNextAction)
+ test/Spec.hs view
@@ -0,0 +1,60 @@+{-# LANGUAGE NumericUnderscores #-}++module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (wait)+import Data.IORef+import qualified Data.Poolboy as PB+import Data.Poolboy.Effectful+import Effectful+import Test.Hspec++--------------------------------------------------------------------------------+-- Main+--------------------------------------------------------------------------------++main :: IO ()+main = hspec spec++spec :: Spec+spec = do+ describe "Effectful Poolboy Wrapper" $ do+ it "enqueueEff runs an Eff action" $ do+ res <-+ runEff $+ withPoolboy+ PB.defaultPoolboySettings+ PB.waitingStopFinishWorkers+ ( \wq -> do+ r <- newIORefEff (0 :: Int)+ enqueue wq (writeIORefEff r 42)+ liftIO $ threadDelay 100_000+ readIORefEff r+ )+ res `shouldBe` 42++ it "enqueueTrackingEff returns Async with correct result" $ do+ res <-+ runEff $+ withPoolboy+ PB.defaultPoolboySettings+ PB.waitingStopFinishWorkers+ ( \wq -> do+ a <- enqueueTracking wq (pure @(Eff _) (99 :: Int))+ liftIO (wait a)+ )+ res `shouldBe` 99++--------------------------------------------------------------------------------+-- Helpers+--------------------------------------------------------------------------------++newIORefEff :: (IOE :> es) => a -> Eff es (IORef a)+newIORefEff = liftIO . newIORef++writeIORefEff :: (IOE :> es) => IORef a -> a -> Eff es ()+writeIORefEff r x = liftIO $ writeIORef r x++readIORefEff :: (IOE :> es) => IORef a -> Eff es a+readIORefEff r = liftIO $ readIORef r