diff --git a/Dflow.cabal b/Dflow.cabal new file mode 100644 --- /dev/null +++ b/Dflow.cabal @@ -0,0 +1,51 @@ +name: Dflow +version: 0.0.1 +cabal-version: >= 1.10 +build-type: Simple +license: BSD3 +license-file: LICENSE.txt +copyright: © Paul Johnson 2012 +author: Paul Johnson +maintainer: +category: Reactivity +tested-with: GHC==7.0.4 +synopsis: Processing Real-time event streams +data-files: README.txt +description: This library provides Real Time Stream Processors (RTSPs). An RTSP + transforms an input event stream into an output event stream. The output + events occur asynchronously with input events. RTSPs can be composed into + pipelines or executed in parallel and their outputs merged. A Real Time + Action (RTA) monad is provided for creating new primitive RTSPs. +stability: Experimental + +library + hs-source-dirs: src + build-depends: + base >= 4 && < 5, + time >=1.1, + stm, + QuickCheck >= 2.4, + containers >= 0.4 + exposed-modules: Control.RTSP + ghc-options: -fspec-constr-count=6 + default-language: Haskell2010 + +test-suite ArbTest + type: exitcode-stdio-1.0 + x-uses-tf: true + build-depends: + base >= 4, + HUnit >= 1.2 && < 2, + QuickCheck >= 2.4, + test-framework >= 0.4.1, + test-framework-quickcheck2 + hs-source-dirs: src, test + ghc-options: -Wall -threaded -rtsopts=all -fspec-constr-count=10 + -- Add "-fhpc" to the previous line to enable test coverage + default-language: Haskell2010 + other-modules: + ArbTest, + Control.RTSP, + Main + + main-is: Main.hs diff --git a/LICENSE.txt b/LICENSE.txt new file mode 100644 --- /dev/null +++ b/LICENSE.txt @@ -0,0 +1,26 @@ +Copyright (c) 2012, Paul Johnson +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + Redistributions in binary form must reproduce the above copyright + notice, this list of conditions and the following disclaimer in + the documentation and/or other materials provided with the + distribution. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.txt b/README.txt new file mode 100644 --- /dev/null +++ b/README.txt @@ -0,0 +1,82 @@ +Dflow +===== + +Real time event processing using data flow declarations. + +The following has been tested using ghc 7.0.4 under Fedora 15 64-bit. + +Installation +------------ + +$ cabal configure +$ cabal build +$ cabal install + + +Tests +----- + +$ cabal configure --enable-test +$ cabal build +$ cabal test + +The tests work by building random networks of event processors and +checking them against an equivalent list implementation. Occasionally +they seem to hit a pathological case that chews up memory and CPU +time. If memory usage goes over 1GB then kill the test and try again. + +Test Coverage +------------- + +Edit the Dflow.cabal file and add "-fhpc" to the GHC options for the +test (see comment in the file). Then do "cabal clean" and follow the +instructions for "Tests" above. Then do: + +$ hpc markup ArbTest.tix --destdir=coverage + +The coverage report is in coverage/hpc_index.html. + +Concepts +-------- + +The main data types for the application programmer are: + +Event: A value that occurs at a certain time. For instance an +"Event Char" might represent a key press. + +RTSP: The Real Time Stream Processor. A value of type "RTSP x y" +takes in events of type "x" and emits events of type "y". RTSPs can +be strung together into pipelines using "." (or ">>>" if you prefer +your data to flow left-to-right). RTSPs are also Monoids, so you can +fork your data through two parallel RTSPs and then merge the results. + +RTA: A monad for building stateful RTSPs. Convert an RTA into an +RTSP using "execRTA" or "accumulateRTA" depending what you want to +do with pending output events when a new input event arrives. + + +You can test an RTSP in "fast time" (that is, without waiting for +real-time delays) by using "simulateRTSP". However the argument list +of input events must be finite. Then you can execute the RTSP in real +time using "execRTSP" and be confident that the real time behaviour +will match the fast-time behaviour. + +To Do +----- + +Events are currently sorted into time order using (in effect) an +O(n) insertion sort. Event streams should use something more +sophisticated internally, such as a heap. + +What API, if any, should be exposed for event streams? On one hand +developers should be able to create new kinds of RTSP primitives +using event streams, but on the other hand anything you can do with +event streams can be done using RTA anyway. + +Figure out some way for RTAs and RTSPs to execute non-blocking IO with +timeouts, probably using continuations in some way. Hence develop true +distribution by having RTSPs on different machines interact in a +declarative manner. + +Make state persistent, together with a replay mechanism for lost +events. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/src/Control/RTSP.hs b/src/Control/RTSP.hs new file mode 100644 --- /dev/null +++ b/src/Control/RTSP.hs @@ -0,0 +1,666 @@ +{- | + +Real Time Stream Processors are used to describe pipelines that process events in real time. An event is described as a +(Time, Value) pair. When an RTSP receives an event it can respond by emitting zero or more events at any time at or after the +receipt of the original event. Further incoming events may influence the stream of emitted events. + +Piplelines of RTSPs can be built up using the '(.)' operator from the Category instance, or alternatively the '(>>>)' operator +(which is merely dot with its arguments swapped). RTSPs can be run in parallel with their outputs merged using "mappend" +from the Monoid instance. + +Within the RTSP implementation all notions of \"delay\" and \"time\" merely refer to the time component of events, and are used +for event ordering. Only the 'execRTSP' function, which runs in the IO monad, executes any actual real-time delay. + +The main data types for the application programmer are: + +['Event'] A value that occurs at a certain time. For instance an @'Event' Char@ might represent a key press. + +['RTSP'] The Real Time Stream Processor. A value of type @'RTSP' x y@ takes in events of type @x@ and emits events of type @y@. +RTSPs can be strung together into pipelines using @(.)@ (or @(>>>)@ if you prefer your data to flow left-to-right). RTSPs are also +monoids, so you can fork your data through two parallel RTSPs and then merge the results. + +['RTA'] A monad for building stateful RTSPs. Convert an 'RTA' into an 'RTSP' using 'execRTA' or 'accumulateRTA' depending what you +want to do with pending output events when a new input event arrives. + +You can test an RTSP in \"fast time\" (that is, without waiting for real-time delays) by using 'simulateRTSP'. Then you can +execute the RTSP in real time using 'execRTSP' and be confident that the real time behaviour will match the fast-time behaviour. + +/Simultaneous Events/ + +The handling of logically simultaneous events in discrete event simulation is a long-standing problem. The three basic approaches +are: + +1. Impose an arbitrary but deterministic order on \"simultaneous\" events. + +2. Collect the simultaneous events and pass them to the application, on the basis that the application programmer can then + impose the appropriate semantics. + +3. Simulate all possible orderings. + +This library takes the first approach. Option 2 would force each RTSP to wait for the next event to see if it was +simultaneous, which is possible in a simulator but not in a real time system. In a real time system option 3 is not feasible, +and would still leave the problem of which ordering to present to the outside world as the \"real\" one. + +When two simultaneous events arrive at an RTSP, the current implementation uses the following rules: + +* Simultaneous output events retain the order of the input events that triggered them. Hence simultaneous events never \"overtake\". + +* In the case of @(id \`mappend\` stream (+ 1))@ the output alternates the left and right expressions, starting with the left. + +However these properties interact in ways that are complex, hard to define formally and not guaranteed to be stable. Code +that depends on the ordering of simultaneous events should therefore be avoided. +-} + +module Control.RTSP ( + -- ** Events + Event (..), + isBefore, + -- ** Real Time Stream Processors + EventStream, + emitsBefore, + nullStream, + esFinished, + esPeek, + esFutures, + esProcess, + esMerge, + RTSP (..), + simulateRTSP, + execRTSP, + stream, + accumulate, + -- ** Manipulating event times + repeatEvent, + delay0, + delay, + -- ** Conditional event processing + Cond, + streamFilter, + cond, + cond1, + ifThenElse, + -- ** Real Time Actions with state + RTA, + get, + put, + modify, + emit, + pause, + now, + execRTA, + accumulateRTA, +) where + + + +import Control.Category +import Control.Concurrent +import Control.Concurrent.STM +import Control.Monad +import Data.List +import Data.Monoid +import Data.Sequence +import Data.Time + +import Prelude hiding ((.),id, repeat) + + +infix 4 `isBefore`, `emitsBefore` + + + +-- | Real time events. +data Event a = Event {eventTime :: UTCTime, eventValue :: a} + deriving (Show, Eq) + +instance Functor Event where + fmap f (Event t v) = Event t (f v) + + +-- | True if the first event occurs strictly before the second. This makes @Event@ a poset (partially ordered set). +-- Infix priority 4 (the same as other comparison operators). +isBefore :: Event a -> Event b -> Bool +isBefore ev1 ev2 = eventTime ev1 < eventTime ev2 + + +{- | +A real-time event stream cannot be described without reference to +unknown future inputs. Hence @EventStream@ embodies two possible futures: + +* An @Event c@ will be emitted at some time in the future, with a new + @EventStream@ representing the future after that event. + +* An incoming @Event b@ will arrive before the next @Event c@ is emitted, + creating a new @EventStream@ representing the response to that event. The + old @Event c@ may or may not be part of the new @EventStream@. + +There are also two degenerate cases: + +* Wait: no event is scheduled to be emitted, and the @EventStream@ + just waits for an incoming event. + +* Finish: no event will ever be emitted, regardless of incoming events. + This is explicitly distinguished so that complex RTSP expressions + can be GC'd if they can be proven to be finished. + +Event streams are like the Mirror of Galadriel, for they show things that +were, things that are, and things that yet may be. But which it is that he +sees, even the wisest cannot always tell. + +Seeing is both good and perilous. An event stream may be modified by +new events, but exceptions or inconsistent results will occur if the incoming +events are not in increasing order of time. +-} +data EventStream b c = + Emit (Event c) (EventStream b c) (RTSP b c) + -- ^ The next event to be emitted, the following EventStream, and the function to + -- handle an incoming event before then. + | Wait (RTSP b c) + -- ^ Degenerate case: no event scheduled to be emitted. + | Finish + -- ^ Semantically equivalent to @Wait eventSink@, but allows completed streams to be GC'd. + deriving (Show) + + +-- | Peek at the events that will be emitted by this EventStream if no incoming event interrupts them. +esPeek :: EventStream b c -> [Event c] +esPeek (Emit ev es1 _) = ev : esPeek es1 +esPeek _ = [] + + +-- | True if the first argument is scheduled to emit an event before the second. This makes @EventStream@ a poset +-- (partially ordered set). Infix priority 4. +emitsBefore :: EventStream b1 c1 -> EventStream b2 c2 -> Bool +emitsBefore Finish _ = False +emitsBefore (Wait _) _ = False +emitsBefore (Emit ev1 _ _) (Emit ev2 _ _) = ev1 `isBefore` ev2 +emitsBefore (Emit _ _ _) _ = True + + +-- Only events satisfying the predicate will be passed on. +esFilter :: (b -> Bool) -> EventStream b b +esFilter = Wait . rtspFilter + + +-- | Only events satisfying the predicate will be passed on. +rtspFilter :: (b -> Bool) -> RTSP b b +rtspFilter p = RTSP $ \ev -> if p $ eventValue ev then Emit ev (esFilter p) (rtspFilter p) else esFilter p + + +-- | All the possible futures of the event stream. +esFutures :: EventStream b c -> [(Event c, EventStream b c)] +esFutures (Emit e es1 _) = (e, es1) : esFutures es1 +esFutures _ = [] + + +-- | True if the event stream is guaranteed not to emit any future events, regardless of input. +esFinished :: EventStream b c -> Bool +esFinished Finish = True +esFinished _ = False + + +-- | Merge the outputs of two event streams. Input events are delivered +-- to both streams. +esMerge :: EventStream b c -> EventStream b c -> EventStream b c +esMerge Finish es = es +esMerge es Finish = es +esMerge (Wait k1) (Wait k2) = + Wait (splitRTSP k1 k2) +esMerge (Emit e es1 k1) es2@(Wait k2) = + Emit e (esMerge es1 es2) (splitRTSP k1 k2) +esMerge es1@(Wait k1) (Emit e es2 k2) = + Emit e (esMerge es1 es2) (splitRTSP k1 k2) +esMerge es1@(Emit e1 es1a k1) es2@(Emit e2 es2a k2) = + if e2 `isBefore` e1 + then Emit e2 (esMerge es1 es2a) (splitRTSP k1 k2) + else Emit e1 (esMerge es1a es2) (splitRTSP k1 k2) + + +-- | Given a new input event to an existing event stream, this returns the modified event stream. When @esProcess@ +-- is called on the result the Event argument to the second call must not occur before the first (they can be +-- simultaneous). More formally, if +-- +-- > esOut = esProcess (esProcess esIn ev1) ev2 +-- +-- then @not (ev2 `isBefore` ev1)@. This precondition is not checked. +esProcess :: Event b -> EventStream b c -> EventStream b c +esProcess _ Finish = Finish +esProcess ev (Wait k) = runRTSP k ev +esProcess eIn (Emit eOut rest k) = + if eIn `isBefore` eOut + then runRTSP k eIn + else Emit eOut (esProcess eIn rest) k + + + +-- | An event stream that never generates anything. +nullStream :: EventStream b c +nullStream = Finish + + +-- | Real Time Stream Processor (RTSP) +-- +-- An EventStream cannot exist independently of some event that caused it to start. Hence the only way to +-- create an EventStream is through an RTSP. +-- +-- * "mempty" is the event sink: it never emits an event. +-- +-- * "mappend" runs its arguments in parallel and merges their outputs. +-- +-- * "id" is the null operation: events are passed through unchanged. +-- +-- * "(.)" is sequential composition: events emitted by the second argument are passed to the first argument. +newtype RTSP b c = RTSP {runRTSP :: Event b -> EventStream b c} + + +instance Show (RTSP b c) where show _ = "-RTSP-" + + +instance Monoid (RTSP b c) where + mempty = eventSink + mappend = splitRTSP + + +instance Monoid (EventStream b c) where + mempty = nullStream + mappend = esMerge + + +instance Functor (EventStream b) where + fmap f (Emit eOut rest k) = Emit (fmap f eOut) (fmap f rest) (fmap f k) + fmap f (Wait k) = Wait (fmap f k) + fmap _ Finish = Finish + + +instance Functor (RTSP b) where + fmap f (RTSP r) = RTSP $ \evt -> fmap f $ r evt + + +{- +The (.) operator for EventStream has to deal with several scenarios: + +1: (Wait k2) . (Wait k1). This is simple because the only possible event + is an input that is piped into k1. The result is the composition of the result with + (Wait k2), achieived using the instance for RTSP. + +2: (Wait k2) . es1@(Emit e1 es1a k1). In this case there are two timelines: + a) e1 : (k2 e1) . es1a -- Event e1 is passed to k2 and the result composed with es1a. + b) ev e1: (Wait k2) . (k1 ev) -- e1 never happens. Instead the input is passed to k1, which + generates a new event stream to be composed with k2. + + The code for case b is the "k" value in the "let" clause. + +3: es2@(Emit e2 es2a k2) es1@(Emit e1 es1a k1). The logic here is similar to scenario + 2, except that the timing of e2 has to be taken into account as well. There are five basic + timelines. Let ev = the next input event and es2b = runRTSP k2 e1 + + a) e2 e1 : Emit e2 (es2a . es1a) (k2 . k1) + b) e1 e2 : k2 e1 . es1a -- e2 never happens because it is overridden by e1 + c) ev ... : es2 . (k1 ev) -- ev overrides e1, and the new output is fed to es2. + d) e1 ev e2 : esProcess (es2b . es1a) ev -- e1 overrides e2, giving es1a and es2b to process ev. + e) e2 ev e1 : Emit e2 (es2a . (k1 ev)) -- e2 is emitted, then ev overrides e1. +-} +instance Category EventStream where + -- id :: EventStream b c + id = Wait id + + -- (.) :: EventStream c d -> EventStream b c -> EventStream b d + Finish . _ = Finish + (Wait _) . Finish = Finish + (Emit e es1 _). Finish = let future = Emit e (es1 . Finish) (RTSP $ \_ -> future) in future + (Wait k2) . (Wait k1) = Wait $ k2 . k1 + es2@(Wait k2) . Emit e1 es1a k1 = + let k = RTSP $ \ev -> + if ev `isBefore` e1 + then es2 . runRTSP k1 ev -- Timeline 2b + else esProcess ev es -- Timeline 2a + es = (runRTSP k2 e1) . es1a -- Future if ev never happens + in case es of + Emit e2 es2b _ -> Emit e2 es2b k + Wait _ -> Wait k + Finish -> Wait k -- if ev `isBefore` e1 then es never happens. + es2@(Emit e2 es2a _) . es1@(Wait k1) = + Emit e2 (es2a . es1) (RTSP $ \ev -> es2 . runRTSP k1 ev) + es2@(Emit e2 es2a k2) . es1@(Emit e1 es1a k1) = + let + es = (runRTSP k2 e1) . es1a + in if e1 `isBefore` e2 + then -- Timelines 3b, 3c and 3d. e2 never happens. + let k = RTSP $ \ev -> + if ev `isBefore` e1 + then es2 . runRTSP k1 ev -- Timeline 3c + else esProcess ev es -- Timeline 3d + in case es of + Emit e3 es3 _ -> Emit e3 es3 k + Wait _ -> Wait k + Finish -> Wait k -- As above. + else -- Timelines 3a, 3c and 3e. + let k = RTSP $ \ev -> es2 . runRTSP k1 ev + in Emit e2 (es2a . es1) k + + + +instance Category RTSP where + -- id :: RTSP b c + id = RTSP $ \ev -> Emit ev id id + -- (.) :: RTSP c d -> RTSP b c -> RTSP b d + r2 . r1 = RTSP $ \e0 -> (Wait r2) . runRTSP r1 e0 + + + +-- | Execute an RTSP against a list of events. Useful for testing. +simulateRTSP :: RTSP b c + -- ^ The processor to execute. + -> [Event b] + -- ^ The events must be finite and in chronological order. This is unchecked. + -> [Event c] +simulateRTSP r = esPeek . foldl (flip esProcess) (Wait r) + + + +-- | Execute an RTSP in the IO monad. The function returns immediately with an action for pushing events into the RTSP. +execRTSP :: + RTSP b (IO ()) + -- ^ The output of the RTSP is a series of action events that will be executed in a separate thread sequentially at the + -- times given. The actions may, of course, fork their own threads as necessary. + -- + -- execRTSP uses 'atomically', so it cannot be called within 'unsafePerformIO'. + -> IO (b -> IO ()) +execRTSP r = do + eventQ <- newTChanIO + let + + putValue v = do + t <- getCurrentTime + atomically $ writeTChan eventQ $ Event t v + + execStream (Emit ev es r1) = do + {- + "c1" and "c2" are threads that race to put a value in "var". "c1" waits until + the next event emission time and then puts "Nothing". "c2" waits for the next + input on "eventQ" and puts "Just" the event. Once this happens "mev2" can get + the result and both threads are killed. + + The tricky bit is avoiding a race condition in "c2" where it reads the + channel and is then killed by the timeout, which would result in a dropped event. + "var" is never emptied, so if "c1" wins the race then "c2" can never complete + before being killed, so the event remains on the queue ready for the next race. + -} + var <- newEmptyTMVarIO + c1 <- timeout (eventTime ev) (atomically $ putTMVar var Nothing) + c2 <- forkIO $ atomically $ do + ev1 <- readTChan eventQ + putTMVar var $ Just ev1 + + mev2 <- atomically $ readTMVar var + killThread c1 + killThread c2 + case mev2 of + Just ev2 -> execStream $ runRTSP r1 ev2 + Nothing -> do + () <- eventValue ev + execStream es + execStream (Wait r1) = do + ev2 <- atomically $ readTChan eventQ + execStream $ runRTSP r1 ev2 + execStream Finish = return () + + _ <- forkIO $ execStream $ Wait r + return putValue + + +-- Execute the given action at the given time. Returns immediately with the ThreadID that will execute the action. +timeout :: UTCTime -> IO () -> IO ThreadId +timeout t action = forkIO $ do + t0 <- getCurrentTime + longDelay (round ((t `diffUTCTime` t0) * 1000000)) + action + where + -- threadDelay takes an Int, which may be as small as 2^29 (a bit over 5 minutes). + longDelay :: Integer -> IO () + longDelay dt = + let (n,dt1) = if dt > 0 then dt `divMod` cycleT else (0,0) + in do + replicateM_ (fromIntegral n) $ threadDelay (fromIntegral cycleT) + threadDelay (fromIntegral dt1) + cycleT = 500000000 -- 500 seconds in uSec. Small enough to fit into an Int. + + +-- | An RTSP that never emits events regardless of its inputs. +eventSink :: RTSP b c +eventSink = RTSP $ \_ -> nullStream + + +-- | A pure function converted into a stream processor +stream :: (b -> c) -> RTSP b c +stream f = fmap f id + + + +-- | Deliver an event to two stream processors and merge the resulting event +-- streams. +splitRTSP :: RTSP b c -> RTSP b c -> RTSP b c +splitRTSP (RTSP r1) (RTSP r2) = RTSP $ \evt -> esMerge (r1 evt) (r2 evt) + where + + + +-- | Convert an list of events into an event stream. Events coming into this +-- stream are ignored. The list must be in chronological order. +streamFromList :: [Event c]-> EventStream b c +streamFromList [] = Finish +streamFromList (e:es) = + Emit e (streamFromList es) (RTSP $ \_ -> streamFromList (e:es)) + + +-- | When a new input event is delivered to an RTSP it causes any future output events to be dropped in favour of the new +-- events. @accumulate@ instead keeps the events from previous inputs interleaved with the new ones. If you use +-- this unnecessarily then you will get duplicated events. +-- +-- If there are @n@ output events due to be emitted before an input event then this will require O(n) time for the input. +accumulate :: RTSP b c -> RTSP b c +accumulate r = rAccum [] r + where + rAccum evs@(ev1:evs1) r1 = RTSP $ \ev2 -> + if ev2 `isBefore` ev1 -- hpc says this is always true. Can this be proved? + then sAccum evs $ runRTSP r1 ev2 + else Emit ev1 (sAccum evs1 $ runRTSP r ev2) (rAccum evs r1) + rAccum [] r1 = RTSP $ \ev -> sAccum [] $ runRTSP r1 ev + sAccum evs1@(ev1:evs1a) es2@(Emit ev2 es2a r2a) = + let + future = + if ev2 `isBefore` ev1 + then Emit ev2 (sAccum evs1 es2a) (rAccum (esPeek future) r2a) + else Emit ev1 (sAccum evs1a es2) (rAccum (esPeek future) r2a) + in future + sAccum [] es2@(Emit ev2 es2a r2) = Emit ev2 (sAccum [] es2a) (rAccum (esPeek es2) r2) + sAccum evs1@(ev1:evs1a) k2@(Wait r2) = Emit ev1 (sAccum evs1a k2) (rAccum evs1 r2) + sAccum [] (Wait r2) = Wait $ accumulate r2 + sAccum evs Finish = streamFromList evs + + +-- | Repeat each input event after the specified delays until a new event arrives, at which point the sequence begins again +-- with the new event value. The list of delays must not be negative and must be in ascending order. All the delays are +-- relative to the first event. +-- +-- Be careful when using list comprehensions to create the argument. A list like +-- +-- > [1..5] :: NominalDiffTime +-- +-- will count up in picoseconds rather than seconds, which is probably not what is wanted. Instead use +-- +-- > map fromInteger [1..5] :: NominalDiffTime +repeatEvent :: [NominalDiffTime] -> RTSP b b +repeatEvent dts1 = RTSP $ \(Event t0 v) -> + let + rStream dt0 (dt:dts2) + | dt0 <= dt = Emit (Event (dt `addUTCTime` t0) v) (rStream dt dts2) (repeatEvent dts1) + | otherwise = error "Control.Applicative.RTSP.streamRepeat: negative time increment." + rStream _ [] = Wait (repeatEvent dts1) + in rStream 0 dts1 + + +-- | Delay input events by the specified time, but given an event stream @{ev1, ev2, ev3...}@, if ev2 arrives before +-- ev1 has been emitted then ev1 will be lost. +delay0 :: NominalDiffTime -> RTSP b b +delay0 dt = repeatEvent [dt] + + +-- | Delay input events by the specified time. +-- +-- Unfortunately this requires O(n) time when there are @n@ events queued up due to the use of "accumulate". +delay :: NominalDiffTime -> RTSP b b +delay = accumulate . delay0 + + +-- | A conditional stream: events matching the predicate will be passed to the stream. +type Cond a b = (a -> Bool, RTSP a b) + + +-- | Conditional stream execution: only certain events will be accepted. +streamFilter :: Cond a b -> RTSP a b +streamFilter (p, r1) = rtspFilter p >>> r1 + + + +-- | Send each event to all the streams that accept it. +cond :: [Cond a b] -> RTSP a b +cond = mconcat . map streamFilter + + +-- | Send each event to the first stream that accepts it, if any. +cond1 :: [Cond a b] -> RTSP a b +cond1 = foldr ifThenElse eventSink + + +-- | Send each event to the conditional stream if it accepts it, otherwise send it to the second argument. +-- +-- @ifThenElse (p, rThen) rElse@ is equivalent to +-- +-- > streamFilter (p, rThen) `mappend` streamFilter (not . p, rElse) +-- +-- However @ifThenElse@ only evaluates @p@ once for each input event. +ifThenElse :: Cond a b -> RTSP a b -> RTSP a b +ifThenElse (p,rThen) rElse = ifRTSP (Wait rThen) (Wait rElse) + where + ifRTSP es1 es2 = RTSP $ \ev -> + if p $ eventValue ev + then ifStream (esProcess ev es1) es2 + else ifStream es1 (esProcess ev es2) + ifStream Finish Finish = Finish + ifStream Finish es@(Wait _) = Wait $ ifRTSP Finish es + ifStream Finish es@(Emit e es1 _) = Emit e (ifStream Finish es1) (ifRTSP Finish es) + ifStream es@(Wait _) Finish = Wait $ ifRTSP es Finish + ifStream es@(Emit e es1 _) Finish = Emit e (ifStream es1 Finish) (ifRTSP es Finish) + ifStream es1@(Wait _) es2@(Wait _) = + Wait $ ifRTSP es1 es2 + ifStream es1@(Emit e es1a _) es2@(Wait _) = + Emit e (ifStream es1a es2) (ifRTSP es1 es2) + ifStream es1@(Wait _) es2@(Emit e es2a _) = + Emit e (ifStream es1 es2a) (ifRTSP es1 es2) + ifStream es1@(Emit e1 es1a _) es2@(Emit e2 es2a _) = + if e1 `isBefore` e2 + then Emit e1 (ifStream es1a es2) (ifRTSP es1 es2) + else Emit e2 (ifStream es1 es2a) (ifRTSP es1 es2) + + +-- | Real-time Actions. This monad is used to build sequential processors that can be turned into stream processors. +-- An RTA emits zero or more events in response to each input event, and has a state that persists from one event to the next. +-- In particular, state changes made after a "pause" will be visible to the next event regardless of the relative times. +newtype RTA s c v = RTA {unRTA :: s -> Seq (Event c) -> UTCTime -> (v, s, Seq (Event c), UTCTime)} + +{- +In the RTA definition code, the following initial variable letters are used: + + b - The type of input events + c - The type of output events + f - A function of whatever type. + q - Queue of output events, of type Seq (Event c) + s - Used for both the type and value of the state. + t - Time, of type UTCTime + v - Used for both the value returned by the current action and its type. + z - Non-termination flag. True if the RTSP should respond to future events. +-} + +instance Functor (RTA s c) where + fmap f rv = RTA $ \s q t -> + let (v1, s1, q1, t1) = unRTA rv s q t + in (f v1, s1, q1, t1) + + +instance Monad (RTA s c) where + return v = RTA $ \ s q t -> (v, s, q, t) + rv >>= f = RTA $ \s q t -> + let (v1, s1, q1, t1) = unRTA rv s q t + in unRTA (f v1) s1 q1 t1 + + + +-- | Get the current time. This is the event time plus any pauses. +now :: RTA s c UTCTime +now = RTA $ \s q t -> (t, s, q, t) + +-- | Get the current state. +get :: RTA s c s +get = RTA $ \s q t -> (s, s, q, t) + +-- | Put the current state. +put :: s -> RTA s c () +put v = RTA $ \_ q t -> ((), v, q, t) + +-- | Apply a function to the current state. +modify :: (s -> s) -> RTA s c () +modify f = fmap f get >>= put + +-- | Emit a value as an event. +emit :: c -> RTA s c () +emit v = RTA $ \s q t -> ((), s, q |> Event t v, t) + + +-- | Pause before the next step. This does not actually delay processing; it merely increments the time of any emitted events. +pause :: NominalDiffTime -> RTA s c () +pause dt + | dt >= 0 = RTA $ \s q t -> ((), s, q, addUTCTime dt t) + | otherwise = error $ "pause: negative interval of " ++ show dt + + + + +-- | Execute an RTA as part of a real time stream processor. +-- +-- When a new event arrives any pending output events will be lost. However any state changes are immediately visible to the +-- next event, even if they occured \"after\" the lost events. For instance, consider this: +-- +-- > execRTA 1 $ \_ -> do +-- > n <- get +-- > pause 10 +-- > emit n +-- > put (n+1) +-- > return True +-- +-- If this receives events at @t=[0,1,3,20]@ then it will emit @[Event 13 3, Event 30 4]@. The events that would have been emitted +-- at @t=[10,11]@ have been lost, but the state change still occured immediately, regardless of the output schedule. +execRTA :: + s + -- ^ The initial state. State persists between input events. + -> (b -> RTA s c Bool) + -- ^ A function from the input value to an action. If the action returns @True@ then subsequent input events + -- will run the action again. If it returns @False@ then the RTSP finishes and will not respond to further events. + -> RTSP b c +execRTA s f = RTSP $ \ev -> + let + t = eventTime ev + v = eventValue ev + (z, s1, q, _) = unRTA (f v) s empty t + queueStream q1 = case viewl q1 of + EmptyL -> if z then Wait $ execRTA s1 f else nullStream + c :< q2 -> Emit c (queueStream q2) (if z then execRTA s1 f else eventSink) + in queueStream q + + +-- | Like "execRTA", except that output events are accumulated. +accumulateRTA :: s -> (b -> RTA s c Bool) -> RTSP b c +accumulateRTA s f = accumulate $ execRTA s f + + + diff --git a/test/ArbTest.hs b/test/ArbTest.hs new file mode 100644 --- /dev/null +++ b/test/ArbTest.hs @@ -0,0 +1,340 @@ +{-# OPTIONS_GHC -fno-warn-orphans -XFlexibleInstances #-} + + +module ArbTest ( + ArbEvents (..), + RtspTest (..), + interpret, + compile, + prop_emitsBefore, + prop_RTSP, + prop_rtspMonoid1, + prop_rtspMonoid2, + prop_rtspMonoidCommutes, + prop_rtspMonoidAssociates, + prop_rtspCategoryId1, + prop_rtspCategoryId2, + prop_rtspCategoryAssociates, + prop_rtspCategoryAssociates2, + prop_rtspIfThenElse, + prop_eventCount, + prop_eventLatch +) where + + +-- import Control.Applicative +import Control.RTSP +import Control.Category +import Control.Monad +import Data.Function (on) +import Data.List (groupBy, sortBy, partition) +import Data.Maybe +import Data.Monoid +import Data.Ord +import Data.Time +import Test.QuickCheck + + +import Prelude hiding ((.), id) + + + +-- Orphan instances for RTSP types + +instance Arbitrary a => Arbitrary (Event a) where + arbitrary = do + dt <- arbitrary + v <- arbitrary + return $ Event (dt `addUTCTime` epoch) v + shrink = shrinkNothing + + +instance Arbitrary NominalDiffTime where + arbitrary = fmap (fromRational . abs) arbitrary + shrink dt = if dt == dtSecs then [] else [dtSecs] + where dtSecs = fromInteger $ floor dt + + + +-- | Base time for events. +epoch :: UTCTime +epoch = UTCTime (fromGregorian 2000 1 1) 0 + + +-- | A list of arbitrary events in chronological order +newtype ArbEvents a = ArbEvents [Event a] + +instance (Show a) => Show (ArbEvents a) where + show (ArbEvents evs) = "[\n" ++ concatMap showEvent evs ++ "]" + where showEvent (Event t v) = " Event " ++ show (diffUTCTime t epoch) ++ " " ++ show v ++ "\n" + +instance (Arbitrary a) => Arbitrary (ArbEvents a) where + arbitrary = do + times <- fmap (scanl (flip addUTCTime) epoch) arbitrary + events <- forM times $ \t -> fmap (Event t) arbitrary + return $ ArbEvents events + shrink (ArbEvents evs) = map ArbEvents $ shrink evs + + +-- | An RTSP test consists of a descriptive string, a list function and an RTSP. The list function +-- has the same effect on the list of events that the RTSP has on a stream. +data RtspTest a = + Id + | Delay Rational + | Dup Rational + | Func (a -> a) String + | If (a -> Bool) String (RtspTest a) (RtspTest a) + | Pipe (RtspTest a) (RtspTest a) + | Par (RtspTest a) (RtspTest a) + +instance Show (RtspTest a) where + show Id = "id" + show (Delay t) = "delay " ++ show t + show (Dup t) = "duplicate " ++ show t + show (Func _ str) = "stream " ++ str + show (If _ str r1 r2) = "(ifThenElse (" ++ str ++ ", " ++ show r1 ++ ") (" ++ show r2 ++ "))" + show (Pipe r1 r2) = "(" ++ show r1 ++ " >>> " ++ show r2 ++ ")" + show (Par r1 r2) = "(" ++ show r1 ++ " `mappend` " ++ show r2 ++ ")" + + +instance (Integral a) => Arbitrary (RtspTest a) where + arbitrary = do + frequency [ + (2, return Id), + (1, fmap (Delay . fromRational . abs) arbitrary), + (1, fmap (Dup . fromRational . abs) arbitrary), + (1, oneof [return $ Func (*2) "(*2)", + return $ Func (+1) "(+1)", + return $ Func (*3) "(*3)"]), + (1, do + (p, str) <- elements [(odd, "odd"), + (even, "even"), + ((== 0) . (`mod` 3), "mult3")] + return (If p str) `ap` arbitrary `ap` arbitrary), + (1, return Pipe `ap` arbitrary `ap` arbitrary), + (1, return Par `ap` arbitrary `ap` arbitrary) + ] + shrink Id = [] + shrink (Delay dt) = Id : map Delay (shrink dt) + shrink (Dup dt) = Delay dt : map Dup (shrink dt) + shrink (Func _ _) = [Id] + shrink (If p str r1 r2) = shrinkBinaryOp (If p str) r1 r2 + shrink (Pipe r1 r2) = shrinkBinaryOp Pipe r1 r2 + shrink (Par r1 r2) = shrinkBinaryOp Par r1 r2 + + +shrinkBinaryOp :: (Integral a) => (RtspTest a -> RtspTest a -> RtspTest a) -> RtspTest a -> RtspTest a -> [RtspTest a] +shrinkBinaryOp op r1 r2 = concat [ + [Id, r1, r2], + map (\r -> op r r2) $ shrink r1, + map (op r1) $ shrink r2] + + +-- | Interpret a test on a list, predicting the output for the equivalent arrow. +interpret :: (Num a) => RtspTest a -> [Event a] -> [Event a] +interpret Id evs = evs +interpret (Delay dt) evs = map (\(Event t v) -> Event (addUTCTime (fromRational dt) t) v) evs +interpret (Dup dt) evs = + foldl merge [] $ map (\(Event t v) -> [Event t v, Event (addUTCTime (fromRational dt) t) v]) evs +interpret (Func f _) evs = map (fmap f) evs +interpret (If p _ r1 r2) evs = merge (interpret r1 thens) (interpret r2 elses) + where + (thens, elses) = partition (p . eventValue) evs +interpret (Pipe r1 r2) evs = interpret r2 $ interpret r1 evs +interpret (Par r1 r2) evs = merge (interpret r1 evs) (interpret r2 evs) + + + +-- | Merge two sorted lists of events. +merge :: [Event a] -> [Event a] -> [Event a] +merge xs [] = xs +merge [] ys = ys +merge xs@(x:xs1) ys@(y:ys1) = if y `isBefore` x then y : merge ys1 xs else x : merge ys xs1 + + + +-- | Compile a test into an arrow. +compile :: (Num a) => RtspTest a -> RTSP a a +compile Id = id +compile (Delay dt) = delay (fromRational dt) +compile (Dup dt) = accumulate $ repeatEvent [0, fromRational dt] +compile (Func f _) = stream f +compile (If p _ r1 r2) = ifThenElse (p, compile r1) (compile r2) +compile (Pipe r1 r2) = compile r1 >>> compile r2 +compile (Par r1 r2) = compile r1 `mappend` compile r2 + + +-- | Two event streams are equivalent regardless of the ordering of simultaneous events +isEquivalent :: (Ord a) => [Event a] -> [Event a] -> Bool +isEquivalent xs ys = normalise xs == normalise ys + where normalise = map (sortBy (comparing eventValue)) . groupBy ((==) `on` eventTime) + + +-- | Assert that "compile" and "interpret" are equivalent. +prop_RTSP :: RtspTest Integer -> ArbEvents Integer -> Property +prop_RTSP tst (ArbEvents evs) = printTestCase failStr $ result1 `isEquivalent` result2 + where + result1 = interpret tst evs + result2 = simulateRTSP (compile tst) evs + failStr = concat [ "interpret => ", show (ArbEvents result1), "\n", + "compile => ", show (ArbEvents result2), "\n"] + + + + +-- | Reify primitive RTA actions. +data RtaTest s c = + Modify (s -> s) | Emit (s -> c) | Pause (s -> NominalDiffTime) + + +instance (Arbitrary s, CoArbitrary s, Arbitrary c) => Arbitrary (RtaTest s c) + where + arbitrary = frequency [ + (3, return Modify `ap` arbitrary), + (1, return Emit `ap` arbitrary), + (5, return Pause `ap` arbitrary) + ] + shrink = shrinkNothing + +instance Show (RtaTest s c) where + show (Modify _) = "Modify" + show (Emit _) = "Emit" + show (Pause _) = "Pause" + + + +instance (Integral s, Arbitrary s, CoArbitrary s, Arbitrary c) => Arbitrary (RTA s c Bool) where + arbitrary = fmap execRtaTests arbitrary + shrink = shrinkNothing + +instance (CoArbitrary b, Arbitrary c) => Arbitrary (RTSP b c) where + arbitrary = do + rtaF <- arbitrary + return $ execRTA (0 :: Integer) rtaF + shrink = shrinkNothing + + +-- | Execute an RtaTest +execRtaTest :: RtaTest s c -> RTA s c () +execRtaTest (Modify f) = fmap f get >>= put +execRtaTest (Emit f) = fmap f get >>= emit +execRtaTest (Pause f) = fmap f get >>= pause + + +-- | Execute a sequence of RtaTests as a single action. +execRtaTests :: (Integral s) => [RtaTest s c] -> RTA s c Bool +execRtaTests ts = do + mapM_ execRtaTest ts + s <- get + -- return True + return $ (s `mod` 20) /= 0 + + +type RtspProp = ArbEvents Integer -> Property + +type RII = RTSP Integer Integer + +rtspEquivalent :: (CoArbitrary b, Arbitrary c, Ord c, Show c) => RTSP b c -> RTSP b c -> ArbEvents b -> Property +rtspEquivalent r1 r2 (ArbEvents evs) = printTestCase failStr $ trace r1 `isEquivalent` trace r2 + where + trace r = simulateRTSP r evs + failStr = concat [ + "Trace1 = ", show $ ArbEvents $ trace r1, "\n", + "Trace2 = ", show $ ArbEvents $ trace r2, "\n"] + + +prop_emitsBefore :: RII -> RII -> Event Integer -> Property +prop_emitsBefore r1 r2 ev = + printTestCase (show (evs1, evs2)) $ + case (evs1, evs2) of + (Nothing, Nothing) -> not (es1 `emitsBefore` es2 || es2 `emitsBefore` es1) + (Just _, Nothing) -> es1 `emitsBefore` es2 && not (es2 `emitsBefore` es1) + (Nothing, Just _ ) -> not (es1 `emitsBefore` es2) && es2 `emitsBefore` es1 + (Just e1, Just e2) -> (e1 `isBefore` e2) == (es1 `emitsBefore` es2) && + (e2 `isBefore` e1) == (es2 `emitsBefore` es1) + where + es1 = runRTSP r1 ev + es2 = runRTSP r2 ev + evs1 = listToMaybe $ esPeek es1 + evs2 = listToMaybe $ esPeek es2 + +prop_rtspMonoid1 :: RII -> RtspProp +prop_rtspMonoid1 r = rtspEquivalent r (mempty `mappend` r) + +prop_rtspMonoid2 :: RII -> RtspProp +prop_rtspMonoid2 r = rtspEquivalent r (r `mappend` mempty) + + +prop_rtspMonoidCommutes :: RII -> RII -> RtspProp +prop_rtspMonoidCommutes r1 r2 = rtspEquivalent (r1 `mappend` r2) (r2 `mappend` r1) + +prop_rtspMonoidAssociates :: RII -> RII -> RII -> RtspProp +prop_rtspMonoidAssociates r1 r2 r3 = rtspEquivalent (r1 `mappend` (r2 `mappend` r3)) ((r1 `mappend` r2) `mappend` r3) + + +prop_rtspCategoryId1 :: RII -> RtspProp +prop_rtspCategoryId1 r = rtspEquivalent r (id >>> r) + +prop_rtspCategoryId2 :: RII -> RtspProp +prop_rtspCategoryId2 r = rtspEquivalent r (r >>> id) + + +prop_rtspCategoryAssociates :: RII -> RII -> RII -> RtspProp +prop_rtspCategoryAssociates r1 r2 r3 = rtspEquivalent (r1 >>> (r2 >>> r3)) ((r1 >>> r2) >>> r3) + +prop_rtspCategoryAssociates2 :: RtspTest Integer -> RtspTest Integer -> RtspTest Integer -> RtspProp +prop_rtspCategoryAssociates2 rt1 rt2 rt3 = rtspEquivalent (r1 >>> (r2 >>> r3)) ((r1 >>> r2) >>> r3) + where + r1 = compile rt1 + r2 = compile rt2 + r3 = compile rt3 + + +-- | @ifThenElse (p, rThen) rElse@ is equivalent to +-- +-- > streamFilter (p, rThen) `mappend` streamFilter (not . p, rElse) +prop_rtspIfThenElse :: RII -> RII -> RtspProp +prop_rtspIfThenElse r1 r2 = rtspEquivalent + (ifThenElse (odd, r1) r2) + (streamFilter (odd, r1) `mappend` streamFilter (not . odd, r2)) +-- where +-- r1 = compile rt1 +-- r2 = compile rt2 + + +traceEquivalent :: (Ord a, Show a) => [Event a] -> [Event a] -> Property +traceEquivalent trace1 trace2 = + printTestCase failStr $ trace1 `isEquivalent` trace2 + where failStr = concat [ + "Trace 1 = ", show $ ArbEvents trace1, "\n", + "Trace 2 = ", show $ ArbEvents trace2, "\n"] + + +-- | Count events. +eventCount :: RTSP b (Integer, b) +eventCount = execRTA 0 $ \v -> do + s <- fmap (+1) get + put s + emit (s, v) + return True + +prop_eventCount :: RtspProp +prop_eventCount (ArbEvents evs) = traceEquivalent + (simulateRTSP eventCount evs) + (zipWith (\n -> fmap (\v -> (n, v))) [1..] evs) + + + +-- | Repeat each event value once a second ten times. +eventLatch :: RTSP b b +eventLatch = accumulateRTA () $ \v -> do + replicateM_ 5 (emit v >> pause 1) + return True + +prop_eventLatch :: RtspProp +prop_eventLatch (ArbEvents evs) = traceEquivalent + (simulateRTSP eventLatch evs) + (sortBy (comparing eventTime) $ concatMap rep evs) + where + rep (Event t v) = [Event (n `addUTCTime` t) v | n <- [0,1..4]] + diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,40 @@ + +module Main where + +import ArbTest +import Test.Framework +import Test.Framework.Providers.QuickCheck2 (testProperty) + +longTest :: TestOptions +longTest = TestOptions Nothing (Just 1000) Nothing Nothing Nothing + +main :: IO () +main = defaultMain tests + +tests :: [Test] +tests = map (longTest `plusTestOptions`) + [testGroup "Arbitrary RTSP list equivalence" + [testProperty "compile == interpret" prop_RTSP, + testProperty "ifThenElse definition" prop_rtspIfThenElse + ], + testGroup "Event order consistency" + [testProperty "emitsBefore consistent with isBefore" prop_emitsBefore + ], + testGroup "Monoid laws, plus commutivity" + [testProperty "Monoid left identity" prop_rtspMonoid1, + testProperty "Monoid right identity" prop_rtspMonoid2, + testProperty "Monoid commutes" prop_rtspMonoidCommutes, + testProperty "Monoid associates" prop_rtspMonoidAssociates + ], + testGroup "Category laws" + [testProperty "Category left identity" prop_rtspCategoryId1, + testProperty "Category right identity" prop_rtspCategoryId2, + testProperty "Category associates" prop_rtspCategoryAssociates, + testProperty "Category associates 2" prop_rtspCategoryAssociates2 + ], + testGroup "Sample RTAs" + [ + testProperty "Event Count" prop_eventCount, + testProperty "Event Latch" prop_eventLatch + ] + ]