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