dataflower 0.1.1.0 → 0.1.2.0

raw patch · 6 files changed

+70/−6 lines, 6 filesdep +pretty-showPVP ok

version bump matches the API change (PVP)

Dependencies added: pretty-show

API changes (from Hackage documentation)

Files

• dataflower.cabal +3/−2
• src/Dataflow.hs +8/−0
• src/Dataflow/Primitives.hs +20/−0
• src/Dataflow/Vertices.hs +31/−2
• src/Test/Dataflow.hs +3/−1
• test/DataflowSpec.hs +5/−1

@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 896552b1d7ddbd42696c3757cf557d2796ed12b99f8cd0576db5a03993194d0f+-- hash: b1bed6fdab70f8f031023bf1ce16822eb629e5e6ad13326cddbf6a46c525563d  name:           dataflower-version:        0.1.1.0+version:        0.1.2.0 synopsis:       A Pure-Haskell Timely Dataflow System description:    See README homepage:       https://github.com/doublecrowngaming/dataflower#readme@@ -37,6 +37,7 @@       base >=4.7 && <5     , hashable     , mtl+    , pretty-show     , stm     , time     , transformers

@@ -23,6 +23,8 @@   statefulVertex,   statelessVertex,   outputTVar,+  trace,+  discard,   Program,   compile,   execute@@ -36,17 +38,23 @@  -- | A 'Program' represents a fully-preprocessed 'Dataflow' that may be -- executed against inputs.+--+-- @since 0.1.0.0 data Program i = Program {   programInput :: Edge i,   programState :: DataflowState }  -- | Take a 'Dataflow' which takes 'i's as input and compile it into a 'Program'.+--+-- @since 0.1.0.0 compile :: Dataflow (Edge i) -> IO (Program i) compile (Dataflow actions) = uncurry Program <$> runStateT actions initDataflowState  -- | Feed a traversable collection of inputs to a 'Program'. All inputs provided will -- have the same 'Timestamp' associated with them.+--+-- @since 0.1.0.0 execute :: Traversable t => t i -> Program i -> IO () execute corpus Program{..} = execDataflow feedInput   where

@@ -39,10 +39,14 @@ newtype Epoch       = Epoch       Natural deriving (Eq, Ord, Hashable, Show)  -- | 'Timestamp's represent instants in the causal timeline.+--+-- @since 0.1.0.0 newtype Timestamp   = Timestamp     Epoch deriving (Eq, Ord, Hashable, Show)  -- | An 'Edge' is a typed reference to a computational vertex that -- takes 'a's as its input.+--+-- @since 0.1.0.0 newtype Edge a      = Edge       VertexID  -- | Class of entities that can be incremented by one.@@ -71,6 +75,8 @@ }  -- | `Dataflow` is the type of all dataflow operations.+--+-- @since 0.1.0.0 newtype Dataflow a = Dataflow { runDataflow :: StateT DataflowState IO a }   deriving (Functor, Applicative, Monad) @@ -130,21 +136,31 @@   Dataflow $ modify $ \s -> s { dfsFinalizers = finalizer : dfsFinalizers s }  -- | Mutable state that holds an `a`.+--+-- @since 0.1.0.0 newtype StateRef a = StateRef (IORef a)  -- | Create a `StateRef` initialized to the provided `a`.+--+-- @since 0.1.0.0 newState :: a -> Dataflow (StateRef a) newState a = StateRef <$> (Dataflow $ lift $ newIORef a)  -- | Read the value stored in the `StateRef`.+--+-- @since 0.1.0.0 readState :: StateRef a -> Dataflow a readState (StateRef ref) = Dataflow $ lift (readIORef ref)  -- | Overwrite the value stored in the `StateRef`.+--+-- @since 0.1.0.0 writeState :: StateRef a -> a -> Dataflow () writeState (StateRef ref) x = Dataflow $ lift $ atomicWriteIORef ref x  -- | Update the value stored in `StateRef`.+--+-- @since 0.1.0.0 modifyState :: StateRef a -> (a -> a) -> Dataflow () modifyState (StateRef ref) op = Dataflow $ lift $ atomicModifyIORef' ref (\x -> (op x, ())) @@ -159,6 +175,8 @@  {-# INLINE send #-} -- | Send an `input` item to be worked on to the indicated vertex.+--+-- @since 0.1.0.0 send :: Edge input -> Timestamp -> input -> Dataflow () send e t i = lookupVertex e >>= invoke t i   where@@ -166,6 +184,8 @@     invoke timestamp datum (StatelessVertex callback)     = callback timestamp datum  -- Notify all relevant vertices that no more input is coming for `Timestamp`.+--+-- @since 0.1.0.0 finalize :: Timestamp -> Dataflow () finalize t = do   finalizers <- Dataflow $ gets dfsFinalizers

