Fix a bug in parDistributeScan and block on buffer in sendToWorker_

Author: adithyaov

src/Streamly/Internal/Data/Fold/Channel/Type.hs

@@ -510,10 +510,12 @@ sendToWorker_ chan a = go
                     (inputItemDoorBell chan)
                     (ChildYield a)
         else do
-            error "sendToWorker_: No space available in the buffer"
+            -- XXX: Is there a reason for us to error out here instead of
+            -- blocking?
+            -- error "sendToWorker_: No space available in the buffer"
             -- Block for space
-            -- () <- liftIO $ takeMVar (inputSpaceDoorBell chan)
-            -- go
+            () <- liftIO $ takeMVar (inputSpaceDoorBell chan)
+            go
 
 -- XXX Cleanup the fold if the stream is interrupted. Add a GC hook.
 

src/Streamly/Internal/Data/Fold/Concurrent.hs

@@ -401,7 +401,7 @@ parDistributeScan cfg getFolds (Stream sstep state) =
                 then do
                     liftIO $ takeMVar db
                     return $ Skip (ScanDrain q db running)
-                else return $ Yield outputs (ScanDrain q db running)
+                else return $ Yield outputs ScanStop
     step _ ScanStop = return Stop
 
 {-# ANN type DemuxState Fuse #-}

src/Streamly/Internal/Data/Scanl/Concurrent.hs

@@ -242,7 +242,8 @@ parDistributeScan cfg getFolds (Stream sstep state) =
                 then do
                     liftIO $ takeMVar db
                     return $ Skip (ScanDrain q db running)
-                else return $ Yield outputs (ScanDrain q db running)
+                else return $ Yield outputs ScanStop
+
     step _ ScanStop = return Stop
 
 {-# ANN type DemuxState Fuse #-}

streamly.cabal

@@ -136,6 +136,7 @@ extra-source-files:
     test/Streamly/Test/Prelude/*.hs
     test/Streamly/Test/Unicode/*.hs
     test/Streamly/Test/Serialize/*.hs
+    test/Streamly/Test/Data/Scanl/*.hs
     test/Streamly/Test/Data/Fold/*.hs
     test/lib/Streamly/Test/Common.hs
     test/lib/Streamly/Test/Prelude/Common.hs

test/Streamly/Test/Data/Scanl/Concurrent.hs

@@ -0,0 +1,62 @@
+-- |
+-- Module      : Streamly.Test.Data.Scanl.Concurrent
+-- Copyright   : (c) 2020 Composewell Technologies
+--
+-- License     : BSD-3-Clause
+-- Maintainer  : [email protected]
+-- Stability   : experimental
+-- Portability : GHC
+
+module Streamly.Test.Data.Scanl.Concurrent (main) where
+
+import Control.Concurrent (threadDelay)
+import Data.Function ( (&) )
+import Data.IORef (newIORef, atomicModifyIORef')
+import Data.List (sort)
+import Streamly.Data.Scanl (Scanl)
+import Test.Hspec as H
+
+import qualified Streamly.Data.Fold as Fold
+import qualified Streamly.Data.Stream as Stream
+import qualified Streamly.Data.Stream.Prelude as Stream
+import qualified Streamly.Internal.Data.Scanl as Scanl
+import qualified Streamly.Internal.Data.Scanl.Prelude as Scanl
+
+moduleName :: String
+moduleName = "Data.Scanl.Concurrent"
+
+---------------------------------------------------------------------------
+-- Main
+---------------------------------------------------------------------------
+
+evenScan :: Scanl IO Int (Maybe Int)
+evenScan =
+    Scanl.filtering even
+        & Scanl.lmapM (\x -> threadDelay 100 >> pure x)
+
+oddScan :: Scanl IO Int (Maybe Int)
+oddScan =
+    Scanl.filtering odd
+        & Scanl.lmapM (\x -> threadDelay 100 >> pure x)
+
+parDistributeScanTest :: (Stream.Config -> Stream.Config) -> IO ()
+parDistributeScanTest concOpts = do
+    ref <- newIORef [evenScan, oddScan]
+    let gen = atomicModifyIORef' ref (\xs -> ([], xs))
+        inpStream = Stream.enumerateFromTo 1 10_000
+    res1 <-
+        Scanl.parDistributeScan concOpts gen inpStream
+            & Stream.concatMap Stream.fromList
+            & Stream.catMaybes
+            & Stream.fold Fold.toList
+    sort res1 `shouldBe` [1..9999]
+
+main :: IO ()
+main = hspec
+  $ H.parallel
+#ifdef COVERAGE_BUILD
+  $ modifyMaxSuccess (const 10)
+#endif
+  $ describe moduleName $ do
+        it "parDistributeScan (maxBuffer 1)"
+            $ parDistributeScanTest (Stream.maxBuffer 1)

test/streamly-tests.cabal

@@ -304,6 +304,14 @@ test-suite Data.RingArray
   main-is: Streamly/Test/Data/RingArray.hs
   ghc-options: -main-is Streamly.Test.Data.RingArray.main
 
+test-suite Data.Scanl.Concurrent
+  import: test-options
+  type: exitcode-stdio-1.0
+  main-is: Streamly/Test/Data/Scanl/Concurrent.hs
+  ghc-options: -main-is Streamly.Test.Data.Scanl.Concurrent.main
+  if flag(use-streamly-core)
+    buildable: False
+
 -- XXX Rename to MutByteArray
 test-suite Data.Serialize
   import: test-options