transient 0.5.1 → 0.5.3
raw patch · 5 files changed
+197/−82 lines, 5 filesdep ~base
Dependency ranges changed: base
Files
- src/Transient/Base.hs +1/−1
- src/Transient/Indeterminism.hs +38/−15
- src/Transient/Internals.hs +145/−51
- src/Transient/Logged.hs +11/−13
- transient.cabal +2/−2
src/Transient/Base.hs view
@@ -26,7 +26,7 @@ ,react -- * State management -,setState, setData, getState, getSData,getData,delState,delData, modifyData,modifyState +,setState, setData, getState, getSData,getData,delState,delData, modifyData,modifyState,try -- * Thread management , threads,addThreads, freeThreads, hookedThreads,oneThread, killChilds
src/Transient/Indeterminism.hs view
@@ -91,42 +91,65 @@ collect n = collect' n 0 -- | search with a timeout --- After the the timeout, it stop unconditionally and return the current results. +-- After the timeout, it stop unconditionally and return the current results. -- It also stops as soon as there are enough results specified in the first parameter. +-- The results are returned by the original thread +-- +-- > timeout t proc=do +-- > r <- collect' 1 t proc +-- > case r of +-- > [] -> empty +-- > r:_ -> return r +-- +-- > timeout 10000 empty <|> liftIO (print "timeout") +-- +-- That executes the alternative and will print "timeout". +-- This would not be produced if collect would not return the results to the original thread +-- +-- `search` is executed in different threads and his state is lost, so don't rely in state +-- to pass information + collect' :: Int -> Int -> TransIO a -> TransIO [a] -collect' n t2 search= do +collect' n t search= do rv <- liftIO $ newEmptyMVar -- !> "NEWMVAR" - results <- liftIO $ newIORef (0,[]) - let worker = do - r <- search - liftIO $ putMVar rv r + let worker = do + r <- abduce >> search + liftIO $ putMVar rv $ Just r stop - timer= if t2 > 0 then async $ threadDelay t2 >> readIORef results >>= return . snd - else empty + timer= do + when (t > 0) . async $ threadDelay t >>putMVar rv Nothing -- readIORef results >>= return . snd + empty - monitor= async loop + monitor= liftIO loop where loop = do - r <- takeMVar rv - (n',rs) <- readIORef results + mr <- takeMVar rv + (n',rs) <- readIORef results + case mr of + Nothing -> return rs + Just r -> do let n''= n' +1 - writeIORef results (n'',r:rs) + let rs'= r:rs + writeIORef results (n'',rs') t' <- getCurrentTime if (n > 0 && n'' >= n) - then readIORef results >>= return . snd else loop - `catch` \(e :: BlockedIndefinitelyOnMVar) -> + then return (rs') + else loop + `catch` \(e :: BlockedIndefinitelyOnMVar) -> readIORef results >>= return . snd - r <- oneThread $ monitor <|> worker <|> timer + r <- oneThread $ worker <|> timer <|> monitor return r + +
src/Transient/Internals.hs view
@@ -32,7 +32,8 @@ import Debug.Trace import System.IO.Unsafe import Unsafe.Coerce -import Control.Exception +import Control.Exception hiding (try,onException) +import qualified Control.Exception (try) import Control.Concurrent import GHC.Conc(unsafeIOToSTM) import Control.Concurrent.STM hiding (retry) @@ -46,15 +47,16 @@ import System.IO import System.Exit -import qualified Data.ByteString.Char8 as BS - +import qualified Data.ByteString.Lazy.Char8 as BS +import Data.ByteString.Builder ---{-# INLINE (!>) #-} ---(!>) :: Show a => b -> a -> b ---(!>) x y= trace (show y) x ---infixr 0 !> +{-# INLINE (!>) #-} +(!>) :: Show a => b -> a -> b +(!>) x y= trace (show y) x +infixr 0 !> +-- (!>) x y= x @@ -285,34 +287,75 @@ readsPrec' _= unsafePerformIO . readWithErr -class (Show a, Read a, Typeable a) => Loggable a where +--class (Show a, Read a, Typeable a) => Loggable a where +class Typeable a => Loggable a where + serialize :: a -> Builder + safeDeserialize :: BS.ByteString -> Maybe a -instance (Show a, Read a,Typeable a) => Loggable a where +deserialize str = r + where + r= case safeDeserialize str of + Just x -> x + Nothing -> error $ "deserialization failure for (string,to type): " ++ show (BS.unpack str,typeOf r) + +--instance (Show a, Read a,Typeable a) => Loggable a where + -- | Dynamic serializable data for logging -data IDynamic= IDyns String | forall a.Loggable a => IDynamic a +data IDynamic= IDyns BS.ByteString | forall a.Loggable a => IDynamic a -instance Show IDynamic where - show (IDynamic x)= show $ show x - show (IDyns s)= show s +instance Loggable IDynamic where + serialize (IDynamic x)= serialize x + serialize (IDyns s)= lazyByteString s + safeDeserialize str= Just $ IDyns str -instance Read IDynamic where - readsPrec n str= map (\(x,s) -> (IDyns x,s)) $ readsPrec' n str +comma= BS.pack "," +instance Show IDynamic where + show = BS.unpack . toLazyByteString . serialize +instance Show Builder where + show = show . toLazyByteString + 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, Show) +data LogElem= Wait | Exec | Var IDynamic deriving (Show) +--instance (Read a,Show a , Typeable a) => Loggable a where +-- serialize = BS.pack . show +-- deserialize= read . BS.unpack +instance Loggable LogElem where + serialize Wait= lazyByteString $ BS.pack "Wait" + serialize Exec= lazyByteString $ BS.pack "Exec" + serialize (Var dyn)= serialize dyn + safeDeserialize str + | str== BS.pack "Wait" = Just Wait + | str== BS.pack "Exec" = Just Exec + | otherwise= Just . Var $ deserialize str + +instance Loggable [LogElem] where + serialize xs= mconcat $ map serialize' xs + where serialize' x= (serialize'' x) <> lazyByteString comma + serialize'' :: LogElem -> Builder + serialize'' x= let serx= toLazyByteString $ serialize x in lazyByteString $ (BS.pack . show $ BS.length serx) <> comma <> serx + + safeDeserialize str + | BS.null str= Just [] + | otherwise = case BS.readInt str of + Just(n,str') -> let (str'', str''')= BS.splitAt (fromIntegral n+1) str' + in Just $ deserialize (BS.tail str'') : if BS.null str''' then [] else deserialize (BS.tail str''') + Nothing -> Nothing + instance Alternative TransIO where empty = Transient $ return Nothing (<|>) = mplus +data Log= Log Recover CurrentPointer LogEntries deriving (Typeable) + data RemoteStatus= WasRemote | WasParallel | NoRemote deriving (Typeable, Eq, Show) instance MonadPlus TransIO where @@ -389,6 +432,8 @@ restoreStack fs return a + + -- (<**) :: TransIO a -> TransIO b -> TransIO a (<**) ma mb= Transient $ do a <- runTrans ma -- !> "ma" @@ -453,23 +498,25 @@ --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) +-- effects not used +-- {-# INLINE baseEffects #-} +--baseEffects :: Effects +-- +--baseEffects x x' f' = do +-- c <- setEventCont x' f' +-- mk <- runTrans x +-- t <- resetEventCont mk c +-- return (t,mk) instance 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 + c <- setEventCont x f + mk <- runTrans x + resetEventCont mk c + case mk of Just k -> runTrans (f k) Nothing -> return Nothing @@ -746,7 +793,17 @@ delState :: ( MonadState EventF m,Typeable a) => a -> m () delState= delData +-- | Executes the computation and reset the state if it fails. +try :: TransIO a -> TransIO a +try mx= do + sd <- gets mfData + mx <|> (modify (\ s ->s{mfData= sd}) >> empty) +-- | Executes the computation and reset the state either if it fails or not +sandbox :: TransIO a -> TransIO a +sandbox mx= do + sd <- gets mfData + mx <*** modify (\s ->s{mfData= sd}) -- | Generator of identifiers that are unique withing the current monadic sequence -- They are not unique in the whole program. @@ -956,22 +1013,21 @@ forkFinally1 :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId forkFinally1 action and_then = - mask $ \restore -> forkIO $ try (restore action) >>= and_then + mask $ \restore -> forkIO $ Control.Exception.try (restore action) >>= and_then - free th env= do --- return () !> ("freeing",th,"in",threadId env) +free th env= do +-- return () !> ("freeing",th,"in",threadId env) let sibling= children env (sbs',found) <- modifyMVar sibling $ \sbs -> do let (sbs', found) = drop [] th sbs return (sbs',(sbs',found)) --- sbs <- takeMVar sibling --- let (sbs', found) = drop [] th sbs -- !> "search "++show th ++ " in " ++ show (map threadId sbs) + if found then do --- putMVar sibling sbs' + -- !> ("new list for",threadId env,map threadId sbs') (typ,_) <- readIORef $ labelth env if (null sbs' && typ /= Listener && isJust (parent env)) @@ -1027,6 +1083,8 @@ -- -- it configures the event handler by calling the setter of the event -- handler with the current continuation +-- +-- react is composable with applicative and alternative operators too react :: Typeable eventdata => ((eventdata -> IO response) -> IO ()) @@ -1047,8 +1105,31 @@ put cont{event=Nothing} return $ unsafeCoerce j +-- | continue the computation in another thread and return `empty` to the computation in the curren thread. +-- +-- Useful for executing alternative computations +abduce = async $ return () +-- Transient $ do +-- st <- get +-- case event st of +-- Just _ -> do +-- put st{event=Nothing} +-- return $ Just () +-- Nothing -> do +-- chs <- liftIO $ newMVar [] +-- +-- label <- liftIO $ newIORef (Alive, BS.pack "abduce") +-- liftIO $ forkIO $ do +-- th <- myThreadId +-- let st' = st{event= Just (),parent=Just st,children= chs, threadId=th,labelth= label} +-- liftIO $ hangThread st st' +-- +-- runCont' st' +-- return() +-- return Nothing + -- * non-blocking keyboard input getLineRef= unsafePerformIO $ newTVarIO Nothing @@ -1065,7 +1146,7 @@ liftIO $ putStrLn $ "Enter "++sret++"\tto: " ++ message liftIO $ modifyMVar_ roption $ \msgs-> return $ sret:msgs waitEvents $ getLine' (==ret) - liftIO $ putStr "option: " >> putStrLn (show ret) + liftIO $ putStr "\noption: " >> putStrLn (show ret) return ret @@ -1276,10 +1357,12 @@ {-# NOINLINE onBack #-} onBack :: (Typeable b, Show b) => TransientIO a -> ( b -> TransientIO a) -> TransientIO a onBack ac bac = registerBack (typeof bac) $ Transient $ do - Backtrack mreason _ <- getData `onNothing` backStateOf (typeof bac) + Backtrack mreason stack <- getData `onNothing` backStateOf (typeof bac) runTrans $ case mreason of Nothing -> ac - Just reason -> bac reason + Just reason -> do + setState $ Backtrack mreason $ tail stack -- to avoid recursive call tot he same handler + bac reason where typeof :: (b -> TransIO a) -> b typeof = undefined @@ -1288,7 +1371,7 @@ onUndo x y= onBack x (\() -> y) --- | Register an action that will be executed when backtracking +-- | Register an action that will be executed when backtracking of a given type {-# NOINLINE registerUndo #-} registerBack :: (Typeable b, Show b) => b -> TransientIO a -> TransientIO a registerBack witness f = Transient $ do @@ -1328,7 +1411,7 @@ -- -- If the backtracking flag is changed the flow proceed forward from that point on. -- --- If the backtrack stack is finished or undoCut executed, `undo` will stop. +-- If the backtrack stack is finished or undoCut executed, the backtracking will stop. back :: (Typeable b, Show b) => b -> TransientIO a back reason = Transient $ do bs <- getData `onNothing` backStateOf reason -- !!>"GOBACK" @@ -1338,17 +1421,12 @@ goBackt (Backtrack _ [] )= return Nothing -- !!> "END" goBackt (Backtrack b (stack@(first : bs)) )= do - (setData $ Backtrack (Just reason) stack) + setData $ Backtrack (Just reason) stack mr <- runClosure first -- !> "RUNCLOSURE" Backtrack back _ <- getData `onNothing` backStateOf reason -- !> "END RUNCLOSURE" --- case back of --- Nothing -> case mr of --- Nothing -> return empty -- !> "FORWARD END" --- Just x -> runContinuation first x -- !> "FORWARD EXEC" --- justreason -> goBackt $ Backtrack justreason bs -- !> ("BACK AGAIN",back) case mr of Nothing -> return empty -- !> "END EXECUTION" @@ -1398,21 +1476,37 @@ SMore x -> return x ------ exceptions --- --- | When a exception is produced in the continuation, the handler is executed. +-- | When a exception is produced anywhere after this statement, the handler is executed. -- | handlers are executed Last in first out. onException :: Exception e => (e -> TransIO ()) -> TransIO () -onException f= onAnyException $ \e -> +onException exc= return () `onException'` exc + + +onException' :: Exception e => TransIO a -> (e -> TransIO a) -> TransIO a +onException' mx f= onAnyException mx $ \e -> case fromException e of - Nothing -> return () + Nothing -> empty Just e' -> f e' where - onAnyException :: (SomeException ->TransIO ()) -> TransIO () - onAnyException f= (return ()) `onBack` f + onAnyException :: TransIO a -> (SomeException ->TransIO a) -> TransIO a + onAnyException mx f= mx `onBack` f --- | stop the backtracking mechanism to execute further handlers +-- | stop the backtracking mechanism from executing further handlers cutExceptions= backCut (undefined :: SomeException) -- | Resume to normal execution at this point continue = forward (undefined :: SomeException) + + +catcht mx exc= sandbox $ do + cutExceptions + onException' mx exc + where + sandbox mx= do + exState <- getState <|> backStateOf (undefined :: SomeException) + mx + <*** setState exState + +
src/Transient/Logged.hs view
@@ -22,8 +22,7 @@ import Data.Typeable import Unsafe.Coerce -import Transient.Base -import Transient.Internals(Loggable) +import Transient.Internals import Transient.Indeterminism(choose) import Transient.Internals(onNothing,reads1,IDynamic(..),Log(..),LogElem(..),RemoteStatus(..),StateIO) import Control.Applicative @@ -31,7 +30,7 @@ import System.Directory import Control.Exception import Control.Monad - +import qualified Data.ByteString as BS #ifndef ghcjs_HOST_OS @@ -71,15 +70,14 @@ let list'= filter ((/=) '.' . head) list file <- choose list' -- !> list' - logstr <- liftIO $ readFile (logs++file) - let log= length logstr `seq` read' logstr + logstr <- liftIO $ BS.readFile (logs++file) + let log= BS.length logstr `seq` deserialize logstr log `seq` setData (Log True (reverse log) log) liftIO $ remove $ logs ++ file proc - where - read'= fst . head . reads1 + where remove f= removeFile f `catch` (\(e::SomeException) -> remove f) @@ -114,7 +112,7 @@ fromIDyn :: Loggable a => IDynamic -> a fromIDyn (IDynamic x)=r where r= unsafeCoerce x -- !> "coerce" ++ " to type "++ show (typeOf r) -fromIDyn (IDyns s)=r `seq`r where r= read s -- !> "read " ++ s ++ " to type "++ show (typeOf r) +fromIDyn (IDyns s)=r `seq`r where r= deserialize s -- !> "read " ++ s ++ " to type "++ show (typeOf r) @@ -201,7 +199,7 @@ else return Nothing _ -> return Nothing where - show1 x= if typeOf x == typeOf "" then unsafeCoerce x else show x + show1 x= serialize x param :: Loggable a => TransIO a param= res where @@ -213,15 +211,15 @@ setData $ Log recover t full return $ cast v Var (IDyns s):t -> do - let mr = reads1 s `asTypeOf` type1 res + let mr = safeDeserialize s `asTypeOf` type1 res case mr of - [] -> return Nothing - (v,r):_ -> do + Nothing -> return Nothing + Just v -> do setData $ Log recover t full return $ Just v _ -> return Nothing where - type1 :: TransIO a -> [(a,String)] + type1 :: TransIO a -> Maybe a type1= error "type1: typelevel"
transient.cabal view
@@ -1,6 +1,6 @@ name: transient -version: 0.5.1 +version: 0.5.3 author: Alberto G. Corona @@ -17,7 +17,7 @@ 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 +synopsis: composing programs with multithreading, events and distributed computing description: See <http://github.com/agocorona/transient> 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