@@ -29,6 +29,7 @@ module Asyncly.AsyncT
2929 , Location(..)
3030 , getLocation
3131 , setLocation
32+
3233 , threads
3334 , waitForChildren
3435 , handleResult
@@ -195,6 +196,36 @@ dbg _ = return ()
195196
196197type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
197198
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+
198229------------------------------------------------------------------------------
199230-- Pick up from where we left in the previous thread
200231------------------------------------------------------------------------------
@@ -478,47 +509,6 @@ makeAsync cbsetter = AsyncT $ do
478509 return ()
479510 spawningParentDone
480511
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−
522512------------------------------------------------------------------------------
523513-- Alternative
524514------------------------------------------------------------------------------
@@ -532,21 +522,6 @@ instance MonadAsync m => Alternative (AsyncT m) where
532522 RemoteNode -> return Nothing
533523 _ -> maybe (runAsyncT m2) (return . Just) r
534524
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−
550525instance MonadAsync m => MonadPlus (AsyncT m) where
551526 mzero = empty
552527 mplus = (<|>)
@@ -555,6 +530,11 @@ instance (Monoid a, MonadAsync m) => Monoid (AsyncT m a) where
555530 mappend x y = mappend <$> x <*> y
556531 mempty = return mempty
557532
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+
558538------------------------------------------------------------------------------
559539-- Num
560540------------------------------------------------------------------------------
@@ -567,48 +547,6 @@ instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where
567547 abs f = f >>= return . abs
568548 signum f = f >>= return . signum
569549
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−
612550-------------------------------------------------------------------------------
613551-- AsyncT transformer
614552-------------------------------------------------------------------------------
@@ -616,6 +554,9 @@ thereafter ma mb = AsyncT $ do
616554instance MonadTrans AsyncT where
617555 lift mx = AsyncT $ lift mx >>= return . Just
618556
557+instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m) where
558+ liftBase = liftBaseDefault
559+
619560-------------------------------------------------------------------------------
620561-- monad-control
621562-------------------------------------------------------------------------------
@@ -629,9 +570,6 @@ instance MonadTransControl AsyncT where
629570 {-# INLINABLE liftWith #-}
630571 {-# INLINABLE restoreT #-}
631572
632−instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m) where
633− liftBase = liftBaseDefault
634−
635573instance (MonadBaseControl b m, MonadAsync m) => MonadBaseControl b (AsyncT m) where
636574 type StM (AsyncT m) a = ComposeSt AsyncT m a
637575 liftBaseWith = defaultLiftBaseWith
@@ -657,3 +595,66 @@ instance (MonadAsync m, MonadRecorder m) => MonadRecorder (AsyncT m) where
657595 getJournal = lift getJournal
658596 putJournal = lift . putJournal
659597 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