diff --git a/benchmark/Main.hs b/benchmark/Main.hs --- a/benchmark/Main.hs +++ b/benchmark/Main.hs @@ -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 () diff --git a/dataflower.cabal b/dataflower.cabal --- a/dataflower.cabal +++ b/dataflower.cabal @@ -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 diff --git a/src/Dataflow.hs b/src/Dataflow.hs --- a/src/Dataflow.hs +++ b/src/Dataflow.hs @@ -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 diff --git a/src/Dataflow/Primitives.hs b/src/Dataflow/Primitives.hs --- a/src/Dataflow/Primitives.hs +++ b/src/Dataflow/Primitives.hs @@ -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 diff --git a/src/Dataflow/Vertices.hs b/src/Dataflow/Vertices.hs --- a/src/Dataflow/Vertices.hs +++ b/src/Dataflow/Vertices.hs @@ -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) diff --git a/src/Test/Dataflow.hs b/src/Test/Dataflow.hs --- a/src/Test/Dataflow.hs +++ b/src/Test/Dataflow.hs @@ -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 [] diff --git a/test/DataflowSpec.hs b/test/DataflowSpec.hs --- a/test/DataflowSpec.hs +++ b/test/DataflowSpec.hs @@ -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 :)