dataflower 0.1.4.0 → 0.2.0.0

raw patch · 5 files changed

+101/−19 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• dataflower.cabal +2/−2
• src/Dataflow.hs +8/−6
• src/Dataflow/Primitives.hs +55/−6
• src/Test/Dataflow.hs +2/−1
• test/DataflowSpec.hs +34/−4

@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: b4d9c5533fc659e6801a98a43183cb91426ce81a603cc0267578d5a1cba664ad+-- hash: f8c905e3886d672d9325bc3d2737600644d15f243a4c93fdfb69cde4c8de0b6c  name:           dataflower-version:        0.1.4.0+version:        0.2.0.0 synopsis:       A Pure-Haskell Timely Dataflow System description:    See README homepage:       https://github.com/doublecrowngaming/dataflower#readme

@@ -32,7 +32,7 @@  import           Control.Monad              (void) import           Control.Monad.IO.Class     (MonadIO (..))-import           Control.Monad.State.Strict (execStateT, runStateT)+import           Control.Monad.State.Strict (evalStateT, execStateT, runStateT) import           Data.Traversable           (Traversable) import           Dataflow.Primitives import           Dataflow.Vertices@@ -56,8 +56,10 @@ -- have the same 'Timestamp' associated with them. -- -- @since 0.1.0.0-execute :: (MonadIO io, Traversable t) => t i -> Program i -> io ()-execute corpus Program{..} = liftIO $ execDataflow feedInput-  where-    feedInput           = input corpus programInput-    execDataflow action = void $ execStateT (runDataflow action) programState+execute :: (MonadIO io, Traversable t) => t i -> Program i -> io (Program i)+execute corpus Program{..} = liftIO $ do+  newProgramState <- evalStateT (runDataflow duplicateDataflowState) programState++  void $ execStateT (runDataflow $ input corpus programInput) newProgramState++  return $ Program programInput newProgramState

@@ -1,12 +1,14 @@ {-# LANGUAGE ExistentialQuantification  #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses      #-}+{-# LANGUAGE RecordWildCards            #-}  module Dataflow.Primitives (   Dataflow(..),   DataflowState,   Vertex(..),   initDataflowState,+  duplicateDataflowState,   StateRef,   newState,   readState,@@ -23,7 +25,9 @@ ) where  import           Control.Arrow              ((>>>))-import           Control.Monad.State.Strict (StateT, gets, modify)+import           Control.Monad              (forM, (>=>))+import           Control.Monad.IO.Class     (liftIO)+import           Control.Monad.State.Strict (StateT, get, gets, modify) import           Control.Monad.Trans        (lift) import           Data.Hashable              (Hashable (..)) import           Data.IORef                 (IORef, atomicModifyIORef',@@ -36,6 +40,7 @@   newtype VertexID    = VertexID        Int deriving (Eq, Ord, Show)+newtype StateID     = StateID         Int deriving (Eq, Ord, Show) newtype Epoch       = Epoch       Natural deriving (Eq, Ord, Hashable, Show)  -- | 'Timestamp's represent instants in the causal timeline.@@ -56,6 +61,9 @@ instance Incrementable VertexID where   inc (VertexID n) = VertexID (n + 1) +instance Incrementable StateID where+  inc (StateID n) = StateID (n + 1)+ instance Incrementable Epoch where   inc (Epoch n) = Epoch (n + 1) @@ -69,8 +77,10 @@  data DataflowState = DataflowState {   dfsVertices       :: Vector ErasedType,+  dfsStates         :: Vector (IORef ErasedType),   dfsFinalizers     :: [Timestamp -> Dataflow ()],   dfsLastVertexID   :: VertexID,+  dfsLastStateID    :: StateID,   dfsLastInputEpoch :: Epoch } @@ -83,11 +93,24 @@ initDataflowState :: DataflowState initDataflowState = DataflowState {   dfsVertices       = empty,+  dfsStates         = empty,   dfsFinalizers     = [],   dfsLastVertexID   = VertexID (-1),+  dfsLastStateID    = StateID (-1),   dfsLastInputEpoch = Epoch 0 } +duplicateDataflowState :: Dataflow (DataflowState)+duplicateDataflowState = Dataflow $ do+  DataflowState{..} <- get++  newStates <- liftIO $ forM dfsStates dupIORef++  return $ DataflowState { dfsStates = newStates, .. }++  where+    dupIORef = readIORef >=> newIORef+ -- | Get the next input Epoch. incrementEpoch :: Dataflow Epoch incrementEpoch =@@ -138,31 +161,57 @@ -- | Mutable state that holds an `a`. -- -- @since 0.1.0.0-newtype StateRef a = StateRef (IORef a)+newtype StateRef a = StateRef StateID  -- | 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)+newState a =+  Dataflow $ do+    sid   <- gets (dfsLastStateID >>> inc)+    ioref <- lift $ newIORef (EraseType a) +    modify $ addState ioref sid++    return (StateRef sid)++  where+    addState ref sid s = s {+      dfsStates      = dfsStates s `snoc` ref,+      dfsLastStateID = sid+    }++lookupStateRef :: StateRef s -> Dataflow (IORef ErasedType)+lookupStateRef (StateRef (StateID sindex)) =+  Dataflow $ do+    states <- gets dfsStates++    return (states ! sindex)+ -- | Read the value stored in the `StateRef`. -- -- @since 0.1.0.0 readState :: StateRef a -> Dataflow a-readState (StateRef ref) = Dataflow $ lift (readIORef ref)+readState sref = do+  ioref <- lookupStateRef sref+  Dataflow $ lift $ (unEraseType <$> readIORef ioref)  -- | 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+writeState sref x = do+  ioref <- lookupStateRef sref+  Dataflow $ lift $ atomicWriteIORef ioref (EraseType 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, ()))+modifyState sref op = do+  ioref <- lookupStateRef sref+  Dataflow $ lift $ atomicModifyIORef' ioref (\x -> (EraseType $ op (unEraseType x), ()))  {-# INLINEABLE input #-} input :: Traversable t => t i -> Edge i -> Dataflow ()

@@ -3,6 +3,7 @@ ) where  import           Control.Concurrent.STM.TVar (newTVarIO, readTVarIO)+import           Control.Monad               (void) import           Control.Monad.IO.Class      (MonadIO (..)) import           Dataflow                    (Dataflow, Edge, compile, execute,                                               outputTVar)@@ -17,6 +18,6 @@     out     <- newTVarIO []     program <- compile (dataflow =<< outputTVar (:) out) -    execute inputs program+    void $ execute inputs program      reverse <$> readTVarIO out

