conduit-concurrent-map 0.1.2 → 0.1.3

raw patch · 3 files changed

+44/−16 lines, 3 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• conduit-concurrent-map.cabal +1/−1
• src/Data/Conduit/ConcurrentMap.hs +28/−15
• test/Main.hs +15/−0

@@ -1,5 +1,5 @@ name:          conduit-concurrent-map-version:       0.1.2+version:       0.1.3 license:       MIT copyright:     2017 Niklas Hambüchen <mail@nh2.me> author:        Niklas Hambüchen <mail@nh2.me>

@@ -96,9 +96,16 @@ --   doesn't change the order). -- * Bounded memory: The conduit will only hold to --   @numThreads * (workerOutputBufferSize + 1)@ as many @b@s.--- * Full utilisation: The conduit will try to keep all cores busy as much as---   it can. This means that it will always try to `await` if there's a free---   core, and will only `yield` once it has to to make a core free.+-- * High utilisation: The conduit will try to keep all cores busy as much as+--   it can. This means that after `await`ing an input, it will only block+--   to wait for an output from a worker thread if it has to because+--   we're at the `workerOutputBufferSize` output buffer bound of `b` elements.+--   (It may, however, `yield` even if the queue is not full.+--   Since `yield` will block the conduit's thread until downstream+--   conduits in the pipeline `await`, utilisation will be poor if other+--   conduits in the pipeline have low throughput.+--   This makes sense because a conduit pipeline's total throughput+--   is bottlenecked by the segment in the pipeline.) --   It also ensures that any worker running for longer than others does not --   prevent other free workers from starting new work, except from when --   we're at the `workerOutputBufferSize` output buffer bound of `b` elements.@@ -308,24 +315,18 @@       --      Cruise phase doesn't happen if the conduit terminates before       --      `numThreads` elements are awaited.       -- 3) Drain phase,-      --      in which we drain off the `numWorkersRampedUp` elements that we-      --      know must be in the queue (due to above invariant),-      --      drain off all elements stored in output buffers,+      --      in which we drain off the `numInQueue` elements in the queue,       --      send all workers the stop signal and wait for their orderly termination.        let loop :: Int -> Int -> ConduitT a b m ()           loop numWorkersRampedUp numInQueue = do              await >>= \case-              Nothing -> do -- Drain phase: Upstream conduit is done, tell all workers to finish.-                for_ [1..numWorkersRampedUp] $ \_ -> do-                  yieldQueueHead -- This will succeed due to the "Cruise phase invariant", see above.-                  putInVar Nothing -- This will not block forever because we just freed an `outVar` in the line above.-                for_ [1..(numThreads - numWorkersRampedUp)] $ \_ -> do -- need to quit workers that were never ramped up too-                  putInVar Nothing-                let numInQueueAfterStopping = numInQueue - numWorkersRampedUp-                for_ [1..numInQueueAfterStopping] $ \_ -> do-                  yieldQueueHead+              Nothing -> do -- Drain phase: Upstream conduit is done.+                for_ [1..numInQueue] $ \_ -> do+                  yieldQueueHead -- Drain the queue.+                for_ [1..numThreads] $ \_ -> do+                  putInVar Nothing -- tell all workers to finish.                 wait workersAsync -- wait for workers to shut down                Just a@@ -340,8 +341,20 @@                     -- Cruise phase:                      putInVar (Just a) >> waitForSignal inVarEnqueued+                    -- `waitForSignal` will not block forever because at least the worker+                    -- in the head of `outQueue` will always be able to take the value:+                    -- Either:+                    -- 1. it is currently running `f`, in which case its `workerOutVar`+                    --    is empty, it will eventually write the `b` into it, and then+                    --    be ready to take the `inVar`.+                    -- 2. or it has already done that and is currently doing `takeMVar invar`+                    --                     -- At the time `waitForSignal inVarEnqueued` completes, we know                     -- that there is a `workerOutVar` in the `outQueue` we can wait for.+                    --+                    -- If it was indeed the `workerOutVar` of the head worker,+                    -- Then we will take that `workerOutVar` below below, to restoring+                    -- the above invariant for the next head worker.                      let numInQueueAfterEnqueued = numInQueue + 1 

@@ -8,6 +8,7 @@ import           Data.Conduit import qualified Data.Conduit.Combinators as CC import qualified Data.Conduit.List as CL+import           Data.Foldable (for_) import           Test.Hspec import           Test.QuickCheck import           Test.QuickCheck.Monadic (monadicIO, run, pick, assert)@@ -71,6 +72,20 @@         .| CL.consume        l `shouldBe` [2,4,6,8,10,12]++    it "does not hang when 3 elements are processed by the same thread in order" $ do+      for_ [(1::Int)..100] $ \t -> do -- try many times due to timing+        sayString ("Test " ++ show t)+        l <- runConduitRes $+             CL.sourceList [1,   0,0,0,   0,0,0,   0,0,0]+          .| concurrentMapM_ 4 2+               (\i -> liftIO $ do+                  threadDelay (i * 1000)+                  return (i*2)+               )+          .| CL.consume++        l `shouldBe` [2,0,0,0,0,0,0,0,0,0]    describe "concurrentMapM_" $ do     it "is like mapM" $ prop_concurrentMapM_is_like_mapM