transient 0.6.0.1 → 0.6.3
raw patch · 8 files changed
+1151/−753 lines, 8 filesdep +primitive
Dependencies added: primitive
Files
- README.md +126/−86
- src/Transient/EVars.hs +0/−1
- src/Transient/Indeterminism.hs +178/−152
- src/Transient/Internals.hs +426/−299
- src/Transient/Logged.hs +35/−25
- src/Transient/Parse.hs +290/−100
- tests/TestSuite.hs +6/−5
- transient.cabal +90/−85
README.md view
@@ -1,86 +1,126 @@- -========= - -[](http://hackage.haskell.org/package/transient) -[](http://stackage.org/lts/package/transient) -[](http://stackage.org/nightly/package/transient) -[](https://travis-ci.org/transient-haskell/transient) -[](https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?utm_source=share-link&utm_medium=link&utm_campaign=share-link) - -[](https://agocorona.github.io/donation.html) - -NOTE: distributed computing and web primitives Are in [transient-universe](https://github.com/transient-haskell/transient-universe) and [axiom](https://github.com/transient-haskell/axiom). Some examples at [transient-examples](https://github.com/transient-haskell/transient-examples) - - -## Some feedback on `transient`: - -1. Rahul Muttineni @rahulmutt nov. 09 2016 03:40 Lead developper of ETA (the JVM Haskell compiler) - - *It's a bit mind bending in that it's like using a higher-level list monad, but it's very, very cool. For beginning Haskellers, what would be really useful is a visualisation of what happens when you do various distributed/parallel stuff.* **It's almost shocking how effortlessly you can run computations across threads/nodes.** - - *The cool part is the composability in the distributed setting. *You can make higher-order monadic functions that allow you to compose & reuse a long chain of distributed transactions via `wormhole` and `teleport`*. Another benefit is that the transaction becomes first class and* **you can see exactly what's going on in one place** *instead of distributing the logic across actors making the code equivalent to event callbacks, as you've stated.* - - https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?at=58228caa35e6cf054773303b - -## What is Transient? - -One of the dreams of software engineering is unrestricted composability. - -This may be put in these terms: - -let `ap1` and `ap2` two applications with arbitrary complexity, with all effects including multiple threads, asynchronous IO, indeterminism, events and perhaps, distributed computing. - -Then the combinations: - - - ap1 <|> ap2 -- Alternative expression - - ap1 >>= \x -> ap2 -- monadic sequence - - ap1 <> ap2 -- monoidal expression - - (,) <$> ap1 <*> ap2 -- Applicative expression - -are possible if the types match, and generate new applications that are composable as well. - -Transient does exactly that. - -The operators `<$>` `<*>` and `<>` express concurrency, the operator `<|>` express parallelism and `>>=` for sequencing of threads, distributed processes or web widgets. So even in the presence of these effects and others, everything is composable. - -For this purpose transient is an extensible effects monad with all major effects and primitives for parallelism, events, asynchronous IO, early termination, non-determinism logging and distributed computing. Since it is possible to extend it with more effects without adding monad transformers, the composability is assured. - -Documentation -============= - -The [Wiki](https://github.com/agocorona/transient/wiki) is more user oriented - -My video sessions in [livecoding.tv](https://www.livecoding.tv/agocorona/videos/) not intended as tutorials or presentations, but show some of the latest features running. - -The articles are more technical: - -- [Philosophy, async, parallelism, thread control, events, Session state](https://www.fpcomplete.com/user/agocorona/EDSL-for-hard-working-IT-programmers?show=tutorials) -- [Backtracking and undoing IO transactions](https://www.fpcomplete.com/user/agocorona/the-hardworking-programmer-ii-practical-backtracking-to-undo-actions?show=tutorials) -- [Non-deterministic list like processing, multithreading](https://www.fpcomplete.com/user/agocorona/beautiful-parallel-non-determinism-transient-effects-iii?show=tutorials) -- [Distributed computing](https://www.fpcomplete.com/user/agocorona/moving-haskell-processes-between-nodes-transient-effects-iv?show=tutorials) -- [Publish-Subscribe variables](https://www.schoolofhaskell.com/user/agocorona/publish-subscribe-variables-transient-effects-v) -- [Distributed streaming, map-reduce](https://www.schoolofhaskell.com/user/agocorona/estimation-of-using-distributed-computing-streaming-transient-effects-vi-1) - -These articles contain executable examples (not now, since the site no longer support the execution of haskell snippets). - - -Plans -===== -Once composability in the large is possible, there are a infinite quantity of ideas that may be realized. There are short term and long term goals. An status of development is regularly published in [](https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?utm_source=share-link&utm_medium=link&utm_campaign=share-link). - -Among the most crazy ones is the possibility of extending this framework to other languages and make them interoperable. treating whole packaged applications as components, and docking them as lego pieces in a new layer of the Operating system where the shell allows such kind of type safe docking. this composable docker allows all kinds of composability, while the current docker platform is just a form of degraded monoid that do not compute. - -Contribute: -========== -Wanna contribute? Make sure that you've read our [contributor guidelines](https://github.com/transient-haskell/transient/blob/master/CONTRIBUTING.md). We'd like to hear from you and your ideas, get in touch with other contributors through: - -- [](https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?utm_source=share-link&utm_medium=link&utm_campaign=share-link) - -- [The issues page for transient](https://github.com/transient-haskell/transient/issues) -- [The issues page for transient-universe](https://github.com/transient-haskell/transient-universe/issues) -- [The issues page for axiom](https://github.com/transient-haskell/axiom/issues) - -Once you learn something interesting, you can [contribute to the wiki](https://github.com/transient-haskell/transient/wiki) - -[You can also donate](https://agocorona.github.io/donation.html) to the lead developer in order to make possible the dedication of more time to fullfil the potential advantages of true software composability across the whole stack. - ++=========++[](http://hackage.haskell.org/package/transient)+[](http://stackage.org/lts/package/transient)+[](http://stackage.org/nightly/package/transient)+[](https://travis-ci.org/transient-haskell/transient)+[](https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?utm_source=share-link&utm_medium=link&utm_campaign=share-link)++[](https://agocorona.github.io/donation.html)++NOTE: distributed computing and web primitives Are in [transient-universe](https://github.com/transient-haskell/transient-universe) and [axiom](https://github.com/transient-haskell/axiom). Some examples at [transient-examples](https://github.com/transient-haskell/transient-examples) +++## Some feedback on `transient`:++1. Rahul Muttineni @rahulmutt nov. 09 2016 03:40 Lead developper of ETA (the JVM Haskell compiler)++ *It's a bit mind bending in that it's like using a higher-level list monad, but it's very, very cool. For beginning Haskellers, what would be really useful is a visualisation of what happens when you do various distributed/parallel stuff.* **It's almost shocking how effortlessly you can run computations across threads/nodes.**++ *The cool part is the composability in the distributed setting. *You can make higher-order monadic functions that allow you to compose & reuse a long chain of distributed transactions via `wormhole` and `teleport`*. Another benefit is that the transaction becomes first class and* **you can see exactly what's going on in one place** *instead of distributing the logic across actors making the code equivalent to event callbacks, as you've stated.*++ https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?at=58228caa35e6cf054773303b++## What is Transient?++One of the dreams of software engineering is unrestricted composability.++This may be put in these terms:++let `ap1` and `ap2` two applications with arbitrary complexity, with all effects including multiple threads, asynchronous IO, indeterminism, events and perhaps, distributed computing.++Then the combinations:++ - ap1 <|> ap2 -- Alternative expression+ - ap1 >>= \x -> ap2 -- monadic sequence+ - ap1 <> ap2 -- monoidal expression+ - (,) <$> ap1 <*> ap2 -- Applicative expression++are possible if the types match, and generate new applications that are composable as well.++Transient does exactly that.++The operators `<$>` `<*>` and `<>` express concurrency, the operator `<|>` express parallelism and `>>=` for sequencing of threads, distributed processes or web widgets. So even in the presence of these effects and others, everything is composable.++For this purpose transient is an extensible effects monad with all major effects and primitives for parallelism, events, asynchronous IO, early termination, non-determinism logging and distributed computing. Since it is possible to extend it with more effects without adding monad transformers, the composability is assured.++Motivating example+==================+This program, will stream "hello world" from N nodes if you enter "fire" in the console++```Haskell+main= keep $ initNode $ inputNodes <|> distribStream++distribStream= do+ local $ option "fire" "fire"+ r <- clustered . local . choose $ repeat "hello world"+ localIO $ print r+```+Read the tutorial to know how to compile and invoke it.++This program will present a link in the browser and stream fibonnacci numbers to the browser when+yo click it. (if you have Docker, you can run it straigh from the console; See [this](https://github.com/transient-haskell/axiom#how-to-install--run-fast)++```Haskell+main= keep . initNode $ webFib++webFib= onBrowser $ do+ local . render $ wlink () (h1 "hello fibonacci numbers")++ r <- atRemote $ do+ r <- local . threads 1 . choose $ take 10 fibs+ localIO $ print r+ localIO $ threadDelay 1000000+ return r++ local . render . rawHtml $ (h2 r)+ where+ fibs = 0 : 1 : zipWith (+) fibs (tail fibs) :: [Int]+```++This program combines both functionalities:++```haskell+main= keep . initNode $ inputNodes <|> webFib <|> distribStream+```++Documentation+=============++The [Wiki](https://github.com/agocorona/transient/wiki) is more user oriented++My video sessions in [livecoding.tv](https://www.livecoding.tv/agocorona/videos/) not intended as tutorials or presentations, but show some of the latest features running.++The articles are more technical:++- [Philosophy, async, parallelism, thread control, events, Session state](https://www.fpcomplete.com/user/agocorona/EDSL-for-hard-working-IT-programmers?show=tutorials)+- [Backtracking and undoing IO transactions](https://www.fpcomplete.com/user/agocorona/the-hardworking-programmer-ii-practical-backtracking-to-undo-actions?show=tutorials)+- [Non-deterministic list like processing, multithreading](https://www.fpcomplete.com/user/agocorona/beautiful-parallel-non-determinism-transient-effects-iii?show=tutorials)+- [Distributed computing](https://www.fpcomplete.com/user/agocorona/moving-haskell-processes-between-nodes-transient-effects-iv?show=tutorials)+- [Publish-Subscribe variables](https://www.schoolofhaskell.com/user/agocorona/publish-subscribe-variables-transient-effects-v)+- [Distributed streaming, map-reduce](https://www.schoolofhaskell.com/user/agocorona/estimation-of-using-distributed-computing-streaming-transient-effects-vi-1)++These articles contain executable examples (not now, since the site no longer support the execution of haskell snippets).+++Plans+=====+Once composability in the large is possible, there are a infinite quantity of ideas that may be realized. There are short term and long term goals. An status of development is regularly published in [](https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?utm_source=share-link&utm_medium=link&utm_campaign=share-link). ++Among the most crazy ones is the possibility of extending this framework to other languages and make them interoperable. treating whole packaged applications as components, and docking them as lego pieces in a new layer of the Operating system where the shell allows such kind of type safe docking. this composable docker allows all kinds of composability, while the current docker platform is just a form of degraded monoid that do not compute.++Contribute:+==========+Wanna contribute? Make sure that you've read our [contributor guidelines](https://github.com/transient-haskell/transient/blob/master/CONTRIBUTING.md). We'd like to hear from you and your ideas, get in touch with other contributors through:++- [](https://gitter.im/Transient-Transient-Universe-HPlay/Lobby?utm_source=share-link&utm_medium=link&utm_campaign=share-link)++- [The issues page for transient](https://github.com/transient-haskell/transient/issues) +- [The issues page for transient-universe](https://github.com/transient-haskell/transient-universe/issues) +- [The issues page for axiom](https://github.com/transient-haskell/axiom/issues) ++Once you learn something interesting, you can [contribute to the wiki](https://github.com/transient-haskell/transient/wiki)++[You can also donate](https://agocorona.github.io/donation.html) to the lead developer in order to make possible the dedication of more time to fullfil the potential advantages of true software composability across the whole stack.+
src/Transient/EVars.hs view
@@ -54,7 +54,6 @@ readEVar (EVar ref1)= do tchan <- liftIO . atomically $ dupTChan ref1 r <- parallel $ atomically $ readTChan tchan- case r of SDone -> empty SMore x -> return x
src/Transient/Indeterminism.hs view
@@ -1,152 +1,178 @@------------------------------------------------------------------------------------ Module : Transient.Indeterminism--- Copyright :--- License : MIT------ Maintainer : agocorona@gmail.com--- Stability :--- Portability :------ | see <https://www.fpcomplete.com/user/agocorona/beautiful-parallel-non-determinism-transient-effects-iii>----------------------------------------------------------------------------------{-# LANGUAGE ScopedTypeVariables, CPP #-}-module Transient.Indeterminism (-choose, choose', collect, collect', group, groupByTime-) where--import Transient.Internals hiding (retry)--import Data.IORef-import Control.Applicative-import Data.Monoid-import Control.Concurrent -import Data.Typeable-import Control.Monad.State-import GHC.Conc-import Data.Time.Clock-import Control.Exception --#ifndef ETA_VERSION-import Data.Atomics-#endif ----- | Converts a list of pure values into a transient task set. You can use the--- 'threads' primitive to control the parallelism.----choose :: Show a => [a] -> TransIO a-choose []= empty-choose xs = do- evs <- liftIO $ newIORef xs- r <- parallel $ do- es <- atomicModifyIORefCAS evs $ \es -> let tes= tail es in (tes,es)- case es of- [x] -> x `seq` return $ SLast x- x:_ -> x `seq` return $ SMore x- checkFinalize r---- | Same as 'choose' except that the 'threads' combinator cannot be used,--- instead the parent thread's limit applies.----choose' :: [a] -> TransIO a-choose' xs = foldl (<|>) empty $ map (async . return) xs----- | Collect the results of a task set in groups of @n@ elements.----group :: Int -> TransIO a -> TransIO [a]-group num proc = do- v <- liftIO $ newIORef (0,[])- x <- proc-- mn <- liftIO $ atomicModifyIORefCAS v $ \(n,xs) ->- let n'=n +1- in if n'== num-- then ((0,[]), Just $ x:xs)- else ((n', x:xs),Nothing)- case mn of- 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`.----groupByTime :: Integer -> TransIO a -> TransIO [a]--groupByTime 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'--- is called.----collect :: Int -> TransIO a -> TransIO [a]-collect n = collect' n 0---- | Like 'collect' but with a timeout. When the timeout is zero it behaves--- exactly like 'collect'. If the timeout (second parameter) is non-zero,--- collection stops after the timeout and the results collected till now are--- returned.----collect' :: Int -> Int -> TransIO a -> TransIO [a]-collect' n t search= do- addThreads 1- rv <- liftIO $ newEmptyMVar -- !> "NEWMVAR"-- results <- liftIO $ newIORef (0,[])-- let worker = do- r <- abduce >> search- liftIO $ putMVar rv $ Just r- stop-- timer= do- when (t > 0) . async $ threadDelay t >>putMVar rv Nothing -- readIORef results >>= return . snd- empty-- monitor= liftIO loop-- where- loop = do- mr <- takeMVar rv- (n',rs) <- readIORef results- case mr of- Nothing -> return rs- Just r -> do- 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) ->- readIORef results >>= return . snd--- r <- oneThread $ worker <|> timer <|> monitor-- return r----+ ----------------------------------------------------------------------------- +-- +-- Module : Transient.Indeterminism +-- Copyright : +-- License : MIT +-- +-- Maintainer : agocorona@gmail.com +-- Stability : +-- Portability : +-- +-- | see <https://www.fpcomplete.com/user/agocorona/beautiful-parallel-non-determinism-transient-effects-iii> +-- +----------------------------------------------------------------------------- +{-# LANGUAGE ScopedTypeVariables, CPP #-} +module Transient.Indeterminism ( +choose, choose', chooseStream, collect, collect', group, groupByTime +) where + +import Transient.Internals hiding (retry) + +import Data.IORef +import Control.Applicative +import Data.Monoid +import Control.Concurrent +import Data.Typeable +import Control.Monad.State +import GHC.Conc +import Data.Time.Clock +import Control.Exception + +#ifndef ETA_VERSION +import Data.Atomics +#endif + + +-- | Converts a list of pure values into a transient task set. You can use the +-- 'threads' primitive to control the parallelism. +-- +choose :: [a] -> TransIO a +choose []= empty +choose xs = chooseStream xs >>= checkFinalize + +-- | transmit the end of stream +chooseStream :: [a] -> TransIO (StreamData a) +chooseStream []= empty +chooseStream xs = do + evs <- liftIO $ newIORef xs + parallel $ do + es <- atomicModifyIORefCAS evs $ \es -> let tes= tail es in (tes,es) + case es of + [x] -> x `seq` return $ SLast x + x:_ -> x `seq` return $ SMore x + + +-- | Same as 'choose', slower in some cases +-- +choose' :: [a] -> TransIO a +choose' xs = foldl (<|>) empty $ map (async . return) xs + + +-- | Collect the results of a task set in groups of @n@ elements. +-- +group :: Int -> TransIO a -> TransIO [a] +group num proc = do + v <- liftIO $ newIORef (0,[]) + x <- proc + + mn <- liftIO $ atomicModifyIORefCAS v $ \(n,xs) -> + let n'=n +1 + in if n'== num + + then ((0,[]), Just $ x:xs) + else ((n', x:xs),Nothing) + case mn of + 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' +-- is called. +-- +collect :: Int -> TransIO a -> TransIO [a] +collect n = collect' n 0 + +-- | Like 'collect' but with a timeout. When the timeout is zero it behaves +-- exactly like 'collect'. If the timeout (second parameter) is non-zero, +-- collection stops after the timeout and the results collected till now are +-- returned. +-- +collect' :: Int -> Int -> TransIO a -> TransIO [a] +collect' n t search= do + addThreads 1 + + rv <- liftIO $ newEmptyMVar -- !> "NEWMVAR" + + results <- liftIO $ newIORef (0,[]) + + let worker = do + r <- abduce >> search + liftIO $ putMVar rv $ Just r + stop + + timer= do + when (t > 0) . async $ threadDelay t >> putMVar rv Nothing + empty + + monitor= liftIO loop + + where + loop = do + mr <- takeMVar rv + + (n',rs) <- readIORef results + case mr of + Nothing -> return rs + Just r -> do + 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) -> + readIORef results >>= return . snd + + + oneThread $ timer <|> worker <|> monitor + + +-- | insert `SDone` response everytime 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) + +groupByTime :: Monoid a => Int -> TransIO a -> TransIO a +groupByTime timeout comp= do + v <- liftIO $ newIORef mempty + gather v <|> run v + where + run v = do + x <- comp + liftIO $ atomicModifyIORefCAS v $ \xs -> (xs <> x,()) + empty + + gather v= waitEvents $ do + threadDelay timeout + atomicModifyIORefCAS v $ \xs -> (mempty , xs) + + + +
src/Transient/Internals.hs view
@@ -1,4 +1,4 @@------------------------------------------------------------------------------+------------------------------------------------------------------------------ -- -- Module : Base -- Copyright :@@ -11,17 +11,14 @@ -- | See http://github.com/agocorona/transient -- Everything in this module is exported in order to allow extensibility. -----------------------------------------------------------------------------+{-# LANGUAGE CPP #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE UndecidableInstances #-}-{-# LANGUAGE Rank2Types #-} {-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE CPP #-}-{-# LANGUAGE InstanceSigs #-} {-# LANGUAGE ConstraintKinds #-} @@ -46,7 +43,10 @@ import Data.IORef import System.Environment import System.IO+import System.IO.Error ++import Data.String import qualified Data.ByteString.Char8 as BS import Data.Typeable @@ -63,7 +63,7 @@ tshow= Debug.Trace.traceShow {-# INLINE (!>) #-} (!>) :: Show a => b -> a -> b-(!>) x y = trace (show y) x +(!>) x y = trace (show y) x infixr 0 !> #else@@ -126,8 +126,8 @@ -- ^ Label the thread with its lifecycle state and a label string } deriving Typeable - + instance MonadState EventF TransIO where get = Transient $ get >>= return . Just put x = Transient $ put x >> return (Just ())@@ -169,6 +169,12 @@ runTransState :: EventF -> TransIO x -> IO (Maybe x, EventF) runTransState st x = runStateT (runTrans x) st +++emptyIfNothing :: Maybe a -> TransIO a+emptyIfNothing = Transient . return++ -- | Get the continuation context: closure, continuation, state, child threads etc getCont :: TransIO EventF getCont = Transient $ Just <$> get@@ -251,102 +257,56 @@ fmap f mx = do x <- mx return $ f x+ instance Applicative TransIO where pure a = Transient . return $ Just a - f <*> g = Transient $ do- rf <- liftIO $ newIORef (Nothing,[])- rg <- liftIO $ newIORef (Nothing,[])--- fs <- getContinuations-- let hasWait (_:Wait:_) = True- hasWait _ = False-- appf k = Transient $ do- Log rec _ full _ <- getData `onNothing` return (Log False [] [] 0)- (liftIO $ writeIORef rf (Just k,full))- -- !> ( show $ unsafePerformIO myThreadId) ++"APPF"-- (x, full2)<- liftIO $ readIORef rg- when (hasWait full ) $- -- (!> (hasWait full,"full",full, "\nfull2",full2)) $- let full'= head full: full2- in (setData $ Log rec full' full')- -- !> ("result1",full')-- return $ Just k <*> x-- appg x = Transient $ do- Log rec _ full _ <- getData `onNothing` return (Log False [] [] 0)- liftIO $ writeIORef rg (Just x, full)-- -- !> ( show $ unsafePerformIO myThreadId) ++ "APPG"-- (k,full1) <- liftIO $ readIORef rf- when (hasWait full) $-- -- (!> ("full", full, "\nfull1",full1)) $-- let full'= head full: full1- in (setData $ Log rec full' full')- -- !> ("result2",full')-- return $ k <*> Just x-- setContinuation f appf fs-- k <- runTrans f- -- !> ( show $ unsafePerformIO myThreadId)++ "RUN f"+ mf <*> mx = do+ + r1 <- liftIO $ newIORef Nothing+ r2 <- liftIO $ newIORef Nothing+ fparallel r1 r2 <|> xparallel r1 r2+ where+ fparallel r1 r2= do+ f <- mf - was <- getData `onNothing` return NoRemote- when (was == WasParallel) $ setData NoRemote+ r <- getState <|> return NoRemote+ -- return () !> ("first",r)+ if r == NoRemote then do+ x <- mx !> "second serial"+ return $ f x+ else do - Log recovery _ full _ <- getData `onNothing` return (Log False [] [] 0)+ liftIO $ (writeIORef r1 $ Just f)+ mx <- liftIO (readIORef r2)+ case mx of+ Nothing -> empty+ Just x -> return $ f x+ + xparallel r1 r2 = 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 - if was== WasRemote || (not recovery && was == NoRemote && isNothing k )- -- !> ("was,recovery,isNothing=",was,recovery, isNothing k)-- -- if the first operand was a remote request- -- (so this node is not master and hasn't to execute the whole expression)- -- or it was not an asyncronous term (a normal term without async or parallel- -- like primitives) and is nothing- then do- restoreStack fs- return Nothing- else do- when (isJust k) $ liftIO $ writeIORef rf (k,full)- -- when necessary since it maybe WasParallel and Nothing-- setContinuation g appg fs-- x <- runTrans g- -- !> ( show $ unsafePerformIO myThreadId) ++ "RUN g"-- Log recovery _ full' _ <- getData `onNothing` return (Log False [] [] 0)- liftIO $ writeIORef rg (x,full')- restoreStack fs- k'' <- if was== WasParallel- then do- (k',_) <- liftIO $ readIORef rf -- since k may have been updated by a parallel f- return k'- else return k- return $ k'' <*> x---- instance Monad (Cont r) where--- return a = Cont ($ a)--- m >>= k = Cont $ \c -> runCont m $ \a -> runCont (k a) c---- instance MonadCont (Cont r) where--- callCC f = Cont $ \c -> runCont (f (\a -> Cont $ \_ -> c a)) c+-- | stop the current computation and does not execute any alternative computation+fullStop :: TransIO stop+fullStop= setData WasRemote >> stop instance Monad TransIO where return = pure x >>= f = Transient $ do+ -- delData WasParallel setEventCont x f mk <- runTrans x resetEventCont mk@@ -355,20 +315,23 @@ Nothing -> return Nothing instance MonadIO TransIO where- -- liftIO mx = do- -- ex <- liftIO' $ (mx >>= return . Right) `catch`- -- (\(e :: SomeException) -> return $ Left e)- -- case ex of- -- Left e -> back e - -- Right x -> return x- -- where - liftIO x = Transient $ liftIO x >>= return . Just- + liftIO x = Transient $ liftIO x >>= return . Just instance Monoid a => Monoid (TransIO a) where- mappend x y = mappend <$> x <*> y mempty = return mempty +#if MIN_VERSION_base(4,11,0) + mappend = (<>) + +instance (Monoid a) => Semigroup (TransIO a) where+ (<>)= mappendt+#else+ mappend= mappendt+#endif++mappendt x y = mappend <$> x <*> y++ instance Alternative TransIO where empty = Transient $ return Nothing (<|>) = mplus@@ -391,14 +354,21 @@ readWithErr n line = (v `seq` return [(v, left)]) `catch` (\(_ :: SomeException) ->- error $ "read error trying to read type: \"" ++ show (typeOf v)+ throw $ ParseError $ "read error trying to read type: \"" ++ show (typeOf v) ++ "\" in: " ++ " <" ++ show line ++ "> ") where (v, left):_ = readsPrec n line +newtype ParseError= ParseError String deriving (Show)++instance Exception ParseError+ read' s= case readsPrec' 0 s of [(x,"")] -> x- _ -> error $ "reading " ++ s++ _ -> throw $ ParseError $ "reading " ++ s ++ readsPrec' n = unsafePerformIO . readWithErr n -- | Constraint type synonym for a value that can be logged.@@ -406,10 +376,10 @@ -- data Serializable a where -- serialize :: a .ç-> BS.ByteString--- deserialize :: BS.ByteString -> a +-- deserialize :: BS.ByteString -> a -- instance Serialize a => Serialie [a] where--- serialize (x:xs)= +-- serialize (x:xs)= -- let s= serialize x -- l= length s -- in makeByteString (#(#l,s#),serialize xs #)@@ -480,13 +450,13 @@ instance AdditionalOperators TransIO where - (**>) :: TransIO a -> TransIO b -> TransIO b+ --(**>) :: TransIO a -> TransIO b -> TransIO b (**>) x y = Transient $ do runTrans x runTrans y - (<***) :: TransIO a -> TransIO b -> TransIO a+ --(<***) :: TransIO a -> TransIO b -> TransIO a (<***) ma mb = Transient $ do fs <- getContinuations@@ -496,7 +466,7 @@ restoreStack fs return a - (<**) :: TransIO a -> TransIO b -> TransIO a+ --(<**) :: TransIO a -> TransIO b -> TransIO a (<**) ma mb = Transient $ do a <- runTrans ma@@ -527,7 +497,7 @@ -- | Set the current closure and continuation for the current statement {-# INLINABLE setEventCont #-} setEventCont :: TransIO a -> (a -> TransIO b) -> StateIO ()-setEventCont x f = modify $ \EventF { fcomp = fs, .. } +setEventCont x f = modify $ \EventF { fcomp = fs, .. } -> EventF { xcomp = x , fcomp = unsafeCoerce f : fs , .. }@@ -538,14 +508,14 @@ -- resetEventCont :: Maybe a -> EventF -> StateIO () {-# INLINABLE resetEventCont #-} resetEventCont mx =- modify $ \EventF { fcomp = fs, .. } + modify $ \EventF { fcomp = fs, .. } -> EventF { xcomp = case mx of Nothing -> empty Just x -> unsafeCoerce (head fs) x , fcomp = tailsafe fs , .. }- + -- | Total variant of `tail` that returns an empty list when given an empty list. {-# INLINE tailsafe #-} tailsafe :: [a] -> [a]@@ -582,33 +552,41 @@ oneThread :: TransIO a -> TransIO a oneThread comp = do st <- get- chs <- liftIO $ newMVar []+ rchs <- liftIO $ newMVar [] label <- liftIO $ newIORef (Alive, BS.pack "oneThread") let st' = st { parent = Just st- , children = chs- , labelth = label }- liftIO $ hangThread st st'+ , children = rchs+ , labelth = label }+ liftIO $ do+ atomicModifyIORefCAS (labelth st) $ \(_, label) -> ((Parent,label),())+ hangThread st st'+ put st'- x <- comp+ + x <- comp th <- liftIO myThreadId- -- !> ("FATHER:", threadId st)- chs <- liftIO $ readMVar chs -- children st'+ !> ("FATHER:", threadId st)+ chs <- liftIO $ readMVar rchs+ liftIO $ mapM_ (killChildren1 th) chs+ !> ("KILLEVENT1 ", map threadId chs ) return x- where killChildren1 :: ThreadId -> EventF -> IO ()- killChildren1 th state = do+ where + + killChildren1 :: ThreadId -> EventF -> IO ()+ killChildren1 th state = 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'- -- !> ("KILLEVENT1 ", map threadId ths' ) + -- | Add a label to the current passing threads so it can be printed by debugging calls like `showThreads`-labelState :: (MonadIO m,MonadState EventF m) => String -> m ()+labelState :: (MonadIO m,MonadState EventF m) => BS.ByteString -> m () labelState l = do st <- get- liftIO $ atomicModifyIORefCAS (labelth st) $ \(status,_) -> ((status, BS.pack l), ())+ liftIO $ atomicModifyIORefCAS (labelth st) $ \(status,_) -> ((status, l), ()) printBlock :: MVar () printBlock = unsafePerformIO $ newMVar ()@@ -633,7 +611,8 @@ showTree 0 st -- | Return the state of the thread that initiated the transient computation-topState :: TransIO EventF+-- topState :: TransIO EventF+topState :: MonadState EventF m => m EventF topState = do st <- get return $ toplevel st@@ -641,33 +620,59 @@ Nothing -> st Just p -> toplevel p --- | Return the state variable of the type desired with which a thread, identified by his number in the treee was initiated-showState :: (Typeable a, MonadIO m, Alternative m) => String -> EventF -> m (Maybe a)-showState th top = resp- where +{-++getStateFromThread th top = resp+ where resp = do let thstring = drop 9 . show $ threadId top if thstring == th then getstate top else do sts <- liftIO $ readMVar $ children top- foldl (<|>) empty $ map (showState th) sts+ foldl (<|>) empty $ map (getStateFromThread th) sts getstate st = case M.lookup (typeOf $ typeResp resp) $ mfData st of Just x -> return . Just $ unsafeCoerce x Nothing -> return Nothing typeResp :: m (Maybe x) -> x typeResp = undefined+-} --- | return all the states of the type desired that are created by direct child threads+-- | find the first computation state which match a filter in the subthree of states+findState :: (MonadIO m, Alternative m) => (EventF -> m Bool) -> EventF -> m EventF +findState filter top= do+ r <- filter top + if r then return top + else do+ sts <- liftIO $ readMVar $ children top+ foldl (<|>) empty $ map (findState filter) sts+++-- | Return the state variable of the type desired for a thread number+getStateFromThread :: (Typeable a, MonadIO m, Alternative m) => String -> EventF -> m (Maybe a)+getStateFromThread th top= getstate =<< findState (matchth th) top+ where+ matchth th th'= do+ let thstring = drop 9 . show $ threadId th'+ return $ if thstring == th then True else False+ getstate st = resp+ where resp= case M.lookup (typeOf $ typeResp resp) $ mfData st of+ Just x -> return . Just $ unsafeCoerce x+ Nothing -> return Nothing+ + typeResp :: m (Maybe x) -> x+ typeResp = undefined++-- | execute all the states of the type desired that are created by direct child threads processStates :: Typeable a => (a-> TransIO ()) -> EventF -> TransIO() processStates display st = do- + getstate st >>= display liftIO $ print $ threadId st sts <- liftIO $ readMVar $ children st mapM_ (processStates display) sts- where + where getstate st = case M.lookup (typeOf $ typeResp display) $ mfData st of Just x -> return $ unsafeCoerce x@@ -675,6 +680,8 @@ typeResp :: (a -> TransIO()) -> a typeResp = undefined ++ -- | Add n threads to the limit of threads. If there is no limit, the limit is set. addThreads' :: Int -> TransIO () addThreads' n= noTrans $ do@@ -721,7 +728,7 @@ st <- get put st {freeTh = False} r <- runTrans process- modify $ \st -> st { freeTh = freeTh st }+ modify $ \s -> s { freeTh = freeTh st } return r -- | Kill all the child threads of the current thread.@@ -749,7 +756,7 @@ when (isJust mparent) $ modifyMVar_ (children $ fromJust mparent) $ \sts -> return $ filter (\st -> threadId st /= thisth) sts- killThread $ thisth+ killThread $ thisth !> ("kill this thread:",thisth) -- * Extensible State: Session Data Management @@ -770,7 +777,7 @@ -- | 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.. +-- 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: -- -- > getSData <|> error "no data of the type desired"@@ -806,7 +813,7 @@ setData x = modify $ \st -> st { mfData = M.insert t (unsafeCoerce x) (mfData st) } where t = typeOf x --- | Accepts a function that takes the current value of the stored data type+-- | 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 ()@@ -819,19 +826,20 @@ Just x -> Just $ unsafeCoerce x Nothing -> Nothing --- | Either modify according with the first parameter or insert according with the second, depending on if the data exist or not.+-- | Either modify according with the first parameter or insert according with the second, depending on if the data exist or not. It returns the+-- old value or the new value accordingly. ----- > runTransient $ do modifyData1 (\h -> h ++ " world") "hello new" ; r <- getSData ; liftIO $ putStrLn r -- > "hello new"--- > runTransient $ do setData "hello" ; modifyData1 (\h -> h ++ " world") "hello new" ; r <- getSData ; liftIO $ putStrLn r -- > "hello world"-modifyData' :: (MonadState EventF m, Typeable a) => (a -> a) -> a ->m a-modifyData' f v= do +-- > 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' f v= do st <- get- let (ma,nmap)= M.insertLookupWithKey alterf t (unsafeCoerce v) (mfData st) + let (ma,nmap)= M.insertLookupWithKey alterf t (unsafeCoerce v) (mfData st) put st { mfData =nmap} return $ if isNothing ma then v else unsafeCoerce $ fromJust ma where t = typeOf v alterf _ _ x = unsafeCoerce $ f $ unsafeCoerce x- + -- | Same as modifyData modifyState :: (MonadState EventF m, Typeable a) => (Maybe a -> Maybe a) -> m () modifyState = modifyData@@ -853,25 +861,33 @@ newtype Ref a = Ref (IORef a) ++ -- | 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:: Typeable a => a -> TransIO ()+setRState:: (MonadIO m,MonadState EventF m, Typeable a) => a -> m () setRState x= do- Ref ref <- getSData+ Ref ref <- getData `onNothing` do+ ref <- Ref <$> liftIO (newIORef x)+ setData ref+ return ref liftIO $ atomicModifyIORefCAS ref $ const (x,())- <|> do- ref <- liftIO (newIORef x)- setData $ Ref ref +getRData :: (MonadIO m, MonadState EventF 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= do- Ref ref <- getSData- liftIO $ readIORef ref+getRState= Transient getRData delRState x= delState (undefined `asTypeOf` ref x)- where ref :: a -> IORef a - ref= undefined+ where ref :: a -> Ref a+ 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@@ -880,7 +896,7 @@ sd <- gets mfData mx <|> (modify (\s -> s { mfData = sd }) >> empty) --- | Executes the computation and reset the state either if it fails or not. +-- | Executes the computation and reset the state either if it fails or not. sandbox :: TransIO a -> TransIO a sandbox mx = do sd <- gets mfData@@ -905,7 +921,7 @@ getPrevId = gets mfSequence instance Read SomeException where- + readsPrec n str = [(SomeException $ ErrorCall s, r)] where [(s , r)] = readsPrec n str @@ -916,6 +932,11 @@ | SDone -- ^ No more tasks, we are done | SError SomeException -- ^ An error occurred deriving (Typeable, Show,Read)+ +instance Functor StreamData where+ fmap f (SMore a)= SMore (f a)+ 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@@ -927,8 +948,11 @@ SMore x -> return x SError e -> back e --- | Run an IO computation asynchronously and generate a single task carrying--- the result of the computation when it completes. See 'parallel' for notes on+-- | 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. async :: IO a -> TransIO a async io = do@@ -952,9 +976,8 @@ spawn :: IO a -> TransIO a spawn = freeThreads . waitEvents --- | A task stream generator that produces an infinite stream of tasks by--- running an IO computation periodically at the specified time interval. The--- task carries the result of the computation. A new task is generated only if+-- | 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. sample :: Eq a => IO a -> Int -> TransIO a@@ -973,15 +996,18 @@ -- | 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--- task is triggered with the result value. If the return value is 'SMore', the--- action is run again to generate the next task, otherwise task creation--- stops.+-- result is returned with the result value. If there are threads available, the res of the+-- computation is executed in a new thread. If the return value is 'SMore', the+-- action is run again to generate the next result, otherwise task creation+-- stop. -- -- Unless the maximum number of threads (set with 'threads') has been reached, -- the task is generated in a new thread and the current thread returns a void -- task. parallel :: IO (StreamData b) -> TransIO (StreamData b) parallel ioaction = Transient $ do+ was <- getData `onNothing` return NoRemote+ when (was /= WasRemote) $ setData WasParallel cont <- get -- !> "PARALLEL" case event cont of@@ -990,9 +1016,9 @@ return $ unsafeCoerce j Nothing -> do liftIO $ atomicModifyIORefCAS (labelth cont) $ \(_, lab) -> ((Parent, lab), ())+ liftIO $ loop cont ioaction- was <- getData `onNothing` return NoRemote- when (was /= WasRemote) $ setData WasParallel+ -- th <- liftIO myThreadId -- return () !> ("finish",th) return Nothing@@ -1045,13 +1071,13 @@ th <- myThreadId let cont'= cont{threadId=th} when(not $ freeTh parent )$ hangThread parent cont'- -- !> ("thread created: ",th,"in",threadId parent )+ -- !> ("thread created: ",th,"in",threadId parent ) proc cont') $ \me -> do case me of- Left e -> exceptBack cont e >> return () -- !> "exceptBack 2" + Left e -> exceptBack cont e >> return () -- !> "exceptBack 2" @@ -1064,19 +1090,21 @@ th <- myThreadId (can,label) <- atomicModifyIORefCAS (labelth cont) $ \(l@(status,label)) -> ((if status== Alive then Dead else status, label),l)+ when (can /= Parent ) $ free th parent - when (can/= Parent ) $ free th parent return () forkFinally1 :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId forkFinally1 action and_then = mask $ \restore -> forkIO $ Control.Exception.try (restore action) >>= and_then- + free th env= do -- return () !> ("freeing",th,"in",threadId env)- let sibling= children env + threadDelay 1000 -- wait for some activity in the children of the parent thread, to avoid+ -- an early elimination+ let sibling= children env (sbs',found) <- modifyMVar sibling $ \sbs -> do let (sbs', found) = drop [] th sbs@@ -1088,13 +1116,14 @@ then do -- !> ("new list for",threadId env,map threadId sbs')+ (typ,_) <- readIORef $ labelth env if (null sbs' && typ /= Listener && isJust (parent env)) -- free the parent then free (threadId env) ( fromJust $ parent env) else return () --- return env+ else return () -- putMVar sibling sbs -- !> (th,"orphan") @@ -1110,10 +1139,9 @@ let headpths= children parentProc modifyMVar_ headpths $ \ths -> return (child:ths)--- ths <- takeMVar headpths--- putMVar headpths (child:ths) + -- !> ("hang", threadId child, threadId parentProc,map threadId ths,unsafePerformIO $ readIORef $ labelth parentProc) -- | kill all the child threads associated with the continuation context@@ -1121,13 +1149,11 @@ ths <- modifyMVar childs $ \ths -> return ([],ths)--- ths <- takeMVar childs--- putMVar childs [] mapM_ (killChildren . children) ths - mapM_ (killThread . threadId) ths !> ("KILL", map threadId ths )+ mapM_ (killThread . threadId) ths !> ("Kill children", map threadId ths ) @@ -1152,14 +1178,15 @@ -> IO response -> TransIO eventdata react setHandler iob= Transient $ do+ was <- getData `onNothing` return NoRemote+ when (was /= WasRemote) $ setData WasParallel 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) cont{event= Just $ unsafeCoerce dat} `catch` exceptBack cont iob- was <- getData `onNothing` return NoRemote- when (was /= WasRemote) $ setData WasParallel+ return Nothing j@(Just _) -> do@@ -1180,7 +1207,7 @@ -- 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-option = optionf False +option = optionf False -- Implements the same functionality than `option` but only wait for one input option1 :: (Typeable b, Show b, Read b, Eq b) =>@@ -1190,103 +1217,136 @@ optionf :: (Typeable b, Show b, Read b, Eq b) => Bool -> b -> String -> TransIO b-optionf flag ret message = do +optionf flag ret message = do let sret= if typeOf ret == typeOf "" then unsafeCoerce ret else show ret- liftIO $ putStrLn $ "Enter "++sret++"\tto: " ++ message- inputf flag sret message Nothing ( == sret)+ liftIO $ putStrLn $ "Enter "++sret++"\t\tto: " ++ message+ inputf flag sret Nothing ( == sret) liftIO $ putStr "\noption: " >> putStrLn sret -- abduce return ret- -inputf flag ident message mv cond= do- r <- react (addListener ident) (return ()) - when (null r) $ liftIO $ writeIORef rconsumed True - let rr= read1 r - when flag $ liftIO $ delListener ident- case rr of- Just x -> if cond x - then do - liftIO $ do- writeIORef rconsumed True - print x; - return x - else do liftIO $ when (isJust mv) $ putStrLn ""; returnm mv- _ -> do liftIO $ when (isJust mv) $ putStrLn ""; returnm mv+-- | inputf <remove after sucessful or not> <listener identifier> <Maybe default value> <validation proc>+inputf :: Loggable a => Bool -> String -> Maybe a -> (a -> Bool) -> TransIO a+inputf flag ident mv cond= do+ str <- react (addListener ident) (return ()) + when flag $ liftIO $ delListener ident + c <- liftIO $ readIORef rconsumed+ if c then returnm mv else do+ if null str then do liftIO $ writeIORef rconsumed True; returnm mv else do + let rr = read1 str + + case rr of+ Just x -> if cond x + then liftIO $ do+ writeIORef rconsumed True + -- print x+ -- hFlush stdout+ return x+ else do liftIO $ when (isJust mv) $ putStrLn ""; returnm mv+ _ -> do liftIO $ when (isJust mv) $ putStrLn ""; returnm mv + where- returnm (Just x)= return x + returnm (Just x)= return x returnm _ = empty- - read1 s= x where- x= if typeOf(typeOfr x) == typeOf "" + + -- read1 :: String -> Maybe a+ read1 s= r + where+ r = if (typeOf $ fromJust r) == typeOf "" then Just $ unsafeCoerce s- else unsafePerformIO $ do- (let r= read s in r `seq` return (Just r)) `catch` \(e :: SomeException) -> (return Nothing)- typeOfr :: Maybe a -> a- typeOfr = undefined+ else case reads s of+ [] -> Nothing+ [(x,"")] -> Just x ++ -- | Waits on stdin and return a value when a console input matches the -- predicate specified in the first argument. The second parameter is a string -- to be displayed on the console before waiting. input :: (Typeable a, Read a,Show a) => (a -> Bool) -> String -> TransIO a input= input' Nothing +-- | `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" prompt mv cond+input' mv cond prompt= do + liftIO $ putStr prompt >> hFlush stdout + inputf True "input" mv cond -rcb= unsafePerformIO $ newIORef M.empty :: IORef (M.Map String (String -> IO())) +++rcb= unsafePerformIO $ newIORef [] :: IORef [ (String,String -> IO())]+ addListener :: String -> (String -> IO ()) -> IO ()-addListener name cb= atomicModifyIORef rcb $ \cbs -> (M.insert name cb cbs,())+addListener name cb= atomicModifyIORef rcb $ \cbs -> (filter ((/=) name . fst) cbs ++ [(name, cb)],()) delListener :: String -> IO ()-delListener name= atomicModifyIORef rcb $ \cbs -> (M.delete name cbs,())- +delListener name= atomicModifyIORef rcb $ \cbs -> (filter ((/=) name . fst) cbs,()) + reads1 s=x where x= if typeOf(typeOfr x) == typeOf "" then unsafeCoerce[(s,"")] else readsPrec' 0 s typeOfr :: [(a,String)] -> a typeOfr = undefined +read1 s= let [(x,"")]= reads1 s in x++++rprompt= unsafePerformIO $ newIORef "> " inputLoop= do- x <- getLine - processLine x- putStr "> "; hFlush stdout- inputLoop + line <- getLine+ processLine line + threadDelay 1000000+ prompt <- readIORef rprompt+ when (not $ null prompt) $ do putStr prompt ; hFlush stdout+ inputLoop++++{-# NOINLINE rconsumed #-} rconsumed = unsafePerformIO $ newIORef False+{-# NOINLINE lineprocessmode #-}+lineprocessmode= unsafePerformIO $ newIORef False processLine r = do- let rs = breakSlash [] r- return () !> rs- mapM' invoke rs + linepro <- readIORef lineprocessmode+ if linepro then do++ mapM' invokeParsers [r]+ + else do+ let rs = breakSlash [] r+ mapM' invokeParsers rs++ where- invoke x= do+ invokeParsers x= do mbs <- readIORef rcb- mapM (\cb -> cb x) $ M.elems mbs - + mapM_ (\cb -> cb x) $ map snd mbs+ mapM' f []= return ()- mapM' f (xss@(x:xs)) =do - f x + mapM' f (xss@(x:xs)) =do+ f x r <- readIORef rconsumed- if r ++ if r then do writeIORef riterloop 0 writeIORef rconsumed False- mapM' f xs + mapM' f xs else do threadDelay 1000 n <- atomicModifyIORef riterloop $ \n -> (n+1,n) if n==100 then do- when (not $ null x) $ putStr x >> putStrLn ": can't read, skip"+ when (not $ null x) $ hPutStr stderr x >> hPutStrLn stderr ": can't read, skip" writeIORef riterloop 0 writeIORef rconsumed False mapM' f xs@@ -1299,18 +1359,18 @@ 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- + tail1 []= [] tail1 x= tail x- + -- | Wait for the execution of `exit` and return the result or the exhaustion of thread activity stay rexit= takeMVar rexit@@ -1319,7 +1379,7 @@ newtype Exit a= Exit a deriving Typeable -- | Runs the transient computation in a child thread and keeps the main thread--- running until all the user threads exit or some thread invokes 'exit'.+-- running until all the user threads exit or some thread 'exit'. -- -- The main thread provides facilities for accepting keyboard input in a -- non-blocking but line-oriented manner. The program reads the standard input@@ -1360,15 +1420,19 @@ forkIO $ do -- liftIO $ putMVar rexit $ Right Nothing runTransient $ do- onException $ \(e :: SomeException ) -> liftIO $ putStr "keep block: " >> print e+ onException $ \(e :: SomeException ) -> liftIO $ putStr "keep block: " >> print e+ onException $ \(e :: IOException) -> do+ when (ioeGetErrorString e == "resource busy") $ do+ liftIO $ do print e ; putStrLn "EXITING!!!"; putMVar rexit Nothing+ empty+ st <- get- setData $ Exit rexit- (abduce >> labelState "input" >> liftIO inputLoop >> empty)+ (abduce >> labelState (fromString "input") >> liftIO inputLoop >> empty) <|> do option "options" "show all options" mbs <- liftIO $ readIORef rcb- liftIO $ mapM_ (\c ->do putStr c; putStr "|") $ M.keys mbs+ liftIO $ mapM_ (\c ->do putStr c; putStr "|") $ map fst mbs liftIO $ putStrLn "" empty <|> do@@ -1378,7 +1442,7 @@ <|> do option "log" "inspect the log of a thread" th <- input (const True) "thread number>"- ml <- liftIO $ showState th st+ ml <- liftIO $ getStateFromThread th st liftIO $ print $ fmap (\(Log _ _ log _) -> reverse log) ml empty <|> do@@ -1413,10 +1477,17 @@ rexit <- newEmptyMVar forkIO $ do runTransient $ do+ onException $ \(e :: SomeException ) -> liftIO $ putStr "keep block: " >> print e+ + 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 - return () + return () threadDelay 10000 forkIO $ execCommandLine stay rexit@@ -1432,7 +1503,7 @@ processLine path --- | Exit the main thread, and thus all the Transient threads (and the+-- | Exit the main thread with a result, and thus all the Transient threads (and the -- application if there is no more code) exit :: Typeable a => a -> TransIO a exit x= do@@ -1470,7 +1541,7 @@ -- | Delete all the undo actions registered till now for the given track id. backCut :: (Typeable b, Show b) => b -> TransientIO () backCut reason= Transient $ do- delData $ Backtrack (Just reason) [] + delData $ Backtrack (Just reason) [] return $ Just () -- | 'backCut' for the default track; equivalent to @backCut ()@.@@ -1486,10 +1557,11 @@ onBack ac bac = registerBack (typeof bac) $ Transient $ do Backtrack mreason stack <- getData `onNothing` (return $ backStateOf (typeof bac)) runTrans $ case mreason of- Nothing -> ac+ Nothing -> ac -- !> "ONBACK NOTHING" Just reason -> do+ -- setState $ Backtrack mreason $ tail stack -- to avoid recursive call tot he same handler- bac reason+ bac reason -- !> ("ONBACK JUST",reason) where typeof :: (b -> TransIO a) -> b typeof = undefined@@ -1507,8 +1579,8 @@ {-# NOINLINE registerUndo #-} registerBack :: (Typeable b, Show b) => b -> TransientIO a -> TransientIO a registerBack witness f = Transient $ do- cont@(EventF _ x _ _ _ _ _ _ _ _ _) <- get - -- if isJust (event cont) then return Nothing else do + cont@(EventF _ x _ _ _ _ _ _ _ _ _) <- get+ -- if isJust (event cont) then return Nothing else do md <- getData `asTypeOf` (Just <$> return (backStateOf witness)) case md of@@ -1534,15 +1606,14 @@ -- XXX Should we enforce retry of the same track which is being undone? If the -- user specifies a different track would it make sense? ----- | For a given undo track id, stop executing more backtracking actions and+-- | For a given undo track type, stop executing more backtracking actions and -- resume normal execution in the forward direction. Used inside an undo -- action. -- forward :: (Typeable b, Show b) => b -> TransIO ()-forward reason= Transient $ do+forward reason= noTrans $ do Backtrack _ stack <- getData `onNothing` ( return $ backStateOf reason) setData $ Backtrack(Nothing `asTypeOf` Just reason) stack- return $ Just () -- | 'forward' for the default undo track; equivalent to @forward ()@. retry= forward ()@@ -1552,42 +1623,57 @@ -- noFinish= continue --- | Start the undo process for the given undo track id. Performs all the undo--- actions registered till now in reverse order. An undo action can use+-- | Start the undo process for a given undo track identifier type. Performs all the undo+-- actions registered for that type in reverse order. An undo action can use -- 'forward' to stop the undo process and resume forward execution. If there--- are no more undo actions registered execution stops and a 'stop' action is--- returned.+-- are no more undo actions registered, execution stop ---back :: (Typeable b, Show b) => b -> TransientIO a+back :: (Typeable b, Show b) => b -> TransIO a back reason = do bs <- getData `onNothing` return (backStateOf reason) - goBackt bs -- !>"GOBACK"+ goBackt bs -- !>"GOBACK" where runClosure :: EventF -> TransIO a runClosure EventF { xcomp = x } = unsafeCoerce x runContinuation :: EventF -> a -> TransIO b- runContinuation EventF { fcomp = fs } = (unsafeCoerce $ compose $ fs)+ runContinuation EventF { fcomp = fs } = (unsafeCoerce $ compose fs) - goBackt (Backtrack _ [] )= empty ++ goBackt (Backtrack _ [] )= empty goBackt (Backtrack b (stack@(first : bs)) )= do setData $ Backtrack (Just reason) stack-- x <- runClosure first -- !> ("RUNCLOSURE",length stack)- return () !> "runclosure"- Backtrack back _ <- getData `onNothing` return (backStateOf reason)- -- !> "END RUNCLOSURE"+ x <- runClosure first -- !> ("RUNCLOSURE",length stack)+ Backtrack back bs' <- getData `onNothing` return (backStateOf reason) case back of- Nothing -> runContinuation first x -- !> "FORWARD EXEC"- justreason ->do+ Nothing -> runContinuation first x -- !> "FORWARD EXEC"+ justreason -> do setData $ Backtrack justreason bs goBackt $ Backtrack justreason bs -- !> ("BACK AGAIN",back)- empty + empty backStateOf :: (Show a, Typeable a) => a -> Backtrack a backStateOf reason= Backtrack (Nothing `asTypeOf` (Just reason)) [] +data BackPoint a = BackPoint (IORef [a -> TransIO()])++-- | a backpoint is a location in the code where callbacks can be installed and will be called when the backtracing pass trough that point.+-- Normally used for exceptions.+backPoint :: (Typeable reason,Show reason) => TransIO (BackPoint reason)+backPoint = do+ point <- liftIO $ newIORef []+ return () `onBack` (\e -> do+ rs <- liftIO $ readIORef point+ mapM_ (\r -> r e) rs)++ return $ BackPoint point++-- | install a callback in a backPoint+onBackPoint (BackPoint ref) handler= liftIO $ atomicModifyIORef ref $ \rs -> (handler:rs,())+++ -- | 'back' for the default undo track; equivalent to @back ()@. -- undo :: TransIO a@@ -1598,7 +1684,7 @@ newtype Finish= Finish String deriving Show -instance Exception Finish +instance Exception Finish -- newtype FinishReason= FinishReason (Maybe SomeException) deriving (Typeable, Show) @@ -1625,7 +1711,6 @@ -- 'initFinish', in reverse order and continue the execution. Either an exception or 'Nothing' can be initFinish = cutExceptions -- passed to 'finish'. The argument passed is made available in the 'onFinish'--- actions invoked. -- finish :: String -> TransIO () finish reason= (throwt $ Finish reason) <|> return()@@ -1649,54 +1734,79 @@ onException :: Exception e => (e -> TransIO ()) -> TransIO () onException exc= return () `onException'` exc +-- | set an exception point. Thi is a point in the backtracking in which exception handlers can be inserted with `onExceptionPoint`+-- it is an specialization of `backPoint` for exceptions.+--+-- 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+-- 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.+--+-- in this case you only need to create a `exceptionPoint` before stablishin the connection, and use `onExceptionPoint`+-- to set a handler that will be called when the connection fail.+exceptionPoint :: Exception e => TransIO (BackPoint e)+exceptionPoint = do+ point <- liftIO $ newIORef []+ return () `onException'` (\e -> do+ rs<- liftIO $ readIORef point+ mapM_ (\r -> r e) rs) -onException' :: Exception e => TransIO a -> (e -> TransIO a) -> TransIO a-onException' mx f= onAnyException mx $ \e ->- case fromException e of- Nothing -> empty- Just e' -> f e'+ return $ BackPoint 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" + case fromException e of++ Nothing -> do+ Backtrack r stack <- getData `onNothing` return (backStateOf e) + setData $ Backtrack r $ tail stack+ back e+ empty+ Just e' -> f e'+++ where onAnyException :: TransIO a -> (SomeException ->TransIO a) -> TransIO a- onAnyException mx f= ioexp f `onBack` f- ioexp f = Transient $ do+ onAnyException mx exc= ioexp `onBack` exc+ + ioexp = Transient $ do st <- get - (mx,st') <- liftIO $ (runStateT + (mr,st') <- liftIO $ (runStateT (do- case event st of + case event st of Nothing -> do- r <- runTrans mx + r <- runTrans mx modify $ \s -> s{event= Just $ unsafeCoerce r}- return () !> "MX"- runCont st + runCont st was <- getData `onNothing` return NoRemote when (was /= WasRemote) $ setData WasParallel return Nothing Just r -> do- modify $ \s -> s{event=Nothing} - return () !> "JUSTTTTTTTTTTT"+ modify $ \s -> s{event=Nothing} return $ unsafeCoerce r) st)- `catch` exceptBack st + `catch` exceptBack st put st'- return mx- + return mr+ exceptBack st = \(e ::SomeException) -> do -- recursive catch itself- runStateT ( runTrans $ back e ) st !> "EXCEPTBACK"+ runStateT ( runTrans $ back e ) st -- !> "EXCEPTBACK" `catch` exceptBack st - -- where- -- -- drop the current exception handler from the stack- -- stex st = - -- let list = mfData st- -- emptyback= backStateOf(undefined :: SomeException)- -- Backtrack b stack = case M.lookup (typeOf emptyback) list of- -- Just x -> unsafeCoerce x- -- Nothing -> emptyback- - -- in st { mfData = M.insert (typeOf emptyback) (unsafeCoerce $ Backtrack b $ tail stack) (mfData st) }-+ -- | Delete all the exception handlers registered till now. cutExceptions :: TransIO () cutExceptions= backCut (undefined :: SomeException)@@ -1711,21 +1821,38 @@ -- 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+ rpassed <- liftIO $ newIORef False sandbox $ do- cutExceptions r <- onException' mx (\e -> do passed <- liftIO $ readIORef rpassed- if not passed then continue >> exc e else empty)+ if not passed then continue >> exc e else do+ Backtrack r stack <- getData `onNothing` return (backStateOf e) + setData $ Backtrack r $ tail stack+ back e+ empty )+ liftIO $ writeIORef rpassed True return r where sandbox mx= do exState <- getState <|> return (backStateOf (undefined :: SomeException)) mx- <*** do setState exState + <*** do setState exState -- | throw an exception in the Transient monad+-- there is a difference between `throw` and `throwt` since the latter preserves the state, while the former does not.+-- Any exception not thrown with `throwt` does not preserve the state.+--+-- > main= keep $ do+-- > onException $ \(e:: SomeException) -> do+-- > v <- getState <|> return "hello"+-- > liftIO $ print v+-- > setState "world"+-- > throw $ ErrorCall "asdasd"+-- +-- the latter print "hello". If you use `throwt` instead, it prints "world" + throwt :: Exception e => e -> TransIO a-throwt= back . toException++throwt = back . toException
src/Transient/Logged.hs view
@@ -48,7 +48,7 @@ import Data.Typeable import Unsafe.Coerce import Transient.Base-import Transient.Internals(Loggable,read')+ import Transient.Indeterminism(choose) import Transient.Internals -- (onNothing,reads1,IDynamic(..),Log(..),LogElem(..),RemoteStatus(..),StateIO) import Control.Applicative@@ -57,7 +57,8 @@ import Control.Exception import Control.Monad import Control.Concurrent.MVar-+import qualified Data.ByteString.Lazy.Char8 as BS+import qualified Data.ByteString.Char8 as BSS #ifndef ghcjs_HOST_OS import System.Random@@ -127,18 +128,18 @@ #endif maybeFromIDyn :: Loggable a => IDynamic -> Maybe a-maybeFromIDyn (IDynamic x)= r+maybeFromIDyn (IDynamic x)= r where- r= if typeOf (Just x) == typeOf r then Just $ unsafeCoerce x else Nothing + r= if typeOf (Just x) == typeOf r then Just $ unsafeCoerce x else Nothing -maybeFromIDyn (IDyns s) = case reads s of- [] -> Nothing- [(x,"")] -> Just x+maybeFromIDyn (IDyns s) = case reads s of+ [] -> Nothing + [(x,"")] -> Just x fromIDyn :: Loggable a => IDynamic -> a fromIDyn (IDynamic x)=r where r= unsafeCoerce x -- !> "coerce" ++ " to type "++ show (typeOf r) -fromIDyn (IDyns s)=r `seq`r where r= read' s -- !> "read " ++ s ++ " to type "++ show (typeOf r)+fromIDyn (IDyns s)=r `seq`r where r= read' s -- !> "read " ++ s ++ " to type "++ show (typeOf r) @@ -161,35 +162,33 @@ runTrans $ case (recover ,rs) of -- !> ("logged enter",recover,rs,reverse full) of (True, Var x: rs') -> do- return () -- !> ("Var:", x)+ 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"+ mx -- !> "Exec" (True, Wait:rs') -> do- setData $ Log True rs' full (hash + 100000) - empty !> "Wait"+ setData $ Log True rs' full (hash + 100000)+ setData WasParallel+ empty -- !> "Wait" _ -> do- -- let add= Exec: full+ setData $ Log False (Exec : rs) (Exec: full) (hash + 1000) -- !> ("setLog False", Exec:rs) - r <- mx <** ( do -- when p1 <|> p2, to avoid the re-execution of p1 at the- -- recovery when p1 is asynchronous- r <- getSData <|> return NoRemote- case r of- WasParallel ->- setData $ Log False (Wait: rs) (Wait: full) (hash+ 100000) - _ -> return ())- + 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+ 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 @@ -207,34 +206,45 @@ received :: Loggable a => a -> TransIO () received n=Transient $ do+ Log recover rs full hash <- getData `onNothing` return ( Log False [][] 0)- case rs of+ return () !> ("RECEIVED log, n", rs,n)++ case rs of [] -> return Nothing Var (IDyns s):t -> if s == show1 n 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 show x+ 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 param :: Loggable a => TransIO a param= res where res= Transient $ do- Log recover rs full hash<- getData `onNothing` return ( Log False [][] 0)+ Log recover rs full hash<- getData `onNothing` return ( Log False [][] 0) + return () !> ("PARAM",rs) case rs of+ [] -> 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 case mr of [] -> return Nothing (v,r):_ -> do- setData $ Log recover t full+ setData $ Log recover t full hash return $ Just v _ -> return Nothing
src/Transient/Parse.hs view
@@ -1,100 +1,290 @@-{-#LANGUAGE ExistentialQuantification, OverloadedStrings, TypeSynonymInstances, FlexibleInstances #-} -module Transient.Parse where -import Transient.Internals -import Data.String -import Data.Typeable -import Control.Applicative -import Data.Char -import Data.Monoid -import System.IO.Unsafe -import Control.Monad -import Control.Monad.State - -import qualified Data.ByteString.Lazy.Char8 as BS - --- | set a stream of strings to be parsed -setParseStream :: (Typeable str,IsString str) => IO str -> TransIO () -setParseStream iox= setState $ ParseContext iox "" - --- | set a string to be parsed -setParseString :: (Typeable str,IsString str) => str -> TransIO () -setParseString x = setState $ ParseContext (error "end of parse string") x - --- | 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 str) str deriving Typeable - - -string :: BS.ByteString -> TransIO BS.ByteString -string s=withData $ \str -> do - let len= BS.length s - ret@(s',_) = BS.splitAt len str - if s == s' - then return ret - else empty - --- many p= do --- r <- p --- rs <- many p <|> return [] --- return $ r:rs - -manyTill p end = scan - where - scan = do{ end; return [] } - <|> - do{ x <- p; xs <- scan; return (x:xs) } - - -dropSpaces= parse $ \str ->((),BS.dropWhile isSpace str) - -dropChar= parse $ \r -> ((), BS.tail r) - -endline c= c== '\r' || c =='\n' - ---tGetLine= tTakeWhile . not . endline - -parseString= do - dropSpaces - tTakeWhile (not . isSpace) - -tTakeWhile :: (Char -> Bool) -> TransIO BS.ByteString -tTakeWhile cond= parse (BS.span cond) - -tTakeWhile' :: (Char -> Bool) -> TransIO BS.ByteString -tTakeWhile' cond= parse ((\(h,t) -> (h, if BS.null t then t else BS.tail t)) . BS.span cond) - -tTake n= parse ( BS.splitAt n) - -tChar= parse $ \s -> (BS.head s,BS.tail s) - -parse :: (BS.ByteString -> (b, BS.ByteString)) -> TransIO b -parse split= withData $ \str -> - if str== mempty then empty - else return $ split str - - --- | bring the data of a parse context as a lazy byteString to a parser --- and actualize the parse context with the result -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 - r <- readMore - (r <>) `liftM` loop - str <- liftIO $ (s <> ) `liftM` loop - mr <- runTrans $ parser str - case mr of - Nothing -> return Nothing - Just (v,str') -> do - setData $ ParseContext readMore str' - return $ Just v - --- | 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 - - let loop = unsafeInterleaveIO $ do - r <- readMore - (r <>) `liftM` loop - liftIO $ (s <> ) `liftM` loop +{-#LANGUAGE FlexibleContexts, ExistentialQuantification, ScopedTypeVariables, OverloadedStrings, TypeSynonymInstances, FlexibleInstances #-}+module Transient.Parse where+import Transient.Internals+import Transient.Indeterminism+import Data.String+import Data.Typeable+import Control.Applicative+import Data.Char+import Data.Monoid++import System.IO.Unsafe+import Control.Monad+import Control.Monad.State+-- import Control.Exception (throw,IOException)+import Control.Concurrent.MVar+++import qualified Data.ByteString.Lazy.Char8 as BS++-- | set a stream of strings to be parsed+setParseStream :: IO (StreamData BS.ByteString) -> TransIO ()++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 ++withParseString :: BS.ByteString -> TransIO a -> TransIO a+withParseString x parse= do+ p@(ParseContext c str) <- getState <|> return(ParseContext (return SDone) mempty)+ setParseString x+ r <- parse+ 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+++-- | succeed if read the string given as parameter+string :: BS.ByteString -> TransIO BS.ByteString+string s= withData $ \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+ else empty -- !> "STRING EMPTY"++-- | fast search for a token+tDropUntilToken token= withData $ \str -> + if BS.null str then empty else drop2 str + where + drop2 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+ where + takeit :: BS.ByteString -> BS.ByteString -> TransIO ( BS.ByteString, BS.ByteString)+ takeit res str= + if BS.null str then return (res,str) 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+ +-- | read an Integer+integer :: TransIO Integer+integer= do+ s <- tTakeWhile isNumber+ if BS.null s then empty else return $ stoi 0 s+ :: TransIO Integer++ where+ 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= do + 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 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+ where+ scan = do{ end; return mempty }+ <|>+ do{ x <- p; xs <- scan; return (x `op` xs) }++between open close p+ = do{ open; x <- p; close; return x }++symbol = string + +parens p = between (symbol "(") (symbol ")") p !> "parens "+braces p = between (symbol "{") (symbol "}") p !> "braces "+angles p = between (symbol "<") (symbol ">") p !> "angles "+brackets p = between (symbol "[") (symbol "]") p !> "brackets "++semi = symbol ";" !> "semi"+comma = symbol "," !> "comma"+dot = symbol "." !> "dot"+colon = symbol ":" !> "colon"++++sepBy p sep = sepBy1 p sep <|> return []+++sepBy1 = chainSepBy1 (:) +++chainSepBy chain p sep= chainSepBy1 chain p sep <|> return mempty++-- take a byteString of elements separated by a separator and apply the desired operator to the parsed results+chainSepBy1+ :: (Monad m, Monoid b, Alternative m) =>+ (a -> b -> b) -> m a -> m x -> m b+chainSepBy1 chain p sep= do{ x <- p+ ; xs <- chainMany chain (sep >> p)+ ; return (x `chain` xs)+ }+ !> "chainSepBy "+ +chainMany chain v= (chain <$> v <*> chainMany chain v) <|> return mempty+ +commaSep p = sepBy p comma+semiSep p = sepBy p semi++commaSep1 p = sepBy1 p comma+semiSep1 p = sepBy1 p semi++dropSpaces= withData $ \str -> return( (),BS.dropWhile isSpace str)+++++dropTillEndOfLine= withData $ \str -> return ((),BS.dropWhile ( /= '\n') str) !> "dropTillEndOfLine"++--manyTill anyChar (tChar '\n' <|> (isDonep >> return ' ') )++parseString= do+ dropSpaces + tTakeWhile (not . isSpace)++-- | take characters while they meet the condition+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)+ + +-- | take characters while they meet the condition and drop the next character+tTakeWhile' :: (Char -> Bool) -> TransIO BS.ByteString+tTakeWhile' cond= withData $ \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) ++ +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)++-- | drop n characters+tDrop n= withData $ \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)++-- | 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) + -- anyChar >>= \x -> if x == c then return c else empty !> ("tChar",x)+++++-- | 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 ++ 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++++-- | 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++ let loop = unsafeInterleaveIO $ do+ mr <- readMore+ case mr of + SMore r -> (r <>) `liftM` loop+ SLast r -> (r <>) `liftM` loop+ SDone -> return mempty+ liftIO $ (s <> ) `liftM` loop)++-- | 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+++ + + + +-- infixl 0 |-++-- | Chain two parsers. The motivation is to parse a chunked HTTP response which contains+-- JSON messages.+--+-- If the REST response is infinite and contains JSON messages, I have to chain the +-- dechunk parser with the JSON decoder of aeson, to produce a stream of aeson messages. +-- Since the boundaries of chunks and JSON messages do not match, it is not possible to add a +-- `decode` to the monadic pipeline. Since the stream is potentially infinite and/or the+-- messages may arrive at any time, I can not wait until all the input finish before decoding +-- the messages.+--+-- I need to generate a ByteString stream with the first parser, which is the input for+-- the second parser. +-- +-- The first parser wait until the second consume the previous chunk, so it is pull-based.+--+-- many parsing stages can be chained with this operator.+--+-- 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+ v <- liftIO $ newEmptyMVar+ initp v <|> initq v++ 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 +
tests/TestSuite.hs view
@@ -13,16 +13,16 @@ import System.Exit import Data.Monoid import Control.Applicative-import Control.Monad.IO.Class+import Control.Monad.State import System.Random import Control.Concurrent import Control.Exception.Base import Data.List -instance Monoid Int where- mempty= 0- mappend = (+)+--instance Monoid Int where+-- mempty = 0+-- mappend = (+) main= do keep' $ do@@ -39,13 +39,14 @@ i <- threads 0 $ choose [1..100] nelems <- liftIO $ randomRIO (1, 10) :: TransIO Int nthreads <- liftIO $ randomRIO (1,nelems)- r <- threads nthreads $ foldr (<>) mempty $map genElem [1..nelems]+ r <- threads nthreads $ foldr (+) 0 $ map genElem [1..nelems] assert (r == sum[1..nelems]) $ return () liftIO $ putStrLn "--------------checking parallel execution, Alternative, events --------" ev <- newEVar r <- collect 3 $ readEVar ev <|> ((choose [1..3] >>= writeEVar ev) >> stop) assert (sort r== [1,2,3]) $ return ()+ liftIO $ print "SUCCESS" exit ()
transient.cabal view
@@ -1,85 +1,90 @@-name: transient -version: 0.6.0.1 -author: Alberto G. Corona -extra-source-files: - ChangeLog.md README.md -maintainer: agocorona@gmail.com -cabal-version: >=1.10 -build-type: Simple -license: MIT -license-file: LICENSE -homepage: https://github.com/transient-haskell/transient -bug-reports: https://github.com/transient-haskell/transient/issues -synopsis: composing programs with multithreading, events and distributed computing -description: See <http://github.com/agocorona/transient> - Distributed primitives are in the transient-universe package. Web primitives are in the axiom package. -category: Control, Concurrency -data-dir: "" - -flag debug - description: Enable debugging outputs - default: False - 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 - - -- libraries not bundled w/ GHC - , mtl - , stm - , random - - ,atomic-primops - - exposed-modules: Transient.Backtrack - Transient.Base - Transient.EVars - Transient.Indeterminism - Transient.Internals - Transient.Logged - Transient.Parse - - exposed: True - buildable: True - default-language: Haskell2010 - hs-source-dirs: src . - - -- eta-options: -ddump-stg -ddump-to-file - - if flag(debug) - cpp-options: -DDEBUG - -source-repository head - type: git - location: https://github.com/agocorona/transient - -test-suite test-transient - - if !impl(ghcjs >=0.1) - build-depends: - base >= 4.8.1 && < 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 - , atomic-primops - type: exitcode-stdio-1.0 - main-is: TestSuite.hs - build-depends: - base >4 - default-language: Haskell2010 - hs-source-dirs: tests src . - ghc-options: -threaded -rtsopts +name: transient+version: 0.6.3+author: Alberto G. Corona+extra-source-files:+ ChangeLog.md README.md+maintainer: agocorona@gmail.com+cabal-version: >=1.10+build-type: Simple+license: MIT+license-file: LICENSE+homepage: https://github.com/transient-haskell/transient+bug-reports: https://github.com/transient-haskell/transient/issues+synopsis: composing programs with multithreading, events and distributed computing+description: See <http://github.com/agocorona/transient>+ Distributed primitives are in the transient-universe package. Web primitives are in the axiom package.+category: Control, Concurrency+data-dir: ""++flag debug+ description: Enable debugging outputs+ default: False+ 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++ -- 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.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++ if flag(debug)+ cpp-options: -DDEBUG++source-repository head+ type: git+ location: https://github.com/agocorona/transient++test-suite test-transient++ if !impl(ghcjs >=0.1)+ build-depends:+ base >= 4.8.1 && < 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+ , atomic-primops+ type: exitcode-stdio-1.0+ main-is: TestSuite.hs+ build-depends:+ base >4+ default-language: Haskell2010+ hs-source-dirs: tests src .+ ghc-options: -threaded -rtsopts