diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,18 +1,18 @@ -Copyright © 2014-2016 Alberto G. Corona - -Permission is hereby granted, free of charge, to any person obtaining a copy of -this software and associated documentation files (the "Software"), to deal in -the Software without restriction, including without limitation the rights to -use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of -the Software, and to permit persons to whom the Software is furnished to do so, -subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS -FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR -COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER -IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +Copyright © 2014-2016 Alberto G. Corona + +Permission is hereby granted, free of charge, to any person obtaining a copy of +this software and associated documentation files (the "Software"), to deal in +the Software without restriction, including without limitation the rights to +use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software is furnished to do so, +subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS +FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER +IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/Setup.hs b/Setup.hs --- a/Setup.hs +++ b/Setup.hs @@ -1,2 +1,2 @@ -import Distribution.Simple -main = defaultMain +import Distribution.Simple +main = defaultMain diff --git a/src/Transient/Base.hs b/src/Transient/Base.hs --- a/src/Transient/Base.hs +++ b/src/Transient/Base.hs @@ -1,35 +1,35 @@ -{-# LANGUAGE ScopedTypeVariables #-} ------------------------------------------------------------------------------ --- --- Module : Base --- Copyright : --- License : GPL (Just (Version {versionBranch = [3], versionTags = []})) --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | See http://github.com/agocorona/transient ------------------------------------------------------------------------------ - -module Transient.Base( - -TransIO(..), TransientIO -,keep, keep', stop -,option, input, exit -,async,waitEvents, spawn, parallel -,react - -,setData,getData,getSData,delData - -, threads,addThreads, freeThreads, hookedThreads,oneThread, killChilds - -, (**>), (<**),(<***), (<|) - -, StreamData(..) -,genId) - -where - - -import Transient.Internals +{-# LANGUAGE ScopedTypeVariables #-} +----------------------------------------------------------------------------- +-- +-- Module : Base +-- Copyright : +-- License : GPL (Just (Version {versionBranch = [3], versionTags = []})) +-- +-- Maintainer : agocorona@gmail.com +-- Stability : +-- Portability : +-- +-- | See http://github.com/agocorona/transient +----------------------------------------------------------------------------- + +module Transient.Base( + +TransIO(..), TransientIO +,keep, keep', stop +,option, input, exit +,async,waitEvents, spawn, parallel, sample +,react + +,setData,getData,getSData,delData + +, threads,addThreads, freeThreads, hookedThreads,oneThread, killChilds + +, (**>), (<**),(<***), (<|) + +, StreamData(..) +,genId) + +where + + +import Transient.Internals diff --git a/src/Transient/EVars.hs b/src/Transient/EVars.hs --- a/src/Transient/EVars.hs +++ b/src/Transient/EVars.hs @@ -1,147 +1,86 @@ -{-# LANGUAGE DeriveDataTypeable #-} -module Transient.EVars where - -import Transient.Base -import Transient.Internals(runTransState,onNothing, EventF(..), killChildren) -import qualified Data.Map as M -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 - - - -data EVar a= EVar (TChan (StreamData a)) deriving Typeable - - --- | creates an EVar. --- --- Evars are event vars. `writeEVar` trigger the execution of all the continuations associated to the `readEVar` of this variable --- (the code that is after them). --- --- It is like the publish-subscribe pattern but without inversion of control, since a readEVar can be inserted at any place in the --- Transient flow. --- --- EVars are created upstream and can be used to communicate two sub-threads of the monad. Following the Transient philosophy they --- do not block his own thread if used with alternative operators, unlike the IORefs and TVars. And unlike STM vars, that are composable, --- they wait for their respective events, while TVars execute the whole expression when any variable is modified. --- --- The execution continues after the writeEVar when all subscribers have been executed. --- --- Now the continuations are executed in parallel. --- --- see https://www.fpcomplete.com/user/agocorona/publish-subscribe-variables-transient-effects-v --- - -newEVar :: TransIO (EVar a) -newEVar = Transient $ do - id <- genId - ref <-liftIO newBroadcastTChanIO - return . Just $ EVar ref - --- | delete al the subscriptions for an evar. -cleanEVar :: EVar a -> TransIO () -cleanEVar (EVar ref1)= liftIO $ atomically $ do - writeTChan ref1 SDone - - --- | read the EVar. It only succeed when the EVar is being updated --- The continuation gets registered to be executed whenever the variable is updated. --- --- if readEVar is re-executed in any kind of loop, since each continuation is different, this will register --- again. The effect is that the continuation will be executed multiple times --- To avoid multiple registrations, use `cleanEVar` -readEVar :: EVar a -> TransIO a -readEVar (EVar ref1)= do - tchan <- liftIO . atomically $ dupTChan ref1 - r <- parallel $ atomically $ readTChan tchan - - case r of - SDone -> empty - SMore x -> return x - SLast x -> return x - SError e -> empty --- error $ "readEVar: "++ show e - --- | update the EVar and execute all readEVar blocks with "last in-first out" priority --- -writeEVar (EVar ref1) x= liftIO $ atomically $ do - writeTChan ref1 $ SMore x - - --- | write the EVar and drop all the `readEVar` handlers. --- --- It is like a combination of `writeEVar` and `cleanEVar` -lastWriteEVar (EVar ref1) x= liftIO $ atomically $ do - writeTChan ref1 $ SLast x - - ----- Finalization --- --- ---type FinishReason= Maybe SomeException --- --- --- ---data Finish= Finish (EVar FinishReason) deriving Typeable --- ----- | initialize the event variable for finalization. ----- all the following computations in different threads will share it ----- it also isolate this event from other branches that may have his own finish variable ---initFinish :: TransIO Finish ---initFinish= do --- fin <- newEVar --- let f = Finish fin --- setData f --- return f --- ----- | set a computation to be called when the finish event happens ---onFinish :: (FinishReason ->TransIO ()) -> TransIO () ---onFinish close= do --- Finish finish <- getSData <|> initFinish --- e <- freeThreads $ readEVar finish --- close e -- !!> "CLOSE" --- stop --- <|> --- return () --- --- --- ----- | trigger the event, so this closes all the resources ---finish :: FinishReason -> TransIO () ---finish e= do --- liftIO $ putStr "finish: " >> print e --- Finish finish <- getSData <|> initFinish --- lastWriteEVar finish e --- ----- | deregister all the finalization actions. ----- A initFinish is needed to register actions again ---unFinish= do --- Finish fin <- getSData --- cleanEVar fin -- !!> "DELEVAR" --- <|> return () -- !!> "NOT DELEVAR" --- --- ----- | kill all the processes generated by the parameter when finish event occurs ---killOnFinish comp= do --- --- chs <- liftIO $ newTVarIO [] --- onFinish $ const $ liftIO $ killChildren chs -- !> "killOnFinish event" --- r <- comp --- modify $ \ s -> s{children= chs} --- return r --- ----- | trigger finish when the stream data return SDone ---checkFinalize v= --- case v of --- SDone -> finish Nothing >> stop --- SLast x -> return x --- SError e -> liftIO ( print e) >> finish Nothing >> stop --- SMore x -> return x - +{-# LANGUAGE DeriveDataTypeable #-} +module Transient.EVars where + +import Transient.Base +import Transient.Internals(runTransState,onNothing, EventF(..), killChildren) +import qualified Data.Map as M +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 + + +-- | creates an EVar. +-- +-- Evars are event vars. `writeEVar` trigger the execution of all the continuations associated to the `readEVar` of this variable +-- (the code that is after them). +-- +-- It is like the publish-subscribe pattern but without inversion of control, since a readEVar can be inserted at any place in the +-- Transient flow. +-- +-- EVars are created upstream and can be used to communicate two sub-threads of the monad. Following the Transient philosophy they +-- do not block his own thread if used with alternative operators, unlike the IORefs and TVars. And unlike STM vars, that are composable, +-- they wait for their respective events, while TVars execute the whole expression when any variable is modified. +-- +-- The execution continues after the writeEVar when all subscribers have been executed. +-- +-- Now the continuations are executed in parallel. +-- +-- see https://www.fpcomplete.com/user/agocorona/publish-subscribe-variables-transient-effects-v +-- + +newEVar :: TransIO (EVar a) +newEVar = Transient $ do + id <- genId + ref <-liftIO newBroadcastTChanIO + return . Just $ EVar ref + +-- | delete al the subscriptions for an evar. +cleanEVar :: EVar a -> TransIO () +cleanEVar (EVar ref1)= liftIO $ atomically $ do + writeTChan ref1 SDone + + +-- | read the EVar. It only succeed when the EVar is being updated +-- The continuation gets registered to be executed whenever the variable is updated. +-- +-- if readEVar is re-executed in any kind of loop, since each continuation is different, this will register +-- again. The effect is that the continuation will be executed multiple times +-- To avoid multiple registrations, use `cleanEVar` +readEVar :: EVar a -> TransIO a +readEVar (EVar ref1)= do + tchan <- liftIO . atomically $ dupTChan ref1 + r <- parallel $ atomically $ readTChan tchan + + case r of + SDone -> empty + SMore x -> return x + SLast x -> return x + SError e -> empty +-- error $ "readEVar: "++ show e + +-- | update the EVar and execute all readEVar blocks with "last in-first out" priority +-- +writeEVar (EVar ref1) x= liftIO $ atomically $ do + writeTChan ref1 $ SMore x + + +-- | write the EVar and drop all the `readEVar` handlers. +-- +-- It is like a combination of `writeEVar` and `cleanEVar` +lastWriteEVar (EVar ref1) x= liftIO $ atomically $ do + writeTChan ref1 $ SLast x + + + diff --git a/src/Transient/Indeterminism.hs b/src/Transient/Indeterminism.hs --- a/src/Transient/Indeterminism.hs +++ b/src/Transient/Indeterminism.hs @@ -1,137 +1,137 @@ ------------------------------------------------------------------------------ --- --- Module : Transient.Indeterminism --- Copyright : --- License : GPL (Just (Version {versionBranch = [3], versionTags = []})) --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | see --- ------------------------------------------------------------------------------ -{-# LANGUAGE BangPatterns #-} -module Transient.Indeterminism ( -choose, choose', collect, collect', group, groupByTime -) where - -import Transient.Base -import Transient.Backtrack(checkFinalize) -import Transient.Internals(killChildren, EventF(..),hangThread) -import Data.IORef -import Control.Applicative -import Data.Monoid -import Control.Concurrent -import Data.Typeable -import Control.Monad.State -import Control.Concurrent.STM as STM -import GHC.Conc -import Data.Time.Clock - - --- | slurp a list of values and process them in parallel . To limit the number of processing --- threads, use `threads` -choose :: Show a => [a] -> TransIO a -choose []= empty -choose xs = do - evs <- liftIO $ newIORef xs - r <- parallel $ do - es <- atomicModifyIORef' evs $ \es -> let !tes= tail es in (tes,es) - case es of - [x] -> x `seq` return $ SLast x - x:_ -> x `seq` return $ SMore x - checkFinalize r - - --- | group the output of a possible multithreaded process in groups of n elements. -group :: Int -> TransIO a -> TransIO [a] -group num proc = do - v <- liftIO $ newIORef (0,[]) - x <- proc - - mn <- liftIO $ atomicModifyIORef' v $ \(n,xs) -> - let !n'=n +1 - in if n'== num - - then ((0,[]), Just xs) - else ((n', x:xs),Nothing) - case mn of - Nothing -> stop - Just xs -> return xs - --- | group result for a time interval, measured with `diffUTCTime` -groupByTime :: Integer -> TransIO a -> TransIO [a] - -groupByTime time proc = do - v <- liftIO $ newIORef (0,[]) - t <- liftIO getCurrentTime - x <- proc - t' <- liftIO getCurrentTime - mn <- liftIO $ atomicModifyIORef' v $ \(n,xs) -> let !n'=n +1 - in - if diffUTCTime t' t < fromIntegral time - then ((n', x:xs),Nothing) - else ((0,[]), Just xs) - case mn of - Nothing -> stop - Just xs -> return xs - --- | alternative definition with more parallelism, as the composition of n `async` sentences -choose' :: [a] -> TransIO a -choose' xs = foldl (<|>) empty $ map (async . return) xs - - --- collect the results of a search done in parallel, usually initiated by --- `choose` . --- --- execute a process and get at least the first n solutions (they could be more). --- if the process end without finding the number of solutions requested, it return the found ones --- if he find the number of solutions requested, it kill the non-free threads of the process and return --- It works monitoring the solutions found and the number of active threads. --- If the first parameter is 0, collect will return all the results -collect :: Int -> TransIO a -> TransIO [a] -collect n = collect' n 0.1 0 - --- | search also between two time intervals. If the first interval has passed and there is no result, ---it stops. --- After the second interval, it stop unconditionally and return the current results. --- It also stops as soon as there are enough results specified in the first parameter. -collect' :: Int -> NominalDiffTime -> NominalDiffTime -> TransIO a -> TransIO [a] -collect' n t1 t2 search= hookedThreads $ do - rv <- liftIO $ atomically $ newTVar (0,[]) -- !> "NEWMVAR" - endflag <- liftIO $ newTVarIO False - st <- newPool - t <- liftIO getCurrentTime - let worker = do - r <- search - liftIO $ atomically $ do - (n1,rs) <- readTVar rv - writeTVar rv (n1+1,r:rs) -- !> "MODIFY" - stop - - monitor= freeThreads $ do - xs <- async $ atomically $ - do (n', xs) <- readTVar rv - ns <- readTVar $ children st - t' <- unsafeIOToSTM getCurrentTime - if - (n > 0 && n' >= n) || - (null ns && (diffUTCTime t' t > t1)) || - (t2 > 0 && diffUTCTime t' t > t2) - -- !> (diffUTCTime t' t, n', length ns) - then return xs else retry - - - liftIO . killChildren $ children st - - return xs - - monitor <|> worker - where - newPool = do - chs <- liftIO $ newTVarIO [] - s <- get - let s'= s{children= chs} - put s' - return s' +----------------------------------------------------------------------------- +-- +-- Module : Transient.Indeterminism +-- Copyright : +-- License : GPL (Just (Version {versionBranch = [3], versionTags = []})) +-- +-- Maintainer : agocorona@gmail.com +-- Stability : +-- Portability : +-- +-- | see +-- +----------------------------------------------------------------------------- +{-# LANGUAGE BangPatterns #-} +module Transient.Indeterminism ( +choose, choose', collect, collect', group, groupByTime +) where + +import Transient.Base +import Transient.Backtrack(checkFinalize) +import Transient.Internals(killChildren, EventF(..),hangThread) +import Data.IORef +import Control.Applicative +import Data.Monoid +import Control.Concurrent +import Data.Typeable +import Control.Monad.State +import Control.Concurrent.STM as STM +import GHC.Conc +import Data.Time.Clock + + +-- | slurp a list of values and process them in parallel . To limit the number of processing +-- threads, use `threads` +choose :: Show a => [a] -> TransIO a +choose []= empty +choose xs = do + evs <- liftIO $ newIORef xs + r <- parallel $ do + es <- atomicModifyIORef' evs $ \es -> let !tes= tail es in (tes,es) + case es of + [x] -> x `seq` return $ SLast x + x:_ -> x `seq` return $ SMore x + checkFinalize r + + +-- | group the output of a possible multithreaded process in groups of n elements. +group :: Int -> TransIO a -> TransIO [a] +group num proc = do + v <- liftIO $ newIORef (0,[]) + x <- proc + + mn <- liftIO $ atomicModifyIORef' v $ \(n,xs) -> + let !n'=n +1 + in if n'== num + + then ((0,[]), Just xs) + else ((n', x:xs),Nothing) + case mn of + Nothing -> stop + Just xs -> return xs + +-- | group result for a time interval, measured with `diffUTCTime` +groupByTime :: Integer -> TransIO a -> TransIO [a] + +groupByTime time proc = do + v <- liftIO $ newIORef (0,[]) + t <- liftIO getCurrentTime + x <- proc + t' <- liftIO getCurrentTime + mn <- liftIO $ atomicModifyIORef' v $ \(n,xs) -> let !n'=n +1 + in + if diffUTCTime t' t < fromIntegral time + then ((n', x:xs),Nothing) + else ((0,[]), Just xs) + case mn of + Nothing -> stop + Just xs -> return xs + +-- | alternative definition with more parallelism, as the composition of n `async` sentences +choose' :: [a] -> TransIO a +choose' xs = foldl (<|>) empty $ map (async . return) xs + + +-- collect the results of a search done in parallel, usually initiated by +-- `choose` . +-- +-- execute a process and get at least the first n solutions (they could be more). +-- if the process end without finding the number of solutions requested, it return the found ones +-- if he find the number of solutions requested, it kill the non-free threads of the process and return +-- It works monitoring the solutions found and the number of active threads. +-- If the first parameter is 0, collect will return all the results +collect :: Int -> TransIO a -> TransIO [a] +collect n = collect' n 0.1 0 + +-- | search also between two time intervals. If the first interval has passed and there is no result, +--it stops. +-- After the second interval, it stop unconditionally and return the current results. +-- It also stops as soon as there are enough results specified in the first parameter. +collect' :: Int -> NominalDiffTime -> NominalDiffTime -> TransIO a -> TransIO [a] +collect' n t1 t2 search= hookedThreads $ do + rv <- liftIO $ atomically $ newTVar (0,[]) -- !> "NEWMVAR" + endflag <- liftIO $ newTVarIO False + st <- newPool + t <- liftIO getCurrentTime + let worker = do + r <- search + liftIO $ atomically $ do + (n1,rs) <- readTVar rv + writeTVar rv (n1+1,r:rs) -- !> "MODIFY" + stop + + monitor= freeThreads $ do + xs <- async $ atomically $ + do (n', xs) <- readTVar rv + ns <- readTVar $ children st + t' <- unsafeIOToSTM getCurrentTime + if + (n > 0 && n' >= n) || + (null ns && (diffUTCTime t' t > t1)) || + (t2 > 0 && diffUTCTime t' t > t2) + -- !> (diffUTCTime t' t, n', length ns) + then return xs else retry + + + liftIO . killChildren $ children st + + return xs + + monitor <|> worker + where + newPool = do + chs <- liftIO $ newTVarIO [] + s <- get + let s'= s{children= chs} + put s' + return s' diff --git a/src/Transient/Internals.hs b/src/Transient/Internals.hs --- a/src/Transient/Internals.hs +++ b/src/Transient/Internals.hs @@ -1,1056 +1,1070 @@ -{-# LANGUAGE ScopedTypeVariables #-} ------------------------------------------------------------------------------ --- --- Module : Base --- Copyright : --- License : GPL (Just (Version {versionBranch = [3], versionTags = []})) --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | See http://github.com/agocorona/transient --- everithing in this module is exported in order to allow extensibility. ------------------------------------------------------------------------------ -{-# LANGUAGE ExistentialQuantification #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE Rank2Types #-} --- show -module Transient.Internals where --- /show - -import Control.Applicative -import Control.Monad.State -import Data.Dynamic -import qualified Data.Map as M -import Data.Monoid -import Debug.Trace -import System.IO.Unsafe -import Unsafe.Coerce -import Control.Exception -import Control.Concurrent -import Control.Concurrent.STM -import System.Mem.StableName -import Data.Maybe -import GHC.Conc -import Data.List -import Data.IORef -import System.Environment -import System.IO (hFlush,stdout) -import System.Exit -{-# INLINE (!>) #-} -(!>) :: Show a => b -> a -> b -(!>) x y= trace (show y) x -infixr 0 !> - - -data TransIO x = Transient {runTrans :: StateT EventF IO (Maybe x)} -type SData= () - -type EventId= Int - -type TransientIO= TransIO - -data EventF = forall a b . EventF{meffects :: Effects - ,event :: Maybe SData - ,xcomp :: TransIO a - ,fcomp :: [b -> TransIO b] - ,mfData :: M.Map TypeRep SData - ,mfSequence :: Int - ,threadId :: ThreadId - ,freeTh :: Bool - ,parent :: Maybe EventF - ,children :: TVar[EventF] - ,maxThread :: Maybe (IORef Int) - } - 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 ()) - state f = Transient $ do - s <- get - let ~(a, s') = f s - put s' - return $ Just a - -type StateIO= StateT EventF IO - - --- | run the transient computation with a blank state -runTransient :: TransIO x -> IO (Maybe x, EventF) -runTransient t= do - th <- myThreadId - let eventf0= EventF baseEffects Nothing empty [] M.empty 0 - th False Nothing (unsafePerformIO $ newTVarIO []) Nothing - - - runStateT (runTrans t) eventf0 - --- | run the transient computation with an state -runTransState st x = runStateT (runTrans x) st - --- | get the continuation context: closure, continuation, state, child threads etc -getCont :: TransIO EventF -getCont = Transient $ Just <$> get - --- | run the closure and the continuation using the state data of the calling thread -runCont :: EventF -> StateIO (Maybe a) -runCont (EventF _ _ x fs _ _ _ _ _ _ _)= runTrans $ do - r <- unsafeCoerce x - compose fs r - --- | run the closure and the continuation using his own state data -runCont' cont= runStateT (runCont cont) cont - --- | warning: radiactive untyped stuff. handle with care -getContinuations :: StateIO [a -> TransIO b] -getContinuations= do - EventF _ _ _ fs _ _ _ _ _ _ _ <- get - return $ unsafeCoerce fs - -{- -runCont cont= do - mr <- runClosure cont - case mr of - Nothing -> return Nothing - Just r -> runContinuation cont r --} - - --- | compose a list of continuations -compose []= const empty -compose (f: fs)= \x -> f x >>= compose fs - - - --- | run the closure (the 'x' in 'x >>= f') of the current bind operation. -runClosure :: EventF -> StateIO (Maybe a) -runClosure (EventF _ _ x _ _ _ _ _ _ _ _) = unsafeCoerce $ runTrans x - - --- | run the continuation (the 'f' in 'x >>= f') of the current bind operation -runContinuation :: EventF -> a -> StateIO (Maybe b) -runContinuation (EventF _ _ _ fs _ _ _ _ _ _ _) = - runTrans . (unsafeCoerce $ compose $ fs) - - -setContinuation :: TransIO a -> (a -> TransIO b) -> [c -> TransIO c] -> StateIO () -setContinuation b c fs = do - (EventF eff ev _ _ d e f g h i j) <- get - put $ EventF eff ev b ( unsafeCoerce c: fs) d e f g h i j - -withContinuation c mx= do - EventF eff ev f1 fs d e f g h i j <- get - put $ EventF eff ev mx ( unsafeCoerce c: fs) d e f g h i j - r <- mx - restoreStack fs - return r - --- | run a chain of continuations. It is up to the programmer to assure by construction that --- each continuation type-check with the next, that the parameter type match the input of the first --- continuation. --- Normally this makes sense if it stop the current flow with `stop` after the invocation -runContinuations :: [a -> TransIO b] -> c -> TransIO d -runContinuations fs x= (compose $ unsafeCoerce fs) x - -instance Functor TransIO where - fmap f mx= -- Transient $ fmap (fmap f) $ runTrans 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,[]) -- !> "NEWIOREF" - - fs <- getContinuations - - let - - hasWait (_:Wait:_)= True - hasWait _ = False - - appf k = Transient $ do - Log rec _ full <- getData `onNothing` return (Log False [] []) - (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 [] []) - 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" - was <- getData `onNothing` return NoRemote - when (was == WasParallel) $ setData NoRemote - - Log recovery _ full <- getData `onNothing` return (Log False [] []) - - - - 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 [] []) - 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 - -restoreStack fs= - modify $ \(EventF eff _ f _ a b c d parent children g1) -> - EventF eff Nothing f fs a b c d parent children g1 - -readWithErr line= - let [(v,left)] = readsPrec 0 line - in (v `seq` return [(v,left)]) - `catch` (\(e::SomeException) -> - error ("read error of " ++ show( typeOf v) ++ " in: "++ "\""++line++"\"")) - - -readsPrec' _= unsafePerformIO . readWithErr - - - --- | dynamic serializable data for logging -data IDynamic= IDyns String | forall a.(Read a, Show a,Typeable a) => IDynamic a - -instance Show IDynamic where - show (IDynamic x)= show $ show x - show (IDyns s)= show s - -instance Read IDynamic where - readsPrec n str= map (\(x,s) -> (IDyns x,s)) $ readsPrec' n str - - -type Recover= Bool -type CurrentPointer= [LogElem] -type LogEntries= [LogElem] -data LogElem= Wait | Exec | Var IDynamic deriving (Read,Show) -data Log= Log Recover CurrentPointer LogEntries deriving Typeable - - -instance Alternative TransIO where - empty = Transient $ return Nothing - (<|>) = mplus - - -data RemoteStatus= WasRemote | WasParallel | NoRemote deriving (Typeable, Eq, Show) - -instance MonadPlus TransIO where - mzero= empty - mplus x y= Transient $ do - mx <- runTrans x -- !!> "RUNTRANS11111" - was <- getData `onNothing` return NoRemote - if was== WasRemote -- !> was - then return Nothing - else - case mx of - Nothing -> runTrans y -- !!> "RUNTRANS22222" - justx -> return justx - --- | a sinonym of empty that can be used in a monadic expression. it stop the --- computation and execute the next alternative computation (composed with `<|>`) -stop :: Alternative m => m stopped -stop= empty - -class AdditionalOperators m where - - -- | executes the second operand even if the frist return empty. - -- A normal imperative (monadic) sequence uses the operator (>>) which in the - -- Transient monad does not execute the next operand if the previous one return empty. - (**>) :: m a -> m b -> m b - - -- | forces the execution of the second operand even if the first stop. It does not execute - -- the second operand as result of internal events occuring in the first operand. - -- Return the first result - (<**) :: m a -> m b -> m a - - atEnd' ::m a -> m b -> m a - atEnd' = (<**) - - -- | forces the execution of the second operand even if the first stop. Return the first result. The second - -- operand is executed also when internal events happens in the first operand and it returns something - (<***) :: m a -> m b -> m a - - atEnd :: m a -> m b -> m a - atEnd= (<***) - - -instance AdditionalOperators TransIO where - --- (**>) :: TransIO a -> TransIO b -> TransIO b - (**>) x y= Transient $ do - runTrans x - runTrans y - --- (<***) :: TransIO a -> TransIO b -> TransIO a - (<***) ma mb= Transient $ do - fs <- getContinuations - setContinuation ma (\x -> mb >> return x) fs - a <- runTrans ma - runTrans mb - restoreStack fs - return a - --- (<**) :: TransIO a -> TransIO b -> TransIO a - (<**) ma mb= Transient $ do - a <- runTrans ma -- !> "ma" - runTrans mb -- !> "mb" - return a - -infixr 1 <*** , <**, **> - - - --- | when the first operand is an asynchronous operation, the second operand is executed once (one single time) --- when the first completes his first asyncronous operation. --- --- This is useful for spawning asynchronous or distributed tasks that are singletons and that should start --- when the first one is set up. --- --- for example a streaming where the event receivers are acivated before the senders. - -(<|) :: TransIO a -> TransIO b -> TransIO a -(<|) ma mb = Transient $ do - fs <- getContinuations - ref <- liftIO $ newIORef False - setContinuation ma (cont ref ) fs - r <- runTrans ma - restoreStack fs - return r - where - cont ref x= Transient $ do - n <- liftIO $ readIORef ref - if n == True - then return $ Just x - else do liftIO $ writeIORef ref True - runTrans mb - return $ Just x - -instance Monoid a => Monoid (TransIO a) where - mappend x y = mappend <$> x <*> y - mempty= return mempty - --- | set the current closure and continuation for the current statement -setEventCont :: TransIO a -> (a -> TransIO b) -> StateIO EventF -setEventCont x f = do - - st@(EventF eff e _ fs d n r applic ch rc bs) <- get -- !> "SET" - let cont= EventF eff e x ( unsafeCoerce f : fs) d n r applic ch rc bs - put cont - return cont - --- | 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 - st@(EventF eff e _ fs d n r nr ch rc bs) <- get -- !> "reset" - let f= \mx -> case mx of - Nothing -> empty - Just x -> (unsafeCoerce $ head fs) x - put $ EventF eff e (f mx) ( tailsafe fs) d n r nr ch rc bs - return id - -tailsafe []=[] -tailsafe (x:xs)= xs - ---refEventCont= unsafePerformIO $ newIORef baseEffects - -{-# INLINE baseEffects #-} -baseEffects :: Effects - -baseEffects x x' f' = do - c <-setEventCont x' f' - mk <- runTrans x - t <- resetEventCont mk c - return (t,mk) - -instance Monad TransIO where - - return = pure - - x >>= f = Transient $ do --- effects <- gets effects -- liftIO $ readIORef refEventCont - (t,mk) <- baseEffects x x f - t $ case mk of - Just k -> runTrans (f k) - - Nothing -> return Nothing - ---instance MonadTrans (Transient ) where --- lift mx = Transient $ mx >>= return . Just - -instance MonadIO TransIO where - liftIO x = Transient $ liftIO x >>= return . Just -- let x= liftIO io in x `seq` lift x - - --- * Threads - -waitQSemB sem= atomicModifyIORef sem $ \n -> if n > 0 then(n-1,True) else (n,False) -signalQSemB sem= atomicModifyIORef sem $ \n -> (n + 1,()) - --- | set the maximun number of threads for a procedure. It is useful to limit the --- parallelization of transient code that uses `parallel` `spawn` and `waitEvents` -threads :: Int -> TransIO a -> TransIO a -threads n proc= Transient $ do - msem <- gets maxThread - sem <- liftIO $ newIORef n - modify $ \s -> s{maxThread= Just sem} - r <- runTrans proc - modify $ \s -> s{maxThread = msem} -- restore it - return r - --- | delete all the previous childs generated by the expression taken as parameter and continue execution --- of the current thread. -oneThread :: TransIO a -> TransientIO a -oneThread comp= do - chs <- liftIO $ newTVarIO [] - r <- comp - modify $ \ s -> s{children= chs} - killChilds - return r - - -showThreads :: TransIO empty -showThreads= do - st' <- gets (fromJust . parent) - liftIO $ showTree 0 st' - stop - where - toplevel st = - case parent st of - Nothing -> st - Just p -> toplevel p - - showThreads' n rchs= do - chs <- atomically $ readTVar rchs - mapM_ (showTree n) chs - - showTree n ch= do - putStr $ take n $ repeat ' ' - print $ threadId ch - showThreads' (n+4) $ children ch - - --- | add n threads to the limit of threads. If there is no limit, it set it -addThreads' :: Int -> TransIO () -addThreads' n= Transient $ do - msem <- gets maxThread - case msem of - Just sem -> liftIO $ modifyIORef sem $ \n' -> n + n' - Nothing -> do - sem <- liftIO (newIORef n) - modify $ \ s -> s{maxThread= Just sem} - return $ Just () - --- | assure that at least there are n threads available -addThreads n= Transient $ do - msem <- gets maxThread - case msem of - Nothing -> return () - Just sem -> liftIO $ modifyIORef sem $ \n' -> if n' > n then n' else n - return $ Just () ---getNonUsedThreads :: TransIO (Maybe Int) ---getNonUsedThreads= Transient $ do --- msem <- gets maxThread --- case msem of --- Just sem -> liftIO $ Just <$> readIORef sem --- Nothing -> return Nothing - - --- | The threads generated in the process passed as parameter will not be killed by `kill*` primitives -freeThreads :: TransIO a -> TransIO a -freeThreads proc= Transient $ do - st <- get - put st{freeTh= True} - r <- runTrans proc - modify $ \s -> s{freeTh= freeTh st} - return r - --- | The threads will be killed when the parent thread dies. That is the default. --- This can be invoked to revert the effect of `freeThreads` -hookedThreads :: TransIO a -> TransIO a -hookedThreads proc= Transient $ do - st <- get - put st{freeTh= False} - r <- runTrans proc - modify $ \st -> st{freeTh= freeTh st} - return r - --- | kill all the child threads of the current thread -killChilds :: TransientIO() -killChilds= Transient $ do - cont <- get - liftIO $ killChildren $ children cont - return $ Just () - --- * extensible state: session data management - --- | Get the state data for the desired type if there is any. -getData :: (MonadState EventF m,Typeable a) => m (Maybe a) -getData = resp where - resp= gets mfData >>= \list -> - case M.lookup ( typeOf $ typeResp resp ) list of - Just x -> return . Just $ unsafeCoerce x - Nothing -> return Nothing - typeResp :: m (Maybe x) -> x - typeResp= undefined - - --- | getData specialized for the Transient monad. if Nothing, the --- monadic computation does not continue. --- --- If there is no such data, `getSData` silently stop the computation. --- That may or may not be the desired behaviour. --- To make sure that this does not get unnoticed, use this construction: --- --- > getSData <|> error "no data" --- --- To have the same semantics and guarantees than `get`, use a default value: --- --- > getInt= getSData <|> return (0 :: Int) --- --- The default value (0 in this case) has the same role than the initial value in a state monad. --- The difference is that you can define as many `get` as you need for all your data types. --- --- To distingish two data with the same types, use newtype definitions. -getSData :: Typeable a => TransIO a -getSData= Transient getData - - - --- | set session data for this type. retrieved with getData or getSData --- Note that this is data in a state monad, that means that the update only affect downstream --- in the monad execution. it is not a global state neither a per user or per thread state --- it is a monadic state like the one of a state monad. -setData :: (MonadState EventF m, Typeable a) => a -> m () -setData x= - let t= typeOf x in modify $ \st -> st{mfData= M.insert t (unsafeCoerce x) (mfData st)} - - -delData :: ( MonadState EventF m,Typeable a) => a -> m () -delData x= modify $ \st -> st{mfData= M.delete (typeOf x ) (mfData st)} - - ---withSData :: ( MonadState EventF m,Typeable a) => (Maybe a -> a) -> m () ---withSData f= modify $ \st -> st{mfData= --- let dat = mfData st --- mx= M.lookup typeofx dat --- mx'= case mx of Nothing -> Nothing; Just x -> unsafeCoerce x --- fx= f mx' --- typeofx= typeOf $ typeoff f --- in M.insert typeofx (unsafeCoerce fx) dat} --- where --- typeoff :: (Maybe a -> a) -> a --- typeoff = undefined ----- - --- | generator of identifiers that are unique withing the current monadic sequence --- They are not unique in the whole program. -genId :: MonadState EventF m => m Int -genId= do - st <- get - let n= mfSequence st - put st{mfSequence= n+1} - return n - -getPrevId :: MonadState EventF m => m Int -getPrevId= do - n <- gets mfSequence - return n - -instance Read SomeException where - readsPrec n str= - let [(s , r)]= read str in [(SomeException $ ErrorCall s,r)] - --- | async calls - -data StreamData a= SMore a | SLast a | SDone | SError SomeException deriving (Typeable, Show,Read) - - --- | variant of `parallel` that repeatedly executes the IO computation and kill the previously created childs --- --- It is useful in single threaded problems where each event discard the computations spawned by --- previous events -waitEvents :: IO b -> TransIO b -waitEvents io= do - mr <- parallel (SMore <$> io) - case mr of - SMore x -> return x - SError e -> throw e - - --- Multithreaded version of `waitEvents` that do not kill the computations spawned by previous events -waitEvents' :: IO b -> TransIO b -waitEvents' io= do - mr <- parallel (SMore <$> io) - case mr of - SMore x -> return x - SError e -> throw e - --- | variant of `parallel` that execute the IO computation once, and kill the previous child threads -async :: IO b -> TransIO b -async io= do - mr <- parallel (SLast <$> io) - case mr of - SLast x -> return x - SError e -> throw e - --- | variant of waitEvents that spawn free threads. It is a little faster at the cost of no thread control -spawn :: IO b -> TransIO b -spawn io= freeThreads $ do - mr <- parallel (SMore <$>io) - case mr of - SMore x -> return x - SError e -> throw e - - - - - - --- | return empty to the current thread, in new thread, execute the IO action, --- this IO action modify an internal buffer. then, executes the closure where `parallel` is located --- In this new execution, since the buffer is filled, `parallel` return the content of this buffer. --- Then it launch the continuation after it with this new value returned by the closure. --- --- If the maximum number of threads, set with `threads` has been reached `parallel` perform --- the work sequentially, in the current thread. --- So `parallel` means that 'it can be parallelized if there are thread available' --- --- if there is a limitation of threads, when a thread finish, the counter of threads available --- is increased so another `parallel` can make use of it. --- --- The behaviour of `parallel` depend on `StreamData`; If `SMore`, `parallel` will excute again the --- IO action. with `SLast`, `SDone` and `SError`, `parallel` will not repeat the IO action anymore. -parallel :: IO (StreamData b) -> TransIO (StreamData b) -parallel 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 - was <- getData `onNothing` return NoRemote - when (was /= WasRemote) $ setData WasParallel - return Nothing - - --- executes the IO action and then the continuation included in the first parameter -loop :: EventF -> IO (StreamData t) -> IO () -loop (cont'@(EventF eff e x fs a b c d _ childs g)) rec = do - chs <- liftIO $ newTVarIO [] - let cont = EventF eff e x fs a b c d (Just cont') chs g - iocont dat= do - runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat} - return () - - -- execute the IO computation and then the closure-continuation - loop'= forkMaybe False cont $ do - mdat <- threadDelay 0 >> rec `catch` \(e :: SomeException) -> return $ SError e - case mdat of - se@(SError _) -> iocont se - SDone -> iocont SDone - last@(SLast _) -> iocont last - - more@(SMore _) -> do - forkMaybe False cont $ iocont more - loop' - loop' - return () - where - forkMaybe True cont proc = forkMaybe' True cont proc - forkMaybe False cont proc = do - dofork <- case maxThread cont of - Nothing -> return True - Just sem -> do - dofork <- waitQSemB sem - if dofork then return True else return False - forkMaybe' dofork cont proc - - forkMaybe' dofork cont proc= - if dofork - then do - forkFinally1 (do - th <- myThreadId - hangThread cont' cont{threadId=th} -- !> "thread created: "++ show th - proc) - $ \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 _ -> when(not $ freeTh cont') $ do -- if was not a free thread - -- if parent is alive - -- then remove himself from the parent list (with free) - -- and pass his active children to his parent - - th <- myThreadId - mparent <- free th cont - return () - -- pass the active children to the parent --- case mparent of --- Nothing -> return() --- Just parent -> atomically $ do --- chs' <- readTVar $ children cont --- chs <- (readTVar $ children parent) --- writeTVar (children parent)$ chs ++ chs' --- return () - - case maxThread cont of - Just sem -> signalQSemB sem -- !> "freed thread" - Nothing -> return () - return () - - - - - else proc -- !> "NO THREAD" - -forkFinally1 :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId -forkFinally1 action and_then = - mask $ \restore -> forkIO $ try (restore action) >>= and_then - -free th env= do - if isNothing $ parent env - then return Nothing -- !!> show th ++ " orphan" - else do - let msibling= fmap children $ parent env - - case msibling of - Nothing -> return Nothing - Just sibling -> do - found <- atomically $ do - sbs <- readTVar sibling - let (sbs', found) = drop [] th sbs -- !!> "search "++show th ++ " in " ++ show (map threadId sbs) - when found $ writeTVar sibling sbs' -- !> ("new list",map threadId sbs') - return found - if (not found && isJust (parent env)) - then free th $ fromJust $ parent env -- !!> "toparent" - else return $ Just env - - where - drop processed th []= (processed,False) - drop processed th (ev:evts)| th == threadId ev= (processed ++ evts, True) - | otherwise= drop (ev:processed) th evts - -hangThread parent child = when(not $ freeTh parent) $ do - let headpths= children parent - atomically $ do - ths <- readTVar headpths - writeTVar headpths $ child:ths -- !!> "thread added: "++ show (threadId child) - --- | kill all the child threads associated with the continuation context -killChildren childs = do - --- forkIO $ do - ths <- atomically $ do - ths <- readTVar childs - writeTVar childs [] - return ths --- mapM_ killChildren ths -- recursive not needed, event handlers do it - - mapM_ (killThread . threadId) ths -- !!> ("KILLEVENT " ++ show (map threadId ths) ++ --- if length ths <20 then "" --- else error "long list of threads" ) --- return () - - -type EventSetter eventdata response= (eventdata -> IO response) -> IO () -type ToReturn response= IO response - --- | deinvert an event handler. --- --- The first parameter is the setter of the event handler to be --- deinverted. Usually it is the primitive provided by a framework to set an event handler --- --- the second parameter is the value to return to the event handler. Usually it is `return()` --- --- it configures the event handler by calling the setter of the event --- handler with the current continuation -react - :: Typeable eventdata - => EventSetter eventdata response - -> ToReturn response - -> TransIO eventdata -react setHandler iob= Transient $ do - cont <- get - case event cont of - Nothing -> do - liftIO $ setHandler $ \dat ->do - runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat} - iob - was <- getData `onNothing` return NoRemote - when (was /= WasRemote) $ setData WasParallel - return Nothing - - j@(Just _) -> do - put cont{event=Nothing} - return $ unsafeCoerce j - --- Just dat -> do --- delData dat --- return (Just dat) - - --- case event cont of --- Nothing -> do --- liftIO $ loop cont ioaction --- was <- getData `onNothing` return NoRemote --- when (was /= WasRemote) $ setData WasParallel --- --- return Nothing --- j@(Just _) -> do --- put cont{event=Nothing} --- return $ unsafeCoerce j - - - --- * non-blocking keyboard input - -getLineRef= unsafePerformIO $ newTVarIO Nothing - - -roption= unsafePerformIO $ newMVar [] - --- | install a event receiver that wait for a string and trigger the continuation when this string arrives. -option :: (Typeable b, Show b, Read b, Eq b) => - b -> String -> TransIO b -option ret message= do - let sret= show ret - - liftIO $ putStrLn $ "Enter "++sret++"\tto: " ++ message - liftIO $ modifyMVar_ roption $ \msgs-> return $ sret:msgs - waitEvents $ getLine' (==ret) - liftIO $ putStrLn $ show ret ++ " chosen" - return ret - - --- | validates an input entered in the keyboard in non blocking mode. non blocking means that --- the user can enter also anything else to activate other option --- unlike `option`, wich watch continuously, input only wait for one valid response -input :: (Typeable a, Read a,Show a) => (a -> Bool) -> String -> TransIO a -input cond prompt= Transient . liftIO $do - putStr prompt >> hFlush stdout - atomically $ do - mr <- readTVar getLineRef - case mr of - Nothing -> retry - Just r -> - case reads1 r of - (s,_):_ -> if cond s -- !> show (cond s) - then do - unsafeIOToSTM $ print s - writeTVar getLineRef Nothing -- !>"match" - return $ Just s - - else return Nothing - _ -> return Nothing - --- | non blocking `getLine` with a validator -getLine' cond= do - atomically $ do - mr <- readTVar getLineRef - case mr of - Nothing -> retry - Just r -> - case reads1 r of -- !> ("received " ++ show r ++ show (unsafePerformIO myThreadId)) of - (s,_):_ -> if cond s -- !> show (cond s) - then do - writeTVar getLineRef Nothing -- !>"match" - return s - - else retry - _ -> retry - -reads1 s=x where - x= if typeOf(typeOfr x) == typeOf "" then unsafeCoerce[(s,"")] else readsPrec' 0 s - typeOfr :: [(a,String)] -> a - typeOfr = undefined - -inputLoop= do --- putStrLn "Press end to exit" - inputLoop' -- !> "started inputLoop" - where - - inputLoop'= do - r<- getLine - processLine r - inputLoop' - -processLine r= do --- when (r=="end") $ atomically $ writeTVar rexit () - let rs = breakSlash [] r - mapM_ (\ r -> -- if (r=="end") then exit' $ Left "terminated by user" else - do - threadDelay 100000 - atomically . writeTVar getLineRef $ Just r ) rs - - - where - breakSlash :: [String] -> String -> [String] - breakSlash [] ""= [""] - breakSlash s ""= s - breakSlash res ('\"':s)= - let (r,rest) = span(/= '\"') s - in breakSlash (res++[r]) $ tail1 rest - - breakSlash res s= - let (r,rest) = span(/= '/') s - in breakSlash (res++[r]) $ tail1 rest - - tail1 []=[] - tail1 x= tail x - - - - --- | wait for the execution of `exit` and return the result -stay rexit= do - mr <- takeMVar rexit - case mr of - Right Nothing -> stay rexit - Right (Just r) -> return r - Left msg -> putStrLn msg >> exitWith ExitSuccess - --- | keep the main thread running, initiate the non blocking keyboard input and execute --- the transient computation. --- --- It also read a slash-separated list of string that are read by --- `option` and `input` as if they were entered by the keyboard --- --- > foo -p options/to/be/read/by/option/and/input - -newtype Exit a= Exit a deriving Typeable - -keep :: Typeable a => TransIO a -> IO a -keep mx = do - rexit <- newEmptyMVar - forkIO $ do - liftIO $ putMVar rexit $ Right Nothing - runTransient $ do - setData $ Exit rexit - async inputLoop - <|> do mx -- ; liftIO (putMVar rexit $ Right Nothing) - -- to avoid "takeMVar blocked in a infinite loop" error - <|> do - option "end" "exit" - killChilds - exit' (Left "terminated by user" `asTypeOf` (type1 mx)) - - - return () - threadDelay 10000 - execCommandLine - stay rexit - where - type1 :: TransIO a -> Either String (Maybe a) - type1= undefined - --- | same than `keep`but do not initiate the asynchronous keyboard input. --- Useful for debugging or for creating background tasks. -keep' :: Typeable a => TransIO a -> IO a -keep' mx = do - rexit <- newEmptyMVar - forkIO $ do - runTransient $ do - setData $ Exit rexit - mx >> liftIO (putMVar rexit $ Right Nothing) - -- to avoid takeMVar in a infinite loop - return () - threadDelay 10000 - execCommandLine - - stay rexit - -execCommandLine= do - args <- getArgs - let mindex = findIndex (\o -> o == "-p" || o == "--path" ) args - when (isJust mindex) $ do - let i= fromJust mindex +1 - when (length args >= i) $ do - let path= args !! i - putStr "Executing: " >> print path - processLine path - --- | force the finalization of the main thread and thus, all the Transient block (and the application --- if there is no more code) -exit :: Typeable a => a -> TransIO a -exit x= do - Exit rexit <- getSData <|> error "exit: not the type expected" `asTypeOf` type1 x - liftIO $ putMVar rexit . Right $ Just x - stop - where - type1 :: a -> TransIO (Exit (MVar (Either String (Maybe a)))) - type1= undefined - -exit' x= do - Exit rexit <- getSData <|> error "exit: not type expected" - liftIO $ putMVar rexit x ; stop - - --- | alternative operator for maybe values. Used in infix mode -onNothing :: Monad m => m (Maybe b) -> m b -> m b -onNothing iox iox'= do - mx <- iox - case mx of - Just x -> return x - Nothing -> iox' +{-# LANGUAGE ScopedTypeVariables #-} +----------------------------------------------------------------------------- +-- +-- Module : Base +-- Copyright : +-- License : GPL (Just (Version {versionBranch = [3], versionTags = []})) +-- +-- Maintainer : agocorona@gmail.com +-- Stability : +-- Portability : +-- +-- | See http://github.com/agocorona/transient +-- everithing in this module is exported in order to allow extensibility. +----------------------------------------------------------------------------- +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE Rank2Types #-} +-- show +module Transient.Internals where +-- /show + +import Control.Applicative +import Control.Monad.State +import Data.Dynamic +import qualified Data.Map as M +import Data.Monoid +import Debug.Trace +import System.IO.Unsafe +import Unsafe.Coerce +import Control.Exception +import Control.Concurrent +import Control.Concurrent.STM +import System.Mem.StableName +import Data.Maybe +import GHC.Conc +import Data.List +import Data.IORef +import System.Environment +import System.IO (hFlush,stdout) +import System.Exit +{-# INLINE (!>) #-} +(!>) :: Show a => b -> a -> b +(!>) x y= trace (show y) x +infixr 0 !> + + +data TransIO x = Transient {runTrans :: StateT EventF IO (Maybe x)} +type SData= () + +type EventId= Int + +type TransientIO= TransIO + +data EventF = forall a b . EventF{meffects :: Effects + ,event :: Maybe SData + ,xcomp :: TransIO a + ,fcomp :: [b -> TransIO b] + ,mfData :: M.Map TypeRep SData + ,mfSequence :: Int + ,threadId :: ThreadId + ,freeTh :: Bool + ,parent :: Maybe EventF + ,children :: TVar[EventF] + ,maxThread :: Maybe (IORef Int) + } + 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 ()) + state f = Transient $ do + s <- get + let ~(a, s') = f s + put s' + return $ Just a + +type StateIO= StateT EventF IO + + +-- | run the transient computation with a blank state +runTransient :: TransIO x -> IO (Maybe x, EventF) +runTransient t= do + th <- myThreadId + let eventf0= EventF baseEffects Nothing empty [] M.empty 0 + th False Nothing (unsafePerformIO $ newTVarIO []) Nothing + + + runStateT (runTrans t) eventf0 + +-- | run the transient computation with an state +runTransState st x = runStateT (runTrans x) st + +-- | get the continuation context: closure, continuation, state, child threads etc +getCont :: TransIO EventF +getCont = Transient $ Just <$> get + +-- | run the closure and the continuation using the state data of the calling thread +runCont :: EventF -> StateIO (Maybe a) +runCont (EventF _ _ x fs _ _ _ _ _ _ _)= runTrans $ do + r <- unsafeCoerce x + compose fs r + +-- | run the closure and the continuation using his own state data +runCont' cont= runStateT (runCont cont) cont + +-- | warning: radiactive untyped stuff. handle with care +getContinuations :: StateIO [a -> TransIO b] +getContinuations= do + EventF _ _ _ fs _ _ _ _ _ _ _ <- get + return $ unsafeCoerce fs + +{- +runCont cont= do + mr <- runClosure cont + case mr of + Nothing -> return Nothing + Just r -> runContinuation cont r +-} + + +-- | compose a list of continuations +compose []= const empty +compose (f: fs)= \x -> f x >>= compose fs + + + +-- | run the closure (the 'x' in 'x >>= f') of the current bind operation. +runClosure :: EventF -> StateIO (Maybe a) +runClosure (EventF _ _ x _ _ _ _ _ _ _ _) = unsafeCoerce $ runTrans x + + +-- | run the continuation (the 'f' in 'x >>= f') of the current bind operation +runContinuation :: EventF -> a -> StateIO (Maybe b) +runContinuation (EventF _ _ _ fs _ _ _ _ _ _ _) = + runTrans . (unsafeCoerce $ compose $ fs) + + +setContinuation :: TransIO a -> (a -> TransIO b) -> [c -> TransIO c] -> StateIO () +setContinuation b c fs = do + (EventF eff ev _ _ d e f g h i j) <- get + put $ EventF eff ev b ( unsafeCoerce c: fs) d e f g h i j + +withContinuation c mx= do + EventF eff ev f1 fs d e f g h i j <- get + put $ EventF eff ev mx ( unsafeCoerce c: fs) d e f g h i j + r <- mx + restoreStack fs + return r + +-- | run a chain of continuations. It is up to the programmer to assure by construction that +-- each continuation type-check with the next, that the parameter type match the input of the first +-- continuation. +-- Normally this makes sense if it stop the current flow with `stop` after the invocation +runContinuations :: [a -> TransIO b] -> c -> TransIO d +runContinuations fs x= (compose $ unsafeCoerce fs) x + +instance Functor TransIO where + fmap f mx= -- Transient $ fmap (fmap f) $ runTrans 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,[]) -- !> "NEWIOREF" + + fs <- getContinuations + + let + + hasWait (_:Wait:_)= True + hasWait _ = False + + appf k = Transient $ do + Log rec _ full <- getData `onNothing` return (Log False [] []) + (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 [] []) + 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" + was <- getData `onNothing` return NoRemote + when (was == WasParallel) $ setData NoRemote + + Log recovery _ full <- getData `onNothing` return (Log False [] []) + + + + 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 [] []) + 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 + +restoreStack fs= + modify $ \(EventF eff _ f _ a b c d parent children g1) -> + EventF eff Nothing f fs a b c d parent children g1 + +readWithErr line= + let [(v,left)] = readsPrec 0 line + in (v `seq` return [(v,left)]) + `catch` (\(e::SomeException) -> + error ("read error of " ++ show( typeOf v) ++ " in: "++ "\""++line++"\"")) + + +readsPrec' _= unsafePerformIO . readWithErr + + + +-- | dynamic serializable data for logging +data IDynamic= IDyns String | forall a.(Read a, Show a,Typeable a) => IDynamic a + +instance Show IDynamic where + show (IDynamic x)= show $ show x + show (IDyns s)= show s + +instance Read IDynamic where + readsPrec n str= map (\(x,s) -> (IDyns x,s)) $ readsPrec' n str + + +type Recover= Bool +type CurrentPointer= [LogElem] +type LogEntries= [LogElem] +data LogElem= Wait | Exec | Var IDynamic deriving (Read,Show) +data Log= Log Recover CurrentPointer LogEntries deriving Typeable + + +instance Alternative TransIO where + empty = Transient $ return Nothing + (<|>) = mplus + + +data RemoteStatus= WasRemote | WasParallel | NoRemote deriving (Typeable, Eq, Show) + +instance MonadPlus TransIO where + mzero= empty + mplus x y= Transient $ do + mx <- runTrans x -- !!> "RUNTRANS11111" + was <- getData `onNothing` return NoRemote + if was== WasRemote -- !> was + then return Nothing + else + case mx of + Nothing -> runTrans y -- !!> "RUNTRANS22222" + justx -> return justx + +-- | a sinonym of empty that can be used in a monadic expression. it stop the +-- computation and execute the next alternative computation (composed with `<|>`) +stop :: Alternative m => m stopped +stop= empty + +class AdditionalOperators m where + + -- | executes the second operand even if the frist return empty. + -- A normal imperative (monadic) sequence uses the operator (>>) which in the + -- Transient monad does not execute the next operand if the previous one return empty. + (**>) :: m a -> m b -> m b + + -- | forces the execution of the second operand even if the first stop. It does not execute + -- the second operand as result of internal events occuring in the first operand. + -- Return the first result + (<**) :: m a -> m b -> m a + + atEnd' ::m a -> m b -> m a + atEnd' = (<**) + + -- | forces the execution of the second operand even if the first stop. Return the first result. The second + -- operand is executed also when internal events happens in the first operand and it returns something + (<***) :: m a -> m b -> m a + + atEnd :: m a -> m b -> m a + atEnd= (<***) + + +instance AdditionalOperators TransIO where + +-- (**>) :: TransIO a -> TransIO b -> TransIO b + (**>) x y= Transient $ do + runTrans x + runTrans y + +-- (<***) :: TransIO a -> TransIO b -> TransIO a + (<***) ma mb= Transient $ do + fs <- getContinuations + setContinuation ma (\x -> mb >> return x) fs + a <- runTrans ma + runTrans mb + restoreStack fs + return a + +-- (<**) :: TransIO a -> TransIO b -> TransIO a + (<**) ma mb= Transient $ do + a <- runTrans ma -- !> "ma" + runTrans mb -- !> "mb" + return a + +infixr 1 <*** , <**, **> + + + +-- | when the first operand is an asynchronous operation, the second operand is executed once (one single time) +-- when the first completes his first asyncronous operation. +-- +-- This is useful for spawning asynchronous or distributed tasks that are singletons and that should start +-- when the first one is set up. +-- +-- for example a streaming where the event receivers are acivated before the senders. + +(<|) :: TransIO a -> TransIO b -> TransIO a +(<|) ma mb = Transient $ do + fs <- getContinuations + ref <- liftIO $ newIORef False + setContinuation ma (cont ref ) fs + r <- runTrans ma + restoreStack fs + return r + where + cont ref x= Transient $ do + n <- liftIO $ readIORef ref + if n == True + then return $ Just x + else do liftIO $ writeIORef ref True + runTrans mb + return $ Just x + +instance Monoid a => Monoid (TransIO a) where + mappend x y = mappend <$> x <*> y + mempty= return mempty + +-- | set the current closure and continuation for the current statement +setEventCont :: TransIO a -> (a -> TransIO b) -> StateIO EventF +setEventCont x f = do + + st@(EventF eff e _ fs d n r applic ch rc bs) <- get -- !> "SET" + let cont= EventF eff e x ( unsafeCoerce f : fs) d n r applic ch rc bs + put cont + return cont + +-- | 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 + st@(EventF eff e _ fs d n r nr ch rc bs) <- get -- !> "reset" + let f= \mx -> case mx of + Nothing -> empty + Just x -> (unsafeCoerce $ head fs) x + put $ EventF eff e (f mx) ( tailsafe fs) d n r nr ch rc bs + return id + +tailsafe []=[] +tailsafe (x:xs)= xs + +--refEventCont= unsafePerformIO $ newIORef baseEffects + +{-# INLINE baseEffects #-} +baseEffects :: Effects + +baseEffects x x' f' = do + c <-setEventCont x' f' + mk <- runTrans x + t <- resetEventCont mk c + return (t,mk) + +instance Monad TransIO where + + return = pure + + x >>= f = Transient $ do +-- effects <- gets effects -- liftIO $ readIORef refEventCont + (t,mk) <- baseEffects x x f + t $ case mk of + Just k -> runTrans (f k) + + Nothing -> return Nothing + +--instance MonadTrans (Transient ) where +-- lift mx = Transient $ mx >>= return . Just + +instance MonadIO TransIO where + liftIO x = Transient $ liftIO x >>= return . Just -- let x= liftIO io in x `seq` lift x + + +-- * Threads + +waitQSemB sem= atomicModifyIORef sem $ \n -> if n > 0 then(n-1,True) else (n,False) +signalQSemB sem= atomicModifyIORef sem $ \n -> (n + 1,()) + +-- | set the maximun number of threads for a procedure. It is useful to limit the +-- parallelization of transient code that uses `parallel` `spawn` and `waitEvents` +threads :: Int -> TransIO a -> TransIO a +threads n proc= Transient $ do + msem <- gets maxThread + sem <- liftIO $ newIORef n + modify $ \s -> s{maxThread= Just sem} + r <- runTrans proc + modify $ \s -> s{maxThread = msem} -- restore it + return r + +-- | delete all the previous childs generated by the expression taken as parameter and continue execution +-- of the current thread. +oneThread :: TransIO a -> TransientIO a +oneThread comp= do + chs <- liftIO $ newTVarIO [] + r <- comp + modify $ \ s -> s{children= chs} + killChilds + return r + + +showThreads :: TransIO empty +showThreads= do + st' <- gets (fromJust . parent) + liftIO $ showTree 0 st' + stop + where + toplevel st = + case parent st of + Nothing -> st + Just p -> toplevel p + + showThreads' n rchs= do + chs <- atomically $ readTVar rchs + mapM_ (showTree n) chs + + showTree n ch= do + putStr $ take n $ repeat ' ' + print $ threadId ch + showThreads' (n+4) $ children ch + + +-- | add n threads to the limit of threads. If there is no limit, it set it +addThreads' :: Int -> TransIO () +addThreads' n= Transient $ do + msem <- gets maxThread + case msem of + Just sem -> liftIO $ modifyIORef sem $ \n' -> n + n' + Nothing -> do + sem <- liftIO (newIORef n) + modify $ \ s -> s{maxThread= Just sem} + return $ Just () + +-- | assure that at least there are n threads available +addThreads n= Transient $ do + msem <- gets maxThread + case msem of + Nothing -> return () + Just sem -> liftIO $ modifyIORef sem $ \n' -> if n' > n then n' else n + return $ Just () +--getNonUsedThreads :: TransIO (Maybe Int) +--getNonUsedThreads= Transient $ do +-- msem <- gets maxThread +-- case msem of +-- Just sem -> liftIO $ Just <$> readIORef sem +-- Nothing -> return Nothing + + +-- | The threads generated in the process passed as parameter will not be killed by `kill*` primitives +freeThreads :: TransIO a -> TransIO a +freeThreads proc= Transient $ do + st <- get + put st{freeTh= True} + r <- runTrans proc + modify $ \s -> s{freeTh= freeTh st} + return r + +-- | The threads will be killed when the parent thread dies. That is the default. +-- This can be invoked to revert the effect of `freeThreads` +hookedThreads :: TransIO a -> TransIO a +hookedThreads proc= Transient $ do + st <- get + put st{freeTh= False} + r <- runTrans proc + modify $ \st -> st{freeTh= freeTh st} + return r + +-- | kill all the child threads of the current thread +killChilds :: TransientIO() +killChilds= Transient $ do + cont <- get + liftIO $ killChildren $ children cont + return $ Just () + +-- * extensible state: session data management + +-- | Get the state data for the desired type if there is any. +getData :: (MonadState EventF m,Typeable a) => m (Maybe a) +getData = resp where + resp= gets mfData >>= \list -> + case M.lookup ( typeOf $ typeResp resp ) list of + Just x -> return . Just $ unsafeCoerce x + Nothing -> return Nothing + typeResp :: m (Maybe x) -> x + typeResp= undefined + + +-- | getData specialized for the Transient monad. if Nothing, the +-- monadic computation does not continue. +-- +-- If there is no such data, `getSData` silently stop the computation. +-- That may or may not be the desired behaviour. +-- To make sure that this does not get unnoticed, use this construction: +-- +-- > getSData <|> error "no data" +-- +-- To have the same semantics and guarantees than `get`, use a default value: +-- +-- > getInt= getSData <|> return (0 :: Int) +-- +-- The default value (0 in this case) has the same role than the initial value in a state monad. +-- The difference is that you can define as many `get` as you need for all your data types. +-- +-- To distingish two data with the same types, use newtype definitions. +getSData :: Typeable a => TransIO a +getSData= Transient getData + + + +-- | set session data for this type. retrieved with getData or getSData +-- Note that this is data in a state monad, that means that the update only affect downstream +-- in the monad execution. it is not a global state neither a per user or per thread state +-- it is a monadic state like the one of a state monad. +setData :: (MonadState EventF m, Typeable a) => a -> m () +setData x= + let t= typeOf x in modify $ \st -> st{mfData= M.insert t (unsafeCoerce x) (mfData st)} + + +delData :: ( MonadState EventF m,Typeable a) => a -> m () +delData x= modify $ \st -> st{mfData= M.delete (typeOf x ) (mfData st)} + + +--withSData :: ( MonadState EventF m,Typeable a) => (Maybe a -> a) -> m () +--withSData f= modify $ \st -> st{mfData= +-- let dat = mfData st +-- mx= M.lookup typeofx dat +-- mx'= case mx of Nothing -> Nothing; Just x -> unsafeCoerce x +-- fx= f mx' +-- typeofx= typeOf $ typeoff f +-- in M.insert typeofx (unsafeCoerce fx) dat} +-- where +-- typeoff :: (Maybe a -> a) -> a +-- typeoff = undefined +---- + +-- | generator of identifiers that are unique withing the current monadic sequence +-- They are not unique in the whole program. +genId :: MonadState EventF m => m Int +genId= do + st <- get + let n= mfSequence st + put st{mfSequence= n+1} + return n + +getPrevId :: MonadState EventF m => m Int +getPrevId= do + n <- gets mfSequence + return n + +instance Read SomeException where + readsPrec n str= + let [(s , r)]= read str in [(SomeException $ ErrorCall s,r)] + +-- | async calls + +data StreamData a= SMore a | SLast a | SDone | SError SomeException deriving (Typeable, Show,Read) + + +-- | variant of `parallel` that repeatedly executes the IO computation and kill the previously created childs +-- +-- It is useful in single threaded problems where each event discard the computations spawned by +-- previous events +waitEvents :: IO b -> TransIO b +waitEvents io= do + mr <- parallel (SMore <$> io) + case mr of + SMore x -> return x + SError e -> throw e + + +-- Multithreaded version of `waitEvents` that do not kill the computations spawned by previous events +waitEvents' :: IO b -> TransIO b +waitEvents' io= do + mr <- parallel (SMore <$> io) + case mr of + SMore x -> return x + SError e -> throw e + +-- | variant of `parallel` that execute the IO computation once, and kill the previous child threads +async :: IO b -> TransIO b +async io= do + mr <- parallel (SLast <$> io) + case mr of + SLast x -> return x + SError e -> throw e + +-- | variant of waitEvents that spawn free threads. It is a little faster at the cost of no thread control +spawn :: IO b -> TransIO b +spawn io= freeThreads $ do + mr <- parallel (SMore <$>io) + case mr of + SMore x -> return x + SError e -> throw e + + +-- | executes an IO action each certain interval of time and return his value if it changes +sample :: Eq a => IO a -> Int -> TransIO a +sample action interval= do + v <- liftIO action + prev <- liftIO $ newIORef v + waitEvents (loop action prev) <|> async (return v) + where + loop action prev= loop' + where + loop'= do + threadDelay interval + v <- action + v' <- readIORef prev + if v /= v' then writeIORef prev v >> return v else loop' + + + + +-- | return empty to the current thread, in new thread, execute the IO action, +-- this IO action modify an internal buffer. then, executes the closure where `parallel` is located +-- In this new execution, since the buffer is filled, `parallel` return the content of this buffer. +-- Then it launch the continuation after it with this new value returned by the closure. +-- +-- If the maximum number of threads, set with `threads` has been reached `parallel` perform +-- the work sequentially, in the current thread. +-- So `parallel` means that 'it can be parallelized if there are thread available' +-- +-- if there is a limitation of threads, when a thread finish, the counter of threads available +-- is increased so another `parallel` can make use of it. +-- +-- The behaviour of `parallel` depend on `StreamData`; If `SMore`, `parallel` will excute again the +-- IO action. with `SLast`, `SDone` and `SError`, `parallel` will not repeat the IO action anymore. +parallel :: IO (StreamData b) -> TransIO (StreamData b) +parallel 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 + was <- getData `onNothing` return NoRemote + when (was /= WasRemote) $ setData WasParallel + return Nothing + + +-- executes the IO action and then the continuation included in the first parameter +loop :: EventF -> IO (StreamData t) -> IO () +loop (cont'@(EventF eff e x fs a b c d _ childs g)) rec = do + chs <- liftIO $ newTVarIO [] + let cont = EventF eff e x fs a b c d (Just cont') chs g + iocont dat= do + runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat} + return () + + -- execute the IO computation and then the closure-continuation + loop'= forkMaybe False cont $ do + mdat <- threadDelay 0 >> rec `catch` \(e :: SomeException) -> return $ SError e + case mdat of + se@(SError _) -> iocont se + SDone -> iocont SDone + last@(SLast _) -> iocont last + + more@(SMore _) -> do + forkMaybe False cont $ iocont more + loop' + loop' + return () + where + forkMaybe True cont proc = forkMaybe' True cont proc + forkMaybe False cont proc = do + dofork <- case maxThread cont of + Nothing -> return True + Just sem -> do + dofork <- waitQSemB sem + if dofork then return True else return False + forkMaybe' dofork cont proc + + forkMaybe' dofork cont proc= + if dofork + then do + forkFinally1 (do + th <- myThreadId + hangThread cont' cont{threadId=th} -- !> "thread created: "++ show th + proc) + $ \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 _ -> when(not $ freeTh cont') $ do -- if was not a free thread + -- if parent is alive + -- then remove himself from the parent list (with free) + -- and pass his active children to his parent + + th <- myThreadId + mparent <- free th cont + return () + -- pass the active children to the parent +-- case mparent of +-- Nothing -> return() +-- Just parent -> atomically $ do +-- chs' <- readTVar $ children cont +-- chs <- (readTVar $ children parent) +-- writeTVar (children parent)$ chs ++ chs' +-- return () + + case maxThread cont of + Just sem -> signalQSemB sem -- !> "freed thread" + Nothing -> return () + return () + + + + + else proc -- !> "NO THREAD" + +forkFinally1 :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId +forkFinally1 action and_then = + mask $ \restore -> forkIO $ try (restore action) >>= and_then + +free th env= do + if isNothing $ parent env + then return Nothing -- !!> show th ++ " orphan" + else do + let msibling= fmap children $ parent env + + case msibling of + Nothing -> return Nothing + Just sibling -> do + found <- atomically $ do + sbs <- readTVar sibling + let (sbs', found) = drop [] th sbs -- !!> "search "++show th ++ " in " ++ show (map threadId sbs) + when found $ writeTVar sibling sbs' -- !> ("new list",map threadId sbs') + return found + if (not found && isJust (parent env)) + then free th $ fromJust $ parent env -- !!> "toparent" + else return $ Just env + + where + drop processed th []= (processed,False) + drop processed th (ev:evts)| th == threadId ev= (processed ++ evts, True) + | otherwise= drop (ev:processed) th evts + +hangThread parent child = when(not $ freeTh parent) $ do + let headpths= children parent + atomically $ do + ths <- readTVar headpths + writeTVar headpths $ child:ths -- !!> "thread added: "++ show (threadId child) + +-- | kill all the child threads associated with the continuation context +killChildren childs = do + +-- forkIO $ do + ths <- atomically $ do + ths <- readTVar childs + writeTVar childs [] + return ths +-- mapM_ killChildren ths -- recursive not needed, event handlers do it + + mapM_ (killThread . threadId) ths -- !!> ("KILLEVENT " ++ show (map threadId ths) ++ +-- if length ths <20 then "" +-- else error "long list of threads" ) +-- return () + + +type EventSetter eventdata response= (eventdata -> IO response) -> IO () +type ToReturn response= IO response + +-- | deinvert an event handler. +-- +-- The first parameter is the setter of the event handler to be +-- deinverted. Usually it is the primitive provided by a framework to set an event handler +-- +-- the second parameter is the value to return to the event handler. Usually it is `return()` +-- +-- it configures the event handler by calling the setter of the event +-- handler with the current continuation +react + :: Typeable eventdata + => EventSetter eventdata response + -> ToReturn response + -> TransIO eventdata +react setHandler iob= Transient $ do + cont <- get + case event cont of + Nothing -> do + liftIO $ setHandler $ \dat ->do + runStateT (runCont cont) cont{event= Just $ unsafeCoerce dat} + iob + was <- getData `onNothing` return NoRemote + when (was /= WasRemote) $ setData WasParallel + return Nothing + + j@(Just _) -> do + put cont{event=Nothing} + return $ unsafeCoerce j + +-- Just dat -> do +-- delData dat +-- return (Just dat) + + +-- case event cont of +-- Nothing -> do +-- liftIO $ loop cont ioaction +-- was <- getData `onNothing` return NoRemote +-- when (was /= WasRemote) $ setData WasParallel +-- +-- return Nothing +-- j@(Just _) -> do +-- put cont{event=Nothing} +-- return $ unsafeCoerce j + + + +-- * non-blocking keyboard input + +getLineRef= unsafePerformIO $ newTVarIO Nothing + + +roption= unsafePerformIO $ newMVar [] + +-- | install a event receiver that wait for a string and trigger the continuation when this string arrives. +option :: (Typeable b, Show b, Read b, Eq b) => + b -> String -> TransIO b +option ret message= do + let sret= show ret + + liftIO $ putStrLn $ "Enter "++sret++"\tto: " ++ message + liftIO $ modifyMVar_ roption $ \msgs-> return $ sret:msgs + waitEvents $ getLine' (==ret) + liftIO $ putStrLn $ show ret ++ " chosen" + return ret + + +-- | validates an input entered in the keyboard in non blocking mode. non blocking means that +-- the user can enter also anything else to activate other option +-- unlike `option`, wich watch continuously, input only wait for one valid response +input :: (Typeable a, Read a,Show a) => (a -> Bool) -> String -> TransIO a +input cond prompt= Transient . liftIO $do + putStr prompt >> hFlush stdout + atomically $ do + mr <- readTVar getLineRef + case mr of + Nothing -> retry + Just r -> + case reads1 r of + (s,_):_ -> if cond s -- !> show (cond s) + then do + unsafeIOToSTM $ print s + writeTVar getLineRef Nothing -- !>"match" + return $ Just s + + else return Nothing + _ -> return Nothing + +-- | non blocking `getLine` with a validator +getLine' cond= do + atomically $ do + mr <- readTVar getLineRef + case mr of + Nothing -> retry + Just r -> + case reads1 r of -- !> ("received " ++ show r ++ show (unsafePerformIO myThreadId)) of + (s,_):_ -> if cond s -- !> show (cond s) + then do + writeTVar getLineRef Nothing -- !>"match" + return s + + else retry + _ -> retry + +reads1 s=x where + x= if typeOf(typeOfr x) == typeOf "" then unsafeCoerce[(s,"")] else readsPrec' 0 s + typeOfr :: [(a,String)] -> a + typeOfr = undefined + +inputLoop= do +-- putStrLn "Press end to exit" + inputLoop' -- !> "started inputLoop" + where + + inputLoop'= do + r<- getLine + processLine r + inputLoop' + +processLine r= do +-- when (r=="end") $ atomically $ writeTVar rexit () + let rs = breakSlash [] r + mapM_ (\ r -> -- if (r=="end") then exit' $ Left "terminated by user" else + do + threadDelay 100000 + atomically . writeTVar getLineRef $ Just r ) rs + + + where + breakSlash :: [String] -> String -> [String] + breakSlash [] ""= [""] + breakSlash s ""= s + breakSlash res ('\"':s)= + let (r,rest) = span(/= '\"') s + in breakSlash (res++[r]) $ tail1 rest + + breakSlash res s= + let (r,rest) = span(/= '/') s + in breakSlash (res++[r]) $ tail1 rest + + tail1 []=[] + tail1 x= tail x + + + + +-- | wait for the execution of `exit` and return the result +stay rexit= do + mr <- takeMVar rexit + case mr of + Right Nothing -> stay rexit + Right (Just r) -> return r + Left msg -> putStrLn msg >> exitWith ExitSuccess + +-- | keep the main thread running, initiate the non blocking keyboard input and execute +-- the transient computation. +-- +-- It also read a slash-separated list of string that are read by +-- `option` and `input` as if they were entered by the keyboard +-- +-- > foo -p options/to/be/read/by/option/and/input + +newtype Exit a= Exit a deriving Typeable + +keep :: Typeable a => TransIO a -> IO a +keep mx = do + rexit <- newEmptyMVar + forkIO $ do + liftIO $ putMVar rexit $ Right Nothing + runTransient $ do + setData $ Exit rexit + async inputLoop + <|> do mx -- ; liftIO (putMVar rexit $ Right Nothing) + -- to avoid "takeMVar blocked in a infinite loop" error + <|> do + option "end" "exit" + killChilds + exit' (Left "terminated by user" `asTypeOf` (type1 mx)) + + + return () + threadDelay 10000 + execCommandLine + stay rexit + where + type1 :: TransIO a -> Either String (Maybe a) + type1= undefined + +-- | same than `keep`but do not initiate the asynchronous keyboard input. +-- Useful for debugging or for creating background tasks. +keep' :: Typeable a => TransIO a -> IO a +keep' mx = do + rexit <- newEmptyMVar + forkIO $ do + runTransient $ do + setData $ Exit rexit + mx >> liftIO (putMVar rexit $ Right Nothing) + -- to avoid takeMVar in a infinite loop + return () + threadDelay 10000 + execCommandLine + + stay rexit + +execCommandLine= do + args <- getArgs + let mindex = findIndex (\o -> o == "-p" || o == "--path" ) args + when (isJust mindex) $ do + let i= fromJust mindex +1 + when (length args >= i) $ do + let path= args !! i + putStr "Executing: " >> print path + processLine path + +-- | force the finalization of the main thread and thus, all the Transient block (and the application +-- if there is no more code) +exit :: Typeable a => a -> TransIO a +exit x= do + Exit rexit <- getSData <|> error "exit: not the type expected" `asTypeOf` type1 x + liftIO $ putMVar rexit . Right $ Just x + stop + where + type1 :: a -> TransIO (Exit (MVar (Either String (Maybe a)))) + type1= undefined + +exit' x= do + Exit rexit <- getSData <|> error "exit: not type expected" + liftIO $ putMVar rexit x ; stop + + +-- | alternative operator for maybe values. Used in infix mode +onNothing :: Monad m => m (Maybe b) -> m b -> m b +onNothing iox iox'= do + mx <- iox + case mx of + Just x -> return x + Nothing -> iox' diff --git a/src/Transient/Logged.hs b/src/Transient/Logged.hs --- a/src/Transient/Logged.hs +++ b/src/Transient/Logged.hs @@ -1,179 +1,179 @@ ------------------------------------------------------------------------------ --- --- Module : Transient.Logged --- Copyright : --- License : GPL-3 --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | --- ------------------------------------------------------------------------------ -{-# LANGUAGE ExistentialQuantification, FlexibleInstances, ScopedTypeVariables, UndecidableInstances #-} -module Transient.Logged(restore,checkpoint,suspend,logged,Loggable) where - -import Data.Typeable -import Unsafe.Coerce -import Transient.Base -import Transient.Indeterminism(choose) -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 System.Random - - -class (Show a, Read a,Typeable a) => Loggable a -instance (Show a, Read a,Typeable a) => Loggable a - -logs= "logs/" - --- re-excutes all the threads whose state has been logged in the "./logs" folder --- .Each log is removed when it is executed. --- --- example: this program, if executed three times will first print hello some times --- but `suspend` will kill the threads and exit it. - --- The second time, it will print "world" and "world22222" and will stay. --- --- The third time that it is executed, it only present "world22222" messages --- --- > main= keep $ restore $ do --- > r <- logged $ choose [1..10 :: Int] --- > logged $ liftIO $ print ("hello",r) --- > suspend () --- > logged $ liftIO $ print ("world",r) --- > checkpoint --- > logged $ liftIO $ print ("world22222",r) - -restore :: TransIO a -> TransIO a -restore proc= do - liftIO $ createDirectory logs `catch` (\(e :: SomeException) -> return ()) - list <- liftIO $ getDirectoryContents logs - `catch` (\(e::SomeException) -> return []) - if length list== 2 then proc else do - - let list'= filter ((/=) '.' . head) list - file <- choose list' -- !> list' - - logstr <- liftIO $ readFile (logs++file) - let log= length logstr `seq` read' logstr - - log `seq` setData (Log True (reverse log) log) - liftIO $ remove $ logs ++ file - proc - where - read'= fst . head . reads1 - - remove f= removeFile f `catch` (\(e::SomeException) -> remove f) - - - --- | save the state of the thread that execute it and exit the transient block initiated with `keep` or similar --- . `keep` will return the value passed by `suspend`. --- If the process is executed again with `restore` it will reexecute the thread from this point on. --- --- it is useful to insert it in `finish` blocks to gather error information, -suspend :: Typeable a => a -> TransIO a -suspend x= do - Log recovery _ log <- getData `onNothing` return (Log False [] []) - if recovery then return x else do - logAll log - exit x - --- | Save the state of every thread at this point. If the process is re-executed with `restore` it will reexecute the thread from this point on.. -checkpoint :: TransIO () -checkpoint = do - Log recovery _ log <- getData `onNothing` return (Log False [] []) - if recovery then return () else logAll log - - -logAll log= do - newlogfile <- liftIO $ (logs ++) <$> replicateM 7 (randomRIO ('a','z')) - liftIO $ writeFile newlogfile $ show log - :: TransIO () - - - - - -fromIDyn :: (Read a, Show a, Typeable 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) - -toIDyn x= IDynamic x - -{- TODO add save/recover from log -rerun :: Log -> TransIO a -> TransIO a - -getLog :: TransIO Log --} - --- | write the result of the computation in the log and return it. --- but if there is data in the internal log, it read the data from the log and --- do not execute the computation. --- --- It accept nested step's. The effect is that if the outer step is executed completely --- the log of the inner steps are erased. If it is not the case, the inner steps are logged --- this reduce the log of large computations to the minimum. That is a feature not present --- in the package Workflow. --- --- > r <- logged $ do --- > logged this :: TransIO () --- > logged that :: TransIO () --- > logged thatOther --- > liftIO $ print r --- --- when `print` is executed, the log is just the value of r. --- --- but at the `thatOther` execution the log is: [Exec,(), ()] --- -logged :: Loggable a => TransientIO a -> TransientIO a -logged mx = Transient $ do - Log recover rs full <- getData `onNothing` return ( Log False [][]) - runTrans $ - case (recover ,rs) of -- !> ("logged enter",recover,rs) of - (True, Var x: rs') -> do - setData $ Log True rs' full - return $ fromIDyn x --- !> ("read in Var:", x) - - (True, Exec:rs') -> do - setData $ Log True rs' full - mx -- !> "Exec" - - (True, Wait:rs') -> do - setData (Log True rs' full) -- !> "Wait" - empty - - _ -> do --- let add= Exec: full - setData $ Log False (Exec : rs) (Exec: full) -- !> ("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 -> --- let add= Wait: full - setData $ Log False (Wait: rs) (Wait: full) - _ -> return ()) - - Log recoverAfter lognew _ <- getData `onNothing` return ( Log False [][]) - let add= Var (toIDyn r): full - if recoverAfter && (not $ null lognew) -- !> ("recoverAfter", recoverAfter) - then (setData $ Log True lognew (reverse lognew ++ add) ) - -- !> ("recover",reverse lognew ,add) - else if recoverAfter && (null lognew) then - setData $ Log False [] add - else - (setData $ Log False (Var (toIDyn r):rs) add) -- !> ("restore", (Var (toIDyn r):rs)) - return r - - - +----------------------------------------------------------------------------- +-- +-- Module : Transient.Logged +-- Copyright : +-- License : GPL-3 +-- +-- Maintainer : agocorona@gmail.com +-- Stability : +-- Portability : +-- +-- | +-- +----------------------------------------------------------------------------- +{-# LANGUAGE ExistentialQuantification, FlexibleInstances, ScopedTypeVariables, UndecidableInstances #-} +module Transient.Logged(restore,checkpoint,suspend,logged,Loggable) where + +import Data.Typeable +import Unsafe.Coerce +import Transient.Base +import Transient.Indeterminism(choose) +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 System.Random + + +class (Show a, Read a,Typeable a) => Loggable a +instance (Show a, Read a,Typeable a) => Loggable a + +logs= "logs/" + +-- re-excutes all the threads whose state has been logged in the "./logs" folder +-- .Each log is removed when it is executed. +-- +-- example: this program, if executed three times will first print hello some times +-- but `suspend` will kill the threads and exit it. + +-- The second time, it will print "world" and "world22222" and will stay. +-- +-- The third time that it is executed, it only present "world22222" messages +-- +-- > main= keep $ restore $ do +-- > r <- logged $ choose [1..10 :: Int] +-- > logged $ liftIO $ print ("hello",r) +-- > suspend () +-- > logged $ liftIO $ print ("world",r) +-- > checkpoint +-- > logged $ liftIO $ print ("world22222",r) + +restore :: TransIO a -> TransIO a +restore proc= do + liftIO $ createDirectory logs `catch` (\(e :: SomeException) -> return ()) + list <- liftIO $ getDirectoryContents logs + `catch` (\(e::SomeException) -> return []) + if length list== 2 then proc else do + + let list'= filter ((/=) '.' . head) list + file <- choose list' -- !> list' + + logstr <- liftIO $ readFile (logs++file) + let log= length logstr `seq` read' logstr + + log `seq` setData (Log True (reverse log) log) + liftIO $ remove $ logs ++ file + proc + where + read'= fst . head . reads1 + + remove f= removeFile f `catch` (\(e::SomeException) -> remove f) + + + +-- | save the state of the thread that execute it and exit the transient block initiated with `keep` or similar +-- . `keep` will return the value passed by `suspend`. +-- If the process is executed again with `restore` it will reexecute the thread from this point on. +-- +-- it is useful to insert it in `finish` blocks to gather error information, +suspend :: Typeable a => a -> TransIO a +suspend x= do + Log recovery _ log <- getData `onNothing` return (Log False [] []) + if recovery then return x else do + logAll log + exit x + +-- | Save the state of every thread at this point. If the process is re-executed with `restore` it will reexecute the thread from this point on.. +checkpoint :: TransIO () +checkpoint = do + Log recovery _ log <- getData `onNothing` return (Log False [] []) + if recovery then return () else logAll log + + +logAll log= do + newlogfile <- liftIO $ (logs ++) <$> replicateM 7 (randomRIO ('a','z')) + liftIO $ writeFile newlogfile $ show log + :: TransIO () + + + + + +fromIDyn :: (Read a, Show a, Typeable 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) + +toIDyn x= IDynamic x + +{- TODO add save/recover from log +rerun :: Log -> TransIO a -> TransIO a + +getLog :: TransIO Log +-} + +-- | write the result of the computation in the log and return it. +-- but if there is data in the internal log, it read the data from the log and +-- do not execute the computation. +-- +-- It accept nested step's. The effect is that if the outer step is executed completely +-- the log of the inner steps are erased. If it is not the case, the inner steps are logged +-- this reduce the log of large computations to the minimum. That is a feature not present +-- in the package Workflow. +-- +-- > r <- logged $ do +-- > logged this :: TransIO () +-- > logged that :: TransIO () +-- > logged thatOther +-- > liftIO $ print r +-- +-- when `print` is executed, the log is just the value of r. +-- +-- but at the `thatOther` execution the log is: [Exec,(), ()] +-- +logged :: Loggable a => TransientIO a -> TransientIO a +logged mx = Transient $ do + Log recover rs full <- getData `onNothing` return ( Log False [][]) + runTrans $ + case (recover ,rs) of -- !> ("logged enter",recover,rs) of + (True, Var x: rs') -> do + setData $ Log True rs' full + return $ fromIDyn x +-- !> ("read in Var:", x) + + (True, Exec:rs') -> do + setData $ Log True rs' full + mx -- !> "Exec" + + (True, Wait:rs') -> do + setData (Log True rs' full) -- !> "Wait" + empty + + _ -> do +-- let add= Exec: full + setData $ Log False (Exec : rs) (Exec: full) -- !> ("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 -> +-- let add= Wait: full + setData $ Log False (Wait: rs) (Wait: full) + _ -> return ()) + + Log recoverAfter lognew _ <- getData `onNothing` return ( Log False [][]) + let add= Var (toIDyn r): full + if recoverAfter && (not $ null lognew) -- !> ("recoverAfter", recoverAfter) + then (setData $ Log True lognew (reverse lognew ++ add) ) + -- !> ("recover",reverse lognew ,add) + else if recoverAfter && (null lognew) then + setData $ Log False [] add + else + (setData $ Log False (Var (toIDyn r):rs) add) -- !> ("restore", (Var (toIDyn r):rs)) + return r + + + diff --git a/src/Transient/Stream/Resource.hs b/src/Transient/Stream/Resource.hs --- a/src/Transient/Stream/Resource.hs +++ b/src/Transient/Stream/Resource.hs @@ -1,76 +1,76 @@ ------------------------------------------------------------------------------ --- --- Module : Transient.Stream.Resource --- Copyright : --- License : GPL-3 --- --- Maintainer : agocorona@gmail.com --- Stability : --- Portability : --- --- | --- ------------------------------------------------------------------------------ -{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable #-} -module Transient.Stream.Resource(sourceFile, sinkFile, process, initFinish, finish, onFinish) where - - -import Transient.Base hiding (loop) -import Transient.Backtrack -import Control.Exception -import Control.Applicative -import Data.Typeable -import Data.Char -import System.IO - -import Control.Concurrent - -import Control.Concurrent.STM -import Control.Monad.State - - - --- | Stream the input to a file -sinkFile :: TransIO String -> String -> TransIO () -sinkFile input file= process input (openFile file WriteMode) hClose' hPutStrLn' - where - hClose' h _= putStr "closing " >> putStrLn file >> hClose h - hPutStrLn' h x= liftIO $ (SMore <$> hPutStrLn h x) - `catch` (\(e::SomeException)-> return $ SError e) - --- | slurp input from a file a line at a time. It creates as much threads as possible. --- to allow single threaded processing, use it with `threads 0` -sourceFile :: String -> TransIO String -sourceFile file= process (return ()) (openFile file ReadMode) hClose' read' - where - hGetLine' h= (SMore <$> hGetLine h) - `catch` (\(e::SomeException)-> return $ SError e) - read' h _ = parallel $ hGetLine' h - - - hClose' h _= putStr "closing ">> putStrLn file >> hClose h - --- | is the general operation for processing a streamed input, with opening resources before --- processing and closing them when finish is called. The process statements suscribe to the --- `Finish` EVar. --- --- When this variable is updated, the close procedure is called. --- --- When the processing return `SDone` or `SError`, the `Finish` variable is updated so all the --- subscribed code, that close the resources, is executed. -process - :: TransIO a -- ^ input computation - -> IO handle -- ^ open computation that gives resources to be used during the computation - -> (handle -> Maybe SomeException -> IO ()) -- ^ close computation that frees the resources - -> (handle -> a -> TransIO (StreamData b)) -- ^ process to be done - -> TransIO b -process input open close proc=do - mh <- liftIO $ (Right <$> open) `catch` (\(e::SomeException)-> return $ Left e) - case mh of - Left e -> liftIO (putStr "process: " >> print e) >> finish (Just e) >> stop - Right h -> do - onFinish (liftIO . close h) - some <- input - v <- proc h some - liftIO $ myThreadId >>= print - checkFinalize v +----------------------------------------------------------------------------- +-- +-- Module : Transient.Stream.Resource +-- Copyright : +-- License : GPL-3 +-- +-- Maintainer : agocorona@gmail.com +-- Stability : +-- Portability : +-- +-- | +-- +----------------------------------------------------------------------------- +{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable #-} +module Transient.Stream.Resource(sourceFile, sinkFile, process, initFinish, finish, onFinish) where + + +import Transient.Base hiding (loop) +import Transient.Backtrack +import Control.Exception +import Control.Applicative +import Data.Typeable +import Data.Char +import System.IO + +import Control.Concurrent + +import Control.Concurrent.STM +import Control.Monad.State + + + +-- | Stream the input to a file +sinkFile :: TransIO String -> String -> TransIO () +sinkFile input file= process input (openFile file WriteMode) hClose' hPutStrLn' + where + hClose' h _= putStr "closing " >> putStrLn file >> hClose h + hPutStrLn' h x= liftIO $ (SMore <$> hPutStrLn h x) + `catch` (\(e::SomeException)-> return $ SError e) + +-- | slurp input from a file a line at a time. It creates as much threads as possible. +-- to allow single threaded processing, use it with `threads 0` +sourceFile :: String -> TransIO String +sourceFile file= process (return ()) (openFile file ReadMode) hClose' read' + where + hGetLine' h= (SMore <$> hGetLine h) + `catch` (\(e::SomeException)-> return $ SError e) + read' h _ = parallel $ hGetLine' h + + + hClose' h _= putStr "closing ">> putStrLn file >> hClose h + +-- | is the general operation for processing a streamed input, with opening resources before +-- processing and closing them when finish is called. The process statements suscribe to the +-- `Finish` EVar. +-- +-- When this variable is updated, the close procedure is called. +-- +-- When the processing return `SDone` or `SError`, the `Finish` variable is updated so all the +-- subscribed code, that close the resources, is executed. +process + :: TransIO a -- ^ input computation + -> IO handle -- ^ open computation that gives resources to be used during the computation + -> (handle -> Maybe SomeException -> IO ()) -- ^ close computation that frees the resources + -> (handle -> a -> TransIO (StreamData b)) -- ^ process to be done + -> TransIO b +process input open close proc=do + mh <- liftIO $ (Right <$> open) `catch` (\(e::SomeException)-> return $ Left e) + case mh of + Left e -> liftIO (putStr "process: " >> print e) >> finish (Just e) >> stop + Right h -> do + onFinish (liftIO . close h) + some <- input + v <- proc h some + liftIO $ myThreadId >>= print + checkFinalize v diff --git a/transient.cabal b/transient.cabal --- a/transient.cabal +++ b/transient.cabal @@ -1,51 +1,51 @@ -name: transient - -version: 0.4.4 - - -author: Alberto G. Corona - -cabal-version: >=1.10 -build-type: Simple - -license: MIT -license-file: LICENSE - -maintainer: agocorona@gmail.com -homepage: http://www.fpcomplete.com/user/agocorona -bug-reports: https://github.com/agocorona/transient/issues - -synopsis: Making composable programs with multithreading, events and distributed computing -description: See - In this release distributed primitives have been moved to the transient-universe package, and web primitives have been moved to the ghcjs-hplay package. -category: Control -data-dir: "" - - -library - build-depends: base > 4 && < 5 - , containers - , mtl - , transformers - , stm - , time - , directory - , random - - exposed-modules: Transient.Backtrack - Transient.Base - Transient.EVars - Transient.Indeterminism - Transient.Internals - Transient.Logged - Transient.Stream.Resource - exposed: True - buildable: True - exposed: True - default-language: Haskell2010 - hs-source-dirs: src . - - -source-repository head - type: git - location: https://github.com/agocorona/transient +name: transient + +version: 0.4.4.1 + +author: Alberto G. Corona +maintainer: agocorona@gmail.com + +cabal-version: >=1.10 + +build-type: Simple + +license: MIT +license-file: LICENSE + +homepage: http://www.fpcomplete.com/user/agocorona +bug-reports: https://github.com/agocorona/transient/issues + +synopsis: Making composable programs with multithreading, events and distributed computing +description: See + In this release distributed primitives have been moved to the transient-universe package, and web primitives have been moved to the ghcjs-hplay package. +category: Control + +data-dir: "" + + +library + build-depends: base > 4 && < 5 + , containers + , mtl + , transformers + , stm + , time + , directory + , random + + exposed-modules: Transient.Backtrack + Transient.Base + Transient.EVars + Transient.Indeterminism + Transient.Internals + Transient.Logged + Transient.Stream.Resource + exposed: True + buildable: True + exposed: True + default-language: Haskell2010 + hs-source-dirs: src . + +source-repository head + type: git + location: https://github.com/agocorona/transient