poolboy 0.2.0.0 → 0.2.1.0

raw patch · 3 files changed

+82/−29 lines, 3 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• poolboy.cabal +1/−1
• src/Data/Poolboy.hs +60/−22
• test/Spec.hs +21/−6

@@ -1,6 +1,6 @@ cabal-version:       3.0 name:                poolboy-version:             0.2.0.0+version:             0.2.1.0 author:              Gautier DI FOLCO maintainer:          gautier.difolco@gmail.com category:            Data

@@ -83,7 +83,7 @@       <$> newTQueueIO       <*> newTQueueIO       <*> newEmptyMVar-      <*> return settings.log+      <*> newQSemN 0    count <-     case settings.workersCount of@@ -91,7 +91,18 @@       FixedWCS x -> return x    changeDesiredWorkersCount wq count-  void $ forkIO $ controller wq []+  void $+    forkIO $+      controller $+        WorkQueueControllerState+          {commands = wq.commands,+            queue = wq.queue,+            stopped = wq.stopped,+            log = settings.log,+            workers = [],+            waitingWorkers = wq.waitingWorkers,+            capabilityCount = 0+          }    return wq @@ -129,9 +140,8 @@ -- | Block until one worker is available waitReadyQueue :: WorkQueue -> IO () waitReadyQueue wq = do-  ready <- newEmptyMVar-  enqueue wq $ putMVar ready ()-  readMVar ready+  waitQSemN wq.waitingWorkers 1+  signalQSemN wq.waitingWorkers 1  -- | Enqueue action and some actions to be run after it enqueueAfter :: Foldable f => WorkQueue -> IO () -> f (IO ()) -> IO ()@@ -145,7 +155,7 @@   { commands :: TQueue Commands,     queue :: TQueue (Either () (IO ())),     stopped :: MVar (),-    log :: String -> IO ()+    waitingWorkers :: QSemN   }  data Commands@@ -153,40 +163,68 @@   | Stop   deriving stock (Show) -controller :: WorkQueue -> [Async ()]  -> IO ()-controller wq workers = do+data WorkQueueControllerState = WorkQueueControllerState+  { commands :: TQueue Commands,+    queue :: TQueue (Either () (IO ())),+    stopped :: MVar (),+    log :: String -> IO (),+    workers :: [Async ()],+    waitingWorkers :: QSemN,+    capabilityCount :: Int+  }++controller :: WorkQueueControllerState -> IO ()+controller wq = do   command <- atomically $ readTQueue wq.commands   let stopOneWorker = atomically $ writeTQueue wq.queue $ Left ()-      getLiveWorkers = filterM (fmap isNothing . poll) workers-  wq.log $ "Command: " <> show command+      getLiveWorkers = filterM (fmap isNothing . poll) wq.workers+      prefix = "Controller: "+  wq.log $ prefix <> "Command: " <> show command   case command of     ChangeDesiredWorkersCount n -> do       liveWorkers <- getLiveWorkers       let diff = length liveWorkers - n-      newWorkers <-+      (newWorkers, newCapabilityCount) <-         if diff > 0           then do             replicateM_ diff stopOneWorker-            return []-          else replicateM (abs diff) $ do-            wq.log "Pre-fork"-            async $ worker wq-      controller wq $ newWorkers <> liveWorkers+            return ([], wq.capabilityCount)+          else do+            let newWorkersCount = abs diff+            newWorkers <- forM [1..newWorkersCount] $ \capability -> do+              wq.log $ prefix <> "Pre-fork"+              asyncOn (capability - 1) $ worker $ WorkQueueWorkerState{queue = wq.queue, waitingWorkers = wq.waitingWorkers, log = wq.log}+            return (newWorkers, wq.capabilityCount + newWorkersCount)+      controller $ wq { workers = newWorkers <> liveWorkers, capabilityCount = newCapabilityCount }     Stop -> do       liveWorkers <- getLiveWorkers       let currentCount = length liveWorkers-      wq.log $ "Stopping " <> show currentCount <> " workers"+      wq.log $ prefix <> "Stopping " <> show currentCount <> " workers"+      waitQSemN wq.waitingWorkers currentCount       replicateM_ currentCount stopOneWorker-      forM_ liveWorkers waitCatch       void $ tryPutMVar wq.stopped () -worker :: WorkQueue -> IO ()+data WorkQueueWorkerState = WorkQueueWorkerState+  { queue :: TQueue (Either () (IO ())),+    waitingWorkers :: QSemN,+    log :: String -> IO ()+  }++worker :: WorkQueueWorkerState -> IO () worker wq = do-  wq.log "New worker"+  workerId <- show <$> myThreadId+  let prefix = "Worker [" <> workerId <> "]: "+  wq.log $ prefix <> "Starting"   let loop = do+        signalQSemN wq.waitingWorkers 1         command <- atomically $ readTQueue wq.queue         case command of           Left () -> do-            wq.log "Stopping"-          Right act -> wq.log "pop" >> void (tryAny act) >> wq.log "poped" >> loop+            wq.log $ prefix <> "Stopping"+          Right act -> do+            waitQSemN wq.waitingWorkers 1+            wq.log (prefix <> "pop")+            void (tryAny act)+            wq.log (prefix <> "poped")+            loop   loop

@@ -15,13 +15,14 @@ spec :: Spec spec =   describe "Poolboy" $ do-    it "threadDelay should be absorbed in mulitple threads" $ do-      computations <--        timeout 2500 $-          withPoolboy (poolboySettingsWith 100) waitingStopFinishWorkers $ \wq -> do-            replicateM_ 100 $ enqueue wq $ threadDelay 1000-      computations `shouldSatisfy` isJust     replicateM_ 1 $+      it "threadDelay should be absorbed in mulitple threads" $ do+        computations <-+          timeout 2500 $+            withPoolboy (poolboySettingsWith 100) waitingStopFinishWorkers $ \wq -> do+              replicateM_ 100 $ enqueue wq $ threadDelay 1000+        computations `shouldSatisfy` isJust+    replicateM_ 1 $       it "should be resilient to errors and Exceptions" $ do         witness <- newIORef False         computations <-@@ -30,6 +31,20 @@               mapM_ (enqueue wq) [error "an error", throw RandomException, writeIORef witness True]         computations `shouldSatisfy` isJust         readIORef witness `shouldReturn` True+    replicateM_ 1 $+      it "enqueueing when working on a stopping work queue should run all jobs" $ do+        counter <- newIORef @Int 0+        let incrAnd wq next =+              enqueueAfter+                wq+                (threadDelay 100 >> atomicModifyIORef' counter (\n -> (n + 1, ())))+                [next]+        computations <-+          timeout 10000 $+            withPoolboy (poolboySettingsWith 10) waitingStopFinishWorkers $ \wq -> do+              replicateM_ 10 $ incrAnd wq $ incrAnd wq $ incrAnd wq $ incrAnd wq $ incrAnd wq $ return ()+        computations `shouldSatisfy` isJust+        readIORef counter `shouldReturn` 50  data RandomException = RandomException   deriving (Show)