@@ -5,7 +5,9 @@ module Dataflow.Vertices (   statefulVertex,   statelessVertex,-  outputTVar+  outputTVar,+  trace,+  discard ) where  import           Control.Concurrent.STM.TVar (TVar, modifyTVar')@@ -14,8 +16,9 @@ import           Dataflow.Primitives         (Dataflow (..), Edge, StateRef,                                               Timestamp (..), Vertex (..),                                               newState, registerFinalizer,-                                              registerVertex)+                                              registerVertex, send) import           Prelude+import           Text.Show.Pretty            (pPrint)   -- | Construct a vertex with internal state. Like 'statelessVertex', 'statefulVertex'@@ -25,6 +28,8 @@ -- -- NB: Until the finalizer has been called for a particular timestamp, a stateful vertex -- must be capable of accepting data for multiple timestamps simultaneously.+--+-- @since 0.1.0.0 statefulVertex ::   state -- ^ The initial state value.   -> (StateRef state -> Timestamp -> i -> Dataflow ()) -- ^ The input handler.@@ -40,6 +45,8 @@ -- -- `send`ing to a stateless vertex is effectively a function call and will execute in the -- caller's thread. By design this is a cheap operation.+--+-- @since 0.1.0.0 statelessVertex :: (Timestamp -> i -> Dataflow ()) -> Dataflow (Edge i) statelessVertex callback = registerVertex $ StatelessVertex callback @@ -47,5 +54,27 @@ -- | Construct an output vertex that stores items into the provided 'TVar'. The first argument -- is an update function so that, for example, the 'TVar' could contain a list of 'o's and the update -- function could then `cons` new items onto the list.+--+-- @since 0.1.0.0 outputTVar :: (o -> w -> w) -> TVar w -> Dataflow (Edge o) outputTVar op register = statelessVertex $ \_ x -> Dataflow $ lift $ atomically $ modifyTVar' register (op x)++-- | Construct a vertex that pretty-prints items and passes them through unchanged.+--+-- @since 0.1.2.0+trace :: Show i => Edge i -> Dataflow (Edge i)+trace next = do+  trace' <- ioVertex $ curry pPrint++  statelessVertex $ \t x -> do+    send trace' t x+    send next t x++  where+    ioVertex callback = registerVertex $ StatelessVertex $ \t i -> Dataflow $ lift $ callback t i++-- | Construct a vertex that discards anything sent to it.+--+-- @since 0.1.2.0+discard :: Dataflow (Edge i)+discard = statelessVertex $ \_ _ -> return ()

@@ -8,7 +8,9 @@                                               outputTVar) import           Prelude -+-- | Run a dataflow with a list of inputs.+--+-- @since 0.1.0.0 runDataflow :: MonadIO io => (Edge o -> Dataflow (Edge i)) -> [i] -> io [o] runDataflow dataflow inputs =   liftIO $ do

@@ -8,7 +8,7 @@  import           Test.Dataflow   (runDataflow) import           Test.Hspec-import           Test.QuickCheck+import           Test.QuickCheck hiding (discard)   spec :: Spec@@ -25,6 +25,10 @@     it "finalizes vertices in the correct order" $ property $       \(numbers :: [Int]) ->         runDataflow (storeAndForward >=> storeAndForward >=> storeAndForward) numbers `shouldReturn` numbers++  describe "discard" $+    it "discards all input" $ property $+      \(numbers :: [Int]) -> runDataflow (const discard) numbers `shouldReturn` ([] :: [Int])  storeAndForward :: Edge i -> Dataflow (Edge i) storeAndForward next = statefulVertex [] store forward