scc-0.2: Control/Concurrent/SCC/ComponentTypes.hs
{-
Copyright 2008 Mario Blazevic
This file is part of the Streaming Component Combinators (SCC) project.
The SCC project is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later
version.
SCC is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License along with SCC. If not, see
<http://www.gnu.org/licenses/>.
-}
{-# LANGUAGE ScopedTypeVariables, MultiParamTypeClasses, FlexibleInstances, FunctionalDependencies,
ExistentialQuantification, KindSignatures, Rank2Types, PatternSignatures #-}
module Control.Concurrent.SCC.ComponentTypes
(-- * Classes
Component (..), BranchComponent (combineBranches),
-- * Types
AnyComponent (AnyComponent), Performer (..), Consumer (..), Producer(..), Splitter(..), Transducer(..),
ComponentConfiguration(..),
-- * Lifting functions
liftPerformer, liftConsumer, liftAtomicConsumer, liftProducer, liftAtomicProducer,
liftTransducer, liftAtomicTransducer, lift121Transducer, liftStatelessTransducer, liftFoldTransducer, liftStatefulTransducer,
liftSimpleSplitter, liftSectionSplitter, liftAtomicSimpleSplitter, liftAtomicSectionSplitter, liftStatelessSplitter,
-- * Utility functions
showComponentTree, optimalTwoParallelConfigurations, optimalTwoSequentialConfigurations, optimalThreeParallelConfigurations
)
where
import Control.Concurrent.SCC.Foundation
import Control.Monad (liftM, when)
import Data.List (minimumBy)
import Data.Maybe
import Data.Typeable (Typeable, cast)
-- | 'AnyComponent' is an existential type wrapper around a 'Component'.
data AnyComponent = forall a. Component a => AnyComponent a
-- | The types of 'Component' class carry metadata and can be configured to use a specific number of threads.
class Component c where
name :: c -> String
-- | Returns the list of all children components.
subComponents :: c -> [AnyComponent]
-- | Returns the maximum number of threads that can be used by the component.
maxUsableThreads :: c -> Int
-- | Configures the component to use the specified number of threads. This function affects 'usedThreads', 'cost',
-- and 'subComponents' methods of the result, while 'name' and 'maxUsableThreads' remain the same.
usingThreads :: Int -> c -> c
-- | The number of threads that the component is configured to use. By default the number is usually 1.
usedThreads :: c -> Int
-- | The cost of using the component as configured.
cost :: c -> Int
cost c = 1 + sum (map cost (subComponents c))
instance Component AnyComponent where
name (AnyComponent c) = name c
subComponents (AnyComponent c) = subComponents c
maxUsableThreads (AnyComponent c) = maxUsableThreads c
usingThreads n (AnyComponent c) = AnyComponent (usingThreads n c)
usedThreads (AnyComponent c) = usedThreads c
cost (AnyComponent c) = cost c
-- | Show details of the given component's configuration.
showComponentTree :: forall c. Component c => c -> String
showComponentTree c = showIndentedComponent 1 c
showIndentedComponent :: forall c. Component c => Int -> c -> String
showIndentedComponent depth c = showRightAligned 4 (cost c) ++ showRightAligned 3 (usedThreads c) ++ replicate depth ' '
++ name c ++ "\n"
++ concatMap (showIndentedComponent (succ depth)) (subComponents c)
showRightAligned :: Show x => Int -> x -> String
showRightAligned width x = let str = show x
in replicate (width - length str) ' ' ++ str
data ComponentConfiguration = ComponentConfiguration {componentChildren :: [AnyComponent],
componentThreads :: Int,
componentCost :: Int}
-- | A component that performs a computation with no inputs nor outputs is a 'Performer'.
data Performer m r = Performer {performerName :: String,
performerMaxThreads :: Int,
performerConfiguration :: ComponentConfiguration,
performerUsingThreads :: Int -> (ComponentConfiguration, forall c. Pipe c m r),
perform :: forall c. Pipe c m r}
-- | A component that consumes values from a 'Source' is called 'Consumer'.
-- data Consumer m x r = Consumer {consumerData :: ComponentData (forall c. Source c x -> Pipe c m r),
-- consume :: forall c. Source c x -> Pipe c m r}
data Consumer m x r = Consumer {consumerName :: String,
consumerMaxThreads :: Int,
consumerConfiguration :: ComponentConfiguration,
consumerUsingThreads :: Int -> (ComponentConfiguration, forall c. Source c x -> Pipe c m r),
consume :: forall c. Source c x -> Pipe c m r}
-- | A component that produces values and puts them into a 'Sink' is called 'Producer'.
data Producer m x r = Producer {producerName :: String,
producerMaxThreads :: Int,
producerConfiguration :: ComponentConfiguration,
producerUsingThreads :: Int -> (ComponentConfiguration, forall c. Sink c x -> Pipe c m r),
produce :: forall c. Sink c x -> Pipe c m r}
-- | The 'Transducer' type represents computations that transform data and return no result.
-- A transducer must continue consuming the given source and feeding the sink while there is data.
data Transducer m x y = Transducer {transducerName :: String,
transducerMaxThreads :: Int,
transducerConfiguration :: ComponentConfiguration,
transducerUsingThreads :: Int -> (ComponentConfiguration,
forall c. Source c x -> Sink c y -> Pipe c m [x]),
transduce :: forall c. Source c x -> Sink c y -> Pipe c m [x]}
-- | The 'Splitter' type represents computations that distribute data acording to some criteria. A splitter should
-- distribute only the original input data, and feed it into the sinks in the same order it has been read from the
-- source. If the two sink arguments of a splitter are the same, the splitter must act as an identity transform.
data Splitter m x = Splitter {splitterName :: String,
splitterMaxThreads :: Int,
splitterConfiguration :: ComponentConfiguration,
splitterUsingThreads :: Int -> (ComponentConfiguration,
forall c. Source c x -> Sink c x -> Sink c x -> Pipe c m [x],
forall c. Source c x -> Sink c (Maybe x) -> Sink c (Maybe x)
-> Pipe c m [x]),
split :: forall c. Source c x -> Sink c x -> Sink c x -> Pipe c m [x],
splitSections :: forall c. Source c x -> Sink c (Maybe x) -> Sink c (Maybe x) -> Pipe c m [x]}
instance Component (Performer m r) where
name = performerName
subComponents = componentChildren . performerConfiguration
maxUsableThreads = performerMaxThreads
usedThreads = componentThreads . performerConfiguration
usingThreads threads performer = let (configuration', perform' :: forall c. Pipe c m r) = performerUsingThreads performer threads
in performer{performerConfiguration= configuration', perform= perform'}
cost = componentCost . performerConfiguration
instance Component (Consumer m x r) where
name = consumerName
subComponents = componentChildren . consumerConfiguration
maxUsableThreads = consumerMaxThreads
usedThreads = componentThreads . consumerConfiguration
usingThreads threads consumer = let (configuration',
consume' :: forall c. Source c x -> Pipe c m r) = consumerUsingThreads consumer threads
in consumer{consumerConfiguration= configuration', consume= consume'}
cost = componentCost . consumerConfiguration
instance Component (Producer m x r) where
name = producerName
subComponents = componentChildren . producerConfiguration
maxUsableThreads = producerMaxThreads
usedThreads = componentThreads . producerConfiguration
usingThreads threads producer = let (configuration',
produce' :: forall c. Sink c x -> Pipe c m r) = producerUsingThreads producer threads
in producer{producerConfiguration= configuration', produce= produce'}
cost = componentCost . producerConfiguration
instance Component (Transducer m x y) where
name = transducerName
subComponents = componentChildren . transducerConfiguration
maxUsableThreads = transducerMaxThreads
usedThreads = componentThreads . transducerConfiguration
usingThreads threads transducer = let (configuration', transduce' :: forall c. Source c x -> Sink c y -> Pipe c m [x])
= transducerUsingThreads transducer threads
in transducer{transducerConfiguration= configuration', transduce= transduce'}
cost = componentCost . transducerConfiguration
instance Component (Splitter m x) where
name = splitterName
subComponents = componentChildren . splitterConfiguration
maxUsableThreads = splitterMaxThreads
usedThreads = componentThreads . splitterConfiguration
usingThreads threads splitter = let (configuration',
split' :: forall c. Source c x -> Sink c x -> Sink c x -> Pipe c m [x],
splitSections' :: forall c. Source c x -> Sink c (Maybe x) -> Sink c (Maybe x)
-> Pipe c m [x])
= splitterUsingThreads splitter threads
in splitter{splitterConfiguration= configuration',
split= split', splitSections= splitSections'}
cost = componentCost . splitterConfiguration
-- | 'BranchComponent' is a type class representing all components that can act as consumers, namely 'Consumer',
-- 'Transducer', and 'Splitter'.
class BranchComponent cc m x r | cc -> m x where
-- | 'combineBranches' is used to combine two components in 'BranchComponent' class into one, using the
-- given 'Consumer' binary combinator.
combineBranches :: String -> Int
-> (forall c. Bool -> (Source c x -> Pipe c m r) -> (Source c x -> Pipe c m r) -> (Source c x -> Pipe c m r))
-> cc -> cc -> cc
instance forall m x r. Monad m => BranchComponent (Consumer m x r) m x r where
combineBranches name cost combinator c1 c2 = liftConsumer name 1 $
\threads-> (ComponentConfiguration [AnyComponent c1, AnyComponent c2] 1 cost,
combinator False (consume c1) (consume c2))
instance forall m x. Monad m => BranchComponent (Consumer m x ()) m x [x] where
combineBranches name cost combinator c1 c2 = liftConsumer name 1 $
\threads-> (ComponentConfiguration [AnyComponent c1, AnyComponent c2] 1 cost,
liftM (const ())
. combinator False
(\source-> consume c1 source >> return [])
(\source-> consume c2 source >> return []))
instance forall m x y. BranchComponent (Transducer m x y) m x [x] where
combineBranches name cost combinator t1 t2
= liftTransducer name (maxUsableThreads t1 + maxUsableThreads t2) $
\threads-> let (configuration, t1', t2', parallel) = optimalTwoParallelConfigurations threads t1 t2
transduce' source sink = combinator parallel
(\source-> transduce t1 source sink)
(\source-> transduce t2 source sink)
source
in (configuration, transduce')
instance forall m x. (ParallelizableMonad m, Typeable x) => BranchComponent (Splitter m x) m x [x] where
combineBranches name cost combinator s1 s2
= liftSimpleSplitter name (maxUsableThreads s1 + maxUsableThreads s2) $
\threads-> let (configuration, s1', s2', parallel) = optimalTwoParallelConfigurations threads s1 s2
split' source true false = combinator parallel
(\source-> split s1 source true false)
(\source-> split s2 source true false)
source
in (configuration, split')
-- | Function 'liftPerformer' takes a component name, maximum number of threads it can use, and its 'usingThreads'
-- method, and returns a 'Performer' component.
liftPerformer :: String -> Int -> (Int -> (ComponentConfiguration, forall c. Pipe c m r)) -> Performer m r
liftPerformer name maxThreads usingThreads = case usingThreads 1
of (configuration, perform) -> Performer name maxThreads configuration
usingThreads perform
-- | Function 'liftConsumer' takes a component name, maximum number of threads it can use, and its 'usingThreads'
-- method, and returns a 'Consumer' component.
liftConsumer :: String -> Int -> (Int -> (ComponentConfiguration, forall c. Source c x -> Pipe c m r)) -> Consumer m x r
liftConsumer name maxThreads usingThreads = case usingThreads 1
of (configuration, consume) -> Consumer name maxThreads configuration
usingThreads consume
-- | Function 'liftProducer' takes a component name, maximum number of threads it can use, and its 'usingThreads'
-- method, and returns a 'Producer' component.
liftProducer :: String -> Int -> (Int -> (ComponentConfiguration, forall c. Sink c x -> Pipe c m r)) -> Producer m x r
liftProducer name maxThreads usingThreads = case usingThreads 1
of (configuration, produce) -> Producer name maxThreads configuration
usingThreads produce
-- | Function 'liftTransducer' takes a component name, maximum number of threads it can use, and its 'usingThreads'
-- method, and returns a 'Transducer' component.
liftTransducer :: String -> Int -> (Int -> (ComponentConfiguration, forall c. Source c x -> Sink c y -> Pipe c m [x]))
-> Transducer m x y
liftTransducer name maxThreads usingThreads = case usingThreads 1
of (configuration, transduce) -> Transducer name maxThreads configuration
usingThreads transduce
-- | Function 'liftAtomicConsumer' lifts a single-threaded 'consume' function into a 'Consumer' component.
liftAtomicConsumer :: String -> Int -> (forall c. Source c x -> Pipe c m r) -> Consumer m x r
liftAtomicConsumer name cost consume = liftConsumer name 1 (\_threads-> (ComponentConfiguration [] 1 cost, consume))
-- | Function 'liftAtomicProducer' lifts a single-threaded 'produce' function into a 'Producer' component.
liftAtomicProducer :: String -> Int -> (forall c. Sink c x -> Pipe c m r) -> Producer m x r
liftAtomicProducer name cost produce = liftProducer name 1 (\_threads-> (ComponentConfiguration [] 1 cost, produce))
-- | Function 'liftAtomicTransducer' lifts a single-threaded 'transduce' function into a 'Transducer' component.
liftAtomicTransducer :: String -> Int -> (forall c. Source c x -> Sink c y -> Pipe c m [x]) -> Transducer m x y
liftAtomicTransducer name cost transduce = liftTransducer name 1 (\_threads-> (ComponentConfiguration [] 1 cost, transduce))
-- | Function 'lift121Transducer' takes a function that maps one input value to one output value each, and lifts it into
-- a 'Transducer'.
lift121Transducer :: (Monad m, Typeable x, Typeable y) => String -> (x -> y) -> Transducer m x y
lift121Transducer name f = liftAtomicTransducer name 1 $
\source sink-> let t = canPut sink
>>= flip when (getSuccess source (\x-> put sink (f x) >> t))
in t >> return []
-- | Function 'liftStatelessTransducer' takes a function that maps one input value into a list of output values, and
-- lifts it into a 'Transducer'.
liftStatelessTransducer :: (Monad m, Typeable x, Typeable y) => String -> (x -> [y]) -> Transducer m x y
liftStatelessTransducer name f = liftAtomicTransducer name 1 $
\source sink-> let t = canPut sink
>>= flip when (getSuccess source (\x-> putList (f x) sink >> t))
in t >> return []
-- | Function 'liftFoldTransducer' creates a stateful transducer that produces only one output value after consuming the
-- entire input. Similar to 'Data.List.foldl'
liftFoldTransducer :: (Monad m, Typeable x, Typeable y) => String -> (s -> x -> s) -> s -> (s -> y) -> Transducer m x y
liftFoldTransducer name f s0 w = liftAtomicTransducer name 1 $
\source sink-> let t s = canPut sink
>>= flip when (get source
>>= maybe
(put sink (w s) >> return ())
(t . f s))
in t s0 >> return []
-- | Function 'liftStatefulTransducer' constructs a 'Transducer' from a state-transition function and the initial
-- state. The transition function may produce arbitrary output at any transition step.
liftStatefulTransducer :: (Monad m, Typeable x, Typeable y) => String -> (state -> x -> (state, [y])) -> state -> Transducer m x y
liftStatefulTransducer name f s0 = liftAtomicTransducer name 1 $
\source sink-> let t s = canPut sink
>>= flip when (getSuccess source
(\x-> let (s', ys) = f s x
in putList ys sink >> t s'))
in t s0 >> return []
-- | Function 'liftStatelessSplitter' takes a function that assigns a Boolean value to each input item and lifts it into
-- a 'Splitter'.
liftStatelessSplitter :: (ParallelizableMonad m, Typeable x) => String -> (x -> Bool) -> Splitter m x
liftStatelessSplitter name f = liftAtomicSimpleSplitter name 1 $
\source true false-> let s = get source
>>= maybe
(return [])
(\x-> (if f x
then put true x
else put false x)
>>= cond s (return [x]))
in s
-- | Function 'liftSimpleSplitter' lifts a simple, non-sectioning splitter function into a full 'Splitter'.
liftSimpleSplitter :: forall m x. (ParallelizableMonad m, Typeable x) =>
String -> Int
-> (Int -> (ComponentConfiguration, forall c. Source c x -> Sink c x -> Sink c x -> Pipe c m [x]))
-> Splitter m x
liftSimpleSplitter name maxThreads usingThreads
= case usingThreads 1
of (configuration, split) -> Splitter name maxThreads configuration usingThreads' split (splitSections split)
where usingThreads' :: Int -> (ComponentConfiguration,
forall c. Source c x -> Sink c x -> Sink c x -> Pipe c m [x],
forall c. Source c x -> Sink c (Maybe x) -> Sink c (Maybe x) -> Pipe c m [x])
usingThreads' threads = case usingThreads threads
of (configuration, splitValues) -> (configuration, splitValues, splitSections splitValues)
splitSections split source true false
= liftM (fst . fst) $
pipeD "liftSimpleSplitter true"
(\true'-> pipeD "liftSimpleSplitter false"
(\false'-> split source true' false')
(decorate false))
(decorate true)
decorate sink source = transduce (lift121Transducer "Just" Just) source sink
-- | Function 'liftSectionSplitter' lifts a sectioning splitter function into a full 'Splitter'
liftSectionSplitter :: forall m x. (ParallelizableMonad m, Typeable x) =>
String -> Int -> (Int -> (ComponentConfiguration,
forall c. Source c x -> Sink c (Maybe x) -> Sink c (Maybe x) -> Pipe c m [x]))
-> Splitter m x
liftSectionSplitter name maxThreads usingThreads
= case usingThreads 1
of (configuration, splitSections) -> Splitter name 1 configuration usingThreads' (splitValues splitSections) splitSections
where usingThreads' :: Int -> (ComponentConfiguration,
forall c. Source c x -> Sink c x -> Sink c x -> Pipe c m [x],
forall c. Source c x -> Sink c (Maybe x) -> Sink c (Maybe x) -> Pipe c m [x])
usingThreads' threads = case usingThreads threads
of (configuration, splitSections) -> (configuration, splitValues splitSections, splitSections)
splitValues splitSections source true false
= liftM (fst . fst) $
pipeD "liftSectionSplitter true"
(\true'-> pipeD "liftSectionSplitter false" (\false'-> splitSections source true' false') (strip false))
(strip true)
strip sink source = canPut sink
>>= flip when (getSuccess source (\x-> maybe (return False) (put sink) x >> strip sink source))
-- | Function 'liftAtomicSimpleSplitter' lifts a single-threaded 'split' function into a 'Splitter' component.
liftAtomicSimpleSplitter :: forall m x. (ParallelizableMonad m, Typeable x) =>
String -> Int -> (forall c. Source c x -> Sink c x -> Sink c x -> Pipe c m [x]) -> Splitter m x
liftAtomicSimpleSplitter name cost split = liftSimpleSplitter name 1 (\_threads-> (ComponentConfiguration [] 1 cost, split))
-- | Function 'liftAtomicSectionSplitter' lifts a single-threaded 'splitSections' function into a full 'Splitter'
-- component.
liftAtomicSectionSplitter :: forall m x. (ParallelizableMonad m, Typeable x) =>
String -> Int -> (forall c. Source c x -> Sink c (Maybe x) -> Sink c (Maybe x) -> Pipe c m [x])
-> Splitter m x
liftAtomicSectionSplitter name cost splitSections = liftSectionSplitter name 1 $
\_threads-> (ComponentConfiguration [] 1 cost, splitSections)
where configuration = ComponentConfiguration [] 1 1
usingThreads :: Int -> (ComponentConfiguration,
forall c. Source c x -> Sink c x -> Sink c x -> Pipe c m [x],
forall c. Source c x -> Sink c (Maybe x) -> Sink c (Maybe x) -> Pipe c m [x])
usingThreads threads = (configuration, splitValues, splitSections)
splitValues source true false
= liftM (fst . fst) $
pipeD "liftSectionSplitter true"
(\true'-> pipeD "liftSectionSplitter false" (\false'-> splitSections source true' false') (strip false))
(strip true)
-- strip sink source = transduce (liftStatelessTransducer (maybe [] (:[]))) source sink
strip sink source = canPut sink
>>= flip when (getSuccess source (\x-> maybe (return False) (put sink) x >> strip sink source))
-- | Function 'optimalTwoParallelConfigurations' configures two components, both of them with the full thread count, and
-- returns the components and a 'ComponentConfiguration' that can be used to build a new component from them.
optimalTwoSequentialConfigurations :: (Component c1, Component c2) => Int -> c1 -> c2 -> (ComponentConfiguration, c1, c2)
optimalTwoSequentialConfigurations threads c1 c2 = (configuration, c1', c2')
where configuration = ComponentConfiguration
[AnyComponent c1', AnyComponent c2']
(usedThreads c1' `max` usedThreads c2')
(cost c1' + cost c2')
c1' = usingThreads threads c1
c2' = usingThreads threads c2
-- | Function 'optimalTwoParallelConfigurations' configures two components assuming they can be run in parallel,
-- splitting the given thread count between them, and returns the configured components, a 'ComponentConfiguration' that
-- can be used to build a new component from them, and a flag that indicates if they should be run in parallel or
-- sequentially for optimal resource usage.
optimalTwoParallelConfigurations :: (Component c1, Component c2) => Int -> c1 -> c2 -> (ComponentConfiguration, c1, c2, Bool)
optimalTwoParallelConfigurations threads c1 c2 = (configuration, c1', c2', parallelize)
where parallelize = threads > 1 && parallelCost + 1 < sequentialCost
configuration = ComponentConfiguration
[AnyComponent c1', AnyComponent c2']
(if parallelize then usedThreads c1' + usedThreads c2' else usedThreads c1' `max` usedThreads c2')
(if parallelize then parallelCost + 1 else sequentialCost)
(c1', c2') = if parallelize then (c1p, c2p) else (c1s, c2s)
(c1p, c2p, parallelCost) = minimumBy
(\(_, _, cost1) (_, _, cost2)-> compare cost1 cost2)
[let c2threads = threads - c1threads `min` maxUsableThreads c2
c1i = usingThreads c1threads c1
c2i = usingThreads c2threads c2
in (c1i, c2i, cost c1i `max` cost c2i)
| c1threads <- [1 .. threads - 1 `min` maxUsableThreads c1]]
c1s = usingThreads threads c1
c2s = usingThreads threads c2
sequentialCost = cost c1s + cost c2s
-- | Function 'optimalThreeParallelConfigurations' configures three components assuming they can be run in parallel,
-- splitting the given thread count between them, and returns the components, a 'ComponentConfiguration' that can be
-- used to build a new component from them, and a flag per component that indicates if it should be run in parallel or
-- sequentially for optimal resource usage.
optimalThreeParallelConfigurations :: (Component c1, Component c2, Component c3) =>
Int -> c1 -> c2 -> c3 -> (ComponentConfiguration, (c1, Bool), (c2, Bool), (c3, Bool))
optimalThreeParallelConfigurations threadCount c1 c2 c3 = undefined