@@ -1,8 +1,6 @@
1{-# LANGUAGE ConstraintKinds #-}
1{-# LANGUAGE ConstraintKinds #-}
2−{-# LANGUAGE EmptyCase #-}
3{-# LANGUAGE FlexibleContexts #-}
2{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
3{-# LANGUAGE FlexibleInstances #-}
5−{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE MultiParamTypeClasses #-}
4{-# LANGUAGE MultiParamTypeClasses #-}
7{-# LANGUAGE RankNTypes #-}
5{-# LANGUAGE RankNTypes #-}
8{-# LANGUAGE UndecidableInstances #-} -- XXX
6{-# LANGUAGE UndecidableInstances #-} -- XXX
@@ -22,19 +20,7 @@ module Asyncly.AsyncT
22 , MonadAsync
20 , MonadAsync
23 , runAsyncly
21 , runAsyncly
39 , interleave
25 , interleave
@@ -79,10 +65,9 @@ import Data.IORef (IORef, modifyIORef, newIORef,
79import Data.Atomics (atomicModifyIORefCAS,
65import Data.Atomics (atomicModifyIORefCAS,
80 atomicModifyIORefCAS_)
66 atomicModifyIORefCAS_)
81import Data.Maybe (isNothing)
67import Data.Maybe (isNothing)
82−import Data.Semigroup (Semigroup(..), cycle1)
68+import Data.Semigroup (Semigroup(..))
83import Data.Set (Set)
69import Data.Set (Set)
84import qualified Data.Set as S
70import qualified Data.Set as S
85−import Prelude hiding (take, drop, zipWith)
87------------------------------------------------------------------------------
72------------------------------------------------------------------------------
88-- Concurrency Semantics
73-- Concurrency Semantics
@@ -591,6 +576,42 @@ instance MonadAsync m => MonadPlus (AsyncT m) where
591 mzero = empty
576 mzero = empty
592 mplus = (<|>)
577 mplus = (<|>)
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
585+ CtxType _ FIFO -> do
586+ c <- getFifoCtx ctype
588+ CtxType _ LIFO -> do
589+ c <- getLifoCtx ctype
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
594+ liftIO $ (enqueue ctx) m
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.
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.
610+async :: MonadAsync m => AsyncT m a -> m (AsyncT m a)
612+ ctx <- pushOneToCtx (CtxType Disjunction LIFO) m
613+ return $ pullFromCtx ctx
594------------------------------------------------------------------------------
615------------------------------------------------------------------------------
596------------------------------------------------------------------------------
617------------------------------------------------------------------------------
@@ -707,159 +728,6 @@ toList m = (runAsyncT m) Nothing stop yield
707 yield a Nothing = return [a]
728 yield a Nothing = return [a]
708 yield a (Just x) = liftM (a :) (toList x)
729 yield a (Just x) = liftM (a :) (toList x)
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
718− stop = return Nothing
720− {-# INLINE yield #-}
721− yield a Nothing = return (Just (a, empty))
722− yield a (Just x) = return (Just (a, x))
724−-- | Build a Stream by unfolding steps starting from a seed.
725−unfoldr :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> AsyncT m a
728− go s = AsyncT $ \_ stp yld -> do
732− Just (a, b) -> yld a (Just (go b))
734−------------------------------------------------------------------------------
736−------------------------------------------------------------------------------
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))
745− else (runAsyncT m) ctx stp yield
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
753− then (runAsyncT m) ctx stp yld
754− else (runAsyncT m) ctx stp yield
756−------------------------------------------------------------------------------
758−------------------------------------------------------------------------------
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
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
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
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
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.
789−-- > f <$> ZipSerial xs1 <*> ... <*> ZipSerial xsN
790−newtype ZipSerial m a = ZipSerial {getZipSerial :: AsyncT m a}
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)
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
803− CtxType _ FIFO -> do
804− c <- getFifoCtx ctype
806− CtxType _ LIFO -> do
807− c <- getLifoCtx ctype
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
812− liftIO $ (enqueue ctx) m
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.
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.
828−async :: MonadAsync m => AsyncT m a -> m (AsyncT m a)
830− ctx <- pushOneToCtx (CtxType Disjunction LIFO) m
831− return $ pullFromCtx ctx
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
840− (runAsyncT (zipWithM f ma mb)) Nothing stp yld
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
849− (runAsyncT (zipWith f ma mb)) Nothing stp yld
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.
855−-- > f <$> ZipAsync xs1 <*> ... <*> ZipAsync xsN
856−newtype ZipAsync m a = ZipAsync {getZipAsync :: AsyncT m a}
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)
863------------------------------------------------------------------------------
731------------------------------------------------------------------------------
864-- Utilities
732-- Utilities
865------------------------------------------------------------------------------
733------------------------------------------------------------------------------