transient 0.6.3 → 0.7.0.0
raw patch · 11 files changed
+1347/−887 lines, 11 filesdep −atomic-primopsdep −primitivedep ~random
Dependencies removed: atomic-primops, primitive
Dependency ranges changed: random
Files
- README.md +1/−1
- src/Transient/Backtrack.hs +6/−231
- src/Transient/Base.hs +10/−33
- src/Transient/EVars.hs +1/−1
- src/Transient/Indeterminism.hs +18/−42
- src/Transient/Internals.hs +480/−325
- src/Transient/Logged.hs +292/−125
- src/Transient/Mailboxes.hs +95/−0
- src/Transient/Parse.hs +408/−99
- tests/TestSuite.hs +12/−9
- transient.cabal +24/−21
README.md view
@@ -47,7 +47,7 @@ Motivating example ==================-This program, will stream "hello world" from N nodes if you enter "fire" in the console+This program, will stream "hello world" from all the nodes connected if you enter "fire" in the console ```Haskell main= keep $ initNode $ inputNodes <|> distribStream
src/Transient/Backtrack.hs view
@@ -15,10 +15,10 @@ -- us to write modular and composable code. -- -- Note that backtracking (undo, finalization or exception handling) does not --- change or automatically roll back the user defined state in any way. It only --- executes the user installed handlers. State changes are only caused via user --- defined actions. Any state changes done within the backtracking actions are --- accumulated on top of the user state as it was when backtracking started. +-- roll back the user defined state in any way. It only +-- executes the user-defined handlers. State changes are only caused via user +-- defined actions. These actions also can change the state as it was when backtracking started. +-- -- This example prints the final state as "world". -- -- @ @@ -30,7 +30,7 @@ -- setState "hello" -- oldState <- getState -- --- liftIO (putStrLn "Register undo") \`onUndo` (do +-- liftIO (putStrLn "Register undo") \`onUndo\` (do -- curState <- getState -- liftIO $ putStrLn $ "Final state: " ++ curState -- liftIO $ putStrLn $ "Old state: " ++ oldState) @@ -59,230 +59,5 @@ import Transient.Internals -import Data.Typeable -import Control.Applicative -import Control.Monad.State -import Unsafe.Coerce -import System.Mem.StableName -import Control.Exception -import Control.Concurrent.STM hiding (retry) -import Data.Maybe --- $defaulttrack --- --- A default undo track with the track id of type @()@ is provided. APIs for --- the default track are simpler as they do not require the track id argument. --- --- @ --- import Control.Concurrent (threadDelay) --- import Control.Monad.IO.Class (liftIO) --- import Transient.Base (keep) --- import Transient.Backtrack (onUndo, undo, retry) --- --- main = keep $ do --- step 1 >> tryAgain >> step 2 >> step 3 >> undo >> return () --- where --- step n = liftIO (putStrLn ("Do Step: " ++ show n)) --- \`onUndo` --- liftIO (putStrLn ("Undo Step: " ++ show n)) --- --- tryAgain = liftIO (putStrLn "Will retry on undo") --- \`onUndo` --- (retry >> liftIO (threadDelay 1000000 >> putStrLn "Retrying...")) --- @ - --- $multitrack --- --- Transient allows you to pair an action with an undo action ('onBack'). As --- actions are executed the corresponding undo actions are saved. At any point --- an 'undo' can be triggered which executes all the undo actions registered --- till now in reverse order. At any point, an undo action can decide to resume --- forward execution by using 'forward'. --- --- Multiple independent undo tracks can be defined for different use cases. An --- undo track is identified by a user defined data type. The data type of each --- track must be distinct. --- --- @ --- import Control.Concurrent (threadDelay) --- import Control.Monad.IO.Class (liftIO) --- import Transient.Base (keep) --- import Transient.Backtrack (onBack, forward, back) --- --- data Track = Track String deriving Show --- --- main = keep $ do --- step 1 >> goForward >> step 2 >> step 3 >> back (Track \"Failed") >> return () --- where --- step n = liftIO (putStrLn $ "Execute Step: " ++ show n) --- \`onBack` --- \(Track r) -> liftIO (putStrLn $ show r ++ " Undo Step: " ++ show n) --- --- goForward = liftIO (putStrLn "Turning point") --- \`onBack` \(Track r) -> --- forward (Track r) --- >> (liftIO $ threadDelay 1000000 --- >> putStrLn "Going forward...") --- @ - --- $finalization --- --- Several finish handlers can be installed (using 'onFinish') that are called --- when the action is finalized using 'finish'. All the handlers installed --- until the last 'initFinish' are invoked in reverse order; thread boundaries --- do not matter. The following example prints "3" and then "2". --- --- @ --- import Control.Monad.IO.Class (liftIO) --- import Transient.Base (keep) --- import Transient.Backtrack (initFinish, onFinish, finish) --- --- main = keep $ do --- onFinish (\\_ -> liftIO $ putStrLn "1") --- initFinish --- onFinish (\\_ -> liftIO $ putStrLn "2") --- onFinish (\\_ -> liftIO $ putStrLn "3") --- finish Nothing --- return () --- @ - --- ---data Backtrack b= Show b =>Backtrack{backtracking :: Maybe b --- ,backStack :: [EventF] } --- deriving Typeable --- --- --- ----- | assures that backtracking will not go further back ---backCut :: (Typeable reason, Show reason) => reason -> TransientIO () ---backCut reason= Transient $ do --- delData $ Backtrack (Just reason) [] --- return $ Just () --- ---undoCut :: TransientIO () ---undoCut = backCut () --- ----- | the second parameter will be executed when backtracking ---{-# NOINLINE onBack #-} ---onBack :: (Typeable b, Show b) => TransientIO a -> ( b -> TransientIO a) -> TransientIO a ---onBack ac bac= registerBack (typeof bac) $ Transient $ do --- Backtrack mreason _ <- getData `onNothing` backStateOf (typeof bac) --- runTrans $ case mreason of --- Nothing -> ac --- Just reason -> bac reason --- where --- typeof :: (b -> TransIO a) -> b --- typeof = undefined --- ---onUndo :: TransientIO a -> TransientIO a -> TransientIO a ---onUndo x y= onBack x (\() -> y) --- --- ----- | register an action that will be executed when backtracking ---{-# NOINLINE registerUndo #-} ---registerBack :: (Typeable b, Show b) => b -> TransientIO a -> TransientIO a ---registerBack witness f = Transient $ do --- cont@(EventF _ _ x _ _ _ _ _ _ _ _) <- get -- !!> "backregister" --- --- md <- getData `asTypeOf` (Just <$> backStateOf witness) --- --- case md of --- Just (bss@(Backtrack b (bs@((EventF _ _ x' _ _ _ _ _ _ _ _):_)))) -> --- when (isNothing b) $ do --- addrx <- addr x --- addrx' <- addr x' -- to avoid duplicate backtracking points --- setData $ if addrx == addrx' then bss else Backtrack mwit (cont:bs) --- Nothing -> setData $ Backtrack mwit [cont] --- --- runTrans f --- where --- mwit= Nothing `asTypeOf` (Just witness) --- addr x = liftIO $ return . hashStableName =<< (makeStableName $! x) --- --- ---registerUndo :: TransientIO a -> TransientIO a ---registerUndo f= registerBack () f --- ----- | restart the flow forward from this point on ---forward :: (Typeable b, Show b) => b -> TransIO () ---forward reason= Transient $ do --- Backtrack _ stack <- getData `onNothing` (backStateOf reason) --- setData $ Backtrack(Nothing `asTypeOf` Just reason) stack --- return $ Just () --- ---retry= forward () --- ---noFinish= forward (FinishReason Nothing) --- ----- | execute backtracking. It execute the registered actions in reverse order. ----- ----- If the backtracking flag is changed the flow proceed forward from that point on. ----- ----- If the backtrack stack is finished or undoCut executed, `undo` will stop. ---back :: (Typeable b, Show b) => b -> TransientIO a ---back reason = Transient $ do --- bs <- getData `onNothing` backStateOf reason -- !!>"GOBACK" --- goBackt bs --- --- where --- --- goBackt (Backtrack _ [] )= return Nothing -- !!> "END" --- goBackt (Backtrack b (stack@(first : bs)) )= do --- (setData $ Backtrack (Just reason) stack) --- --- mr <- runClosure first -- !> "RUNCLOSURE" --- --- Backtrack back _ <- getData `onNothing` backStateOf reason --- -- !> "END RUNCLOSURE" --- case back of --- Nothing -> case mr of --- Nothing -> return empty -- !> "FORWARD END" --- Just x -> runContinuation first x -- !> "FORWARD EXEC" --- justreason -> goBackt $ Backtrack justreason bs -- !> ("BACK AGAIN",back) --- ---backStateOf :: (Monad m, Show a, Typeable a) => a -> m (Backtrack a) ---backStateOf reason= return $ Backtrack (Nothing `asTypeOf` (Just reason)) [] --- ---undo :: TransIO a ---undo= back () --- --------- finalization --- ---newtype FinishReason= FinishReason (Maybe SomeException) deriving (Typeable, Show) --- ----- | initialize the event variable for finalization. ----- all the following computations in different threads will share it ----- it also isolate this event from other branches that may have his own finish variable ---initFinish= backCut (FinishReason Nothing) --- ----- | set a computation to be called when the finish event happens ---onFinish :: ((Maybe SomeException) ->TransIO ()) -> TransIO () ---onFinish f= onFinish' (return ()) f --- --- ----- | set a computation to be called when the finish event happens this only apply for ---onFinish' ::TransIO a ->((Maybe SomeException) ->TransIO a) -> TransIO a ---onFinish' proc f= proc `onBack` \(FinishReason reason) -> --- f reason --- --- ----- | trigger the event, so this closes all the resources ---finish :: Maybe SomeException -> TransIO a ---finish reason= back (FinishReason reason) --- --- ----- | kill all the processes generated by the parameter when finish event occurs ---killOnFinish comp= do --- chs <- liftIO $ newTVarIO [] --- onFinish $ const $ liftIO $ killChildren chs -- !> "killOnFinish event" --- r <- comp --- modify $ \ s -> s{children= chs} --- return r --- ----- | trigger finish when the stream of data ends ---checkFinalize v= --- case v of --- SDone -> finish Nothing >> stop --- SLast x -> return x --- SError e -> liftIO ( print e) >> finish Nothing >> stop --- SMore x -> return x +-- Code moved to Internals in order to manage exceptions in spawned threads.
src/Transient/Base.hs view
@@ -241,56 +241,33 @@ ,keep, keep', stop, exit -- * Asynchronous console IO-,option, input,input'+,option,option1, input,input' -- * Task Creation--- $taskgen , StreamData(..)-,parallel, async, waitEvents, sample, spawn, react, abduce+,parallel, async, waitEvents, sample, spawn, react, abduce, fork,sync -- * State management-,setData, getSData, getData, delData, modifyData, modifyData', try, setState, getState, delState, getRState,setRState, modifyState+,setData, getSData, getData, delData, modifyData, modifyData', try, setState, getState, delState, newRState,getRState,setRState, modifyState+,labelState, findState, killState -- * Thread management , threads,addThreads, freeThreads, hookedThreads,oneThread, killChilds +-- * backtracking+,undo,onUndo,retry,back,onBack,forward,backPoint,onBackPoint,finish,onFinish+ -- * Exceptions--- $exceptions -,onException, onException', cutExceptions, continue, catcht, throwt+,onException, onException',whileException, cutExceptions, continue, catcht, throwt,exceptionPoint, onExceptionPoint -- * Utilities ,genId+,module Transient.Logged ) where import Transient.Internals---- $taskgen------ These primitives are used to create asynchronous and concurrent tasks from--- an IO action.------- $exceptions------ Exception handlers are implemented using the backtracking mechanism.--- (see 'Transient.Backtrack.back'). Several exception handlers can be--- installed using 'onException'; handlers are run in reverse order when an--- exception is raised. The following example prints "3" and then "2".------ @--- {-\# LANGUAGE ScopedTypeVariables #-}--- import Transient.Base (keep, onException, cutExceptions)--- import Control.Monad.IO.Class (liftIO)--- import Control.Exception (ErrorCall)------ main = keep $ do--- onException $ \\(e:: ErrorCall) -> liftIO $ putStrLn "1"--- cutExceptions--- onException $ \\(e:: ErrorCall) -> liftIO $ putStrLn "2"--- onException $ \\(e:: ErrorCall) -> liftIO $ putStrLn "3"--- liftIO $ error "Raised ErrorCall exception" >> return ()--- @+import Transient.Logged hiding (exec, wait)
src/Transient/EVars.hs view
@@ -53,7 +53,7 @@ readEVar :: EVar a -> TransIO a readEVar (EVar ref1)= do tchan <- liftIO . atomically $ dupTChan ref1- r <- parallel $ atomically $ readTChan tchan+ r <- parallel $ atomically $ readTChan tchan case r of SDone -> empty SMore x -> return x
src/Transient/Indeterminism.hs view
@@ -13,7 +13,7 @@ ----------------------------------------------------------------------------- {-# LANGUAGE ScopedTypeVariables, CPP #-} module Transient.Indeterminism ( -choose, choose', chooseStream, collect, collect', group, groupByTime +choose, choose', chooseStream, collect, collect', group, groupByTime, burst ) where import Transient.Internals hiding (retry) @@ -21,16 +21,11 @@ import Data.IORef import Control.Applicative import Data.Monoid -import Control.Concurrent -import Data.Typeable +import Control.Concurrent import Control.Monad.State -import GHC.Conc -import Data.Time.Clock import Control.Exception +import qualified Data.ByteString.Char8 as BS -#ifndef ETA_VERSION -import Data.Atomics -#endif -- | Converts a list of pure values into a transient task set. You can use the @@ -46,7 +41,7 @@ chooseStream xs = do evs <- liftIO $ newIORef xs parallel $ do - es <- atomicModifyIORefCAS evs $ \es -> let tes= tail es in (tes,es) + es <- atomicModifyIORef evs $ \es -> let tes= tail es in (tes,es) case es of [x] -> x `seq` return $ SLast x x:_ -> x `seq` return $ SMore x @@ -65,7 +60,7 @@ v <- liftIO $ newIORef (0,[]) x <- proc - mn <- liftIO $ atomicModifyIORefCAS v $ \(n,xs) -> + mn <- liftIO $ atomicModifyIORef v $ \(n,xs) -> let n'=n +1 in if n'== num @@ -75,28 +70,7 @@ Nothing -> stop Just xs -> return xs --- | Collect the results of a task set, grouping all results received within --- every time interval specified by the first parameter as `diffUTCTime`. --- -{- -groupByTime1 time proc = do - t <- liftIO getCurrentTime - - v <- liftIO $ newIORef (0,t,[]) - - x <- proc - t' <- liftIO getCurrentTime - mn <- liftIO $ atomicModifyIORefCAS v $ \(n,t,xs) -> let n'=n +1 - in - if diffUTCTime t' t < fromIntegral time - then ((n',t, x:xs),Nothing) - else ((0 ,t',[]), Just $ x:xs) - case mn of - Nothing -> stop - Just xs -> return xs --} - -- | Collect the results of the first @n@ tasks. Synchronizes concurrent tasks -- to collect the results safely and kills all the non-free threads before -- returning the results. Results are returned in the thread where 'collect' @@ -112,8 +86,8 @@ -- collect' :: Int -> Int -> TransIO a -> TransIO [a] collect' n t search= do - addThreads 1 + rv <- liftIO $ newEmptyMVar -- !> "NEWMVAR" results <- liftIO $ newIORef (0,[]) @@ -124,7 +98,9 @@ stop timer= do - when (t > 0) . async $ threadDelay t >> putMVar rv Nothing + when (t > 0) $ do + --addThreads 1 + async $ threadDelay t >> putMVar rv Nothing empty monitor= liftIO loop @@ -137,28 +113,29 @@ case mr of Nothing -> return rs Just r -> do - let n''= n' +1 + let n''= n' + 1 let rs'= r:rs writeIORef results (n'',rs') - t' <- getCurrentTime if (n > 0 && n'' >= n) then return (rs') else loop - `catch` \(e :: BlockedIndefinitelyOnMVar) -> + `catch` \(_ :: BlockedIndefinitelyOnMVar) -> readIORef results >>= return . snd - oneThread $ timer <|> worker <|> monitor + oneThread $ timer <|> worker <|> monitor --- | insert `SDone` response everytime there is a timeout since the last response +-- | insert `SDone` response every time there is a timeout since the last response burst :: Int -> TransIO a -> TransIO (StreamData a) burst timeout comp= do r <- oneThread comp return (SMore r) <|> (async (threadDelay timeout) >> return SDone) - + +-- | Collect the results of a task set, grouping all results received within +-- every time interval specified by the first parameter as `diffUTCTime`. groupByTime :: Monoid a => Int -> TransIO a -> TransIO a groupByTime timeout comp= do v <- liftIO $ newIORef mempty @@ -166,13 +143,12 @@ where run v = do x <- comp - liftIO $ atomicModifyIORefCAS v $ \xs -> (xs <> x,()) + liftIO $ atomicModifyIORef v $ \xs -> (xs <> x,()) empty gather v= waitEvents $ do threadDelay timeout - atomicModifyIORefCAS v $ \xs -> (mempty , xs) + atomicModifyIORef v $ \xs -> (mempty , xs) -
src/Transient/Internals.hs view
@@ -1,6 +1,6 @@ ------------------------------------------------------------------------------ ----- Module : Base+-- Module : Transient.Internals -- Copyright : -- License : MIT --@@ -8,7 +8,7 @@ -- Stability : -- Portability : ----- | See http://github.com/agocorona/transient+-- | See http://github.com/transient-haskell/transient -- Everything in this module is exported in order to allow extensibility. ----------------------------------------------------------------------------- {-# LANGUAGE CPP #-}@@ -16,17 +16,19 @@ {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ConstraintKinds #-}+--{-# LANGUAGE MonoLocalBinds #-} module Transient.Internals where import Control.Applicative import Control.Monad.State-import Data.Dynamic+--import Data.Dynamic import qualified Data.Map as M import System.IO.Unsafe import Unsafe.Coerce@@ -34,10 +36,9 @@ import qualified Control.Exception (try) import Control.Concurrent -- import GHC.Real---import GHC.Conc(unsafeIOToSTM)+-- import GHC.Conc(unsafeIOToSTM) -- import Control.Concurrent.STM hiding (retry) -- import qualified Control.Concurrent.STM as STM (retry)-import System.Mem.StableName import Data.Maybe import Data.List import Data.IORef@@ -48,11 +49,10 @@ import Data.String import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy.Char8 as BSL import Data.Typeable--#ifndef ETA_VERSION-import Data.Atomics-#endif+import Control.Monad.Fail+import System.Directory #ifdef DEBUG @@ -63,11 +63,11 @@ tshow= Debug.Trace.traceShow {-# INLINE (!>) #-} (!>) :: Show a => b -> a -> b-(!>) x y = trace (show y) x+(!>) x y = trace (show (unsafePerformIO myThreadId, y)) x infixr 0 !> #else-tshow :: Show a => a -> x -> x+tshow :: a -> x -> x tshow _ y= y {-# INLINE (!>) #-} (!>) :: a -> b -> a@@ -75,15 +75,12 @@ #endif+tr x= return () !> x -#ifdef ETA_VERSION-atomicModifyIORefCAS = atomicModifyIORef-#endif type StateIO = StateT EventF IO - newtype TransIO a = Transient { runTrans :: StateIO (Maybe a) } type SData = ()@@ -92,14 +89,13 @@ type TransientIO = TransIO -data LifeCycle = Alive | Parent | Listener | Dead+data LifeCycle = Alive | Parent | Listener | Dead deriving (Eq, Show) -- | EventF describes the context of a TransientIO computation: data EventF = forall a b. EventF { event :: Maybe SData- -- ^ Not yet consumed result (event) from the last asynchronous run of the- -- computation+ -- ^ Not yet consumed result (event) from the last asynchronous computation , xcomp :: TransIO a , fcomp :: [b -> TransIO b]@@ -124,10 +120,19 @@ , labelth :: IORef (LifeCycle, BS.ByteString) -- ^ Label the thread with its lifecycle state and a label string+ , parseContext :: ParseContext+ , execMode :: ExecMode } deriving Typeable +data ParseContext = ParseContext { more :: TransIO (StreamData BSL.ByteString)+ , buffer :: BSL.ByteString+ , done :: IORef Bool} deriving Typeable +-- | To define primitives for all the transient monads: TransIO, Cloud and Widget+class MonadState EventF m => TransMonad m+instance MonadState EventF m => TransMonad m + instance MonadState EventF TransIO where get = Transient $ get >>= return . Just put x = Transient $ put x >> return (Just ())@@ -137,12 +142,22 @@ put s' return $ Just a --- | Run a "non transient" computation within the underlying state monad, so it is--- guaranteed that the computation neither can stop neither can trigger additional--- events/threads.-noTrans :: StateIO x -> TransIO x+-- | Run a computation in the underlying state monad. it is a little lighter and+-- performant and it should not contain advanced effects beyond state.+noTrans :: StateIO x -> TransIO x noTrans x = Transient $ x >>= return . Just +-- | filters away the Nothing responses of the State monad.+-- in principle the state monad should return a single response, but, for performance reasons,+-- it can run inside elements of transient monad (using `runTrans`) which may produce +-- many results+liftTrans :: StateIO (Maybe b) -> TransIO b+liftTrans mx= do+ r <- noTrans mx+ case r of+ Nothing -> empty+ Just x -> return x + emptyEventF :: ThreadId -> IORef (LifeCycle, BS.ByteString) -> MVar [EventF] -> EventF emptyEventF th label childs = EventF { event = mempty@@ -155,7 +170,9 @@ , parent = Nothing , children = childs , maxThread = Nothing- , labelth = label }+ , labelth = label+ , parseContext = ParseContext (return SDone) mempty undefined+ , execMode = Serial} -- | Run a transient computation with a default initial state runTransient :: TransIO a -> IO (Maybe a, EventF)@@ -239,7 +256,7 @@ -- | Restore the continuations to the provided ones. -- | NOTE: Events are also cleared out.-restoreStack :: MonadState EventF m => [a -> TransIO a] -> m ()+restoreStack :: TransMonad m => [a -> TransIO a] -> m () restoreStack fs = modify $ \EventF {..} -> EventF { event = Nothing, fcomp = fs, .. } -- | Run a chain of continuations.@@ -262,51 +279,58 @@ instance Applicative TransIO where pure a = Transient . return $ Just a - mf <*> mx = do+ mf <*> mx = do -- do f <- mf; x <- mx ; return $ f x r1 <- liftIO $ newIORef Nothing r2 <- liftIO $ newIORef Nothing- fparallel r1 r2 <|> xparallel r1 r2+ fparallel r1 r2 <|> xparallel r1 r2+ where+ fparallel r1 r2= do f <- mf -- r <- getState <|> return NoRemote- -- return () !> ("first",r)- if r == NoRemote then do- x <- mx !> "second serial"- return $ f x- else do-- liftIO $ (writeIORef r1 $ Just f)- mx <- liftIO (readIORef r2)- case mx of- Nothing -> empty+ liftIO $ (writeIORef r1 $ Just f)+ mr <- liftIO (readIORef r2) + case mr of+ Nothing -> empty Just x -> return $ f x xparallel r1 r2 = do+ + mr <- liftIO (readIORef r1)+ case mr of+ Nothing -> do - r <- getState <|> return NoRemote- -- return () !> ("SECOND par",r)- if r == WasParallel then do- -- delData WasParallel- x <- mx- liftIO $ (writeIORef r2 $ Just x)- mf <- liftIO (readIORef r1)- case mf of- Nothing -> empty- Just f -> return $ f x- else empty+ p <- gets execMode+ + if p== Serial then empty else do+ x <- mx+ liftIO $ (writeIORef r2 $ Just x)+ + mr <- liftIO (readIORef r1)+ case mr of+ Nothing -> empty + Just f -> return $ f x + + Just f -> do+ x <- mx+ liftIO $ (writeIORef r2 $ Just x)+ return $ f x ++ ++data ExecMode = Remote | Parallel | Serial+ deriving (Typeable, Eq, Show)+ -- | stop the current computation and does not execute any alternative computation fullStop :: TransIO stop-fullStop= setData WasRemote >> stop+fullStop= do modify $ \s ->s{execMode= Remote} ; stop instance Monad TransIO where return = pure x >>= f = Transient $ do- -- delData WasParallel setEventCont x f mk <- runTrans x resetEventCont mk@@ -341,8 +365,9 @@ mplus x y = Transient $ do mx <- runTrans x - was <- getData `onNothing` return NoRemote- if was == WasRemote+ was <- gets execMode -- getData `onNothing` return Serial+ + if was == Remote then return Nothing else case mx of@@ -350,6 +375,9 @@ justx -> return justx +instance MonadFail TransIO where+ fail _ = mzero+ readWithErr :: (Typeable a, Read a) => Int -> String -> IO [(a, String)] readWithErr n line = (v `seq` return [(v, left)])@@ -358,8 +386,11 @@ ++ "\" in: " ++ " <" ++ show line ++ "> ") where (v, left):_ = readsPrec n line -newtype ParseError= ParseError String deriving (Show)+newtype ParseError= ParseError String +instance Show ParseError where+ show (ParseError s)= "ParseError " ++ s + instance Exception ParseError read' s= case readsPrec' 0 s of@@ -371,45 +402,10 @@ readsPrec' n = unsafePerformIO . readWithErr n --- | Constraint type synonym for a value that can be logged.-type Loggable a = (Show a, Read a, Typeable a) --- data Serializable a where--- serialize :: a .ç-> BS.ByteString--- deserialize :: BS.ByteString -> a --- instance Serialize a => Serialie [a] where--- serialize (x:xs)=--- let s= serialize x--- l= length s--- in makeByteString (#(#l,s#),serialize xs #) --- | Dynamic serializable data for logging.-data IDynamic =- IDyns String- | forall a. Loggable a => IDynamic a -instance Show IDynamic where- show (IDynamic x) = show (show x)- show (IDyns s) = show s--instance Read IDynamic where- readsPrec n str = map (\(x,s) -> (IDyns x,s)) $ readsPrec' n str--type Recover = Bool-type CurrentPointer = [LogElem]-type LogEntries = [LogElem]-type Hash = Int--data LogElem = Wait | Exec | Var IDynamic- deriving (Read, Show)--data Log = Log Recover CurrentPointer LogEntries Hash- deriving (Typeable, Show)--data RemoteStatus = WasRemote | WasParallel | NoRemote- deriving (Typeable, Eq, Show)- -- | A synonym of 'empty' that can be used in a monadic expression. It stops -- the computation, which allows the next computation in an 'Alternative' -- ('<|>') composition to run.@@ -448,6 +444,8 @@ atEnd :: m a -> m b -> m a atEnd = (<***) ++ instance AdditionalOperators TransIO where --(**>) :: TransIO a -> TransIO b -> TransIO b@@ -473,7 +471,7 @@ runTrans mb return a -infixr 1 <***, <**, **>+infixl 4 <***, <**, **> -- | Run @b@ once, discarding its result when the first task in task set @a@ -- has finished. Useful to start a singleton task after the first task has been@@ -530,9 +528,10 @@ -- * Threads -waitQSemB sem = atomicModifyIORefCAS sem $ \n ->- if n > 0 then(n - 1, True) else (n, False)-signalQSemB sem = atomicModifyIORefCAS sem $ \n -> (n + 1, ())+waitQSemB onemore sem = atomicModifyIORef sem $ \n ->+ let one = if onemore then 1 else 0+ in if n + one > 0 then(n - 1, True) else (n, False)+signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ()) -- | Sets the maximum number of threads that can be created for the given task -- set. When set to 0, new tasks start synchronously in the current thread.@@ -542,52 +541,100 @@ msem <- gets maxThread sem <- liftIO $ newIORef n modify $ \s -> s { maxThread = Just sem }- r <- process <** (modify $ \s -> s { maxThread = msem }) -- restore it+ r <- process <*** (modify $ \s -> s { maxThread = msem }) -- restore it return r --- | Terminate all the child threads in the given task set and continue--- execution in the current thread. Useful to reap the children when a task is--- done.----oneThread :: TransIO a -> TransIO a-oneThread comp = do+-- clone the current state as a child of the current state, with the same thread+cloneInChild name= do st <- get rchs <- liftIO $ newMVar []- label <- liftIO $ newIORef (Alive, BS.pack "oneThread")+ label <- liftIO $ newIORef (Alive, if not $ null name then BS.pack name else mempty) let st' = st { parent = Just st , children = rchs , labelth = label } liftIO $ do- atomicModifyIORefCAS (labelth st) $ \(_, label) -> ((Parent,label),())- hangThread st st'+ atomicModifyIORef (labelth st) $ \(_, label) -> ((Parent,label),())+ hangThread st st' -- parent could have more than one children with the same threadId - put st'+ return st'++-- remove the current child task from the tree of tasks. +-- If the child and parent threads are different, the child is killed+removeChild :: (MonadIO m,TransMonad m) => m ()+removeChild = do+ st <- get+ let mparent = parent st+ case mparent of+ Nothing -> return ()+ Just parent -> do + sts <- liftIO $ modifyMVar (children parent) $ \ths -> do+ let (xs,sts)= partition (\st' -> threadId st' /= threadId st) ths+ ys <- case sts of+ [] -> return []+ st':_ -> readMVar $ children st'+ return (xs ++ ys,sts)+ + put parent+ case sts of+ [] -> return()+ st':_ -> do+ (status,_) <- liftIO $ readIORef $ labelth st'+ if status == Listener || threadId parent == threadId st then return () else liftIO $ (killThread . threadId) st'+-- | Terminate all the child threads in the given task set and continue+-- execution in the current thread. Useful to reap the children when a task is+-- done, restart a task when a new event happens etc.+--+oneThread :: TransIO a -> TransIO a+oneThread comp = do+ st <- cloneInChild "oneThread" + let rchs= children st x <- comp th <- liftIO myThreadId- !> ("FATHER:", threadId st)+ -- !> ("FATHER:", threadId st) chs <- liftIO $ readMVar rchs liftIO $ mapM_ (killChildren1 th) chs- !> ("KILLEVENT1 ", map threadId chs )+ -- !> ("KILLEVENT1 ", map threadId chs ) return x where killChildren1 :: ThreadId -> EventF -> IO () killChildren1 th state = do+ forkIO $ do ths' <- modifyMVar (children state) $ \ths -> do let (inn, ths')= partition (\st -> threadId st == th) ths return (inn, ths') mapM_ (killChildren1 th) ths' mapM_ (killThread . threadId) ths'+ return() -- | Add a label to the current passing threads so it can be printed by debugging calls like `showThreads`-labelState :: (MonadIO m,MonadState EventF m) => BS.ByteString -> m ()+labelState :: (MonadIO m,TransMonad m) => BS.ByteString -> m () labelState l = do st <- get- liftIO $ atomicModifyIORefCAS (labelth st) $ \(status,_) -> ((status, l), ())+ liftIO $ atomicModifyIORef (labelth st) $ \(status,prev) -> ((status, prev <> BS.pack "," <> l), ()) +-- | return the threadId associated with an state (you can see all of them with the console option 'ps')+threadState thid= do + st <- findState match =<< topState+ return $ threadId st :: TransIO ThreadId+ where+ match st= do+ (_,lab) <-liftIO $ readIORef $ labelth st+ return $ if lab == thid then True else False++-- | kill the thread subtree labeled as such (you can see all of them with the console option 'ps')+killState thid= do+ st <- findState match =<< topState+ liftIO $ killBranch' st+ where+ match st= do+ (_,lab) <-liftIO $ readIORef $ labelth st+ return $ if lab == thid then True else False+ + printBlock :: MVar () printBlock = unsafePerformIO $ newMVar () @@ -604,7 +651,7 @@ if BS.null label then putStr . show $ threadId ch else do BS.putStr label; putStr . drop 8 . show $ threadId ch- when (state == Dead) $ putStr " dead"+ when (state == Dead) $ putStr " dead" -- putStr " " >> putStr (take 3 $ show state) -- putStrLn $ if mythread == threadId ch then " <--" else "" chs <- readMVar $ children ch mapM_ (showTree $ n + 2) $ reverse chs@@ -612,7 +659,7 @@ -- | Return the state of the thread that initiated the transient computation -- topState :: TransIO EventF-topState :: MonadState EventF m => m EventF+topState :: TransMonad m => m EventF topState = do st <- get return $ toplevel st@@ -750,19 +797,20 @@ -- | Kill the childs and the thread of an state killBranch' :: EventF -> IO () killBranch' cont = do- killChildren $ children cont- let thisth = threadId cont- mparent = parent cont- when (isJust mparent) $- modifyMVar_ (children $ fromJust mparent) $ \sts ->- return $ filter (\st -> threadId st /= thisth) sts- killThread $ thisth !> ("kill this thread:",thisth)-+ forkIO $ do+ killChildren $ children cont+ let thisth = threadId cont+ mparent = parent cont+ when (isJust mparent) $+ modifyMVar_ (children $ fromJust mparent) $ \sts ->+ return $ filter (\st -> threadId st /= thisth) sts+ killThread $ thisth !> ("kill this thread:",thisth)+ return () -- * Extensible State: Session Data Management --- | Same as 'getSData' but with a more general type. If the data is found, a+-- | Same as 'getSData' but with a more conventional interface. If the data is found, a -- 'Just' value is returned. Otherwise, a 'Nothing' value is returned.-getData :: (MonadState EventF m, Typeable a) => m (Maybe a)+getData :: (TransMonad m, Typeable a) => m (Maybe a) getData = resp where resp = do list <- gets mfData@@ -776,12 +824,17 @@ -- | Retrieve a previously stored data item of the given data type from the -- monad state. The data type to retrieve is implicitly determined by the data type.--- If the data item is not found, empty is executed, so the alternative computation will be executed, if any, or--- Otherwise, the computation will stop..--- If you want to print an error message or a default value, you can use an 'Alternative' composition. For example:+-- If the data item is not found, empty is executed, so the alternative computation will be executed, if any. +-- Otherwise, the computation will stop.+-- If you want to print an error message or return a default value, you can use an 'Alternative' composition. For example: -- -- > getSData <|> error "no data of the type desired" -- > getInt = getSData <|> return (0 :: Int)+--+-- The later return either the value set or 0.+--+-- It is highly recommended not to use it directly, since his relatively complex behaviour may be confusing sometimes.+-- Use instead a monomorphic alias like "getInt" defined above. getSData :: Typeable a => TransIO a getSData = Transient getData @@ -809,14 +862,14 @@ -- Person name age <- getSData -- liftIO $ print (name, age) -- @-setData :: (MonadState EventF m, Typeable a) => a -> m ()+setData :: (TransMonad m, Typeable a) => a -> m () setData x = modify $ \st -> st { mfData = M.insert t (unsafeCoerce x) (mfData st) } where t = typeOf x -- | Accepts a function which takes the current value of the stored data type -- and returns the modified value. If the function returns 'Nothing' the value -- is deleted otherwise updated.-modifyData :: (MonadState EventF m, Typeable a) => (Maybe a -> Maybe a) -> m ()+modifyData :: (TransMonad m, Typeable a) => (Maybe a -> Maybe a) -> m () modifyData f = modify $ \st -> st { mfData = M.alter alterf t (mfData st) } where typeResp :: (Maybe a -> b) -> a typeResp = undefined@@ -831,7 +884,7 @@ -- -- > runTransient $ do modifyData' (\h -> h ++ " world") "hello new" ; r <- getSData ; liftIO $ putStrLn r -- > "hello new" -- > runTransient $ do setData "hello" ; modifyData' (\h -> h ++ " world") "hello new" ; r <- getSData ; liftIO $ putStrLn r -- > "hello world"-modifyData' :: (MonadState EventF m, Typeable a) => (a -> a) -> a -> m a+modifyData' :: (TransMonad m, Typeable a) => (a -> a) -> a -> m a modifyData' f v= do st <- get let (ma,nmap)= M.insertLookupWithKey alterf t (unsafeCoerce v) (mfData st)@@ -840,20 +893,20 @@ where t = typeOf v alterf _ _ x = unsafeCoerce $ f $ unsafeCoerce x --- | Same as modifyData-modifyState :: (MonadState EventF m, Typeable a) => (Maybe a -> Maybe a) -> m ()+-- | Same as `modifyData`+modifyState :: (TransMonad m, Typeable a) => (Maybe a -> Maybe a) -> m () modifyState = modifyData --- | Same as 'setData'-setState :: (MonadState EventF m, Typeable a) => a -> m ()+-- | Same as `setData`+setState :: (TransMonad m, Typeable a) => a -> m () setState = setData -- | Delete the data item of the given type from the monad state.-delData :: (MonadState EventF m, Typeable a) => a -> m ()+delData :: (TransMonad m, Typeable a) => a -> m () delData x = modify $ \st -> st { mfData = M.delete (typeOf x) (mfData st) } --- | Same as 'delData'-delState :: (MonadState EventF m, Typeable a) => a -> m ()+-- | Same as `delData`+delState :: (TransMonad m, Typeable a) => a -> m () delState = delData @@ -862,26 +915,34 @@ newtype Ref a = Ref (IORef a) +-- | Initializes a new mutable reference (similar to STRef in the state monad)+-- It is polimorphic. Each type has his own reference+-- It return the associated IORef, so it can be updated in the IO monad+newRState:: (MonadIO m,TransMonad m, Typeable a) => a -> m (IORef a)+newRState x= do + ref@(Ref rx) <- Ref <$> liftIO (newIORef x)+ setData ref+ return rx -- | mutable state reference that can be updated (similar to STRef in the state monad) -- They are identified by his type. -- Initialized the first time it is set.-setRState:: (MonadIO m,MonadState EventF m, Typeable a) => a -> m ()+setRState:: (MonadIO m,TransMonad m, Typeable a) => a -> m () setRState x= do Ref ref <- getData `onNothing` do ref <- Ref <$> liftIO (newIORef x) setData ref return ref- liftIO $ atomicModifyIORefCAS ref $ const (x,())+ liftIO $ atomicModifyIORef ref $ const (x,()) -getRData :: (MonadIO m, MonadState EventF m, Typeable a) => m (Maybe a)+getRData :: (MonadIO m, TransMonad m, Typeable a) => m (Maybe a) getRData= do mref <- getData case mref of Just (Ref ref) -> Just <$> (liftIO $ readIORef ref) Nothing -> return Nothing- - ++ getRState :: Typeable a => TransIO a getRState= Transient getRData @@ -890,11 +951,11 @@ ref = undefined -- | Run an action, if it does not succeed, undo any state changes--- that it might have caused and allow aternative actions to run with the original state+-- that may have been caused by the action and allow aternative actions to run with the original state try :: TransIO a -> TransIO a try mx = do- sd <- gets mfData- mx <|> (modify (\s -> s { mfData = sd }) >> empty)+ s <- get+ mx <|> (modify (const s) >> empty) -- | Executes the computation and reset the state either if it fails or not. sandbox :: TransIO a -> TransIO a@@ -904,20 +965,20 @@ -- | generates an identifier that is unique within the current program execution genGlobalId :: MonadIO m => m Int-genGlobalId= liftIO $ atomicModifyIORefCAS rglobalId $ \n -> (n +1,n)+genGlobalId= liftIO $ atomicModifyIORef rglobalId $ \n -> (n +1,n) rglobalId= unsafePerformIO $ newIORef (0 :: Int) -- | Generator of identifiers that are unique within the current monadic -- sequence They are not unique in the whole program.-genId :: MonadState EventF m => m Int+genId :: TransMonad m => m Int genId = do st <- get let n = mfSequence st put st { mfSequence = n + 1 } return n -getPrevId :: MonadState EventF m => m Int+getPrevId :: TransMonad m => m Int getPrevId = gets mfSequence instance Read SomeException where@@ -925,11 +986,11 @@ readsPrec n str = [(SomeException $ ErrorCall s, r)] where [(s , r)] = readsPrec n str --- | 'StreamData' represents a task in a task stream being generated.+-- | 'StreamData' represents an result in an stream being generated. data StreamData a =- SMore a -- ^ More tasks to come- | SLast a -- ^ This is the last task- | SDone -- ^ No more tasks, we are done+ SMore a -- ^ More to come+ | SLast a -- ^ This is the last one+ | SDone -- ^ No more, we are done | SError SomeException -- ^ An error occurred deriving (Typeable, Show,Read) @@ -938,9 +999,9 @@ fmap f (SLast a)= SLast (f a) fmap _ SDone= SDone --- | A task stream generator that produces an infinite stream of tasks by--- running an IO computation in a loop. A task is triggered carrying the output--- of the computation. See 'parallel' for notes on the return value.+-- | A task stream generator that produces an infinite stream of results by+-- running an IO computation in a loop, each result may be processed in different threads (tasks)+-- depending on the thread limits stablished with `threads`. waitEvents :: IO a -> TransIO a waitEvents io = do mr <- parallel (SMore <$> io)@@ -951,9 +1012,8 @@ -- | Run an IO computation asynchronously carrying -- the result of the computation in a new thread when it completes. -- If there are no threads available, the async computation and his continuation is executed--- before any alternative computation.--- See 'parallel' for notes on--- the return value.+-- in the same thread before any alternative computation.+ async :: IO a -> TransIO a async io = do mr <- parallel (SLast <$> io)@@ -961,39 +1021,48 @@ SLast x -> return x SError e -> back e --- | Force an async computation to run synchronously. It can be useful in an--- 'Alternative' composition to run the alternative only after finishing a--- computation. Note that in Applicatives it might result in an undesired--- serialization.+-- | Avoid the execution of alternative computations when the computation is asynchronous+--+-- > sync (async whatever) <|> liftIO (print "hello") -- never print "hello" sync :: TransIO a -> TransIO a sync x = do- setData WasRemote- r <- x- delData WasRemote+ was <- gets execMode -- getSData <|> return Serial+ r <- x <** modify (\s ->s{execMode= Remote}) -- setData Remote+ modify $ \s -> s{execMode= was} return r --- | @spawn = freeThreads . waitEvents@+-- | create task threads faster, but with no thread control: @spawn = freeThreads . waitEvents@ spawn :: IO a -> TransIO a spawn = freeThreads . waitEvents -- | An stream generator that run an IO computation periodically at the specified time interval. The -- task carries the result of the computation. A new result is generated only if--- the output of the computation is different from the previous one. See--- 'parallel' for notes on the return value.+-- the output of the computation is different from the previous one. sample :: Eq a => IO a -> Int -> TransIO a sample action interval = do v <- liftIO action prev <- liftIO $ newIORef v- waitEvents (loop action prev) <|> async (return v)- where loop action prev = loop'- where loop' = do- threadDelay interval- v <- action- v' <- readIORef prev- if v /= v' then writeIORef prev v >> return v else loop'+ waitEvents (loop action prev) <|> return v+ where + loop action prev = loop'+ where + loop' = do+ threadDelay interval+ v <- action+ v' <- readIORef prev+ if v /= v' then writeIORef prev v >> return v else loop' +-- | Runs the rest of the computation in a new thread. Returns 'empty' to the current thread+abduce = async $ return () ++-- | fork an independent process. It is equivalent to forkIO. The thread created +-- is managed with the thread control primitives of transient+fork :: TransIO () -> TransIO ()+fork proc= (abduce >> proc >> empty) <|> return()+ + -- | Run an IO action one or more times to generate a stream of tasks. The IO -- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new -- result is returned with the result value. If there are threads available, the res of the@@ -1006,41 +1075,40 @@ -- task. parallel :: IO (StreamData b) -> TransIO (StreamData b) parallel ioaction = Transient $ do- was <- getData `onNothing` return NoRemote- when (was /= WasRemote) $ setData WasParallel+ --was <- gets execMode -- getData `onNothing` return Serial+ --when (was /= Remote) $ modify $ \s -> s{execMode= Parallel}+ modify $ \s -> s{execMode=let rs= execMode s in if rs /= Remote then Parallel else rs} cont <- get- -- !> "PARALLEL"+ -- !> "PARALLEL" case event cont of j@(Just _) -> do put cont { event = Nothing } return $ unsafeCoerce j Nothing -> do- liftIO $ atomicModifyIORefCAS (labelth cont) $ \(_, lab) -> ((Parent, lab), ())+ liftIO $ atomicModifyIORef (labelth cont) $ \(_, lab) -> ((Parent, lab), ()) liftIO $ loop cont ioaction --- th <- liftIO myThreadId--- return () !> ("finish",th) return Nothing -- | Execute the IO action and the continuation loop :: EventF -> IO (StreamData t) -> IO ()-loop parentc rec = forkMaybe parentc $ \cont -> do+loop parentc rec = forkMaybe True parentc $ \cont -> do -- Execute the IO computation and then the closure-continuation- liftIO $ atomicModifyIORefCAS (labelth cont) $ const ((Listener,BS.pack "wait"),())+ liftIO $ atomicModifyIORef (labelth cont) $ \(_,label) -> ((Listener,label),()) let loop'= do- mdat <- rec `catch` \(e :: SomeException) -> return $ SError e+ mdat <- rec `catch` \(e :: SomeException) -> return $ SError e case mdat of se@(SError _) -> setworker cont >> iocont se cont SDone -> setworker cont >> iocont SDone cont last@(SLast _) -> setworker cont >> iocont last cont more@(SMore _) -> do- forkMaybe cont $ iocont more+ forkMaybe False cont $ iocont more loop' where- setworker cont= liftIO $ atomicModifyIORefCAS (labelth cont) $ const ((Alive,BS.pack "work"),())+ setworker cont= liftIO $ atomicModifyIORef (labelth cont) $ \(_,lab) -> ((Alive,lab),()) iocont dat cont = do @@ -1053,13 +1121,16 @@ return () where {-# INLINABLE forkMaybe #-}- forkMaybe parent proc = do+ forkMaybe :: Bool -> EventF -> (EventF -> IO ()) -> IO ()+ forkMaybe onemore parent proc = do case maxThread parent of Nothing -> forkIt parent proc Just sem -> do- dofork <- waitQSemB sem- if dofork then forkIt parent proc else proc parent-+ dofork <- waitQSemB onemore sem+ if dofork then forkIt parent proc + else proc parent + `catch` \e ->exceptBack parent e >> return()+ forkIt parent proc= do chs <- liftIO $ newMVar []@@ -1075,20 +1146,19 @@ proc cont') $ \me -> do- case me of- Left e -> exceptBack cont e >> return () -- !> "exceptBack 2"+ Left e -> (exceptBack cont e >> return ()) -- !> "exceptBack 2" - _ -> do- case maxThread cont of- Just sem -> signalQSemB sem -- !> "freed thread"+ _ -> return () + case maxThread cont of+ Just sem -> signalQSemB sem -- !> "freed thread" Nothing -> return ()- when(not $ freeTh parent ) $ do -- if was not a free thread+ when(not $ freeTh parent ) $ do -- if was not a free thread - th <- myThreadId- (can,label) <- atomicModifyIORefCAS (labelth cont) $ \(l@(status,label)) ->+ th <- myThreadId + (can,label) <- atomicModifyIORef (labelth cont) $ \(l@(status,label)) -> ((if status== Alive then Dead else status, label),l) when (can /= Parent ) $ free th parent @@ -1097,7 +1167,7 @@ forkFinally1 :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId- forkFinally1 action and_then =+ forkFinally1 action and_then = mask $ \restore -> forkIO $ Control.Exception.try (restore action) >>= and_then free th env= do@@ -1146,45 +1216,50 @@ -- | kill all the child threads associated with the continuation context killChildren childs = do--+ forkIO $ do ths <- modifyMVar childs $ \ths -> return ([],ths)- mapM_ (killChildren . children) ths--- mapM_ (killThread . threadId) ths !> ("Kill children", map threadId ths )-+ mapM_ (\th -> do+ (status,_) <- readIORef $ labelth th+ when (status /= Listener && status /= Parent) $ killThread $ threadId th !> ("killChildren",threadId th, status)) ths >> return () + return () --- | Make a transient task generator from an asynchronous callback handler.+-- | capture a callback handler so that the execution of the current computation continues +-- whenever an event occurs. The effect is called "de-inversion of control" ----- The first parameter is a callback. The second parameter is a value to be+-- The first parameter is a callback setter. The second parameter is a value to be -- returned to the callback; if the callback expects no return value it--- can just be a @return ()@. The callback expects a setter function taking the--- @eventdata@ as an argument and returning a value to the callback; this--- function is supplied by 'react'.+-- can just be @return ()@. The callback setter expects a function taking the+-- @eventdata@ as an argument and returning a value; this+-- function is the continuation, which is supplied by 'react'. -- -- Callbacks from foreign code can be wrapped into such a handler and hooked -- into the transient monad using 'react'. Every time the callback is called it--- generates a new task for the transient monad.+-- continues the execution on the current transient computation. ---+-- > +-- > do+-- > event <- react onEvent $ return ()+-- > ....+-- > react- :: Typeable eventdata- => ((eventdata -> IO response) -> IO ())+ :: ((eventdata -> IO response) -> IO ()) -> IO response -> TransIO eventdata-react setHandler iob= Transient $ do- was <- getData `onNothing` return NoRemote- when (was /= WasRemote) $ setData WasParallel+react setHandler iob= do+ st <- cloneInChild "react"+ liftIO $ atomicModifyIORef (labelth st) $ \(_,label) -> ((Listener,label),())+ Transient $ do+ modify $ \s -> s{execMode=let rs= execMode s in if rs /= Remote then Parallel else rs} cont <- get+ case event cont of Nothing -> do liftIO $ setHandler $ \dat ->do- runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat} `catch` exceptBack cont+ runStateT (runCont cont) st{event= Just $ unsafeCoerce dat} `catch` exceptBack cont iob return Nothing@@ -1192,9 +1267,7 @@ j@(Just _) -> do put cont{event=Nothing} return $ unsafeCoerce j---- | Runs the rest of the computation in a new thread. Returns 'empty' to the current thread-abduce = async $ return ()+ -- * Non-blocking keyboard input @@ -1203,7 +1276,7 @@ -- | listen stdin and triggers a new task every time the input data -- matches the first parameter. The value contained by the task is the matched--- value i.e. the first argument itself. The second parameter is a message to the user for+-- value i.e. the first argument itself. The second parameter is a message for -- the user. The label is displayed in the console when the option match. option :: (Typeable b, Show b, Read b, Eq b) => b -> String -> TransIO b@@ -1219,43 +1292,52 @@ Bool -> b -> String -> TransIO b optionf flag ret message = do let sret= if typeOf ret == typeOf "" then unsafeCoerce ret else show ret- liftIO $ putStrLn $ "Enter "++sret++"\t\tto: " ++ message- inputf flag sret Nothing ( == sret)+ let msg= "Enter "++sret++"\t\tto: " ++ message++"\n"+ inputf flag sret msg Nothing ( == sret) liftIO $ putStr "\noption: " >> putStrLn sret -- abduce return ret --- | inputf <remove after sucessful or not> <listener identifier> <Maybe default value> <validation proc>-inputf :: Loggable a => Bool -> String -> Maybe a -> (a -> Bool) -> TransIO a-inputf flag ident mv cond= do- str <- react (addListener ident) (return ())-- when flag $ liftIO $ delListener ident - c <- liftIO $ readIORef rconsumed- if c then returnm mv else do- if null str then do liftIO $ writeIORef rconsumed True; returnm mv else do - let rr = read1 str - - case rr of- Just x -> if cond x - then liftIO $ do- writeIORef rconsumed True - -- print x- -- hFlush stdout- return x- else do liftIO $ when (isJust mv) $ putStrLn ""; returnm mv- _ -> do liftIO $ when (isJust mv) $ putStrLn ""; returnm mv +-- | General asynchronous console input.+-- +-- inputf <remove input listener after sucessful or not> <listener identifier> <prompt> +-- <Maybe default value> <validation proc>+inputf :: (Show a, Read a,Typeable a) => Bool -> String -> String -> Maybe a -> (a -> Bool) -> TransIO a+inputf remove ident message mv cond = do+ let loop= do+ liftIO $ putStr message >> hFlush stdout + str <- react (addConsoleAction ident message) (return ())+ when remove $ do removeChild; liftIO $ delConsoleAction ident + c <- liftIO $ readIORef rconsumed+ if c then returnm mv else do - where+ let rr = read1 str + + case (rr,str) of+ (Nothing,_) -> do (liftIO $ when (isJust mv) $ putStrLn ""); returnm mv + (Just x,"") -> do (liftIO $ do writeIORef rconsumed True; print x); returnm mv + (Just x,_) -> if cond x + then liftIO $ do+ writeIORef rconsumed True + print x+ -- hFlush stdout+ return x+ + else do liftIO $ when (isJust mv) $ putStrLn ""+ returnm mv+ loop+ where returnm (Just x)= return x returnm _ = empty -- read1 :: String -> Maybe a read1 s= r where- r = if (typeOf $ fromJust r) == typeOf ""- then Just $ unsafeCoerce s- else case reads s of+ typ= typeOf $ fromJust r+ r = if typ == typeOf "" then Just $ unsafeCoerce s+ else if typ == typeOf (BS.pack "") then Just $ unsafeCoerce $ BS.pack s+ else if typ == typeOf (BSL.pack "") then Just $ unsafeCoerce $ BSL.pack s+ else case reads s of [] -> Nothing [(x,"")] -> Just x @@ -1270,21 +1352,22 @@ -- | `input` with a default value input' :: (Typeable a, Read a,Show a) => Maybe a -> (a -> Bool) -> String -> TransIO a input' mv cond prompt= do - liftIO $ putStr prompt >> hFlush stdout - inputf True "input" mv cond- --+ --liftIO $ putStr prompt >> hFlush stdout + inputf True "input" prompt mv cond -rcb= unsafePerformIO $ newIORef [] :: IORef [ (String,String -> IO())]--addListener :: String -> (String -> IO ()) -> IO ()-addListener name cb= atomicModifyIORef rcb $ \cbs -> (filter ((/=) name . fst) cbs ++ [(name, cb)],())+rcb= unsafePerformIO $ newIORef [] :: IORef [ (String,String,String -> IO())] -delListener :: String -> IO ()-delListener name= atomicModifyIORef rcb $ \cbs -> (filter ((/=) name . fst) cbs,())+addConsoleAction :: String -> String -> (String -> IO ()) -> IO ()+addConsoleAction name message cb= atomicModifyIORef rcb $ \cbs -> + ((name,message, cb) : filter ((/=) name . fst) cbs ,())+ where+ fst (x,_,_)= x +delConsoleAction :: String -> IO ()+delConsoleAction name= atomicModifyIORef rcb $ \cbs -> (filter ((/=) name . fst) cbs,())+ where+ fst (x,_,_)= x reads1 s=x where@@ -1295,17 +1378,18 @@ read1 s= let [(x,"")]= reads1 s in x - rprompt= unsafePerformIO $ newIORef "> " inputLoop= do+ prompt <- readIORef rprompt+ when (not $ null prompt) $ do putStr prompt ; hFlush stdout line <- getLine+ --threadDelay 1000000+ processLine line - threadDelay 1000000- prompt <- readIORef rprompt- when (not $ null prompt) $ do putStr prompt ; hFlush stdout+ inputLoop-+ `catch` \(SomeException _) -> inputLoop -- myThreadId >>= killThread {-# NOINLINE rconsumed #-}@@ -1328,9 +1412,9 @@ where invokeParsers x= do mbs <- readIORef rcb- mapM_ (\cb -> cb x) $ map snd mbs+ mapM_ (\cb -> cb x) $ map (\(_,_,p)-> p) mbs - mapM' f []= return ()+ mapM' _ []= return () mapM' f (xss@(x:xs)) =do f x r <- readIORef rconsumed@@ -1344,7 +1428,7 @@ else do threadDelay 1000 n <- atomicModifyIORef riterloop $ \n -> (n+1,n)- if n==100+ if n==1 then do when (not $ null x) $ hPutStr stderr x >> hPutStrLn stderr ": can't read, skip" writeIORef riterloop 0@@ -1353,20 +1437,23 @@ else mapM' f xss riterloop= unsafePerformIO $ newIORef 0- breakSlash :: [String] -> String -> [String]- breakSlash [] ""= [""]- breakSlash s ""= s- breakSlash res ('\"':s)=- let (r,rest) = span(/= '\"') s- in breakSlash (res++[r]) $ tail1 rest+breakSlash :: [String] -> String -> [String]+breakSlash [] ""= [""]+breakSlash s ""= s+breakSlash res ('\"':s)=+ let (r,rest) = span(/= '\"') s+ in breakSlash (res++[r]) $ tail1 rest - breakSlash res s=- let (r,rest) = span(\x -> x /= '/' && x /= ' ') s- in breakSlash (res++[r]) $ tail1 rest+breakSlash res s=+ let (r,rest) = span(\x -> (not $ elem x "/,:") && x /= ' ') s+ in breakSlash (res++[r]) $ tail1 rest - tail1 []= []- tail1 x= tail x+tail1 []= []+tail1 x= tail x +-- >>> breakSlash [] "test.hs/0/\"-prof -auto\""+-- ["test.hs","0","-prof -auto"]+-- @@ -1417,10 +1504,19 @@ keep mx = do liftIO $ hSetBuffering stdout LineBuffering rexit <- newEmptyMVar- forkIO $ do+ void $ forkIO $ do -- liftIO $ putMVar rexit $ Right Nothing- runTransient $ do- onException $ \(e :: SomeException ) -> liftIO $ putStr "keep block: " >> print e+ let logFile= "trans.log"++ void $ runTransient $ do+ liftIO $ removeFile logFile `catch` \(e :: IOError) -> return ()+ onException $ \(e :: SomeException) -> do+ --top <- topState+ liftIO $ print e+ --showThreads top`+ --liftIO $ appendFile logFile $ show e ++ "\n" -- `catch` \(e:: IOError) -> exc+ empty+ onException $ \(e :: IOException) -> do when (ioeGetErrorString e == "resource busy") $ do liftIO $ do print e ; putStrLn "EXITING!!!"; putMVar rexit Nothing@@ -1428,11 +1524,17 @@ st <- get setData $ Exit rexit- (abduce >> labelState (fromString "input") >> liftIO inputLoop >> empty)- <|> do+ + do option "options" "show all options" mbs <- liftIO $ readIORef rcb- liftIO $ mapM_ (\c ->do putStr c; putStr "|") $ map fst mbs++ liftIO $ mapM_ (\c ->do putStr c; putStr "|") $ map (\(fst,_,_) -> fst)mbs++ d <- input' (Just "n") (\x -> x=="y" || x =="n" || x=="Y" || x=="N") "\nDetails? N/y "+ when (d == "y") $+ let line (x,y,_)= putStr y -- do putStr x; putStr "\t\t"; putStrLn y+ in liftIO $ mapM_ line mbs liftIO $ putStrLn "" empty <|> do@@ -1440,11 +1542,18 @@ liftIO $ showThreads st empty <|> do+ option "errs" "show exceptions log"+ c <- liftIO $ readFile logFile `catch` \(e:: IOError) -> return ""+ liftIO . putStrLn $ if null c then "no errors logged" else c+ empty+ {-+ <|> do option "log" "inspect the log of a thread" th <- input (const True) "thread number>" ml <- liftIO $ getStateFromThread th st liftIO $ print $ fmap (\(Log _ _ log _) -> reverse log) ml empty+ -} <|> do option "end" "exit" liftIO $ putStrLn "exiting..."@@ -1454,9 +1563,16 @@ empty <|> mx+#ifndef ghcjs_HOST_OS+ <|> do + abduce + liftIO $ execCommandLine+ labelState (fromString "input") + liftIO inputLoop + empty+#endif return ()- threadDelay 10000- execCommandLine+ stay rexit where@@ -1464,26 +1580,37 @@ type1= undefined -- | Same as `keep` but does not read from the standard input, and therefore--- the async input APIs ('option' and 'input') cannot be used in the monad.--- However, keyboard input can still be passed via command line arguments as+-- the async input APIs ('option' and 'input') cannot respond interactively.+-- However, input can still be passed via command line arguments as -- described in 'keep'. Useful for debugging or for creating background tasks, -- as well as to embed the Transient monad inside another computation. It--- returns either the value returned by `exit`. or Nothing, when there are no+-- returns either the value returned by `exit` or Nothing, when there are no -- more threads running --+ keep' :: Typeable a => TransIO a -> IO (Maybe a) keep' mx = do liftIO $ hSetBuffering stdout LineBuffering- rexit <- newEmptyMVar- forkIO $ do- runTransient $ do- onException $ \(e :: SomeException ) -> liftIO $ putStr "keep block: " >> print e- + rexit <- newEmptyMVar + void $ forkIO $ do+ void $ runTransient $ do+ onException $ \(e :: SomeException ) -> do+ top <- topState+ liftIO $ do+ th <- myThreadId+ putStr $ show th+ putStr ": "+ print e+ putStrLn "Threads:"+ showThreads top+ empty+ onException $ \(e :: IOException) -> do when (ioeGetErrorString e == "resource busy") $ do liftIO $ do print e ; putStrLn "EXITING!!!"; putMVar rexit Nothing liftIO $ putMVar rexit Nothing empty+ setData $ Exit rexit mx @@ -1498,8 +1625,9 @@ when (isJust mindex) $ do let i= fromJust mindex +1 when (length args >= i) $ do- let path= args !! i- putStr "Executing: " >> print path+ let path= args !! i + --print $ drop (i-1) args+ --putStr "Executing: " >> print path processLine path @@ -1516,6 +1644,7 @@ + -- | If the first parameter is 'Nothing' return the second parameter otherwise -- return the first parameter.. onNothing :: Monad m => m (Maybe b) -> m b -> m b@@ -1555,13 +1684,11 @@ {-# NOINLINE onBack #-} onBack :: (Typeable b, Show b) => TransientIO a -> ( b -> TransientIO a) -> TransientIO a onBack ac bac = registerBack (typeof bac) $ Transient $ do+ tr "onBack" Backtrack mreason stack <- getData `onNothing` (return $ backStateOf (typeof bac)) runTrans $ case mreason of Nothing -> ac -- !> "ONBACK NOTHING"- Just reason -> do-- -- setState $ Backtrack mreason $ tail stack -- to avoid recursive call tot he same handler- bac reason -- !> ("ONBACK JUST",reason)+ Just reason -> bac reason -- !> ("ONBACK JUST",reason) where typeof :: (b -> TransIO a) -> b typeof = undefined@@ -1579,25 +1706,23 @@ {-# NOINLINE registerUndo #-} registerBack :: (Typeable b, Show b) => b -> TransientIO a -> TransientIO a registerBack witness f = Transient $ do- cont@(EventF _ x _ _ _ _ _ _ _ _ _) <- get+ tr "registerBack"+ cont@(EventF _ x _ _ _ _ _ _ _ _ _ _ _) <- get -- if isJust (event cont) then return Nothing else do md <- getData `asTypeOf` (Just <$> return (backStateOf witness)) case md of- Just (Backtrack b []) -> setData $ Backtrack b [cont]- Just (bss@(Backtrack b (bs@((EventF _ x' _ _ _ _ _ _ _ _ _):_)))) ->- when (isNothing b) $ do- addrx <- addr x- addrx' <- addr x' -- to avoid duplicate backtracking points- setData $ if addrx == addrx' then bss else Backtrack b (cont:bs)- --setData $ Backtrack b (cont:bs)+ Just (Backtrack b []) -> setData $ Backtrack b [cont] + Just (bss@(Backtrack b (bs@((EventF _ x' _ _ _ _ _ _ _ _ _ _ _):_)))) ->+ when (isNothing b) $ + setData $ Backtrack b (cont:bs) Nothing -> setData $ Backtrack mwit [cont] - runTrans f+ runTrans $ return () >> f where mwit= Nothing `asTypeOf` (Just witness)- addr x = liftIO $ return . hashStableName =<< (makeStableName $! x)+ --addr x = liftIO $ return . hashStableName =<< (makeStableName $! x) registerUndo :: TransientIO a -> TransientIO a@@ -1605,6 +1730,7 @@ -- XXX Should we enforce retry of the same track which is being undone? If the -- user specifies a different track would it make sense?+-- see https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?at=5ef46626e0e5673398d33afb -- -- | For a given undo track type, stop executing more backtracking actions and -- resume normal execution in the forward direction. Used inside an undo@@ -1615,6 +1741,11 @@ Backtrack _ stack <- getData `onNothing` ( return $ backStateOf reason) setData $ Backtrack(Nothing `asTypeOf` Just reason) stack +-- | put at the end of an backtrack handler intended to backtrack to other previous handlers.+-- This is the default behaviour in transient. `backtrack` is in order to keep the type compiler happy+backtrack :: TransIO a+backtrack= return $ error "backtrack should be called at the end of an exception handler with no `forward`, `continue` or `retry` on it"+ -- | 'forward' for the default undo track; equivalent to @forward ()@. retry= forward () @@ -1630,6 +1761,7 @@ -- back :: (Typeable b, Show b) => b -> TransIO a back reason = do+ tr "back" bs <- getData `onNothing` return (backStateOf reason) goBackt bs -- !>"GOBACK" @@ -1642,15 +1774,18 @@ goBackt (Backtrack _ [] )= empty goBackt (Backtrack b (stack@(first : bs)) )= do- setData $ Backtrack (Just reason) stack- x <- runClosure first -- !> ("RUNCLOSURE",length stack)+ setData $ Backtrack (Just reason) bs --stack+ x <- runClosure first !> ("RUNCLOSURE",length stack) Backtrack back bs' <- getData `onNothing` return (backStateOf reason) case back of- Nothing -> runContinuation first x -- !> "FORWARD EXEC"+ Nothing -> do+ setData $ Backtrack (Just reason) stack+ st <- get+ runContinuation first x `catcht` (\e -> liftIO(exceptBack st e) >> empty) !> "FORWARD EXEC" justreason -> do- setData $ Backtrack justreason bs- goBackt $ Backtrack justreason bs -- !> ("BACK AGAIN",back)+ --setData $ Backtrack justreason bs+ goBackt $ Backtrack justreason bs !> ("BACK AGAIN",back) empty backStateOf :: (Show a, Typeable a) => a -> Backtrack a@@ -1670,9 +1805,9 @@ return $ BackPoint point -- | install a callback in a backPoint-onBackPoint (BackPoint ref) handler= liftIO $ atomicModifyIORef ref $ \rs -> (handler:rs,()) -+onBackPoint :: MonadIO m => BackPoint t -> (t -> TransIO ()) -> m ()+onBackPoint (BackPoint ref) handler= liftIO $ atomicModifyIORef ref $ \rs -> (handler:rs,()) -- | 'back' for the default undo track; equivalent to @back ()@. --@@ -1740,7 +1875,7 @@ -- When an exception backtracking reach the backPoint it executes all the handlers registered for it. -- -- Use case: suppose that when a connection fails, you need to stop a process.--- This process may not be directly involved in the connection. Perhaps it was initiated after the socket is being read+-- This process may not be started before the connection. Perhaps it was initiated after the socket read -- so an exception will not backtrack trough the process, since it is downstream, not upstream. The process may -- be even unrelated to the connection, in other branch of the computation. --@@ -1757,16 +1892,16 @@ --- in conjunction with `backPoint` it set a handler that will be called when backtracking pass trough the point+-- | in conjunction with `backPoint` it set a handler that will be called when backtracking pass +-- trough the point onExceptionPoint :: Exception e => BackPoint e -> (e -> TransIO()) -> TransIO () onExceptionPoint= onBackPoint onException' :: Exception e => TransIO a -> (e -> TransIO a) -> TransIO a onException' mx f= onAnyException mx $ \e -> do- return () !> "EXCEPTION HANDLER EXEC" + --return () !> "EXCEPTION HANDLER EXEC" case fromException e of- Nothing -> do Backtrack r stack <- getData `onNothing` return (backStateOf e) setData $ Backtrack r $ tail stack@@ -1787,11 +1922,12 @@ (do case event st of Nothing -> do- r <- runTrans mx+ r <- runTrans mx modify $ \s -> s{event= Just $ unsafeCoerce r} runCont st- was <- getData `onNothing` return NoRemote- when (was /= WasRemote) $ setData WasParallel+ -- was <- gets execMode -- getData `onNothing` return Serial+ -- when (was /= Remote) $ modify $ \s -> s{execMode= Parallel}+ modify $ \s -> s{execMode=let rs= execMode s in if rs /= Remote then Parallel else rs} return Nothing @@ -1802,10 +1938,17 @@ put st' return mr -exceptBack st = \(e ::SomeException) -> do -- recursive catch itself+exceptBack st = \(e ::SomeException) -> do + tr "catched" runStateT ( runTrans $ back e ) st -- !> "EXCEPTBACK"- `catch` exceptBack st+ -- `catch` exceptBack st -- removed +-- re execute the first argument as long as the exception is produced within the argument. +-- The second argument is executed before every re-execution+-- if the second argument executes `empty` the execution is aborted.++whileException :: Exception e => TransIO b -> (e -> TransIO()) -> TransIO b +whileException mx fixexc = mx `catcht` \e -> do fixexc e; whileException mx fixexc -- | Delete all the exception handlers registered till now. cutExceptions :: TransIO ()@@ -1815,20 +1958,32 @@ -- handlers and resume normal execution from this point on. continue :: TransIO () continue = forward (undefined :: SomeException) -- !> "CONTINUE"- -- | catch an exception in a Transient block -- -- The semantic is the same than `catch` but the computation and the exception handler can be multirhreaded catcht :: Exception e => TransIO b -> (e -> TransIO b) -> TransIO b catcht mx exc= do+ st <- get+ (mx,st') <- liftIO $ runStateT ( runTrans $ mx ) st `catch` \e -> runStateT ( runTrans $ exc e ) st+ put st' + case mx of+ Just x -> return x+ Nothing -> empty+-- | catch an exception in a Transient block+--+-- The semantic is the same than `catch` but the computation and the exception handler can be multirhreaded+catcht' :: Exception e => TransIO b -> (e -> TransIO b) -> TransIO b+catcht' mx exc= do rpassed <- liftIO $ newIORef False sandbox $ do r <- onException' mx (\e -> do passed <- liftIO $ readIORef rpassed+ return () !> ("CATCHT passed", passed) if not passed then continue >> exc e else do Backtrack r stack <- getData `onNothing` return (backStateOf e) setData $ Backtrack r $ tail stack- back e+ back e + return () !> "AFTER BACK" empty ) liftIO $ writeIORef rpassed True
src/Transient/Logged.hs view
@@ -1,3 +1,4 @@+ {-#Language OverloadedStrings, FlexibleContexts #-} ----------------------------------------------------------------------------- -- -- Module : Transient.Logged@@ -17,8 +18,7 @@ -- contains purely application level state, and is therefore independent of the -- underlying machine architecture. The saved logs can be sent across the wire -- to another machine and the computation can then be resumed on that machine.--- We can also save the log to gather diagnostic information, especially in--- 'finish' blocks.+-- We can also save the log to gather diagnostic information. -- -- The following example illustrates the APIs. In its first run 'suspend' saves -- the state in a directory named @logs@ and exits, in the second run it@@ -37,36 +37,183 @@ ----------------------------------------------------------------------------- {-# LANGUAGE CPP, ExistentialQuantification, FlexibleInstances, ScopedTypeVariables, UndecidableInstances #-} module Transient.Logged(-Loggable, logged, received, param,+Loggable(..), logged, received, param, getLog, exec,wait, emptyLog,+ #ifndef ghcjs_HOST_OS- suspend, checkpoint, restore,+ suspend, checkpoint, rerun, restore, #endif--- * low level-fromIDyn,maybeFromIDyn,toIDyn++Log(..), toLazyByteString, byteString, lazyByteString, Raw(..) ) where import Data.Typeable import Unsafe.Coerce-import Transient.Base+import Transient.Internals import Transient.Indeterminism(choose)-import Transient.Internals -- (onNothing,reads1,IDynamic(..),Log(..),LogElem(..),RemoteStatus(..),StateIO)+--import Transient.Internals -- (onNothing,reads1,IDynamic(..),Log(..),LogElem(..),execMode(..),StateIO)+import Transient.Parse import Control.Applicative import Control.Monad.State import System.Directory-import Control.Exception -import Control.Monad-import Control.Concurrent.MVar+import Control.Exception+--import Control.Monad import qualified Data.ByteString.Lazy.Char8 as BS import qualified Data.ByteString.Char8 as BSS+import qualified Data.Map as M+-- #ifndef ghcjs_HOST_OS+import Data.ByteString.Builder+import Data.Monoid+import System.Random+-- #else+--import Data.JSString hiding (empty)+-- #endif +++-- #ifndef ghcjs_HOST_OS+-- pack= BSS.pack++-- #else+{-+newtype Builder= Builder(JSString -> JSString)+instance Monoid Builder where+ mappend (Builder fx) (Builder fy)= Builder $ \next -> fx (fy next)+ mempty= Builder id++instance Semigroup Builder where+ (<>)= mappend++byteString :: JSString -> Builder+byteString ss= Builder $ \s -> ss <> s+lazyByteString = byteString+++toLazyByteString :: Builder -> JSString+toLazyByteString (Builder b)= b mempty+-}+-- #endif++exec= byteString "e/"+wait= byteString "w/"++class (Show a, Read a,Typeable a) => Loggable a where+ serialize :: a -> Builder+ serialize = byteString . BSS.pack . show++ deserializePure :: BS.ByteString -> Maybe(a, BS.ByteString)+ deserializePure s = r+ where+ r= case reads $ BS.unpack s of -- `traceShow` ("deserialize",typeOf $ typeOf1 r,s) of+ [] -> Nothing !> "Nothing"+ (r,t): _ -> return (r, BS.pack t)++ typeOf1 :: Maybe(a, BS.ByteString) -> a+ typeOf1= undefined++ deserialize :: TransIO a+ deserialize = x+ where+ x= withGetParseString $ \s -> case deserializePure s of+ Nothing -> empty+ Just x -> return x++instance Loggable ()++instance Loggable Bool where + serialize b= if b then "t" else "f"+ deserialize = withGetParseString $ \s -> + if (BS.head $ BS.tail s) /= '/' + then empty + else+ let h= BS.head s+ tail= BS.tail s+ in if h== 't' then return (True,tail) else if h== 'f' then return (False, tail) else empty ++instance Loggable Int+instance Loggable Integer++instance (Typeable a, Loggable a) => Loggable[a] +-- serialize x= if typeOf x= typeOf (undefined :: String) then BS.pack x else BS.pack $ show x+-- deserialize= let [(s,r)]= +++++instance Loggable Char+instance Loggable Float+instance Loggable Double+instance Loggable a => Loggable (Maybe a)+instance (Loggable a,Loggable b) => Loggable (a,b)+instance (Loggable a,Loggable b, Loggable c) => Loggable (a,b,c)+instance (Loggable a,Loggable b, Loggable c,Loggable d) => Loggable (a,b,c,d)+instance (Loggable a,Loggable b, Loggable c,Loggable d,Loggable e) => Loggable (a,b,c,d,e)+instance (Loggable a,Loggable b, Loggable c,Loggable d,Loggable e,Loggable f) => Loggable (a,b,c,d,e,f)+instance (Loggable a,Loggable b, Loggable c,Loggable d,Loggable e,Loggable f,Loggable g) => Loggable (a,b,c,d,e,f,g)+instance (Loggable a,Loggable b, Loggable c,Loggable d,Loggable e,Loggable f,Loggable g,Loggable h) => Loggable (a,b,c,d,e,f,g,h)+instance (Loggable a,Loggable b, Loggable c,Loggable d,Loggable e,Loggable f,Loggable g,Loggable h,Loggable i) => Loggable (a,b,c,d,e,f,g,h,i)+++instance (Loggable a, Loggable b) => Loggable (Either a b)+-- #ifdef ghcjs_HOST_OS+++-- intDec i= Builder $ \s -> pack (show i) <> s+-- int64Dec i= Builder $ \s -> pack (show i) <> s++-- #endif+instance (Loggable k, Ord k, Loggable a) => Loggable (M.Map k a) where+ serialize v= intDec (M.size v) <> M.foldlWithKey' (\s k x -> s <> "/" <> serialize k <> "/" <> serialize x ) mempty v+ deserialize= do+ len <- int+ list <- replicateM len $+ (,) <$> (tChar '/' *> deserialize)+ <*> (tChar '/' *> deserialize)+ return $ M.fromList list+ #ifndef ghcjs_HOST_OS-import System.Random+instance Loggable BS.ByteString where+ serialize str = lazyByteString str+ deserialize= tTakeWhile (/= '/') #endif +#ifndef ghcjs_HOST_OS+instance Loggable BSS.ByteString where+ serialize str = byteString str+ deserialize = tTakeWhile (/= '/') >>= return . BS.toStrict+#endif+instance Loggable SomeException +newtype Raw= Raw BS.ByteString deriving (Read,Show)+instance Loggable Raw where+ serialize (Raw str)= lazyByteString str+ deserialize= Raw <$> do+ s <- notParsed+ BS.length s `seq` return s --force the read till the end +data Log = Log{ recover :: Bool, buildLog :: Builder, fulLog :: Builder, lengthFull:: Int, hashClosure :: Int}++++ #ifndef ghcjs_HOST_OS++++-- | Reads the saved logs from the @logs@ subdirectory of the current+-- directory, restores the state of the computation from the logs, and runs the+-- computation. The log files are maintained.+-- It could be used for the initial configuration of a program.+rerun :: String -> TransIO a -> TransIO a+rerun path proc = do+ liftIO $ do+ r <- doesDirectoryExist path+ when (not r) $ createDirectory path+ setCurrentDirectory path+ restore' proc False+++ logs= "logs/" -- | Reads the saved logs from the @logs@ subdirectory of the current@@ -74,23 +221,26 @@ -- computation. The log files are removed after the state has been restored. -- restore :: TransIO a -> TransIO a-restore proc= do+restore proc= restore' proc True++restore' proc delete= do liftIO $ createDirectory logs `catch` (\(e :: SomeException) -> return ()) list <- liftIO $ getDirectoryContents logs `catch` (\(e::SomeException) -> return []) if null list || length list== 2 then proc else do let list'= filter ((/=) '.' . head) list- file <- choose list' -- !> list'+ file <- choose list' - logstr <- liftIO $ readFile (logs++file)- let log= length logstr `seq` read' logstr+ log <- liftIO $ BS.readFile (logs++file) - log `seq` setData (Log True (reverse log) log)- liftIO $ remove $ logs ++ file+ let logb= lazyByteString log+ setData Log{recover= True,buildLog= logb,fulLog= logb,lengthFull= 0, hashClosure= 0}+ setParseString log+ when delete $ liftIO $ remove $ logs ++ file proc where- read'= fst . head . reads1+ -- read'= fst . head . reads1 remove f= removeFile f `catch` (\(e::SomeException) -> remove f) @@ -103,151 +253,168 @@ -- subdirectory of the current directory. Each thread's log is saved in a -- separate file. ---suspend :: Typeable a => a -> TransIO a+suspend :: Typeable a => a -> TransIO a suspend x= do- Log recovery _ log _ <- getData `onNothing` return (Log False [] [] 0)- if recovery then return x else do- logAll log+ log <- getLog+ if (recover log) then return x else do+ logAll $ fulLog log exit x + -- | Saves the accumulated logs of the current computation, like 'suspend', but -- does not exit.-checkpoint :: TransIO ()+checkpoint :: TransIO () checkpoint = do- Log recovery _ log _ <- getData `onNothing` return (Log False [] [] 0)- if recovery then return () else logAll log+ log <- getLog+ if (recover log) then return () else logAll $ fulLog log logAll log= liftIO $do newlogfile <- (logs ++) <$> replicateM 7 (randomRIO ('a','z')) logsExist <- doesDirectoryExist logs when (not logsExist) $ createDirectory logs- writeFile newlogfile $ show log+ BS.writeFile newlogfile $ toLazyByteString log -- :: TransIO ()--#endif--maybeFromIDyn :: Loggable a => IDynamic -> Maybe a-maybeFromIDyn (IDynamic x)= r- where- r= if typeOf (Just x) == typeOf r then Just $ unsafeCoerce x else Nothing --maybeFromIDyn (IDyns s) = case reads s of- [] -> Nothing - [(x,"")] -> Just x +#else+rerun :: TransIO a -> TransIO a+rerun = const empty -fromIDyn :: Loggable a => IDynamic -> a-fromIDyn (IDynamic x)=r where r= unsafeCoerce x -- !> "coerce" ++ " to type "++ show (typeOf r)+suspend :: TransIO ()+suspend= empty -fromIDyn (IDyns s)=r `seq`r where r= read' s -- !> "read " ++ s ++ " to type "++ show (typeOf r)+checkpoint :: TransIO ()+checkpoint= empty +restore :: TransIO a -> TransIO a+restore= const empty +#endif -toIDyn x= IDynamic x+getLog :: TransMonad m => m Log+getLog= getData `onNothing` return emptyLog +emptyLog= Log False mempty mempty 0 0 --- | Run the computation, write its result in a log in the parent computation+-- | Run the computation, write its result in a log in the state -- and return the result. If the log already contains the result of this -- computation ('restore'd from previous saved state) then that result is used -- instead of running the computation again. ----- 'logged' can be used for computations inside a 'logged' computation. Once+-- 'logged' can be used for computations inside a nother 'logged' computation. Once -- the parent computation is finished its internal (subcomputation) logs are -- discarded. --++ logged :: Loggable a => TransIO a -> TransIO a-logged mx = Transient $ do- Log recover rs full hash <- getData `onNothing` return ( Log False [][] 0)- runTrans $- case (recover ,rs) of -- !> ("logged enter",recover,rs,reverse full) of- (True, Var x: rs') -> do- return () -- !> ("Var:", x)- setData $ Log True rs' full (hash+ 10000000)- return $ fromIDyn x- - - (True, Exec:rs') -> do- setData $ Log True rs' full (hash + 1000)- mx -- !> "Exec"- - (True, Wait:rs') -> do- setData $ Log True rs' full (hash + 100000)- setData WasParallel- empty -- !> "Wait"- - _ -> do+logged mx = do+ log <- getLog - setData $ Log False (Exec : rs) (Exec: full) (hash + 1000) -- !> ("setLog False", Exec:rs)- - r <- mx <** do setData $ Log False (Wait: rs) (Wait: full) (hash+ 100000)- -- when p1 <|> p2, to avoid the re-execution of p1 at the- -- recovery when p1 is asynchronous or return empty+ let full= fulLog log+ rest <- giveParseString - Log recoverAfter lognew _ _ <- getData `onNothing` return ( Log False [][] 0)- let add= Var (toIDyn r): full- if recoverAfter && (not $ null lognew) -- !> ("recoverAfter", recoverAfter)- then do- setData WasParallel- (setData $ Log True lognew (reverse lognew ++ add) (hash + 10000000) )- -- !> ("recover",reverse (reverse lognew ++add))- else if recoverAfter && (null lognew) then do - -- showClosure- setData $ Log False [] add (hash + 10000000) -- !> ("recover2",reverse add)- else do- -- showClosure- (setData $ Log False (Var (toIDyn r):rs) add (hash +10000000)) -- !> ("restore", reverse $ (Var (toIDyn r):rs))- - return r+ if recover log -- !> ("recover",recover log)+ then+ if not $ BS.null rest + then recoverIt log -- !> "RECOVER" -- exec1 log <|> wait1 log <|> value log+ else do+ setData log{buildLog=mempty}+ notRecover full log !> "NOTRECOVER" + else notRecover full log+ where+ notRecover full log= do + let rs = buildLog log + setData $ Log False (rs <> exec) (full <> exec) (lengthFull log +1) (hashClosure log + 1000) -- !> ("setLog False", Exec:rs) --------- parsing the log for API's+ r <- mx <** do setData $ Log False ( rs <> wait) (full <> wait) (lengthFull log +1) (hashClosure log + 100000)+ -- when p1 <|> p2, to avoid the re-execution of p1 at the+ -- recovery when p1 is asynchronous or return empty -received :: Loggable a => a -> TransIO ()-received n=Transient $ do+ log' <- getLog - Log recover rs full hash <- getData `onNothing` return ( Log False [][] 0)- return () !> ("RECEIVED log, n", rs,n) - case rs of - [] -> return Nothing- Var (IDyns s):t -> if s == show1 n+ let len= lengthFull log'+ add= full <> serialize r <> byteString "/" -- Var (toIDyn r): full+ recoverAfter= recover log'+ lognew= buildLog log'++ rest <- giveParseString+ if recoverAfter && not (BS.null rest) then do- return() !> "valid"- setData $ Log recover t full hash- return $ Just ()- else return Nothing- _ -> return Nothing- where- show1 x= if typeOf x == typeOf "" then unsafeCoerce x - else if typeOf x== typeOf (undefined :: BS.ByteString) then unsafeCoerce x- else if typeOf x== typeOf (undefined :: BSS.ByteString) then unsafeCoerce x- else show x+ modify $ \s -> s{execMode= Parallel} + setData $ log'{recover= True, fulLog= lognew <> add, lengthFull= lengthFull log+ len,hashClosure= hashClosure log + 10000000}+ -- !> ("recover", toLazyByteString $ lognew <> add) -param :: Loggable a => TransIO a-param= res where- res= Transient $ do- Log recover rs full hash<- getData `onNothing` return ( Log False [][] 0) - return () !> ("PARAM",rs)- case rs of+ else - [] -> return Nothing- Var (IDynamic v):t ->do- return () !> ("IDyn", show v)- setData $ Log recover t full hash- return $ cast v- Var (IDyns s):t -> do- return () !> ("IDyn",s)- let mr = reads1 s `asTypeOf` type1 res+ setData $ Log{recover= False, buildLog=rs <> serialize r <> byteString "/", fulLog= add,lengthFull= len+1, hashClosure=hashClosure log +10000000}+ - case mr of- [] -> return Nothing- (v,r):_ -> do- setData $ Log recover t full hash- return $ Just v- _ -> return Nothing - where- type1 :: TransIO a -> [(a,String)]- type1= error "type1: typelevel"+ return r++ recoverIt log= do+ s <- giveParseString+ case BS.splitAt 2 s of+ ("e/",r) -> do+ setData $ log{ + lengthFull= lengthFull log +1, hashClosure= hashClosure log + 1000}+ setParseString r -- !> "Exec"+ mx++ ("w/",r) -> do+ setData $ log{ + lengthFull= lengthFull log +1, hashClosure= hashClosure log + 100000}+ setParseString r+ modify $ \s -> s{execMode= Parallel} --setData Parallel+ empty -- !> "Wait"++ _ -> value log++ value log= r+ where+ typeOfr :: TransIO a -> a+ typeOfr _= undefined+ r= do+ -- return() !> "logged value"+ x <- deserialize <|> do+ psr <- giveParseString+ error (show("error parsing",psr,"to",typeOf $ typeOfr r))+ + tChar '/'++ setData $ log{recover= True -- , = rs'+ ,lengthFull= lengthFull log +1,hashClosure= hashClosure log + 10000000}+ return x++++++-------- parsing the log for API's++received :: (Loggable a, Eq a) => a -> TransIO ()+received n= Transient.Internals.try $ do+ r <- param+ if r == n then return () else empty++param :: (Loggable a, Typeable a) => TransIO a+param = r where+ r= do+ let t = typeOf $ type1 r+ (Transient.Internals.try $ tChar '/' >> return ())<|> return () --maybe there is a '/' to drop+ --(Transient.Internals.try $ tTakeWhile (/= '/') >>= liftIO . print >> empty) <|> return ()+ if t == typeOf (undefined :: String) then return . unsafeCoerce . BS.unpack =<< tTakeWhile' (/= '/')+ else if t == typeOf (undefined :: BS.ByteString) then return . unsafeCoerce =<< tTakeWhile' (/= '/')+ else if t == typeOf (undefined :: BSS.ByteString) then return . unsafeCoerce . BS.toStrict =<< tTakeWhile' (/= '/')+ else deserialize -- <* tChar '/'+++ where+ type1 :: TransIO x -> x+ type1 = undefined++
+ src/Transient/Mailboxes.hs view
@@ -0,0 +1,95 @@+ {-# LANGUAGE ExistentialQuantification #-}+module Transient.Mailboxes where++import Transient.Internals+import Transient.EVars+import qualified Data.Map as M+import Data.IORef+import Data.Typeable+import System.IO.Unsafe+import Unsafe.Coerce+import Control.Monad.IO.Class++mailboxes :: IORef (M.Map MailboxId (EVar SData))+mailboxes= unsafePerformIO $ newIORef M.empty++data MailboxId = forall a .(Typeable a, Ord a) => MailboxId a TypeRep+--type SData= ()+instance Eq MailboxId where+ id1 == id2 = id1 `compare` id2== EQ++instance Ord MailboxId where+ MailboxId n t `compare` MailboxId n' t'=+ case typeOf n `compare` typeOf n' of+ EQ -> case n `compare` unsafeCoerce n' of+ EQ -> t `compare` t'+ LT -> LT+ GT -> GT++ other -> other++instance Show MailboxId where+ show ( MailboxId _ t) = show t++-- | write to the mailbox+-- Mailboxes are node-wide, for all processes that share the same connection data, that is, are under the+-- same `listen` or `connect`+-- while EVars are only visible by the process that initialized it and his children.+-- Internally, the mailbox is in a well known EVar stored by `listen` in the `Connection` state.+putMailbox :: Typeable val => val -> TransIO ()+putMailbox = putMailbox' (0::Int)++-- | write to a mailbox identified by an identifier besides the type+putMailbox' :: (Typeable key, Ord key, Typeable val) => key -> val -> TransIO ()+putMailbox' idbox dat= do+ let name= MailboxId idbox $ typeOf dat+ mbs <- liftIO $ readIORef mailboxes+ let mev = M.lookup name mbs+ case mev of+ Nothing -> newMailbox name >> putMailbox' idbox dat+ Just ev -> writeEVar ev $ unsafeCoerce dat+++newMailbox :: MailboxId -> TransIO ()+newMailbox name= do+-- return () -- !> "newMailBox"+ ev <- newEVar+ liftIO $ atomicModifyIORef mailboxes $ \mv -> (M.insert name ev mv,())++++-- | get messages from the mailbox that matches with the type expected.+-- The order of reading is defined by `readTChan`+-- This is reactive. it means that each new message trigger the execution of the continuation+-- each message wake up all the `getMailbox` computations waiting for it.+getMailbox :: Typeable val => TransIO val+getMailbox = getMailbox' (0 :: Int)++-- | read from a mailbox identified by an identifier besides the type+getMailbox' :: (Typeable key, Ord key, Typeable val) => key -> TransIO val+getMailbox' mboxid = x where+ x = do+ let name= MailboxId mboxid $ typeOf $ typeOfM x+ mbs <- liftIO $ readIORef mailboxes+ let mev = M.lookup name mbs+ case mev of+ Nothing ->newMailbox name >> getMailbox' mboxid+ Just ev ->unsafeCoerce $ readEVar ev++ typeOfM :: TransIO a -> a+ typeOfM = undefined++-- | delete all subscriptions for that mailbox expecting this kind of data+deleteMailbox :: Typeable a => a -> TransIO ()+deleteMailbox = deleteMailbox' (0 ::Int)++-- | clean a mailbox identified by an Int and the type+deleteMailbox' :: (Typeable key, Ord key, Typeable a) => key -> a -> TransIO ()+deleteMailbox' mboxid witness= do+ let name= MailboxId mboxid $ typeOf witness+ mbs <- liftIO $ readIORef mailboxes+ let mev = M.lookup name mbs+ case mev of+ Nothing -> return()+ Just ev -> do cleanEVar ev+ liftIO $ atomicModifyIORef mailboxes $ \bs -> (M.delete name bs,())
src/Transient/Parse.hs view
@@ -1,9 +1,22 @@ {-#LANGUAGE FlexibleContexts, ExistentialQuantification, ScopedTypeVariables, OverloadedStrings, TypeSynonymInstances, FlexibleInstances #-}-module Transient.Parse where+module Transient.Parse(+-- * Setting the stream+setParseStream, setParseString, withParseString, withParseStream,+-- * parsing+string, tDropUntilToken, tTakeUntilToken, integer, hex, int, double, tChar,anyChar,+manyTill, chainManyTill,between, symbol,parens, braces,angles,brackets,+semi, comma, dot,colon, sepBy, sepBy1, chainSepBy, chainSepBy1,chainMany,+commaSep, semiSep, commaSep1, dropSpaces,dropTillEndOfLine,+parseString, tTakeWhile,tTakeUntil, tTakeWhile', tTake, tDrop, tDropUntil, tPutStr,+isDone,dropUntilDone, +-- * giving the parse string+withGetParseString, giveParseString,+-- * debug+notParsed, getParseBuffer,clearParseBuffer, showNext,+-- Composing parsing processes+(|-)) where+ import Transient.Internals-import Transient.Indeterminism-import Data.String-import Data.Typeable import Control.Applicative import Data.Char import Data.Monoid@@ -13,62 +26,111 @@ import Control.Monad.State -- import Control.Exception (throw,IOException) import Control.Concurrent.MVar--+import Data.Maybe(fromJust) import qualified Data.ByteString.Lazy.Char8 as BS+import Data.ByteString.Builder+import Control.Exception hiding (try)+import Data.IORef+import Control.Concurrent+import Data.Maybe -- | set a stream of strings to be parsed-setParseStream :: IO (StreamData BS.ByteString) -> TransIO ()+setParseStream :: TransMonad m => TransIO (StreamData BS.ByteString) -> m ()+setParseStream iox= modify $ \s -> s{execMode=Serial,parseContext= ParseContext iox "" (unsafePerformIO $ newIORef False)} -- setState $ ParseContext iox "" -setParseStream iox= do delData NoRemote; setState $ ParseContext iox "" -- | set a string to be parsed-setParseString :: BS.ByteString -> TransIO ()-setParseString x = do delData NoRemote; setState $ ParseContext (return SDone) x +setParseString :: TransMonad m => BS.ByteString -> m ()+setParseString x = modify $ \s -> s{execMode=Serial,parseContext= ParseContext (return SDone) x (unsafePerformIO $ newIORef False)} -- setState $ ParseContext (return SDone) x -withParseString :: BS.ByteString -> TransIO a -> TransIO a++withParseString :: BS.ByteString -> TransIO a -> TransIO a withParseString x parse= do- p@(ParseContext c str) <- getState <|> return(ParseContext (return SDone) mempty)+ p <- gets parseContext -- getState <|> return(ParseContext (return SDone) mempty) setParseString x r <- parse- setState (ParseContext c (str :: BS.ByteString))+ modify $ \s -> s{parseContext= p} --setState (ParseContext c (str :: BS.ByteString)) return r ++withParseStream stream parse= do+ p <- gets parseContext -- getState <|> return(ParseContext (return SDone) mempty)+ setParseStream stream+ r <- parse+ modify $ \s -> s{parseContext= p} --setState (ParseContext c (str :: BS.ByteString))+ return r+ -- | The parse context contains either the string to be parsed or a computation that gives an stream of -- strings or both. First, the string is parsed. If it is empty, the stream is pulled for more.-data ParseContext str = IsString str => ParseContext (IO (StreamData str)) str deriving Typeable+-- data ParseContext str = IsString str => ParseContext (IO (StreamData str)) str deriving Typeable -- | succeed if read the string given as parameter string :: BS.ByteString -> TransIO BS.ByteString-string s= withData $ \str -> do+string s= withGetParseString $ \str -> do let len= BS.length s ret@(s',_) = BS.splitAt len str- if s == s' -- !> ("parse string looked, found",s,s')- then return ret++ if s == s' -- !> ("parse string looked, found",s,s')++ then return ret else empty -- !> "STRING EMPTY" --- | fast search for a token-tDropUntilToken token= withData $ \str -> +-- | fast search for a token.+-- If the token is not found, the parse is left in the original state.+tDropUntilToken token= withGetParseString $ \str -> if BS.null str then empty else drop2 str where drop2 str=- if token `BS.isPrefixOf` str !> (BS.take 2 str)+ if token `BS.isPrefixOf` str -- !> (BS.take 2 str) then return ((),BS.drop (BS.length token) str) else if not $ BS.null str then drop2 $ BS.tail str else empty ++ tTakeUntilToken :: BS.ByteString -> TransIO BS.ByteString-tTakeUntilToken token= withData $ \str -> takeit mempty str+tTakeUntilToken token= withGetParseString $ \str -> takeit mempty str where - takeit :: BS.ByteString -> BS.ByteString -> TransIO ( BS.ByteString, BS.ByteString)+ takeit :: Builder -> BS.ByteString -> TransIO ( BS.ByteString, BS.ByteString) takeit res str= - if BS.null str then return (res,str) else + if BS.null str then empty else if token `BS.isPrefixOf` str - then return (res !> ("tTakeUntilString",res),BS.drop (BS.length token) str)- else if not $ BS.null str then takeit ( BS.snoc res (BS.head str)) $ BS.tail str else empty+ then return (toLazyByteString res ,BS.drop (BS.length token) str)+ else if not $ BS.null str then takeit ( res <> (lazyByteString $ BS.singleton $ BS.head str)) $ BS.tail str else empty + -- | read an Integer integer :: TransIO Integer+integer= withGetParseString $ \str -> + case BS.readInteger str of + Just x -> return x+ Nothing -> empty++-- | parse an hexadecimal number+hex :: TransIO Int+hex = withGetParseString $ \s -> parsehex (-1) s+ where++ parsehex v s=+ case (BS.null s,v) of+ (True, -1) -> empty+ (True,_) -> return (v, mempty)+ _ -> do+++ let h= BS.head s !> ("HEX",BS.head s)++ t= BS.tail s+ v'= if v== -1 then 0 else v+ x = if h >= '0' && h <= '9' then v' * 16 + ord(h) -ord '0'+ else if h >= 'A' && h <= 'F' then v' * 16 + ord h -ord 'A' +10+ else if h >= 'a' && h <= 'f' then v' * 16 + ord h -ord 'a' +10+ else -1+ case (v,x) of+ (-1,-1) -> empty+ (v, -1) -> return (v,s) + (_, x) -> parsehex x t +{- integer= do s <- tTakeWhile isNumber if BS.null s then empty else return $ stoi 0 s@@ -78,36 +140,60 @@ stoi :: Integer -> BS.ByteString -> Integer stoi x s| BS.null s = x | otherwise= stoi (x *10 + fromIntegral(ord (BS.head s) - ord '0')) (BS.tail s)-+-} -- | read an Int int :: TransIO Int+int= withGetParseString $ \str -> + case BS.readInt str of + Just x -> return x+ Nothing -> empty+{- int= do - s <- tTakeWhile' isNumber+ s <- tTakeWhile isNumber if BS.null s then empty else return $ stoi 0 s where stoi :: Int -> BS.ByteString -> Int stoi x s| BS.null s = x | otherwise= stoi (x *10 + (ord (BS.head s) - ord '0')) (BS.tail s)+-}+-- | read a double in floating point/scientific notation+double :: TransIO Double +double= do + ent <- integer -- takes the sign too+ frac <- fracf+ exp <- expf+ + return $ (fromIntegral ent * (10 ^ exp)) +- (( (fromIntegral $ fst $ fromJust $ BS.readInteger frac)) + /(10 ^ (fromIntegral (BS.length frac) - exp)))+ where + (+-) a b= if a >= 0 then a + b else a - b + fracf= do + tChar '.' + tTakeWhile isDigit+ <|> return "0"+ + expf= do + tChar 'e' <|> tChar 'E'+ int+ <|> return 0 --- | read many results with a parser (at least one) until a `end` parser succeed. - -+-- | read many results with a parser (at least one) until a `end` parser succeed. manyTill :: TransIO a -> TransIO b -> TransIO [a] manyTill= chainManyTill (:) -chainManyTill op p end= op <$> p <*> scan+--chainManyTill :: Monoid m => (m -> a -> a) -> TransIO m -> TransIO t -> TransIO a+chainManyTill op p end= scan where- scan = do{ end; return mempty }+ scan = do{try end; return mempty } <|> do{ x <- p; xs <- scan; return (x `op` xs) } -between open close p- = do{ open; x <- p; close; return x }+between open close p = do{ open; x <- p; close; return x } symbol = string @@ -149,105 +235,247 @@ commaSep1 p = sepBy1 p comma semiSep1 p = sepBy1 p semi -dropSpaces= withData $ \str -> return( (),BS.dropWhile isSpace str)+dropSpaces= withGetParseString $ \str -> return( (),BS.dropWhile isSpace str) +dropTillEndOfLine= withGetParseString $ \str -> return ((),BS.dropWhile ( /= '\n') str) !> "dropTillEndOfLine" -dropTillEndOfLine= withData $ \str -> return ((),BS.dropWhile ( /= '\n') str) !> "dropTillEndOfLine"----manyTill anyChar (tChar '\n' <|> (isDonep >> return ' ') )- parseString= do+ tr "parseString" dropSpaces - tTakeWhile (not . isSpace)+ r <- tTakeWhile (not . isSpace)+ return r --- | take characters while they meet the condition++-- | take characters while they meet the condition. if no char matches, it returns empty tTakeWhile :: (Char -> Bool) -> TransIO BS.ByteString tTakeWhile cond= -- parse (BS.span cond)- withData $ \s -> let (h,t)= BS.span cond s in if BS.null h then empty else return (h,t) !> ("tTakeWhile",h)+ withGetParseString $ \s -> do + let ret@(h,_)= BS.span cond s+ --return () !> ("takewhile'",h,t)+ if BS.null h then empty else return ret+ -- | take characters while they meet the condition and drop the next character tTakeWhile' :: (Char -> Bool) -> TransIO BS.ByteString-tTakeWhile' cond= withData $ \s -> do+tTakeWhile' cond= withGetParseString $ \s -> do let (h,t)= BS.span cond s return () !> ("takewhile'",h,t)- if BS.null h then empty else return (h, if BS.null t then t else BS.tail t) + if BS.null h then empty else return (h, if BS.null t then t else BS.tail t) just1 f x= let (h,t)= f x in (Just h,t) -- | take n characters -tTake n= withData $ \s -> return $ BS.splitAt n s -- !> ("tTake",n,BS.take n s)+tTake n= withGetParseString $ \s -> return $ BS.splitAt n s !> ("tTake",n) -- | drop n characters-tDrop n= withData $ \s -> return $ ((),BS.drop n s)+tDrop n= withGetParseString $ \s -> return $ ((),BS.drop n s) --- | read a char-anyChar= withData $ \s -> if BS.null s then empty else return (BS.head s,BS.tail s) -- !> ("anyChar",s)+-- | read a char. If there is no input left it fails with empty+anyChar= withGetParseString $ \s -> if BS.null s then empty else return (BS.head s ,BS.tail s ) -- !> ("anyChar",s) -- | verify that the next character is the one expected-tChar c= withData $ \s -> if BS.null s || BS.head s /= c then empty else return (BS.head s,BS.tail s) !> ("tChar", BS.head s) +tChar c= withGetParseString $ \s -> if BS.null s || BS.head s /= c then empty else return (BS.head s,BS.tail s) -- !> ("tChar", BS.head s) -- anyChar >>= \x -> if x == c then return c else empty !> ("tChar",x) +{-+withGetParseString2 :: (BS.ByteString -> TransIO (a,BS.ByteString)) -> TransIO a+withGetParseString2 parser= do + ParseContext readMore s done <- gets parseContext + let str = s <> iter + iter = + let mr = lazy !> "READMORE"+ in case mr of+ SMore r -> r <> iter !> "SMORE"+ SLast r -> writeIORef done True `seq` r+ SDone -> writeIORef done True `seq` mempty --- | bring the lazy byteString state to a parser--- and actualize the byteString state with the result--- The tuple that the parser should return should be : (what it returns, what should remain to be parsed)-withData :: (BS.ByteString -> TransIO (a,BS.ByteString)) -> TransIO a-withData parser= Transient $ do- ParseContext readMore s <- getData `onNothing` error "parser: no context"- - let loop = unsafeInterleaveIO $ do- mr <- readMore + lazy = unsafePerformIO $ do+ r <- readIORef done+ if r then return SDone else do+ (x,_) <- runTransient readMore + tr x+ return $ fromJust x - return () !> ("readMore",mr)- case mr of - SMore r -> return r <> loop - SLast r -> return r- SDone -> return mempty -- !> "withData SDONE" - str <- liftIO $ return s <> loop- --if str == mempty then return Nothing else do- mr <- runTrans $ parser str- case mr of- Nothing -> return Nothing -- !> "NOTHING"- Just (v,str') -> do- setData $ ParseContext readMore str'- return $ Just v+ (v,str') <- parser str+ modify $ \s -> s{parseContext= ParseContext readMore str' done}+ return v+ where + +-- >>> :set -XOverloadedStrings+-- >>> :m + Transient.Internals Transient.Parse Control.Monad.IO.Class Data.ByteString.Lazy +-- >>> keep' $ do setParseStream (return $ SMore "hello") ; r <- withGetParseString2 $ \s-> return(Data.ByteString.Lazy.take 13 s,Data.ByteString.Lazy.drop 13 s); liftIO $ print r+-- >>> keep' $ do setParseString "time-1.9.3/lib/Data/Time/Clock/Internal/SystemTime.hs:1:1: error:" ; r <- (,,) <$> tTakeWhile' (/=':') <*> int <* tChar ':' <*> int; liftIO $ print r+-- "hellohellohel"+-- Nothing+-- ("time-1.9.3/lib/Data/Time/Clock/Internal/SystemTime.hs",1,1)+-- Nothing+-- +-} ++--+ --++{-+withGetParseString3 :: (BS.ByteString -> TransIO (a,BS.ByteString)) -> TransIO a+withGetParseString3 parser= do++ ParseContext readMore s done <- gets parseContext + + modify $ \st -> st{execMode= Serial}+ str <- return s <> iter readMore+ (v,str') <- parser str+ modify $ \s -> s{parseContext= ParseContext readMore str' done}+ return v+ where+ iter readMore= do+ -- modify $ \s -> s{execMode= Remote}+ mr <- readMore !> "READMORE"+ case mr of+ SMore r -> do liftIO $ print "SMORE"; return r <> iter readMore + SLast r -> return r+ SDone -> return mempty+ + + lazy mx= unsafePerformIO $ do+ (x,_) <- runTransient mx + return $ fromJust x+-}++++-- | bring the lazy byteString state to a parser which return the rest of the stream together with the result+-- and actualize the byteString state with it+-- The tuple that the parser returns should be : (what it returns, what should remain to be parsed)++++withGetParseString :: (BS.ByteString -> TransIO (a,BS.ByteString)) -> TransIO a+withGetParseString parser= Transient $ do+ + ParseContext readMore s done <- gets parseContext + + let loop = unsafeInterleaveIO $ do+ r <-readIORef done+ if r then return mempty else do+ (mr,_) <- runTransient readMore+ case mr of + Nothing -> mempty + Just(SMore r) -> return r <> do + d <- readIORef done+ if d then mempty else loop++ Just(SLast r) -> do tr "LAST"; writeIORef done True ; return r+ Just SDone -> do tr "DONE"; writeIORef done True ; return mempty -- !> "withGetParseString SDONE" ++ -- str <- liftIO $ (s <> ) `liftM` loop+ str <- liftIO $ return s <> loop+ --if BS.null str then return Nothing else do+ --return () !> ("withGetParseString", BS.take 3 str)+ mr <- runTrans $ parser str+ case mr of+ Nothing -> return Nothing -- !> "NOTHING"+ Just (v,str') -> do+ --return () !> (v,str') + modify $ \s-> s{parseContext= ParseContext readMore str' done}+ return $ Just v++++-- >>> keep' $ do x <- return "hello" <> lazy (liftIO $ print "world" >> return "world"); liftIO $ print $ take 3 x+++-- >>> :set -XOverloadedStrings+-- >>> :m + Transient.Internals Transient.Parse Control.Monad.IO.Class+-- >>> keep' $ do x <- withParseStream (return $ SMore "hello world") $ tTake 2 ; liftIO $ print x+-- *** Exception: ghc: signal: 15+--++++ -- | bring the data of the parse context as a lazy byteString-giveData= (noTrans $ do- ParseContext readMore s <- getData `onNothing` error "parser: no context"- :: StateIO (ParseContext BS.ByteString) -- change to strict BS+giveParseString :: TransIO BS.ByteString+giveParseString= (noTrans $ do+ ParseContext readMore s done<- gets parseContext -- getData `onNothing` error "parser: no context"+ -- :: StateIO (ParseContext BS.ByteString) -- change to strict BS let loop = unsafeInterleaveIO $ do- mr <- readMore+ (mr,_) <- runTransient readMore+ tr ("read",mr)+ case mr of - SMore r -> (r <>) `liftM` loop- SLast r -> (r <>) `liftM` loop- SDone -> return mempty+ Nothing -> mempty+ Just(SMore r) -> (r <>) `liftM` loop+ Just(SLast r) -> (r <>) `liftM` loop+ Just SDone -> return mempty liftIO $ (s <> ) `liftM` loop) +-- | drop from the stream until a condition is met+tDropUntil cond= withGetParseString $ \s -> f s+ where + f s= if BS.null s then return ((),s) else if cond s then return ((),s) else f $ BS.tail s++-- | take from the stream until a condition is met+tTakeUntil cond= withGetParseString $ \s -> f s+ where + f s= if BS.null s then return (s,s) else if cond s then return (s,s) else f $ BS.tail s++-- | add the String at the beginning of the stream to be parsed+tPutStr s'= withGetParseString $ \s -> return ((),s'<> s) -- | True if the stream has finished isDone :: TransIO Bool isDone= noTrans $ do - return () !> "isDone"- ParseContext readMore s <- getData `onNothing` error "parser: no context"- :: StateIO (ParseContext BS.ByteString) -- change to strict BS- if not $ BS.null s then return False else do- mr <- liftIO readMore - case mr of - SMore r -> do setData $ ParseContext readMore r ; return False- SLast r -> do setData $ ParseContext readMore r ; return False- SDone -> return True+ ParseContext _ _ done<- gets parseContext + liftIO $ readIORef done +dropUntilDone= (withGetParseString $ \s -> do+ tr "dropUntilDone"+ ParseContext _ _ done <- gets parseContext+ let loop s= do+ if (unsafePerformIO $ readIORef done)== True || BS.null s then return((), s) else loop $ BS.tail s+ -- end <- s `seq` liftIO $ readIORef done+ -- if end then return((), s) else loop $ BS.tail s+ loop s)+ <|> return()+ + - +-- | return the portion of the string not parsed+-- it is useful for testing purposes:+--+-- > result <- myParser <|> (do rest <- notParsed ; liftIO (print "not parsed this:"++ rest))+--+-- would print where myParser stopped working. +-- This does not work with (infinite) streams. Use `getParseBuffer` instead+notParsed:: TransIO BS.ByteString+notParsed= withGetParseString $ \s -> return (s,mempty) !> "notParsed"++-- | get the current buffer already read but not yet parsed+getParseBuffer :: TransIO BS.ByteString+getParseBuffer= do+ ParseContext _ s _<- gets parseContext+ return s++-- | empty the buffer+clearParseBuffer :: TransIO ()+clearParseBuffer= + modify$ \s -> s{parseContext= let ParseContext readMore _ d= parseContext s in ParseContext readMore mempty d}++-- | Used for debugging. It shows the next N characters in the parse buffer +showNext msg n= do + r <- tTake n+ liftIO $ print (msg,r);+ modify $ \s -> s{parseContext= (parseContext s){buffer= r <>buffer(parseContext s)}}+ @@ -273,18 +501,99 @@ -- The output is nondeterministic: it can return 0, 1 or more results -- -- example: https://t.co/fmx1uE2SUd+-- (|--) :: TransIO (StreamData BS.ByteString) -> TransIO b -> TransIO b+-- p |-- q = do+-- --addThreads 1+-- v <- liftIO $ newIORef undefined -- :: TransIO (MVar (StreamData BS.ByteString -> IO ()))+-- initq v <|> initp v+-- -- `catcht` \(_ :: BlockedIndefinitelyOnMVar) -> empty+-- -- TODO #2 use react instrad of MVar's? need buffering-contention+-- where+-- initq v= do+-- --abduce+-- r <-withParseStream (takev v ) q+-- liftIO $ print "AFGRT WITH"+-- return r++-- initp v= do+-- --abduce++-- return () !> "INITP"+-- repeatIt+-- where+-- repeatIt= do +-- r <- p+-- putv v r+-- return () !> "AFTER PUTV"+-- repeatIt+-- empty+-- -- return () !> ("putMVar")+-- -- t <-liftIO $ (putv v r >> return True) `catcht` \(_ :: BlockedIndefinitelyOnMVar) -> return False+-- -- if t then repeatIt else empty++-- takev v= do +-- return () !> "BEFORE TAKEV"+-- --modify $ \s -> s{execMode= Remote}+-- r <- react (writeIORef v) (return()) +-- return () !> ("TAKEV",r)+-- liftIO $ threadDelay 5000000+-- return r+ ++ +-- putv v s= liftIO $ do+-- proc <- readIORef v -- :: TransIO (StreamData BS.ByteString -> IO())+-- return () !> ("PUTV", s)+-- proc s+++++++++++ (|-) :: TransIO (StreamData BS.ByteString) -> TransIO b -> TransIO b p |- q = do- v <- liftIO $ newEmptyMVar- initp v <|> initq v+ --addThreads 1+ pcontext <- liftIO $ newIORef $ Just undefined+ v <- liftIO $ newEmptyMVar+ initp v pcontext <|> initq v pcontext+-- `catcht` \(_ :: BlockedIndefinitelyOnMVar) -> empty - where- initq v= do- --abduce- setParseStream (takeMVar v >>= \v -> (return v !> ("!- operator return",v))) -- each time the parser need more data, takes the var- q - - initp v= abduce >> repeatIt- where- repeatIt= (do r <- p; liftIO (putMVar v r !> "putMVar") ; empty) <|> repeatIt + where+ initq v pcontext= do+ --abduce+ setParseStream (do r <- liftIO $ takeMVar v; tr ("rec",fmap (BS.take 10) r); return r)-- `catch` \(_:: SomeException) -> return SDone ) + r <- q+ dropUntilDone+ Just p <- liftIO $ readIORef pcontext+ liftIO $ writeIORef pcontext Nothing !> "WRITENOTHING"+ pc <- gets parseContext+ modify $ \ s -> s{parseContext= p{done=done pc}}+ return r + initp v pcontext= do+ abduce+ ParseContext _ _ done <- gets parseContext++ let repeatIt= do+ pc <- liftIO $ readIORef pcontext+ if isNothing pc then tr "FINNNNNNNNNNNNNNNNNNNNNNNN" >> empty else do+ d <- liftIO $ readIORef done+ if d then do tr "sendDone";liftIO $ putMVar v SDone; repeatIt else do+ r <- p ++ liftIO $ putMVar v r -- `catch` \(_ :: BlockedIndefinitelyOnMVar) -> return False++ p <- gets parseContext+ liftIO $ writeIORef pcontext $ Just p+ case r of+ SDone -> empty+ SLast _ -> empty+ SMore _ -> repeatIt++ repeatIt
tests/TestSuite.hs view
@@ -1,4 +1,4 @@-#!/usr/bin/env ./execthirdline.sh+#!/usr/bin/env ./execthirdlinedocker.sh -- development -- set -e && docker run -it -v /c/Users/magocoal/OneDrive/Haskell/devel:/devel agocorona/transient:05-02-2017 bash -c "runghc -j2 -isrc -i/devel/transient/src /devel/transient/tests/$1 $2 $3 $4" @@ -6,10 +6,13 @@ -- set -e && executable=`basename -s .hs ${1}` && docker run -it -v $(pwd):/work agocorona/transient:05-02-2017 bash -c "ghc /work/${1} && /work/${executable} ${2} ${3}" +import qualified Prelude as Pr(return)+import Prelude hiding ((>>=),(>>),return)+ import Transient.Base import Transient.EVars import Transient.Indeterminism-import Transient.Backtrack+ import System.Exit import Data.Monoid import Control.Applicative@@ -26,8 +29,8 @@ main= do keep' $ do- let genElem :: a -> TransIO a- genElem x= do+ let -- genElem :: a -> TransIO a+ genElem x= do -- generates synchronous and asynchronous results with various delays isasync <- liftIO randomIO delay <- liftIO $ randomRIO (1, 1000) liftIO $ threadDelay delay@@ -35,11 +38,11 @@ liftIO $ putStrLn "--Testing thread control + Monoid + Applicative + async + indetermism---" - collect 100 $ do- i <- threads 0 $ choose [1..100]- nelems <- liftIO $ randomRIO (1, 10) :: TransIO Int- nthreads <- liftIO $ randomRIO (1,nelems)- r <- threads nthreads $ foldr (+) 0 $ map genElem [1..nelems]+ collect 100 $ do -- gather the result of 100 iterations+ i <- threads 0 $ choose [1..100] -- test 100 times. 'loop' for 100 times+ nelems <- liftIO $ randomRIO (1, 100) -- :: TransIO Int+ nthreads <- liftIO $ randomRIO (0,nelems) -- different numbers of threads+ r <- threads nthreads $ foldr (+) 0 $ map genElem [1..nelems] -- sum sync and async results using applicative assert (r == sum[1..nelems]) $ return () liftIO $ putStrLn "--------------checking parallel execution, Alternative, events --------"
transient.cabal view
@@ -1,5 +1,5 @@ name: transient-version: 0.6.3+version: 0.7.0.0 author: Alberto G. Corona extra-source-files: ChangeLog.md README.md@@ -22,48 +22,49 @@ manual: True library- -- Note: `stack sdist/upload` will add missing bounds (via "pvp-bounds: both") in `build-depends`- -- support GHC 7.10.3 and later; lower bounds below denote GHC 7.10.3's bundled versions- build-depends: base >= 4.8.0 && < 5- , containers >= 0.5.6- , transformers >= 0.4.2- , time >= 1.5- , directory >= 1.2.2- , bytestring >= 0.10.6+-- if impl(ghcjs >=0.1)+-- build-depends:+-- ghcjs-base -any ++ build-depends: base >= 4.8.0 && < 5+ , containers >= 0.5.6+ , transformers >= 0.4.2+ , time >= 1.5+ , directory >= 1.2.2+ , bytestring >= 0.10.6++ -- libraries not bundled w/ GHC , mtl , stm , random- , primitive < 0.6.4- if impl(eta)- build-depends:- - else- build-depends:- atomic-primops + exposed-modules: Transient.Backtrack Transient.Base Transient.EVars+ Transient.Mailboxes Transient.Indeterminism Transient.Internals Transient.Logged Transient.Parse ++ exposed: True buildable: True default-language: Haskell2010 hs-source-dirs: src .- ghc-options: - -- eta-options: -ddump-stg -ddump-to-file + ghc-options: + if flag(debug) cpp-options: -DDEBUG source-repository head type: git- location: https://github.com/agocorona/transient+ location: https://github.com/agocorona/transient-stack/transient test-suite test-transient @@ -79,8 +80,10 @@ -- libraries not bundled w/ GHC , mtl , stm- , random- , atomic-primops+ , random == 1.1+++ type: exitcode-stdio-1.0 main-is: TestSuite.hs build-depends: