packages feed

dataflower (empty) → 0.1.0.0

raw patch · 10 files changed

+463/−0 lines, 10 filesdep +QuickCheckdep +basedep +criterionsetup-changed

Dependencies added: QuickCheck, base, criterion, dataflower, hashable, hspec, mtl, stm, time, transformers, vector

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Double Crown Gaming Co. (c) 2020++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.++    * Neither the name of Double Crown Gaming Co. nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++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+OWNER 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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ benchmark/Main.hs view
@@ -0,0 +1,22 @@+module Main where++import           Criterion.Main+import           Data.Typeable  (Typeable)+import           Dataflow+import           Prelude+import           Test.Dataflow  (runDataflow)+++main :: IO ()+main = defaultMain [+  bgroup "dataflow" [+      bench "passthrough 1,000,000 inputs" $ nfIO (runDataflow passthrough [0..1000000 :: Int]),+      bench "discard     1,000,000 inputs" $ nfIO (runDataflow blackhole [0..1000000 :: Int])+    ]+  ]+  where+    passthrough :: Typeable a => Edge a -> Dataflow (Edge a)+    passthrough next = statelessVertex $ \t x -> send next t x++    blackhole :: Typeable a => Edge a -> Dataflow (Edge a)+    blackhole _ = statelessVertex $ \_ _ -> return ()
+ dataflower.cabal view
@@ -0,0 +1,78 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.31.2.+--+-- see: https://github.com/sol/hpack+--+-- hash: 8f7b0ebc453194c8e6a5cca3c5cef03c62cfa135a28e2c15fcf3ff5a58e10d3a++name:           dataflower+version:        0.1.0.0+synopsis:       A Pure-Haskell Timely Dataflow System+description:    See README+homepage:       https://github.com/doublecrowngaming/dataflower#readme+bug-reports:    https://github.com/doublecrowngaming/dataflower/issues+maintainer:     Jesse Kempf+license:        BSD3+license-file:   LICENSE+build-type:     Simple++source-repository head+  type: git+  location: https://github.com/doublecrowngaming/dataflower++library+  exposed-modules:+      Dataflow+      Test.Dataflow+  other-modules:+      Dataflow.Primitives+      Dataflow.Vertices+      Paths_dataflower+  hs-source-dirs:+      src+  default-extensions: StrictData+  ghc-options: -Wall -O2+  build-depends:+      base >=4.7 && <5+    , hashable+    , mtl+    , stm+    , time+    , transformers+    , vector+  default-language: Haskell2010++test-suite spec+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules:+      DataflowSpec+      Paths_dataflower+  hs-source-dirs:+      test+  default-extensions: StrictData+  ghc-options: -Wall -O2+  build-depends:+      QuickCheck+    , base >=4.7 && <5+    , dataflower+    , hspec >=2.0.0+    , stm+  default-language: Haskell2010++benchmark performance+  type: exitcode-stdio-1.0+  main-is: Main.hs+  other-modules:+      Paths_dataflower+  hs-source-dirs:+      benchmark+  default-extensions: StrictData+  ghc-options: -Wall -O2 -O2+  build-depends:+      base >=4.7 && <5+    , criterion+    , dataflower+    , stm+  default-language: Haskell2010
+ src/Dataflow.hs view
@@ -0,0 +1,55 @@+{-# LANGUAGE RecordWildCards #-}++{-|+Module      : Dataflow+Description : Timely Dataflow for Haskell+Copyright   : (c) Double Crown Gaming Co. 2020+License     : BSD3+Maintainer  : jesse.kempf@doublecrown.co+Stability   : experimental++Timely Dataflow in pure Haskell.+-}++module Dataflow (+  Dataflow,+  Edge,+  Timestamp,+  StateRef,+  send,+  readState,+  writeState,+  modifyState,+  statefulVertex,+  statelessVertex,+  outputTVar,+  Program,+  compile,+  execute+) where++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++-- | A 'Program' represents a fully-preprocessed 'Dataflow' that may be+-- executed against inputs.+data Program i = Program {+  programInput :: Edge i,+  programState :: DataflowState+}++-- | Take a 'Dataflow' which takes 'i's as input and compile it into a 'Program'.+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.+execute :: (Traversable t, Typeable i) => t i -> Program i -> IO ()+execute corpus Program{..} = execDataflow feedInput+  where+    feedInput           = input corpus programInput+    execDataflow action = void $ execStateT (runDataflow action) programState
+ src/Dataflow/Primitives.hs view
@@ -0,0 +1,166 @@+{-# LANGUAGE ExistentialQuantification  #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses      #-}++module Dataflow.Primitives (+  Dataflow(..),+  DataflowState,+  Vertex(..),+  initDataflowState,+  StateRef,+  newState,+  readState,+  writeState,+  modifyState,+  Edge,+  Timestamp(..),+  registerVertex,+  registerFinalizer,+  incrementEpoch,+  input,+  send,+  finalize+) where++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,+                                             readIORef)+import           Data.Vector                (Vector, empty, snoc, (!))+import           Numeric.Natural            (Natural)+import           Prelude+++newtype VertexID    = VertexID        Int deriving (Eq, Ord, Show)+newtype Epoch       = Epoch       Natural deriving (Eq, Ord, Hashable, Show)++-- | 'Timestamp's represent instants in the causal timeline.+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.+newtype Edge a      = Edge       VertexID++-- | Class of entities that can be incremented by one.+class Incrementable a where+  inc :: a -> a++instance Incrementable VertexID where+  inc (VertexID n) = VertexID (n + 1)++instance Incrementable Epoch where+  inc (Epoch n) = Epoch (n + 1)+++data DataflowState = DataflowState {+  dfsVertices       :: Vector Dynamic,+  dfsFinalizers     :: [Timestamp -> Dataflow ()],+  dfsLastVertexID   :: VertexID,+  dfsLastInputEpoch :: Epoch+}++-- | `Dataflow` is the type of all dataflow operations.+newtype Dataflow a = Dataflow { runDataflow :: StateT DataflowState IO a }+  deriving (Functor, Applicative, Monad)++initDataflowState :: DataflowState+initDataflowState = DataflowState {+  dfsVertices       = empty,+  dfsFinalizers     = [],+  dfsLastVertexID   = VertexID (-1),+  dfsLastInputEpoch = Epoch 0+}++-- | Get the next input Epoch.+incrementEpoch :: Dataflow Epoch+incrementEpoch =+  Dataflow $ do+    epoch <- gets (dfsLastInputEpoch >>> inc)++    modify $ \s -> s { dfsLastInputEpoch = epoch }++    return epoch+++data Vertex i = forall s.+    StatefulVertex+      (StateRef s)+      (StateRef s -> Timestamp -> i -> Dataflow ())+  | StatelessVertex+      (Timestamp -> i -> Dataflow ())++-- | Retrieve the vertex for a given edge.+lookupVertex :: Typeable i => 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")++-- | Store a provided vertex and obtain an 'Edge' that refers to it.+registerVertex :: Typeable i => Vertex i -> Dataflow (Edge i)+registerVertex vertex =+  Dataflow $ do+    vid <- gets (dfsLastVertexID >>> inc)++    modify $ addVertex vertex vid++    return (Edge vid)++  where+    addVertex vtx vid s = s {+      dfsVertices     = dfsVertices s `snoc` toDyn vtx,+      dfsLastVertexID = vid+    }++-- | Store a provided finalizer.+registerFinalizer :: (Timestamp -> Dataflow ()) -> Dataflow ()+registerFinalizer finalizer =+  Dataflow $ modify $ \s -> s { dfsFinalizers = finalizer : dfsFinalizers s }++-- | Mutable state that holds an `a`.+newtype StateRef a = StateRef (IORef a)++-- | Create a `StateRef` initialized to the provided `a`.+newState :: a -> Dataflow (StateRef a)+newState a = StateRef <$> (Dataflow $ lift $ newIORef a)++-- | Read the value stored in the `StateRef`.+readState :: StateRef a -> Dataflow a+readState (StateRef ref) = Dataflow $ lift (readIORef ref)++-- | Overwrite the value stored in the `StateRef`.+writeState :: StateRef a -> a -> Dataflow ()+writeState (StateRef ref) x = Dataflow $ lift $ atomicWriteIORef ref x++-- | Update the value stored in `StateRef`.+modifyState :: StateRef a -> (a -> a) -> Dataflow ()+modifyState (StateRef ref) op = Dataflow $ lift $ atomicModifyIORef' ref (\x -> (op x, ()))++{-# INLINEABLE input #-}+input :: (Traversable t, Typeable i) => t i -> Edge i -> Dataflow ()+input inputs next = do+  timestamp <- Timestamp <$> incrementEpoch++  mapM_ (send next timestamp) inputs++  finalize timestamp++{-# NOINLINE send #-}+-- | Send an `input` item to be worked on to the indicated vertex.+send :: Typeable input => 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+    invoke timestamp datum (StatelessVertex callback)     = callback timestamp datum++-- Notify all relevant vertices that no more input is coming for `Timestamp`.+finalize :: Timestamp -> Dataflow ()+finalize t = do+  finalizers <- Dataflow $ gets dfsFinalizers++  mapM_ (\p -> p t) finalizers
+ src/Dataflow/Vertices.hs view
@@ -0,0 +1,52 @@+{-# LANGUAGE ExplicitForAll      #-}+{-# LANGUAGE OverloadedLists     #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Dataflow.Vertices (+  statefulVertex,+  statelessVertex,+  outputTVar+) where++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,+                                              registerVertex)+import           Prelude+++-- | Construct a vertex with internal state. Like 'statelessVertex', 'statefulVertex'+-- requires a procedure to invoke on each input. It also needs an initial 'state' value+-- and a procedure to call when all inputs for a given 'Timestamp' value have been+-- delivered.+--+-- 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 =>+  state -- ^ The initial state value.+  -> (StateRef state -> Timestamp -> i -> Dataflow ()) -- ^ The input handler.+  -> (StateRef state -> Timestamp -> Dataflow ()) -- ^ The finalizer.+  -> Dataflow (Edge i)+statefulVertex initState callback finalizer = do+  stateRef <- newState initState++  registerFinalizer $ finalizer stateRef+  registerVertex    $ StatefulVertex stateRef callback++-- | Construct a vertex with no internal state. The given procedure is invoked on each input.+--+-- `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 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 op register = statelessVertex $ \_ x -> Dataflow $ lift $ atomically $ modifyTVar' register (op x)
+ src/Test/Dataflow.hs view
@@ -0,0 +1,21 @@+module Test.Dataflow (+  runDataflow+) where++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 dataflow inputs =+  liftIO $ do+    out     <- newTVarIO []+    program <- compile (dataflow =<< outputTVar (:) out)++    execute inputs program++    reverse <$> readTVarIO out
+ test/DataflowSpec.hs view
@@ -0,0 +1,36 @@+{-# LANGUAGE ScopedTypeVariables #-}++module DataflowSpec (spec) where++import           Control.Monad   ((>=>))+import           Data.Typeable   (Typeable)+import           Dataflow+import           Prelude         hiding (map)++import           Test.Dataflow   (runDataflow)+import           Test.Hspec+import           Test.QuickCheck+++spec :: Spec+spec = do+  it "can pass through data without modification" $ property $ do+    let passthrough next = statelessVertex $ \t x -> send next t x++    \(numbers :: [Integer]) -> runDataflow passthrough numbers `shouldReturn` numbers++  describe "finalize" $ do+    it "finalizes vertices" $ property $+      \(numbers :: [Int]) -> runDataflow storeAndForward numbers `shouldReturn` numbers++    it "finalizes vertices in the correct order" $ property $+      \(numbers :: [Int]) ->+        runDataflow (storeAndForward >=> storeAndForward >=> storeAndForward) numbers `shouldReturn` numbers++storeAndForward :: (Show i, Typeable i) => Edge i -> Dataflow (Edge i)+storeAndForward next = statefulVertex [] store forward+  where+    store sref _ i = modifyState sref (i :)+    forward sref t = do+      mapM_ (send next t) =<< reverse <$> readState sref+      writeState sref []
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}