broccoli 0.1.0.0 → 0.2.0.0
raw patch · 2 files changed
+166/−106 lines, 2 files
Files
- Control/Broccoli.hs +164/−104
- broccoli.cabal +2/−2
Control/Broccoli.hs view
@@ -5,21 +5,22 @@ module Control.Broccoli ( X, E,- Output(..),- runProgram,- edge,- accumulate,+ never, snapshot, snapshot_,- out,- filterE,+ accumulate,+ edge, justE, maybeE,- never,- debugX,- debugE,+ filterE,+ Setup,+ runProgram, newX,- newE+ newE,+ input,+ output,+ debugX,+ debugE ) where import Control.Applicative@@ -38,7 +39,7 @@ 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 :: TVar a -> X a+ PortX :: MVar [ThreadId] -> TVar a -> X a -- | An event that carries values of type a when it occurs. data E a where@@ -48,10 +49,7 @@ ProductE :: (b -> c -> a) -> E b -> E c -> E a SnapshotE :: E b -> X a -> E a JustE :: E (Maybe a) -> E a- PortE :: TChan a -> E a---- | A special IO action which will react to events. See 'out'.-newtype Output = Output (IO ())+ PortE :: MVar [ThreadId] -> TChan a -> E a instance Functor X where fmap f x = FmapX f x@@ -67,6 +65,47 @@ mempty = NeverE mappend e1 e2 = MappendE e1 e2 +-- | A monad for hooking up inputs and outputs to a program.+data Setup a = Setup (MVar [ThreadId] -> IO a)++instance Monad Setup where+ return x = Setup (\_ -> return x)+ (Setup r) >>= f = Setup r' where+ r' mv = do+ x <- r mv+ let Setup r'' = f x+ r'' mv++instance Applicative Setup where+ pure = return+ (<*>) = ap++instance Functor Setup where+ fmap f (Setup io) = Setup (\mv -> f <$> io mv)++setupIO :: IO a -> Setup a+setupIO io = Setup (\_ -> io)++getThreads :: Setup (MVar [ThreadId])+getThreads = Setup (\mv -> return mv)++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++getThreadsX :: X a -> Maybe (MVar [ThreadId])+getThreadsX x = case x of+ PureX _ -> Nothing+ FmapX _ x' -> getThreadsX x'+ ApplX x1 x2 -> getFirst $ First (getThreadsX x1) <> First (getThreadsX x2)+ PortX mv _ -> Just mv+ dupE :: E a -> IO (E a) dupE e = case e of NeverE -> return NeverE@@ -77,9 +116,6 @@ e1' <- dupE e1 e2' <- dupE e2 return (MappendE e1' e2')- PortE ch -> do- ch' <- atomically (dupTChan ch)- return (PortE ch') ProductE f e1 e2 -> do e1' <- dupE e1 e2' <- dupE e2@@ -90,11 +126,13 @@ JustE e' -> do e'' <- dupE e' return (JustE e'')+ PortE mv ch -> do+ ch' <- atomically (dupTChan ch)+ return (PortE mv ch') readE :: E a -> IO a readE e = case e of NeverE -> hang- PortE ch -> atomically (readTChan ch) MappendE e1 e2 -> race (readE e1) (readE e2) FmapE f e' -> f <$> readE e' ProductE f e1 e2 -> do@@ -109,6 +147,7 @@ case m of Nothing -> loop Just x -> return x+ PortE _ ch -> atomically (readTChan ch) readX :: X a -> STM a readX x = case x of@@ -118,7 +157,7 @@ f <- readX ff x <- readX xx return (f x)- PortX tv -> readTVar tv+ PortX _ tv -> readTVar tv hang :: IO a hang = do@@ -130,31 +169,8 @@ e <- dupE e0 readE e -runEvent :: E a -> (a -> IO ()) -> IO ()-runEvent e0 act = do- e <- dupE e0- forever (readE e >>= act)- --- --- | Create a new signal with an initial value. Get back an IO action to--- change the value of the signal. This is the only way the signal will--- change.-newX :: a -> IO (a -> IO (), X a)-newX v0 = do- tv <- newTVarIO v0- return- ( \x -> atomically (writeTVar tv x)- , PortX tv )---- | Create a new event and the associated IO action to trigger the event.-newE :: IO (a -> IO (), E a)-newE = do- ch <- newBroadcastTChanIO- return- ( \x -> atomically (writeTChan ch x)- , PortE ch )- -- | 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)@@ -179,70 +195,114 @@ never :: E a never = mempty --- | Events will occur when an "edge" is detected in a signal. The detection--- algorithm checks the two values before and after a discrete change in--- the signal.+-- | 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+ Nothing -> pure s0+ Just mv -> PortX mv tv where+ tv = unsafePerformIO $ do+ state <- newTVarIO s0+ threadId <- forkIO $ do+ putStrLn "accum forked"+ 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:))+ 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. edge :: X a -> (a -> a -> Maybe b) -> E b-edge x diff = PortE ch where- ch = unsafePerformIO $ do- out <- newBroadcastTChanIO- forkIO $ do- v0 <- atomically (readX x)- ref <- newIORef v0- forever $ do- v <- readIORef ref- (d, v') <- atomically $ do- v' <- readX x- case diff v v' of- Just d -> return (d, v')- Nothing -> retry- writeIORef ref v'- atomically (writeTChan out d)- return out+edge x diff = case getThreadsX x of+ Nothing -> never+ Just mv -> PortE mv ch where+ ch = unsafePerformIO $ do+ out <- newBroadcastTChanIO+ threadId <- forkIO $ do+ putStrLn "edge forked"+ v0 <- atomically (readX x)+ ref <- newIORef v0+ forever $ do+ v <- readIORef ref+ (d, v') <- atomically $ do+ v' <- readX x+ case diff v v' of+ Just d -> return (d, v')+ Nothing -> retry+ writeIORef ref v'+ atomically (writeTChan out d)+ modifyMVar_ mv (return . (threadId:))+ return out --- | Create a signal out of an input event and a state machine.-accumulate :: E a -> s -> (a -> s -> s) -> X s-accumulate e0 s0 trans = PortX tv where- tv = unsafePerformIO $ do- state <- newTVarIO s0- forkIO $ do- e <- dupE e0- forever $ do- x <- readE e- atomically $ do- s <- readTVar state- let s' = trans x s- writeTVar state s'- return state --- | A handler for events.-out :: E a -> (a -> IO ()) -> Output-out e0 act = Output (runEvent e0 act)+-- | Creates a new event and an IO action to trigger it.+newE :: Setup (E a, a -> IO ())+newE = do+ mv <- getThreads+ bch <- setupIO newBroadcastTChanIO+ return (PortE mv bch, atomically . writeTChan bch) --- | Prints the values of a signal as they change.-debugX :: (Eq a, Show a) => X a -> Output-debugX x = Output $ do- v0 <- atomically (readX x)- print v0- let diff a b = if a == b then Nothing else Just b- runEvent (edge x diff) print+-- | 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) --- | Prints the values of events as they occur.-debugE :: (Show a) => E a -> Output-debugE e = out e print+-- | Spawn a thread to execute an action for each event occurrence.+output :: E a -> (a -> IO ()) -> Setup ()+output e0 act = do+ mv <- getThreads+ setupIO $ do+ e <- dupE e0+ tid <- (forkIO . forever) (readE e >>= act)+ modifyMVar_ mv (return . (tid:))+ return () --- | Run a set of Outputs. This spawns several threads then waits for an--- event. The output threads will then be killed to stop further processing.--- However other threads which no longer have any effect will probably remain,--- taking up resources. After threads have been spawned but before waiting,--- the given "boot" action will be executed.-runProgram :: IO () -- ^ action to execute after initialization- -> E () -- ^ event that will shutdown the system- -> [Output] -- ^ set of output event handlers to run- -> IO ()-runProgram notifyBoot stop outs = do- tids <- forM outs (\(Output io) -> forkIO io)- threadDelay 5000- notifyBoot- waitE stop- forM_ tids killThread+-- | Spawn an input thread to generate source signals and events.+input :: IO () -> Setup ()+input handler = do+ mv <- getThreads+ setupIO $ do+ tid <- forkIO handler+ modifyMVar_ mv (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+ mv <- newMVar []+ (boot, exit) <- setup mv+ --threadDelay 5000+ boot+ waitE exit+ 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+ e' <- dupE e+ (forkIO . forever) (readE e' >>= print)+ return e++-- | Print out transitions in a signal. Only for debugging purposes.+debugX :: (Eq a, Show a) => X a -> X a+debugX x =+ let diff a b = if a == b then Nothing else Just (a,b) in+ let e = edge x diff in+ unsafePerformIO $ do+ forkIO $ do+ putStrLn "edge forked"+ e' <- dupE e+ forever (readE e' >>= print)+ return x
broccoli.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: broccoli-version: 0.1.0.0+version: 0.2.0.0 synopsis: Small library for interactive functional programs. description: Small library for interactive functional programs. license: BSD3@@ -32,4 +32,4 @@ source-repository this type: git location: https://github.com/evanrinehart/broccoli- tag: 0.1.0.0+ tag: 0.2.0.0