dataflower 0.1.0.0 → 0.1.1.0

raw patch · 7 files changed

+27/−24 lines, 7 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• benchmark/Main.hs +3/−3
• dataflower.cabal +2/−2
• src/Dataflow.hs +1/−2
• src/Dataflow/Primitives.hs +16/−9
• src/Dataflow/Vertices.hs +3/−4
• src/Test/Dataflow.hs +1/−2
• test/DataflowSpec.hs +1/−2

@@ -1,7 +1,6 @@ module Main where  import           Criterion.Main-import           Data.Typeable  (Typeable) import           Dataflow import           Prelude import           Test.Dataflow  (runDataflow)@@ -11,12 +10,13 @@ main = defaultMain [   bgroup "dataflow" [       bench "passthrough 1,000,000 inputs" $ nfIO (runDataflow passthrough [0..1000000 :: Int]),+      bench "discard         1,000 inputs" $ nfIO (runDataflow blackhole [0..1000    :: Int]),       bench "discard     1,000,000 inputs" $ nfIO (runDataflow blackhole [0..1000000 :: Int])     ]   ]   where-    passthrough :: Typeable a => Edge a -> Dataflow (Edge a)+    passthrough :: Edge a -> Dataflow (Edge a)     passthrough next = statelessVertex $ \t x -> send next t x -    blackhole :: Typeable a => Edge a -> Dataflow (Edge a)+    blackhole :: Edge a -> Dataflow (Edge a)     blackhole _ = statelessVertex $ \_ _ -> return ()

@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 8f7b0ebc453194c8e6a5cca3c5cef03c62cfa135a28e2c15fcf3ff5a58e10d3a+-- hash: 896552b1d7ddbd42696c3757cf557d2796ed12b99f8cd0576db5a03993194d0f  name:           dataflower-version:        0.1.0.0+version:        0.1.1.0 synopsis:       A Pure-Haskell Timely Dataflow System description:    See README homepage:       https://github.com/doublecrowngaming/dataflower#readme

@@ -31,7 +31,6 @@ import           Control.Monad              (void) import           Control.Monad.State.Strict (execStateT, runStateT) import           Data.Traversable           (Traversable)-import           Data.Typeable              (Typeable) import           Dataflow.Primitives import           Dataflow.Vertices @@ -48,7 +47,7 @@  -- | Feed a traversable collection of inputs to a 'Program'. All inputs provided will -- have the same 'Timestamp' associated with them.-execute :: (Traversable t, Typeable i) => t i -> Program i -> IO ()+execute :: Traversable t => t i -> Program i -> IO () execute corpus Program{..} = execDataflow feedInput   where     feedInput           = input corpus programInput

