~z      !"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\]^_`abcdefghijklmnopqrstuvwxy(c) 2017 Harendra KumarBSD3harendra.kumar@gmail.com experimentalGHCNone +;<=>?DQV]^A monad that can perform concurrent or parallel IO operations. Streams that can be composed concurrently require the underlying monad to be .zDThe type 'Stream m a' represents a monadic stream of values of type a% constructed using actions in monad ma. It uses stop, singleton and yield continuations equivalent to the following direct style type:;data Stream m a = Stop | Singleton a | Yield a (Stream m a)To facilitate parallel composition we maintain a local state in an SVar that is shared across and is used for synchronization of the streams being composed.The singleton case can be expressed in terms of stop and yield but we have it as a separate case to optimize composition operations for streams with single element. We build singleton streams in the implementation of {$ for Applicative and Monad, and in | for MonadTrans.}VAn SVar or a Stream Var is a conduit to the output from multiple streams running concurrently and asynchronously. An SVar can be thought of as an asynchronous IO handle. We can write any number of streams to an SVar in a non-blocking manner and then read them back at any time at any pace. The SVar would run the streams asynchronously and accumulate results. An SVar may not really execute the stream completely and accumulate all the results. However, it ensures that the reader can read the results at whatever paces it wants to read. The SVar monitors and adapts to the consumer's pace.An SVar is a mini scheduler, it has an associated runqueue that holds the stream tasks to be picked and run by a pool of worker threads. It has an associated output queue where the output stream elements are placed by the worker threads. A doorBell is used by the worker threads to intimate the consumer thread about availability of new results in the output queue. More workers are added to the SVar by ~" on demand if the output produced is not keeping pace with the consumer. On bounded SVars, workers block on the output queue to provide throttling of the producer when the consumer is not pulling fast enough. The number of workers may even get reduced depending on the consuming pace.{New work is enqueued either at the time of creation of the SVar or as a result of executing the parallel combinators i.e. <| and <|>< when the already enqueued computations get evaluated. See .hIdentify the type of the SVar. Two computations using the same style can be scheduled on the same SVar.=Sorting out-of-turn outputs in a heap for Ahead style streams7Events that a child thread may send to a parent thread.+faster than consM because there is no bind.Same as  once . return+ but may be faster because there is no bindConcatenates two streams sequentially i.e. the first stream is exhausted completely before yielding any element from the second stream.This function is used by the producer threads to queue output for the consumer thread to consume. Returns whether the queue has more space.}This is safe even if we are adding more threads concurrently because if a child thread is adding another thread then anyway  will not be empty.=In contrast to pushWorker which always happens only from the consumer thread, a pushWorkerPar can happen concurrently from multiple threads on the producer side. So we need to use a thread safe modification of runningThreads. Alternatively, we can use a CreateThread event to avoid using a CAS based modification.~Pull a stream from an SVar.Create a new empty SVar.;Create a new SVar and enqueue one stream computation on it.;Create a new SVar and enqueue one stream computation on it.Write a stream to an }Q in a non-blocking manner. The stream can then be read back from the SVar using fromSVar./Join two computations on the currently running } queue for concurrent execution. When we are using parallel composition, an SVar is passed around as a state variable. We try to schedule a new parallel computation on the SVar passed to us. The first time, when no SVar exists, a new SVar is created. Subsequently, n may get called when a computation already scheduled on the SVar is further evaluated. For example, when (a  b) is evaluated it calls a  to put a and b! on the current scheduler queue.The \ required by the current composition context is passed as one of the parameters. If the scheduling and composition style of the new computation being scheduled is different than the style of the current SVar, then we create a new SVar and schedule it on that. The newly created SVar joins as one of the computations on the current SVar queue.+Cases when we need to switch to a new SVar:(x  y)  (t 1 u) -- all of them get scheduled on the same SVar(x  y)  (t  u) -- t and uN get scheduled on a new child SVar because of the scheduling policy change.if we adapt a stream of type  to a stream of type Parallel1, we create a new SVar at the transitioning bind.When the stream is switching from disjunctive composition to conjunctive composition and vice-versa we create a new SVar to isolate the scheduling of the two.kXXX we can implement it more efficienty by directly implementing instead of combining streams using ahead.kXXX we can implement it more efficienty by directly implementing instead of combining streams using async.lXXX we can implement it more efficienty by directly implementing instead of combining streams using wAsync.nXXX we can implement it more efficienty by directly implementing instead of combining streams using parallel.!z}~z} (c) 2017 Harendra KumarBSD3harendra.kumar@gmail.com experimentalGHCNone +.0;<=>?KQVxD>An IO stream whose applicative instance zips streams wAsyncly.>An IO stream whose applicative instance zips streams serially.4A parallely composing IO stream of elements of type a. See  documentation for more details.@A round robin parallely composing IO stream of elements of type a. See  documentation for more details. OA demand driven left biased parallely composing IO stream of elements of type a. See  documentation for more details. 'A serial IO stream of elements of type a" with concurrent lookahead. See  documentation for more details. 5An interleaving serial IO stream of elements of type a. See ! documentation for more details. 'A serial IO stream of elements of type a. See ! documentation for more details. Like P but zips in parallel, it generates all the elements to be zipped concurrently. main = (toList . 8 $ (,,) <$> s1 <*> s2 <*> s3) >>= print where s1 = fromFoldable [1, 2] s2 = fromFoldable [3, 4] s3 = fromFoldable [5, 6]  [(1,3,5),(2,4,6)] The 6 instance of this type works the same way as that of .The applicative instance of } zips a number of streams serially i.e. it produces one element from each stream serially and then zips all those elements. main = (toList . 6 $ (,,) <$> s1 <*> s2 <*> s3) >>= print where s1 = fromFoldable [1, 2] s2 = fromFoldable [3, 4] s3 = fromFoldable [5, 6]  [(1,3,5),(2,4,6)] The 6 instance of this type works the same way as that of .=Async composition with simultaneous traversal of all streams.The Semigroup instance of  concurrently merges two streams, running both strictly concurrently and yielding elements from both streams as they arrive. When multiple streams are combined using  each one is evaluated in its own thread and the results produced are presented in the combined stream on a first come first serve basis. and  are concurrent lookahead streams each with a specific type of consumption pattern (depth first or breadth first). Since they are lookahead, they may introduce certain default latency in starting more concurrent tasks for efficiency reasons or may put a default limitation on the resource consumption (e.g. number of concurrent threads for lookahead). If we look at the implementation detail, they both can share a pool of worker threads to evaluate the streams in the desired pattern and at the desired rate. However, 9 uses a separate runtime thread to evaluate each stream. is similar to z, as both of them evaluate the constituent streams fairly in a round robin fashion. However, the key difference is that ! is lazy or pull driven whereas  is strict or push driven. z immediately starts concurrent evaluation of both the streams (in separate threads) and later picks the results whereas  may wait for a certain latency threshold before initiating concurrent evaluation of the next stream. The concurrent scheduling of the next stream or the degree of concurrency is driven by the feedback from the consumer. In case of @ each stream is evaluated in a separate thread and results are pushedd to a shared output buffer, the evaluation rate is controlled by blocking when the buffer is full.@Concurrent lookahead streams are generally more efficient than  and can work pretty efficiently even for smaller tasks because they do not necessarily use a separate thread for each task. So they should be preferred over d especially when efficiency is a concern and simultaneous strict evaluation is not a requirement. ] is useful for cases when the streams are required to be evaluated simultaneously irrespective of how the consumer consumes them e.g. when we want to race two tasks and want to start both strictly at the same time or if we have timers in the parallel tasks and our results depend on the timers being started at the same time. We can say that A is almost the same (modulo some implementation differences) as \ when the latter is used with unlimited lookahead and zero latency in initiating lookahead. main = (toList . 5; $ (fromFoldable [1,2]) <> (fromFoldable [3,4])) >>= print   [1,3,2,4] zWhen streams with more than one element are merged, it yields whichever stream yields first without any bias, unlike the   style streams.Any exceptions generated by a constituent stream are propagated to the output stream. The output and exceptions from a single stream are guaranteed to arrive in the same order in the resulting stream as they were generated in the input stream. However, the relative ordering of elements from different streams in the resulting stream can vary depending on scheduling and generation delays.Similarly, the  instance of  runs all& iterations of the loop concurrently. import Streamly import qualified Streamly.Prelude( as S import Control.Concurrent main =  . 5 $ do n <- return 3 <> return 2 <> return 1 S.once $ do threadDelay (n * 1000000) myThreadId >>= \tid -> putStrLn (show tid ++ ": Delay " ++ show n)  ?ThreadId 40: Delay 1 ThreadId 39: Delay 2 ThreadId 38: Delay 3 Note that parallel composition can only combine a finite number of streams as it needs to retain state for each unfinished stream.eWide async composition or async composition with breadth first traversal. The Semigroup instance of  concurrently  traverses the composed streams using a depth first travesal or in a round robin fashion, yielding elements from both streams alternately. main = (toList . 4; $ (fromFoldable [1,2]) <> (fromFoldable [3,4])) >>= print   [1,3,2,4] Any exceptions generated by a constituent stream are propagated to the output stream. The output and exceptions from a single stream are guaranteed to arrive in the same order in the resulting stream as they were generated in the input stream. However, the relative ordering of elements from different streams in the resulting stream can vary depending on scheduling and generation delays.Similarly, the  instance of  runs all@ iterations fairly concurrently using a round robin scheduling. import Streamly import qualified Streamly.Prelude( as S import Control.Concurrent main =  . 4 $ do n <- return 3 <> return 2 <> return 1 S.once $ do threadDelay (n * 1000000) myThreadId >>= \tid -> putStrLn (show tid ++ ": Delay " ++ show n)  ?ThreadId 40: Delay 1 ThreadId 39: Delay 2 ThreadId 38: Delay 3 Unlike L all iterations are guaranteed to run fairly concurrently, unconditionally.Note that async composition with breadth first traversal can only combine a finite number of streams as it needs to retain state for each unfinished stream.\Deep async composition or async composition with depth first traversal. In a left to right  composition it tries to yield elements from the left stream as long as it can, but it can run the right stream in parallel if it needs to, based on demand. The right stream can be run if the left stream blocks on IO or cannot produce elements fast enough for the consumer. main = (toList . 3; $ (fromFoldable [1,2]) <> (fromFoldable [3,4])) >>= print   [1,2,3,4] Any exceptions generated by a constituent stream are propagated to the output stream. The output and exceptions from a single stream are guaranteed to arrive in the same order in the resulting stream as they were generated in the input stream. However, the relative ordering of elements from different streams in the resulting stream can vary depending on scheduling and generation delays.!Similarly, the monad instance of  may run each iteration concurrently based on demand. More concurrent iterations are started only if the previous iterations are not able to produce enough output for the consumer. import Streamly import qualified Streamly.Prelude( as S import Control.Concurrent main =  . 3 $ do n <- return 3 <> return 2 <> return 1 S.once $ do threadDelay (n * 1000000) myThreadId >>= \tid -> putStrLn (show tid ++ ": Delay " ++ show n)  ?ThreadId 40: Delay 1 ThreadId 39: Delay 2 ThreadId 38: Delay 3 ?All iterations may run in the same thread if they do not block.Note that async composition with depth first traversal can be used to combine infinite number of streams as it explores only a bounded number of streams at a time.fDeep ahead composition or ahead composition with depth first traversal. The semigroup composition of 4 appends streams in a depth first manner just like M except that it can produce elements concurrently ahead of time. It is like  except that , produces the output as it arrives whereas * orders the output in the traversal order. main = (toList . 2; $ (fromFoldable [1,2]) <> (fromFoldable [3,4])) >>= print   [1,2,3,4] VAny exceptions generated by a constituent stream are propagated to the output stream.!Similarly, the monad instance of c may run each iteration concurrently ahead of time but presents the results in the same order as . import Streamly import qualified Streamly.Prelude( as S import Control.Concurrent main =  . 2 $ do n <- return 3 <> return 2 <> return 1 S.once $ do threadDelay (n * 1000000) myThreadId >>= \tid -> putStrLn (show tid ++ ": Delay " ++ show n)  ?ThreadId 40: Delay 1 ThreadId 39: Delay 2 ThreadId 38: Delay 3 ?All iterations may run in the same thread if they do not block.Note that ahead composition with depth first traversal can be used to combine infinite number of streams as it explores only a bounded number of streams at a time.SWide serial composition or serial composition with a breadth first traversal. The  instance of  traverses the two streams in a breadth first manner. In other words, it interleaves two streams, yielding one element from each stream alternately. !import Streamly import qualified Streamly.Prelude as S main = (toList . 0; $ (fromFoldable [1,2]) <> (fromFoldable [3,4])) >>= print   [1,3,2,4] Similarly, the o instance interleaves the iterations of the inner and the outer loop, nesting loops in a breadth first manner. main =  . 0\ $ do x <- return 1 <> return 2 y <- return 3 <> return 4 S.once $ print (x, y)  (1,3) (2,3) (1,4) (2,4) Note that a serial composition with breadth first traversal can only combine a finite number of streams as it needs to retain state for each unfinished stream.ODeep serial composition or serial composition with depth first traversal. The  instance of  appends two streams serially in a depth first manner, yielding all elements from the first stream, and then all elements from the second stream. !import Streamly import qualified Streamly.Prelude as S main = (toList . /; $ (fromFoldable [1,2]) <> (fromFoldable [3,4])) >>= print   [1,2,3,4] The  instance runs the monadic continuation+ for each element of the stream, serially. main =  . /9 $ do x <- return 1 <> return 2 S.once $ print x  1 2 0 nests streams serially in a depth first manner. main =  . /\ $ do x <- return 1 <> return 2 y <- return 3 <> return 4 S.once $ print (x, y)  (1,3) (1,4) (2,3) (2,4) This behavior of  is exactly like a list transformer. We call the monadic code being run for each element of the stream a monadic continuation. In imperative paradigm we can think of this composition as nested foro loops and the monadic continuation is the body of the loop. The loop iterates for all elements of the stream.The /: combinator can be omitted as the default stream type is . Note that serial composition with depth first traversal can be used to combine an infinite number of streams as it explores only one stream at a time.Same as .DClass of types that can represent a stream of elements of some type a in some monad m._Constructs a stream by adding a monadic action at the head of an existing stream. For example: M> toList $ getLine `consM` getLine `consM` nil hello world ["hello","world"] Concurrent (do not use 5 to construct infinite streams)Operator equivalent of . We can read it as "parallel colon" to remember that | comes before :. C> toList $ getLine |: getLine |: nil hello world ["hello","world"]  let delay = threadDelay 1000000 >> print 1 runStream $ serially $ delay |: delay |: delay |: nil runStream $ parallely $ delay |: delay |: delay |: nil Concurrent (do not use 5 to construct infinite streams)An empty stream. > toList nil [] _Constructs a stream by adding a monadic action at the head of an existing stream. For example: M> toList $ getLine `consM` getLine `consM` nil hello world ["hello","world"] tConstruct a stream by adding a pure value at the head of an existing stream. For pure values it can be faster than . For example: 2> toList $ 1 `cons` 2 `cons` 3 `cons` nil [1,2,3] Operator equivalent of . &> toList $ 1 .: 2 .: 3 .: nil [1,2,3] 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 remaining stream if any otherwise 9. The third parameter is to represent an "empty" stream.2Build a singleton stream from a callback function.Read an SVar to get a stream.Fold a stream using its church encoding. The second argument is the "step" function consuming an element and the remaining stream, if any. The third argument is for consuming an "empty" stream that yields nothing.ZRun a streaming composition, discard the results. By default it interprets the stream as O, to run other types of streams use the type adapting combinators for example  runStream . 3. Same as Write a stream to an }Q in a non-blocking manner. The stream can then be read back from the SVar using .!EMake a stream asynchronous, triggers the computation and returns a stream in the underlying monad representing the output generated by the original computation. The returned action is exhaustible and must be drained once. If not drained fully we may have a thread blocked forever and once exhausted it will always return empty."iParallel function application operator for streams; just like the regular function application operator } except that it is concurrent. The following code prints a value every second even though each stage adds a 1 second delay. qrunStream $ S.mapM (\x -> threadDelay 1000000 >> print x) |$ S.repeatM (threadDelay 1000000 >> return 1)  Concurrent#yParallel reverse function application operator for streams; just like the regular reverse function application operator & except that it is concurrent. rrunStream $ S.repeatM (threadDelay 1000000 >> return 1) |& S.mapM (\x -> threadDelay 1000000 >> print x)  Concurrent$2Parallel function application operator; applies a run or fold_ function to a stream such that the fold consumer and the stream producer run in parallel. A run or foldF function reduces the stream to a value in the underlying monad. The .I at the end of the operator is a mnemonic for termination of the stream. o S.foldlM' (\_ a -> threadDelay 1000000 >> print a) () |$. S.repeatM (threadDelay 1000000 >> return 1)  Concurrent%lParallel reverse function application operator for applying a run or fold functions to a stream. Just like $' except that the operands are reversed. p S.repeatM (threadDelay 1000000 >> return 1) |&. S.foldlM' (\_ a -> threadDelay 1000000 >> print a) ()  Concurrent&Polymorphic version of the  operation  of . Appends two streams sequentially, yielding all elements from the first stream, and then all elements from the second stream.'Polymorphic version of the  operation  of N. Interleaves two streams, yielding one element from each stream alternately.(Same as '.)Polymorphic version of the  operation  of A. Merges two streams sequentially but with concurrent lookahead.*Polymorphic version of the  operation  of g. Merges two streams possibly concurrently, preferring the elements from the left one when available.+Same as *.,Polymorphic version of the  operation  of F. Merges two streams concurrently choosing elements from both fairly.-Polymorphic version of the  operation  of " Merges two streams concurrently..AAdapt any specific stream type to any other specific stream type./(Fix the type of a polymorphic stream as .0(Fix the type of a polymorphic stream as .1Same as 0.2(Fix the type of a polymorphic stream as .3(Fix the type of a polymorphic stream as .4(Fix the type of a polymorphic stream as .5(Fix the type of a polymorphic stream as .6(Fix the type of a polymorphic stream as .7Same as 6.8(Fix the type of a polymorphic stream as  .9Same as 8.:Same as  runStream.;Same as runStream . wSerially.<Same as runStream . asyncly.=Same as runStream . parallely.>Same as runStream . zipping.?Same as runStream . zippingAsync.@ A variant of  that allows you to fold a @ container of streams using the specified stream sum operation.  foldWith * $ map return [1..3]A A variant of 9 that allows you to map a monadic streaming action on a F container and then fold it using the specified stream sum operation.  foldMapWith * return [1..3]BLike Ad but with the last two arguments reversed i.e. the monadic streaming function is the last argument.F  !"#$%&'()*+,-./0123456789:;<=>?@AB  5555"0#1$0%1(5(c) 2017 Harendra KumarBSD3harendra.kumar@gmail.com experimentalGHCNone ;<=>?QV5CBuild a stream by unfolding a pure step function starting from a seed. The step function returns the next element in the stream and the next seed value. When it is done it returns # and the stream ends. For example, elet f b = if b > 3 then Nothing else Just (b, b + 1) in toList $ unfoldr f 0  [0,1,2,3] DBuild a stream by unfolding a monadic step function starting from a seed. The step function returns the next element in the stream and the next seed value. When it is done it returns # and the stream ends. For example, let f b = if b > 3 then return Nothing else print b >> return (Just (b, b + 1)) in runStream $ unfoldrM f 0   0 1 2 3 When run concurrently, the next unfold step can run concurrently with the processing of the output of the previous step. Note that more than one step cannot run concurrently as the next step depends on the output of the previous step. (asyncly $ S.unfoldrM (\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0) & S.foldlM' (\_ a -> threadDelay 1000000 >> print a) ()  Concurrent Since: 0.1.0EConstruct a stream from a  containing pure values.FConstruct a stream from a  containing monadic actions. runStream $ serially $ S.fromFoldableM $ replicate 10 (threadDelay 1000000 >> print 1) runStream $ asyncly $ S.fromFoldableM $ replicate 10 (threadDelay 1000000 >> print 1) Concurrent (do not use with 5 on infinite containers)GSame as E.HGCreate a singleton stream by executing a monadic action once. Same as  m `consM` nil but more efficient. (> toList $ once getLine hello ["hello"] I1Generate a stream by performing a monadic action n times. runStream $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1) runStream $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)  ConcurrentJCGenerate a stream by repeatedly executing a monadic action forever. runStream $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1) runStream $ asyncly $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1) &Concurrent, infinite (do not use with 5)KIIterate a pure function from a seed value, streaming the results forever.LMIterate a monadic function from a seed value, streaming the results forever.When run concurrently, the next iteration can run concurrently with the processing of the previous iteration. Note that more than one iteration cannot run concurrently as the next iteration depends on the output of the previous iteration. runStream $ serially $ S.take 10 $ S.iterateM (\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0 runStream $ asyncly $ S.take 10 $ S.iterateM (\x -> threadDelay 1000000 >> print x >> return (x + 1)) 0  ConcurrentM6Read lines from an IO Handle into a stream of Strings.NGLazy right associative fold. For example, to fold a stream into a list: H>> runIdentity $ foldr (:) [] (serially $ fromFoldable [1,2,3]) [1,2,3] OYLazy right fold with a monadic step function. For example, to fold a stream into a list: `>> runIdentity $ foldrM (\x xs -> return (x : xs)) [] (serially $ fromFoldable [1,2,3]) [1,2,3] P3Strict left scan with an extraction function. Like Ry, but applies a user supplied extraction function (the third argument) at each step. This is designed to work with the foldl library. The suffix x is a mnemonic for extraction.QRStrict left scan. Like U, but returns the folded value at each step, generating a stream of all intermediate fold results. The first element of the stream is the user supplied initial value, and the last element of the stream is the same as the result of U.SStrict left fold with an extraction function. Like the standard strict left fold, but applies a user supplied extraction function (the third argument) to the folded value at the end. This is designed to work with the foldl library. The suffix x is a mnemonic for extraction.TUStrict left associative fold.VLike S#, but with a monadic step function.WXLike U" but with a monadic step function.YLDecompose a stream into its head and tail. If the stream is empty, returns &. If the stream is non-empty, returns  Just (a, ma), where a is the head of the stream and ma its tail.Z*Write a stream of Strings to an IO Handle.[5Convert a stream into a list in the underlying monad.\ Take first n/ elements from the stream and discard the rest.]2Include only those elements that pass a predicate.^<End the stream as soon as the predicate fails on an element._Discard first n, elements from the stream and take the rest.`dDrop elements in the stream as long as the predicate succeeds and then take the rest of the stream.a?Determine whether all elements of a stream satisfy a predicate.bFDetermine whether any of the elements of a stream satisfy a predicate.c8Determine the sum of all elements of a stream of numbersd<Determine the product of all elements of a stream of numberse0Extract the first element of the stream, if any.f8Extract all but the first element of the stream, if any.g/Extract the last element of the stream, if any.h&Determine whether the stream is empty.i6Determine whether an element is present in the stream.j:Determine whether an element is not present in the stream.k#Determine the length of the stream.lPReturns the elements of the stream in reverse order. The stream must be finite.m*Determine the minimum element in a stream.n*Determine the maximum element in a stream.o_Replace each element of the stream with the result of a monadic action applied on the element. runStream $ S.replicateM 10 (return 1) & (serially . S.mapM (\x -> threadDelay 1000000 >> print x)) runStream $ S.replicateM 10 (return 1) & (asyncly . S.mapM (\x -> threadDelay 1000000 >> print x)) Concurrent (do not use with 5 on infinite streams)pMap a 0 returning function to a stream, filter out the 9 elements, and return a stream of values extracted from .qLike p but maps a monadic function.Concurrent (do not use with 5 on infinite streams)r[Apply a monadic action to each element of the stream and discard the output of the action.sOReduce a stream of monadic actions to a stream of the output of those actions. runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1) & (serially . S.sequence) runStream $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1) & (asyncly . S.sequence) Concurrent (do not use with 5 on infinite streams)t7Zip two streams serially using a pure zipping function.u:Zip two streams serially using a monadic zipping function.v}Zip two streams concurrently (i.e. both the elements being zipped are generated concurrently) using a pure zipping function.w{Zip two streams asyncly (i.e. both the elements being zipped are generated concurrently) using a monadic zipping function.:CDEFGHIJKLMNOPQRSTUVWXYZ[\]^_`abcdefghijklmnopqrstuvw:CDHIJKLEFYNOUXSVr[abefghkijnmcdRP]\^_`lopqstuvwMZGQTW(c) 2017 Harendra KumarBSD3harendra.kumar@gmail.com experimentalGHCNone>  !"#$%&'()*+,-./0123456789:;<=>?@AB? "#$%!&')*,-@AB/0324568.  :;<=>?179(+(c) 2017 Harendra KumarBSD3harendra.kumar@gmail.com experimentalGHCSafeaxXRun an action forever periodically at the given frequency specified in per second (Hz).yRun a computation on every clock tick, the clock runs at the specified frequency. It allows running a computation at high frequency efficiently by maintaining a local clock and adjusting it with the provided base clock at longer intervals. The first argument is a base clock returning some notion of time in microseconds. The second argument is the frequency in per second (Hz). The third argument is the action to run, the action is provided the local time as an argument.xyxy(c) 2017 Harendra KumarBSD3harendra.kumar@gmail.comNonex   !"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\]^_`abcdefghijklmnopqrstuvwxyz{|}~ (185*'S%247     "  $streamly-0.3.0-7aqAHzs9TO4vIK3QJZx9xStreamlyStreamly.Prelude Streamly.Time Streamly.CoreStreamly.Streams Data.FoldablefoldStreamly.TutorialbaseData.Semigroup<> Semigroupstimessconcat MonadAsyncZipAsync ZipSerialParallelWAsyncAsyncAheadWSerialSerial ZipAsyncM ZipStream ZipSerialM ParallelTWAsyncTAsyncTAheadT InterleavedTWSerialTStreamTSerialT StreamingIsStreamconsM|:nilcons.: runStream runStreamingmkAsync|$|&|$.|&.serialwSerial<=>aheadasync<|wAsyncparalleladaptserially wSerially interleavingaheadlyasynclywAsyncly parallely zipSeriallyzipping zipAsyncly zippingAsync runStreamTrunInterleavedT runAsyncT runParallelT runZipStream runZipAsyncfoldWith foldMapWith forEachWithunfoldrunfoldrM fromFoldable fromFoldableMeachonce replicateMrepeatMiterateiterateM fromHandlefoldrfoldrMscanxscanscanl'foldxfoldlfoldl'foldxMfoldlMfoldlM'unconstoHandletoListtakefilter takeWhiledrop dropWhileallanysumproductheadtaillastnullelemnotElemlengthreverseminimummaximummapMmapMaybe mapMaybeMmapM_sequencezipWithzipWithM zipAsyncWith zipAsyncWithMperiodic withClockStreamGHC.Basepuretransformers-0.5.2.0Control.Monad.Trans.ClassliftSVar fromStreamVarjoinStreamVarAsync SVarStyleAheadHeapEntry ChildEvent singletonsendallThreadsDonerunningThreads pushWorkerPar newEmptySVar newStreamVar1newStreamVarAhead toStreamVar forkSVarAsync consMAhead consMAsync consMWAsync consMParallelAsyncVar WAsyncVar ParallelVarAheadVarrepeat applyWithrunWith svarStyle outputQueuedoorBell outputHeap workQueueenqueue queueEmptywaitingForWorkrunqueue activeWorkersAheadEntryPureAheadEntryStream ChildYield ChildStopMonad consMSerial streamBuildNothing fromCallbackfromSVar streamFoldtoSVar$FoldablefoldMaptoStream fromStream getZipAsyncM getZipSerialM getParallelT getWAsyncT getAsyncT getAheadT getWSerialT getSerialTMaybeJust