diff --git a/Control-Engine.cabal b/Control-Engine.cabal --- a/Control-Engine.cabal +++ b/Control-Engine.cabal @@ -1,5 +1,5 @@ name: Control-Engine -version: 0.0.6 +version: 1.0.0.0 license: BSD3 license-file: LICENSE author: Thomas DuBuisson @@ -12,8 +12,8 @@ category: Control stability: stable build-type: Simple -cabal-version: >= 1.2 -tested-with: GHC == 6.8.3 +cabal-version: >= 1.6 +tested-with: GHC == 6.10.3 extra-source-files: Flag small_base @@ -21,9 +21,9 @@ Library if flag(small_base) - Build-Depends: base >= 3, stm + Build-Depends: base == 4.*, stm, concurrent, BoundedChan >= 1.0.0.2 else - Build-Depends: base >= 3, stm + Build-Depends: base == 4.*, stm, BoundedChan >= 1.0.0.2 hs-source-dirs: exposed-modules: Control.Engine, Control.ThreadPool ghc-options: diff --git a/Control/Engine.hs b/Control/Engine.hs --- a/Control/Engine.hs +++ b/Control/Engine.hs @@ -1,56 +1,74 @@ -{-| Implemented here is a thread pool library on crack. - +{-| /1.0 Introduction/ Typically, a thread pool is a set of execution contexts that will execute - tasks from an input queue. Typically, thread pools are used to parallize + tasks from an input queue. Thread pools are used to parallize the processing of incoming work across all available CPUs without going through the expense of starting a new thread for every new task. In 'Control.Engine' you will find a somewhat unique implementation. The 'Engine' is not only a set of threads running a common mutator on the input - queue, producing an output queue, but also include hooks, task injection, and - state management. + queue, placing results on an output queue, but also include hooks, task + injection, and state management. + @ + +--------+ chan1 +------------------------+ chan2 +--------+ + | In Hk + ---> | PreMH, Mutator, PostMH | -----> | Out Hk | + +--------+ +------------------------+ +--------+ + ^ ^ ^ + | | | + | | IO Ref | + +-----------------+-----------------------------+ + | + +------------+ + | State TVar | + +------------+ + @ - /Queues :: (Chan a)/ - from "Control.Concurrent.Chan". + /Queues :: (BoundedChan a)/ - from "Control.Concurrent.BoundedChan". + The system uses two primary queues. One for transporting data from + pre-mutator hooks to the mutator. One for data from the mutator to the + post-mutator hooks. These channels are size-bounded - which is needed + mostly due to the inflexibility of the scheduler. /Hooks :: (a -> IO Maybe a)/ Hooks can be added and removed during execution without creating a new engine. They allow the developer to modify tasks: - * prior to parallization (for sequential preprocessing) + * Input hooks - prior to parallization (for sequential preprocessing) - * in parallel, prior to main mutation funciton + * Pre-Mutator hooks - in parallel, prior to main mutation funciton - * in parallel, after mutation function + * Post-Mutator hooks - in parallel, after mutation function - * post parallization (for sequential post processing) + * Output hooks - post parallization (for sequential post processing) /State Management/ + Control-Engine manages state for you. Semantically, all workers and hooks + will see a correct state but it won't always be the most recent or consistent + between threads. + The stateManager waits for any updates to the mutator state or hooks. If any - modifications are made then the new set of hooks (or state) is provided + modifications are made then the new set of hooks or state is provided to the workers. Correctness is handled by keeping the master copies as TVars - ("Control.Concurrent.STM"). While the mutators and hooks read from an 'MVar' - ("Control.Concurrent.MVar") to avoid contention. + ("Control.Concurrent.STM"). While the mutators and hooks read state from an + 'IORef' ("Control.Concurrent.IORef") to avoid contention. The thinking here is that changing the hooks and state is a rare / low contention action while the need for this information will be constant and performance critical. How successful this stratagy is has yet to be shown. - - /Injection/ - One injection point allows injection of a result that had no preceding - task. The second injector allows the initial hooks ('Input' hooks) to be - bypassed. + One injection point allows injection of a result that had no preceding task - + thus the result is only seen by the output hooks. Another injector allows the + input hooks to be bypassed. -} module Control.Engine @@ -75,17 +93,17 @@ , injectPostMutator ) where -import Control.Concurrent +import Control.Concurrent (forkIO) import Control.Concurrent.STM -import Control.Concurrent.Chan +import Control.Concurrent.BoundedChan import Control.Monad import Data.IORef import Data.List (insert) -- |An 'Engine' represents a pool of threads ready to execute tasks. data Engine job result state = - Eng { chan1 :: Chan job - , chan2 :: Chan result + Eng { chan1 :: BoundedChan job + , chan2 :: BoundedChan result , tvInHook :: TVar [Hook state job] , tvPreMutateHook :: TVar [Hook state job] , tvPostMutateHook :: TVar [Hook state result] @@ -106,28 +124,39 @@ -- You should consider using "Control.ThreadPool" instead. -- -- Evaluation of the result is forced using seq. -initSimpleEngine :: Int -> (job -> result) -> IO (Chan job, Chan result) +-- Input, output, and intermediate channels are length bounded to a multiple +-- of the number of workers. +initSimpleEngine :: Int -> (job -> result) -> IO (BoundedChan job, BoundedChan result) initSimpleEngine nr mutator = do - input <- newChan - output <- newChan + input <- newBoundedChan chanBound + output <- newBoundedChan chanBound let m = const (return . Just . mutator) - initEngine nr (readChan input) (\o -> o `seq` writeChan output o) m () + initEngine nr chanBound (readChan input) (\o -> o `seq` writeChan output o) m () return (input, output) + where + chanBound = nr * 8 -- |Simpler than calling 'initEngine', but it allows no state or interaction -- with the hooks and injectors. No strictness is forced. -initSimpleEngineIO :: Int -> (job -> IO result) -> IO (Chan job, Chan result) +-- +-- Input, output, and intermediate channels are length bounded to a multiple +-- of the number of workers. +initSimpleEngineIO :: Int -> (job -> IO result) -> IO (BoundedChan job, BoundedChan result) initSimpleEngineIO nr mutator = do - input <- newChan - output <- newChan + input <- newBoundedChan chanBound + output <- newBoundedChan chanBound let m = (\_ j -> mutator j >>= return . Just) - initEngine nr (readChan input) (writeChan output) m () + initEngine nr chanBound (readChan input) (writeChan output) m () return (input, output) + where + chanBound = nr * 8 -- |To initilize an engine you must provide: -- -- * the number of threads -- +-- * the maxiumum channel size for intermediate channels +-- -- * an action that will get the input -- -- * an action that will consume output @@ -138,10 +167,17 @@ -- -- No strictness is forced - be sure you force evaluation if wanted. -- All hooks start out empty. -initEngine :: (Eq st) => Int -> (IO job) -> (result -> IO ()) -> (st -> job -> IO (Maybe result)) -> st -> IO (Engine job result st) -initEngine nrWorkers input output mutator initialState = do - c1 <- newChan - c2 <- newChan +initEngine :: (Eq st) => + Int -> + Int -> + (IO job) -> + (result -> IO ()) -> + (st -> job -> IO (Maybe result)) -> + st -> + IO (Engine job result st) +initEngine nrWorkers chanBound input output mutator initialState = do + c1 <- newBoundedChan chanBound + c2 <- newBoundedChan chanBound inputHooks <- newIORef [] outputHooks <- newIORef [] @@ -164,7 +200,7 @@ forM_ [1..nrWorkers] $ \_ -> forkIO (worker c1 eref mutator c2) return engine -worker :: Chan job -> RefEngine job result st -> (st -> job -> IO (Maybe result)) -> Chan result -> IO () +worker :: BoundedChan job -> RefEngine job result st -> (st -> job -> IO (Maybe result)) -> BoundedChan result -> IO () worker c1 eref mutator c2 = forever $ readChan c1 >>= worker' where worker' msg = do @@ -207,7 +243,7 @@ updateState new -- Input.hs -inputManager :: (IO job) -> Chan job -> RefEngine job result st -> IO () +inputManager :: (IO job) -> BoundedChan job -> RefEngine job result st -> IO () inputManager input outChan eref = forever $ input >>= handleMsg where handleMsg msg = do @@ -219,7 +255,7 @@ Nothing -> return () -- Output.hs -outputManager :: (result -> IO ()) -> Chan result -> RefEngine job result state -> IO () +outputManager :: (result -> IO ()) -> BoundedChan result -> RefEngine job result state -> IO () outputManager output msgChan eref = forever $ do m <- readChan msgChan hook <- readIORef (refOutHook eref)