broccoli 0.3.0.0 → 0.4.0.0
raw patch · 2 files changed
+314/−98 lines, 2 filesdep +containers
Dependencies added: containers
Files
- Control/Broccoli.hs +310/−96
- broccoli.cabal +4/−2
Control/Broccoli.hs view
@@ -2,6 +2,7 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TupleSections #-} module Control.Broccoli ( X, E,@@ -20,10 +21,19 @@ justE, maybeE, filterE,+ multiplex,+ delayE,+ delayX,+ dilate,+ timeWarp,+ timeWarp',+ delayE',+ rasterize, -- * Setup IO interface Setup, Time,+ DeltaT, Boot, runProgram, newX,@@ -38,13 +48,19 @@ import Control.Applicative import Data.Monoid-import Control.Monad+import Control.Monad (forever, ap, msum) import Data.IORef import Control.Concurrent import Control.Concurrent.STM import System.IO.Unsafe import Data.Time+import Data.Sequence+import Data.Foldable (toList)+import Data.List (intercalate) +--import Debug.Trace+--import Unsafe.Coerce+ -- | @X a@ represents time signals with values of type @a@. -- -- > X a = Time -> a@@ -52,8 +68,10 @@ 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- TimeX :: a ~ Time => Context -> X a- PortX :: Context -> TVar (a,Time) -> X a+ MappendX :: Monoid a => X a -> X a -> X a+ MultiX :: forall a b . a ~ [b] => [X b] -> X a+ TimeX :: a ~ Time => Context -> (Time -> Time) -> X a+ PortX :: a -> Context -> TVar (a,Time) -> X a -- | @E a@ represents events with values of type @a@. -- @@ -64,8 +82,8 @@ MappendE :: E a -> E a -> E a SnapshotE :: a ~ (b,c) => E b -> X c -> E a JustE :: E (Maybe a) -> E a- --DelayedE :: E a -> Int -> E a PortE :: Context -> TChan (a, Time) -> E a+ SingleE :: Context -> TMVar a -> E a instance Functor X where fmap f x = FmapX f x@@ -74,6 +92,10 @@ pure x = PureX x f <*> x = ApplX f x +instance Monoid a => Monoid (X a) where+ mempty = pure mempty+ mappend = MappendX+ instance Functor E where fmap f e = FmapE f e @@ -100,15 +122,18 @@ 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 +type DeltaT = Double+ -- | The boot event occurs once at the beginning of a simulation. type Boot = () data Context = Context { cxThreads :: MVar [ThreadId]- , cxEpoch :: UTCTime }+ , cxEpoch :: UTCTime+ , cxDefer :: Time -> IO () -> IO ()+ } setupIO :: IO a -> Setup a setupIO io = Setup (\_ -> io)@@ -116,21 +141,15 @@ getContext :: Setup Context getContext = Setup (\cx -> return cx) -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- getContextX :: X a -> Maybe Context getContextX x = case x of PureX _ -> Nothing FmapX _ x' -> getContextX x' ApplX x1 x2 -> getFirst $ First (getContextX x1) <> First (getContextX x2)- TimeX cx -> Just cx- PortX cx _ -> Just cx+ MappendX x1 x2 -> getFirst $ First (getContextX x1) <> First (getContextX x2)+ TimeX cx _ -> Just cx+ PortX _ cx _ -> Just cx+ MultiX xs -> msum (map getContextX xs) getContextE :: E a -> Maybe Context getContextE e = case e of@@ -140,6 +159,7 @@ SnapshotE e' x -> getFirst $ First (getContextE e') <> First (getContextX x) JustE e' -> getContextE e' PortE cx _ -> Just cx+ SingleE cx _ -> Just cx dupE :: E a -> IO (E a) dupE e = case e of@@ -160,6 +180,10 @@ PortE mv ch -> do ch' <- atomically (dupTChan ch) return (PortE mv ch')+ SingleE cx mv -> do+ x <- atomically (readTMVar mv)+ mv' <- newTMVarIO x+ return (SingleE cx mv') data Promise a = Promise { force :: IO a } @@ -177,7 +201,6 @@ TryLater | DropThis | NotNowNotEver |- --Delayed a Int | Normal a Double deriving Show @@ -187,7 +210,12 @@ MappendE e1 e2 -> do mx <- readE e1 case mx of- TryLater -> readE e2+ TryLater -> do+ my <- readE e2+ case my of+ NotNowNotEver -> return TryLater+ other -> return other+ NotNowNotEver -> readE e2 ok -> return ok FmapE f e' -> do mx <- readE e'@@ -220,6 +248,11 @@ else do (v, t) <- readTChan ch return (Normal v t)+ SingleE _ mv -> do+ attempt <- tryTakeTMVar mv+ case attempt of+ Nothing -> return NotNowNotEver+ Just v -> return (Normal v 0) readX :: Double -> X a -> STM (a, Time) readX time sig = case sig of@@ -231,8 +264,16 @@ (f,t1) <- readX time ff (x,t2) <- readX time xx return (f x, max t1 t2)- TimeX _ -> return (time, time)- PortX _ tv -> readTVar tv+ MappendX xx1 xx2 -> do+ (x1, t1) <- readX time xx1+ (x2, t2) <- readX time xx2+ return (x1 <> x2, max t1 t2)+ TimeX _ tmap -> return (tmap time, tmap time)+ PortX _ _ tv -> readTVar tv+ MultiX xs -> do+ pairs <- mapM (readX time) xs+ let maxT = maximum (map snd pairs)+ return (map fst pairs, maxT) waitE :: E a -> IO a waitE e0 = do@@ -253,17 +294,41 @@ DropThis -> readEIO e NotNowNotEver -> hang +showEventResult :: EventResult a -> String+showEventResult r = case r of+ TryLater -> "TryLater"+ Normal _ t -> "Normal ? " ++ show t+ DropThis -> "DropThis"+ NotNowNotEver -> "NotNowNotEver"+ hang :: IO a hang = do threadDelay (100 * 10^(6::Int)) hang++unsafeNewPortX :: Context -> a -> (TVar (a, Time) -> IO ()) -> X a+unsafeNewPortX cx v0 workLoop = PortX v0 cx tv where+ tv = unsafePerformIO $ do+ out <- newTVarIO (v0, 0)+ threadId <- forkIO (workLoop out)+ modifyMVar_ (cxThreads cx) (return . (threadId:))+ return out++unsafeNewPortE :: Context -> (TChan (a, Time) -> IO ()) -> E a+unsafeNewPortE cx workLoop = PortE cx ch where+ ch = unsafePerformIO $ do+ out <- newBroadcastTChanIO+ threadId <- forkIO (workLoop out)+ modifyMVar_ (cxThreads cx) (return . (threadId:))+ return out+ --- -- | An event which gets the value of a signal when another event occurs. snapshot :: E a -> X b -> E (a,b) snapshot e x = SnapshotE e x --- | Like snapshot but ignores the original event's payload.+-- | Like 'snapshot' but ignores the original event's payload. snapshot_ :: E a -> X b -> E b snapshot_ e x = snd <$> snapshot e x @@ -295,80 +360,152 @@ voidE :: E a -> E () voidE e = () <$ e +-- | Value of a signal at time zero.+initialValue :: X a -> a+initialValue sig = case sig of+ PureX v -> v+ TimeX _ tmap -> tmap 0+ PortX v0 _ _ -> v0+ FmapX f x -> f (initialValue x)+ ApplX ff xx -> (initialValue ff) (initialValue xx)+ MappendX x1 x2 -> mappend (initialValue x1) (initialValue x2)+ MultiX xs -> map initialValue xs+ -- | Sum over events using an initial state and a state transition function.-accum :: E a -> s -> (a -> s -> s) -> X s-accum e0 s0 trans = case getContextE e0 of+accum :: s -> (a -> s -> s) -> E a -> X s+accum s0 trans e0 = case getContextE e0 of Nothing -> pure s0- Just cx -> PortX cx tv where- tv = unsafePerformIO $ do- state <- newTVarIO (s0, 0)- threadId <- forkIO $ do- e <- dupE e0- forever $ do- atomically $ do- 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+ Just cx -> unsafeNewPortX cx s0 $ \tv -> do+ e <- dupE e0+ forever $ do+ atomically $ do+ mx <- readE e+ case mx of+ TryLater -> retry+ DropThis -> return ()+ Normal x t -> do+ (s,_) <- readTVar tv+ let s' = trans x s+ writeTVar tv (s',t)+ NotNowNotEver -> error "impossible (2)" -- | 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)+trap x0 = accum 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 getContextX x of+-- | An event that occurs on "edges" detected in a signal. The signal will+-- be rasterized if necessary for this to make sense.+edge :: (a -> a -> Maybe b) -> X a -> E b+edge diff sig = case getContextX sig of Nothing -> never- Just cx -> PortE cx ch where- ch = unsafePerformIO $ do- out <- newBroadcastTChanIO- threadId <- forkIO $ do- (v0,_) <- atomically (readX 0 x)- ref <- newIORef v0- 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 -> 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+ Just cx -> unsafeNewPortE cx $ \out -> do+ let x = rasterize sig -- implicit rasterization ...+ ref <- newIORef (initialValue sig)+ forever $ do+ v <- readIORef ref+ (d, v', t) <- atomically $ do+ -- signal was rasterized so we don't need to fabricate any time here+ (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)) -chron :: UTCTime -> IO Double-chron epoch = do- now <- getCurrentTime- let time = diffUTCTime now epoch- return (realToFrac time)+showSignal :: X a -> String+showSignal sig = case sig of+ PureX _ -> "Pure ?"+ TimeX _ _ -> "TimeX ? ?"+ PortX _ _ _ -> "PortX ? ? ?"+ FmapX _ x -> "FmapX ? (" ++ showSignal x ++ ")"+ ApplX ff xx -> "ApplX ("++showSignal ff++") ("++showSignal xx++")"+ MappendX x1 x2 -> "Mappend ("++showSignal x1++") ("++showSignal x2++")"+ MultiX xs -> "MultiX ["++intercalate "," (map showSignal xs)++"]" -newInternalBoot :: Context -> IO (E (), IO ())-newInternalBoot cx = do- bch <- newBroadcastTChanIO- return- ( PortE cx bch- , atomically (writeTChan bch ((), 0)) )+-- | Rasterize a signal. If there are several edge tests on a continuous+-- signal then it's better to explicitly rasterize before.+rasterize :: X a -> X a+rasterize sig = case getContextX sig of+ Nothing -> sig+ Just cx -> case containsTimeX sig of+ False -> sig+ True -> unsafeNewPortX cx (initialValue sig) $ \tv -> do+ let period = 0.01+ counterRef <- newIORef 0+ forever $ do+ counter <- readIORef counterRef+ let target = counter * period+ atomically (readX target sig >>= writeTVar tv)+ now <- chron (cxEpoch cx)+ let newTarget = (counter+1) * period+ writeIORef counterRef (counter+1)+ let ms = ceiling ((newTarget - now) * million)+ threadDelay ms +containsTimeX :: X a -> Bool+containsTimeX x = case x of+ PureX _ -> False+ FmapX _ x' -> containsTimeX x'+ ApplX x1 x2 -> containsTimeX x1 || containsTimeX x2+ MappendX x1 x2 -> containsTimeX x1 || containsTimeX x2+ TimeX _ _ -> True+ PortX _ _ _ -> False+ MultiX xs -> or (map containsTimeX xs)++-- | From many signals, one.+multiplex :: [X a] -> X [a]+multiplex sigs = MultiX sigs++-- | Like 'delayE' but the amount of delay is determined on a per-event basis.+delayE' :: E (a, DeltaT) -> E a+delayE' src = case getContextE src of+ Nothing -> never+ Just cx -> unsafeNewPortE cx $ \out -> do+ e <- dupE src+ forever $ do+ ((v,dt), t) <- readEIO e+ let t' = t + max dt 0+ let io = atomically (writeTChan out (v, t'))+ cxDefer cx t' io++-- | Delay occurrences of an event.+delayE :: DeltaT -> E a -> E a+delayE delta e = delayE' (fmap (,delta) e)++-- | Shift a signal ahead in time.+delayX :: DeltaT -> X a -> X a+delayX delta = timeWarp' (subtract delta) (+delta)++-- | Slow down a signal by a factor.+dilate :: Double -> X a -> X a+dilate rate = timeWarp' (/rate) (*rate)++-- | Time warp a signal.+timeWarp :: (Time -> Time) -> X a -> X a+timeWarp g sig = case sig of+ PureX _ -> sig+ TimeX cx f -> TimeX cx (f . g)+ PortX _ _ _ -> error "timeWarp: Can't handle events. Try timeWarp'."+ FmapX f x -> FmapX f (timeWarp g x)+ ApplX ff xx -> ApplX (timeWarp g ff) (timeWarp g xx)+ MappendX x1 x2 -> MappendX (timeWarp g x1) (timeWarp g x2)+ MultiX xs -> MultiX (map (timeWarp g) xs)++-- | Like 'timeWarp' but works with events. The inverse of the warp function+-- must exist and be provided.+timeWarp' :: (Time -> Time)+ -> (Time -> Time) -- ^ inverse of warp+ -> X a+ -> X a+timeWarp' g ginv sig = case sig of+ PureX _ -> sig+ TimeX cx f -> TimeX cx (f . g)+ PortX v0 cx tv -> warpPortX v0 cx tv ginv+ FmapX f x -> FmapX f (timeWarp' g ginv x)+ ApplX ff xx -> ApplX (timeWarp' g ginv ff) (timeWarp' g ginv xx)+ MappendX x1 x2 -> MappendX (timeWarp' g ginv x1) (timeWarp' g ginv x2)+ MultiX xs -> MultiX (map (timeWarp' g ginv) xs)+ -- | 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 ())@@ -377,7 +514,7 @@ let epoch = cxEpoch cx tv <- setupIO (newTVarIO (v,0)) return- ( PortX cx tv+ ( PortX v cx tv , \x -> do now <- chron epoch atomically (writeTVar tv (x,now)))@@ -400,8 +537,8 @@ -- | 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+output :: (Time -> a -> IO ()) -> E a -> Setup ()+output act e0 = do cx <- getContext setupIO $ do e <- dupE e0@@ -422,18 +559,18 @@ return () -- | Runs a program defined by the setup computation. The simulation ends--- if the returned event occurs.+-- if the returned event occurs. The provided time signal is in units+-- of seconds with zero at the beginning of the simulation. runProgram :: (E Boot -> X Time -> Setup (E ())) -> IO () runProgram setup = do mv <- newMVar [] epoch <- getCurrentTime- let cx = Context mv epoch- let time = TimeX cx- (onBoot, boot) <- newInternalBoot cx+ defer <- newScheduler epoch mv+ let cx = Context mv epoch defer+ let time = TimeX cx id+ onBoot <- SingleE cx <$> newTMVarIO () let Setup act = setup onBoot time exit <- act cx- --threadDelay 5000- boot waitE exit _ <- withMVar mv (mapM killThread) return ()@@ -447,11 +584,11 @@ putStrLn (toString x) return e --- | Print out transitions in a signal. Only for debugging purposes.+-- | Print out transitions in a signal, rasterizing if necessary. Only for debugging purposes. 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 sig diff in+ let e = edge diff sig in unsafePerformIO $ do _ <- forkIO $ do e' <- dupE e@@ -460,3 +597,80 @@ putStrLn (toString x) return sig +chron :: UTCTime -> IO Double+chron epoch = do+ now <- getCurrentTime+ let time = diffUTCTime now epoch+ return (realToFrac time)++-- move changes in a signal to the future+warpPortX :: a -> Context -> TVar (a, Time) -> (Time -> Time) -> X a+warpPortX v0 cx srcTV ginv = unsafeNewPortX cx v0 $ \tv -> do+ latest <- newIORef 0+ forever $ do+ t <- readIORef latest+ (v,t') <- atomically $ do+ (v, t') <- readTVar srcTV+ if t' > t then return (v,t') else retry+ writeIORef latest t'+ let t'' = ginv t'+ let io = atomically (writeTVar tv (v,t''))+ -- at this point we can decide to drop events that were mapped into the past+ -- or to map them to the current time+ cxDefer cx (max t'' t') io++newScheduler :: UTCTime -> MVar [ThreadId] -> IO (Time -> IO () -> IO ())+newScheduler epoch threads = do+ seqMv <- newMVar Data.Sequence.empty+ wake <- newChan+ tid <- forkIO (dispatcher epoch wake seqMv)+ modifyMVar_ threads (return . (tid:))+ return $ \targetT io -> modifyMVar_ seqMv $ \queue -> do+ let (seqL, seqR) = spanl (\(t, _) -> t <= targetT) queue+ let seq' = seqL >< ((targetT, io) <| seqR)+ writeChan wake ()+ return seq'++dispatcher :: UTCTime -> Chan () -> MVar (Seq (Time, IO ())) -> IO ()+dispatcher epoch wake mv = forever $ do+ now <- chron epoch+ nextWake <- modifyMVar mv $ \queue -> do+ let (seqL, seqR) = spanl (\(t, _) -> t <= now) queue+ mapM_ snd (toList seqL)+ case viewl seqR of+ EmptyL -> return (seqR, Nothing)+ (t', _) :< _ -> return (seqR, Just t')+ case nextWake of+ Nothing -> readChan wake+ Just tNext -> do+ let ms = ceiling (min (tNext - now) 10 * million)+ cancel <- cancellableThread (threadDelay ms >> writeChan wake ())+ readChan wake+ cancel++million :: Double+million = toEnum (10^(6::Int))++cancellableThread :: IO () -> IO (IO ())+cancellableThread io = do+ mv <- newEmptyMVar+ tid <- forkIO $ do+ io+ _ <- tryTakeMVar mv+ return ()+ putMVar mv tid+ return $ do+ mtarget <- tryTakeMVar mv+ case mtarget of+ Nothing -> return ()+ Just target -> killThread target++sampleX :: X a -> Time -> a+sampleX sig t = case sig of+ PureX v -> v+ TimeX _ tmap -> tmap t+ PortX _ _ _ -> error "sampleX expects time-only signals"+ FmapX f x -> f (sampleX x t)+ ApplX ff xx -> (sampleX ff t) (sampleX xx t)+ MappendX x1 x2 -> mappend (sampleX x1 t) (sampleX x2 t)+ MultiX xs -> map (flip sampleX t) xs
broccoli.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: broccoli-version: 0.3.0.0+version: 0.4.0.0 synopsis: Small library for interactive functional programs. description: Small library for interactive functional programs. license: BSD3@@ -21,8 +21,10 @@ other-extensions: GADTs, RankNTypes build-depends: base >=4.7 && <4.8, time >= 1.4 && <1.6,+ containers >=0.5 && <0.6, stm >=2.4 && <2.5 -- hs-source-dirs: + ghc-options: -Wall default-language: Haskell2010 source-repository head@@ -32,4 +34,4 @@ source-repository this type: git location: https://github.com/evanrinehart/broccoli- tag: 0.3.0.0+ tag: 0.4.0.0