@@ -25,7 +25,6 @@ import           Control.Arrow              ((>>>)) import           Control.Monad.State.Strict (StateT, gets, modify) import           Control.Monad.Trans        (lift)-import           Data.Dynamic               (Dynamic, Typeable, fromDyn, toDyn) import           Data.Hashable              (Hashable (..)) import           Data.IORef                 (IORef, atomicModifyIORef',                                              atomicWriteIORef, newIORef,@@ -33,6 +32,7 @@ import           Data.Vector                (Vector, empty, snoc, (!)) import           Numeric.Natural            (Natural) import           Prelude+import           Unsafe.Coerce              (unsafeCoerce)   newtype VertexID    = VertexID        Int deriving (Eq, Ord, Show)@@ -56,8 +56,15 @@   inc (Epoch n) = Epoch (n + 1)  +-- | 'ErasedType' erases the type it wraps.+data ErasedType = forall i. EraseType i++unEraseType :: ErasedType -> a+unEraseType (EraseType x) = unsafeCoerce x++ data DataflowState = DataflowState {-  dfsVertices       :: Vector Dynamic,+  dfsVertices       :: Vector ErasedType,   dfsFinalizers     :: [Timestamp -> Dataflow ()],   dfsLastVertexID   :: VertexID,   dfsLastInputEpoch :: Epoch@@ -94,15 +101,15 @@       (Timestamp -> i -> Dataflow ())  -- | Retrieve the vertex for a given edge.-lookupVertex :: Typeable i => Edge i -> Dataflow (Vertex i)+lookupVertex :: Edge i -> Dataflow (Vertex i) lookupVertex (Edge (VertexID vindex)) =   Dataflow $ do     vertices <- gets dfsVertices -    return $ fromDyn (vertices ! vindex) (error "Programming error: Vertex and Edge were of different types")+    return $ unEraseType (vertices ! vindex)  -- | Store a provided vertex and obtain an 'Edge' that refers to it.-registerVertex :: Typeable i => Vertex i -> Dataflow (Edge i)+registerVertex :: Vertex i -> Dataflow (Edge i) registerVertex vertex =   Dataflow $ do     vid <- gets (dfsLastVertexID >>> inc)@@ -113,7 +120,7 @@    where     addVertex vtx vid s = s {-      dfsVertices     = dfsVertices s `snoc` toDyn vtx,+      dfsVertices     = dfsVertices s `snoc` EraseType vtx,       dfsLastVertexID = vid     } @@ -142,7 +149,7 @@ modifyState (StateRef ref) op = Dataflow $ lift $ atomicModifyIORef' ref (\x -> (op x, ()))  {-# INLINEABLE input #-}-input :: (Traversable t, Typeable i) => t i -> Edge i -> Dataflow ()+input :: Traversable t => t i -> Edge i -> Dataflow () input inputs next = do   timestamp <- Timestamp <$> incrementEpoch @@ -150,9 +157,9 @@    finalize timestamp -{-# NOINLINE send #-}+{-# INLINE send #-} -- | Send an `input` item to be worked on to the indicated vertex.-send :: Typeable input => Edge input -> Timestamp -> input -> Dataflow ()+send :: Edge input -> Timestamp -> input -> Dataflow () send e t i = lookupVertex e >>= invoke t i   where     invoke timestamp datum (StatefulVertex sref callback) = callback sref timestamp datum

@@ -11,7 +11,6 @@ import           Control.Concurrent.STM.TVar (TVar, modifyTVar') import           Control.Monad.STM           (atomically) import           Control.Monad.Trans.Class   (lift)-import           Data.Typeable               (Typeable) import           Dataflow.Primitives         (Dataflow (..), Edge, StateRef,                                               Timestamp (..), Vertex (..),                                               newState, registerFinalizer,@@ -26,7 +25,7 @@ -- -- NB: Until the finalizer has been called for a particular timestamp, a stateful vertex -- must be capable of accepting data for multiple timestamps simultaneously.-statefulVertex :: Typeable i =>+statefulVertex ::   state -- ^ The initial state value.   -> (StateRef state -> Timestamp -> i -> Dataflow ()) -- ^ The input handler.   -> (StateRef state -> Timestamp -> Dataflow ()) -- ^ The finalizer.@@ -41,12 +40,12 @@ -- -- `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.-statelessVertex :: Typeable i => (Timestamp -> i -> Dataflow ()) -> Dataflow (Edge i)+statelessVertex :: (Timestamp -> i -> Dataflow ()) -> Dataflow (Edge i) statelessVertex callback = registerVertex $ StatelessVertex callback  {-# NOINLINE outputTVar #-} -- | 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.-outputTVar :: Typeable o => (o -> w -> w) -> TVar w -> Dataflow (Edge o)+outputTVar :: (o -> w -> w) -> TVar w -> Dataflow (Edge o) outputTVar op register = statelessVertex $ \_ x -> Dataflow $ lift $ atomically $ modifyTVar' register (op x)

@@ -4,13 +4,12 @@  import           Control.Concurrent.STM.TVar (newTVarIO, readTVarIO) import           Control.Monad.IO.Class      (MonadIO (..))-import           Data.Typeable               (Typeable) import           Dataflow                    (Dataflow, Edge, compile, execute,                                               outputTVar) import           Prelude  -runDataflow :: (Typeable i, Typeable o, MonadIO io) => (Edge o -> Dataflow (Edge i)) -> [i] -> io [o]+runDataflow :: MonadIO io => (Edge o -> Dataflow (Edge i)) -> [i] -> io [o] runDataflow dataflow inputs =   liftIO $ do     out     <- newTVarIO []

@@ -3,7 +3,6 @@ module DataflowSpec (spec) where  import           Control.Monad   ((>=>))-import           Data.Typeable   (Typeable) import           Dataflow import           Prelude         hiding (map) @@ -27,7 +26,7 @@       \(numbers :: [Int]) ->         runDataflow (storeAndForward >=> storeAndForward >=> storeAndForward) numbers `shouldReturn` numbers -storeAndForward :: (Show i, Typeable i) => Edge i -> Dataflow (Edge i)+storeAndForward :: Edge i -> Dataflow (Edge i) storeAndForward next = statefulVertex [] store forward   where     store sref _ i = modifyState sref (i :)