Dflow (empty) → 0.0.1

raw patch · 7 files changed

+1207/−0 lines, 7 filesdep +HUnitdep +QuickCheckdep +basesetup-changed

Dependencies added: HUnit, QuickCheck, base, containers, stm, test-framework, test-framework-quickcheck2, time

Files

Dflow.cabal +51/−0
LICENSE.txt +26/−0
README.txt +82/−0
Setup.hs +2/−0
src/Control/RTSP.hs +666/−0
test/ArbTest.hs +340/−0
test/Main.hs +40/−0

+ Dflow.cabal view

@@ -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:     <paul@cogito.org.uk>+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

+ LICENSE.txt view

@@ -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.

+ README.txt view

@@ -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.

+ Setup.hs view

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

+ src/Control/RTSP.hs view

@@ -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+++

+ test/ArbTest.hs view

@@ -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]]+

+ test/Main.hs view

@@ -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+       ]      +   ]