on ede29df help sign in
379 lines 12 kB history raw
1{-# LANGUAGE BangPatterns #-}
2{-# LANGUAGE CPP #-}
3{-# LANGUAGE ConstraintKinds #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE FlexibleInstances #-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE PatternSynonyms #-}
9{-# LANGUAGE ViewPatterns #-}
10{-# LANGUAGE RankNTypes #-}
11{-# LANGUAGE UndecidableInstances #-} -- XXX
13-- |
14-- Module : Streamly.Streams.StreamK.Type
15-- Copyright : (c) 2017 Harendra Kumar
16--
17-- License : BSD3
18-- Maintainer : harendra.kumar@gmail.com
19-- Stability : experimental
20-- Portability : GHC
21--
22--
23-- Continuation passing style (CPS) stream implementation. The symbol 'K' below
24-- denotes a function as well as a Kontinuation.
25--
26module Streamly.Streams.StreamK.Type
27 (
28 -- * A class for streams
29 IsStream (..)
30 , adapt
32 -- * The stream type
33 , Stream ()
35 -- * Construction
36 , mkStream
37 , fromStopK
38 , fromYieldK
39 , consK
41 -- * Elimination
42 , foldStream
43 , foldStreamShared
44 , foldStreamSVar
46 -- instances
47 , consMStream
49 , nil
50 , serial
51 , map
52 , yieldM
54 , Streaming -- deprecated
55 )
56where
58import Control.Monad (void)
59import Control.Monad.IO.Class (MonadIO(liftIO))
60import Control.Monad.Trans.Class (MonadTrans(lift))
61import Data.Semigroup (Semigroup(..))
62import Prelude hiding (map)
64import Streamly.SVar
66------------------------------------------------------------------------------
67-- Basic stream type
68------------------------------------------------------------------------------
70-- | The type @Stream m a@ represents a monadic stream of values of type 'a'
71-- constructed using actions in monad 'm'. It uses stop, singleton and yield
72-- continuations equivalent to the following direct style type:
73--
74-- @
75-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
76-- @
77--
78-- To facilitate parallel composition we maintain a local state in an 'SVar'
79-- that is shared across and is used for synchronization of the streams being
80-- composed.
81--
82-- The singleton case can be expressed in terms of stop and yield but we have
83-- it as a separate case to optimize composition operations for streams with
84-- single element. We build singleton streams in the implementation of 'pure'
85-- for Applicative and Monad, and in 'lift' for MonadTrans.
86--
87-- XXX remove the Stream type parameter from State as it is always constant.
88-- We can remove it from SVar as well
89--
90newtype Stream m a =
91 MkStream (forall r.
92 State Stream m a -- state
93 -> (a -> Stream m a -> m r) -- yield
94 -> (a -> m r) -- singleton
95 -> m r -- stop
96 -> m r
97 )
99------------------------------------------------------------------------------
100-- Types that can behave as a Stream
101------------------------------------------------------------------------------
103infixr 5 `consM`
104infixr 5 |:
106-- XXX Use a different SVar based on the stream type. But we need to make sure
107-- that we do not lose performance due to polymorphism.
108--
109-- | Class of types that can represent a stream of elements of some type 'a' in
110-- some monad 'm'.
111--
112-- @since 0.2.0
113class IsStream t where
114 toStream :: t m a -> Stream m a
115 fromStream :: Stream m a -> t m a
116 -- | Constructs a stream by adding a monadic action at the head of an
117 -- existing stream. For example:
118 --
119 -- @
120 -- > toList $ getLine \`consM` getLine \`consM` nil
121 -- hello
122 -- world
123 -- ["hello","world"]
124 -- @
125 --
126 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
127 --
128 -- @since 0.2.0
129 consM :: MonadAsync m => m a -> t m a -> t m a
130 -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
131 -- to remember that @|@ comes before ':'.
132 --
133 -- @
134 -- > toList $ getLine |: getLine |: nil
135 -- hello
136 -- world
137 -- ["hello","world"]
138 -- @
139 --
140 -- @
141 -- let delay = threadDelay 1000000 >> print 1
142 -- runStream $ serially $ delay |: delay |: delay |: nil
143 -- runStream $ parallely $ delay |: delay |: delay |: nil
144 -- @
145 --
146 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
147 --
148 -- @since 0.2.0
149 (|:) :: MonadAsync m => m a -> t m a -> t m a
150 -- We can define (|:) just as 'consM' but it is defined explicitly for each
151 -- type because we want to use SPECIALIZE pragma on the definition.
153-- | Same as 'IsStream'.
154--
155-- @since 0.1.0
156{-# DEPRECATED Streaming "Please use IsStream instead." #-}
157type Streaming = IsStream
159-------------------------------------------------------------------------------
160-- Type adapting combinators
161-------------------------------------------------------------------------------
163-- XXX Move/reset the State here by reconstructing the stream with cleared
164-- state. Can we make sure we do not do that when t1 = t2? If we do this then
165-- we do not need to do that explicitly using svarStyle. It would act as
166-- unShare when the stream type is the same.
167--
168-- | Adapt any specific stream type to any other specific stream type.
169--
170-- @since 0.1.0
171adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
172adapt = fromStream . toStream
174------------------------------------------------------------------------------
175-- Building a stream
176------------------------------------------------------------------------------
178-- XXX The State is always parameterized by "Stream" which means State is not
179-- different for different stream types. So we have to manually make sure that
180-- when converting from one stream to another we migrate the state correctly.
181-- This can be fixed if we use a different SVar type for different streams.
182-- Currently we always use "SVar Stream" and therefore a different State type
183-- parameterized by that stream.
184--
185-- | Build a stream from an 'SVar', a stop continuation, a singleton stream
186-- continuation and a yield continuation.
187{-# INLINE mkStream #-}
188mkStream:: IsStream t
189 => (forall r. State Stream m a
190 -> (a -> t m a -> m r)
191 -> (a -> m r)
192 -> m r
193 -> m r)
194 -> t m a
195mkStream k = fromStream $ MkStream $ \st yld sng stp ->
196 let yieldk a r = yld a (toStream r)
197 in k st yieldk sng stp
199-- | A terminal function that has no continuation to follow.
200type StopK m = forall r. m r -> m r
202-- | A monadic continuation, it is a function that yields a value of type "a"
203-- and calls the argument (a -> m r) as a continuation with that value. We can
204-- also think of it as a callback with a handler (a -> m r). Category
205-- theorists call it a codensity type, a special type of right kan extension.
206type YieldK m a = forall r. (a -> m r) -> m r
208_wrapM :: Monad m => m a -> YieldK m a
209_wrapM m = \k -> m >>= k
211-- | Make an empty stream from a stop function.
212fromStopK :: IsStream t => StopK m -> t m a
213fromStopK k = mkStream $ \_ _ _ stp -> k stp
215-- | Make a singleton stream from a yield function.
216fromYieldK :: IsStream t => YieldK m a -> t m a
217fromYieldK k = mkStream $ \_ _ sng _ -> k sng
219-- | Add a yield function at the head of the stream.
220consK :: IsStream t => YieldK m a -> t m a -> t m a
221consK k r = mkStream $ \_ yld _ _ -> k (\x -> yld x r)
223-- XXX Build a stream from a repeating callback function.
225------------------------------------------------------------------------------
226-- Folding a stream
227------------------------------------------------------------------------------
229-- | Fold a stream by providing an SVar, a stop continuation, a singleton
230-- continuation and a yield continuation. The stream would share the current
231-- SVar passed via the State.
232{-# INLINE foldStreamShared #-}
233foldStreamShared
234 :: IsStream t
235 => State Stream m a
236 -> (a -> t m a -> m r)
237 -> (a -> m r)
238 -> m r
239 -> t m a
240 -> m r
241foldStreamShared st yld sng stp m =
242 let yieldk a x = yld a (fromStream x)
243 MkStream k = toStream m
244 in k st yieldk sng stp
246-- | Fold a stream by providing a State, stop continuation, a singleton
247-- continuation and a yield continuation. The stream will not use the SVar
248-- passed via State.
249{-# INLINE foldStream #-}
250foldStream
251 :: IsStream t
252 => State Stream m a
253 -> (a -> t m a -> m r)
254 -> (a -> m r)
255 -> m r
256 -> t m a
257 -> m r
258foldStream st yld sng stp m =
259 let yieldk a x = yld a (fromStream x)
260 MkStream k = toStream m
261 in k (adaptState st) yieldk sng stp
263-- Run the stream using a run function associated with the SVar that runs the
264-- streams with a captured snapshot of the monadic state.
265{-# INLINE foldStreamSVar #-}
266foldStreamSVar
267 :: (IsStream t, MonadIO m)
268 => SVar Stream m a
269 -> State Stream m a -- state
270 -> (a -> t m a -> m r) -- yield
271 -> (a -> m r) -- singleton
272 -> m r -- stop
273 -> t m a
274 -> m ()
275foldStreamSVar sv st yld sng stp m =
276 let mrun = runInIO $ svarMrun sv
277 in void $ liftIO $ mrun $ foldStreamShared st yld sng stp m
279-------------------------------------------------------------------------------
280-- Instances
281-------------------------------------------------------------------------------
283-- NOTE: specializing the function outside the instance definition seems to
284-- improve performance quite a bit at times, even if we have the same
285-- SPECIALIZE in the instance definition.
286{-# INLINE consMStream #-}
287{-# SPECIALIZE consMStream :: IO a -> Stream IO a -> Stream IO a #-}
288consMStream :: (Monad m) => m a -> Stream m a -> Stream m a
289consMStream m r = MkStream $ \_ yld _ _ -> m >>= \a -> yld a r
291-------------------------------------------------------------------------------
292-- IsStream Stream
293-------------------------------------------------------------------------------
295instance IsStream Stream where
296 toStream = id
297 fromStream = id
299 {-# INLINE consM #-}
300 {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
301 consM :: Monad m => m a -> Stream m a -> Stream m a
302 consM = consMStream
304 {-# INLINE (|:) #-}
305 {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
306 (|:) :: Monad m => m a -> Stream m a -> Stream m a
307 (|:) = consMStream
309------------------------------------------------------------------------------
310-- Semigroup
311------------------------------------------------------------------------------
313-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
314-- Appends two streams sequentially, yielding all elements from the first
315-- stream, and then all elements from the second stream.
316--
317-- @since 0.2.0
318{-# INLINE serial #-}
319serial :: IsStream t => t m a -> t m a -> t m a
320serial m1 m2 = go m1
321 where
322 go m = mkStream $ \st yld sng stp ->
323 let stop = foldStream st yld sng stp m2
324 single a = yld a m2
325 yieldk a r = yld a (go r)
326 in foldStream st yieldk single stop m
328instance Semigroup (Stream m a) where
329 (<>) = serial
331------------------------------------------------------------------------------
332-- Monoid
333------------------------------------------------------------------------------
335-- | An empty stream.
336--
337-- @
338-- > toList nil
339-- []
340-- @
341--
342-- @since 0.1.0
343{-# INLINE nil #-}
344nil :: IsStream t => t m a
345nil = mkStream $ \_ _ _ stp -> stp
347instance Monoid (Stream m a) where
348 mempty = nil
349 mappend = (<>)
351-------------------------------------------------------------------------------
352-- Functor
353-------------------------------------------------------------------------------
355{-# INLINE map #-}
356map :: IsStream t => (a -> b) -> t m a -> t m b
357map f m = go m
358 where
359 go m1 =
360 mkStream $ \st yld sng stp ->
361 let single = sng . f
362 yieldk a r = yld (f a) (go r)
363 in foldStream (adaptState st) yieldk single stp m1
365-- in fact use the Stream type everywhere and only use polymorphism in the high
366-- level modules/prelude.
367instance Monad m => Functor (Stream m) where
368 fmap = map
370-------------------------------------------------------------------------------
371-- Transformers
372-------------------------------------------------------------------------------
374{-# INLINE yieldM #-}
375yieldM :: (Monad m, IsStream t) => m a -> t m a
376yieldM m = fromStream $ mkStream $ \_ _ single _ -> m >>= single
378instance MonadTrans Stream where
379 lift = yieldM