@@ -5,6 +5,7 @@ module Strands.Threads
6 , waitEvents
6 , waitEvents
@@ -17,6 +18,8 @@ module Strands.Threads
18 , waitLogged
19 , waitLogged
19 , gatherLogged
20 , gatherLogged
@@ -391,6 +394,10 @@ async io = do
391 SLast x -> return x
394 SLast x -> return x
392 -- SError e -> back e
395 -- SError e -> back e
397+each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
398+ => [a] -> AsyncT m a
399+each xs = foldl (<|>) empty $ map (async . return) xs
394-- | Force an async computation to run synchronously. It can be useful in an
401-- | Force an async computation to run synchronously. It can be useful in an
395-- 'Alternative' composition to run the alternative only after finishing a
402-- 'Alternative' composition to run the alternative only after finishing a
396-- computation. Note that in Applicatives it might result in an undesired
403-- computation. Note that in Applicatives it might result in an undesired
@@ -551,39 +558,43 @@ wait m = waitAsync (const mzero) Nothing m
552logged :: (Loggable a, MonadIO m) => AsyncT m a -> AsyncT m a
559logged :: (Loggable a, MonadIO m) => AsyncT m a -> AsyncT m a
553logged m = AsyncT $ do
560logged m = AsyncT $ do
554− ctx <- get
561+ ctx <- get
555− case logsRef ctx of
562+ case logsRef ctx of
556− Nothing -> runAsyncT m
563+ Nothing -> runAsyncT m
557− Just _ -> withLog ctx m
564+ Just _ -> runLogged ctx (runAndLogResult m)
560− withLog ctx action =
568+ runLogged ctx action =
561− case journal ctx of
569+ case journal ctx of
562− -- no replay
570+ -- no replay
563− j@(CtxLog _ []) -> do
571+ CtxLog ls [] -> do
564− preLog ctx j
572+ put $ ctx {journal = CtxLog (Executing : ls) []}
565− x <- runAsyncT action
573+ action
569− -- replaying the log
575+ -- replaying the log
570− CtxLog ls (r:rs) -> do
576+ CtxLog ls (r:rs) -> do
571− let j = CtxLog ls rs
577+ -- dbg $ "Replay: j: " ++ show j
573− case r of
578+ case r of
574− Executing -> do
579+ Executing -> do
575− x <- runAsyncT action
580+ put $ ctx {journal = CtxLog (r : ls) rs}
576− postLog ctx (CtxLog ls []) x
581+ action
578− Result val -> do
582+ Result val -> do
579− let x = fmap read val
583+ let x = fmap read val
580− postLog ctx j x
584+ put $ ctx {journal = CtxLog (r : ls) rs}
581− return x
585+ return x
583− -- add an Executing entry at the head of the log
587+ runAndLogResult action = runAsyncT $ do
584− preLog = appendLog Executing
588+ x <- action
585− postLog ctx j x = appendLog (Result (fmap show x)) ctx j
589+ logResult x
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))
586− appendLog l ctx (CtxLog ls rs) = put $ ctx {journal = CtxLog (l : ls) rs}
596+ put $ ctx {journal = CtxLog (entry : ls) []}
588-- XXX enable or disable the suspension points by sending an exception to the
599-- XXX enable or disable the suspension points by sending an exception to the
589-- computation.
600-- computation.
@@ -612,10 +623,22 @@ suspend = logged $ AsyncT $ do
612 _ -> error "Bug: replay inside suspend"
623 _ -> error "Bug: replay inside suspend"
613 where logResult x = Result (fmap show x)
624 where logResult x = Result (fmap show x)
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
630+ put $ ctx {journal = CtxLog [] (reverse entries)}
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
615-- | Run an 'AsyncT m' computation with logging enabled and collect the results
639-- | Run an 'AsyncT m' computation with logging enabled and collect the results
616-- or logs generated by each thread of the computation.
640-- or logs generated by each thread of the computation.
617−gatherLogged :: forall m a. (MonadIO m, MonadCatch m)
641+gatherLogged :: (MonadIO m, MonadCatch m) => AsyncT m a -> m ([a], [Log])
618− => AsyncT m a -> m ([a], [Log])
619gatherLogged m = do
642gatherLogged m = do
620 resultsRef <- liftIO $ newIORef []
643 resultsRef <- liftIO $ newIORef []
621 lref <- liftIO $ newIORef []
644 lref <- liftIO $ newIORef []
@@ -627,8 +650,7 @@ gatherLogged m = do
627-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish
650-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish
628-- and discard the results. If the computation suspends collect the logs to
651-- and discard the results. If the computation suspends collect the logs to
629-- replay later.
652-- replay later.
630−waitLogged :: forall m a. (MonadIO m, MonadCatch m)
653+waitLogged :: (MonadIO m, MonadCatch m) => AsyncT m a -> m [Log]
631− => AsyncT m a -> m [Log]
632waitLogged m = do
654waitLogged m = do
633 lref <- liftIO $ newIORef []
655 lref <- liftIO $ newIORef []
634 waitAsync (const mzero) (Just lref) m
656 waitAsync (const mzero) (Just lref) m