6c9d241 · streamly · git.iko.soy

git.iko.soy/streamly

ran src through hlint, made most changes; added .hlint.yaml.

commit6c9d241f73a1be87c998e9605289871e1d4a0177 authorKeith <wygulmage@users.noreply.github.com> dateMon 1 Oct 2018, 11:02 (8 years ago) parent1c9e948

10 files changed+208−157

compare with master

unifiedsplit

.hlint.yaml+70 src/Streamly/Prelude.hs+5−5 src/Streamly/SVar.hs+65−84 src/Streamly/Streams/Ahead.hs+11−13 src/Streamly/Streams/Async.hs+3−3 src/Streamly/Streams/Parallel.hs+2−2 src/Streamly/Streams/SVar.hs+10−8 src/Streamly/Streams/Serial.hs+4−4 src/Streamly/Streams/StreamD.hs+8−8 src/Streamly/Streams/StreamK.hs+30−30

.hlint.yamlnew file+70

@@ -0,0 +1,70 @@

1+# HLint configuration file

2+# https://github.com/ndmitchell/hlint

3+##########################

5+# This file contains a template configuration file, which is typically

6+# placed as .hlint.yaml in the root of your project

9+# Warnings currently triggered by your code

10+- suggest: {name: "Unused LANGUAGE pragma"}

11+- suggest: {name: "Eta reduce"}

12+- suggest: {name: "Reduce duplication"}

13+- ignore: {name: "Use list literal pattern"}

14+- ignore: {name: "Use tuple-section"} # requires GHC extension

15+- ignore: {name: "Use fromMaybe"} # may want to use this suggestion, but it didn't match the common idiom of the library

16+- ignore: {name: "Use unless"} # low power-to-weight

17+

18+

19+# Specify additional command line arguments

20+#

21+# - arguments: [--color, --cpp-simple, -XQuasiQuotes]

22+

23+

24+# Control which extensions/flags/modules/functions can be used

25+#

26+# - extensions:

27+# - default: false # all extension are banned by default

28+# - name: [PatternGuards, ViewPatterns] # only these listed extensions can be used

29+# - {name: CPP, within: CrossPlatform} # CPP can only be used in a given module

30+#

31+# - flags:

32+# - {name: -w, within: []} # -w is allowed nowhere

33+#

34+# - modules:

35+# - {name: [Data.Set, Data.HashSet], as: Set} # if you import Data.Set qualified, it must be as 'Set'

36+# - {name: Control.Arrow, within: []} # Certain modules are banned entirely

37+#

38+# - functions:

39+# - {name: unsafePerformIO, within: []} # unsafePerformIO can only appear in no modules

40+

41+

42+# Add custom hints for this project

43+#

44+# Will suggest replacing "wibbleMany [myvar]" with "wibbleOne myvar"

45+# - error: {lhs: "wibbleMany [x]", rhs: wibbleOne x}

46+

47+

48+# Turn on hints that are off by default

49+#

50+# Ban "module X(module X) where", to require a real export list

51+# - warn: {name: Use explicit module export list}

52+#

53+# Replace a $ b $ c with a . b $ c

54+# - group: {name: dollar, enabled: true}

55+#

56+# Generalise map to fmap, ++ to <>

57+- group: {name: generalise, enabled: true}

58+

59+

60+# Ignore some builtin hints

61+# - ignore: {name: Use let}

62+# - ignore: {name: Use const, within: SpecialModule} # Only within certain modules

63+

64+

65+# Define some custom infix operators

66+# - fixity: infixr 3 ~^#^~

67+

68+

69+# To generate a suitable file for HLint do:

70+# $ hlint --default > .hlint.yaml

src/Streamly/Prelude.hs+5−5

@@ -379,7 +379,7 @@ unfoldrMSerial step seed = fromStreamS (S.unfoldrM step seed)

379379-- @since 0.4.0

