@@ -5,6 +5,7 @@ module Strands.Threads
55 ( parallel
66 , waitEvents
77 , async
8+ , each
89 , sample
910 , sync
1011 --, react
@@ -17,6 +18,8 @@ module Strands.Threads
1718 , suspend
1819 , waitLogged
1920 , gatherLogged
21+ , withLog
22+ , eachWithLog
2023 )
2124where
2225
@@ -391,6 +394,10 @@ async io = do
391394 SLast x -> return x
392395 -- SError e -> back e
393396
397+each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
398+ => [a] -> AsyncT m a
399+each xs = foldl (<|>) empty $ map (async . return) xs
400+
394401-- | Force an async computation to run synchronously. It can be useful in an
395402-- 'Alternative' composition to run the alternative only after finishing a
396403-- computation. Note that in Applicatives it might result in an undesired
@@ -551,39 +558,43 @@ wait m = waitAsync (const mzero) Nothing m
551558
552559logged :: (Loggable a, MonadIO m) => AsyncT m a -> AsyncT m a
553560logged m = AsyncT $ do
554− ctx <- get
555− case logsRef ctx of
556− Nothing -> runAsyncT m
557− Just _ -> withLog ctx m
561+ ctx <- get
562+ case logsRef ctx of
563+ Nothing -> runAsyncT m
564+ Just _ -> runLogged ctx (runAndLogResult m)
565+
558566 where
559567
560− withLog ctx action =
561− case journal ctx of
562− -- no replay
563− j@(CtxLog _ []) -> do
564− preLog ctx j
565− x <- runAsyncT action
566− postLog ctx j x
567− return x
568−
569− -- replaying the log
570− CtxLog ls (r:rs) -> do
571− let j = CtxLog ls rs
572− preLog ctx j
573− case r of
574− Executing -> do
575− x <- runAsyncT action
576− postLog ctx (CtxLog ls []) x
577− return x
578− Result val -> do
579− let x = fmap read val
580− postLog ctx j x
581− return x
582−
583− -- add an Executing entry at the head of the log
584− preLog = appendLog Executing
585− postLog ctx j x = appendLog (Result (fmap show x)) ctx j
586− appendLog l ctx (CtxLog ls rs) = put $ ctx {journal = CtxLog (l : ls) rs}
568+ runLogged ctx action =
569+ case journal ctx of
570+ -- no replay
571+ CtxLog ls [] -> do
572+ put $ ctx {journal = CtxLog (Executing : ls) []}
573+ action
574+
575+ -- replaying the log
576+ CtxLog ls (r:rs) -> do
577+ -- dbg $ "Replay: j: " ++ show j
578+ case r of
579+ Executing -> do
580+ put $ ctx {journal = CtxLog (r : ls) rs}
581+ action
582+ Result val -> do
583+ let x = fmap read val
584+ put $ ctx {journal = CtxLog (r : ls) rs}
585+ return x
586+
587+ runAndLogResult action = runAsyncT $ do
588+ x <- action
589+ logResult x
590+ return x
591+
592+ -- replaces the head of the log with the supplied result
593+ logResult x = AsyncT $ do
594+ ctx@Context{journal = CtxLog (_ : ls) _} <- get
595+ let entry = Result (fmap show (Just x))
596+ put $ ctx {journal = CtxLog (entry : ls) []}
597+ return $ Just ()
587598
588599-- XXX enable or disable the suspension points by sending an exception to the
589600-- computation.
@@ -612,10 +623,22 @@ suspend = logged $ AsyncT $ do
612623 _ -> error "Bug: replay inside suspend"
613624 where logResult x = Result (fmap show x)
614625
626+-- | Compose a computation using previously captured logs
627+withLog :: Monad m => AsyncT m a -> Log -> AsyncT m a
628+withLog m (Log entries) = AsyncT $ do
629+ ctx <- get
630+ put $ ctx {journal = CtxLog [] (reverse entries)}
631+ runAsyncT m
632+
633+-- | Compose a computation using previously captured list of logs. Composes a
634+-- group of computations, one for each log.
635+eachWithLog :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
636+ => AsyncT m a -> [Log] -> AsyncT m a
637+eachWithLog m logs = each logs >>= withLog m
638+
615639-- | Run an 'AsyncT m' computation with logging enabled and collect the results
616640-- or logs generated by each thread of the computation.
617−gatherLogged :: forall m a. (MonadIO m, MonadCatch m)
618− => AsyncT m a -> m ([a], [Log])
641+gatherLogged :: (MonadIO m, MonadCatch m) => AsyncT m a -> m ([a], [Log])
619642gatherLogged m = do
620643 resultsRef <- liftIO $ newIORef []
621644 lref <- liftIO $ newIORef []
@@ -627,8 +650,7 @@ gatherLogged m = do
627650-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish
628651-- and discard the results. If the computation suspends collect the logs to
629652-- replay later.
630−waitLogged :: forall m a. (MonadIO m, MonadCatch m)
631− => AsyncT m a -> m [Log]
653+waitLogged :: (MonadIO m, MonadCatch m) => AsyncT m a -> m [Log]
632654waitLogged m = do
633655 lref <- liftIO $ newIORef []
634656 waitAsync (const mzero) (Just lref) m