Spock-worker 0.1.0.1 → 0.1.0.2
raw patch · 4 files changed
+226/−57 lines, 4 filesdep +HTFdep +lifted-basedep ~Spock
Dependencies added: HTF, lifted-base
Dependency ranges changed: Spock
Files
- Spock-worker.cabal +22/−6
- src/Web/Spock/Worker.hs +41/−17
- src/Web/Spock/Worker/Queue.hs +154/−34
- src/Web/Spock/Worker/Tests.hs +9/−0
Spock-worker.cabal view
@@ -1,8 +1,5 @@--- Initial Spock-worker.cabal generated by cabal init. For further --- documentation, see http://haskell.org/cabal/users-guide/- name: Spock-worker-version: 0.1.0.1+version: 0.1.0.2 synopsis: Background workers for Spock description: Adds a background-job queue to Spock homepage: http://github.com/agrafix/Spock-worker@@ -18,14 +15,33 @@ library exposed-modules: Web.Spock.Worker other-modules: Web.Spock.Worker.Queue+ hs-source-dirs: src+ default-language: Haskell2010 build-depends: base >=4.6 && <4.7,- Spock >=0.4.2.1 && <0.4.3,+ Spock >=0.4.2.4 && <0.4.3,+ HTF ==0.11.2.*, text >=0.11.3.1 && <1.2, time ==1.4.*, stm ==2.4.*, vector ==0.10.9.*, containers ==0.5.*, mtl,- transformers+ transformers,+ lifted-base++test-suite SpockWorkerTests+ type: exitcode-stdio-1.0+ main-is: Web/Spock/Worker/Tests.hs hs-source-dirs: src default-language: Haskell2010+ build-depends: base >=4.6 && <4.7,+ Spock >=0.4.2.4 && <0.4.3,+ HTF ==0.11.2.*,+ text >=0.11.3.1 && <1.2,+ time ==1.4.*,+ stm ==2.4.*,+ vector ==0.10.9.*,+ containers ==0.5.*,+ mtl,+ transformers,+ lifted-base
src/Web/Spock/Worker.hs view
@@ -1,4 +1,5 @@-{-# LANGUAGE RankNTypes #-}+{-# OPTIONS_GHC -F -pgmF htfpp #-}+{-# LANGUAGE RankNTypes, ScopedTypeVariables, OverloadedStrings #-} module Web.Spock.Worker ( -- * Worker WorkQueue@@ -9,14 +10,20 @@ , WorkResult (..) -- * Error Handeling , ErrorHandler, InternalError+ -- * Tests+ , htf_thisModulesTests ) where +import Test.Framework+ import Control.Concurrent import Control.Concurrent.STM import Control.Monad.Trans import Control.Monad.Trans.Error+import Control.Exception.Lifted as EX+import Control.Exception import Data.Time import Web.Spock@@ -59,24 +66,41 @@ newWorker maxSize workHandler errorHandler = do heart <- getSpockHeart q <- liftIO $ Q.newQueue maxSize- _ <- liftIO $ forkIO (runSpockIO heart $ core q)+ _ <- liftIO $ forkIO (workProcessor q workHandler errorHandler heart) return (WorkQueue q)++workProcessor :: Q.WorkerQueue UTCTime a+ -> WorkHandler conn sess st a+ -> ErrorHandler a+ -> WebState conn sess st+ -> IO ()+workProcessor q workHandler errorHandler spockCore =+ loop where- core q =- do work <- liftIO $ atomically $ Q.dequeue q- res <-- do workRes <- runErrorT $ workHandler work- case workRes of- Left err -> liftIO (errorHandler err work)- Right r -> return r- now <- liftIO $ getCurrentTime- case res of- WorkRepeatIn secs ->- addWork (WorkIn secs) work (WorkQueue q)- WorkRepeatAt time ->- addWork (WorkAt time) work (WorkQueue q)- _ ->- return ()+ 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+ Right r -> return r+ loop =+ do now <- getCurrentTime+ mWork <- atomically $ Q.dequeue now q+ case mWork of+ Nothing ->+ do threadDelay (1000 * 1000) -- 1 sec+ loop+ 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 -- | Add a new job to the background worker. If the queue is full this will block addWork :: MonadIO m => WorkExecution -> a -> WorkQueue a -> m ()
src/Web/Spock/Worker/Queue.hs view
@@ -1,53 +1,173 @@+{-# OPTIONS_GHC -F -pgmF htfpp #-}+{-# LANGUAGE ScopedTypeVariables #-} module Web.Spock.Worker.Queue- ( WorkerQueue, newQueue, size, enqueue, dequeue, isFull )+ ( WorkerQueue, newQueue, size, enqueue, dequeue, isFull+ , htf_thisModulesTests+ ) where +import Test.Framework++import Control.Monad+import Control.Concurrent import Control.Concurrent.STM import Control.Applicative+import Data.List (sortBy)+import Data.Ord (comparing)+import Data.Maybe+ import qualified Data.Map.Strict as M import qualified Data.Vector as V -data WorkerQueue p v- = WorkerQueue- { wq_container :: TVar (M.Map p (V.Vector v))- , wq_maxSize :: Int+data PureQueue p v+ = PureQueue+ { pq_container :: M.Map p (V.Vector v)+ , pq_maxSize :: Int } +emptyPQ :: Int -> PureQueue p v+emptyPQ maxSize =+ PureQueue M.empty maxSize++sizePQ :: PureQueue p v -> Int+sizePQ (PureQueue m _) =+ M.size m++isFullPQ :: PureQueue p v -> Bool+isFullPQ pq@(PureQueue _ maxSize) =+ sizePQ pq >= maxSize++toListPQ :: Ord p => PureQueue p v -> [(p, [v])]+toListPQ pq@(PureQueue m _) =+ map (\(k, v) -> (k, V.toList v)) (M.toList m)++fromListPQ :: Ord p => Int -> [(p, [v])] -> Maybe (PureQueue p v)+fromListPQ limit kv+ | length kv > limit = Nothing+ | otherwise =+ Just $+ foldl (\(PureQueue content _) (k, v) ->+ (PureQueue (M.insert k (V.fromList v) content) limit)+ ) (emptyPQ limit) kv++maxPrioPQ :: Ord p => PureQueue p v -> p+maxPrioPQ (PureQueue m _) =+ fst (M.findMax m)++enqueuePQ :: Ord p => p -> v -> PureQueue p v -> (Bool, PureQueue p v)+enqueuePQ prio value pq@(PureQueue m _) =+ if isFullPQ pq+ then ( False, pq )+ else ( True+ , pq+ { pq_container =+ M.insertWith (V.++) prio (V.singleton value) m+ }+ )++dequeuePQ :: Ord p => p -> PureQueue p v -> (Maybe v, PureQueue p v)+dequeuePQ bound pq =+ removePrio $ dequeuePQ' bound pq+ where+ removePrio (Nothing, q) = (Nothing, q)+ removePrio (Just (p, v), q) = (Just v, q)++dequeuePQ' :: Ord p => p -> PureQueue p v -> (Maybe (p, v), PureQueue p v)+dequeuePQ' prioBound pq@(PureQueue m _)+ | M.null m = (Nothing, pq)+ | minK > prioBound = (Nothing, pq)+ | otherwise =+ case V.toList vec of+ [workEl] ->+ (Just (minK, workEl), updatePQ (M.delete minK))+ (workEl:xs) ->+ (Just (minK, workEl), updatePQ (M.adjust (const (V.fromList xs)) minK))+ [] ->+ error "Library-Error: This should never happen."+ where+ (minK, vec) =+ M.findMin m+ updatePQ fun =+ pq { pq_container = fun (pq_container pq) }++newtype WorkerQueue p v =+ WorkerQueue (TVar (PureQueue p v))+ newQueue :: Int -> IO (WorkerQueue p v) newQueue limit =- (flip WorkerQueue) limit <$> newTVarIO M.empty+ WorkerQueue <$> newTVarIO (emptyPQ limit) size :: WorkerQueue p v -> STM Int-size (WorkerQueue q _) =- M.size <$> readTVar q+size (WorkerQueue qVar) =+ readTVar qVar >>= (return . sizePQ) isFull :: WorkerQueue p v -> STM Bool-isFull wq@(WorkerQueue _ sizeLimit) =- do currSize <- size wq- return (currSize >= sizeLimit)+isFull (WorkerQueue qVar) =+ readTVar qVar >>= (return . isFullPQ) enqueue :: Ord p => p -> v -> WorkerQueue p v -> STM ()-enqueue priority value wq@(WorkerQueue q _) =- do full <- isFull wq- if full- then retry- else modifyTVar' q (M.insertWith (V.++) priority (V.singleton value))+enqueue prio value (WorkerQueue qVar) =+ do q <- readTVar qVar+ let (ok, newQ) = enqueuePQ prio value q+ if ok+ then writeTVar qVar newQ+ else retry -dequeue :: Ord p => WorkerQueue p v -> STM v-dequeue (WorkerQueue q _) =- do m <- readTVar q- if M.null m- then retry- else runDequeue m- where- runDequeue m =- do let (minPrio, vals) = M.findMin m- case V.toList vals of- [workEl] ->- do writeTVar q (M.delete minPrio m)- return workEl- (workEl:xs) ->- do writeTVar q (M.adjust (const (V.fromList xs)) minPrio m)- return workEl- [] ->- error "Library-Error: This should never happen."+dequeue :: Ord p => p -> WorkerQueue p v -> STM (Maybe v)+dequeue minP (WorkerQueue qVar) =+ do q <- readTVar qVar+ let (mVal, newQ) = dequeuePQ minP q+ when (isJust mVal) $ writeTVar qVar newQ+ return mVal++-- -------------+-- TESTS+-- -------------++tAddToMap k val m =+ M.insertWith (++) k [val] m+tDeq maxP q+ | sizePQ q == 0 =+ M.empty+ | otherwise =+ let (mVal, newQ) = dequeuePQ' maxP q+ in case mVal of+ Nothing ->+ M.empty+ Just (k, val) ->+ tAddToMap k val (tDeq maxP newQ)+tMappifyInput xs =+ foldl (\m (k, v) ->+ tAddToMap k v m+ ) M.empty xs++prop_enqueueDequeuePQ xs =+ let pq = foldl (\q (prio :: Int, el :: Int) ->+ let (ok, newPQ) = enqueuePQ prio el q+ in if ok then newPQ else (error "Failed to enqueue!")+ ) (emptyPQ (length xs)) xs+ maxP = maxPrioPQ pq+ in (tMappifyInput xs == tDeq maxP pq)++prop_onlyDequeueBelowPrio prio xs =+ let xs' = M.toList $ tMappifyInput xs+ Just pq = fromListPQ (length xs) (xs' :: [(Int, [Int])])+ filtered = filter (\(p, _) -> p <= prio) xs+ in (tMappifyInput filtered == tDeq prio pq)++prop_isFull limit xs =+ limit > 0 ==>+ let xs' = M.toList $ tMappifyInput xs+ mPq = fromListPQ limit (xs' :: [(Int, [Int])])+ in case mPq of+ Just pq ->+ if limit == (length xs')+ then isFullPQ pq+ else limit > (length xs')+ Nothing ->+ limit < (length xs')++test_dontEnqueueIfFull =+ let pq = emptyPQ 0+ (ok, newPQ) = enqueuePQ (0 :: Int) False pq+ in assertBool (not ok)
+ src/Web/Spock/Worker/Tests.hs view
@@ -0,0 +1,9 @@+{-# OPTIONS_GHC -F -pgmF htfpp #-}++module Main where++import Test.Framework+import {-@ HTF_TESTS @-} Web.Spock.Worker.Queue+import {-@ HTF_TESTS @-} Web.Spock.Worker++main = htfMain htf_importedTests