@@ -1,1105 +0,0 @@
1−module Main (main) where
3−import Control.Concurrent (threadDelay)
4−import Control.Exception (Exception, try, ErrorCall(..), catch)
5−import Control.Monad (void)
6−import Control.Monad.Catch (throwM, MonadThrow)
7−import Control.Monad.Error.Class (throwError, MonadError)
8−import Control.Monad.IO.Class (MonadIO(liftIO))
9−import Control.Monad.State (MonadState, get, modify, runStateT
10− , StateT(..), evalStateT)
11−import Control.Monad.Trans.Except (runExceptT, ExceptT)
12−import Data.Foldable (forM_, fold)
13−import Data.Function ((&))
14−import Data.List (sort)
15−import Data.Maybe (fromJust, isJust)
16−#if !(MIN_VERSION_base(4,11,0))
17−import Data.Semigroup (Semigroup(..))
19−import System.Mem (performMajorGC)
22−import Test.Hspec as H
24−import Streamly.Prelude
25− ( AsyncT, SerialT, WSerialT, MonadAsync, IsStream, (.:), nil, adapt
26− , ahead, async, aheadly, asyncly, serially, parallely, wAsync, wAsyncly
27− , wSerially, maxBuffer)
28−import qualified Streamly.Prelude as S
29−import qualified Streamly.Internal.Data.Stream.IsStream as IP
31−singleton :: IsStream t => a -> t m a
32−singleton a = a .: nil
34−toListSerial :: SerialT IO a -> IO [a]
35−toListSerial = S.toList . serially
37−toListInterleaved :: WSerialT IO a -> IO [a]
38−toListInterleaved = S.toList . wSerially
40−toListAsync :: AsyncT IO a -> IO [a]
41−toListAsync = S.toList . asyncly
47− describe "restricts concurrency and cleans up extra tasks" $ do
48− it "take 1 asyncly" $ checkCleanup 2 asyncly (S.take 1)
49− it "take 1 wAsyncly" $ checkCleanup 2 wAsyncly (S.take 1)
50− it "take 1 aheadly" $ checkCleanup 2 aheadly (S.take 1)
52− it "takeWhile (< 0) asyncly" $ checkCleanup 2 asyncly (S.takeWhile (< 0))
53− it "takeWhile (< 0) wAsyncly" $ checkCleanup 2 wAsyncly (S.takeWhile (< 0))
54− it "takeWhile (< 0) aheadly" $ checkCleanup 2 aheadly (S.takeWhile (< 0))
57− let timed :: (IsStream t, Monad (t IO)) => Int -> t IO Int
58− timed x = S.yieldM (threadDelay (x * 100000)) >> return x
60− -- These are not run parallely because the timing gets affected
61− -- unpredictably when other tests are running on the same machine.
63− -- Also, they fail intermittently due to scheduling delays, so not run on
65− describe "Nested parallel and serial compositions" $ do
70− -- This is not correct, the result can also be [4,4,8,0,8,0,2,2]
71− -- because of parallelism of [8,0] and [8,0].
72− it "Nest <|>, <>, <|> (1)" $
75− ((t 8 <|> t 4) <> (t 2 <|> t 0))
76− <|> ((t 8 <|> t 4) <> (t 2 <|> t 0)))
77− `shouldReturn` ([4,4,8,8,0,0,2,2])
79− it "Nest <|>, <>, <|> (2)" $
80− (S.toList . wAsyncly) (
81− s (p (t 4 <> t 8) <> p (t 1 <> t 2))
82− <> s (p (t 4 <> t 8) <> p (t 1 <> t 2)))
83− `shouldReturn` ([4,4,8,8,1,1,2,2])
84− -- FIXME: These two keep failing intermittently on Mac OS X
85− -- Need to examine and fix the tests.
87− it "Nest <|>, <=>, <|> (1)" $
90− ((t 8 <|> t 4) <=> (t 2 <|> t 0))
91− <|> ((t 9 <|> t 4) <=> (t 2 <|> t 0)))
92− `shouldReturn` ([4,4,0,0,8,2,9,2])
93− it "Nest <|>, <=>, <|> (2)" $
96− ((t 4 <|> t 8) <=> (t 1 <|> t 2))
97− <|> ((t 4 <|> t 9) <=> (t 1 <|> t 2)))
98− `shouldReturn` ([4,4,1,1,8,2,9,2])
100− it "Nest <|>, <|>, <|>" $
101− (S.toList . wAsyncly) (
102− ((t 4 <> t 8) <> (t 0 <> t 2))
103− <> ((t 4 <> t 8) <> (t 0 <> t 2)))
104− `shouldReturn` ([0,0,2,2,4,4,8,8])
106− -- parallely fails on CI machines, may need more difference in times of
107− -- the events, but that would make tests even slower.
108− it "take 1 parallely" $ checkCleanup 3 parallely (S.take 1)
109− it "takeWhile (< 0) parallely" $ checkCleanup 3 parallely (S.takeWhile (< 0))
111− testFoldOpsCleanup "head" S.head
112− testFoldOpsCleanup "null" S.null
113− testFoldOpsCleanup "elem" (S.elem 0)
114− testFoldOpsCleanup "notElem" (S.notElem 0)
115− testFoldOpsCleanup "elemIndex" (S.elemIndex 0)
117− testFoldOpsCleanup "notElem" (S.notElem 0)
118− testFoldOpsCleanup "find" (S.find (==0))
119− testFoldOpsCleanup "findIndex" (S.findIndex (==0))
120− testFoldOpsCleanup "all" (S.all (==1))
121− testFoldOpsCleanup "any" (S.any (==0))
122− testFoldOpsCleanup "and" (S.and . S.map (==1))
123− testFoldOpsCleanup "or" (S.or . S.map (==0))
126− ---------------------------------------------------------------------------
127− -- Semigroup/Monoidal Composition strict ordering checks
128− ---------------------------------------------------------------------------
130− -- test both (<>) and mappend to make sure we are using correct instance
131− -- for Monoid that is using the right version of semigroup. Instance
132− -- deriving can cause us to pick wrong instances sometimes.
134− describe "WSerial interleaved (<>) ordering check" $ interleaveCheck wSerially (<>)
135− describe "WSerial interleaved mappend ordering check" $ interleaveCheck wSerially mappend
137− -- describe "WAsync interleaved (<>) ordering check" $ interleaveCheck wAsyncly (<>)
138− -- describe "WAsync interleaved mappend ordering check" $ interleaveCheck wAsyncly mappend
140− describe "Async (<>) time order check" $ parallelCheck asyncly (<>)
141− describe "Async mappend time order check" $ parallelCheck asyncly mappend
143− -- XXX this keeps failing intermittently, need to investigate
144− -- describe "WAsync (<>) time order check" $ parallelCheck wAsyncly (<>)
145− -- describe "WAsync mappend time order check" $ parallelCheck wAsyncly mappend
147− describe "Parallel (<>) time order check" $ parallelCheck parallely (<>)
148− describe "Parallel mappend time order check" $ parallelCheck parallely mappend
149− it "fromCallback" $ testFromCallback `shouldReturn` (50*101)
151−checkCleanup :: IsStream t
153− -> (t IO Int -> SerialT IO Int)
154− -> (t IO Int -> t IO Int)
156−checkCleanup d t op = do
157− r <- newIORef (-1 :: Int)
158− S.drain . serially $ do
159− _ <- t $ op $ delay r 0 S.|: delay r 1 S.|: delay r 2 S.|: S.nil
166− delay ref i = threadDelay (i*d*100000) >> writeIORef ref i >> return i
169−checkCleanupFold :: IsStream t
170− => (t IO Int -> SerialT IO Int)
171− -> (SerialT IO Int -> IO (Maybe Int))
173−checkCleanupFold t op = do
174− r <- newIORef (-1 :: Int)
175− _ <- op $ t $ delay r 0 S.|: delay r 1 S.|: delay r 2 S.|: S.nil
181− delay ref i = threadDelay (i*200000) >> writeIORef ref i >> return i
183−testFoldOpsCleanup :: String -> (SerialT IO Int -> IO a) -> Spec
184−testFoldOpsCleanup name f = do
185− let testOp op x = op x >> return Nothing
186− it (name <> " asyncly") $ checkCleanupFold asyncly (testOp f)
187− it (name <> " wAsyncly") $ checkCleanupFold wAsyncly (testOp f)
188− it (name <> " aheadly") $ checkCleanupFold aheadly (testOp f)
189− it (name <> " parallely") $ checkCleanupFold parallely (testOp f)
192−parallelTests :: SpecWith ()
193−parallelTests = H.parallel $ do
195− ---------------------------------------------------------------------------
196− -- TBD Monoidal composition combinations
197− ---------------------------------------------------------------------------
199− -- TBD need more such combinations to be tested.
200− describe "serial <> and serial <>" $
201− composeAndComposeSimple serially serially (repeat [1 .. 9])
202− describe "ahead <> and ahead <>" $
203− composeAndComposeSimple aheadly aheadly (repeat [1 .. 9])
204− describe "ahead <> and serial <>" $
205− composeAndComposeSimple aheadly serially (repeat [1 .. 9])
206− describe "serial <> and ahead <>" $
207− composeAndComposeSimple serially aheadly (repeat [1 .. 9])
209− describe "<> and <=>" $ composeAndComposeSimple
214− , [1, 3, 2, 4, 6, 5, 7, 9, 8]
215− , [1, 3, 2, 4, 6, 5, 7, 9, 8]
218− describe "<=> and <=>" $ composeAndComposeSimple
221− [ [1, 4, 2, 7, 3, 5, 8, 6, 9]
222− , [1, 7, 4, 8, 2, 9, 5, 3, 6]
223− , [1, 4, 3, 7, 2, 6, 9, 5, 8]
224− , [1, 7, 4, 9, 3, 8, 6, 2, 5]
227− describe "<=> and <>" $ composeAndComposeSimple
230− [ [1, 4, 2, 7, 3, 5, 8, 6, 9]
231− , [1, 7, 4, 8, 2, 9, 5, 3, 6]
232− , [1, 4, 2, 7, 3, 5, 8, 6, 9]
233− , [1, 7, 4, 8, 2, 9, 5, 3, 6]
236− ---------------------------------------------------------------------------
237− -- Monoidal composition recursion loops
238− ---------------------------------------------------------------------------
240− describe "Serial loops" $ loops serially id reverse
241− describe "Ahead loops" $ loops aheadly id reverse
242− describe "Async parallel loops" $ loops asyncly sort sort
243− describe "WAsync loops" $ loops wAsyncly sort sort
244− describe "parallel loops" $ loops parallely sort sort
246− ---------------------------------------------------------------------------
247− -- Bind and monoidal composition combinations
248− ---------------------------------------------------------------------------
250− describe "Bind and compose Stream 1" $ bindAndComposeSimple serially serially
251− describe "Bind and compose Stream 2" $ bindAndComposeSimple serially wSerially
252− describe "Bind and compose Stream 3" $ bindAndComposeSimple serially asyncly
253− describe "Bind and compose Stream 4" $ bindAndComposeSimple serially wAsyncly
254− describe "Bind and compose Stream 5" $ bindAndComposeSimple serially parallely
255− describe "Bind and compose Stream 6" $ bindAndComposeSimple serially aheadly
257− describe "Bind and compose Ahead Stream 0" $ bindAndComposeSimple aheadly aheadly
258− describe "Bind and compose Ahead Stream 1" $ bindAndComposeSimple aheadly serially
259− describe "Bind and compose Ahead Stream 2" $ bindAndComposeSimple aheadly wSerially
260− describe "Bind and compose Ahead Stream 3" $ bindAndComposeSimple aheadly asyncly
261− describe "Bind and compose Ahead Stream 4" $ bindAndComposeSimple aheadly wAsyncly
262− describe "Bind and compose Ahead Stream 5" $ bindAndComposeSimple aheadly parallely
264− describe "Bind and compose Costream 1" $ bindAndComposeSimple wSerially serially
265− describe "Bind and compose Costream 2" $ bindAndComposeSimple wSerially wSerially
266− describe "Bind and compose Costream 3" $ bindAndComposeSimple wSerially asyncly
267− describe "Bind and compose Costream 4" $ bindAndComposeSimple wSerially wAsyncly
268− describe "Bind and compose Costream 5" $ bindAndComposeSimple wSerially parallely
269− describe "Bind and compose Costream 6" $ bindAndComposeSimple wSerially aheadly
271− describe "Bind and compose Async 1" $ bindAndComposeSimple asyncly serially
272− describe "Bind and compose Async 2" $ bindAndComposeSimple asyncly wSerially
273− describe "Bind and compose Async 3" $ bindAndComposeSimple asyncly asyncly
274− describe "Bind and compose Async 4" $ bindAndComposeSimple asyncly wAsyncly
275− describe "Bind and compose Async 5" $ bindAndComposeSimple asyncly parallely
276− describe "Bind and compose Async 6" $ bindAndComposeSimple asyncly aheadly
278− describe "Bind and compose WAsync 1" $ bindAndComposeSimple wAsyncly serially
279− describe "Bind and compose WAsync 2" $ bindAndComposeSimple wAsyncly wSerially
280− describe "Bind and compose WAsync 3" $ bindAndComposeSimple wAsyncly asyncly
281− describe "Bind and compose WAsync 4" $ bindAndComposeSimple wAsyncly wAsyncly
282− describe "Bind and compose WAsync 5" $ bindAndComposeSimple wAsyncly parallely
283− describe "Bind and compose WAsync 6" $ bindAndComposeSimple wAsyncly aheadly
285− describe "Bind and compose Parallel 1" $ bindAndComposeSimple parallely serially
286− describe "Bind and compose Parallel 2" $ bindAndComposeSimple parallely wSerially
287− describe "Bind and compose Parallel 3" $ bindAndComposeSimple parallely asyncly
288− describe "Bind and compose Parallel 4" $ bindAndComposeSimple parallely wAsyncly
289− describe "Bind and compose Parallel 5" $ bindAndComposeSimple parallely parallely
290− describe "Bind and compose Parallel 6" $ bindAndComposeSimple parallely aheadly
292− let fldr, fldl :: (IsStream t, Semigroup (t IO Int)) => [t IO Int] -> t IO Int
293− fldr = foldr (<>) nil
294− fldl = foldl (<>) nil
296− forM_ [fldr, fldl] $ \k ->
297− describe "Bind and compose" $ bindAndComposeHierarchy serially serially k
298− forM_ [fldr, fldl] $ \k ->
299− describe "Bind and compose" $ bindAndComposeHierarchy serially wSerially k
300− forM_ [fldr, fldl] $ \k ->
301− describe "Bind and compose" $ bindAndComposeHierarchy serially asyncly k
302− forM_ [fldr, fldl] $ \k ->
303− describe "Bind and compose" $ bindAndComposeHierarchy serially wAsyncly k
304− forM_ [fldr, fldl] $ \k ->
305− describe "Bind and compose" $ bindAndComposeHierarchy serially parallely k
306− forM_ [fldr, fldl] $ \k ->
307− describe "Bind and compose serially aheadly" $ bindAndComposeHierarchy serially aheadly k
309− forM_ [fldr, fldl] $ \k ->
310− describe "Bind and compose aheadly serially" $ bindAndComposeHierarchy aheadly serially k
311− forM_ [fldr, fldl] $ \k ->
312− describe "Bind and compose aheadly wSerially" $ bindAndComposeHierarchy aheadly wSerially k
313− forM_ [fldr, fldl] $ \k ->
314− describe "Bind and compose aheadly asyncly" $ bindAndComposeHierarchy aheadly asyncly k
315− forM_ [fldr, fldl] $ \k ->
316− describe "Bind and compose aheadly wAsyncly" $ bindAndComposeHierarchy aheadly wAsyncly k
317− forM_ [fldr, fldl] $ \k ->
318− describe "Bind and compose aheadly parallely" $ bindAndComposeHierarchy aheadly parallely k
319− forM_ [fldr, fldl] $ \k ->
320− describe "Bind and compose serially aheadly" $ bindAndComposeHierarchy aheadly aheadly k
322− forM_ [fldr, fldl] $ \k ->
323− describe "Bind and compose" $ bindAndComposeHierarchy wSerially serially k
324− forM_ [fldr, fldl] $ \k ->
325− describe "Bind and compose" $ bindAndComposeHierarchy wSerially wSerially k
326− forM_ [fldr, fldl] $ \k ->
327− describe "Bind and compose" $ bindAndComposeHierarchy wSerially asyncly k
328− forM_ [fldr, fldl] $ \k ->
329− describe "Bind and compose" $ bindAndComposeHierarchy wSerially wAsyncly k
330− forM_ [fldr, fldl] $ \k ->
331− describe "Bind and compose" $ bindAndComposeHierarchy wSerially parallely k
332− forM_ [fldr, fldl] $ \k ->
333− describe "Bind and compose wserially aheadly" $ bindAndComposeHierarchy wSerially aheadly k
335− forM_ [fldr, fldl] $ \k ->
336− describe "Bind and compose" $ bindAndComposeHierarchy asyncly serially k
337− forM_ [fldr, fldl] $ \k ->
338− describe "Bind and compose" $ bindAndComposeHierarchy asyncly wSerially k
339− forM_ [fldr, fldl] $ \k ->
340− describe "Bind and compose" $ bindAndComposeHierarchy asyncly asyncly k
341− forM_ [fldr, fldl] $ \k ->
342− describe "Bind and compose" $ bindAndComposeHierarchy asyncly wAsyncly k
343− forM_ [fldr, fldl] $ \k ->
344− describe "Bind and compose" $ bindAndComposeHierarchy asyncly parallely k
345− forM_ [fldr, fldl] $ \k ->
346− describe "Bind and compose asyncly aheadly" $ bindAndComposeHierarchy asyncly aheadly k
348− forM_ [fldr, fldl] $ \k ->
349− describe "Bind and compose" $ bindAndComposeHierarchy wAsyncly serially k
350− forM_ [fldr, fldl] $ \k ->
351− describe "Bind and compose" $ bindAndComposeHierarchy wAsyncly wSerially k
352− forM_ [fldr, fldl] $ \k ->
353− describe "Bind and compose" $ bindAndComposeHierarchy wAsyncly asyncly k
354− forM_ [fldr, fldl] $ \k ->
355− describe "Bind and compose" $ bindAndComposeHierarchy wAsyncly wAsyncly k
356− forM_ [fldr, fldl] $ \k ->
357− describe "Bind and compose" $ bindAndComposeHierarchy wAsyncly parallely k
358− forM_ [fldr, fldl] $ \k ->
359− describe "Bind and compose wAsyncly aheadly" $ bindAndComposeHierarchy wAsyncly aheadly k
361− forM_ [fldr, fldl] $ \k ->
362− describe "Bind and compose" $ bindAndComposeHierarchy parallely serially k
363− forM_ [fldr, fldl] $ \k ->
364− describe "Bind and compose" $ bindAndComposeHierarchy parallely wSerially k
365− forM_ [fldr, fldl] $ \k ->
366− describe "Bind and compose" $ bindAndComposeHierarchy parallely asyncly k
367− forM_ [fldr, fldl] $ \k ->
368− describe "Bind and compose" $ bindAndComposeHierarchy parallely wAsyncly k
369− forM_ [fldr, fldl] $ \k ->
370− describe "Bind and compose" $ bindAndComposeHierarchy parallely parallely k
371− forM_ [fldr, fldl] $ \k ->
372− describe "Bind and compose parallely aheadly" $ bindAndComposeHierarchy parallely aheadly k
374− -- Nest two lists using different styles of product compositions
375− it "Nests two streams using monadic serial composition" nestTwoSerial
376− it "Nests two streams using monadic ahead composition" nestTwoAhead
377− it "Nests two streams using monadic interleaved composition" nestTwoInterleaved
378− it "Nests two streams using monadic Async composition" nestTwoAsync
379− it "Nests two streams using monadic WAsync composition" nestTwoWAsync
380− it "Nests two streams using monadic parallel composition" nestTwoParallel
382− it "Nests two streams using applicative serial composition" nestTwoSerialApp
383− it "Nests two streams using applicative ahead composition" nestTwoAheadApp
384− it "Nests two streams using applicative interleaved composition" nestTwoInterleavedApp
385− it "Nests two streams using applicative Async composition" nestTwoAsyncApp
386− it "Nests two streams using applicative WAsync composition" nestTwoWAsyncApp
387− it "Nests two streams using applicative parallel composition" nestTwoParallelApp
389− ---------------------------------------------------------------------------
390− -- TBD Bind and Bind combinations
391− ---------------------------------------------------------------------------
393− -- TBD combine all binds and all compose in one example
394− describe "Miscellaneous combined examples" mixedOps
395− describe "Miscellaneous combined examples aheadly" mixedOpsAheadly
396− describe "Simple MonadError and MonadThrow" simpleMonadError
399− describe "Composed MonadError serially" $ composeWithMonadError serially
400− describe "Composed MonadError wSerially" $ composeWithMonadError wSerially
401− describe "Composed MonadError asyncly" $ composeWithMonadError asyncly
402− describe "Composed MonadError wAsyncly" $ composeWithMonadError wAsyncly
405− describe "Composed MonadThrow serially" $ composeWithMonadThrow serially
406− describe "Composed MonadThrow wSerially" $ composeWithMonadThrow wSerially
407− describe "Composed MonadThrow asyncly" $ composeWithMonadThrow asyncly
408− describe "Composed MonadThrow wAsyncly" $ composeWithMonadThrow wAsyncly
409− describe "Composed MonadThrow parallely" $ composeWithMonadThrow parallely
410− describe "Composed MonadThrow aheadly" $ composeWithMonadThrow aheadly
412− describe "take on infinite concurrent stream" $ takeInfinite asyncly
413− describe "take on infinite concurrent stream" $ takeInfinite wAsyncly
414− describe "take on infinite concurrent stream" $ takeInfinite aheadly
416− ---------------------------------------------------------------------------
417− -- Some ad-hoc tests that failed at times
418− ---------------------------------------------------------------------------
420− it "takes n from stream of streams" (takeCombined 1 aheadly)
421− it "takes n from stream of streams" (takeCombined 2 asyncly)
422− it "takes n from stream of streams" (takeCombined 3 wAsyncly)
424− ---------------------------------------------------------------------------
425− -- Left folds are strict enough
426− ---------------------------------------------------------------------------
429− it "foldx is strict enough" checkFoldxStrictness
430− it "scanx is strict enough" checkScanxStrictness
431− it "foldxM is strict enough" (checkFoldMStrictness foldxMStrictCheck)
433− it "foldl' is strict enough" checkFoldl'Strictness
434− it "scanl' is strict enough" checkScanl'Strictness
435− it "foldlM' is strict enough" (checkFoldMStrictness foldlM'StrictCheck)
436− it "scanlM' is strict enough" (checkScanlMStrictness scanlM'StrictCheck)
438− ---------------------------------------------------------------------------
439− -- Right folds are lazy enough
440− ---------------------------------------------------------------------------
442− it "foldrM is lazy enough" checkFoldrLaziness
444− ---------------------------------------------------------------------------
445− -- Monadic state transfer in concurrent tasks
446− ---------------------------------------------------------------------------
448− -- XXX Can we write better test cases to hit every case?
450− it "async: state is saved and used if the work is partially enqueued"
451− (checkMonadicStateTransfer async)
453− it "wAsync: state is saved and used if the work is partially enqueued"
454− (checkMonadicStateTransfer wAsync)
456− it "ahead: state is saved and used if the work is partially enqueued"
457− (checkMonadicStateTransfer ahead)
459− ---------------------------------------------------------------------------
460− -- Monadic state snapshot in concurrent tasks
461− ---------------------------------------------------------------------------
463− it "asyncly maintains independent states in concurrent tasks"
464− (monadicStateSnapshot asyncly)
465− it "asyncly limited maintains independent states in concurrent tasks"
466− (monadicStateSnapshot (asyncly . S.take 10000))
467− it "wAsyncly maintains independent states in concurrent tasks"
468− (monadicStateSnapshot wAsyncly)
469− it "wAsyncly limited maintains independent states in concurrent tasks"
470− (monadicStateSnapshot (wAsyncly . S.take 10000))
471− it "aheadly maintains independent states in concurrent tasks"
472− (monadicStateSnapshot aheadly)
473− it "aheadly limited maintains independent states in concurrent tasks"
474− (monadicStateSnapshot (aheadly . S.take 10000))
475− it "parallely maintains independent states in concurrent tasks"
476− (monadicStateSnapshot parallely)
478− it "async maintains independent states in concurrent tasks"
479− (monadicStateSnapshotOp async)
480− it "ahead maintains independent states in concurrent tasks"
481− (monadicStateSnapshotOp ahead)
482− it "wAsync maintains independent states in concurrent tasks"
483− (monadicStateSnapshotOp wAsync)
484− it "parallel maintains independent states in concurrent tasks"
485− (monadicStateSnapshotOp S.parallel)
487− ---------------------------------------------------------------------------
488− -- Slower tests are at the end
489− ---------------------------------------------------------------------------
491− ---------------------------------------------------------------------------
493− ---------------------------------------------------------------------------
495− it "asyncly crosses thread limit (2000 threads)" $
496− S.drain (asyncly $ fold $
497− replicate 2000 $ S.yieldM $ threadDelay 1000000)
500− it "aheadly crosses thread limit (4000 threads)" $
501− S.drain (aheadly $ fold $
502− replicate 4000 $ S.yieldM $ threadDelay 1000000)
505−-- Each snapshot carries an independent state. Multiple parallel tasks should
506−-- not affect each other's state. This is especially important when we run
507−-- multiple tasks in a single thread.
508−snapshot :: (IsStream t, MonadAsync m, MonadState Int m) => t m ()
510− -- We deliberately use a replicate count 1 here, because a lower count
511− -- catches problems that a higher count doesn't.
513− -- Even though we modify the state here it should not reflect in other
514− -- parallel tasks, it is local to each concurrent task.
515− modify (+1) >> get >>= liftIO . (`shouldSatisfy` (==1))
516− modify (+1) >> get >>= liftIO . (`shouldSatisfy` (==2))
518−snapshot1 :: (IsStream t, MonadAsync m, MonadState Int m) => t m ()
519−snapshot1 = S.replicateM 1000 $
520− modify (+1) >> get >>= liftIO . (`shouldSatisfy` (==2))
522−snapshot2 :: (IsStream t, MonadAsync m, MonadState Int m) => t m ()
523−snapshot2 = S.replicateM 1000 $
524− modify (+1) >> get >>= liftIO . (`shouldSatisfy` (==2))
529− , Semigroup (t m ())
532− , MonadState Int (t m)
536− -- Each task in a concurrent composition inherits the state and maintains
537− -- its own modifications to it, not affecting the parent computation.
538− snapshot <> (modify (+1) >> (snapshot1 <> snapshot2))
539− -- The above modify statement does not affect our state because that is
540− -- used in a parallel composition. In a serial composition it will affect
542− get >>= liftIO . (`shouldSatisfy` (== (0 :: Int)))
546− , Semigroup (t (StateT Int IO) ())
547− , MonadIO (t (StateT Int IO))
548− , MonadState Int (t (StateT Int IO))
550− => (t (StateT Int IO) () -> SerialT (StateT Int IO) ()) -> IO ()
551−monadicStateSnapshot t = void $ runStateT (S.drain $ t stateComp) 0
554− :: ( AsyncT (StateT Int IO) ()
555− -> AsyncT (StateT Int IO) ()
556− -> AsyncT (StateT Int IO) ()
558− -> SerialT (StateT Int IO) ()
560− -- Each task in a concurrent composition inherits the state and maintains
561− -- its own modifications to it, not affecting the parent computation.
562− asyncly (snapshot `op` (modify (+1) >> (snapshot1 `op` snapshot2)))
563− -- The above modify statement does not affect our state because that is
564− -- used in a parallel composition. In a serial composition it will affect
566− get >>= liftIO . (`shouldSatisfy` (== (0 :: Int)))
568−monadicStateSnapshotOp
569− :: ( AsyncT (StateT Int IO) ()
570− -> AsyncT (StateT Int IO) ()
571− -> AsyncT (StateT Int IO) ()
574−monadicStateSnapshotOp op = void $ runStateT (S.drain $ stateCompOp op) 0
576−takeCombined :: (Monad m, Semigroup (t m Int), Show a, Eq a, IsStream t)
577− => Int -> (t m Int -> SerialT IO a) -> IO ()
578−takeCombined n t = do
579− let constr = S.fromFoldable
580− r <- (S.toList . t) $
581− S.take n (constr ([] :: [Int]) <> constr ([] :: [Int]))
584−checkMonadicStateTransfer
585− :: (IsStream t1, IsStream t2)
586− => ( t1 (StateT Int IO) ()
587− -> t2 (StateT Int IO) ()
588− -> SerialT (StateT Int IO) a3 )
590−checkMonadicStateTransfer op = evalStateT str (0 :: Int)
595− (serially $ S.mapM snapshoti $ S.fromList [1..10]) `op`
596− (serially $ S.mapM snapshoti $ S.fromList [1..10])
600− lift1 $ x `shouldBe` y
601− lift1 m = StateT $ \s -> do
605−checkFoldrLaziness :: IO ()
606−checkFoldrLaziness = do
607− S.foldrM (\x xs -> if odd x then return True else xs)
608− (return False) (S.fromList (2:4:5:undefined :: [Int]))
611− S.toList (IP.foldrS (\x xs -> if odd x then return True else xs)
613− $ (S.fromList (2:4:5:undefined) :: SerialT IO Int))
614− `shouldReturn` [True]
616− S.toList (IP.foldrT (\x xs -> if odd x then return True else xs)
618− $ (S.fromList (2:4:5:undefined) :: SerialT IO Int))
619− `shouldReturn` [True]
622−checkFoldxStrictness :: IO ()
623−checkFoldxStrictness = do
624− let s = return (1 :: Int) `S.consM` error "failure"
625− catch (S.foldx (\_ a -> if a == 1 then error "success" else "done")
627− (\(ErrorCall err) -> return err)
628− `shouldReturn` "success"
631−checkFoldl'Strictness :: IO ()
632−checkFoldl'Strictness = do
633− let s = return (1 :: Int) `S.consM` error "failure"
634− catch (S.foldl' (\_ a -> if a == 1 then error "success" else "done")
636− (\(ErrorCall err) -> return err)
637− `shouldReturn` "success"
640−checkScanxStrictness :: IO ()
641−checkScanxStrictness = do
642− let s = return (1 :: Int) `S.consM` error "failure"
647− then error "success"
651− >> return "finished"
653− (\(ErrorCall err) -> return err)
654− `shouldReturn` "success"
657−checkScanl'Strictness :: IO ()
658−checkScanl'Strictness = do
659− let s = return (1 :: Int) `S.consM` error "failure"
665− then error "success"
669− >> return "finished"
671− (\(ErrorCall err) -> return err)
672− `shouldReturn` "success"
674−foldlM'StrictCheck :: IORef Int -> SerialT IO Int -> IO ()
675−foldlM'StrictCheck ref = S.foldlM' (\_ _ -> writeIORef ref 1) (return ())
678−foldxMStrictCheck :: IORef Int -> SerialT IO Int -> IO ()
679−foldxMStrictCheck ref = S.foldxM (\_ _ -> writeIORef ref 1) (return ()) return
682−checkFoldMStrictness :: (IORef Int -> SerialT IO Int -> IO ()) -> IO ()
683−checkFoldMStrictness f = do
685− let s = return 1 `S.consM` error "x"
686− catch (f ref s) (\(_ :: ErrorCall) -> return ())
687− readIORef ref `shouldReturn` 1
689−scanlM'StrictCheck :: IORef Int -> SerialT IO Int -> SerialT IO ()
690−scanlM'StrictCheck ref = S.scanlM' (\_ _ -> writeIORef ref 1) ()
692−checkScanlMStrictness :: (IORef Int -> SerialT IO Int -> SerialT IO ()) -> IO ()
693−checkScanlMStrictness f = do
695− let s = return 1 `S.consM` error "x"
696− catch (S.drain $ f ref s) (\(_ :: ErrorCall) -> return ())
697− readIORef ref `shouldReturn` 1
699−takeInfinite :: IsStream t => (t IO Int -> SerialT IO Int) -> Spec
702− S.drain (t $ S.take 1 $ S.repeatM (print "hello" >> return (1::Int)))
705−-- XXX need to test that we have promptly cleaned up everything after the error
706−-- XXX We can also check the output that we are expected to get before the
709−newtype ExampleException = ExampleException String deriving (Eq, Show)
711−instance Exception ExampleException
713−simpleMonadError :: Spec
714−simpleMonadError = do
716− it "simple runExceptT" $ do
717− (runExceptT $ S.drain $ return ())
718− `shouldReturn` (Right () :: Either String ())
719− it "simple runExceptT with error" $ do
720− (runExceptT $ S.drain $ throwError "E") `shouldReturn` Left "E"
723− try (S.drain $ return ())
724− `shouldReturn` (Right () :: Either ExampleException ())
725− it "simple try with throw error" $
726− try (S.drain $ throwM $ ExampleException "E")
727− `shouldReturn` (Left (ExampleException "E") :: Either ExampleException ())
729−composeWithMonadThrow
731− , Semigroup (t IO Int)
734− => (t IO Int -> SerialT IO Int) -> Spec
735−composeWithMonadThrow t = do
736− it "Compose throwM, nil" $
737− try (tl (throwM (ExampleException "E") <> S.nil))
738− `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])
739− it "Compose nil, throwM" $
740− try (tl (S.nil <> throwM (ExampleException "E")))
741− `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])
742− oneLevelNestedSum "serially" serially
743− oneLevelNestedSum "wSerially" wSerially
744− oneLevelNestedSum "asyncly" asyncly
745− oneLevelNestedSum "wAsyncly" wAsyncly
746− -- XXX add two level nesting
748− oneLevelNestedProduct "serially" serially
749− oneLevelNestedProduct "wSerially" wSerially
750− oneLevelNestedProduct "asyncly" asyncly
751− oneLevelNestedProduct "wAsyncly" wAsyncly
755− oneLevelNestedSum desc t1 =
756− it ("One level nested sum " <> desc) $ do
757− let nested = S.fromFoldable [1..10] <> throwM (ExampleException "E")
758− <> S.fromFoldable [1..10]
759− try (tl (S.nil <> t1 nested <> S.fromFoldable [1..10]))
760− `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])
762− oneLevelNestedProduct desc t1 =
763− it ("One level nested product" <> desc) $ do
764− let s1 = t $ S.concatMapFoldableWith (<>) return [1..4]
765− s2 = t1 $ S.concatMapFoldableWith (<>) return [5..8]
770− then throwM (ExampleException "E")
773− `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])
775−_composeWithMonadError
777− , Semigroup (t (ExceptT String IO) Int)
778− , MonadError String (t (ExceptT String IO))
780− => (t (ExceptT String IO) Int -> SerialT (ExceptT String IO) Int) -> Spec
781−_composeWithMonadError t = do
782− let tl = S.toList . t
783− it "Compose throwError, nil" $
784− runExceptT (tl (throwError "E" <> S.nil)) `shouldReturn` Left "E"
785− it "Compose nil, error" $
786− runExceptT (tl (S.nil <> throwError "E")) `shouldReturn` Left "E"
788−nestTwoSerial :: Expectation
790− let s1 = S.concatMapFoldableWith (<>) return [1..4]
791− s2 = S.concatMapFoldableWith (<>) return [5..8]
796− ) `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
798−nestTwoAhead :: Expectation
800− let s1 = S.concatMapFoldableWith (<>) return [1..4]
801− s2 = S.concatMapFoldableWith (<>) return [5..8]
802− in (S.toList . aheadly) (do
806− ) `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
808−nestTwoSerialApp :: Expectation
810− let s1 = S.concatMapFoldableWith (<>) return [1..4]
811− s2 = S.concatMapFoldableWith (<>) return [5..8]
812− in toListSerial ((+) <$> s1 <*> s2)
813− `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
815−nestTwoAheadApp :: Expectation
817− let s1 = S.concatMapFoldableWith (<>) return [1..4]
818− s2 = S.concatMapFoldableWith (<>) return [5..8]
819− in (S.toList . aheadly) ((+) <$> s1 <*> s2)
820− `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
822−nestTwoInterleaved :: Expectation
824− let s1 = S.concatMapFoldableWith (<>) return [1..4]
825− s2 = S.concatMapFoldableWith (<>) return [5..8]
826− in toListInterleaved (do
830− ) `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
832−nestTwoInterleavedApp :: Expectation
833−nestTwoInterleavedApp =
834− let s1 = S.concatMapFoldableWith (<>) return [1..4]
835− s2 = S.concatMapFoldableWith (<>) return [5..8]
836− in toListInterleaved ((+) <$> s1 <*> s2)
837− `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
839−nestTwoAsync :: Expectation
841− let s1 = S.concatMapFoldableWith (<>) return [1..4]
842− s2 = S.concatMapFoldableWith (<>) return [5..8]
843− in sort <$> toListAsync (do
847− `shouldReturn` sort ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
849−nestTwoAsyncApp :: Expectation
851− let s1 = S.concatMapFoldableWith (<>) return [1..4]
852− s2 = S.concatMapFoldableWith (<>) return [5..8]
853− in sort <$> toListAsync ((+) <$> s1 <*> s2)
854− `shouldReturn` sort ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])
856−nestTwoWAsync :: Expectation
858− let s1 = S.concatMapFoldableWith (<>) return [1..4]
859− s2 = S.concatMapFoldableWith (<>) return [5..8]
860− in sort <$> (S.toList . wAsyncly) (do
864− `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
866−nestTwoParallel :: Expectation
868− let s1 = S.concatMapFoldableWith (<>) return [1..4]
869− s2 = S.concatMapFoldableWith (<>) return [5..8]
870− in sort <$> (S.toList . parallely) (do
874− `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
876−nestTwoWAsyncApp :: Expectation
878− let s1 = S.concatMapFoldableWith (<>) return [1..4]
879− s2 = S.concatMapFoldableWith (<>) return [5..8]
880− in sort <$> (S.toList . wAsyncly) ((+) <$> s1 <*> s2)
881− `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
883−nestTwoParallelApp :: Expectation
885− let s1 = S.concatMapFoldableWith (<>) return [1..4]
886− s2 = S.concatMapFoldableWith (<>) return [5..8]
887− in sort <$> (S.toList . parallely) ((+) <$> s1 <*> s2)
888− `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])
890−interleaveCheck :: IsStream t
891− => (t IO Int -> SerialT IO Int)
892− -> (t IO Int -> t IO Int -> t IO Int)
894−interleaveCheck t f =
895− it "Interleave four" $
896− (S.toList . t) ((singleton 0 `f` singleton 1) `f` (singleton 100 `f` singleton 101))
897− `shouldReturn` [0, 100, 1, 101]
899−parallelCheck :: (IsStream t, Monad (t IO))
900− => (t IO Int -> SerialT IO Int)
901− -> (t IO Int -> t IO Int -> t IO Int)
903−parallelCheck t f = do
904− it "Parallel ordering left associated" $
905− (S.toList . t) (((event 4 `f` event 3) `f` event 2) `f` event 1)
906− `shouldReturn` [1..4]
908− it "Parallel ordering right associated" $
909− (S.toList . t) (event 4 `f` (event 3 `f` (event 2 `f` event 1)))
910− `shouldReturn` [1..4]
912− where event n = S.yieldM (threadDelay (n * 200000)) >> return n
914−composeAndComposeSimple
915− :: ( IsStream t1, Semigroup (t1 IO Int)
916− , IsStream t2, Monoid (t2 IO Int), Monad (t2 IO)
917−#if __GLASGOW_HASKELL__ < 804
918− , Semigroup (t2 IO Int)
921− => (t1 IO Int -> SerialT IO Int)
922− -> (t2 IO Int -> t2 IO Int)
924−composeAndComposeSimple t1 t2 answer = do
925− let rfold = adapt . t2 . S.concatMapFoldableWith (<>) return
926− it "Compose right associated outer expr, right folded inner" $
927− (S.toList . t1) (rfold [1,2,3] <> (rfold [4,5,6] <> rfold [7,8,9]))
928− `shouldReturn` head answer
930− it "Compose left associated outer expr, right folded inner" $
931− (S.toList . t1) ((rfold [1,2,3] <> rfold [4,5,6]) <> rfold [7,8,9])
932− `shouldReturn` (answer !! 1)
934− let lfold xs = adapt $ t2 $ foldl (<>) mempty $ fmap return xs
935− it "Compose right associated outer expr, left folded inner" $
936− (S.toList . t1) (lfold [1,2,3] <> (lfold [4,5,6] <> lfold [7,8,9]))
937− `shouldReturn` (answer !! 2)
939− it "Compose left associated outer expr, left folded inner" $
940− (S.toList . t1) ((lfold [1,2,3] <> lfold [4,5,6]) <> lfold [7,8,9])
941− `shouldReturn` (answer !! 3)
944− :: (IsStream t, Semigroup (t IO Int), Monad (t IO))
945− => (t IO Int -> t IO Int)
949−loops t tsrt hsrt = do
950− it "Tail recursive loop" $ (tsrt <$> (S.toList . adapt) (loopTail 0))
951− `shouldReturn` [0..3]
953− it "Head recursive loop" $ (hsrt <$> (S.toList . adapt) (loopHead 0))
954− `shouldReturn` [0..3]
958− -- this print line is important for the test (causes a bind)
959− S.yieldM $ putStrLn "LoopHead..."
960− t $ (if x < 3 then loopHead (x + 1) else nil) <> return x
963− -- this print line is important for the test (causes a bind)
964− S.yieldM $ putStrLn "LoopTail..."
965− t $ return x <> (if x < 3 then loopTail (x + 1) else nil)
968− :: ( IsStream t1, IsStream t2, Semigroup (t2 IO Int), Monad (t2 IO))
969− => (t1 IO Int -> SerialT IO Int)
970− -> (t2 IO Int -> t2 IO Int)
972−bindAndComposeSimple t1 t2 = do
973− -- XXX need a bind in the body of concatForFoldableWith instead of a simple return
974− it "Compose many (right fold) with bind" $
975− (sort <$> (S.toList . t1)
976− (adapt . t2 $ S.concatForFoldableWith (<>) [1..10 :: Int] return))
977− `shouldReturn` [1..10]
979− it "Compose many (left fold) with bind" $
980− let forL xs k = foldl (<>) nil $ fmap k xs
981− in (sort <$> (S.toList . t1) (adapt . t2 $ forL [1..10 :: Int] return))
982− `shouldReturn` [1..10]
984−bindAndComposeHierarchy
985− :: ( IsStream t1, Monad (t1 IO)
986− , IsStream t2, Monad (t2 IO))
987− => (t1 IO Int -> SerialT IO Int)
988− -> (t2 IO Int -> t2 IO Int)
989− -> ([t2 IO Int] -> t2 IO Int)
991−bindAndComposeHierarchy t1 t2 g =
992− it "Bind and compose nested" $
993− (sort <$> (S.toList . t1) bindComposeNested)
994− `shouldReturn` (sort (
1000− <> replicate 7 15) :: [Int])
1004− -- bindComposeNested :: WAsyncT IO Int
1005− bindComposeNested =
1006− let c1 = tripleCompose (return 1) (return 2) (return 3)
1007− c2 = tripleCompose (return 4) (return 5) (return 6)
1008− c3 = tripleCompose (return 7) (return 8) (return 9)
1009− b = tripleBind c1 c2 c3
1010−-- it seems to be causing a huge space leak in hspec so disabling this for now
1011−-- c = tripleCompose b b b
1012−-- m = tripleBind c c c
1016− tripleCompose a b c = adapt . t2 $ g [a, b, c]
1017− tripleBind mx my mz =
1020− >>= \z -> return (x + y + z)
1024− it "Compose many ops" $
1025− (sort <$> toListSerial composeMixed)
1026− `shouldReturn` ([8,9,9,9,9,9,10,10,10,10,10,10,10,10,10,10,11,11
1027− ,11,11,11,11,11,11,11,11,12,12,12,12,12,13
1031− composeMixed :: SerialT IO Int
1033− S.yieldM $ return ()
1034− S.yieldM $ putStr ""
1038− x1 <- wAsyncly $ return 1 <> return 2
1039− S.yieldM $ return ()
1040− S.yieldM $ putStr ""
1041− y1 <- asyncly $ return 1 <> return 2
1043− x11 <- return 1 <> return 2
1044− y11 <- asyncly $ return 1 <> return 2
1045− z11 <- wSerially $ return 1 <> return 2
1046− S.yieldM $ return ()
1047− S.yieldM $ putStr ""
1048− return (x11 + y11 + z11)
1049− return (x1 + y1 + z1)
1052−mixedOpsAheadly :: Spec
1054− it "Compose many ops" $
1055− (sort <$> toListSerial composeMixed)
1056− `shouldReturn` ([8,9,9,9,9,9,10,10,10,10,10,10,10,10,10,10,11,11
1057− ,11,11,11,11,11,11,11,11,12,12,12,12,12,13
1061− composeMixed :: SerialT IO Int
1063− S.yieldM $ return ()
1064− S.yieldM $ putStr ""
1068− x1 <- wAsyncly $ return 1 <> return 2
1069− S.yieldM $ return ()
1070− S.yieldM $ putStr ""
1071− y1 <- aheadly $ return 1 <> return 2
1073− x11 <- return 1 <> return 2
1074− y11 <- aheadly $ return 1 <> return 2
1075− z11 <- parallely $ return 1 <> return 2
1076− S.yieldM $ return ()
1077− S.yieldM $ putStr ""
1078− return (x11 + y11 + z11)
1079− return (x1 + y1 + z1)
1082−testFromCallback :: IO Int
1083−testFromCallback = do
1084− ref <- newIORef Nothing
1085− let stream = S.map Just (IP.fromCallback (setCallback ref))
1086− `S.parallel` runCallback ref
1087− S.sum $ S.map fromJust $ S.takeWhile isJust stream
1091− setCallback ref cb = do
1092− writeIORef ref (Just cb)
1094− runCallback ref = S.yieldM $ do
1096− S.repeatM (readIORef ref)
1101− S.fromList [1..100]
1102− & IP.delayPost 0.001
1103− & S.mapM_ (fromJust cb)