Control-Engine 0.0.6 → 1.0.0.0

raw patch · 2 files changed

+78/−42 lines, 2 filesdep +BoundedChandep +concurrentdep ~basePVP ok

version bump matches the API change (PVP)

Dependencies added: BoundedChan, concurrent

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• Control-Engine.cabal +5/−5
• Control/Engine.hs +73/−37

@@ -1,5 +1,5 @@ name:		Control-Engine-version:	0.0.6+version:	1.0.0.0 license:	BSD3 license-file:	LICENSE author:		Thomas DuBuisson <thomas.dubuisson@gmail.com>@@ -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:

@@ -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)