Spock-worker 0.1.0.5 → 0.2.0.0
raw patch · 2 files changed
+71/−30 lines, 2 filesdep ~basedep ~vector
Dependency ranges changed: base, vector
Files
- Spock-worker.cabal +7/−7
- src/Web/Spock/Worker.hs +64/−23
Spock-worker.cabal view
@@ -1,5 +1,5 @@ name: Spock-worker-version: 0.1.0.5+version: 0.2.0.0 synopsis: Background workers for Spock description: Adds a background-job queue to Spock homepage: http://github.com/agrafix/Spock-worker@@ -21,9 +21,9 @@ Spock >=0.4.2.4 && <0.7, HTF ==0.11.2.*, text >=0.11.3.1 && <1.2,- time ==1.4.*,- stm ==2.4.*,- vector ==0.10.9.*,+ time >=1.4 && <1.5,+ stm >=2.4 && <2.5,+ vector >=0.10 && <0.11, containers ==0.5.*, mtl, transformers,@@ -38,9 +38,9 @@ Spock >=0.4.2.4 && <0.7, HTF ==0.11.2.*, text >=0.11.3.1 && <1.2,- time ==1.4.*,- stm ==2.4.*,- vector ==0.10.9.*,+ time >=1.4 && <1.5,+ stm >=2.4 && <2.5,+ vector >=0.10 && <0.11, containers ==0.5.*, mtl, transformers,
src/Web/Spock/Worker.hs view
@@ -4,12 +4,14 @@ ( -- * Worker WorkQueue , WorkHandler+ , WorkerConfig (..)+ , WorkerConcurrentStrategy (..) , newWorker , addWork , WorkExecution (..) , WorkResult (..) -- * Error Handeling- , ErrorHandler, InternalError+ , ErrorHandler(..), InternalError -- * Tests , htf_thisModulesTests )@@ -20,6 +22,7 @@ import Control.Concurrent import Control.Concurrent.STM +import Control.Monad import Control.Monad.Trans import Control.Monad.Trans.Error import Control.Exception.Lifted as EX@@ -33,8 +36,9 @@ -- | Describe how you want to handle errors. Make sure you catch all exceptions -- that can happen inside this handler, otherwise the worker will crash!-type ErrorHandler a- = InternalError -> a -> IO WorkResult+data ErrorHandler conn sess st a+ = ErrorHandlerIO (InternalError -> a -> IO WorkResult)+ | ErrorHandlerSpock (InternalError -> a -> (WebStateM conn sess st) WorkResult) -- | Describe how you want jobs in the queue to be performed type WorkHandler conn sess st a@@ -58,49 +62,86 @@ | WorkRepeatAt UTCTime deriving (Show, Eq) +-- | Configure the concurrent behaviour of a worker. If you want tasks executed+-- concurrently, consider using 'WorkerConcurrentBounded'+data WorkerConcurrentStrategy+ = WorkerNoConcurrency+ | WorkerConcurrentBounded Int+ | WorkerConcurrentUnbounded++-- | Configure how the worker handles it's task and define the queue size+data WorkerConfig+ = WorkerConfig+ { wc_queueLimit :: Int+ , wc_concurrent :: WorkerConcurrentStrategy+ }+ -- | Create a new background worker and limit the size of the job queue.-newWorker :: Int+newWorker :: WorkerConfig -> WorkHandler conn sess st a- -> ErrorHandler a+ -> ErrorHandler conn sess st a -> SpockM conn sess st (WorkQueue a)-newWorker maxSize workHandler errorHandler =+newWorker wc workHandler errorHandler = do heart <- getSpockHeart- q <- liftIO $ Q.newQueue maxSize- _ <- liftIO $ forkIO (workProcessor q workHandler errorHandler heart)+ q <- liftIO $ Q.newQueue (wc_queueLimit wc)+ _ <- liftIO $ forkIO (workProcessor q workHandler errorHandler heart (wc_concurrent wc)) return (WorkQueue q) workProcessor :: Q.WorkerQueue UTCTime a -> WorkHandler conn sess st a- -> ErrorHandler a+ -> ErrorHandler conn sess st a -> WebState conn sess st+ -> WorkerConcurrentStrategy -> IO ()-workProcessor q workHandler errorHandler spockCore =- loop+workProcessor q workHandler errorHandler spockCore concurrentStrategy =+ do runningTasksVar <- newTVarIO 0+ loop runningTasksVar where runWork work = do workRes <- EX.catch (runSpockIO spockCore $ runErrorT $ workHandler work) (\(e::SomeException) -> return $ Left (show e)) case workRes of- Left err -> errorHandler err work+ Left err ->+ case errorHandler of+ ErrorHandlerIO h ->+ h err work+ ErrorHandlerSpock h ->+ runSpockIO spockCore $ h err work Right r -> return r- loop =++ loop runningTasksV = do now <- getCurrentTime mWork <- atomically $ Q.dequeue now q case mWork of Nothing -> do threadDelay (1000 * 1000) -- 1 sec- loop+ loop runningTasksV Just work ->- do res <- runWork work- case res of- WorkRepeatIn secs ->- addWork (WorkIn secs) work (WorkQueue q)- WorkRepeatAt time ->- addWork (WorkAt time) work (WorkQueue q)- _ ->- return ()- loop+ do case concurrentStrategy of+ WorkerConcurrentBounded limit ->+ do atomically $+ do runningTasks <- readTVar runningTasksV+ when (runningTasks >= limit) retry+ _ <- forkIO $ launchWork runningTasksV work+ return ()+ WorkerNoConcurrency ->+ launchWork runningTasksV work+ WorkerConcurrentUnbounded ->+ do _ <- forkIO $ launchWork runningTasksV work+ return ()+ loop runningTasksV++ launchWork runningTasksV work =+ do atomically $ modifyTVar runningTasksV (\x -> x + 1)+ res <- (runWork work `EX.finally` (atomically $ modifyTVar runningTasksV (\x -> x - 1)))+ case res of+ WorkRepeatIn secs ->+ addWork (WorkIn secs) work (WorkQueue q)+ WorkRepeatAt time ->+ addWork (WorkAt time) work (WorkQueue q)+ _ ->+ return () -- | Add a new job to the background worker. If the queue is full this will block addWork :: MonadIO m => WorkExecution -> a -> WorkQueue a -> m ()