git.iko.soy/streamly
on master help sign in
Reorganize code top to bottom in dependency order
d64225fHarendra Kumar
src/Asyncly/AsyncT.hs+102−101
compare with master
unifiedsplit
@@ -29,6 +29,7 @@ module Asyncly.AsyncT
29 , Location(..)
29 , Location(..)
30 , getLocation
30 , getLocation
31 , setLocation
31 , setLocation
32+
32 , threads
33 , threads
33 , waitForChildren
34 , waitForChildren
34 , handleResult
35 , handleResult
@@ -195,6 +196,36 @@ dbg _ = return ()
195
196
196type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
197type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
197
198
199+------------------------------------------------------------------------------
200+-- Monad
201+------------------------------------------------------------------------------
202+
203+instance MonadAsync m => Monad (AsyncT m) where
204+ return a = AsyncT . return $ Just a
205+ m >>= f = AsyncT $ do
206+ modify $ \Context { continuation = g, .. } ->
207+ let k x = f x >>= unsafeCoerce g
208+ in Context { continuation = unsafeCoerce k, .. }
209+
210+ x <- runAsyncT m
211+ Context { continuation = g } <- get
212+ runAsyncT $ maybe mzero (unsafeCoerce g) x
213+
214+------------------------------------------------------------------------------
215+-- Functor
216+------------------------------------------------------------------------------
217+
218+instance MonadAsync m => Functor (AsyncT m) where
219+ fmap = liftM
220+
221+------------------------------------------------------------------------------
222+-- Applicative
223+------------------------------------------------------------------------------
224+
225+instance MonadAsync m => Applicative (AsyncT m) where
226+ pure = return
227+ (<*>) = ap
228+
198------------------------------------------------------------------------------
229------------------------------------------------------------------------------
199-- Pick up from where we left in the previous thread
230-- Pick up from where we left in the previous thread
200------------------------------------------------------------------------------
231------------------------------------------------------------------------------
@@ -478,47 +509,6 @@ makeAsync cbsetter = AsyncT $ do
478 return ()
509 return ()
479 spawningParentDone
510 spawningParentDone
480
511
481−-- scatter
482−each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
483− => [a] -> AsyncT m a
484−each xs = foldl (<|>) empty $ map return xs
485−
486−------------------------------------------------------------------------------
487−-- Controlling thread quota
488−------------------------------------------------------------------------------
489−
490−-- XXX Should n be Word32 instead?
491−-- | Runs a computation under a given thread limit. A limit of 0 means all new
492−-- tasks start synchronously in the current thread unless overridden by
493−-- 'async'.
494−threads :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
495−threads n process = AsyncT $ do
496− oldCr <- gets threadCredit
497− newCr <- liftIO $ newIORef n
498− modify $ \s -> s { threadCredit = newCr }
499− r <- runAsyncT $ process
500− >>* (AsyncT $ do
501− modify $ \s -> s { threadCredit = oldCr }
502− return (Just ())
503− ) -- restore old credit
504− return r
505−
506−------------------------------------------------------------------------------
507−-- Functor
508−------------------------------------------------------------------------------
509−
510−instance MonadAsync m => Functor (AsyncT m) where
511− fmap = liftM
512−
513−------------------------------------------------------------------------------
514−-- Applicative
515−------------------------------------------------------------------------------
516−
517−
518−instance MonadAsync m => Applicative (AsyncT m) where
519− pure a = AsyncT . return $ Just a
520− (<*>) = ap
521−
522------------------------------------------------------------------------------
512------------------------------------------------------------------------------
523-- Alternative
513-- Alternative
524------------------------------------------------------------------------------
514------------------------------------------------------------------------------
@@ -532,21 +522,6 @@ instance MonadAsync m => Alternative (AsyncT m) where
532 RemoteNode -> return Nothing
522 RemoteNode -> return Nothing
533 _ -> maybe (runAsyncT m2) (return . Just) r
523 _ -> maybe (runAsyncT m2) (return . Just) r
534
524
535−------------------------------------------------------------------------------
536−-- Monad
537−------------------------------------------------------------------------------
538−
539−instance MonadAsync m => Monad (AsyncT m) where
540− return = pure
541− m >>= f = AsyncT $ do
542− modify $ \Context { continuation = g, .. } ->
543− let k x = f x >>= unsafeCoerce g
544− in Context { continuation = unsafeCoerce k, .. }
545−
546− x <- runAsyncT m
547− Context { continuation = g } <- get
548− runAsyncT $ maybe mzero (unsafeCoerce g) x
549−
550instance MonadAsync m => MonadPlus (AsyncT m) where
525instance MonadAsync m => MonadPlus (AsyncT m) where
551 mzero = empty
526 mzero = empty
552 mplus = (<|>)
527 mplus = (<|>)
@@ -555,6 +530,11 @@ instance (Monoid a, MonadAsync m) => Monoid (AsyncT m a) where
555 mappend x y = mappend <$> x <*> y
530 mappend x y = mappend <$> x <*> y
556 mempty = return mempty
531 mempty = return mempty
557
532
533+-- scatter
534+each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
535+ => [a] -> AsyncT m a
536+each xs = foldl (<|>) empty $ map return xs
537+
558------------------------------------------------------------------------------
538------------------------------------------------------------------------------
559-- Num
539-- Num
560------------------------------------------------------------------------------
540------------------------------------------------------------------------------
@@ -567,48 +547,6 @@ instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where
567 abs f = f >>= return . abs
547 abs f = f >>= return . abs
568 signum f = f >>= return . signum
548 signum f = f >>= return . signum
569
549
570−------------------------------------------------------------------------------
571−-- Special compositions
572−------------------------------------------------------------------------------
573−
574−infixr 1 >>*, *>>, >>|
575−
576−-- XXX This can be moved to utility functions as it is purely app level
577−-- | Run @m b@ right after the first event in @m a@ is generated but before it
578−-- is yielded. The result of @m b@ is discarded.
579−afterfirst :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
580−afterfirst ma mb = do
581− ref <- liftIO $ newIORef False
582− x <- ma
583− done <- liftIO $ readIORef ref
584− when (not done) $ (liftIO $ writeIORef ref True) >>* mb
585− return x
586−
587−-- | Same as 'afterfirst'.
588−(>>|) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
589−(>>|) = afterfirst
590−
591−-- | Run @m a@ before running @m b@. The result of @m a@ is discarded.
592−before :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b
593−before ma mb = AsyncT $ do
594− _ <- runAsyncT (ma >> mzero)
595− runAsyncT mb
596−
597−-- | Same as 'before'.
598−(*>>) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b
599−(*>>) = before
600−
601−-- | Run @m b@ after running @m a@. The result of @m b@ is discarded.
602−thereafter :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
603−thereafter ma mb = AsyncT $ do
604− a <- runAsyncT ma
605− _ <- runAsyncT (mb >> mzero)
606− return a
607−
608−-- | Same as 'thereafter'.
609−(>>*) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
610−(>>*) = thereafter
611−
612-------------------------------------------------------------------------------
550-------------------------------------------------------------------------------
613-- AsyncT transformer
551-- AsyncT transformer
614-------------------------------------------------------------------------------
552-------------------------------------------------------------------------------
@@ -616,6 +554,9 @@ thereafter ma mb = AsyncT $ do
616instance MonadTrans AsyncT where
554instance MonadTrans AsyncT where
617 lift mx = AsyncT $ lift mx >>= return . Just
555 lift mx = AsyncT $ lift mx >>= return . Just
618
556
557+instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m) where
558+ liftBase = liftBaseDefault
559+
619-------------------------------------------------------------------------------
560-------------------------------------------------------------------------------
620-- monad-control
561-- monad-control
621-------------------------------------------------------------------------------
562-------------------------------------------------------------------------------
@@ -629,9 +570,6 @@ instance MonadTransControl AsyncT where
629 {-# INLINABLE liftWith #-}
570 {-# INLINABLE liftWith #-}
630 {-# INLINABLE restoreT #-}
571 {-# INLINABLE restoreT #-}
631
572
632−instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m) where
633− liftBase = liftBaseDefault
634−
635instance (MonadBaseControl b m, MonadAsync m) => MonadBaseControl b (AsyncT m) where
573instance (MonadBaseControl b m, MonadAsync m) => MonadBaseControl b (AsyncT m) where
636 type StM (AsyncT m) a = ComposeSt AsyncT m a
574 type StM (AsyncT m) a = ComposeSt AsyncT m a
637 liftBaseWith = defaultLiftBaseWith
575 liftBaseWith = defaultLiftBaseWith
@@ -657,3 +595,66 @@ instance (MonadAsync m, MonadRecorder m) => MonadRecorder (AsyncT m) where
657 getJournal = lift getJournal
595 getJournal = lift getJournal
658 putJournal = lift . putJournal
596 putJournal = lift . putJournal
659 play = lift . play
597 play = lift . play
598+
599+------------------------------------------------------------------------------
600+-- Special compositions
601+------------------------------------------------------------------------------
602+
603+-- | Run @m a@ before running @m b@. The result of @m a@ is discarded.
604+before :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b
605+before ma mb = AsyncT $ do
606+ _ <- runAsyncT (ma >> mzero)
607+ runAsyncT mb
608+
609+infixr 1 *>>
610+-- | Same as 'before'.
611+(*>>) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b
612+(*>>) = before
613+
614+-- | Run @m b@ after running @m a@. The result of @m b@ is discarded.
615+thereafter :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
616+thereafter ma mb = AsyncT $ do
617+ a <- runAsyncT ma
618+ _ <- runAsyncT (mb >> mzero)
619+ return a
620+
621+infixr 1 >>*
622+-- | Same as 'thereafter'.
623+(>>*) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
624+(>>*) = thereafter
625+
626+-- XXX This can be moved to utility functions as it is purely app level
627+-- | Run @m b@ right after the first event in @m a@ is generated but before it
628+-- is yielded. The result of @m b@ is discarded.
629+afterfirst :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
630+afterfirst ma mb = do
631+ ref <- liftIO $ newIORef False
632+ x <- ma
633+ done <- liftIO $ readIORef ref
634+ when (not done) $ (liftIO $ writeIORef ref True) >>* mb
635+ return x
636+
637+infixr 1 >>|
638+-- | Same as 'afterfirst'.
639+(>>|) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
640+(>>|) = afterfirst
641+
642+------------------------------------------------------------------------------
643+-- Controlling thread quota
644+------------------------------------------------------------------------------
645+
646+-- XXX Should n be Word32 instead?
647+-- | Runs a computation under a given thread limit. A limit of 0 means all new
648+-- tasks start synchronously in the current thread unless overridden by
649+-- 'async'.
650+threads :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
651+threads n process = AsyncT $ do
652+ oldCr <- gets threadCredit
653+ newCr <- liftIO $ newIORef n
654+ modify $ \s -> s { threadCredit = newCr }
655+ r <- runAsyncT $ process
656+ >>* (AsyncT $ do
657+ modify $ \s -> s { threadCredit = oldCr }
658+ return (Just ())
659+ ) -- restore old credit
660+ return r