transient 0.5.8 → 0.5.9
raw patch · 7 files changed
+122/−94 lines, 7 filesdep +atomic-primops
Dependencies added: atomic-primops
Files
- README.md +26/−2
- src/Transient/Base.hs +2/−2
- src/Transient/EVars.hs +3/−8
- src/Transient/Indeterminism.hs +10/−7
- src/Transient/Internals.hs +70/−65
- src/Transient/Logged.hs +7/−7
- transient.cabal +4/−3
README.md view
@@ -7,9 +7,11 @@ [](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) +[](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) @@ -39,7 +41,7 @@ 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. +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. @@ -60,3 +62,25 @@ - [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/Base.hs view
@@ -232,10 +232,10 @@ module Transient.Base( -- * The Monad-TransIO(..), TransientIO+TransIO, TransientIO -- * Task Composition Operators-, (**>), (<**), (<***), (<|)+, (**>), (<**), (<***) -- * Running the monad ,keep, keep', stop, exit
src/Transient/EVars.hs view
@@ -1,23 +1,17 @@ {-# LANGUAGE DeriveDataTypeable #-} module Transient.EVars where -import Transient.Base-import Transient.Internals(runTransState,onNothing, EventF(..), killChildren)-import qualified Data.Map as M+import Transient.Internals import Data.Typeable -import Control.Concurrent import Control.Applicative import Control.Concurrent.STM import Control.Monad.IO.Class-import Control.Exception(SomeException) -import Data.List(nub)-import Control.Monad.State-import Data.IORef + data EVar a= EVar (TChan (StreamData a)) deriving Typeable @@ -70,6 +64,7 @@ -- | update the EVar and execute all readEVar blocks with "last in-first out" priority --+writeEVar :: EVar a -> a -> TransIO () writeEVar (EVar ref1) x= liftIO $ atomically $ do writeTChan ref1 $ SMore x
src/Transient/Indeterminism.hs view
@@ -27,6 +27,7 @@ import GHC.Conc import Data.Time.Clock import Control.Exception+import Data.Atomics -- | Converts a list of pure values into a transient task set. You can use the@@ -37,7 +38,7 @@ choose xs = do evs <- liftIO $ newIORef xs r <- parallel $ do- es <- atomicModifyIORef' evs $ \es -> let tes= tail es in (tes,es)+ 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@@ -57,7 +58,7 @@ v <- liftIO $ newIORef (0,[]) x <- proc - mn <- liftIO $ atomicModifyIORef' v $ \(n,xs) ->+ mn <- liftIO $ atomicModifyIORefCAS v $ \(n,xs) -> let n'=n +1 in if n'== num @@ -73,15 +74,17 @@ groupByTime :: Integer -> TransIO a -> TransIO [a] groupByTime time proc = do- v <- liftIO $ newIORef (0,[]) t <- liftIO getCurrentTime++ v <- liftIO $ newIORef (0,t,[])+ x <- proc t' <- liftIO getCurrentTime- mn <- liftIO $ atomicModifyIORef' v $ \(n,xs) -> let n'=n +1+ mn <- liftIO $ atomicModifyIORefCAS v $ \(n,t,xs) -> let n'=n +1 in if diffUTCTime t' t < fromIntegral time- then ((n', x:xs),Nothing)- else ((0,[]), Just $ x:xs)+ then ((n',t, x:xs),Nothing)+ else ((0 ,t',[]), Just $ x:xs) case mn of Nothing -> stop Just xs -> return xs@@ -109,7 +112,7 @@ -- collect' :: Int -> Int -> TransIO a -> TransIO [a] collect' n t search= do-+ addThreads 1 rv <- liftIO $ newEmptyMVar -- !> "NEWMVAR" results <- liftIO $ newIORef (0,[])
src/Transient/Internals.hs view
@@ -47,6 +47,8 @@ import System.IO import qualified Data.ByteString.Char8 as BS+import Data.Atomics+import Data.Typeable #ifdef DEBUG @@ -83,9 +85,7 @@ -- | EventF describes the context of a TransientIO computation: data EventF = forall a b. EventF- { meffects :: ()-- , event :: Maybe SData+ { event :: Maybe SData -- ^ Not yet consumed result (event) from the last asynchronous run of the -- computation @@ -114,10 +114,8 @@ -- ^ Label the thread with its lifecycle state and a label string } deriving Typeable -type Effects =- forall a b c. TransIO a -> TransIO a -> (a -> TransIO b)- -> StateIO (StateIO (Maybe c) -> StateIO (Maybe c), Maybe a) + instance MonadState EventF TransIO where get = Transient $ get >>= return . Just put x = Transient $ put x >> return (Just ())@@ -135,8 +133,7 @@ emptyEventF :: ThreadId -> IORef (LifeCycle, BS.ByteString) -> MVar [EventF] -> EventF emptyEventF th label childs =- EventF { meffects = mempty- , event = mempty+ EventF { event = mempty , xcomp = empty , fcomp = [] , mfData = mempty@@ -189,6 +186,7 @@ -} -- | Compose a list of continuations.+{-# INLINE compose #-} compose :: [a -> TransIO a] -> (a -> TransIO b) compose [] = const empty compose (f:fs) = \x -> f x >>= compose fs@@ -327,12 +325,19 @@ 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+ instance Monad TransIO where return = pure x >>= f = Transient $ do- c <- setEventCont x f+ setEventCont x f mk <- runTrans x- resetEventCont mk c+ resetEventCont mk case mk of Just k -> runTrans (f k) Nothing -> return Nothing@@ -493,44 +498,34 @@ return $ Just x -- | Set the current closure and continuation for the current statement-setEventCont :: TransIO a -> (a -> TransIO b) -> StateIO EventF-setEventCont x f = do- EventF { fcomp = fs, .. } <- get- let cont = EventF { xcomp = x- , fcomp = unsafeCoerce f : fs- , .. }- put cont- return cont+{-# INLINABLE setEventCont #-}+setEventCont :: TransIO a -> (a -> TransIO b) -> StateIO ()+setEventCont x f = modify $ \EventF { fcomp = fs, .. } + -> EventF { xcomp = x+ , fcomp = unsafeCoerce f : fs+ , .. } + -- | Reset the closure and continuation. Remove inner binds than the previous -- computations may have stacked in the list of continuations.--- resetEventCont :: Maybe a -> EventF -> StateIO (TransIO b -> TransIO b)-resetEventCont mx _ = do- EventF { fcomp = fs, .. } <- get- let f mx = case mx of- Nothing -> empty- Just x -> unsafeCoerce (head fs) x- put $ EventF { xcomp = f mx- , fcomp = tailsafe fs- , .. }- return id+-- resetEventCont :: Maybe a -> EventF -> StateIO ()+{-# INLINABLE resetEventCont #-}+resetEventCont mx =+ 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] tailsafe [] = []-tailsafe (x:xs) = xs+tailsafe (_:xs) = xs ---refEventCont= unsafePerformIO $ newIORef baseEffects --- effects not used--- {-# INLINE baseEffects #-}---baseEffects :: Effects------baseEffects x x' f' = do--- c <- setEventCont x' f'--- mk <- runTrans x--- t <- resetEventCont mk c--- return (t,mk) --instance MonadTrans (Transient ) where@@ -538,9 +533,9 @@ -- * Threads -waitQSemB sem = atomicModifyIORef sem $ \n ->+waitQSemB sem = atomicModifyIORefCAS sem $ \n -> if n > 0 then(n - 1, True) else (n, False)-signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())+signalQSemB sem = atomicModifyIORefCAS sem $ \n -> (n + 1, ()) -- | Sets the maximum number of threads that can be created for the given task -- set. When set to 0, new tasks start synchronously in the current thread.@@ -586,7 +581,7 @@ labelState :: (MonadIO m,MonadState EventF m) => String -> m () labelState l = do st <- get- liftIO $ atomicModifyIORef (labelth st) $ \(status,_) -> ((status, BS.pack l), ())+ liftIO $ atomicModifyIORefCAS (labelth st) $ \(status,_) -> ((status, BS.pack l), ()) printBlock :: MVar () printBlock = unsafePerformIO $ newMVar ()@@ -807,7 +802,7 @@ setRState:: Typeable a => a -> TransIO () setRState x= do Ref ref <- getSData- liftIO $ atomicModifyIORef ref $ const (x,())+ liftIO $ atomicModifyIORefCAS ref $ const (x,()) <|> do ref <- liftIO (newIORef x) setData $ Ref ref@@ -817,8 +812,12 @@ 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.+delRState x= delState (undefined `asTypeOf` ref x)+ where ref :: a -> IORef 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 try :: TransIO a -> TransIO a try mx = do sd <- gets mfData@@ -926,7 +925,7 @@ put cont { event = Nothing } return $ unsafeCoerce j Nothing -> do- liftIO $ atomicModifyIORef (labelth cont) $ \(_, lab) -> ((Parent, lab), ())+ liftIO $ atomicModifyIORefCAS (labelth cont) $ \(_, lab) -> ((Parent, lab), ()) liftIO $ loop cont ioaction was <- getData `onNothing` return NoRemote when (was /= WasRemote) $ setData WasParallel@@ -938,7 +937,7 @@ loop :: EventF -> IO (StreamData t) -> IO () loop parentc rec = forkMaybe parentc $ \cont -> do -- Execute the IO computation and then the closure-continuation- liftIO $ atomicModifyIORef (labelth cont) $ const ((Listener,BS.pack "wait"),())+ liftIO $ atomicModifyIORefCAS (labelth cont) $ const ((Listener,BS.pack "wait"),()) let loop'= do mdat <- rec `catch` \(e :: SomeException) -> return $ SError e case mdat of@@ -951,7 +950,7 @@ loop' where- setworker cont= liftIO $ atomicModifyIORef (labelth cont) $ const ((Alive,BS.pack "work"),())+ setworker cont= liftIO $ atomicModifyIORefCAS (labelth cont) $ const ((Alive,BS.pack "work"),()) iocont dat cont = do @@ -963,7 +962,7 @@ loop' return () where-+ {-# INLINABLE forkMaybe #-} forkMaybe parent proc = do case maxThread parent of Nothing -> forkIt parent proc@@ -992,13 +991,14 @@ - _ ->+ _ -> do case maxThread cont of Just sem -> signalQSemB sem -- !> "freed thread"- Nothing -> when(not $ freeTh parent ) $ do -- if was not a free thread+ Nothing -> return ()+ when(not $ freeTh parent ) $ do -- if was not a free thread th <- myThreadId- (can,label) <- atomicModifyIORef (labelth cont) $ \(l@(status,label)) ->+ (can,label) <- atomicModifyIORefCAS (labelth cont) $ \(l@(status,label)) -> ((if status== Alive then Dead else status, label),l) @@ -1011,7 +1011,7 @@ mask $ \restore -> forkIO $ Control.Exception.try (restore action) >>= and_then free th env= do--- return () !> ("freeing",th,"in",threadId env)+ -- return () !> ("freeing",th,"in",threadId env) let sibling= children env (sbs',found) <- modifyMVar sibling $ \sbs -> do@@ -1421,17 +1421,19 @@ {-# NOINLINE registerUndo #-} registerBack :: (Typeable b, Show b) => b -> TransientIO a -> TransientIO a registerBack witness f = Transient $ do- cont@(EventF _ _ x _ _ _ _ _ _ _ _ _) <- get -- !!> "backregister"+ cont@(EventF _ x _ _ _ _ _ _ _ _ _) <- get -- !!> "backregister" 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- addrx' <- addr x' -- to avoid duplicate backtracking points- setData $ if addrx == addrx' then bss else Backtrack mwit (cont:bs)+ Just (Backtrack b []) -> setData $ Backtrack b [cont]+ Just (bss@(Backtrack b (bs@((EventF _ x' _ _ _ _ _ _ _ _ _):_)))) ->+ -- when (isNothing b) $ do+ -- addrx <- addr x+ -- addrx' <- addr x' -- to avoid duplicate backtracking points+ -- setData $ if addrx == addrx' then bss else Backtrack mwit (cont:bs)+ setData $ Backtrack b (cont:bs)+ Nothing -> setData $ Backtrack mwit [cont] runTrans f@@ -1490,7 +1492,10 @@ Nothing -> return empty -- !> "END EXECUTION" Just x -> case back of Nothing -> runContinuation first x -- !> "FORWARD EXEC"- justreason -> goBackt $ Backtrack justreason bs -- !> ("BACK AGAIN",back)+ justreason ->do+ setData $ Backtrack justreason bs+ goBackt $ Backtrack justreason bs+ empty -- !> ("BACK AGAIN",back) backStateOf :: (Monad m, Show a, Typeable a) => a -> m (Backtrack a) backStateOf reason= return $ Backtrack (Nothing `asTypeOf` (Just reason)) []@@ -1529,13 +1534,13 @@ -- | Execute all the finalization actions registered up to the last--- 'initFinish', in reverse order. Either an exception or 'Nothing' can be+-- '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.+-- actions invoked. ---finish :: String -> TransIO a-finish reason= throwt $ Finish reason+finish :: String -> TransIO ()+finish reason= (throwt $ Finish reason) <|> return()
src/Transient/Logged.hs view
@@ -49,13 +49,13 @@ import Transient.Base import Transient.Internals(Loggable) import Transient.Indeterminism(choose)-import Transient.Internals(onNothing,reads1,IDynamic(..),Log(..),LogElem(..),RemoteStatus(..),StateIO)+import Transient.Internals -- (onNothing,reads1,IDynamic(..),Log(..),LogElem(..),RemoteStatus(..),StateIO) import Control.Applicative import Control.Monad.IO.Class import System.Directory import Control.Exception import Control.Monad-+import Control.Concurrent.MVar #ifndef ghcjs_HOST_OS@@ -119,7 +119,7 @@ logAll log= do newlogfile <- liftIO $ (logs ++) <$> replicateM 7 (randomRIO ('a','z')) liftIO $ writeFile newlogfile $ show log- :: TransIO ()+ -- :: TransIO () #endif @@ -144,8 +144,8 @@ -- the parent computation is finished its internal (subcomputation) logs are -- discarded. ---logged :: Loggable a => TransientIO a -> TransientIO a-logged mx = Transient $ do+logged :: Loggable a => TransIO a -> TransIO a+logged mx = Transient $ do Log recover rs full <- getData `onNothing` return ( Log False [][]) runTrans $ case (recover ,rs) of -- !> ("logged enter",recover,rs) of@@ -193,7 +193,7 @@ -------- parsing the log for API's received :: Loggable a => a -> TransIO ()-received n=Transient $ do+received n=Transient $ do Log recover rs full <- getData `onNothing` return ( Log False [][]) case rs of [] -> return Nothing@@ -208,7 +208,7 @@ param :: Loggable a => TransIO a param= res where- res= Transient $ do+ res= Transient $ do Log recover rs full <- getData `onNothing` return ( Log False [][]) case rs of [] -> return Nothing
transient.cabal view
@@ -1,5 +1,5 @@ name: transient-version: 0.5.8+version: 0.5.9 author: Alberto G. Corona extra-source-files: ChangeLog.md README.md@@ -26,7 +26,6 @@ -- support GHC 7.10.3 and later; lower bounds below denote GHC 7.10.3's bundled versions 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@@ -35,7 +34,9 @@ , mtl , stm , random+ , atomic-primops + exposed-modules: Transient.Backtrack Transient.Base Transient.EVars@@ -76,4 +77,4 @@ base >4 default-language: Haskell2010 hs-source-dirs: tests src .- ghc-options: -threaded -rtsopts+ ghc-options: -threaded -rtsopts