@@ -808,3 +808,87 @@ tapCount ::
808 -> Stream m a
808 -> Stream m a
809 -> Stream m a
809 -> Stream m a
810tapCount = parTapCount
810tapCount = parTapCount
812+-------------------------------------------------------------------------------
814+-------------------------------------------------------------------------------
816+-- Clone a stream into n streams, perform some processing on them and then zip
817+-- or merge the results in different ways?
819+-- For serial processing combining n scans into a single scan on the source
820+-- stream would be the most efficient way of doing this. But this has a
821+-- limitation that we process one element at a time through the combined scan.
822+-- This is a way to combine the states of different scans in a modular way.
823+-- This works well for cases where each scan consumes and produces one element
824+-- at a time. If different scans produce elements by consuming different number
825+-- of elements then this may become complicated, inconvenient to use.
827+-- It does not make much sense to clone a stream to multiple free streams
828+-- unless we enforce processing those streams in independent threads. If
829+-- we are anyway running them in the same thread then there is not much point
830+-- of cloning, we can just map a function on the stream to do multiple tasks in
833+-- Cloning a stream to multiple free streams can provide independent buffering
834+-- and speed of evaluation to each cloned stream pipeline. For example, we can
835+-- parseBreak each stream independently using a different parser. The
836+-- evaluation would be push driven. The source stream would be evaluated in a
837+-- separate thread and we would push the generated elements to all the cloned
840+-- 1. If the cloned streams have infinite buffers then this can lead to the
841+-- source stream getting evaluated faster than consumers and buffering the
842+-- entire source stream in cloned streams.
844+-- 2. If the cloned streams have limited buffers, then they will all go at the
845+-- speed of the slowest stream if they are run concurrently.
847+-- 3. If the cloned streams have limited buffers and are evaluated serially
848+-- then we may run into deadlock if we are deep evaluating one stream and the
849+-- source gets blocked because other stream's buffer are full.
851+-- This is somewhat like list sharing. And it will have the same space leak
852+-- issues if used incorrectly. In fact, we can evaluate the source stream to
853+-- generate a lazy list using unsafePerformIO and share that lazy list among
854+-- multiple consumers. The evaluation of the list would drive the stream. And
855+-- the list would be naturally shared across consumers which can use different
856+-- buffering. This would be more like the lazy IO model. However, it may be
857+-- better to use streams instead of lists because streams use a monad and lists
858+-- are pure - pure lists can lead to the same issues as lazy IO when used in
861+-- Therefore, for safety, it makes better sense to use consumers (Stream m a ->
862+-- m b) rather than generating free streams as results. Each such consumer can
863+-- be enforced to run in its own thread. We can also pass a result collector
864+-- callback in a ReaderT env to collect the results from all these consumers
865+-- into a single stream.
867+-- parTap -- tap the stream to a single consumer
868+-- parDistribute -- a finite list of consumers is specified. all consumers
869+-- are guaranteed to get the entire stream from beginning. Run each consumer in
870+-- a separate thread.
871+-- parDistributeStream -- consumers can join the distribution channel
872+-- dynamically, they will get the source stream from now onwards.
874+-- XXX We could use Stream or StreamK, what are the pros and cons? The StreamK
875+-- version can be used to implement parDistribute using foldr?
877+{-# INLINE parTap #-}
878+parTap :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> Stream m a
879+parTap f m = undefined
881+-- Can we just use a parEval fold in tap?
882+-- We can easily convert the Fold to "Stream m a -> m b" form. Check if this
883+-- provides the same perf as above.
884+{-# INLINE parTap #-}
885+parTap :: MonadAsync m => Fold m a b -> Stream m a -> Stream m a
886+parTap f xs = undefined
888+-- Can we just use a parallel distribute fold in tap?
889+-- Maybe better to use a custom impl of distribute?
890+{-# INLINE parDistribute #-}
891+parDistribute :: (Foldable f, , MonadAsync m)
892+ => f (Stream m a -> m b) -> Stream m a -> Stream m a
893+parDistribute = flip (Prelude.foldr parTap)