broccoli 0.2.0.1 → 0.3.0.0
raw patch · 2 files changed
+285/−128 lines, 2 filesdep +timedep −unamb
Dependencies added: time
Dependencies removed: unamb
Files
- Control/Broccoli.hs +282/−125
- broccoli.cabal +3/−3
Control/Broccoli.hs view
@@ -5,51 +5,67 @@ module Control.Broccoli ( X, E,++ -- * Event and signal combinators never,+ unionE,+ (<>),+ (-|-),+ voidE, snapshot, snapshot_,- accumulate,+ accum, edge,+ trap, justE, maybeE, filterE,++ -- * Setup IO interface Setup,+ Time,+ Boot, runProgram, newX, newE, input, output,++ -- * Debug debugX,- debugE+ debugE, ) where import Control.Applicative-import Data.Functor import Data.Monoid import Control.Monad-import Data.Unamb import Data.IORef import Control.Concurrent import Control.Concurrent.STM-import Data.Function import System.IO.Unsafe+import Data.Time --- | A value of type a that varies.+-- | @X a@ represents time signals with values of type @a@.+-- +-- > X a = Time -> a data X a where PureX :: a -> X a FmapX :: forall a b . (b -> a) -> X b -> X a ApplX :: forall a b . X (b -> a) -> X b -> X a- PortX :: MVar [ThreadId] -> TVar a -> X a+ TimeX :: a ~ Time => Context -> X a+ PortX :: Context -> TVar (a,Time) -> X a --- | An event that carries values of type a when it occurs.+-- | @E a@ represents events with values of type @a@.+-- +-- > E a = [(Time, a)] data E a where NeverE :: E a FmapE :: forall a b . (b -> a) -> E b -> E a MappendE :: E a -> E a -> E a- ProductE :: (b -> c -> a) -> E b -> E c -> E a- SnapshotE :: E b -> X a -> E a+ SnapshotE :: a ~ (b,c) => E b -> X c -> E a JustE :: E (Maybe a) -> E a- PortE :: MVar [ThreadId] -> TChan a -> E a+ --DelayedE :: E a -> Int -> E a+ PortE :: Context -> TChan (a, Time) -> E a instance Functor X where fmap f x = FmapX f x@@ -61,12 +77,13 @@ instance Functor E where fmap f e = FmapE f e +-- | mempty = 'never', mappend = 'unionE' instance Monoid (E a) where- mempty = NeverE- mappend e1 e2 = MappendE e1 e2+ mempty = never+ mappend = unionE -- | A monad for hooking up inputs and outputs to a program.-data Setup a = Setup (MVar [ThreadId] -> IO a)+data Setup a = Setup (Context -> IO a) instance Monad Setup where return x = Setup (\_ -> return x)@@ -83,29 +100,47 @@ instance Functor Setup where fmap f (Setup io) = Setup (\mv -> f <$> io mv) +-- | Time is measured from the beginning of a simulation in seconds.+type Time = Double++-- | The boot event occurs once at the beginning of a simulation.+type Boot = ()++data Context = Context+ { cxThreads :: MVar [ThreadId]+ , cxEpoch :: UTCTime }+ setupIO :: IO a -> Setup a setupIO io = Setup (\_ -> io) -getThreads :: Setup (MVar [ThreadId])-getThreads = Setup (\mv -> return mv)+getContext :: Setup Context+getContext = Setup (\cx -> return cx) -getThreadsE :: E a -> Maybe (MVar [ThreadId])-getThreadsE e = case e of- NeverE -> Nothing- FmapE _ e' -> getThreadsE e'- MappendE e1 e2 -> getFirst $ First (getThreadsE e1) <> First (getThreadsE e2)- ProductE _ e1 e2 -> getFirst $ First (getThreadsE e1) <> First (getThreadsE e2)- SnapshotE e' x -> getFirst $ First (getThreadsE e') <> First (getThreadsX x)- JustE e' -> getThreadsE e'- PortE mv _ -> Just mv+containsTimeX :: X a -> Bool+containsTimeX x = case x of+ PureX _ -> False+ FmapX _ x' -> containsTimeX x'+ ApplX x1 x2 -> containsTimeX x1 || containsTimeX x2+ TimeX _ -> True+ PortX _ _ -> False -getThreadsX :: X a -> Maybe (MVar [ThreadId])-getThreadsX x = case x of+getContextX :: X a -> Maybe Context+getContextX x = case x of PureX _ -> Nothing- FmapX _ x' -> getThreadsX x'- ApplX x1 x2 -> getFirst $ First (getThreadsX x1) <> First (getThreadsX x2)- PortX mv _ -> Just mv+ FmapX _ x' -> getContextX x'+ ApplX x1 x2 -> getFirst $ First (getContextX x1) <> First (getContextX x2)+ TimeX cx -> Just cx+ PortX cx _ -> Just cx +getContextE :: E a -> Maybe Context+getContextE e = case e of+ NeverE -> Nothing+ FmapE _ e' -> getContextE e'+ MappendE e1 e2 -> getFirst $ First (getContextE e1) <> First (getContextE e2)+ SnapshotE e' x -> getFirst $ First (getContextE e') <> First (getContextX x)+ JustE e' -> getContextE e'+ PortE cx _ -> Just cx+ dupE :: E a -> IO (E a) dupE e = case e of NeverE -> return NeverE@@ -116,10 +151,6 @@ e1' <- dupE e1 e2' <- dupE e2 return (MappendE e1' e2')- ProductE f e1 e2 -> do- e1' <- dupE e1- e2' <- dupE e2- return (ProductE f e1' e2') SnapshotE e' x -> do e'' <- dupE e' return (SnapshotE e'' x)@@ -130,54 +161,111 @@ ch' <- atomically (dupTChan ch) return (PortE mv ch') -readE :: E a -> IO a+data Promise a = Promise { force :: IO a }++instance Functor Promise where+ fmap f p = f <$> Promise (force p)++instance Applicative Promise where+ pure x = Promise (return x)+ ff <*> xx = Promise $ do+ f <- force ff+ x <- force xx+ return (f x)++data EventResult a =+ TryLater |+ DropThis |+ NotNowNotEver |+ --Delayed a Int |+ Normal a Double+ deriving Show++readE :: E a -> STM (EventResult a) readE e = case e of- NeverE -> hang- MappendE e1 e2 -> race (readE e1) (readE e2)- FmapE f e' -> f <$> readE e'- ProductE f e1 e2 -> do- x <- readE e1- y <- readE e2- return (f x y)+ NeverE -> return NotNowNotEver+ MappendE e1 e2 -> do+ mx <- readE e1+ case mx of+ TryLater -> readE e2+ ok -> return ok+ FmapE f e' -> do+ mx <- readE e'+ case mx of+ Normal x t -> return (Normal (f x) t)+ TryLater -> return TryLater+ DropThis -> return DropThis+ NotNowNotEver -> return NotNowNotEver SnapshotE e' x -> do- readE e'- atomically (readX x)- JustE e' -> fix $ \loop -> do- m <- readE e'- case m of- Nothing -> loop- Just x -> return x- PortE _ ch -> atomically (readTChan ch)+ ma <- readE e'+ case ma of+ Normal a t -> do+ (b, _) <- readX t x+ return (Normal (a,b) t)+ TryLater -> return TryLater+ DropThis -> return DropThis+ NotNowNotEver -> return NotNowNotEver+ JustE e' -> do+ mx <- readE e'+ case mx of+ Normal Nothing _ -> return DropThis+ Normal (Just x) t -> return (Normal x t)+ TryLater -> return TryLater+ DropThis -> return DropThis+ NotNowNotEver -> return NotNowNotEver+ PortE _ ch -> do+ emp <- isEmptyTChan ch+ if emp+ then return TryLater+ else do+ (v, t) <- readTChan ch+ return (Normal v t) -readX :: X a -> STM a-readX x = case x of- PureX v -> return v- FmapX f xx -> f <$> readX xx+readX :: Double -> X a -> STM (a, Time)+readX time sig = case sig of+ PureX v -> return (v, 0)+ FmapX f xx -> do+ (x, t) <- readX time xx+ return (f x, t) ApplX ff xx -> do- f <- readX ff- x <- readX xx- return (f x)+ (f,t1) <- readX time ff+ (x,t2) <- readX time xx+ return (f x, max t1 t2)+ TimeX _ -> return (time, time) PortX _ tv -> readTVar tv -hang :: IO a-hang = do- threadDelay (100 * 10^(6::Int))- hang- waitE :: E a -> IO a waitE e0 = do e <- dupE e0- readE e+ (x, _) <- readEIO e+ return x +readEIO :: E a -> IO (a, Time)+readEIO e = do+ result <- atomically $ do+ mx <- readE e+ case mx of+ TryLater -> retry+ other -> return other+ case result of+ TryLater -> error "impossible"+ Normal a t -> return (a, t)+ DropThis -> readEIO e+ NotNowNotEver -> hang++hang :: IO a+hang = do+ threadDelay (100 * 10^(6::Int))+ hang --- -- | An event which gets the value of a signal when another event occurs. snapshot :: E a -> X b -> E (a,b)-snapshot e x = ProductE (,) e (SnapshotE e x)+snapshot e x = SnapshotE e x -- | Like snapshot but ignores the original event's payload. snapshot_ :: E a -> X b -> E b-snapshot_ e x = SnapshotE e x+snapshot_ e x = snd <$> snapshot e x -- | Filter out events with the value of Nothing. justE :: E (Maybe a) -> E a@@ -193,113 +281,182 @@ -- | An event that never happens. never :: E a-never = mempty+never = NeverE +-- | All the occurrences from two events together. Same as '<>'.+unionE :: E a -> E a -> E a+unionE = MappendE++-- | Same as 'unionE' but on events that might have a different type.+(-|-) :: E a -> E b -> E (Either a b)+e1 -|- e2 = (Left <$> e1) <> (Right <$> e2)++-- | Forget the values associated with the events.+voidE :: E a -> E ()+voidE e = () <$ e+ -- | Sum over events using an initial state and a state transition function.-accumulate :: E a -> s -> (a -> s -> s) -> X s-accumulate e0 s0 trans = case getThreadsE e0 of+accum :: E a -> s -> (a -> s -> s) -> X s+accum e0 s0 trans = case getContextE e0 of Nothing -> pure s0- Just mv -> PortX mv tv where+ Just cx -> PortX cx tv where tv = unsafePerformIO $ do- state <- newTVarIO s0+ state <- newTVarIO (s0, 0) threadId <- forkIO $ do e <- dupE e0 forever $ do- x <- readE e atomically $ do- s <- readTVar state- let s' = trans x s- writeTVar state s'- modifyMVar_ mv (return . (threadId:))+ mx <- readE e+ case mx of+ TryLater -> retry+ DropThis -> return ()+ Normal x t -> do+ (s,_) <- readTVar state+ let s' = trans x s+ writeTVar state (s',t)+ NotNowNotEver -> error "impossible (2)"+ modifyMVar_ (cxThreads cx) (return . (threadId:)) return state --- | An event that occurs when an edge is detected in a signal. The edge test--- is applied to values before and after a discrete transition in the signal.--- The test should return Nothing when the two values are the same.+-- | A signal that remembers the most recent occurrence of an event.+trap :: a -> E a -> X a+trap x0 e = accum e x0 (\x _ -> x)++-- | An event that occurs when an edge is detected in a signal. When a signal+-- changes discretely the edge test is evaluated on the values immediately+-- before and after a change. edge :: X a -> (a -> a -> Maybe b) -> E b-edge x diff = case getThreadsX x of+edge x diff = case getContextX x of Nothing -> never- Just mv -> PortE mv ch where+ Just cx -> PortE cx ch where ch = unsafePerformIO $ do out <- newBroadcastTChanIO threadId <- forkIO $ do- v0 <- atomically (readX x)+ (v0,_) <- atomically (readX 0 x) ref <- newIORef v0- forever $ do- v <- readIORef ref- (d, v') <- atomically $ do- v' <- readX x+ if containsTimeX x+ then forever $ do+ now <- chron (cxEpoch cx)+ v <- readIORef ref+ (v', _) <- atomically (readX now x) case diff v v' of- Just d -> return (d, v')- Nothing -> retry- writeIORef ref v'- atomically (writeTChan out d)- modifyMVar_ mv (return . (threadId:))+ Just d -> do+ writeIORef ref v'+ atomically (writeTChan out (d,now))+ Nothing -> return ()+ else forever $ do+ v <- readIORef ref+ (d, v', t) <- atomically $ do+ (v', t) <- readX undefined x+ case diff v v' of+ Just d -> return (d, v', t)+ Nothing -> retry+ writeIORef ref v'+ atomically (writeTChan out (d, t))+ modifyMVar_ (cxThreads cx) (return . (threadId:)) return out +chron :: UTCTime -> IO Double+chron epoch = do+ now <- getCurrentTime+ let time = diffUTCTime now epoch+ return (realToFrac time) --- | Creates a new event and an IO action to trigger it.+newInternalBoot :: Context -> IO (E (), IO ())+newInternalBoot cx = do+ bch <- newBroadcastTChanIO+ return+ ( PortE cx bch+ , atomically (writeTChan bch ((), 0)) )++-- | Creates a new input signal with an initial value. Use 'input' to feed+-- data to the signal during the simulation.+newX :: a -> Setup (X a, a -> IO ())+newX v = do+ cx <- getContext+ let epoch = cxEpoch cx+ tv <- setupIO (newTVarIO (v,0))+ return+ ( PortX cx tv+ , \x -> do+ now <- chron epoch+ atomically (writeTVar tv (x,now)))++-- | Creates a new input event and a command to trigger it. Use 'input' to+-- to provide external stimulus during the simulation. newE :: Setup (E a, a -> IO ()) newE = do- mv <- getThreads+ cx <- getContext+ let epoch = cxEpoch cx bch <- setupIO newBroadcastTChanIO- return (PortE mv bch, atomically . writeTChan bch)+ return+ ( PortE cx bch+ , \x -> do+ now <- chron epoch+ atomically (writeTChan bch (x,now))) --- | Creates a new signal and an IO action to update it. The argument is--- the initial value of the signal.-newX :: a -> Setup (X a, a -> IO ())-newX v = do- mv <- getThreads- tv <- setupIO (newTVarIO v)- return (PortX mv tv, atomically . writeTVar tv) --- | Spawn a thread to execute an action for each event occurrence.-output :: E a -> (a -> IO ()) -> Setup ()++-- | Setup a thread to react to events. The callback will be provided with+-- the time of the event which is measured in seconds since the start of+-- the simulation.+output :: E a -> (Time -> a -> IO ()) -> Setup () output e0 act = do- mv <- getThreads+ cx <- getContext setupIO $ do e <- dupE e0- tid <- (forkIO . forever) (readE e >>= act)- modifyMVar_ mv (return . (tid:))+ tid <- forkIO . forever $ do+ (x, t) <- readEIO e+ act t x+ modifyMVar_ (cxThreads cx) (return . (tid:)) return () --- | Spawn an input thread to generate source signals and events.+-- | A thread to generate source signals and events will be started+-- when setup is complete. input :: IO () -> Setup () input handler = do- mv <- getThreads+ cx <- getContext setupIO $ do tid <- forkIO handler- modifyMVar_ mv (return . (tid:))+ modifyMVar_ (cxThreads cx) (return . (tid:)) return () --- | Run the setup action to create input and output threads. The returned IO--- action will be executed when setup is complete. runProgram blocks until--- the returned event occurs, at which time it kills all the threads and--- returns.-runProgram :: Setup (IO (), E ()) -> IO ()-runProgram (Setup setup) = do+-- | Runs a program defined by the setup computation. The simulation ends+-- if the returned event occurs.+runProgram :: (E Boot -> X Time -> Setup (E ())) -> IO ()+runProgram setup = do mv <- newMVar []- (boot, exit) <- setup mv+ epoch <- getCurrentTime+ let cx = Context mv epoch+ let time = TimeX cx+ (onBoot, boot) <- newInternalBoot cx+ let Setup act = setup onBoot time+ exit <- act cx --threadDelay 5000 boot waitE exit- withMVar mv (mapM killThread)+ _ <- withMVar mv (mapM killThread) return () -- | Print out events as they occur. Only for debugging purposes.-debugE :: Show a => E a -> E a-debugE e = unsafePerformIO $ do+debugE :: (a -> String) -> E a -> E a+debugE toString e = unsafePerformIO $ do e' <- dupE e- (forkIO . forever) (readE e' >>= print)+ _ <- forkIO . forever $ do+ (x, _) <- readEIO e'+ putStrLn (toString x) return e -- | Print out transitions in a signal. Only for debugging purposes.-debugX :: (Eq a, Show a) => X a -> X a-debugX x =+debugX :: Eq a => ((a, a) -> String) -> X a -> X a+debugX toString sig = let diff a b = if a == b then Nothing else Just (a,b) in- let e = edge x diff in+ let e = edge sig diff in unsafePerformIO $ do- forkIO $ do+ _ <- forkIO $ do e' <- dupE e- forever (readE e' >>= print)- return x+ forever $ do+ (x, _) <- readEIO e'+ putStrLn (toString x)+ return sig+
broccoli.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: broccoli-version: 0.2.0.1+version: 0.3.0.0 synopsis: Small library for interactive functional programs. description: Small library for interactive functional programs. license: BSD3@@ -20,7 +20,7 @@ -- other-modules: other-extensions: GADTs, RankNTypes build-depends: base >=4.7 && <4.8,- unamb >=0.2 && <0.3,+ time >= 1.4 && <1.6, stm >=2.4 && <2.5 -- hs-source-dirs: default-language: Haskell2010@@ -32,4 +32,4 @@ source-repository this type: git location: https://github.com/evanrinehart/broccoli- tag: 0.2.0.1+ tag: 0.3.0.0