@@ -1,8 +1,6 @@
11{-# LANGUAGE ConstraintKinds #-}
2−{-# LANGUAGE EmptyCase #-}
32{-# LANGUAGE FlexibleContexts #-}
43{-# LANGUAGE FlexibleInstances #-}
5−{-# LANGUAGE GeneralizedNewtypeDeriving#-}
64{-# LANGUAGE MultiParamTypeClasses #-}
75{-# LANGUAGE RankNTypes #-}
86{-# LANGUAGE UndecidableInstances #-} -- XXX
@@ -22,19 +20,7 @@ module Asyncly.AsyncT
2220 , MonadAsync
2321 , runAsyncly
2422 , toList
25− , uncons
2623 , async
27− , unfoldr
28−
29− , take
30− , drop
31−
32− , zipWithM
33− , zipWith
34− , ZipSerial (..)
35− , zipAsyncWithM
36− , zipAsyncWith
37− , ZipAsync (..)
3824
3925 , interleave
4026 , (<=>)
@@ -79,10 +65,9 @@ import Data.IORef (IORef, modifyIORef, newIORef,
7965import Data.Atomics (atomicModifyIORefCAS,
8066 atomicModifyIORefCAS_)
8167import Data.Maybe (isNothing)
82−import Data.Semigroup (Semigroup(..), cycle1)
68+import Data.Semigroup (Semigroup(..))
8369import Data.Set (Set)
8470import qualified Data.Set as S
85−import Prelude hiding (take, drop, zipWith)
8671
8772------------------------------------------------------------------------------
8873-- Concurrency Semantics
@@ -591,6 +576,42 @@ instance MonadAsync m => MonadPlus (AsyncT m) where
591576 mzero = empty
592577 mplus = (<|>)
593578
579+-- | The computation specified in the argument is pushed to a new thread and
580+-- the context is computation pulls output from the thread.
581+pushOneToCtx :: MonadAsync m => CtxType -> AsyncT m a -> m (Context m a)
582+pushOneToCtx ctype m = do
583+ ctx <- liftIO $
584+ case ctype of
585+ CtxType _ FIFO -> do
586+ c <- getFifoCtx ctype
587+ return c
588+ CtxType _ LIFO -> do
589+ c <- getLifoCtx ctype
590+ return c
591+ -- Note: We must have all the work on the queue before sending the
592+ -- pushworker, otherwise the pushworker may exit before we even get a
593+ -- chance to push.
594+ liftIO $ (enqueue ctx) m
595+ pushWorker ctx
596+ return ctx
597+
598+-- XXX The async API is useful for exploring each stream arbitrarily when
599+-- zipping or merging two streams. We can use a newtype wrapper with a monad
600+-- instance that composes like regular streaming libraries to facilitate linear
601+-- composition. We will also need a yield API for that.
602+
603+-- | Run a computation asynchronously, triggers the computation and returns
604+-- another computation (i.e. a promise) that when executed produces the output
605+-- from the original computation. Note that the returned action must be
606+-- executed exactly once and drained completely. If not executed or not drained
607+-- fully we will may have a thread blocked forever and if executed more than
608+-- once a ContextUsedAfterEOF exception will be raised.
609+
610+async :: MonadAsync m => AsyncT m a -> m (AsyncT m a)
611+async m = do
612+ ctx <- pushOneToCtx (CtxType Disjunction LIFO) m
613+ return $ pullFromCtx ctx
614+
594615------------------------------------------------------------------------------
595616-- Num
596617------------------------------------------------------------------------------
@@ -707,159 +728,6 @@ toList m = (runAsyncT m) Nothing stop yield
707728 yield a Nothing = return [a]
708729 yield a (Just x) = liftM (a :) (toList x)
709730
710−-- | Decompose a stream into its head and tail. If the stream is empty, returns
711−-- 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)', where 'a' is
712−-- the head of the stream and 'ma' its tail.
713−uncons :: MonadAsync m => AsyncT m a -> m (Maybe (a, AsyncT m a))
714−uncons m = (runAsyncT m) Nothing stop yield
715−
716− where
717−
718− stop = return Nothing
719−
720− {-# INLINE yield #-}
721− yield a Nothing = return (Just (a, empty))
722− yield a (Just x) = return (Just (a, x))
723−
724−-- | Build a Stream by unfolding steps starting from a seed.
725−unfoldr :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> AsyncT m a
726−unfoldr step = go
727− where
728− go s = AsyncT $ \_ stp yld -> do
729− mayb <- step s
730− case mayb of
731− Nothing -> stp
732− Just (a, b) -> yld a (Just (go b))
733−
734−------------------------------------------------------------------------------
735−-- Transformation
736−------------------------------------------------------------------------------
737−
738−-- | Take first 'n' elements from the stream and discard the rest.
739−take :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
740−take n m = AsyncT $ \ctx stp yld -> do
741− let yield a Nothing = yld a Nothing
742− yield a (Just x) = yld a (Just (take (n - 1) x))
743− if (n <= 0)
744− then stp
745− else (runAsyncT m) ctx stp yield
746−
747−-- | Discard first 'n' elements from the stream and take the rest.
748−drop :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
749−drop n m = AsyncT $ \ctx stp yld -> do
750− let yield _ Nothing = stp
751− yield _ (Just x) = (runAsyncT $ drop (n - 1) x) ctx stp yld
752− if (n <= 0)
753− then (runAsyncT m) ctx stp yld
754− else (runAsyncT m) ctx stp yield
755−
756−------------------------------------------------------------------------------
757−-- Zipping Streams
758−------------------------------------------------------------------------------
759−
760−-- | Zip two AsyncT streams serially using a monadic zipping function.
761−zipWithM :: MonadAsync m =>
762− (a -> b -> AsyncT m c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
763−zipWithM f m1 m2 = AsyncT $ \_ stp yld -> do
764− let merge a ra =
765− let yield2 b Nothing = (runAsyncT (f a b)) Nothing stp yld
766− yield2 b (Just rb) =
767− (runAsyncT ((f a b) <> (zipWithM f ra rb))) Nothing stp yld
768− in (runAsyncT m2) Nothing stp yield2
769− let yield1 a Nothing = merge a empty
770− yield1 a (Just ra) = merge a ra
771− (runAsyncT m1) Nothing stp yield1
772−
773−-- | Zip two AsyncT streams serially using a pure zipping function.
774−zipWith :: MonadAsync m
775− => (a -> b -> c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
776−zipWith f m1 m2 = AsyncT $ \_ stp yld -> do
777− let merge a ra =
778− let yield2 b Nothing = yld (f a b) Nothing
779− yield2 b (Just rb) = yld (f a b) (Just (zipWith f ra rb))
780− in (runAsyncT m2) Nothing stp yield2
781− let yield1 a Nothing = merge a empty
782− yield1 a (Just ra) = merge a ra
783− (runAsyncT m1) Nothing stp yield1
784−
785−-- | Wrapper around AsyncT type with a serial zipping Applicative instance.
786−-- Note that the binary function interleave (\<=>) is a special case of
787−-- ZipSerial Applicative.
788−--
789−-- > f <$> ZipSerial xs1 <*> ... <*> ZipSerial xsN
790−newtype ZipSerial m a = ZipSerial {getZipSerial :: AsyncT m a}
791− deriving (Functor)
792−
793−instance MonadAsync m => Applicative (ZipSerial m) where
794− pure a = ZipSerial $ cycle1 (pure a)
795− (ZipSerial xs) <*> (ZipSerial ys) = ZipSerial (zipWith id xs ys)
796−
797−-- | The computation specified in the argument is pushed to a new thread and
798−-- the context is computation pulls output from the thread.
799−pushOneToCtx :: MonadAsync m => CtxType -> AsyncT m a -> m (Context m a)
800−pushOneToCtx ctype m = do
801− ctx <- liftIO $
802− case ctype of
803− CtxType _ FIFO -> do
804− c <- getFifoCtx ctype
805− return c
806− CtxType _ LIFO -> do
807− c <- getLifoCtx ctype
808− return c
809− -- Note: We must have all the work on the queue before sending the
810− -- pushworker, otherwise the pushworker may exit before we even get a
811− -- chance to push.
812− liftIO $ (enqueue ctx) m
813− pushWorker ctx
814− return ctx
815−
816−-- XXX The async API is useful for exploring each stream arbitrarily when
817−-- zipping or merging two streams. We can use a newtype wrapper with a monad
818−-- instance that composes like regular streaming libraries to facilitate linear
819−-- composition. We will also need a yield API for that.
820−
821−-- | Run a computation asynchronously, triggers the computation and returns
822−-- another computation (i.e. a promise) that when executed produces the output
823−-- from the original computation. Note that the returned action must be
824−-- executed exactly once and drained completely. If not executed or not drained
825−-- fully we will may have a thread blocked forever and if executed more than
826−-- once a ContextUsedAfterEOF exception will be raised.
827−
828−async :: MonadAsync m => AsyncT m a -> m (AsyncT m a)
829−async m = do
830− ctx <- pushOneToCtx (CtxType Disjunction LIFO) m
831− return $ pullFromCtx ctx
832−
833−-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
834−-- concurrently) using a monadic zipping function.
835−zipAsyncWithM :: MonadAsync m
836− => (a -> b -> AsyncT m c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
837−zipAsyncWithM f m1 m2 = AsyncT $ \_ stp yld -> do
838− ma <- async m1
839− mb <- async m2
840− (runAsyncT (zipWithM f ma mb)) Nothing stp yld
841−
842−-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
843−-- concurrently) using a pure zipping function.
844−zipAsyncWith :: MonadAsync m
845− => (a -> b -> c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
846−zipAsyncWith f m1 m2 = AsyncT $ \_ stp yld -> do
847− ma <- async m1
848− mb <- async m2
849− (runAsyncT (zipWith f ma mb)) Nothing stp yld
850−
851−-- | Wrapper around AsyncT type with a parallel zipping Applicative instance.
852−-- Note that the binary operator (\<|>) from the Alternative instance of AsyncT
853−-- is a special case of ZipAsync Applicative.
854−--
855−-- > f <$> ZipAsync xs1 <*> ... <*> ZipAsync xsN
856−newtype ZipAsync m a = ZipAsync {getZipAsync :: AsyncT m a}
857− deriving (Functor)
858−
859−instance MonadAsync m => Applicative (ZipAsync m) where
860− pure a = ZipAsync $ cycle1 (pure a)
861− (ZipAsync xs) <*> (ZipAsync ys) = ZipAsync (zipAsyncWith id xs ys)
862−
863731------------------------------------------------------------------------------
864732-- Utilities
865733------------------------------------------------------------------------------