streamly 0.1.0 → 0.1.1
raw patch · 11 files changed
+460/−182 lines, 11 filesdep ~exceptionsdep ~http-conduit
Dependency ranges changed: exceptions, http-conduit
Files
- Changelog.md +11/−0
- README.md +37/−24
- src/Streamly/Core.hs +85/−30
- src/Streamly/Prelude.hs +137/−62
- src/Streamly/Streams.hs +49/−8
- src/Streamly/Time.hs +1/−1
- src/Streamly/Tutorial.hs +3/−2
- stack-8.0.yaml +0/−17
- stack.yaml +1/−3
- streamly.cabal +7/−8
- test/Main.hs +129/−27
Changelog.md view
@@ -1,3 +1,14 @@+## 0.1.1++### Enhancements+* Make `cons` right associative and provide an operator form `.:` for it+* Add `null`, `tail`, `reverse`, `replicateM`, `scan` stream operations+* Improve performance of some stream operations (`foldl`, `dropWhile`)++### Bug Fixes+* Fix the `product` operation. Earlier, it always returned 0 due to a bug+* Fix the `last` operation, which returned `Nothing` for singleton streams+ ## 0.1.0 * Initial release
README.md view
@@ -1,10 +1,5 @@ # Streamly -[](https://gitter.im/composewell/streamly)-[](https://travis-ci.org/composewell/streamly)-[](https://ci.appveyor.com/project/harendra-kumar/streamly)-[](https://coveralls.io/github/composewell/streamly?branch=master)- ## Stream`ing` `Concurrent`ly Streamly is a monad transformer unifying non-determinism@@ -23,14 +18,23 @@ the comprehensive tutorial module `Streamly.Tutorial` first. Also see `Streamly.Examples` for some working examples. +`Streamly` has best in class performance even though it generalizes streaming+to concurrent composition that does not mean it sacrifices non-concurrent+performance. See+[streaming-benchmarks](https://github.com/composewell/streaming-benchmarks) for+detailed performance comparison with regular streaming libraries.+ ## Non-determinism The monad instance composes like a list monad. ``` haskell-loops = $ do- x <- each [1,2]- y <- each [3,4]+import Streamly+import qualified Streamly.Prelude as S++loops = do+ x <- S.each [1,2]+ y <- S.each [3,4] liftIO $ putStrLn $ show (x, y) main = runStreaming $ serially $ loops@@ -68,13 +72,17 @@ concurrently sum the square roots of all combinations: ``` haskell+import Streamly+import qualified Streamly.Prelude as S+ main = do- print $ sum $ asyncly $ do- -- Squaring is concurrent (<|)- x2 <- forEachWith (<|) [1..100] $ \x -> return $ x * x- y2 <- forEachWith (<|) [1..100] $ \y -> return $ y * y- -- sqrt is concurrent (asyncly)- return $ sqrt (x2 + y2)+ s <- S.sum $ asyncly $ do+ -- Squaring is concurrent (<|)+ x2 <- forEachWith (<|) [1..100] $ \x -> return $ x * x+ y2 <- forEachWith (<|) [1..100] $ \y -> return $ y * y+ -- sqrt is concurrent (asyncly)+ return $ sqrt (x2 + y2)+ print s ``` Of course, the actions running in parallel could be arbitrary IO actions. To@@ -116,7 +124,7 @@ ```haskell import Streamly-import Streamly.Prelude as S+import qualified Streamly.Prelude as S import Data.Function ((&)) main = S.each [1..10]@@ -143,11 +151,13 @@ Streams can be combined together in multiple ways: ```haskell-return 1 <> return 2 -- serial, combine atoms-S.each [1..10] <> S.each [11..20] -- serial-S.each [1..10] <| S.each [11..20] -- demand driven parallel-S.each [1..10] <=> S.each [11..20] -- serial but interleaved-S.each [1..10] <|> S.each [11..20] -- fully parallel+main = do+ let p s = (toList . serially) s >>= print+ p $ return 1 <> return 2 -- serial, combine atoms+ p $ S.each [1..10] <> S.each [11..20] -- serial+ p $ S.each [1..10] <| S.each [11..20] -- demand driven parallel+ p $ S.each [1..10] <=> S.each [11..20] -- serial but interleaved+ p $ S.each [1..10] <|> S.each [11..20] -- fully parallel ``` As we have already seen streams can be combined using monadic composition in a@@ -163,10 +173,13 @@ ## Contributing -The code is available under BSD-3 license [on-github](https://github.com/composewell/streamly). Join the [gitter-chat](https://gitter.im/composewell/streamly) channel for discussions. All-contributions are welcome!+The code is available under BSD-3 license+[on github](https://github.com/composewell/streamly). Join the+[gitter chat](https://gitter.im/composewell/streamly) channel for discussions.+You can find some of the+[todo items on the github wiki](https://github.com/composewell/streamly/wiki/Things-To-Do).+Please ask on the gitter channel or [contact the maintainer directly](mailto:harendra.kumar@gmail.com)+for more details on each item. All contributions are welcome! This library was originally inspired by the `transient` package authored by Alberto G. Corona.
src/Streamly/Core.hs view
@@ -1,6 +1,7 @@ {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE UndecidableInstances #-} -- XXX@@ -89,7 +90,7 @@ ------------------------------------------------------------------------------ -- | Conjunction is used for monadic/product style composition. Disjunction is--- used for fold/sum style composition. We need to distiguish the two types of+-- used for fold/sum style composition. We need to distinguish the two types of -- SVars so that the scheduling of the two is independent. data SVarTag = Conjunction | Disjunction deriving Eq @@ -140,13 +141,65 @@ ------------------------------------------------------------------------------ -- TBD use a functor instead of the bare type a?--- XXX remove the Maybe, use "empty" as the base case --- | Represents a monadic stream of values of type 'a' constructed using--- actions in monad 'm'. Streams can be composed sequentially or in parallel;--- in product style compositions (monadic bind multiplies streams in a ListT--- fashion) or in sum style compositions like 'Semigroup', 'Monoid',--- 'Alternative' or variants of these.+-- | The type 'Stream m a' represents a monadic stream of values of type 'a'+-- constructed using actions in monad 'm'. It uses a stop continuation and a+-- yield continuation. You can consider it a rough equivalent of direct style+-- type:+--+-- data Stream m a = Stop | Yield a (Maybe (Stream m a))+--+-- Our goal is to be able to represent finite as well infinite streams and+-- being able to compose a large number of small streams efficiently. In+-- addition we want to compose streams in parallel, to facilitate that we+-- maintain a local state in an SVar that is shared across and is used for+-- synchronization of the streams being composed.+--+-- Using this type, there are two ways to indicate the end of a stream, one is+-- by calling the stop continuation and the other one is by yielding the last+-- value along with 'Nothing' as the rest of the stream.+--+-- Why do we have this redundancy? Why can't we use (a -> Stream m a -> m r) as+-- the type of the yield continuation and always use the stop continuation to+-- indicate the end of the stream? The reason is that when we compose a large+-- number of short or singleton streams then using the stop continuation+-- becomes expensive, just to know that there is no next element we have to+-- call the continuation, introducing an indirection, it seems when using CPS+-- GHC is not able to optimize this out as efficiently as it can be in direct+-- style because of the function call involved. In direct style it will just be+-- a constructor check and a memory access instead of a function call. So we+-- could use:+--+-- data Stream m a = Stop | Yield a (Stream m a)+--+-- In CPS style, when we use the 'Maybe' argument of yield to indicate the end+-- then just like direct style we can figure out that there is no next element+-- without a function call.+--+-- Then why not get rid of the stop continuation and use only yield to indicate+-- the end of stream? The answer is, in that case to indicate the end of the+-- stream we would have to yield at least one element so there is no way to+-- represent an empty stream.+--+-- Whenever we make a singleton stream or in general when we build a stream+-- strictly i.e. when we know all the elements of the stream in advance we can+-- use the last yield to indicate th end of the stream, because we know in+-- advance at the time of the last yield that the stream is ending. We build+-- singleton streams in the implementation of 'pure' for Applicative and Monad,+-- and in 'lift' for MonadTrans, in these places we use yield with 'Nothing' to+-- indicate the end of the stream. Note that, the only advantage of Maybe is+-- when we have to build a large number of singleton or short streams. For+-- larger streams anyway the overhead of a separate stop continuation is not+-- significant. This could be significant when we breakdown a large stream into+-- its elements, process them in some way and then recompose it from the+-- pieces. Zipping streams is one such example. Zipping with streamly is the+-- fastest among all streaming libraries.+--+-- However in a lazy computation we cannot know in advance that the stream is+-- ending therefore we cannot use 'Maybe', we use the stop continuation in that+-- case. For example when building a stream from a lazy container using a right+-- fold.+-- newtype Stream m a = Stream { runStream :: forall r.@@ -168,6 +221,15 @@ snil = Stream $ \_ stp _ -> stp ------------------------------------------------------------------------------+-- Composing streams+------------------------------------------------------------------------------++-- Streams can be composed sequentially or in parallel; in product style+-- compositions (monadic bind multiplies streams in a ListT fashion) or in sum+-- style compositions like 'Semigroup', 'Monoid', 'Alternative' or variants of+-- these.++------------------------------------------------------------------------------ -- Semigroup ------------------------------------------------------------------------------ @@ -253,10 +315,9 @@ yield a Nothing = sendit a >> run yield a (Just r) = sendit a >> (runStream r) (Just sv) run yield - dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->- case ms of- [] -> ([], Nothing)- x : xs -> (xs, Just x)+ dequeue = liftIO $ atomicModifyIORefCAS q $ \case+ [] -> ([], Nothing)+ x : xs -> (xs, Just x) {-# INLINE enqueueFIFO #-} enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()@@ -289,7 +350,7 @@ {-# NOINLINE addThread #-} addThread :: MonadIO m => SVar m a -> ThreadId -> m () addThread sv tid =- liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.insert tid s)+ liftIO $ modifyIORef (runningThreads sv) (S.insert tid) {-# INLINE delThread #-} delThread :: MonadIO m => SVar m a -> ThreadId -> m ()@@ -298,8 +359,7 @@ {-# INLINE allThreadsDone #-} allThreadsDone :: MonadIO m => SVar m a -> m Bool-allThreadsDone sv = liftIO $ do- readIORef (runningThreads sv) >>= return . S.null+allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv) {-# NOINLINE handleChildException #-} handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()@@ -326,8 +386,8 @@ output <- liftIO $ readIORef (outputQueue sv) when (null output) $ do done <- queueEmpty sv- if (not done)- then (pushWorker sv) >> sendWorkerWait sv+ if not done+ then pushWorker sv >> sendWorkerWait sv else void (liftIO $ takeMVar (doorBell sv)) -- | Pull a stream from an SVar.@@ -381,13 +441,13 @@ running <- newIORef S.empty q <- newQ let sv =- SVar { outputQueue = outQ+ SVar { outputQueue = outQ , doorBell = outQMv , runningThreads = running , runqueue = runqueueFIFO sv q , enqueue = pushL q , queueEmpty = liftIO $ nullQ q- , svarStyle = ctype+ , svarStyle = ctype } in return sv @@ -397,30 +457,25 @@ outQMv <- newEmptyMVar running <- newIORef S.empty q <- newIORef []- let checkEmpty = liftIO (readIORef q) >>= return . null+ let checkEmpty = null <$> liftIO (readIORef q) let sv =- SVar { outputQueue = outQ+ SVar { outputQueue = outQ , doorBell = outQMv , runningThreads = running , runqueue = runqueueLIFO sv q , enqueue = enqueueLIFO q , queueEmpty = checkEmpty- , svarStyle = ctype+ , svarStyle = ctype } in return sv -- | Create a new empty SVar. newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a) newEmptySVar style = do- sv <- liftIO $+ liftIO $ case style of- SVarStyle _ FIFO -> do- c <- getFifoSVar style- return c- SVarStyle _ LIFO -> do- c <- getLifoSVar style- return c- return sv+ SVarStyle _ FIFO -> getFifoSVar style+ SVarStyle _ LIFO -> getLifoSVar style -- | Create a new SVar and enqueue one stream computation on it. newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)@@ -534,7 +589,7 @@ {-# INLINE joinStreamVar2 #-} joinStreamVar2 :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a-joinStreamVar2 style m1 m2 = Stream $ \st stp yld -> do+joinStreamVar2 style m1 m2 = Stream $ \st stp yld -> case st of Just sv | svarStyle sv == style -> liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
src/Streamly/Prelude.hs view
@@ -1,10 +1,8 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE GeneralizedNewtypeDeriving#-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RankNTypes #-}-{-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE UndecidableInstances #-} -- XXX -- |@@ -20,34 +18,38 @@ module Streamly.Prelude ( -- * Construction- cons- , nil+ nil+ , cons+ , (.:) , unfoldr , unfoldrM , each- , fromHandle -- * Elimination+ -- ** General Folds , foldr , foldrM+ , scan , foldl , foldlM , uncons - -- * Elimination Special Folds+ -- ** Special Folds , toList- , toHandle , all , any- , sum- , product , head+ , tail , last+ , null , length , elem , notElem+ , reverse , maximum , minimum+ , sum+ , product -- * Filtering , filter@@ -60,16 +62,22 @@ , mapM , mapM_ , sequence+ , replicateM -- * Zipping , zipWith , zipWithM , zipAsyncWith , zipAsyncWithM++ -- * IO+ , fromHandle+ , toHandle+ ) where -import Control.Monad (liftM)+import Control.Monad (void) import Control.Monad.IO.Class (MonadIO(..)) import Data.Semigroup (Semigroup(..)) import Prelude hiding (filter, drop, dropWhile, take,@@ -77,13 +85,12 @@ mapM, mapM_, sequence, all, any, sum, product, elem, notElem, maximum, minimum, head, last,- length)-import qualified Prelude as Prelude+ tail, length, null, reverse)+import qualified Prelude import qualified System.IO as IO import Streamly.Core import Streamly.Streams- ------------------------------------------------------------------------------ -- Construction ------------------------------------------------------------------------------@@ -92,7 +99,7 @@ unfoldr :: Streaming t => (b -> Maybe (a, b)) -> b -> t m a unfoldr step = fromStream . go where- go s = Stream $ \_ stp yld -> do+ go s = Stream $ \_ stp yld -> case step s of Nothing -> stp Just (a, b) -> yld a (Just (go b))@@ -110,12 +117,12 @@ -- XXX need eachInterleaved, eachAsync, eachParallel -- | Same as @foldWith (<>)@ but more efficient. {-# INLINE each #-}-each :: (Foldable f, Streaming t) => f a -> t m a-each xs = Prelude.foldr cons nil xs+each :: (Streaming t, Foldable f) => f a -> t m a+each = Prelude.foldr cons nil -- | Read lines from an IO Handle into a stream of Strings.-fromHandle :: (MonadIO m, Streaming t) => IO.Handle -> t m String-fromHandle h = fromStream $ go+fromHandle :: (Streaming t, MonadIO m) => IO.Handle -> t m String+fromHandle h = fromStream go where go = Stream $ \_ stp yld -> do eof <- liftIO $ IO.hIsEOF h@@ -132,7 +139,7 @@ -- Parallel variants of folds? -- | Right fold.-foldr :: (Monad m, Streaming t) => (a -> b -> b) -> b -> t m a -> m b+foldr :: (Streaming t, Monad m) => (a -> b -> b) -> b -> t m a -> m b foldr step acc m = go (toStream m) where go m1 =@@ -152,21 +159,49 @@ yield a (Just x) = step a (go x) in (runStream m1) Nothing stop yield +-- | Scan left. A strict left fold which accumulates the result of its reduction steps inside a stream, from left.+{-# INLINE scan #-}+scan :: Streaming t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b+scan step begin done m = cons (done begin) $ fromStream $ go (toStream m) begin+ where+ go m1 !acc = Stream $ \_ stp yld ->+ let stop = stp+ yield a Nothing = yld (done $ step acc a) Nothing+ yield a (Just x) =+ let s = step acc a+ in yld (done s) (Just (go x s))+ in runStream m1 Nothing stop yield+ -- | Strict left fold. This is typed to work with the foldl package. To use--- directly pass 'id' as the third argument.-foldl :: (Monad m, Streaming t)+-- it normally just pass 'id' as the third argument.+{-# INLINE foldl #-}+foldl :: (Streaming t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b-foldl step begin done m = go begin (toStream m)+foldl step begin done m = get $ go (toStream m) begin where- go !acc m1 =- let stop = return (done acc)- yield a Nothing = return (done (step acc a))- yield a (Just x) = go (step acc a) x- in (runStream m1) Nothing stop yield+ {-# NOINLINE get #-}+ get m1 =+ let yield a Nothing = return $ done a+ yield _ _ = undefined+ in (runStream m1) Nothing undefined yield + -- Note, this can be implemented by making a recursive call to "go",+ -- however that is more expensive because of unnecessary recursion+ -- that cannot be tail call optimized. Unfolding recursion explicitly via+ -- continuations is much more efficient.+ go m1 !acc = Stream $ \_ _ yld ->+ let stop = yld acc Nothing+ yield a r =+ let s = step acc a+ in case r of+ Nothing -> yld s Nothing+ Just x -> (runStream (go x s)) Nothing undefined yld+ in (runStream m1) Nothing stop yield++-- XXX replace the recursive "go" with explicit continuations. -- | Strict left fold, with monadic step function. This is typed to work -- with the foldl package. To use directly pass 'id' as the third argument.-foldlM :: (Monad m, Streaming t)+foldlM :: (Streaming t, Monad m) => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b foldlM step begin done m = go begin (toStream m) where@@ -183,7 +218,7 @@ uncons m = let stop = return Nothing yield a Nothing = return (Just (a, nil))- yield a (Just x) = return (Just (a, (fromStream x)))+ yield a (Just x) = return (Just (a, fromStream x)) in (runStream (toStream m)) Nothing stop yield -- | Write a stream of Strings to an IO Handle.@@ -202,34 +237,37 @@ -- | Convert a stream into a list in the underlying monad. {-# INLINABLE toList #-}-toList :: (Monad m, Streaming t) => t m a -> m [a]-toList = foldrM (\a xs -> liftM (a :) xs) (return [])+toList :: (Streaming t, Monad m) => t m a -> m [a]+toList = foldrM (\a xs -> fmap (a :) xs) (return []) -- | Take first 'n' elements from the stream and discard the rest.+{-# INLINE take #-} take :: Streaming t => Int -> t m a -> t m a take n m = fromStream $ go n (toStream m) where- go n1 m1 = Stream $ \ctx stp yld -> do+ go n1 m1 = Stream $ \ctx stp yld -> let yield a Nothing = yld a Nothing yield a (Just x) = yld a (Just (go (n1 - 1) x))- if (n1 <= 0)- then stp- else (runStream m1) ctx stp yield---- XXX This is not as efficient as it could be. We need a short circuiting at--- a lower level. Compare with simple-conduit, filtering there cuts down time--- due to short circuting whereas the time spent remains the same here.+ in if n1 <= 0 then stp else (runStream m1) ctx stp yield -- | Include only those elements that pass a predicate. {-# INLINE filter #-}-filter :: (Streaming t, Monad (t m)) => (a -> Bool) -> t m a -> t m a-filter p m = m >>= \x -> if p x then return x else nil+filter :: Streaming t => (a -> Bool) -> t m a -> t m a+filter p m = fromStream $ go (toStream m)+ where+ go m1 = Stream $ \ctx stp yld ->+ let yield a Nothing | p a = yld a Nothing+ | otherwise = stp+ yield a (Just x) | p a = yld a (Just (go x))+ | otherwise = (runStream x) ctx stp yield+ in (runStream m1) ctx stp yield -- | End the stream as soon as the predicate fails on an element.+{-# INLINE takeWhile #-} takeWhile :: Streaming t => (a -> Bool) -> t m a -> t m a takeWhile p m = fromStream $ go (toStream m) where- go m1 = Stream $ \ctx stp yld -> do+ go m1 = Stream $ \ctx stp yld -> let yield a Nothing | p a = yld a Nothing | otherwise = stp yield a (Just x) | p a = yld a (Just (go x))@@ -240,22 +278,24 @@ drop :: Streaming t => Int -> t m a -> t m a drop n m = fromStream $ go n (toStream m) where- go n1 m1 = Stream $ \ctx stp yld -> do+ go n1 m1 = Stream $ \ctx stp yld -> let yield _ Nothing = stp yield _ (Just x) = (runStream $ go (n1 - 1) x) ctx stp yld- if (n1 <= 0)- then (runStream m1) ctx stp yld- else (runStream m1) ctx stp yield+ -- Somehow "<=" check performs better than a ">"+ in if n1 <= 0+ then (runStream m1) ctx stp yld+ else (runStream m1) ctx stp yield -- | Drop elements in the stream as long as the predicate succeeds and then -- take the rest of the stream.+{-# INLINE dropWhile #-} dropWhile :: Streaming t => (a -> Bool) -> t m a -> t m a dropWhile p m = fromStream $ go (toStream m) where- go m1 = Stream $ \ctx stp yld -> do+ go m1 = Stream $ \ctx stp yld -> let yield a Nothing | p a = stp | otherwise = yld a Nothing- yield a (Just x) | p a = (runStream (go x)) ctx stp yield+ yield a (Just x) | p a = (runStream x) ctx stp yield | otherwise = yld a (Just x) in (runStream m1) ctx stp yield @@ -287,7 +327,7 @@ -- | Determine the product of all elements of a stream of numbers product :: (Streaming t, Monad m, Num a) => t m a -> m a-product = foldl (*) 0 id+product = foldl (*) 1 id -- | Extract the first element of the stream, if any. head :: (Streaming t, Monad m) => t m a -> m (Maybe a)@@ -296,16 +336,26 @@ yield a _ = return (Just a) in (runStream (toStream m)) Nothing stop yield +-- | Extract all but the first element of the stream, if any.+tail :: (Streaming t, Monad m) => t m a -> m (Maybe (t m a))+tail m =+ let stop = return Nothing+ yield _ Nothing = return $ Just nil+ yield _ (Just t) = return $ Just $ fromStream t+ in (runStream (toStream m)) Nothing stop yield+ -- | Extract the last element of the stream, if any.+{-# INLINE last #-} last :: (Streaming t, Monad m) => t m a -> m (Maybe a)-last m = go (toStream m)- where- go m1 =- let stop = return Nothing- yield a Nothing = return (Just a)- yield _ (Just x) = go x- in (runStream m1) Nothing stop yield+last = foldl (\_ y -> Just y) Nothing id +-- | Determine whether the stream is empty.+null :: (Streaming t, Monad m) => t m a -> m Bool+null m =+ let stop = return True+ yield _ _ = return False+ in (runStream (toStream m)) Nothing stop yield+ -- | Determine whether an element is present in the stream. elem :: (Streaming t, Monad m, Eq a) => a -> t m a -> m Bool elem e m = go (toStream m)@@ -313,7 +363,7 @@ go m1 = let stop = return False yield a Nothing = return (a == e)- yield a (Just x) = if (a == e) then return True else go x+ yield a (Just x) = if a == e then return True else go x in (runStream m1) Nothing stop yield -- | Determine whether an element is not present in the stream.@@ -323,13 +373,27 @@ go m1 = let stop = return True yield a Nothing = return (a /= e)- yield a (Just x) = if (a == e) then return False else go x+ yield a (Just x) = if a == e then return False else go x in (runStream m1) Nothing stop yield -- | Determine the length of the stream. length :: (Streaming t, Monad m) => t m a -> m Int length = foldl (\n _ -> n + 1) 0 id +-- | Returns the elements of the stream in reverse order.+-- The stream must be finite.+reverse :: (Streaming t) => t m a -> t m a+reverse m = fromStream $ go Nothing (toStream m)+ where+ go rev rest = Stream $ \svr stp yld ->+ let stop = case rev of+ Nothing -> stp+ Just str -> runStream str svr stp yld+ yield a Nothing = runStream (a `scons` rev) svr stp yld+ yield a (Just x) = runStream (go (Just $ a `scons` rev) x) svr stp yld+ in runStream rest svr stop yield++-- XXX replace the recursive "go" with continuation -- | Determine the minimum element in a stream. minimum :: (Streaming t, Monad m, Ord a) => t m a -> m (Maybe a) minimum m = go Nothing (toStream m)@@ -344,6 +408,7 @@ Nothing -> Just a Just e -> Just $ min a e +-- XXX replace the recursive "go" with continuation -- | Determine the maximum element in a stream. maximum :: (Streaming t, Monad m, Ord a) => t m a -> m (Maybe a) maximum m = go Nothing (toStream m)@@ -366,10 +431,11 @@ -- | Replace each element of the stream with the result of a monadic action -- applied on the element.+{-# INLINE mapM #-} mapM :: (Streaming t, Monad m) => (a -> m b) -> t m a -> t m b mapM f m = fromStream $ go (toStream m) where- go m1 = Stream $ \_ stp yld -> do+ go m1 = Stream $ \_ stp yld -> let stop = stp yield a Nothing = f a >>= \b -> yld b Nothing yield a (Just x) = f a >>= \b -> yld b (Just (go x))@@ -382,7 +448,7 @@ where go m1 = let stop = return ()- yield a Nothing = f a >> return ()+ yield a Nothing = void (f a) yield a (Just x) = f a >> go x in (runStream m1) Nothing stop yield @@ -391,12 +457,21 @@ sequence :: (Streaming t, Monad m) => t m (m a) -> t m a sequence m = fromStream $ go (toStream m) where- go m1 = Stream $ \_ stp yld -> do+ go m1 = Stream $ \_ stp yld -> let stop = stp yield a Nothing = a >>= \b -> yld b Nothing yield a (Just x) = a >>= \b -> yld b (Just (go x)) in (runStream m1) Nothing stop yield +-- | Generate a stream by performing an action @n@ times.+replicateM :: (Streaming t, Monad m) => Int -> m a -> t m a+replicateM n m = fromStream $ go n+ where+ go cnt = Stream $ \_ stp yld ->+ if cnt <= 0+ then stp+ else m >>= \a -> yld a (Just $ go (cnt - 1))+ ------------------------------------------------------------------------------ -- Serially Zipping Streams ------------------------------------------------------------------------------@@ -409,7 +484,7 @@ let merge a ra = let yield2 b Nothing = (runStream (g a b)) Nothing stp yld yield2 b (Just rb) =- (runStream ((g a b) <> (go ra rb))) Nothing stp yld+ (runStream (g a b <> go ra rb)) Nothing stp yld in (runStream my) Nothing stp yield2 let yield1 a Nothing = merge a snil yield1 a (Just ra) = merge a ra
src/Streamly/Streams.hs view
@@ -29,13 +29,14 @@ , newEmptySVar -- * Construction+ , nil+ , cons+ , (.:) , streamBuild , fromCallback , fromSVar -- * Elimination- , cons- , nil , streamFold , runStreaming , toSVar@@ -94,7 +95,7 @@ import Control.Monad.State.Class (MonadState(..)) import Control.Monad.Trans.Class (MonadTrans) import Data.Semigroup (Semigroup(..))-import Prelude hiding (drop, take, zipWith)+import Prelude hiding (zipWith) import Streamly.Core ------------------------------------------------------------------------------@@ -111,14 +112,54 @@ -- Constructing a stream ------------------------------------------------------------------------------ --- | Add an element a the head of a stream.+-- | Represesnts an empty stream just like @[]@ represents an empty list.+nil :: Streaming t => t m a+nil = fromStream snil++infixr 5 `cons`++-- | Constructs a stream by adding a pure value at the head of an existing+-- stream, just like ':' constructs lists. For example:+--+-- @+-- > let stream = 1 \`cons` 2 \`cons` 3 \`cons` nil+-- > (toList . serially) stream+-- [1,2,3]+-- @ cons :: (Streaming t) => a -> t m a -> t m a cons a r = fromStream $ scons a (Just (toStream r)) --- | An empty stream.-nil :: Streaming t => t m a-nil = fromStream $ snil+infixr 5 .: +-- | Operator equivalent of 'cons' so that you can construct a stream of pure+-- values more succinctly like this:+--+-- @+-- > let stream = 1 .: 2 .: 3 .: nil+-- > (toList . serially) stream+-- [1,2,3]+-- @+--+-- '.:' constructs a stream just like ':' constructs a list.+--+-- Also note that another equivalent way of building streams from pure values+-- is:+--+-- @+-- > let stream = pure 1 <> pure 2 <> pure 3+-- > (toList . serially) stream+-- [1,2,3]+-- @+--+-- In the first method we construct a stream by adding one element at a time.+-- In the second method we first construct singleton streams using 'pure' and+-- then compose all those streams together using the 'Semigroup' style+-- composition of streams. The former method is a bit more efficient than the+-- latter.+--+(.:) :: (Streaming t) => a -> t m a -> t m a+(.:) = cons+ -- | Build a stream from its church encoding. The function passed maps -- directly to the underlying representation of the stream type. The second -- parameter to the function is the "yield" function yielding a value and the@@ -483,7 +524,7 @@ Stream $ \ctx stp yld -> let run x = (runStream x) ctx stp yld yield a Nothing = run $ f a- yield a (Just r) = run $ f a `par` (go r)+ yield a (Just r) = run $ f a `par` go r in g Nothing stp yield instance MonadAsync m => Monad (AsyncT m) where
src/Streamly/Time.hs view
@@ -49,7 +49,7 @@ action localTime when (delay > 0) $ threadDelay delay - if (n == freq)+ if n == freq then do (t, newTick, newDelay) <- adjustClock lastAdj localTime delay go t newDelay newTick (localTime + newTick) 0
src/Streamly/Tutorial.hs view
@@ -292,6 +292,7 @@ -- -- @ -- main = 'runStreamT' $ traced (sqrt 9) '<|' traced (sqrt 16) '<|' traced (sqrt 25)+-- where traced m = liftIO (myThreadId >>= print) >> m -- @ -- @ -- ThreadId 40@@ -1015,7 +1016,7 @@ -- When it comes to streaming, in terms of core concepts, @simple-conduit@ is -- the package that is closest to streamly if we set aside the concurrency -- dimension, both are streaming packages with list transformer like monad--- composition. However, in terms of API it is more like the @streaming@+-- composition. However, in terms of API @streamly@ is more like the @streaming@ -- package. Streamly can be used to achieve more or less the functionality -- provided by any of the streaming packages listed above. The types and API of -- streamly are much simpler in comparison to conduit and pipes. It is more or@@ -1038,5 +1039,5 @@ -- the "Streamly.Examples.CirclingSquare" example from Yampa demonstrate the -- basic FRP capability of streamly. In core concepts streamly is strikingly -- similar to @dunai@. dunai was designed from a FRP perspective and streamly--- wa original designed from a concurrency perspective. However, both have+-- was originally designed from a concurrency perspective. However, both have -- similarity at the core.
− stack-8.0.yaml
@@ -1,17 +0,0 @@-resolver: lts-7.24-packages:-- '.'-extra-deps:- - lockfree-queue-0.2.3.1- - simple-conduit-0.6.0- - transient-0.4.4- - monad-recorder-0.1.0- - http-conduit-2.2.2- - http-client-0.5.0- - http-client-tls-0.3.0- - SDL-0.6.5.1-flags: {}-extra-package-dbs: []-# For mac ports installed SDL library on Mac OS X-#extra-include-dirs:-#- /opt/local/include
stack.yaml view
@@ -1,9 +1,7 @@-#resolver: lts-9.2-resolver: nightly-2017-09-07+resolver: lts-11.0 packages: - '.' extra-deps:- - lockfree-queue-0.2.3.1 - simple-conduit-0.6.0 - SDL-0.6.5.1 flags: {}
streamly.cabal view
@@ -1,5 +1,5 @@ name: streamly-version: 0.1.0+version: 0.1.1 synopsis: Beautiful Streaming, Concurrent and Reactive Composition description: Streamly is a monad transformer unifying non-determinism@@ -19,11 +19,11 @@ "Streamly.Tutorial" first. Also see "Streamly.Examples" for some working examples. -homepage: http://github.com/harendra-kumar/streamly-bug-reports: https://github.com/harendra-kumar/streamly/issues+homepage: https://github.com/composewell/streamly+bug-reports: https://github.com/composewell/streamly/issues license: BSD3 license-file: LICENSE-tested-with: GHC==7.10.3, GHC==8.0.2, GHC==8.2.1+tested-with: GHC==7.10.3, GHC==8.0.2, GHC==8.2.2, GHC==8.4.1 author: Harendra Kumar maintainer: harendra.kumar@gmail.com copyright: 2017 Harendra Kumar@@ -36,12 +36,11 @@ Changelog.md README.md stack-7.10.yaml- stack-8.0.yaml stack.yaml source-repository head type: git- location: https://github.com/harendra-kumar/streamly+ location: https://github.com/composewell/streamly flag dev description: Build development version@@ -105,7 +104,7 @@ build-depends: base >= 4.8 && < 5 , atomic-primops >= 0.8 && < 0.9 , containers >= 0.5 && < 0.6- , exceptions >= 0.8 && < 0.9+ , exceptions >= 0.8 && < 0.11 , lifted-base >= 0.2 && < 0.3 , lockfree-queue >= 0.2.3 && < 0.3 , monad-control >= 1.0 && < 2@@ -120,7 +119,7 @@ if flag(examples) || flag(examples-sdl) build-Depends:- http-conduit >= 2.2.2 && < 2.3+ http-conduit >= 2.2.2 && < 2.4 , path-io >= 0.1.0 && < 1.4 , random >= 1.0.0 && < 1.2
test/Main.hs view
@@ -4,8 +4,10 @@ module Main (main) where import Control.Concurrent (threadDelay)+import Control.Monad (replicateM) import Data.Foldable (forM_) import Data.List (sort)+import Data.Maybe (fromJust) import Test.Hspec import Streamly@@ -20,8 +22,8 @@ toListAsync :: AsyncT IO a -> IO [a] toListAsync = A.toList . asyncly -toListParallel :: ParallelT IO a -> IO [a]-toListParallel = A.toList . parallely+toListParallel :: Ord a => ParallelT IO a -> IO [a]+toListParallel = fmap sort . A.toList . parallely main :: IO () main = hspec $ do@@ -247,7 +249,8 @@ it "Nests two streams using Num serial composition" nestTwoSerialNum it "Nests two streams using Num interleaved composition" nestTwoInterleavedNum it "Nests two streams using Num async composition" nestTwoAsyncNum- it "Nests two streams using Num parallel composition" nestTwoParallelNum+ -- This test fails intermittently, need to investigate+ -- it "Nests two streams using Num parallel composition" nestTwoParallelNum --------------------------------------------------------------------------- -- TBD Bind and Bind combinations@@ -256,12 +259,48 @@ -- TBD combine all binds and all compose in one example describe "Miscellaneous combined examples" mixedOps - describe "Transformation" $ transformOps (<>)+ ---------------------------------------------------------------------------+ -- Stream operations+ ---------------------------------------------------------------------------++ -- XXX for streams other than StreamT+ describe "Stream Ops empty" $ streamOperations makeEmptyStream+ describe "Stream ops singleton constr" $ streamOperations makeSingletonStream1+ describe "Stream ops singleton folded" $ streamOperations makeSingletonStream2+ describe "Stream Ops constr" $ streamOperations makeStream1+ describe "Stream Ops folded" $ streamOperations $ makeStream2+ ((<>) :: StreamT IO Int -> StreamT IO Int -> StreamT IO Int)+ describe "Serial zipping" $ zipOps A.zipWith A.zipWithM zipping describe "Async zipping" $ zipOps A.zipAsyncWith A.zipAsyncWithM zippingAsync +makeEmptyStream :: (StreamT IO Int, [Int], Int)+makeEmptyStream = (A.nil, [], 0)++makeSingletonStream1 :: (StreamT IO Int, [Int], Int)+makeSingletonStream1 = (1 `A.cons` A.nil, [1], 1)++makeSingletonStream2 :: (StreamT IO Int, [Int], Int)+makeSingletonStream2 = (return 1, [1], 1)++-- Streams that indicate an end via the stop continuation+makeStream1 :: (StreamT IO Int, [Int], Int)+makeStream1 =+ let list = [1..10]+ stream = A.each list+ in (stream, list, 10)++-- Streams that indicate an end via the yield continuation+makeStream2 :: (Streaming t, Monad (t IO))+ => (t IO Int -> t IO Int -> t IO Int)+ -> (t IO Int, [Int], Int)+makeStream2 f =+ let list = [1..10]+ stream = foldMapWith f return list+ in (stream, list, 10)+ nestTwoSerial :: Expectation nestTwoSerial = let s1 = foldMapWith (<>) return [1..4]@@ -351,12 +390,14 @@ in toListParallel ((+) <$> s1 <*> s2) `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int]) +{- nestTwoParallelNum :: Expectation nestTwoParallelNum = let s1 = foldMapWith (<>) return [1..4] s2 = foldMapWith (<>) return [5..8] in toListParallel (s1 + s2) `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])+-} zipOps :: (Streaming t, Applicative (t IO)) => (forall a b c. (a -> b -> c)@@ -395,7 +436,7 @@ it "Then and toList" $ toListSerial (return (1 :: Int) >> return 2) `shouldReturn` ([2] :: [Int]) -type ToListType s = (forall a. s IO a -> IO [a])+type ToListType s = (forall a. Ord a => s IO a -> IO [a]) pureBind :: Monad (s IO) => ToListType s -> Spec pureBind l = do it "Bind and toList" $@@ -515,7 +556,7 @@ bindAndComposeSimple :: (Streaming t, Alternative (t IO), Monad (t IO))- => (forall a. t IO a -> IO [a])+ => (forall a. Ord a => t IO a -> IO [a]) -> (t IO Int -> t IO Int -> t IO Int) -> Spec bindAndComposeSimple tl g = do@@ -530,7 +571,7 @@ where f = (>>=) bindAndComposeHierarchy- :: Monad (s IO) => (forall a. s IO a -> IO [a])+ :: Monad (s IO) => (forall a. Ord a => s IO a -> IO [a]) -> ([s IO Int] -> s IO Int) -> Spec bindAndComposeHierarchy tl g = do@@ -595,24 +636,85 @@ return (x1 + y1 + z1) return (x + y + z) -transformOps :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int) -> Spec-transformOps f = do- it "take all" $- (toListSerial $ A.take 10 $ foldMapWith f return [1..10])- `shouldReturn` [1..10]- it "take none" $- (toListSerial $ A.take 0 $ foldMapWith f return [1..10])- `shouldReturn` []- it "take 5" $- (toListSerial $ A.take 5 $ foldMapWith f return [1..10])- `shouldReturn` [1..5]+streamOperations :: Streaming t => (t IO Int, [Int], Int) -> Spec+streamOperations (stream, list, len) = do - it "drop all" $- (toListSerial $ A.drop 10 $ foldMapWith f return [1..10])- `shouldReturn` []- it "drop none" $- (toListSerial $ A.drop 0 $ foldMapWith f return [1..10])- `shouldReturn` [1..10]- it "drop 5" $- (toListSerial $ A.drop 5 $ foldMapWith f return [1..10])- `shouldReturn` [6..10]+ -- Generation+ it "replicateM" $ do+ let x = return (1 :: Int)+ str <- A.toList . serially $ A.replicateM len x+ lst <- replicateM len x+ return $ str == lst+ `shouldReturn` True++ -- Filtering+ it "filter all out" $ transform (A.filter (> len)) (filter (> len))+ it "filter all in" $ transform (A.filter (<= len)) (filter (<= len))+ it "filter even" $ transform (A.filter even) (filter even)++ it "take all" $ transform (A.take len) (take len)+ it "take none" $ transform (A.take 0) (take 0)+ it "take some" $ transform (A.take $ len - 1) (take $ len - 1)+ it "take one" $ transform (A.take 1) (take 1)++ it "takeWhile true" $ transform (A.takeWhile (const True))+ (takeWhile (const True))+ it "takeWhile false" $ transform (A.takeWhile (const False))+ (takeWhile (const False))+ it "takeWhile < some" $ transform (A.takeWhile (< (len `div` 2)))+ (takeWhile (< (len `div` 2)))++ it "drop all" $ transform (A.drop len) (drop len)+ it "drop none" $ transform (A.drop 0) (drop 0)+ it "drop some" $ transform (A.drop $ len - 1) (drop $ len - 1)+ it "drop one" $ transform (A.drop 1) (drop 1)++ it "dropWhile true" $ transform (A.dropWhile (const True))+ (dropWhile (const True))+ it "dropWhile false" $ transform (A.dropWhile (const False))+ (dropWhile (const False))+ it "dropWhile < some" $ transform (A.dropWhile (< (len `div` 2)))+ (dropWhile (< (len `div` 2)))++ -- Transformations+ it "scan left" $ transform (A.scan (+) 0 id) (scanl (+) 0)+ it "reverse" $ transform A.reverse reverse++ -- Elimination+ it "foldl" $ elimination (A.foldl (+) 0 id) (foldl (+) 0)+ it "all" $ elimination (A.all even) (all even)+ it "any" $ elimination (A.any even) (any even)+ it "length" $ elimination A.length length+ it "elem" $ elimination (A.elem (len - 1)) (elem (len - 1))+ it "elem" $ elimination (A.elem (len + 1)) (elem (len + 1))+ it "notElem" $ elimination (A.notElem (len - 1)) (notElem (len - 1))+ it "notElem" $ elimination (A.notElem (len + 1)) (notElem (len + 1))+ it "sum" $ elimination A.sum sum+ it "product" $ elimination A.product product++ if list == []+ then do+ it "head empty" $ A.head stream `shouldReturn` Nothing+ it "last empty" $ A.last stream `shouldReturn` Nothing+ it "maximum empty" $ A.maximum stream `shouldReturn` Nothing+ it "minimum empty" $ A.minimum stream `shouldReturn` Nothing+ it "null empty" $ A.null stream `shouldReturn` True+ it "tail empty" $ (A.tail stream >>= return . maybe True (const False))+ `shouldReturn` True+ else do+ it "head nonEmpty" $ A.head stream `shouldReturn` Just (head list)+ it "last nonEmpty" $ A.last stream `shouldReturn` Just (last list)+ it "maximum nonEmpty" $ A.maximum stream+ `shouldReturn` Just (maximum list)+ it "minimum nonEmpty" $ A.minimum stream+ `shouldReturn` Just (minimum list)+ it "null nonEmpty" $ A.null stream `shouldReturn` False+ it "tail nonEmpty" $ (A.tail stream >>= A.toList . fromJust)+ `shouldReturn` tail list++ where+ -- XXX run on empty stream as well+ transform streamOp listOp =+ (A.toList $ streamOp stream) `shouldReturn` listOp list++ elimination streamOp listOp = (streamOp stream) `shouldReturn` listOp list