Control-Engine 0.0.1 → 0.0.2

raw patch · 2 files changed

+65/−39 lines, 2 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Control-Engine.cabal +1/−1
• Control/Engine.hs +64/−38

@@ -1,5 +1,5 @@ name:		Control-Engine-version:	0.0.1+version:	0.0.2 license:	BSD3 license-file:	LICENSE author:		Thomas DuBuisson <thomas.dubuisson@gmail.com>

@@ -1,38 +1,47 @@-{- | 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- - 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.- -- - 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)- - * in parallel, prior to main mutation funciton- - * in parallel, after mutation function- - * post parallization (for sequential post processing)- -- - State Management- - 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- - to the workers.  This allows the state of the entire engine to be atomically- - modified (it is all STM) but allows the workers to use cheap and quick- - MVars.- -- - The thinking here is that changing the hooks and state is a rare / low+{-| 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+  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.++  Queues :: (Chan a)+  From "Control.Concurrent.Chan". ++  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)++   * in parallel, prior to main mutation funciton++   * in parallel, after mutation function++   * post parallization (for sequential post processing)+ +  State Management+  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+  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.++  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.- -- - Injection- - One injection point allows injection of a 'result' that had no preceding- - 'task'.  With another the initial hooks ('Input' hooks) can be bypassed.+  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.  -}  module Control.Engine@@ -81,7 +90,7 @@     }  -- |If all you want is a basic thread pool, this will work.--- You should consider using Control.ThreadPool instead.+-- You should consider using "Control.ThreadPool" instead. -- -- Evaluation of the result is forced using seq. initSimpleEngine :: Int -> (job -> result) -> IO (Chan job, Chan result)@@ -103,10 +112,15 @@ 	return (input, output)  -- |To initilize an engine you must provide:+-- --    * the number of threads+-- --    * an action that will get the input+-- --    * an action that will consume output+-- --    * a mutator function to perform on all inputs+-- --    * an initial state for the mutator function -- -- No strictness is forced - be sure you force evaluation if wanted.@@ -210,9 +224,9 @@ 		Nothing -> return ()  -- Hooks.hs--- A hook is simply a mutation on the task.  To order the hooks they all have--- priorities (lower value priorites happen first).  For accounting and to--- remove old hooks there is a description field.+-- |A hook is simply a mutation on the task.  The priority is used to order+-- hook execution (lower value priorites happen first).  For accounting and to+-- remove old hooks the description field is used. data Hook st msg = Hk 		{ hkFunc	:: st -> msg -> IO (Maybe msg) 		, hkPriority	:: Int@@ -237,34 +251,46 @@   apply Nothing  f = return Nothing   apply (Just a) f = (hkFunc f st) a +-- |Adds a hook that will be performed in serial on all jobs added to+-- the input queue. addInputHook :: Engine job result state -> Hook state job -> IO () addInputHook e h = atomically $ do 	readTVar (tvInHook e) >>= writeTVar (tvInHook e) . insert h +-- |Adds a hook that will be performed in serial on all results+-- before they are added to the output queue. addOutputHook :: Engine job result state -> Hook state result -> IO () addOutputHook e h = atomically $ do 	readTVar (tvOutHook e) >>= writeTVar (tvOutHook e) . insert h +-- |Adds a hook that will be performed in parallel before the main mutator+-- function. addPreMutateHook :: Engine job result state -> Hook state job -> IO () addPreMutateHook e h = atomically $ do 	readTVar (tvPreMutateHook e) >>= writeTVar (tvPreMutateHook e) . insert h +-- |Adds a hook that will be performed in parallel after the main mutator+-- function. addPostMutateHook :: Engine job result state -> Hook state result -> IO () addPostMutateHook e h = atomically $ do 	readTVar (tvPostMutateHook e) >>= writeTVar (tvPostMutateHook e) . insert h +-- |Deletes all input hooks matching the provided desciption delInputHook :: Engine j r s -> String -> IO () delInputHook e s = atomically $ do 	readTVar (tvInHook e) >>= writeTVar (tvInHook e) . filter ( (/= s) . hkDescription) +-- |Deletes all pre-mutate hooks matching the provided desciption delPreMutateHook :: Engine j r s -> String -> IO () delPreMutateHook e s = atomically $ do 	readTVar (tvPreMutateHook e) >>= writeTVar (tvPreMutateHook e) . filter ( (/= s) . hkDescription) +-- |Deletes all post-mutate hooks matching the provided desciption delPostMutateHook :: Engine j r s -> String -> IO () delPostMutateHook e s = atomically $ do 	readTVar (tvPostMutateHook e) >>= writeTVar (tvPostMutateHook e) . filter ( (/= s) . hkDescription) +-- |Deletes all output hooks matching the provided desciption delOutputHook :: Engine j r s -> String -> IO () delOutputHook e s = atomically $ do 	readTVar (tvOutHook e) >>= writeTVar (tvOutHook e) . filter ( (/= s) . hkDescription)