380380{-# INLINE yield #-}

381381yield :: IsStream t => a -> t m a

382−yield a = K.yield a

382+yield = K.yield

383383

384384-- | Create a singleton stream from a monadic action. Same as @m \`consM` nil@

385385-- but more efficient.

@@ -393,7 +393,7 @@ yield a = K.yield a

393393-- @since 0.4.0

394394{-# INLINE yieldM #-}

395395yieldM :: (Monad m, IsStream t) => m a -> t m a

396−yieldM m = K.yieldM m

396+yieldM = K.yieldM

397397

398398-- | Generate a stream by performing a monadic action @n@ times. Can be

399399-- expressed as @stimes n (yieldM m)@.

@@ -627,14 +627,14 @@ foldlM' step begin m = S.foldlM' step begin $ toStreamS m

627627-- @since 0.1.1

628628{-# INLINE null #-}

629629null :: Monad m => SerialT m a -> m Bool

630−null m = K.null m

630+null = K.null

631631

632632-- | Extract the first element of the stream, if any.

633633--

634634-- @since 0.1.0

635635{-# INLINE head #-}

636636head :: Monad m => SerialT m a -> m (Maybe a)

637−head m = K.head m

637+head = K.head

638638

639639-- | Extract all but the first element of the stream, if any.

640640--

@@ -814,7 +814,7 @@ toHandle h m = go (toStream m)

814814 let stop = return ()

815815 single a = liftIO (IO.hPutStrLn h a)

816816 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r

817− in (K.unStream m1) defState stop single yieldk

817+ in K.unStream m1 defState stop single yieldk

818818

819819------------------------------------------------------------------------------

820820-- Transformation by Folding (Scans)

src/Streamly/SVar.hs+65−84

@@ -496,11 +496,9 @@ getStreamRate = _maxStreamRate

496496setStreamLatency :: Int -> State t m a -> State t m a

497497setStreamLatency n st =

498498 st { _streamLatency =

499− if n < 0

499+ if n <= 0

500500 then Nothing

501− else if n == 0

502− then Nothing

503− else Just (fromIntegral n)

501+ else Just (fromIntegral n)

504502 }

505503

506504getStreamLatency :: State t m a -> Maybe NanoSecs

@@ -513,14 +511,14 @@ getStreamLatency = _streamLatency

513511cleanupSVar :: SVar t m a -> IO ()

514512cleanupSVar sv = do

515513 workers <- readIORef (workerThreads sv)

516− Prelude.mapM_ (\tid -> throwTo tid ThreadAbort)

514+ Prelude.mapM_ (`throwTo` ThreadAbort)

517515 (S.toList workers)

518516

519517cleanupSVarFromWorker :: SVar t m a -> IO ()

520518cleanupSVarFromWorker sv = do

521519 workers <- readIORef (workerThreads sv)

522520 self <- myThreadId

523− mapM_ (\tid -> throwTo tid ThreadAbort)

521+ mapM_ (`throwTo` ThreadAbort)

524522 (S.toList workers \\ [self])

525523

526524-------------------------------------------------------------------------------

@@ -751,6 +749,21 @@ atomicModifyIORefCAS ref fn = do

751749 then return result

752750 else loop tkt (tries - 1)

753751

752+{-# INLINE ringDoorBell #-}

753+ringDoorBell :: SVar t m a -> IO ()

754+ringDoorBell sv = do

755+ storeLoadBarrier

756+ w <- readIORef $ needDoorBell sv

757+ when w $ do

758+ -- Note: the sequence of operations is important for correctness here.

759+ -- We need to set the flag to false strictly before sending the

760+ -- outputDoorBell, otherwise the outputDoorBell may get processed too early and

761+ -- then we may set the flag to False to later making the consumer lose

762+ -- the flag, even without receiving a outputDoorBell.

763+ atomicModifyIORefCAS_ (needDoorBell sv) (const False)

764+ void $ tryPutMVar (outputDoorBell sv) ()

765+

766+

754767------------------------------------------------------------------------------

755768-- Spawning threads and collecting result in streamed fashion

756769------------------------------------------------------------------------------

@@ -770,7 +783,7 @@ type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

770783{-# INLINE rawForkIO #-}

771784rawForkIO :: IO () -> IO ThreadId

772785rawForkIO action = IO $ \ s ->

773− case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)

786+ case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)

774787

775788{-# INLINE doFork #-}

776789doFork :: MonadBaseControl IO m

@@ -857,7 +870,7 @@ send sv msg = do

857870 Unlimited -> return True

858871 Limited lim -> do

859872 active <- readIORef (workerCount sv)

860− return $ len < ((fromIntegral lim) - active)

873+ return $ len < (fromIntegral lim - active)

861874

862875workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecs))

863876workerCollectLatency winfo = do

@@ -865,7 +878,7 @@ workerCollectLatency winfo = do

865878 cnt1 <- readIORef (workerYieldCount winfo)

866879 let cnt = cnt1 - cnt0

867880

868− if (cnt > 0)

881+ if cnt > 0

869882 then do

870883 t1 <- getTime Monotonic

871884 let period = fromInteger $ toNanoSecs (t1 - t0)

@@ -919,7 +932,7 @@ checkRatePeriodic :: SVar t m a

919932checkRatePeriodic sv yinfo winfo ycnt = do

920933 i <- readIORef (workerPollingInterval yinfo)

921934 -- XXX use generation count to check if the interval has been updated

922− if (i /= 0 && (ycnt `mod` i) == 0)

935+ if i /= 0 && (ycnt `mod` i) == 0

923936 then do

924937 workerUpdateLatency yinfo winfo

925938 -- XXX not required for parallel streams

@@ -961,12 +974,11 @@ workerStopUpdate winfo info = do

961974sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()

962975sendStop sv mwinfo = do

963976 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1

964− case mwinfo of

965− Just winfo ->

966− case yieldRateInfo sv of

967− Nothing -> return ()

968− Just info -> workerStopUpdate winfo info

969− Nothing -> return ()

977+ case (mwinfo, yieldRateInfo sv) of

978+ (Just winfo, Just info) ->

979+ workerStopUpdate winfo info

980+ _ ->

981+ return ()

970982 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)

971983

972984-------------------------------------------------------------------------------

@@ -981,16 +993,7 @@ sendStop sv mwinfo = do

981993enqueueLIFO :: SVar t m a -> IORef [t m a] -> t m a -> IO ()

982994enqueueLIFO sv q m = do

983995 atomicModifyIORefCAS_ q $ \ms -> m : ms

984− storeLoadBarrier

985− w <- readIORef $ needDoorBell sv

986− when w $ do

987− -- Note: the sequence of operations is important for correctness here.

988− -- We need to set the flag to false strictly before sending the

989− -- outputDoorBell, otherwise the outputDoorBell may get processed too early and

990− -- then we may set the flag to False to later making the consumer lose

991− -- the flag, even without receiving a outputDoorBell.

992− atomicModifyIORefCAS_ (needDoorBell sv) (const False)

993− void $ tryPutMVar (outputDoorBell sv) ()

996+ ringDoorBell sv

994997

995998-------------------------------------------------------------------------------

996999-- WAsync

@@ -1004,16 +1007,7 @@ enqueueLIFO sv q m = do

10041007enqueueFIFO :: SVar t m a -> LinkedQueue (t m a) -> t m a -> IO ()

10051008enqueueFIFO sv q m = do

10061009 pushL q m

1007− storeLoadBarrier

1008− w <- readIORef $ needDoorBell sv

1009− when w $ do

1010− -- Note: the sequence of operations is important for correctness here.

1011− -- We need to set the flag to false strictly before sending the

1012− -- outputDoorBell, otherwise the outputDoorBell may get processed too early and

1013− -- then we may set the flag to False to later making the consumer lose

1014− -- the flag, even without receiving a outputDoorBell.

1015− atomicModifyIORefCAS_ (needDoorBell sv) (const False)

1016− void $ tryPutMVar (outputDoorBell sv) ()

1010+ ringDoorBell sv

10171011

10181012-------------------------------------------------------------------------------

10191013-- Ahead

@@ -1085,16 +1079,7 @@ enqueueAhead sv q m = do

10851079 atomicModifyIORefCAS_ q $ \ case

10861080 ([], n) -> ([m], n + 1) -- increment sequence

10871081 _ -> error "not empty"

1088− storeLoadBarrier

1089− w <- readIORef $ needDoorBell sv

1090− when w $ do

1091− -- Note: the sequence of operations is important for correctness here.

1092− -- We need to set the flag to false strictly before sending the

1093− -- outputDoorBell, otherwise the outputDoorBell may get processed too early and

1094− -- then we may set the flag to False to later making the consumer lose

1095− -- the flag, even without receiving a outputDoorBell.

1096− atomicModifyIORefCAS_ (needDoorBell sv) (const False)

1097− void $ tryPutMVar (outputDoorBell sv) ()

1082+ ringDoorBell sv

10981083

10991084-- enqueue without incrementing the sequence number

11001085{-# INLINE reEnqueueAhead #-}

@@ -1103,11 +1088,7 @@ reEnqueueAhead sv q m = do

11031088 atomicModifyIORefCAS_ q $ \ case

11041089 ([], n) -> ([m], n) -- DO NOT increment sequence

11051090 _ -> error "not empty"

1106− storeLoadBarrier

1107− w <- readIORef $ needDoorBell sv

1108− when w $ do

1109− atomicModifyIORefCAS_ (needDoorBell sv) (const False)

1110− void $ tryPutMVar (outputDoorBell sv) ()

1091+ ringDoorBell sv

11111092

11121093-- Normally the thread that has the token should never go away. The token gets

11131094-- handed over to another thread, but someone or the other has the token at any

@@ -1138,7 +1119,7 @@ queueEmptyAhead q = liftIO $ do

11381119{-# INLINE dequeueAhead #-}

11391120dequeueAhead :: MonadIO m

11401121 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))

1141−dequeueAhead q = liftIO $ do

1122+dequeueAhead q = liftIO $

11421123 atomicModifyIORefCAS q $ \case

11431124 ([], n) -> (([], n), Nothing)

11441125 (x : [], n) -> (([], n), Just (x, n))

@@ -1235,7 +1216,7 @@ addThread sv tid =

12351216{-# INLINE delThread #-}

12361217delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()

12371218delThread sv tid =

1238− liftIO $ modifyIORef (workerThreads sv) $ (\s -> S.delete tid s)

1219+ liftIO $ modifyIORef (workerThreads sv) (S.delete tid)

12391220

12401221-- If present then delete else add. This takes care of out of order add and

12411222-- delete i.e. a delete arriving before we even added a thread.

@@ -1245,14 +1226,13 @@ delThread sv tid =

12451226modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()

12461227modifyThread sv tid = do

12471228 changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->

1248− if (S.member tid old)

1249− then let new = (S.delete tid old) in (new, new)

1250− else let new = (S.insert tid old) in (new, old)

1251− if null changed

1252− then liftIO $ do

1253− writeBarrier

1254− void $ tryPutMVar (outputDoorBell sv) ()

1255− else return ()

1229+ if S.member tid old

1230+ then let new = S.delete tid old in (new, new)

1231+ else let new = S.insert tid old in (new, old)

1232+ when (null changed) $

1233+ liftIO $ do

1234+ writeBarrier

1235+ void $ tryPutMVar (outputDoorBell sv) ()

12561236

12571237-- | This is safe even if we are adding more threads concurrently because if

12581238-- a child thread is adding another thread then anyway 'workerThreads' will

@@ -1292,7 +1272,7 @@ pushWorker yieldMax sv = do

12921272 cntRef <- newIORef 0

12931273 t <- getTime Monotonic

12941274 lat <- newIORef (0, t)

1295− return $ Just $ WorkerInfo

1275+ return $ Just WorkerInfo

12961276 { workerYieldMax = yieldMax

12971277 , workerYieldCount = cntRef

12981278 , workerLatencyStart = lat

@@ -1309,10 +1289,11 @@ pushWorker yieldMax sv = do

13091289-- using a CAS based modification.

13101290{-# NOINLINE pushWorkerPar #-}

13111291pushWorkerPar :: MonadAsync m => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()

1312−pushWorkerPar sv wloop = do

1292+pushWorkerPar sv wloop =

13131293 -- We do not use workerCount in case of ParallelVar but still there is no

13141294 -- harm in maintaining it correctly.

13151295#ifdef DIAGNOSTICS

1296+ do

13161297 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1

13171298 recordMaxWorkers sv

13181299 -- This allocation matters when significant number of workers are being

@@ -1347,14 +1328,14 @@ dispatchWorker yieldCount sv = do

13471328 -- Note, "done" may not mean that the work is actually finished if there

13481329 -- are workers active, because there may be a worker which has not yet

13491330 -- queued the leftover work.

1350− if (not done)

1331+ if not done

13511332 then do

13521333 qDone <- liftIO $ isQueueDone sv

13531334 -- Note that the worker count is only decremented during event

13541335 -- processing in fromStreamVar and therefore it is safe to read and

13551336 -- use it without a lock.

13561337 active <- liftIO $ readIORef $ workerCount sv

1357− if (not qDone)

1338+ if not qDone

13581339 then do

13591340 -- Note that we may deadlock if the previous workers (tasks in the

13601341 -- stream) wait/depend on the future workers (tasks in the stream)

@@ -1408,7 +1389,7 @@ rateRecoveryTime :: NanoSecs

14081389rateRecoveryTime = 1000000

14091390

14101391nanoToMicroSecs :: NanoSecs -> Int

1411−nanoToMicroSecs s = (fromIntegral s) `div` 1000

1392+nanoToMicroSecs s = fromIntegral s `div` 1000

14121393

14131394-- We either block, or send one worker with limited yield count or one or more

14141395-- workers with unlimited yield count.

@@ -1493,7 +1474,7 @@ estimateWorkers workerLimit svarYields gainLossYields

14931474 in assert (sleepTime >= 0) $

14941475 -- if s is less than 0 it means our maxSleepTime is less

14951476 -- than the worker latency.

1496− if (s > 0) then BlockWait s else ManyWorkers 1 (Count 0)

1477+ if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)

14971478 where

14981479 withLimit n =

14991480 case workerLimit of

@@ -1519,7 +1500,7 @@ getWorkerLatency yinfo = do

15191500 pendingTime = colTime + time

15201501 new =

15211502 if pendingCount > 0

1522− then let lat = pendingTime `div` (fromIntegral pendingCount)

1503+ then let lat = pendingTime `div` fromIntegral pendingCount

15231504 -- XXX Give more weight to new?

15241505 in (lat + prev) `div` 2

15251506 else prev

@@ -1583,9 +1564,9 @@ collectLatency _ss yinfo = do

15831564 lcount' = lcount + pendingCount

15841565 tripleWith lat = (lcount', ltime, lat)

15851566

1586− if (pendingCount > 0)

1567+ if pendingCount > 0

15871568 then do

1588− let new = pendingTime `div` (fromIntegral pendingCount)

1569+ let new = pendingTime `div` fromIntegral pendingCount

15891570#ifdef DIAGNOSTICS

15901571 minLat <- readIORef (minWorkerLatency _ss)

15911572 when (new < minLat || minLat == 0) $

@@ -1599,7 +1580,7 @@ collectLatency _ss yinfo = do

15991580 -- return the previous latency derived from the previous batch.

16001581 if (pendingCount > fromIntegral magicMaxBuffer)

16011582 || (pendingTime > minThreadDelay)

1602− || (let r = (fromIntegral new) / (fromIntegral prev) :: Double

1583+ || (let r = fromIntegral new / fromIntegral prev :: Double

16031584 in prev > 0 && (r > 2 || r < 0.5))

16041585 || (prev == 0)

16051586 then do

@@ -1698,18 +1679,17 @@ dispatchWorkerPaced sv = do

16981679 liftIO $ writeIORef periodRef period

16991680

17001681 cnt <- liftIO $ readIORef $ workerCount sv

1701− if (cnt < netWorkers)

1682+ if cnt < netWorkers

17021683 then do

17031684 let total = netWorkers - cnt

17041685 batch = max 1 $ fromIntegral $

17051686 minThreadDelay `div` targetLat

1706− r <- dispatchN (min total batch)

17071687 -- XXX stagger the workers over a period?

17081688 -- XXX cannot sleep, as that would mean we cannot process the

17091689 -- outputs. need to try a different mechanism to stagger.

17101690 -- when (total > batch) $

17111691 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay

1712− return r

1692+ dispatchN (min total batch)

17131693 else return False

17141694

17151695 where

@@ -1723,7 +1703,7 @@ dispatchWorkerPaced sv = do

17231703 else yields + buf

17241704 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)

17251705

1726− dispatchN n = do

1706+ dispatchN n =

17271707 if n == 0

17281708 then return True

17291709 else do

@@ -1736,12 +1716,13 @@ sendWorkerDelayPaced :: SVar t m a -> IO ()

17361716sendWorkerDelayPaced _ = return ()

17371717

17381718sendWorkerDelay :: SVar t m a -> IO ()

1739−sendWorkerDelay _sv = do

1719+sendWorkerDelay _sv =

17401720 -- XXX we need a better way to handle this than hardcoded delays. The

17411721 -- delays may be different for different systems.

17421722 -- If there is a usecase where this is required we can create a combinator

17431723 -- to set it as a config in the state.

17441724 {-

1725+ do

17451726 ncpu <- getNumCapabilities

17461727 if ncpu <= 1

17471728 then

@@ -1816,7 +1797,7 @@ sendWorkerWait delay dispatch sv = do

18161797 -- doorbell on the next enqueue.

18171798

18181799 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True

1819− liftIO $ storeLoadBarrier

1800+ liftIO storeLoadBarrier

18201801 canDoMore <- dispatch sv

18211802

18221803 -- XXX test for the case when we miss sending a worker when the worker

@@ -1865,12 +1846,12 @@ readOutputQBounded sv = do

18651846 cnt <- liftIO $ readIORef $ workerCount sv

18661847 when (cnt <= 0) $ do

18671848 done <- liftIO $ isWorkDone sv

1868− when (not done) $ pushWorker 0 sv

1849+ when (not done) (pushWorker 0 sv)

18691850

18701851 {-# INLINE blockingRead #-}

18711852 blockingRead = do

18721853 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv

1873− liftIO $ (readOutputQRaw sv >>= return . fst)

1854+ liftIO (fst `fmap` readOutputQRaw sv)

18741855

18751856readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]

18761857readOutputQPaced sv = do

@@ -1888,7 +1869,7 @@ readOutputQPaced sv = do

18881869 {-# INLINE blockingRead #-}

18891870 blockingRead = do

18901871 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv

1891− liftIO $ (readOutputQRaw sv >>= return . fst)

1872+ liftIO (fst `fmap` readOutputQRaw sv)

18921873

18931874postProcessBounded :: MonadAsync m => SVar t m a -> m Bool

18941875postProcessBounded sv = do

@@ -1907,7 +1888,7 @@ postProcessBounded sv = do

19071888 r <- liftIO $ isWorkDone sv

19081889 -- Note that we need to guarantee a worker, therefore we cannot just

19091890 -- use dispatchWorker which may or may not send a worker.

1910− when (not r) $ pushWorker 0 sv

1891+ when (not r) (pushWorker 0 sv)

19111892 -- XXX do we need to dispatch many here?

19121893 -- void $ dispatchWorker sv

19131894 return r

@@ -2150,7 +2131,7 @@ getParallelSVar st = do

21502131 case yieldRateInfo sv of

21512132 Nothing -> return ()

21522133 Just yinfo -> void $ collectLatency (svarStats sv) yinfo

2153− readOutputQRaw sv >>= return . fst

2134+ fst `fmap` readOutputQRaw sv

21542135

21552136sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)

21562137sendFirstWorker sv m = do

@@ -2160,7 +2141,7 @@ sendFirstWorker sv m = do

21602141 liftIO $ enqueue sv m

21612142 case yieldRateInfo sv of

21622143 Nothing -> pushWorker 0 sv

2163− Just yinfo -> do

2144+ Just yinfo ->

21642145 if svarLatencyTarget yinfo == maxBound

21652146 then liftIO $ threadDelay maxBound

21662147 else pushWorker 1 sv

@@ -2190,7 +2171,7 @@ newParallelVar st = liftIO $ getParallelSVar st

21902171-- be read back from the SVar using 'fromSVar'.

21912172toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()

21922173toStreamVar sv m = do

2193− liftIO $ (enqueue sv) m

2174+ liftIO $ enqueue sv m

21942175 done <- allThreadsDone sv

21952176 -- XXX This is safe only when called from the consumer thread or when no

21962177 -- consumer is present. There may be a race if we are not running in the

src/Streamly/Streams/Ahead.hs+11−13

@@ -148,15 +148,13 @@ underMaxHeap sv hp = do

148148 -- XXX simplify this

149149 let maxHeap = case maxBufferLimit sv of

150150 Limited lim -> Limited $

151− if (fromIntegral lim) >= len

152− then lim - (fromIntegral len)

153− else 0

151+ max 0 (lim - fromIntegral len)

154152 Unlimited -> Unlimited

155153

156154 case maxHeap of

157155 Limited lim -> do

158156 active <- readIORef (workerCount sv)

159− return $ H.size hp + active <= (fromIntegral lim)

157+ return $ H.size hp + active <= fromIntegral lim

160158 Unlimited -> return True

161159

162160-- Return value:

@@ -166,7 +164,7 @@ preStopCheck ::

166164 SVar Stream m a

167165 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Maybe Int)

168166 -> IO Bool

169−preStopCheck sv heap = do

167+preStopCheck sv heap =

170168 -- check the stop condition under a lock before actually

171169 -- stopping so that the whole herd does not stop at once.

172170 withIORef heap $ \(hp, _) -> do

@@ -230,7 +228,7 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

230228 sendStop sv winfo

231229 else runStreamWithYieldLimit True seqNo r

232230

233− loopHeap seqNo ent = do

231+ loopHeap seqNo ent =

234232 case ent of

235233 AheadEntryPure a -> do

236234 -- Use 'send' directly so that we do not account this in worker

@@ -239,7 +237,7 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

239237 -- transferring available results from heap to outputQueue.

240238 void $ liftIO $ send sv (ChildYield a)

241239 nextHeap seqNo

242− AheadEntryStream r -> do

240+ AheadEntryStream r ->

243241 if stopping

244242 then stopIfNeeded ent seqNo r

245243 else runStreamWithYieldLimit True seqNo r

@@ -249,7 +247,7 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

249247 case res of

250248 Ready (Entry seqNo hent) -> loopHeap seqNo hent

251249 Clearing -> liftIO $ sendStop sv winfo

252− _ -> do

250+ _ ->

253251 if stopping

254252 then do

255253 r <- liftIO $ preStopCheck sv heap

@@ -265,7 +263,7 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

265263 Just (m, seqNo) -> do

266264 yieldLimitOk <- liftIO $ decrementYieldLimit sv

267265 if yieldLimitOk

268− then do

266+ then

269267 if seqNo == prevSeqNo + 1

270268 then processWithToken q heap st sv winfo m seqNo

271269 else processWithoutToken q heap st sv winfo m seqNo

@@ -368,7 +366,7 @@ processWithoutToken q heap st sv winfo m sno = do

368366 status <-

369367 case yieldRateInfo sv of

370368 Nothing -> return HContinue

371− Just yinfo -> do

369+ Just yinfo ->

372370 case winfo of

373371 Just info -> do

374372 rateOk <- liftIO $ workerRateControl sv yinfo info

@@ -445,7 +443,7 @@ processWithToken q heap st sv winfo action sno = do

445443 Just (m, seqNo) -> do

446444 yieldLimitOk <- liftIO $ decrementYieldLimit sv

447445 if yieldLimitOk

448− then do

446+ then

449447 if seqNo == prevSeqNo + 1

450448 then do

451449 let stop = do

@@ -525,7 +523,7 @@ workLoopAhead q heap st sv winfo = do

525523 Just (m, seqNo) -> do

526524 yieldLimitOk <- liftIO $ decrementYieldLimit sv

527525 if yieldLimitOk

528− then do

526+ then

529527 if seqNo == 0

530528 then processWithToken q heap st sv winfo m seqNo

531529 else processWithoutToken q heap st sv winfo m seqNo

@@ -560,7 +558,7 @@ forkSVarAhead m1 m2 = Stream $ \st stp sng yld -> do

560558

561559{-# INLINE aheadS #-}

562560aheadS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

563−aheadS m1 m2 = Stream $ \st stp sng yld -> do

561+aheadS m1 m2 = Stream $ \st stp sng yld ->

564562 case streamVar st of

565563 Just sv | svarStyle sv == AheadVar -> do

566564 liftIO $ enqueue sv m2

src/Streamly/Streams/Async.hs+3−3

@@ -469,11 +469,11 @@ newAsyncVar st m = do

469469-- @since 0.2.0

470470{-# INLINABLE mkAsync #-}

471471mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)

472−mkAsync m = newAsyncVar defState (toStream m) >>= return . fromSVar

472+mkAsync m = fmap fromSVar (newAsyncVar defState (toStream m))

473473

474474{-# INLINABLE mkAsync' #-}

475475mkAsync' :: (IsStream t, MonadAsync m) => State Stream m a -> t m a -> m (t m a)

476−mkAsync' st m = newAsyncVar st (toStream m) >>= return . fromSVar

476+mkAsync' st m = fmap fromSVar (newAsyncVar st (toStream m))

477477

478478-- | Create a new SVar and enqueue one stream computation on it.

479479{-# INLINABLE newWAsyncVar #-}

@@ -563,7 +563,7 @@ forkSVarAsync style m1 m2 = Stream $ \st stp sng yld -> do

563563{-# INLINE joinStreamVarAsync #-}

564564joinStreamVarAsync :: MonadAsync m

565565 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

566−joinStreamVarAsync style m1 m2 = Stream $ \st stp sng yld -> do

566+joinStreamVarAsync style m1 m2 = Stream $ \st stp sng yld ->

567567 case streamVar st of

568568 Just sv | svarStyle sv == style ->

569569 liftIO (enqueue sv m2) >> unStream m1 st stp sng yld

src/Streamly/Streams/Parallel.hs+2−2

@@ -91,7 +91,7 @@ forkSVarPar m r = Stream $ \st stp sng yld -> do

9191 sv <- newParallelVar st

9292 pushWorkerPar sv (runOne st{streamVar = Just sv} m)

9393 pushWorkerPar sv (runOne st{streamVar = Just sv} r)

94− (unStream (fromSVar sv)) (rstState st) stp sng yld

94+ unStream (fromSVar sv) (rstState st) stp sng yld

9595

9696{-# INLINE joinStreamVarPar #-}

9797joinStreamVarPar :: MonadAsync m

@@ -119,7 +119,7 @@ consMParallel m r = K.yieldM m `parallelStream` r

119119-- @since 0.2.0

120120{-# INLINE parallel #-}

121121parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

122−parallel m1 m2 = fromStream $ Stream $ \st stp sng yld -> do

122+parallel m1 m2 = fromStream $ Stream $ \st stp sng yld ->

123123 unStream (parallelStream (toStream m1) (toStream m2))

124124 st stp sng yld

125125

src/Streamly/Streams/SVar.hs+10−8

@@ -81,8 +81,9 @@ fromStreamVar sv = Stream $ \st stp sng yld -> do

8181

8282 where

8383

84− allDone stp = do

84+ allDone stp =

8585#ifdef DIAGNOSTICS

86+ do

8687 t <- liftIO $ getTime Monotonic

8788 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)

8889#ifdef DIAGNOSTICS_VERBOSE

@@ -114,7 +115,7 @@ fromStreamVar sv = Stream $ \st stp sng yld -> do

114115

115116{-# INLINE fromSVar #-}

116117fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a

117−fromSVar sv = do

118+fromSVar sv =

118119 fromStream $ Stream $ \st stp sng yld -> do

119120 ref <- liftIO $ newIORef ()

120121 _ <- liftIO $ mkWeakIORef ref hook

@@ -124,8 +125,9 @@ fromSVar sv = do

124125 unStream (fromStreamVar sv{svarRef = Just ref}) st stp sng yld

125126 where

126127

127− hook = do

128+ hook =

128129#ifdef DIAGNOSTICS_VERBOSE

130+ do

129131 printSVar sv "SVar Garbage Collected"

130132#endif

131133 cleanupSVar sv

@@ -154,7 +156,7 @@ toSVar sv m = toStreamVar sv (toStream m)

154156-- @since 0.4.0

155157{-# INLINE_NORMAL maxThreads #-}

156158maxThreads :: IsStream t => Int -> t m a -> t m a

157−maxThreads n m = fromStream $ Stream $ \st stp sng yld -> do

159+maxThreads n m = fromStream $ Stream $ \st stp sng yld ->

158160 unStream (toStream m) (setMaxThreads n st) stp sng yld

159161

160162{-

@@ -179,7 +181,7 @@ maxThreadsSerial _ = id

179181-- @since 0.4.0

180182{-# INLINE_NORMAL maxBuffer #-}

181183maxBuffer :: IsStream t => Int -> t m a -> t m a

182−maxBuffer n m = fromStream $ Stream $ \st stp sng yld -> do

184+maxBuffer n m = fromStream $ Stream $ \st stp sng yld ->

183185 unStream (toStream m) (setMaxBuffer n st) stp sng yld

184186

185187{-

@@ -203,7 +205,7 @@ maxBufferSerial _ = id

203205-- @since 0.5.0

204206{-# INLINE_NORMAL rate #-}

205207rate :: IsStream t => Maybe Rate -> t m a -> t m a

206−rate r m = fromStream $ Stream $ \st stp sng yld -> do

208+rate r m = fromStream $ Stream $ \st stp sng yld ->

207209 case r of

208210 Just (Rate low goal _ _) | goal < low ->

209211 error "rate: Target rate cannot be lower than minimum rate."

@@ -281,7 +283,7 @@ constRate r = rate (Just $ Rate r r r 0)

281283--

282284{-# INLINE_NORMAL _serialLatency #-}

283285_serialLatency :: IsStream t => Int -> t m a -> t m a

284−_serialLatency n m = fromStream $ Stream $ \st stp sng yld -> do

286+_serialLatency n m = fromStream $ Stream $ \st stp sng yld ->

285287 unStream (toStream m) (setStreamLatency n st) stp sng yld

286288

287289{-

@@ -296,7 +298,7 @@ serialLatencySerial _ = id

296298-- inherited by everything in enclosed scope.

297299{-# INLINE_NORMAL maxYields #-}

298300maxYields :: IsStream t => Maybe Int64 -> t m a -> t m a

299−maxYields n m = fromStream $ Stream $ \st stp sng yld -> do

301+maxYields n m = fromStream $ Stream $ \st stp sng yld ->

300302 unStream (toStream m) (setYieldLimit n st) stp sng yld

301303

302304{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}

src/Streamly/Streams/Serial.hs+4−4

@@ -177,7 +177,7 @@ serial m1 m2 = fromStream $ Stream $ \st stp sng yld ->

177177instance Monad m => Monad (SerialT m) where

178178 return = pure

179179 (SerialT (Stream m)) >>= f = SerialT $ Stream $ \st stp sng yld ->

180− let run x = (unStream x) (rstState st) stp sng yld

180+ let run x = unStream x (rstState st) stp sng yld

181181 single a = run $ toStream (f a)

182182 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

183183 in m (rstState st) stp single yieldk

@@ -289,10 +289,10 @@ instance IsStream WSerialT where

289289{-# INLINE interleave #-}

290290interleave :: Stream m a -> Stream m a -> Stream m a

291291interleave m1 m2 = Stream $ \st stp sng yld -> do

292− let stop = (unStream m2) (rstState st) stp sng yld

292+ let stop = unStream m2 (rstState st) stp sng yld

293293 single a = yld a m2

294294 yieldk a r = yld a (interleave m2 r)

295− (unStream m1) (rstState st) stop single yieldk

295+ unStream m1 (rstState st) stop single yieldk

296296

297297-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.

298298-- Interleaves two streams, yielding one element from each stream alternately.

@@ -332,7 +332,7 @@ instance Monoid (WSerialT m a) where

332332instance Monad m => Monad (WSerialT m) where

333333 return = pure

334334 (WSerialT (Stream m)) >>= f = WSerialT $ Stream $ \st stp sng yld ->

335− let run x = (unStream x) (rstState st) stp sng yld

335+ let run x = unStream x (rstState st) stp sng yld

336336 single a = run $ toStream (f a)

337337 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

338338 in m (rstState st) stp single yieldk

src/Streamly/Streams/StreamD.hs+8−8

@@ -182,7 +182,7 @@ uncons (Stream step state) = go state

182182 go st = do

183183 r <- step defState st

184184 return $ case r of

185− Yield x s -> Just (x, (Stream step s))

185+ Yield x s -> Just (x, Stream step s)

186186 Stop -> Nothing

187187

188188------------------------------------------------------------------------------

@@ -217,8 +217,8 @@ enumFromStepN from stride n =

217217 from `seq` stride `seq` n `seq` Stream step (from, n)

218218 where

219219 {-# INLINE_LATE step #-}

220− step _ (x, i) | i > 0 = return $ Yield x (x + stride, i - 1)

221− | otherwise = return $ Stop

220+ step _ (x, i) | i > 0 = return $ Yield x (x + stride, i - 1)

221+ | otherwise = return Stop

222222

223223-------------------------------------------------------------------------------

224224-- Generation by Conversion

@@ -246,7 +246,7 @@ yieldM m = Stream step True

246246-- | Convert a list of monadic actions to a 'Stream'

247247{-# INLINE_LATE fromListM #-}

248248fromListM :: MonadAsync m => [m a] -> Stream m a

249−fromListM zs = Stream step zs

249+fromListM = Stream step

250250 where

251251 {-# INLINE_LATE step #-}

252252 step _ (m:ms) = m >>= \x -> return $ Yield x ms

@@ -255,7 +255,7 @@ fromListM zs = Stream step zs

255255-- | Convert a list of pure values to a 'Stream'

256256{-# INLINE_LATE fromList #-}

257257fromList :: Monad m => [a] -> Stream m a

258−fromList zs = Stream step zs

258+fromList = Stream step

259259 where

260260 {-# INLINE_LATE step #-}

261261 step _ (x:xs) = return $ Yield x xs

@@ -264,7 +264,7 @@ fromList zs = Stream step zs

264264-- XXX pass the state to streamD

265265{-# INLINE_LATE fromStreamK #-}

266266fromStreamK :: Monad m => K.Stream m a -> Stream m a

267−fromStreamK m = Stream step m

267+fromStreamK = Stream step

268268 where

269269 step gst m1 =

270270 let stop = return Stop

@@ -530,7 +530,7 @@ takeWhileM f (Stream step state) = Stream step' state

530530 Yield x s -> do

531531 b <- f x

532532 return $ if b then Yield x s else Stop

533− Stop -> return $ Stop

533+ Stop -> return Stop

534534

535535{-# INLINE takeWhile #-}

536536takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a

@@ -593,7 +593,7 @@ filterM f (Stream step state) = Stream step' state

593593 if b

594594 then return $ Yield x s

595595 else step' gst s

596− Stop -> return $ Stop

596+ Stop -> return Stop

597597

598598{-# INLINE filter #-}

599599filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a

src/Streamly/Streams/StreamK.hs+30−30

@@ -366,7 +366,7 @@ uncons m =

366366 let stop = return Nothing

367367 single a = return (Just (a, nil))

368368 yieldk a r = return (Just (a, fromStream r))

369− in (unStream (toStream m)) defState stop single yieldk

369+ in unStream (toStream m) defState stop single yieldk

370370

371371-------------------------------------------------------------------------------

372372-- Generation

@@ -454,7 +454,7 @@ foldStream

454454 -> m r

455455foldStream st blank single step m =

456456 let yieldk a x = step a (fromStream x)

457− in (unStream (toStream m)) st blank single yieldk

457+ in unStream (toStream m) st blank single yieldk

458458

459459-- | Lazy right associative fold.

460460foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b

@@ -464,7 +464,7 @@ foldr step acc m = go (toStream m)

464464 let stop = return acc

465465 single a = return (step a acc)

466466 yieldk a r = go r >>= \b -> return (step a b)

467− in (unStream m1) defState stop single yieldk

467+ in unStream m1 defState stop single yieldk

468468

469469-- | Lazy right fold with a monadic step function.

470470{-# INLINE foldrM #-}

@@ -475,7 +475,7 @@ foldrM step acc m = go (toStream m)

475475 let stop = return acc

476476 single a = step a acc

477477 yieldk a r = go r >>= step a

478− in (unStream m1) defState stop single yieldk

478+ in unStream m1 defState stop single yieldk

479479

480480{-# INLINE foldr1 #-}

481481foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)

@@ -483,12 +483,12 @@ foldr1 step m = do

483483 r <- uncons m

484484 case r of

485485 Nothing -> return Nothing

486− Just (h, t) -> go h (toStream t) >>= return . Just

486+ Just (h, t) -> fmap Just (go h (toStream t))

487487 where

488488 go p m1 =

489489 let stp = return p

490490 single a = return $ step a p

491− yieldk a r = go a r >>= return . (step p)

491+ yieldk a r = fmap (step p) (go a r)

492492 in unStream m1 defState stp single yieldk

493493

494494-- | Strict left fold with an extraction function. Like the standard strict

@@ -503,7 +503,7 @@ foldx step begin done m = get $ go (toStream m) begin

503503 {-# NOINLINE get #-}

504504 get m1 =

505505 let single = return . done

506− in (unStream m1) undefined undefined single undefined

506+ in unStream m1 undefined undefined single undefined

507507

508508 -- Note, this can be implemented by making a recursive call to "go",

509509 -- however that is more expensive because of unnecessary recursion

@@ -514,13 +514,13 @@ foldx step begin done m = get $ go (toStream m) begin

514514 single a = sng $ step acc a

515515 yieldk a r =

516516 let stream = go r (step acc a)

517− in (unStream stream) defState undefined sng yld

518− in (unStream m1) defState stop single yieldk

517+ in unStream stream defState undefined sng yld

518+ in unStream m1 defState stop single yieldk

519519

520520-- | Strict left associative fold.

521521{-# INLINE foldl' #-}

522522foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b

523−foldl' step begin m = foldx step begin id m

523+foldl' step begin = foldx step begin id

524524

525525-- XXX replace the recursive "go" with explicit continuations.

526526-- | Like 'foldx', but with a monadic step function.

@@ -532,11 +532,11 @@ foldxM step begin done m = go begin (toStream m)

532532 let stop = acc >>= done

533533 single a = acc >>= \b -> step b a >>= done

534534 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r

535− in (unStream m1) defState stop single yieldk

535+ in unStream m1 defState stop single yieldk

536536

537537-- | Like 'foldl'' but with a monadic step function.

538538foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b

539−foldlM' step begin m = foldxM step (return begin) return m

539+foldlM' step begin = foldxM step (return begin) return

540540

541541------------------------------------------------------------------------------

542542-- Specialized folds

@@ -598,7 +598,7 @@ elem e m = go (toStream m)

598598 let stop = return False

599599 single a = return (a == e)

600600 yieldk a r = if a == e then return True else go r

601− in (unStream m1) defState stop single yieldk

601+ in unStream m1 defState stop single yieldk

602602

603603{-# INLINE notElem #-}

604604notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

@@ -608,7 +608,7 @@ notElem e m = go (toStream m)

608608 let stop = return True

609609 single a = return (a /= e)

610610 yieldk a r = if a == e then return False else go r

611− in (unStream m1) defState stop single yieldk

611+ in unStream m1 defState stop single yieldk

612612

613613all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool

614614all p m = go (toStream m)

@@ -725,7 +725,7 @@ mapM_ f m = go (toStream m)

725725 let stop = return ()

726726 single a = void (f a)

727727 yieldk a r = f a >> go r

728− in (unStream m1) defState stop single yieldk

728+ in unStream m1 defState stop single yieldk

729729

730730------------------------------------------------------------------------------

731731-- Converting folds

@@ -757,7 +757,7 @@ scanx step begin done m =

757757

758758{-# INLINE scanl' #-}

759759scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b

760−scanl' step begin m = scanx step begin id m

760+scanl' step begin = scanx step begin id

761761

762762-------------------------------------------------------------------------------

763763-- Filtering

@@ -771,7 +771,7 @@ filter p m = fromStream $ go (toStream m)

771771 let single a | p a = sng a

772772 | otherwise = stp

773773 yieldk a r | p a = yld a (go r)

774− | otherwise = (unStream r) (rstState st) stp single yieldk

774+ | otherwise = unStream r (rstState st) stp single yieldk

775775 in unStream m1 (rstState st) stp single yieldk

776776

777777{-# INLINE take #-}

@@ -814,7 +814,7 @@ dropWhile p m = fromStream $ go (toStream m)

814814 go m1 = Stream $ \st stp sng yld ->

815815 let single a | p a = stp

816816 | otherwise = sng a

817− yieldk a r | p a = (unStream r) (rstState st) stp single yieldk

817+ yieldk a r | p a = unStream r (rstState st) stp single yieldk

818818 | otherwise = yld a r

819819 in unStream m1 (rstState st) stp single yieldk

820820

@@ -837,8 +837,8 @@ mapM f m = go (toStream m)

837837 where

838838 go m1 = fromStream $ Stream $ \st stp sng yld ->

839839 let single a = f a >>= sng

840− yieldk a r = unStream (toStream (f a |: (go r))) st stp sng yld

841− in (unStream m1) (rstState st) stp single yieldk

840+ yieldk a r = unStream (toStream (f a |: go r)) st stp sng yld

841+ in unStream m1 (rstState st) stp single yieldk

842842

843843-- Be careful when modifying this, this uses a consM (|:) deliberately to allow

844844-- other stream types to overload it.

@@ -849,7 +849,7 @@ sequence m = go (toStream m)

849849 go m1 = fromStream $ Stream $ \st stp sng yld ->

850850 let single ma = ma >>= sng

851851 yieldk ma r = unStream (toStream $ ma |: go r) st stp sng yld

852− in (unStream m1) (rstState st) stp single yieldk

852+ in unStream m1 (rstState st) stp single yieldk

853853

854854-------------------------------------------------------------------------------

855855-- Inserting

@@ -881,7 +881,7 @@ mapMaybe f m = go (toStream m)

881881 Nothing -> stp

882882 yieldk a r = case f a of

883883 Just b -> yld b (toStream $ go r)

884− Nothing -> (unStream r) (rstState st) stp single yieldk

884+ Nothing -> unStream r (rstState st) stp single yieldk

885885 in unStream m1 (rstState st) stp single yieldk

886886

887887------------------------------------------------------------------------------

@@ -890,15 +890,15 @@ mapMaybe f m = go (toStream m)

890890

891891{-# INLINE zipWithS #-}

892892zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

893−zipWithS f m1 m2 = go m1 m2

893+zipWithS f = go

894894 where

895895 go mx my = Stream $ \st stp sng yld -> do

896896 let merge a ra =

897897 let single2 b = sng (f a b)

898898 yield2 b rb = yld (f a b) (go ra rb)

899899 in unStream my (rstState st) stp single2 yield2

900− let single1 a = merge a nil

901− yield1 a ra = merge a ra

900+ let single1 a = merge a nil

901+ yield1 = merge

902902 unStream mx (rstState st) stp single1 yield1

903903

904904-- | Zip two streams serially using a pure zipping function.

@@ -920,8 +920,8 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

920920 single2 b = f a b >>= sng

921921 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)

922922 in unStream my (rstState st) stp single2 yield2

923− let single1 a = merge a nil

924− yield1 a ra = merge a ra

923+ let single1 a = merge a nil

924+ yield1 = merge

925925 unStream mx (rstState st) stp single1 yield1

926926

927927------------------------------------------------------------------------------

@@ -935,7 +935,7 @@ serial :: Stream m a -> Stream m a -> Stream m a

935935serial m1 m2 = go m1

936936 where

937937 go (Stream m) = Stream $ \st stp sng yld ->

938− let stop = (unStream m2) (rstState st) stp sng yld

938+ let stop = unStream m2 (rstState st) stp sng yld

939939 single a = yld a m2

940940 yieldk a r = yld a (go r)

941941 in m (rstState st) stop single yieldk

@@ -972,7 +972,7 @@ bindWith par m f = go m

972972 where

973973 go (Stream g) =

974974 Stream $ \st stp sng yld ->

975− let run x = (unStream x) st stp sng yld

975+ let run x = unStream x st stp sng yld

976976 single a = run $ f a

977977 yieldk a r = run $ f a `par` go r

978978 in g (rstState st) stp single yieldk

@@ -995,7 +995,7 @@ withLocal f m =

995995 Stream $ \st stp sng yld ->

996996 let single = local f . sng

997997 yieldk a r = local f $ yld a (withLocal f r)

998− in (unStream m) (rstState st) (local f stp) single yieldk

998+ in unStream m (rstState st) (local f stp) single yieldk

999999

10001000------------------------------------------------------------------------------

10011001-- MonadError