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.1 +version: 0.0.2 license: BSD3 license-file: LICENSE author: Thomas DuBuisson diff --git a/Control/Engine.hs b/Control/Engine.hs --- a/Control/Engine.hs +++ b/Control/Engine.hs @@ -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)