transient 0.4.1 → 0.4.2

raw patch · 5 files changed

+120/−124 lines, 5 files

Files

src/Transient/EVars.hs +33/−45
src/Transient/Indeterminism.hs +10/−7
src/Transient/Internals.hs +51/−44
src/Transient/Logged.hs +1/−1
transient.cabal +25/−27

src/Transient/EVars.hs view

@@ -17,7 +17,7 @@ 
 
 
-data EVar a= EVar Int (TVar (Int,Int))  (TChan (StreamData a)) deriving  Typeable
+data EVar a= EVar  (TChan (StreamData a)) deriving  Typeable
 
 
 -- | creates an EVar.
@@ -42,55 +42,43 @@ newEVar ::  TransIO (EVar a)
 newEVar  = Transient $ do
    id <- genId
-   rn <- liftIO $ newTVarIO (0,0)
-   ref <-liftIO  newTChanIO
-   return . Just $ EVar id  rn ref
+   ref <-liftIO  newBroadcastTChanIO
+   return . Just $ EVar  ref
 
 -- | delete al the subscriptions for an evar.
 cleanEVar :: EVar a -> TransIO ()
-cleanEVar (EVar id rn ref1)= liftIO $ atomically $ do
+cleanEVar (EVar  ref1)= liftIO $ atomically $ do
     writeTChan  ref1 SDone
-    writeTVar rn (0,0)
 
+
 -- | 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 id rn ref1)= do
-     liftIO $ atomically $ readTVar rn >>= \(n,n') -> writeTVar rn $ (n+1,n'+1)
-     r <- parallel $ atomically $ do
-                r <- peekTChan ref1
-----                return ()               !> "peekTChan executed"
-                (n,n') <- readTVar rn  -- !> "readtvar rn"
---                return ()              !> ("rn",n)
-                if n'> 1 then do
-                           writeTVar rn (n,n'-1)
-                           return r
-                         else do
-                           readTChan ref1
-                           writeTVar rn (n,n)
-                           return r
+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 -> do
-            liftIO . atomically $ readTVar rn >>= \(n,n') -> writeTVar rn $ (n-1,n'-1)
-            error $ "readEVar: "++ show e
+        SError e -> error ""
+--            error $ "readEVar: "++ show e
 
 -- |  update the EVar and execute all readEVar blocks with "last in-first out" priority
 --
-writeEVar (EVar id rn ref1) x= liftIO $ atomically $ do
+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 id rn ref1) x= liftIO $ atomically $ do
+lastWriteEVar (EVar ref1) x= liftIO $ atomically $ do
        writeTChan  ref1 $ SLast x
 
 
@@ -99,20 +87,13 @@ 
 type FinishReason= Maybe SomeException
 
--- | 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
 
 
-
 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
@@ -120,16 +101,15 @@       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 <- readEVar finish
-                               close e  -- !!> "CLOSE"
-                               stop
-                         <|>
-                           return ()
+       Finish finish <- getSData <|> initFinish
+       e <-  freeThreads $ readEVar finish
+       close e  -- !!> "CLOSE"
+       stop
+     <|>
+       return ()
 
 
 
@@ -148,12 +128,20 @@    <|> return ()   -- !!> "NOT DELEVAR"
 
 
-killOnFinish comp=   do
+-- | kill all the processes generated by the parameter when finish event occurs
+killOnFinish comp= do
 
    chs <- liftIO $ newTVarIO []
-   onFinish $ const $ do
-
-      liftIO $ killChildren chs   -- !> "killOnFinish event"
+   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
+

src/Transient/Indeterminism.hs view

@@ -17,6 +17,7 @@ ) where
 
 import Transient.Base
+import Transient.EVars
 import Transient.Internals(killChildren, EventF(..),hangThread)
 import Data.IORef
 import Control.Applicative
@@ -31,20 +32,22 @@ 
 -- | slurp a list of values and process them in parallel . To limit the number of processing
 -- threads, use `threads`
-choose  ::  [a] -> TransIO a
+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]  -> return $ SLast x
-            x:_  -> return $ SMore x
-    return $ toData r
+           case es  of
+            [x]  -> x `seq` return $ SLast x
+            x:_  -> x `seq` return $ SMore x
+    toData r
 
 toData r= case r of
-      SMore x -> x
-      SLast x -> x
+      SMore x -> return x
+      SLast x -> return x
+      SError e ->  finish (Just e) >> empty
+
 
 -- | group the output of a possible multithreaded process in groups of n elements.
 group :: Int -> TransIO a -> TransIO [a]

src/Transient/Internals.hs view

