6c9d241 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

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

6c9d241Keith8 years ago

.hlint.yamlnew file+70

compare with master

unifiedsplit

@@ -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)

379-- @since 0.4.0

380{-# INLINE yield #-}

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

382−yield a = K.yield a

382+yield = K.yield

383

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

385-- but more efficient.

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

393-- @since 0.4.0

394{-# INLINE yieldM #-}

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

396−yieldM m = K.yieldM m

396+yieldM = K.yieldM

397

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

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

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

627-- @since 0.1.1

628{-# INLINE null #-}

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

630−null m = K.null m

630+null = K.null

631

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

633--

634-- @since 0.1.0

635{-# INLINE head #-}

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

637−head m = K.head m

637+head = K.head

638

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

640--

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

814 let stop = return ()

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

816 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

818

819------------------------------------------------------------------------------

820-- Transformation by Folding (Scans)

src/Streamly/SVar.hs+65−84

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

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

497setStreamLatency n st =

498 st { _streamLatency =

499− if n < 0

499+ if n <= 0

500 then Nothing

501− else if n == 0

502− then Nothing

503− else Just (fromIntegral n)

501+ else Just (fromIntegral n)

504 }

502 }

505

503

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

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

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

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

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

514cleanupSVar sv = do

512cleanupSVar sv = do

515 workers <- readIORef (workerThreads sv)

513 workers <- readIORef (workerThreads sv)

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

514+ Prelude.mapM_ (`throwTo` ThreadAbort)

517 (S.toList workers)

515 (S.toList workers)

518

516

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

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

520cleanupSVarFromWorker sv = do

518cleanupSVarFromWorker sv = do

521 workers <- readIORef (workerThreads sv)

519 workers <- readIORef (workerThreads sv)

522 self <- myThreadId

520 self <- myThreadId

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

521+ mapM_ (`throwTo` ThreadAbort)

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

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

525

523

526-------------------------------------------------------------------------------

524-------------------------------------------------------------------------------

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

751 then return result

749 then return result

752 else loop tkt (tries - 1)

750 else loop tkt (tries - 1)

753

751

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+

754------------------------------------------------------------------------------

767------------------------------------------------------------------------------

755-- Spawning threads and collecting result in streamed fashion

768-- Spawning threads and collecting result in streamed fashion

756------------------------------------------------------------------------------

769------------------------------------------------------------------------------

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

770{-# INLINE rawForkIO #-}

783{-# INLINE rawForkIO #-}

771rawForkIO :: IO () -> IO ThreadId

784rawForkIO :: IO () -> IO ThreadId

772rawForkIO action = IO $ \ s ->

785rawForkIO 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 #)

774

787

775{-# INLINE doFork #-}

788{-# INLINE doFork #-}

776doFork :: MonadBaseControl IO m

789doFork :: MonadBaseControl IO m

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

857 Unlimited -> return True

870 Unlimited -> return True

858 Limited lim -> do

871 Limited lim -> do

859 active <- readIORef (workerCount sv)

872 active <- readIORef (workerCount sv)

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

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

861

874

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

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

863workerCollectLatency winfo = do

876workerCollectLatency winfo = do

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

865 cnt1 <- readIORef (workerYieldCount winfo)

878 cnt1 <- readIORef (workerYieldCount winfo)

866 let cnt = cnt1 - cnt0

879 let cnt = cnt1 - cnt0

867

880

868− if (cnt > 0)

881+ if cnt > 0

869 then do

882 then do

870 t1 <- getTime Monotonic

883 t1 <- getTime Monotonic

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

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

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

919checkRatePeriodic sv yinfo winfo ycnt = do

932checkRatePeriodic sv yinfo winfo ycnt = do

920 i <- readIORef (workerPollingInterval yinfo)

933 i <- readIORef (workerPollingInterval yinfo)

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

934 -- 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

923 then do

936 then do

924 workerUpdateLatency yinfo winfo

937 workerUpdateLatency yinfo winfo

925 -- XXX not required for parallel streams

938 -- XXX not required for parallel streams

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

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

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

962sendStop sv mwinfo = do

975sendStop sv mwinfo = do

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

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

964− case mwinfo of

965− Just winfo ->

966− case yieldRateInfo sv of

977+ case (mwinfo, yieldRateInfo sv) of

967− Nothing -> return ()

978+ (Just winfo, Just info) ->

968− Just info -> workerStopUpdate winfo info

979+ workerStopUpdate winfo info

980+ _ ->

969− Nothing -> return ()

981+ return ()

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

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

971

983

972-------------------------------------------------------------------------------

984-------------------------------------------------------------------------------

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

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

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

982enqueueLIFO sv q m = do

994enqueueLIFO sv q m = do

983 atomicModifyIORefCAS_ q $ \ms -> m : ms

995 atomicModifyIORefCAS_ q $ \ms -> m : ms

984− storeLoadBarrier

996+ ringDoorBell sv

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) ()

994

997

995-------------------------------------------------------------------------------

998-------------------------------------------------------------------------------

996-- WAsync

999-- WAsync

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

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

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

1005enqueueFIFO sv q m = do

1008enqueueFIFO sv q m = do

1006 pushL q m

1009 pushL q m

1007− storeLoadBarrier

1010+ ringDoorBell sv

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) ()

1017

1011

1018-------------------------------------------------------------------------------

1012-------------------------------------------------------------------------------

1019-- Ahead

1013-- Ahead

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

1085 atomicModifyIORefCAS_ q $ \ case

1079 atomicModifyIORefCAS_ q $ \ case

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

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

1087 _ -> error "not empty"

1081 _ -> error "not empty"

1088− storeLoadBarrier

1082+ ringDoorBell sv

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) ()

1098

1083

1099-- enqueue without incrementing the sequence number

1084-- enqueue without incrementing the sequence number

1100{-# INLINE reEnqueueAhead #-}

1085{-# INLINE reEnqueueAhead #-}

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

1103 atomicModifyIORefCAS_ q $ \ case

1088 atomicModifyIORefCAS_ q $ \ case

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

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

1105 _ -> error "not empty"

1090 _ -> error "not empty"

1106− storeLoadBarrier

1091+ ringDoorBell sv

1107− w <- readIORef $ needDoorBell sv

1108− when w $ do

1109− atomicModifyIORefCAS_ (needDoorBell sv) (const False)

1110− void $ tryPutMVar (outputDoorBell sv) ()

1111

1092

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

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

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

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

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

1138{-# INLINE dequeueAhead #-}

1119{-# INLINE dequeueAhead #-}

1139dequeueAhead :: MonadIO m

1120dequeueAhead :: MonadIO m

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

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

1141−dequeueAhead q = liftIO $ do

1122+dequeueAhead q = liftIO $

1142 atomicModifyIORefCAS q $ \case

1123 atomicModifyIORefCAS q $ \case

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

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

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

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

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

1235{-# INLINE delThread #-}

1216{-# INLINE delThread #-}

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

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

1237delThread sv tid =

1218delThread sv tid =

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

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

1239

1220

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

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

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

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

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

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

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

1246modifyThread sv tid = do

1227modifyThread sv tid = do

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

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

1248− if (S.member tid old)

1229+ if S.member tid old

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

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

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

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

1251− if null changed

1232+ when (null changed) $

1252− then liftIO $ do

1233+ liftIO $ do

1253− writeBarrier

1234+ writeBarrier

1254− void $ tryPutMVar (outputDoorBell sv) ()

1235+ void $ tryPutMVar (outputDoorBell sv) ()

1255− else return ()

1256

1236

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

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

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

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

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

1292 cntRef <- newIORef 0

1272 cntRef <- newIORef 0

1293 t <- getTime Monotonic

1273 t <- getTime Monotonic

1294 lat <- newIORef (0, t)

1274 lat <- newIORef (0, t)

1295− return $ Just $ WorkerInfo

1275+ return $ Just WorkerInfo

1296 { workerYieldMax = yieldMax

1276 { workerYieldMax = yieldMax

1297 , workerYieldCount = cntRef

1277 , workerYieldCount = cntRef

1298 , workerLatencyStart = lat

1278 , workerLatencyStart = lat

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

1309-- using a CAS based modification.

1289-- using a CAS based modification.

1310{-# NOINLINE pushWorkerPar #-}

1290{-# NOINLINE pushWorkerPar #-}

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

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

1312−pushWorkerPar sv wloop = do

1292+pushWorkerPar sv wloop =

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

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

1314 -- harm in maintaining it correctly.

1294 -- harm in maintaining it correctly.

1315#ifdef DIAGNOSTICS

1295#ifdef DIAGNOSTICS

1296+ do

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

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

1317 recordMaxWorkers sv

1298 recordMaxWorkers sv

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

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

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

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

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

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

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

1349 -- queued the leftover work.

1330 -- queued the leftover work.

1350− if (not done)

1331+ if not done

1351 then do

1332 then do

1352 qDone <- liftIO $ isQueueDone sv

1333 qDone <- liftIO $ isQueueDone sv

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

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

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

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

1355 -- use it without a lock.

1336 -- use it without a lock.

1356 active <- liftIO $ readIORef $ workerCount sv

1337 active <- liftIO $ readIORef $ workerCount sv

1357− if (not qDone)

1338+ if not qDone

1358 then do

1339 then do

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

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

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

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

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

1408rateRecoveryTime = 1000000

1389rateRecoveryTime = 1000000

1409

1390

1410nanoToMicroSecs :: NanoSecs -> Int

1391nanoToMicroSecs :: NanoSecs -> Int

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

1392+nanoToMicroSecs s = fromIntegral s `div` 1000

1412

1393

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

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

1414-- workers with unlimited yield count.

1395-- workers with unlimited yield count.

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

1493 in assert (sleepTime >= 0) $

1474 in assert (sleepTime >= 0) $

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

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

1495 -- than the worker latency.

1476 -- 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)

1497 where

1478 where

1498 withLimit n =

1479 withLimit n =

1499 case workerLimit of

1480 case workerLimit of

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

1519 pendingTime = colTime + time

1500 pendingTime = colTime + time

1520 new =

1501 new =

1521 if pendingCount > 0

1502 if pendingCount > 0

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

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

1523 -- XXX Give more weight to new?

1504 -- XXX Give more weight to new?

1524 in (lat + prev) `div` 2

1505 in (lat + prev) `div` 2

1525 else prev

1506 else prev

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

1583 lcount' = lcount + pendingCount

1564 lcount' = lcount + pendingCount

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

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

1585

1566

1586− if (pendingCount > 0)

1567+ if pendingCount > 0

1587 then do

1568 then do

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

1569+ let new = pendingTime `div` fromIntegral pendingCount

1589#ifdef DIAGNOSTICS

1570#ifdef DIAGNOSTICS

1590 minLat <- readIORef (minWorkerLatency _ss)

1571 minLat <- readIORef (minWorkerLatency _ss)

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

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

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

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

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

1600 if (pendingCount > fromIntegral magicMaxBuffer)

1581 if (pendingCount > fromIntegral magicMaxBuffer)

1601 || (pendingTime > minThreadDelay)

1582 || (pendingTime > minThreadDelay)

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

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

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

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

1604 || (prev == 0)

1585 || (prev == 0)

1605 then do

1586 then do

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

1698 liftIO $ writeIORef periodRef period

1679 liftIO $ writeIORef periodRef period

1699

1680

1700 cnt <- liftIO $ readIORef $ workerCount sv

1681 cnt <- liftIO $ readIORef $ workerCount sv

1701− if (cnt < netWorkers)

1682+ if cnt < netWorkers

1702 then do

1683 then do

1703 let total = netWorkers - cnt

1684 let total = netWorkers - cnt

1704 batch = max 1 $ fromIntegral $

1685 batch = max 1 $ fromIntegral $

1705 minThreadDelay `div` targetLat

1686 minThreadDelay `div` targetLat

1706− r <- dispatchN (min total batch)

1707 -- XXX stagger the workers over a period?

1687 -- XXX stagger the workers over a period?

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

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

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

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

1710 -- when (total > batch) $

1690 -- when (total > batch) $

1711 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay

1691 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay

1712− return r

1692+ dispatchN (min total batch)

1713 else return False

1693 else return False

1714

1694

1715 where

1695 where

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

1723 else yields + buf

1703 else yields + buf

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

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

1725

1705

1726− dispatchN n = do

1706+ dispatchN n =

1727 if n == 0

1707 if n == 0

1728 then return True

1708 then return True

1729 else do

1709 else do

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

1736sendWorkerDelayPaced _ = return ()

1716sendWorkerDelayPaced _ = return ()

1737

1717

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

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

1739−sendWorkerDelay _sv = do

1719+sendWorkerDelay _sv =

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

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

1741 -- delays may be different for different systems.

1721 -- delays may be different for different systems.

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

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

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

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

1744 {-

1724 {-

1725+ do

1745 ncpu <- getNumCapabilities

1726 ncpu <- getNumCapabilities

1746 if ncpu <= 1

1727 if ncpu <= 1

1747 then

1728 then

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

1816 -- doorbell on the next enqueue.

1797 -- doorbell on the next enqueue.

1817

1798

1818 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True

1799 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True

1819− liftIO $ storeLoadBarrier

1800+ liftIO storeLoadBarrier

1820 canDoMore <- dispatch sv

1801 canDoMore <- dispatch sv

1821

1802

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

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

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

1865 cnt <- liftIO $ readIORef $ workerCount sv

1846 cnt <- liftIO $ readIORef $ workerCount sv

1866 when (cnt <= 0) $ do

1847 when (cnt <= 0) $ do

1867 done <- liftIO $ isWorkDone sv

1848 done <- liftIO $ isWorkDone sv

1868− when (not done) $ pushWorker 0 sv

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

1869

1850

1870 {-# INLINE blockingRead #-}

1851 {-# INLINE blockingRead #-}

1871 blockingRead = do

1852 blockingRead = do

1872 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv

1853 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv

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

1854+ liftIO (fst `fmap` readOutputQRaw sv)

1874

1855

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

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

1876readOutputQPaced sv = do

1857readOutputQPaced sv = do

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

1888 {-# INLINE blockingRead #-}

1869 {-# INLINE blockingRead #-}

1889 blockingRead = do

1870 blockingRead = do

1890 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv

1871 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv

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

1872+ liftIO (fst `fmap` readOutputQRaw sv)

1892

1873

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

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

1894postProcessBounded sv = do

1875postProcessBounded sv = do

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

1907 r <- liftIO $ isWorkDone sv

1888 r <- liftIO $ isWorkDone sv

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

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

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

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

1910− when (not r) $ pushWorker 0 sv

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

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

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

1912 -- void $ dispatchWorker sv

1893 -- void $ dispatchWorker sv

1913 return r

1894 return r

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

2150 case yieldRateInfo sv of

2131 case yieldRateInfo sv of

2151 Nothing -> return ()

2132 Nothing -> return ()

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

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

2153− readOutputQRaw sv >>= return . fst

2134+ fst `fmap` readOutputQRaw sv

2154

2135

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

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

2156sendFirstWorker sv m = do

2137sendFirstWorker sv m = do

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

2160 liftIO $ enqueue sv m

2141 liftIO $ enqueue sv m

2161 case yieldRateInfo sv of

2142 case yieldRateInfo sv of

2162 Nothing -> pushWorker 0 sv

2143 Nothing -> pushWorker 0 sv

2163− Just yinfo -> do

2144+ Just yinfo ->

2164 if svarLatencyTarget yinfo == maxBound

2145 if svarLatencyTarget yinfo == maxBound

2165 then liftIO $ threadDelay maxBound

2146 then liftIO $ threadDelay maxBound

2166 else pushWorker 1 sv

2147 else pushWorker 1 sv

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

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

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

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

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

2192toStreamVar sv m = do

2173toStreamVar sv m = do

2193− liftIO $ (enqueue sv) m

2174+ liftIO $ enqueue sv m

2194 done <- allThreadsDone sv

2175 done <- allThreadsDone sv

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

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

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

2177 -- 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

148 -- XXX simplify this

149 let maxHeap = case maxBufferLimit sv of

150 Limited lim -> Limited $

151− if (fromIntegral lim) >= len

151+ max 0 (lim - fromIntegral len)

152− then lim - (fromIntegral len)

153− else 0

154 Unlimited -> Unlimited

152 Unlimited -> Unlimited

155

153

156 case maxHeap of

154 case maxHeap of

157 Limited lim -> do

155 Limited lim -> do

158 active <- readIORef (workerCount sv)

156 active <- readIORef (workerCount sv)

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

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

160 Unlimited -> return True

158 Unlimited -> return True

161

159

162-- Return value:

160-- Return value:

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

166 SVar Stream m a

164 SVar Stream m a

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

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

168 -> IO Bool

166 -> IO Bool

169−preStopCheck sv heap = do

167+preStopCheck sv heap =

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

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

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

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

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

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

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

230 sendStop sv winfo

228 sendStop sv winfo

231 else runStreamWithYieldLimit True seqNo r

229 else runStreamWithYieldLimit True seqNo r

232

230

233− loopHeap seqNo ent = do

231+ loopHeap seqNo ent =

234 case ent of

232 case ent of

235 AheadEntryPure a -> do

233 AheadEntryPure a -> do

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

234 -- 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

239 -- transferring available results from heap to outputQueue.

237 -- transferring available results from heap to outputQueue.

240 void $ liftIO $ send sv (ChildYield a)

238 void $ liftIO $ send sv (ChildYield a)

241 nextHeap seqNo

239 nextHeap seqNo

242− AheadEntryStream r -> do

240+ AheadEntryStream r ->

243 if stopping

241 if stopping

244 then stopIfNeeded ent seqNo r

242 then stopIfNeeded ent seqNo r

245 else runStreamWithYieldLimit True seqNo r

243 else runStreamWithYieldLimit True seqNo r

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

249 case res of

247 case res of

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

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

251 Clearing -> liftIO $ sendStop sv winfo

249 Clearing -> liftIO $ sendStop sv winfo

252− _ -> do

250+ _ ->

253 if stopping

251 if stopping

254 then do

252 then do

255 r <- liftIO $ preStopCheck sv heap

253 r <- liftIO $ preStopCheck sv heap

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

265 Just (m, seqNo) -> do

263 Just (m, seqNo) -> do

266 yieldLimitOk <- liftIO $ decrementYieldLimit sv

264 yieldLimitOk <- liftIO $ decrementYieldLimit sv

267 if yieldLimitOk

265 if yieldLimitOk

268− then do

266+ then

269 if seqNo == prevSeqNo + 1

267 if seqNo == prevSeqNo + 1

270 then processWithToken q heap st sv winfo m seqNo

268 then processWithToken q heap st sv winfo m seqNo

271 else processWithoutToken q heap st sv winfo m seqNo

269 else processWithoutToken q heap st sv winfo m seqNo

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

368 status <-

366 status <-

369 case yieldRateInfo sv of

367 case yieldRateInfo sv of

370 Nothing -> return HContinue

368 Nothing -> return HContinue

371− Just yinfo -> do

369+ Just yinfo ->

372 case winfo of

370 case winfo of

373 Just info -> do

371 Just info -> do

374 rateOk <- liftIO $ workerRateControl sv yinfo info

372 rateOk <- liftIO $ workerRateControl sv yinfo info

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

445 Just (m, seqNo) -> do

443 Just (m, seqNo) -> do

446 yieldLimitOk <- liftIO $ decrementYieldLimit sv

444 yieldLimitOk <- liftIO $ decrementYieldLimit sv

447 if yieldLimitOk

445 if yieldLimitOk

448− then do

446+ then

449 if seqNo == prevSeqNo + 1

447 if seqNo == prevSeqNo + 1

450 then do

448 then do

451 let stop = do

449 let stop = do

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

525 Just (m, seqNo) -> do

523 Just (m, seqNo) -> do

526 yieldLimitOk <- liftIO $ decrementYieldLimit sv

524 yieldLimitOk <- liftIO $ decrementYieldLimit sv

527 if yieldLimitOk

525 if yieldLimitOk

528− then do

526+ then

529 if seqNo == 0

527 if seqNo == 0

530 then processWithToken q heap st sv winfo m seqNo

528 then processWithToken q heap st sv winfo m seqNo

531 else processWithoutToken q heap st sv winfo m seqNo

529 else processWithoutToken q heap st sv winfo m seqNo

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

560

558

561{-# INLINE aheadS #-}

559{-# INLINE aheadS #-}

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

560aheadS :: 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 ->

564 case streamVar st of

562 case streamVar st of

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

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

566 liftIO $ enqueue sv m2

564 liftIO $ enqueue sv m2

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

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

469-- @since 0.2.0

470{-# INLINABLE mkAsync #-}

471mkAsync :: (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))

473

474{-# INLINABLE mkAsync' #-}

475mkAsync' :: (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))

477

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

479{-# INLINABLE newWAsyncVar #-}

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

563{-# INLINE joinStreamVarAsync #-}

564joinStreamVarAsync :: MonadAsync m

565 => 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 ->

567 case streamVar st of

568 Just sv | svarStyle sv == style ->

569 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

91 sv <- newParallelVar st

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

93 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

96{-# INLINE joinStreamVarPar #-}

97joinStreamVarPar :: MonadAsync m

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

119-- @since 0.2.0

120{-# INLINE parallel #-}

121parallel :: (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 ->

123 unStream (parallelStream (toStream m1) (toStream m2))

124 st stp sng yld

125

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

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

82 where

84− allDone stp = do

84+ allDone stp =

85#ifdef DIAGNOSTICS

86+ do

86 t <- liftIO $ getTime Monotonic

87 t <- liftIO $ getTime Monotonic

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

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

88#ifdef DIAGNOSTICS_VERBOSE

89#ifdef DIAGNOSTICS_VERBOSE

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

114

115

115{-# INLINE fromSVar #-}

116{-# INLINE fromSVar #-}

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

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

117−fromSVar sv = do

118+fromSVar sv =

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

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

119 ref <- liftIO $ newIORef ()

120 ref <- liftIO $ newIORef ()

120 _ <- liftIO $ mkWeakIORef ref hook

121 _ <- liftIO $ mkWeakIORef ref hook

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

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

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

125 where

126 where

126

127

127− hook = do

128+ hook =

128#ifdef DIAGNOSTICS_VERBOSE

129#ifdef DIAGNOSTICS_VERBOSE

130+ do

129 printSVar sv "SVar Garbage Collected"

131 printSVar sv "SVar Garbage Collected"

130#endif

132#endif

131 cleanupSVar sv

133 cleanupSVar sv

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

154-- @since 0.4.0

156-- @since 0.4.0

155{-# INLINE_NORMAL maxThreads #-}

157{-# INLINE_NORMAL maxThreads #-}

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

158maxThreads :: 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 ->

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

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

159

161

160{-

162{-

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

179-- @since 0.4.0

181-- @since 0.4.0

180{-# INLINE_NORMAL maxBuffer #-}

182{-# INLINE_NORMAL maxBuffer #-}

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

183maxBuffer :: 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 ->

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

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

184

186

185{-

187{-

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

203-- @since 0.5.0

205-- @since 0.5.0

204{-# INLINE_NORMAL rate #-}

206{-# INLINE_NORMAL rate #-}

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

207rate :: 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 ->

207 case r of

209 case r of

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

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

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

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

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

281--

283--

282{-# INLINE_NORMAL _serialLatency #-}

284{-# INLINE_NORMAL _serialLatency #-}

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

285_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 ->

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

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

286

288

287{-

289{-

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

296-- inherited by everything in enclosed scope.

298-- inherited by everything in enclosed scope.

297{-# INLINE_NORMAL maxYields #-}

299{-# INLINE_NORMAL maxYields #-}

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

300maxYields :: 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 ->

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

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

301

303

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

304{-# 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 ->

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

178 return = pure

179 (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

181 single a = run $ toStream (f a)

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

183 in m (rstState st) stp single yieldk

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

289{-# INLINE interleave #-}

290interleave :: Stream m a -> Stream m a -> Stream m a

291interleave 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

293 single a = yld a m2

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

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

295+ unStream m1 (rstState st) stop single yieldk

296

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

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

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

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

333 return = pure

334 (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

336 single a = run $ toStream (f a)

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

338 in m (rstState st) stp single yieldk

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

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

182 go st = do

183 r <- step defState st

184 return $ case r of

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

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

186 Stop -> Nothing

187

188------------------------------------------------------------------------------

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

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

218 where

219 {-# INLINE_LATE step #-}

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

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

221− | otherwise = return $ Stop

221+ | otherwise = return Stop

222

223-------------------------------------------------------------------------------

224-- Generation by Conversion

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

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

247{-# INLINE_LATE fromListM #-}

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

249−fromListM zs = Stream step zs

249+fromListM = Stream step

250 where

251 {-# INLINE_LATE step #-}

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

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

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

256{-# INLINE_LATE fromList #-}

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

258−fromList zs = Stream step zs

258+fromList = Stream step

259 where

260 {-# INLINE_LATE step #-}

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

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

264-- XXX pass the state to streamD

265{-# INLINE_LATE fromStreamK #-}

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

267−fromStreamK m = Stream step m

267+fromStreamK = Stream step

268 where

269 step gst m1 =

270 let stop = return Stop

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

530 Yield x s -> do

531 b <- f x

532 return $ if b then Yield x s else Stop

533− Stop -> return $ Stop

533+ Stop -> return Stop

534

535{-# INLINE takeWhile #-}

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

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

593 if b

594 then return $ Yield x s

595 else step' gst s

596− Stop -> return $ Stop

596+ Stop -> return Stop

597

598{-# INLINE filter #-}

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

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

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

366 let stop = return Nothing

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

368 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

370

371-------------------------------------------------------------------------------

372-- Generation

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

454 -> m r

455foldStream st blank single step m =

456 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

458

459-- | Lazy right associative fold.

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

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

464 let stop = return acc

465 single a = return (step a acc)

466 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

468

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

470{-# INLINE foldrM #-}

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

475 let stop = return acc

476 single a = step a acc

477 yieldk a r = go r >>= step a

478− in (unStream m1) defState stop single yieldk

478+ in unStream m1 defState stop single yieldk

479

480{-# INLINE foldr1 #-}

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

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

483 r <- uncons m

484 case r of

485 Nothing -> return Nothing

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

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

487 where

488 go p m1 =

489 let stp = return p

490 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)

492 in unStream m1 defState stp single yieldk

493

494-- | 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

503 {-# NOINLINE get #-}

504 get m1 =

505 let single = return . done

506− in (unStream m1) undefined undefined single undefined

506+ in unStream m1 undefined undefined single undefined

507

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

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

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

514 single a = sng $ step acc a

515 yieldk a r =

516 let stream = go r (step acc a)

517− in (unStream stream) defState undefined sng yld

517+ in unStream stream defState undefined sng yld

518− in (unStream m1) defState stop single yieldk

518+ in unStream m1 defState stop single yieldk

519

520-- | Strict left associative fold.

521{-# INLINE foldl' #-}

522foldl' :: (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

524

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

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

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

532 let stop = acc >>= done

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

534 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

536

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

538foldlM' :: (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

540

541------------------------------------------------------------------------------

542-- Specialized folds

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

598 let stop = return False

599 single a = return (a == e)

600 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

602

603{-# INLINE notElem #-}

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

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

608 let stop = return True

609 single a = return (a /= e)

610 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

612

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

614all p m = go (toStream m)

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

725 let stop = return ()

726 single a = void (f a)

727 yieldk a r = f a >> go r

728− in (unStream m1) defState stop single yieldk

728+ in unStream m1 defState stop single yieldk

729

730------------------------------------------------------------------------------

731-- Converting folds

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

757

758{-# INLINE scanl' #-}

759scanl' :: 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

761

762-------------------------------------------------------------------------------

763-- Filtering

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

771 let single a | p a = sng a

772 | otherwise = stp

773 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

775 in unStream m1 (rstState st) stp single yieldk

776

777{-# INLINE take #-}

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

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

815 let single a | p a = stp

816 | 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

818 | otherwise = yld a r

819 in unStream m1 (rstState st) stp single yieldk

820

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

837 where

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

839 let single a = f a >>= sng

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

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

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

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

842

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

844-- other stream types to overload it.

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

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

850 let single ma = ma >>= sng

851 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

853

854-------------------------------------------------------------------------------

855-- Inserting

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

881 Nothing -> stp

882 yieldk a r = case f a of

883 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

885 in unStream m1 (rstState st) stp single yieldk

886

887------------------------------------------------------------------------------

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

890

891{-# INLINE zipWithS #-}

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

893−zipWithS f m1 m2 = go m1 m2

893+zipWithS f = go

894 where

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

896 let merge a ra =

897 let single2 b = sng (f a b)

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

899 in unStream my (rstState st) stp single2 yield2

900− let single1 a = merge a nil

900+ let single1 a = merge a nil

901− yield1 a ra = merge a ra

901+ yield1 = merge

902 unStream mx (rstState st) stp single1 yield1

903

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

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

920 single2 b = f a b >>= sng

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

922 in unStream my (rstState st) stp single2 yield2

923− let single1 a = merge a nil

923+ let single1 a = merge a nil

924− yield1 a ra = merge a ra

924+ yield1 = merge

925 unStream mx (rstState st) stp single1 yield1

926

927------------------------------------------------------------------------------

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

935serial m1 m2 = go m1

936 where

937 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

939 single a = yld a m2

940 yieldk a r = yld a (go r)

941 in m (rstState st) stop single yieldk

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

972 where

973 go (Stream g) =

974 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

976 single a = run $ f a

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

978 in g (rstState st) stp single yieldk

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

995 Stream $ \st stp sng yld ->

996 let single = local f . sng

997 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

999

1000------------------------------------------------------------------------------

1001-- MonadError