chp-1.3.2: Control/Concurrent/CHP/Arrow.hs
-- Communicating Haskell Processes.
-- Copyright (c) 2008, University of Kent.
-- All rights reserved.
--
-- Redistribution and use in source and binary forms, with or without
-- modification, are permitted provided that the following conditions are
-- met:
--
-- * Redistributions of source code must retain the above copyright
-- notice, this list of conditions and the following disclaimer.
-- * Redistributions in binary form must reproduce the above copyright
-- notice, this list of conditions and the following disclaimer in the
-- documentation and/or other materials provided with the distribution.
-- * Neither the name of the University of Kent nor the names of its
-- contributors may be used to endorse or promote products derived from
-- this software without specific prior written permission.
--
-- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
-- IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
-- THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
-- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
-- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
-- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
-- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
-- PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
-- LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
-- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
-- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-- | Provides an instance of Arrow for process pipelines. As described in
-- the original paper on arrows, they can be used to represent stream processing,
-- so CHP seemed like a possible fit for an arrow.
--
-- Whether this is /actually/ an instance of Arrow depends on technicalities.
-- This can be demonstrated with the arrow law @arr id >>> f = f = f >>> arr
-- id@. Whether CHP satisfies this arrow law depends on the definition of
-- equality.
--
-- * If equality means that given the same input value, both arrows produce the
-- same corresponding output value, this is an arrow.
--
-- * If equality means you give the arrows the same single input and wait for the single output,
-- and the output is the same, this is an arrow.
--
-- * If equality means that you can feed the arrows lots of inputs (one after
-- the other) and the behaviour should be the same with regards to communication,
-- this is not an arrow.
--
-- The problem lies in the buffering inherent in arrows. Imagine if @f@ is
-- a single function. @f@ is effectively a buffer of one. You can feed it
-- a single value, but no more than that until you read its output. However,
-- if you have @arr id >>> f@, that can accept two inputs (one held by the
-- @arr id@ process and one held by @f@) before you must accept the output.
--
-- I am fairly confident that the arrow laws are satisfied for the
-- definition of equality that given the same single input, they will
-- produce the same single output. If you don't worry too much about the
-- behavioural difference, and just take arrows as another way to wire
-- together a certain class of process network, you should do fine.
--
-- Added in version 1.0.2.
module Control.Concurrent.CHP.Arrow (ProcessPipeline, runPipeline, arrowProcess, arrStrict) where
-- I have got this module to work on GHC 6.8 and 6.10 by following the CPP-variant
-- instructions on this page: http://haskell.org/haskellwiki/Upgrading_packages
#if __GLASGOW_HASKELL__ >= 609
import Control.Category
import Prelude hiding ((.), id)
#endif
import Control.Arrow
#if __GLASGOW_HASKELL__ < 610
hiding (pure)
#endif
import Control.Monad
import Control.Parallel.Strategies
import Control.Concurrent.CHP
import qualified Control.Concurrent.CHP.Common as CHP
import Control.Concurrent.CHP.Utils
-- | The type that is an instance of 'Arrow' for process pipelines. See 'runPipeline'.
data ProcessPipeline a b = ProcessPipeline
{ runPipeline :: Chanin a -> Chanout b -> CHP ()
-- ^ Given a 'ProcessPipeline' (formed using its 'Arrow' instance) and
-- the channels to plug into the ends of the pipeline, returns the process
-- representing the pipeline.
--
-- The pipeline will run forever (until poisoned) and you must run it in
-- parallel to whatever is feeding it the inputs and reading off the outputs.
-- Imagine that you want a process pipeline that takes in a pair of numbers,
-- doubles the first and adds one to the second. You could encode this
-- in an arrow using:
--
-- > runPipeline (arr (*2) *** arr (+1))
--
-- Arrows are more useful where you already have processes written that
-- process data and you want to easily wire them together. The arrow notation
-- is probably easier for doing that than declaring all the channels yourself
-- and composing everything in parallel.
}
-- | Adds a wrapper that forms this process into the right data type to be
-- part of an arrow.
--
-- Any process you apply this to should produce exactly one output per
-- input, or else you will find odd behaviour resulting (including deadlock).
-- So for example, /don't/ use @arrowProcess ('Control.Concurrent.CHP.Common.filter'
-- ...)@ or @arrowProcess 'Control.Concurrent.CHP.Common.stream'@
--
-- Added in version 1.1.0
arrowProcess :: (Chanin a -> Chanout b -> CHP ()) -> ProcessPipeline a b
arrowProcess = ProcessPipeline
-- | Like the arr function of the ProcessPipeline arrow instance, but fully evaluates
-- the result before sending it. If you are building process pipelines with arrows to
-- try and get some parallel speed-up, you should try this function instead of
-- arr itself.
--
-- Added in version 1.3.2
arrStrict :: NFData b => (a -> b) -> ProcessPipeline a b
arrStrict = ProcessPipeline . CHP.map'
#if __GLASGOW_HASKELL__ >= 609
instance Category ProcessPipeline where
(ProcessPipeline q) . (ProcessPipeline p) = ProcessPipeline (p |->| q)
id = ProcessPipeline CHP.id
#endif
instance Arrow ProcessPipeline where
#if __GLASGOW_HASKELL__ < 609
(ProcessPipeline p) >>> (ProcessPipeline q) = ProcessPipeline (p |->| q)
#endif
arr = ProcessPipeline . CHP.map
first (ProcessPipeline p) = ProcessPipeline $ \in_ out -> do
c <- newChannel
c' <- newChannel
d <- newChannel
runParallel_
[ CHP.split in_ (writer c) (writer d)
, p (reader c) (writer c')
, CHP.join (,) (reader c') (reader d) out
]
second (ProcessPipeline p) = ProcessPipeline $ \in_ out -> do
c <- newChannel
c' <- newChannel
d <- newChannel
runParallel_
[ CHP.split in_ (writer d) (writer c)
, p (reader c) (writer c')
, CHP.join (,) (reader d) (reader c') out
]
(ProcessPipeline p) *** (ProcessPipeline q) = ProcessPipeline $ \in_ out -> do
c <- newChannel
c' <- newChannel
d <- newChannel
d' <- newChannel
runParallel_
[ CHP.split in_ (writer c) (writer d)
, p (reader c) (writer c')
, q (reader d) (writer d')
, CHP.join (,) (reader c') (reader d') out
]
(ProcessPipeline p) &&& (ProcessPipeline q) = ProcessPipeline $ \in_ out -> do
c <- newChannel
c' <- newChannel
d <- newChannel
d' <- newChannel
runParallel_
[ CHP.parDelta in_ [writer c, writer d]
, p (reader c) (writer c')
, q (reader d) (writer d')
, CHP.join (,) (reader c') (reader d') out
]
instance ArrowChoice ProcessPipeline where
left (ProcessPipeline p) = ProcessPipeline $ \in_ out -> do
c <- oneToOneChannel
d <- oneToOneChannel
(forever $ do x <- readChannel in_
case x of
Left l -> do writeChannel (writer c) l
l' <- readChannel (reader d)
writeChannel out (Left l')
Right r -> writeChannel out (Right r)
) <||> p (reader c) (writer d)
return ()
right (ProcessPipeline p) = ProcessPipeline $ \in_ out -> do
c <- oneToOneChannel
d <- oneToOneChannel
(forever $ do x <- readChannel in_
case x of
Right r -> do writeChannel (writer c) r
r' <- readChannel (reader d)
writeChannel out (Right r')
Left l -> writeChannel out (Left l)
) <||> p (reader c) (writer d)
return ()
(ProcessPipeline p) ||| (ProcessPipeline q)
= ProcessPipeline $ \in_ out -> do
c <- oneToOneChannel
c' <- oneToOneChannel
d <- oneToOneChannel
d' <- oneToOneChannel
runParallel_
[ forever $ do x <- readChannel in_
x' <- case x of
Left l -> do writeChannel (writer c) l
readChannel (reader c')
Right r -> do writeChannel (writer d) r
readChannel (reader d')
writeChannel out x'
, p (reader c) (writer c')
, q (reader d) (writer d')
]
(ProcessPipeline p) +++ (ProcessPipeline q)
= ProcessPipeline $ \in_ out -> do
c <- oneToOneChannel
c' <- oneToOneChannel
d <- oneToOneChannel
d' <- oneToOneChannel
runParallel_
[ forever $ do x <- readChannel in_
x' <- case x of
Left l -> do writeChannel (writer c) l
l' <- readChannel (reader c')
return (Left l')
Right r -> do writeChannel (writer d) r
r' <- readChannel (reader d')
return (Right r')
writeChannel out x'
, p (reader c) (writer c')
, q (reader d) (writer d')
]