@@ -2,13 +2,15 @@  module DataflowSpec (spec) where -import           Control.Monad   ((>=>))+import           Control.Concurrent.STM.TVar (newTVarIO, readTVarIO)+import           Control.Monad               (void, (>=>)) import           Dataflow-import           Prelude         hiding (map)+import           Prelude -import           Test.Dataflow   (runDataflow)+import           Test.Dataflow               (runDataflow) import           Test.Hspec-import           Test.QuickCheck hiding (discard)+import           Test.QuickCheck             hiding (discard)+import           Test.QuickCheck.Modifiers   (NonEmptyList (..))   spec :: Spec@@ -18,6 +20,24 @@      \(numbers :: [Integer]) -> runDataflow passthrough numbers `shouldReturn` numbers +  describe "execute" $ do+    it "isolates the state of runs from each other" $ property $ \(NonEmpty numbers) -> do+      out     <- newTVarIO []+      program <- compile (integrate =<< outputTVar (:) out)++      void $ execute numbers program+      void $ execute numbers program++      (reverse <$> readTVarIO out) `shouldReturn` (scanl1 (+) numbers ++ scanl1 (+) numbers)++    it "bundles all the execution state into a Program" $ property $ \(NonEmpty numbers) -> do+      out     <- newTVarIO 0+      program <- compile (integrate =<< outputTVar const out)++      void $ execute numbers program >>= execute numbers >>= execute numbers++      readTVarIO out `shouldReturn` (3 * sum numbers)+   describe "finalize" $ do     it "finalizes vertices" $ property $       \(numbers :: [Int]) -> runDataflow storeAndForward numbers `shouldReturn` numbers@@ -37,3 +57,13 @@     forward sref t = do       mapM_ (send next t) =<< reverse <$> readState sref       writeState sref []++integrate :: Edge Int -> Dataflow (Edge Int)+integrate next = statefulVertex 0 recv finalize+  where+    recv s t i   = do+      modifyState s (+ i)++      send next t =<< readState s++    finalize _ _ = return ()