streamly-0.11.0: test/Streamly/Test/Data/Stream/Concurrent.hs
-- |
-- Module : Streamly.Test.Data.Stream.Concurrent
-- Copyright : (c) 2020 Composewell Technologies
--
-- License : BSD-3-Clause
-- Maintainer : streamly@composewell.com
-- Stability : experimental
-- Portability : GHC
module Streamly.Test.Data.Stream.Concurrent (main) where
import Control.Concurrent (threadDelay)
import Control.Exception (Exception, try)
import Control.Monad (replicateM)
import Control.Monad.Catch (throwM)
import Data.Function ( (&) )
import Data.IORef ( newIORef, readIORef, writeIORef)
import Data.List (sort)
import Data.Maybe ( isJust, fromJust )
import Data.Word (Word8)
import Streamly.Data.Stream (Stream)
import Test.Hspec.QuickCheck
import Test.QuickCheck (Testable, Property, choose, forAll, withMaxSuccess)
import Test.QuickCheck.Monadic (monadicIO, run)
import Test.Hspec as H
import qualified Streamly.Data.Fold as Fold
import qualified Streamly.Internal.Data.Stream as Stream
import qualified Streamly.Internal.Data.Stream.Prelude as Stream
import qualified Streamly.Internal.Data.Stream.Prelude as Async
import Streamly.Test.Common (listEquals)
moduleName :: String
moduleName = "Data.Stream.Concurrent"
-------------------------------------------------------------------------------
-- Utilities
-------------------------------------------------------------------------------
sortEq :: Ord a => [a] -> [a] -> Bool
sortEq a b = sort a == sort b
cmp :: (Show a, Ord a) => ([a] -> [a] -> Bool) -> [a] -> Stream IO a -> Property
cmp eq list s =
monadicIO $ do
stream <- run $ sort <$> Stream.fold Fold.toList s
listEquals eq stream list
prop1 :: Testable prop => String -> prop -> SpecWith ()
prop1 x y = modifyMaxSuccess (const 1) $ prop x y
-- Coverage build takes too long with default number of tests
maxTestCount :: Int
#ifdef DEVBUILD
maxTestCount = 100
#else
maxTestCount = 10
#endif
transformCombineFromList ::
([Int] -> Stream IO Int)
-> ([Int] -> [Int] -> Bool)
-> ([Int] -> [Int])
-> (Stream IO Int -> Stream IO Int)
-> [Int]
-> [Int]
-> [Int]
-> Property
transformCombineFromList constr eq listOp op a b c =
withMaxSuccess maxTestCount $
monadicIO $ do
let s1 = op (Async.parList id [constr b, constr c])
let s2 = Async.parList id [constr a, s1]
stream <- run (Stream.fold Fold.toList s2)
let list = a <> listOp (b <> c)
listEquals eq stream list
commonOpConfigs :: [(String, Async.Config -> Async.Config)]
commonOpConfigs =
[ ("default", id)
#ifndef COVERAGE_BUILD
, ("rate AvgRate 10000", Async.avgRate 10000)
, ("rate Nothing", Async.rate Nothing)
, ("maxBuffer 0", Async.maxBuffer 0)
, ("maxThreads 0", Async.maxThreads 0)
, ("maxThreads 1", Async.maxThreads 1)
, ("eager", Async.eager True)
-- XXX Need to use an unsorted eq operation for ahead
, ("ordered", Async.ordered True)
#ifdef USE_LARGE_MEMORY
, ("maxThreads -1", Async.maxThreads (-1))
#endif
#endif
]
opConfigs :: [(String, Async.Config -> Async.Config)]
opConfigs = commonOpConfigs
++ [
#ifndef COVERAGE_BUILD
("maxBuffer 1", Async.maxBuffer 1)
#endif
]
makeSpec :: [(String, a)] -> (a -> Spec) -> Spec
makeSpec cfg spec = mapM_ (\(desc, arg) -> describe desc $ spec arg) cfg
asyncSpec :: ((Async.Config -> Async.Config) -> Spec) -> Spec
asyncSpec =
makeSpec $ opConfigs
#ifndef COVERAGE_BUILD
<> [("maxBuffer (-1)", Async.maxBuffer (-1))]
#endif
-------------------------------------------------------------------------------
-- Compose with MonadThrow
-------------------------------------------------------------------------------
newtype ExampleException = ExampleException String deriving (Eq, Show, Ord)
instance Exception ExampleException
exceptionPropagation ::
(Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec
exceptionPropagation f = do
it "append throwM, nil" $
try (tl (Stream.fromEffect (throwM (ExampleException "E")) `f` Stream.nil))
`shouldReturn`
(Left (ExampleException "E") :: Either ExampleException [Int])
it "append nil, throwM" $
try (tl (Stream.nil `f` Stream.fromEffect (throwM (ExampleException "E"))))
`shouldReturn`
(Left (ExampleException "E") :: Either ExampleException [Int])
it "append nested throwM" $ do
let nested =
(Stream.fromList [1..10])
`f` Stream.fromEffect (throwM (ExampleException "E"))
`f` (Stream.fromList [1..10])
try (tl (Stream.nil `f` nested `f` (Stream.fromList [1..10])))
`shouldReturn`
(Left (ExampleException "E")
:: Either ExampleException [Int])
it "sequence throwM" $
let stream = Stream.fromList [throwM (ExampleException "E")]
in try (tl (Stream.nil `f` Async.parSequence id stream))
`shouldReturn`
(Left (ExampleException "E") :: Either ExampleException [Int])
it "concatMap throwM" $ do
let s1 = Async.parList id $ fmap Stream.fromPure [1..4]
s2 = Async.parList id $ fmap Stream.fromPure [5..8]
try $ tl (
let bind = flip (Async.parConcatMap id)
in bind s1 $ \x ->
bind s2 $ \y ->
if x + y > 10
then Stream.fromEffect (throwM (ExampleException "E"))
else Stream.fromPure (x + y)
)
`shouldReturn`
(Left (ExampleException "E") :: Either ExampleException [Int])
where
tl = Stream.fold Fold.toList
---------------------------------------------------------------------------
-- Time ordering
---------------------------------------------------------------------------
#ifdef DEVBUILD
timeOrdering :: ([Stream IO Int] -> Stream IO Int) -> Spec
timeOrdering f = do
it "Parallel event ordering check" $
Stream.fold Fold.toList (f [event 4, event 3, event 2, event 1])
`shouldReturn` [1..4]
where event n = Stream.fromEffect (threadDelay (n * 200000) >> return n)
#endif
-------------------------------------------------------------------------------
-- Some ad-hoc tests that failed at times
-------------------------------------------------------------------------------
takeCombined :: Int -> IO ()
takeCombined n = do
let constr = Stream.fromList
let s = Async.parList id [constr ([] :: [Int]), constr ([] :: [Int])]
r <- Stream.fold Fold.toList $ Stream.take n s
r `shouldBe` []
---------------------------------------------------------------------------
-- Main
---------------------------------------------------------------------------
constructWithLenM
:: (Int -> Stream IO Int)
-> (Int -> IO [Int])
-> Word8
-> Property
constructWithLenM mkStream mkList len =
withMaxSuccess maxTestCount
$ monadicIO $ do
stream <-
run
$ Stream.fold Fold.toList
$ mkStream (fromIntegral len)
list <- run $ mkList (fromIntegral len)
listEquals (==) stream list
sequenceReplicate
:: (Async.Config -> Async.Config)
-> Word8
-> Property
sequenceReplicate cfg = constructWithLenM stream list
where
list = flip replicateM (return 1 :: IO Int)
stream = Async.parSequence cfg . flip Stream.replicate (return 1 :: IO Int)
drainMapM :: Monad m => (a -> m b) -> Stream m a -> m ()
drainMapM f = Stream.fold (Fold.drainMapM f)
testFromCallback :: IO Int
testFromCallback = do
ref <- newIORef Nothing
let stream =
Stream.parList (Stream.eager True)
[ fmap Just (Stream.fromCallback (setCallback ref))
, runCallback ref
]
Stream.fold Fold.sum $ fmap fromJust $ Stream.takeWhile isJust stream
where
setCallback ref cb = do
writeIORef ref (Just cb)
runCallback ref = Stream.fromEffect $ do
cb <-
Stream.repeatM (readIORef ref)
& Stream.delayPost 0.1
& Stream.mapMaybe id
& Stream.fold Fold.one
Stream.fromList [1..100]
& Stream.delayPost 0.001
& drainMapM (fromJust cb)
threadDelay 100000
return Nothing
main :: IO ()
main = hspec
$ H.parallel
#ifdef COVERAGE_BUILD
$ modifyMaxSuccess (const 10)
#endif
$ describe moduleName $ do
let transform = transformCombineFromList Stream.fromList sortEq
prop "parBuffered" $
transform
(fmap (+2))
(fmap (+1) . Async.parBuffered id . fmap (+1))
asyncSpec $ prop "parSequence" . sequenceReplicate
asyncSpec $
prop "parMapM (+1)"
. transform (fmap (+1))
. (`Async.parMapM` (\x -> return (x + 1)))
-- XXX Need to use eq instead of sortEq for ahead oeprations
-- Binary append
asyncSpec $
let appWith cfg = Async.parList cfg [Stream.nil, Stream.nil]
in prop1 "parList [] []" . cmp sortEq ([] :: [Int]) . appWith
asyncSpec $
let appWith cfg = Async.parList cfg [Stream.nil, Stream.fromPure 1]
in prop1 "parList [] [1]" . cmp sortEq [1 :: Int] . appWith
asyncSpec $
let appWith cfg = Async.parList cfg [Stream.fromPure 1, Stream.nil]
in prop1 "parList [1] []" . cmp sortEq [1 :: Int] . appWith
asyncSpec $
let appWith cfg =
Async.parList cfg [Stream.fromPure 0, Stream.fromPure 1]
in prop1 "parList [0] [1]" . cmp sortEq [0, 1 :: Int] . appWith
asyncSpec $
let appWith cfg =
Async.parList
cfg [Stream.fromPure 0, Stream.nil, Stream.fromPure 1]
in prop1 "parList [0] [] [1]" . cmp sortEq [0, 1 :: Int] . appWith
asyncSpec $
let appWith cfg =
Async.parTwo cfg
(Async.parTwo cfg
(Async.parTwo cfg
(Stream.fromPure 0) (Stream.fromPure 1))
(Stream.fromPure 2))
(Stream.fromPure 3)
in prop1 "parTwo left associated"
. cmp sortEq [0, 1, 2, 3 :: Int] . appWith
asyncSpec $
let appWith cfg =
Async.parTwo cfg
(Stream.fromPure 0)
(Async.parTwo cfg
(Stream.fromPure 1)
(Async.parTwo cfg
(Stream.fromPure 2) (Stream.fromPure 3))
)
in prop1 "parTwo right associated"
. cmp sortEq [0, 1, 2, 3 :: Int] . appWith
asyncSpec $
let leaf x y cfg =
Async.parTwo cfg (Stream.fromPure x)
(Stream.fromPure y)
leaf11 cfg =
Async.parTwo cfg (leaf 0 1 cfg) $ leaf 2 (3 :: Int) cfg
leaf12 cfg =
Async.parTwo cfg (leaf 4 5 cfg) $ leaf 6 7 cfg
appWith cfg =
Async.parTwo cfg (leaf11 cfg) (leaf12 cfg)
in prop1 "parTwo balanced"
. cmp sortEq [0, 1, 2, 3, 4, 5, 6,7] . appWith
asyncSpec $
let appWith cfg =
Async.parTwo cfg
(Stream.fromList [1,2,3,4,5 :: Int])
(Stream.fromList [6,7,8,9,10])
in prop1 "parTwo" . cmp (==) [1,2,3,4,5,6,7,8,9,10] . appWith
asyncSpec $
let par2 cfg =
Async.parTwo
cfg
(Stream.fromPure 1)
(Stream.fromPure 2)
s1 cfg =
Async.parCrossApply
cfg
(Stream.fromPure (,))
(par2 cfg)
s2 cfg =
Async.parCrossApply
cfg
(s1 cfg)
(Stream.fromPure 3) :: Stream IO (Int, Int)
in prop1
"parCrossApply (async arg1)" . cmp (==) ( [(1, 3), (2, 3)]) . s2
asyncSpec $
let par2 cfg =
Async.parTwo
cfg
(Stream.fromPure (2 :: Int))
(Stream.fromPure 3)
s1 = Stream.fromPure (1 :: Int,)
s2 cfg = Async.parCrossApply cfg s1 (par2 cfg)
in prop1 "apply (async arg2)" . cmp (==) ([(1, 2), (1, 3)]) . s2
-- concat
asyncSpec $
let stream cfg =
Async.parConcat cfg
$ fmap Stream.fromPure
$ Stream.fromList [1..100 :: Int]
in prop1 "parConcat" . cmp sortEq [1..100] . stream
asyncSpec $
let f cfg =
forAll (choose (0, 100)) $ \n ->
transform
(concatMap (const [1..n]))
(Async.parConcatMap
cfg (const (Stream.fromList [1..n]))
)
in prop "parConcatMap" . f
#ifdef DEVBUILD
describe "Time ordering" $ timeOrdering (Async.parList id)
#endif
let async = Async.parTwo id
describe "Exception propagation" $ exceptionPropagation async
-- Ad-hoc tests
it "takes n from stream of streams" $ takeCombined 2
it "fromCallback" $ testFromCallback `shouldReturn` (50*101)