@@ -150,6 +150,7 @@     (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
 
+
 -- | 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.
@@ -158,14 +159,14 @@ runContinuations fs x= (compose $ unsafeCoerce fs)  x
 
 instance   Functor TransIO where
-  fmap f mx=   -- Transient $ fmap (fmap f) $ runTrans mx
+  fmap f mx=  --   Transient $ fmap (fmap f) $ runTrans mx
     do
      x <- mx
      return $ f x
 
--- to set the identifier number...
-newtype IDNUM = IDNUM Int deriving Show
 
+
+
 instance Applicative TransIO where
   pure a  = Transient . return $ Just a
 
@@ -184,9 +185,9 @@                    Log rec _ full <- getData `onNothing` return (Log False [] [])
                    liftIO $ writeIORef rf  (Just k,full)       --  !> "APPF"
                    (x, full2)<- liftIO $ readIORef rg
-                   when (hasWait full2) $
-                        let full'= head full2: full
-                        in setData $ Log rec full' full'
+                   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
 
@@ -194,9 +195,9 @@                    Log rec _ full <- getData `onNothing` return (Log False [] [])
                    liftIO $ writeIORef rg (Just x, full)       -- !> "APPG"
                    (k,full1) <- liftIO $ readIORef rf
-                   when (hasWait  full) $
+                   when (hasWait  full) $                    -- !> ("full", full, "\nfull1",full1)) $
                         let full'= head full: full1
-                        in setData $ Log rec full' full'
+                        in (setData $ Log rec full' full')   -- !> ("result2",full')
 
                    return $ k <*> Just x
 
@@ -222,10 +223,6 @@              liftIO $ writeIORef rf  (k,full)
 
 
---             mfdata <- gets mfData
-             n <- gets mfSequence
-             setData $ IDNUM n
-
              setContinuation g appg fs
 
              x <- runTrans g             --  !> "RUN g"
@@ -233,7 +230,6 @@              liftIO $ writeIORef rg  (x,full')
              restoreStack fs
              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
@@ -291,44 +287,55 @@ stop :: Alternative m => m stop
 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.
-(**>) :: TransIO a -> TransIO b -> TransIO b
-(**>) x y=  Transient $ do
-          runTrans x
-          runTrans y
+    -- | 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
 
-infixr 1  <***  ,  <**, **>
+    -- | 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
 
--- | 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
-(<***) :: 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
+    atEnd' ::m a -> m b -> m a
+    atEnd' = (<**)
 
-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
 
--- | 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
-(<**) :: TransIO a -> TransIO b -> TransIO a
-(<**) ma mb= Transient $ do
-              a <- runTrans ma    -- !> "ma"
-              runTrans  mb        -- !> "mb"
-              return a
+    atEnd :: m a -> m b -> m a
+    atEnd= (<***)
 
 
-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.
 --
@@ -362,7 +369,7 @@ 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
+   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
@@ -371,7 +378,7 @@ -- 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
+   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
@@ -383,6 +390,7 @@ 
 --refEventCont= unsafePerformIO $ newIORef baseEffects
 
+{-# INLINE baseEffects #-}
 baseEffects :: Effects
 
 baseEffects x  x' f' = do
@@ -652,7 +660,6 @@ -- 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

src/Transient/Logged.hs view

@@ -58,7 +58,7 @@ --
 --  but at the `thatOther` execution the log is: [Exec,(), ()]
 --
-logged :: (Show a, Read a, Typeable a) => TransientIO a -> TransientIO a
+logged :: Loggable a => TransientIO a -> TransientIO a
 logged mx =  Transient $ do
    Log recover rs full <- getData `onNothing` return ( Log False  [][])
    runTrans $

transient.cabal view

@@ -1,49 +1,47 @@ name: transient
+version: 0.4.2
+author: Alberto G. Corona
 
-version: 0.4.1
 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 <http://github.com/agocorona/transient>
-             In this release, distributed primitives have been moved to the transient-universe package
-             Web primitives have been moved to the package ghcjs-hplay
+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
-author: Alberto G. Corona
 data-dir: ""
 
-source-repository head
-    type: git
-    location: https://github.com/agocorona/transient
 
 library
-
-        build-depends: base >4 && <5, containers, mtl, transformers, stm, time
+    build-depends:     base          > 4  &&  < 5
+                     , containers
+                     , mtl
+                     , transformers
+                     , stm
+                     , time
 
-        exposed-modules: Transient.Indeterminism
-                         Transient.Base Transient.EVars Transient.Backtrack
-                         Transient.Logged Transient.Stream.Resource
-                         Transient.Internals
-        exposed: True
-        buildable: True
+    exposed-modules: Transient.Backtrack
+                     Transient.Base
+                     Transient.EVars
+                     Transient.Indeterminism
+                     Transient.Internals
+                     Transient.Logged
+                     Transient.Stream.Resource
     exposed: True
     buildable: True
+    exposed: True
+    buildable: True
     default-language: Haskell2010
     hs-source-dirs: src .
---
---
---test-suite test-transient
---    build-depends: base >4 && <5, mtl , random ,
---                       containers , directory , filepath , stm ,
---                       HTTP , network , transformers , process
---    type: exitcode-stdio-1.0
---    main-is: TestSuite.hs
---    buildable: True
---    default-language: Haskell2010
---    hs-source-dirs: tests src .
 
 
+source-repository head
+    type: git
+    location: https://github.com/agocorona/transient