transient 0.5.6 → 0.5.8
raw patch · 4 files changed
+199/−164 lines, 4 files
Files
- README.md +62/−62
- src/Transient/Base.hs +1/−1
- src/Transient/Internals.hs +135/−100
- transient.cabal +1/−1
README.md view
@@ -1,62 +1,62 @@--=========--[](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)--NOTE: distributed computing and web primitives have been moved to [transient-universe](https://github.com/agocorona/transient-universe) and [ghcjs-hplay](https://github.com/agocorona/ghcjs-hplay)---## 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 and/or distributed processes. 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).+ +========= + +[](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) + +NOTE: distributed computing and web primitives have been moved to [transient-universe](https://github.com/transient-haskell/transient-universe) and [axiom](https://github.com/transient-haskell/axiom) + + +## 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 and/or distributed processes. 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).
src/Transient/Base.hs view
@@ -249,7 +249,7 @@ ,parallel, async, waitEvents, sample, spawn, react -- * State management-,setData, getSData, getData, delData, modifyData, try, setState, getState, delState, modifyState+,setData, getSData, getData, delData, modifyData, try, setState, getState, delState, getRState,setRState, modifyState -- * Thread management , threads,addThreads, freeThreads, hookedThreads,oneThread, killChilds
src/Transient/Internals.hs view
@@ -56,7 +56,7 @@ {-# INLINE (!>) #-} (!>) :: Show a => b -> a -> b-(!>) x y = trace (show y) x+(!>) x y = trace (show y) x infixr 0 !> #else@@ -342,11 +342,11 @@ -- ex <- liftIO' $ (mx >>= return . Right) `catch` -- (\(e :: SomeException) -> return $ Left e) -- case ex of- -- Left e -> back e -- finish $ Just e+ -- Left e -> back e -- Right x -> return x -- where liftIO x = Transient $ liftIO x >>= return . Just- -- let x= liftIO io in x `seq` lift x+ instance Monoid a => Monoid (TransIO a) where mappend x y = mappend <$> x <*> y@@ -796,6 +796,27 @@ delState :: (MonadState EventF m, Typeable a) => a -> m () delState = delData ++-- STRefs for the Transient monad++newtype Ref a = Ref (IORef a)++-- | mutable state reference that can be updated (similar to STRef in the state monad)+--+-- Initialized the first time it is set.+setRState:: Typeable a => a -> TransIO ()+setRState x= do+ Ref ref <- getSData+ liftIO $ atomicModifyIORef ref $ const (x,())+ <|> do+ ref <- liftIO (newIORef x)+ setData $ Ref ref++getRState :: Typeable a => TransIO a+getRState= do+ Ref ref <- getSData+ liftIO $ readIORef ref+ -- | Run an action, if the result is a void action undo any state changes -- that it might have caused. try :: TransIO a -> TransIO a@@ -803,7 +824,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@@ -885,31 +906,8 @@ v' <- readIORef prev if v /= v' then writeIORef prev v >> return v else loop' ---serial :: IO (StreamData b) -> TransIO (StreamData b)---serial ioaction= Transient $ do--- cont <- get -- !> "PARALLEL"--- case event cont of--- j@(Just _) -> do--- put cont{event=Nothing}--- return $ unsafeCoerce j--- Nothing -> do--- liftIO $ loop cont ioaction--- return Nothing------ where loop cont ioaction= do--- let iocont dat= do--- runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat}--- return ()--- mdat <- ioaction `catch` \(e :: SomeException) -> return $ SError e--- case mdat of--- se@(SError _) -> iocont se--- SDone -> iocont SDone--- last@(SLast _) -> iocont last------ more@(SMore _) -> do--- iocont more--- loop cont ioaction + -- | 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@@ -980,7 +978,7 @@ label <- newIORef (Alive, BS.pack "work") let cont = parent{parent=Just parent,children= chs, labelth= label} - forkFinally1 (do+ forkFinally1 (do th <- myThreadId let cont'= cont{threadId=th} when(not $ freeTh parent )$ hangThread parent cont'@@ -989,17 +987,12 @@ proc cont') $ \me -> do --- case me of -- !> "THREAD END" of--- Left e -> do----- when (fromException e /= Just ThreadKilled)$--- liftIO $ print e--- killChildren $ children cont----- !> "KILL RECEIVED" ++ (show $ unsafePerformIO myThreadId)------ Right _ ->+ case me of+ Left e -> exceptBack cont e >> return () + _ -> case maxThread cont of Just sem -> signalQSemB sem -- !> "freed thread" Nothing -> when(not $ freeTh parent ) $ do -- if was not a free thread@@ -1016,7 +1009,7 @@ 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@@ -1113,26 +1106,9 @@ -- 'empty' in an 'Alternative' composition. abduce = async $ return ()--- Transient $ do--- st <- get--- case event st of--- Just _ -> do--- put st{event=Nothing}--- return $ Just ()--- Nothing -> do--- chs <- liftIO $ newMVar []------ label <- liftIO $ newIORef (Alive, BS.pack "abduce")--- liftIO $ forkIO $ do--- th <- myThreadId--- let st' = st{event= Just (),parent=Just st,children= chs, threadId=th,labelth= label}--- liftIO $ hangThread st st'------ runCont' st'--- return()--- return Nothing + -- * non-blocking keyboard input getLineRef= unsafePerformIO $ newTVarIO Nothing@@ -1176,17 +1152,25 @@ case mr of Nothing -> STM.retry Just r ->- case reads1 r of- (s,_):_ -> if cond s -- !> show (cond s)+ case reads2 r of+ (s,_):_ -> if cond s !> show (cond s) then do unsafeIOToSTM $ print s- writeTVar getLineRef Nothing -- !>"match"+ writeTVar getLineRef Nothing !>"match" return $ Just s else return mv- _ -> return mv+ _ -> return mv !> "return " + where+ reads2 s= x where+ x= if typeOf(typeOfr x) == typeOf "" + then unsafeCoerce[(s,"")] + else unsafePerformIO $ return (reads s) `catch` \(e :: SomeException) -> (return []) + typeOfr :: [(a,String)] -> a+ typeOfr = undefined+ -- | Non blocking `getLine` with a validator getLine' cond= do atomically $ do@@ -1240,7 +1224,7 @@ in breakSlash (res++[r]) $ tail1 rest breakSlash res s=- let (r,rest) = span(/= '/') s+ let (r,rest) = span(\x -> x /= '/' && x /= ' ') s in breakSlash (res++[r]) $ tail1 rest tail1 []=[]@@ -1400,7 +1384,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 ()@.@@ -1418,7 +1402,7 @@ runTrans $ case mreason of Nothing -> ac Just reason -> do- setState $ Backtrack mreason $ tail stack -- to avoid recursive call tot he same handler+ -- setState $ Backtrack mreason $ tail stack -- to avoid recursive call tot he same handler bac reason where typeof :: (b -> TransIO a) -> b@@ -1442,6 +1426,7 @@ md <- getData `asTypeOf` (Just <$> backStateOf witness) case md of+ Just (Backtrack _ []) -> empty Just (bss@(Backtrack b (bs@((EventF _ _ x' _ _ _ _ _ _ _ _ _):_)))) -> when (isNothing b) $ do addrx <- addr x@@ -1477,7 +1462,7 @@ -- | Abort finish. Stop executing more finish actions and resume normal -- execution. Used inside 'onFinish' actions. ---noFinish= forward (FinishReason Nothing)+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@@ -1487,8 +1472,8 @@ -- back :: (Typeable b, Show b) => b -> TransientIO a back reason = Transient $ do- bs <- getData `onNothing` backStateOf reason -- !!>"GOBACK"- goBackt bs+ bs <- getData `onNothing` backStateOf reason + goBackt bs -- !!>"GOBACK" where @@ -1496,16 +1481,16 @@ goBackt (Backtrack b (stack@(first : bs)) )= do setData $ Backtrack (Just reason) stack - mr <- runClosure first -- !> "RUNCLOSURE"+ mr <- runClosure first -- !> ("RUNCLOSURE",length stack) Backtrack back _ <- getData `onNothing` backStateOf reason -- !> "END RUNCLOSURE" case mr of- Nothing -> return empty -- !> "END EXECUTION"+ Nothing -> return empty -- !> "END EXECUTION" Just x -> case back of- Nothing -> runContinuation first x -- !> "FORWARD EXEC"- justreason -> goBackt $ Backtrack justreason bs -- !> ("BACK AGAIN",back)+ Nothing -> 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)) []@@ -1518,35 +1503,39 @@ ------ finalization -newtype FinishReason= FinishReason (Maybe SomeException) deriving (Typeable, Show)+newtype Finish= Finish String deriving Show +instance Exception Finish ++-- newtype FinishReason= FinishReason (Maybe SomeException) deriving (Typeable, Show)+ -- | Clear all finish actions registered till now.-initFinish= backCut (FinishReason Nothing)+-- initFinish= backCut (FinishReason Nothing) -- | Register an action that to be run when 'finish' is called. 'onFinish' can -- be used multiple times to register multiple actions. Actions are run in -- reverse order. Used in infix style. ---onFinish :: ((Maybe SomeException) ->TransIO ()) -> TransIO ()-onFinish f= onFinish' (return ()) f+onFinish :: (Finish ->TransIO ()) -> TransIO ()+onFinish f= onException' (return ()) f -- | Run the action specified in the first parameter and register the second -- parameter as a finish action to be run when 'finish' is called. Used in -- infix style. ---onFinish' ::TransIO a ->((Maybe SomeException) ->TransIO a) -> TransIO a-onFinish' proc f= proc `onBack` \(FinishReason reason) ->- f reason+onFinish' ::TransIO a ->(Finish ->TransIO a) -> TransIO a+onFinish' proc f= proc `onException'` f -- | Execute all the finalization actions registered up to the last -- 'initFinish', in reverse order. Either an exception or 'Nothing' can be+initFinish = cutExceptions -- passed to 'finish'. The argument passed is made available in the 'onFinish' -- actions invoked. ---finish :: Maybe SomeException -> TransIO a-finish reason= back (FinishReason reason)+finish :: String -> TransIO a+finish reason= throwt $ Finish reason @@ -1555,15 +1544,15 @@ case v of SDone -> stop SLast x -> return x- SError e -> back e+ SError e -> throwt e SMore x -> return x ------ exceptions --- ----- | Install an exception handler. On exception, currently installed handlers--- are executed in reverse (i.e. last in first out) order. Note that multiple--- handlers can be installed for the same exception type.+-- | Install an exception handler. Handlers are executed in reverse (i.e. last in, first out) order when such exception happens in the+-- continuation. Note that multiple handlers can be installed for the same exception type. --+-- The semantic is thus very different than the one of `Control.Exception.Base.onException` onException :: Exception e => (e -> TransIO ()) -> TransIO () onException exc= return () `onException'` exc @@ -1575,9 +1564,9 @@ Just e' -> f e' where onAnyException :: TransIO a -> (SomeException ->TransIO a) -> TransIO a- onAnyException mx f= ioexp `onBack` f- - ioexp = Transient $ do + onAnyException mx f= ioexp `onBack` f++ ioexp = Transient $ do st <- get (mx,st') <- liftIO $ (runStateT (do @@ -1586,37 +1575,83 @@ r <- runTrans mx -- !> "mx" modify $ \s -> s{event= Just $ unsafeCoerce r}- - runCont st -- !> "runcont"+ ++ runCont st was <- getData `onNothing` return NoRemote when (was /= WasRemote) $ setData WasParallel return Nothing+ Just r -> do- modify $ \s -> s{event=Nothing} - return $ unsafeCoerce r) st)- `catch` \(e ::SomeException) -> runStateT ( runTrans $ back e ) st+ modify $ \s -> s{event=Nothing} + return $ unsafeCoerce r) st)+ `catch` exceptBack st put st' return mx+ +exceptBack st = \(e ::SomeException) -> do -- recursive catch itself+ -- return () !> "CATCH" + runStateT ( runTrans $ back e ) st+ `catch` exceptBack st ++ + -- | Delete all the exception handlers registered till now.-cutExceptions= backCut (undefined :: SomeException)+cutExceptions :: TransIO ()+cutExceptions= backCut (undefined :: SomeException) --- | Used inside an exception handler. Stop executing any further exception+-- | Use it inside an exception handler. it stop executing any further exception -- handlers and resume normal execution from this point on.----continue = forward (undefined :: SomeException)-+continue :: TransIO ()+continue = forward (undefined :: SomeException) !> "CONTINUE" -catcht mx exc= sandbox $ do+-- | 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 cutExceptions- onException' mx exc+ r <- onException' mx (\e -> do+ passed <- liftIO $ readIORef rpassed+ if not passed then continue >> exc e else empty)+ liftIO $ writeIORef rpassed True+ return r where- sandbox mx= do+ sandbox mx= do exState <- getState <|> backStateOf (undefined :: SomeException) mx- <*** setState exState+ <*** do setState exState +-- | throw an exception in the Transient monad throwt :: Exception e => e -> TransIO a throwt= back . toException ++-- catcht1 :: Exception e => TransIO a -> (e ->TransIO a) -> TransIO a+-- catcht1 mx exc= Transient $ do +-- st <- get+ +-- case event st of +-- Nothing -> do+-- (mx,st') <- liftIO $ (runStateT(runTrans mx) st ) `catch` \(e ::SomeException) -> +-- runStateT ( runTrans $ f e ) st+-- put st'+-- modify $ \s -> s{event= Just $ unsafeCoerce mx}+-- runCont st+-- was <- getData `onNothing` return NoRemote+-- when (was /= WasRemote) $ setData WasParallel++-- return Nothing++-- Just r -> do+-- modify $ \s -> s{event=Nothing} +-- return $ unsafeCoerce r+ +-- where+-- f e=case fromException e of+-- Nothing -> empty+-- Just e' -> exc e'
transient.cabal view
@@ -1,5 +1,5 @@ name: transient-version: 0.5.6+version: 0.5.8 author: Alberto G. Corona extra-source-files: ChangeLog.md README.md