diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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 diff --git a/src/Transient/Backtrack.hs b/src/Transient/Backtrack.hs --- a/src/Transient/Backtrack.hs +++ b/src/Transient/Backtrack.hs @@ -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. diff --git a/src/Transient/Base.hs b/src/Transient/Base.hs --- a/src/Transient/Base.hs +++ b/src/Transient/Base.hs @@ -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) diff --git a/src/Transient/EVars.hs b/src/Transient/EVars.hs --- a/src/Transient/EVars.hs +++ b/src/Transient/EVars.hs @@ -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 diff --git a/src/Transient/Indeterminism.hs b/src/Transient/Indeterminism.hs --- a/src/Transient/Indeterminism.hs +++ b/src/Transient/Indeterminism.hs @@ -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) - diff --git a/src/Transient/Internals.hs b/src/Transient/Internals.hs --- a/src/Transient/Internals.hs +++ b/src/Transient/Internals.hs @@ -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 -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 +-- +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 diff --git a/src/Transient/Logged.hs b/src/Transient/Logged.hs --- a/src/Transient/Logged.hs +++ b/src/Transient/Logged.hs @@ -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 + + diff --git a/src/Transient/Mailboxes.hs b/src/Transient/Mailboxes.hs new file mode 100644 --- /dev/null +++ b/src/Transient/Mailboxes.hs @@ -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,()) diff --git a/src/Transient/Parse.hs b/src/Transient/Parse.hs --- a/src/Transient/Parse.hs +++ b/src/Transient/Parse.hs @@ -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 diff --git a/tests/TestSuite.hs b/tests/TestSuite.hs --- a/tests/TestSuite.hs +++ b/tests/TestSuite.hs @@ -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 --------" diff --git a/transient.cabal b/transient.cabal --- a/transient.cabal +++ b/transient.cabal @@ -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: