module Simulation.Aivika.Processor
(
Processor(..),
emptyProcessor,
arrProcessor,
accumProcessor,
withinProcessor,
processorUsingId,
prefetchProcessor,
delayProcessor,
bufferProcessor,
bufferProcessorLoop,
queueProcessor,
queueProcessorLoopMerging,
queueProcessorLoopSeq,
queueProcessorLoopParallel,
processorSeq,
processorParallel,
processorQueuedParallel,
processorPrioritisingOutputParallel,
processorPrioritisingInputParallel,
processorPrioritisingInputOutputParallel,
arrivalProcessor,
joinProcessor,
failoverProcessor,
channelProcessor,
processorChannel,
queuedChannelProcessor,
queuedProcessorChannel,
traceProcessor) where
import qualified Control.Category as C
import Control.Arrow
import Data.Monoid
import Simulation.Aivika.Simulation
import Simulation.Aivika.Dynamics
import Simulation.Aivika.Event
import Simulation.Aivika.Composite
import Simulation.Aivika.Cont
import Simulation.Aivika.Process
import Simulation.Aivika.Stream
import Simulation.Aivika.QueueStrategy
import Simulation.Aivika.Signal
import Simulation.Aivika.Channel
import Simulation.Aivika.Internal.Arrival
newtype Processor a b =
Processor { forall a b. Processor a b -> Stream a -> Stream b
runProcessor :: Stream a -> Stream b
}
instance C.Category Processor where
id :: forall a. Processor a a
id = forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a. a -> a
id
Processor Stream b -> Stream c
x . :: forall b c a. Processor b c -> Processor a b -> Processor a c
. Processor Stream a -> Stream b
y = forall a b. (Stream a -> Stream b) -> Processor a b
Processor (Stream b -> Stream c
x forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stream a -> Stream b
y)
instance Arrow Processor where
arr :: forall b c. (b -> c) -> Processor b c
arr = forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall a b. (a -> b) -> Stream a -> Stream b
mapStream
first :: forall b c d. Processor b c -> Processor (b, d) (c, d)
first (Processor Stream b -> Stream c
f) =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream (b, d)
xys ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do (Stream b
xs, Stream d
ys) <- forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall a b. Stream (a, b) -> Simulation (Stream a, Stream b)
unzipStream Stream (b, d)
xys
forall a. Stream a -> Process (a, Stream a)
runStream forall a b. (a -> b) -> a -> b
$ forall a b. Stream a -> Stream b -> Stream (a, b)
zipStreamParallel (Stream b -> Stream c
f Stream b
xs) Stream d
ys
second :: forall b c d. Processor b c -> Processor (d, b) (d, c)
second (Processor Stream b -> Stream c
f) =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream (d, b)
xys ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do (Stream d
xs, Stream b
ys) <- forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall a b. Stream (a, b) -> Simulation (Stream a, Stream b)
unzipStream Stream (d, b)
xys
forall a. Stream a -> Process (a, Stream a)
runStream forall a b. (a -> b) -> a -> b
$ forall a b. Stream a -> Stream b -> Stream (a, b)
zipStreamParallel Stream d
xs (Stream b -> Stream c
f Stream b
ys)
Processor Stream b -> Stream c
f *** :: forall b c b' c'.
Processor b c -> Processor b' c' -> Processor (b, b') (c, c')
*** Processor Stream b' -> Stream c'
g =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream (b, b')
xys ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do (Stream b
xs, Stream b'
ys) <- forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall a b. Stream (a, b) -> Simulation (Stream a, Stream b)
unzipStream Stream (b, b')
xys
forall a. Stream a -> Process (a, Stream a)
runStream forall a b. (a -> b) -> a -> b
$ forall a b. Stream a -> Stream b -> Stream (a, b)
zipStreamParallel (Stream b -> Stream c
f Stream b
xs) (Stream b' -> Stream c'
g Stream b'
ys)
Processor Stream b -> Stream c
f &&& :: forall b c c'.
Processor b c -> Processor b c' -> Processor b (c, c')
&&& Processor Stream b -> Stream c'
g =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream b
xs -> forall a b. Stream a -> Stream b -> Stream (a, b)
zipStreamParallel (Stream b -> Stream c
f Stream b
xs) (Stream b -> Stream c'
g Stream b
xs)
instance ArrowChoice Processor where
left :: forall b c d. Processor b c -> Processor (Either b d) (Either c d)
left (Processor Stream b -> Stream c
f) =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream (Either b d)
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do Stream (Either b d)
ys <- forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall a. Stream a -> Simulation (Stream a)
memoStream Stream (Either b d)
xs
forall a. Stream a -> Process (a, Stream a)
runStream forall a b. (a -> b) -> a -> b
$ forall a b c.
Stream (Either a b) -> Stream c -> Stream (Either c b)
replaceLeftStream Stream (Either b d)
ys (Stream b -> Stream c
f forall a b. (a -> b) -> a -> b
$ forall a b. Stream (Either a b) -> Stream a
leftStream Stream (Either b d)
ys)
right :: forall b c d. Processor b c -> Processor (Either d b) (Either d c)
right (Processor Stream b -> Stream c
f) =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream (Either d b)
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do Stream (Either d b)
ys <- forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall a. Stream a -> Simulation (Stream a)
memoStream Stream (Either d b)
xs
forall a. Stream a -> Process (a, Stream a)
runStream forall a b. (a -> b) -> a -> b
$ forall a b c.
Stream (Either a b) -> Stream c -> Stream (Either a c)
replaceRightStream Stream (Either d b)
ys (Stream b -> Stream c
f forall a b. (a -> b) -> a -> b
$ forall a b. Stream (Either a b) -> Stream b
rightStream Stream (Either d b)
ys)
instance ArrowZero Processor where
zeroArrow :: forall b c. Processor b c
zeroArrow = forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ forall a b. a -> b -> a
const forall a. Stream a
emptyStream
instance ArrowPlus Processor where
(Processor Stream b -> Stream c
f) <+> :: forall b c. Processor b c -> Processor b c -> Processor b c
<+> (Processor Stream b -> Stream c
g) =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream b
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do [Stream b
xs1, Stream b
xs2] <- forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall a. Int -> Stream a -> Simulation [Stream a]
splitStream Int
2 Stream b
xs
forall a. Stream a -> Process (a, Stream a)
runStream forall a b. (a -> b) -> a -> b
$ forall a. Stream a -> Stream a -> Stream a
mergeStreams (Stream b -> Stream c
f Stream b
xs1) (Stream b -> Stream c
g Stream b
xs2)
emptyProcessor :: Processor a b
emptyProcessor :: forall b c. Processor b c
emptyProcessor = forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ forall a b. a -> b -> a
const forall a. Stream a
emptyStream
arrProcessor :: (a -> Process b) -> Processor a b
arrProcessor :: forall a b. (a -> Process b) -> Processor a b
arrProcessor = forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall a b. (a -> Process b) -> Stream a -> Stream b
mapStreamM
accumProcessor :: (acc -> a -> Process (acc, b)) -> acc -> Processor a b
accumProcessor :: forall acc a b.
(acc -> a -> Process (acc, b)) -> acc -> Processor a b
accumProcessor acc -> a -> Process (acc, b)
f acc
acc = forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ forall acc a b.
(acc -> a -> Process (acc, b)) -> acc -> Stream a -> Stream b
accumStream acc -> a -> Process (acc, b)
f acc
acc
withinProcessor :: Process () -> Processor a a
withinProcessor :: forall a. Process () -> Processor a a
withinProcessor Process ()
m =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$
forall a b. (a -> Process b) -> Stream a -> Stream b
mapStreamM forall a b. (a -> b) -> a -> b
$ \a
a ->
do { Process ()
m; forall (m :: * -> *) a. Monad m => a -> m a
return a
a }
processorUsingId :: ProcessId -> Processor a b -> Processor a b
processorUsingId :: forall a b. ProcessId -> Processor a b -> Processor a b
processorUsingId ProcessId
pid (Processor Stream a -> Stream b
f) =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ forall a. Process (a, Stream a) -> Stream a
Cons forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall a. ProcessId -> Process a -> Process a
processUsingId ProcessId
pid forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall a. Stream a -> Process (a, Stream a)
runStream forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stream a -> Stream b
f
processorQueuedParallel :: (EnqueueStrategy si,
EnqueueStrategy so)
=> si
-> so
-> [Processor a b]
-> Processor a b
processorQueuedParallel :: forall si so a b.
(EnqueueStrategy si, EnqueueStrategy so) =>
si -> so -> [Processor a b] -> Processor a b
processorQueuedParallel si
si so
so [Processor a b]
ps =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream a
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do let n :: Int
n = forall (t :: * -> *) a. Foldable t => t a -> Int
length [Processor a b]
ps
[Stream a]
input <- forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall s a.
EnqueueStrategy s =>
s -> Int -> Stream a -> Simulation [Stream a]
splitStreamQueueing si
si Int
n Stream a
xs
let results :: [Stream b]
results = forall a b c. (a -> b -> c) -> b -> a -> c
flip forall a b. (a -> b) -> [a] -> [b]
map (forall a b. [a] -> [b] -> [(a, b)]
zip [Stream a]
input [Processor a b]
ps) forall a b. (a -> b) -> a -> b
$ \(Stream a
input, Processor a b
p) ->
forall a b. Processor a b -> Stream a -> Stream b
runProcessor Processor a b
p Stream a
input
output :: Stream b
output = forall s a. EnqueueStrategy s => s -> [Stream a] -> Stream a
concatQueuedStreams so
so [Stream b]
results
forall a. Stream a -> Process (a, Stream a)
runStream Stream b
output
processorPrioritisingOutputParallel :: (EnqueueStrategy si,
PriorityQueueStrategy so po)
=> si
-> so
-> [Processor a (po, b)]
-> Processor a b
processorPrioritisingOutputParallel :: forall si so po a b.
(EnqueueStrategy si, PriorityQueueStrategy so po) =>
si -> so -> [Processor a (po, b)] -> Processor a b
processorPrioritisingOutputParallel si
si so
so [Processor a (po, b)]
ps =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream a
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do let n :: Int
n = forall (t :: * -> *) a. Foldable t => t a -> Int
length [Processor a (po, b)]
ps
[Stream a]
input <- forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall s a.
EnqueueStrategy s =>
s -> Int -> Stream a -> Simulation [Stream a]
splitStreamQueueing si
si Int
n Stream a
xs
let results :: [Stream (po, b)]
results = forall a b c. (a -> b -> c) -> b -> a -> c
flip forall a b. (a -> b) -> [a] -> [b]
map (forall a b. [a] -> [b] -> [(a, b)]
zip [Stream a]
input [Processor a (po, b)]
ps) forall a b. (a -> b) -> a -> b
$ \(Stream a
input, Processor a (po, b)
p) ->
forall a b. Processor a b -> Stream a -> Stream b
runProcessor Processor a (po, b)
p Stream a
input
output :: Stream b
output = forall s p a.
PriorityQueueStrategy s p =>
s -> [Stream (p, a)] -> Stream a
concatPriorityStreams so
so [Stream (po, b)]
results
forall a. Stream a -> Process (a, Stream a)
runStream Stream b
output
processorPrioritisingInputParallel :: (PriorityQueueStrategy si pi,
EnqueueStrategy so)
=> si
-> so
-> [(Stream pi, Processor a b)]
-> Processor a b
processorPrioritisingInputParallel :: forall si pi so a b.
(PriorityQueueStrategy si pi, EnqueueStrategy so) =>
si -> so -> [(Stream pi, Processor a b)] -> Processor a b
processorPrioritisingInputParallel si
si so
so [(Stream pi, Processor a b)]
ps =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream a
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do [Stream a]
input <- forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall s p a.
PriorityQueueStrategy s p =>
s -> [Stream p] -> Stream a -> Simulation [Stream a]
splitStreamPrioritising si
si (forall a b. (a -> b) -> [a] -> [b]
map forall a b. (a, b) -> a
fst [(Stream pi, Processor a b)]
ps) Stream a
xs
let results :: [Stream b]
results = forall a b c. (a -> b -> c) -> b -> a -> c
flip forall a b. (a -> b) -> [a] -> [b]
map (forall a b. [a] -> [b] -> [(a, b)]
zip [Stream a]
input [(Stream pi, Processor a b)]
ps) forall a b. (a -> b) -> a -> b
$ \(Stream a
input, (Stream pi
_, Processor a b
p)) ->
forall a b. Processor a b -> Stream a -> Stream b
runProcessor Processor a b
p Stream a
input
output :: Stream b
output = forall s a. EnqueueStrategy s => s -> [Stream a] -> Stream a
concatQueuedStreams so
so [Stream b]
results
forall a. Stream a -> Process (a, Stream a)
runStream Stream b
output
processorPrioritisingInputOutputParallel :: (PriorityQueueStrategy si pi,
PriorityQueueStrategy so po)
=> si
-> so
-> [(Stream pi, Processor a (po, b))]
-> Processor a b
processorPrioritisingInputOutputParallel :: forall si pi so po a b.
(PriorityQueueStrategy si pi, PriorityQueueStrategy so po) =>
si -> so -> [(Stream pi, Processor a (po, b))] -> Processor a b
processorPrioritisingInputOutputParallel si
si so
so [(Stream pi, Processor a (po, b))]
ps =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream a
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do [Stream a]
input <- forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$ forall s p a.
PriorityQueueStrategy s p =>
s -> [Stream p] -> Stream a -> Simulation [Stream a]
splitStreamPrioritising si
si (forall a b. (a -> b) -> [a] -> [b]
map forall a b. (a, b) -> a
fst [(Stream pi, Processor a (po, b))]
ps) Stream a
xs
let results :: [Stream (po, b)]
results = forall a b c. (a -> b -> c) -> b -> a -> c
flip forall a b. (a -> b) -> [a] -> [b]
map (forall a b. [a] -> [b] -> [(a, b)]
zip [Stream a]
input [(Stream pi, Processor a (po, b))]
ps) forall a b. (a -> b) -> a -> b
$ \(Stream a
input, (Stream pi
_, Processor a (po, b)
p)) ->
forall a b. Processor a b -> Stream a -> Stream b
runProcessor Processor a (po, b)
p Stream a
input
output :: Stream b
output = forall s p a.
PriorityQueueStrategy s p =>
s -> [Stream (p, a)] -> Stream a
concatPriorityStreams so
so [Stream (po, b)]
results
forall a. Stream a -> Process (a, Stream a)
runStream Stream b
output
processorParallel :: [Processor a b] -> Processor a b
processorParallel :: forall a b. [Processor a b] -> Processor a b
processorParallel = forall si so a b.
(EnqueueStrategy si, EnqueueStrategy so) =>
si -> so -> [Processor a b] -> Processor a b
processorQueuedParallel FCFS
FCFS FCFS
FCFS
processorSeq :: [Processor a a] -> Processor a a
processorSeq :: forall a. [Processor a a] -> Processor a a
processorSeq [] = forall b c. Processor b c
emptyProcessor
processorSeq [Processor a a
p] = Processor a a
p
processorSeq (Processor a a
p : [Processor a a]
ps) = Processor a a
p forall {k} (cat :: k -> k -> *) (a :: k) (b :: k) (c :: k).
Category cat =>
cat a b -> cat b c -> cat a c
>>> forall a. Processor a a
prefetchProcessor forall {k} (cat :: k -> k -> *) (a :: k) (b :: k) (c :: k).
Category cat =>
cat a b -> cat b c -> cat a c
>>> forall a. [Processor a a] -> Processor a a
processorSeq [Processor a a]
ps
bufferProcessor :: (Stream a -> Process ())
-> Stream b
-> Processor a b
bufferProcessor :: forall a b. (Stream a -> Process ()) -> Stream b -> Processor a b
bufferProcessor Stream a -> Process ()
consume Stream b
output =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream a
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do Process () -> Process ()
spawnProcess (Stream a -> Process ()
consume Stream a
xs)
forall a. Stream a -> Process (a, Stream a)
runStream Stream b
output
bufferProcessorLoop :: (Stream a -> Stream c -> Process ())
-> Stream d
-> Processor d (Either e b)
-> Processor e c
-> Processor a b
bufferProcessorLoop :: forall a c d e b.
(Stream a -> Stream c -> Process ())
-> Stream d
-> Processor d (Either e b)
-> Processor e c
-> Processor a b
bufferProcessorLoop Stream a -> Stream c -> Process ()
consume Stream d
preoutput Processor d (Either e b)
cond Processor e c
body =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream a
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do (Stream e
reverted, Stream b
output) <-
forall (m :: * -> *) a. SimulationLift m => Simulation a -> m a
liftSimulation forall a b. (a -> b) -> a -> b
$
forall a b. Stream (Either a b) -> Simulation (Stream a, Stream b)
partitionEitherStream forall a b. (a -> b) -> a -> b
$
forall a b. Processor a b -> Stream a -> Stream b
runProcessor Processor d (Either e b)
cond Stream d
preoutput
Process () -> Process ()
spawnProcess
(Stream a -> Stream c -> Process ()
consume Stream a
xs forall a b. (a -> b) -> a -> b
$ forall a b. Processor a b -> Stream a -> Stream b
runProcessor Processor e c
body Stream e
reverted)
forall a. Stream a -> Process (a, Stream a)
runStream Stream b
output
queueProcessor :: (a -> Process ())
-> Process b
-> Processor a b
queueProcessor :: forall a b. (a -> Process ()) -> Process b -> Processor a b
queueProcessor a -> Process ()
enqueue Process b
dequeue =
forall a b. (Stream a -> Process ()) -> Stream b -> Processor a b
bufferProcessor
(forall a. (a -> Process ()) -> Stream a -> Process ()
consumeStream a -> Process ()
enqueue)
(forall a. Process a -> Stream a
repeatProcess Process b
dequeue)
queueProcessorLoopMerging :: (Stream a -> Stream d -> Stream e)
-> (e -> Process ())
-> Process c
-> Processor c (Either f b)
-> Processor f d
-> Processor a b
queueProcessorLoopMerging :: forall a d e c f b.
(Stream a -> Stream d -> Stream e)
-> (e -> Process ())
-> Process c
-> Processor c (Either f b)
-> Processor f d
-> Processor a b
queueProcessorLoopMerging Stream a -> Stream d -> Stream e
merge e -> Process ()
enqueue Process c
dequeue =
forall a c d e b.
(Stream a -> Stream c -> Process ())
-> Stream d
-> Processor d (Either e b)
-> Processor e c
-> Processor a b
bufferProcessorLoop
(\Stream a
bs Stream d
cs ->
forall a. (a -> Process ()) -> Stream a -> Process ()
consumeStream e -> Process ()
enqueue forall a b. (a -> b) -> a -> b
$
Stream a -> Stream d -> Stream e
merge Stream a
bs Stream d
cs)
(forall a. Process a -> Stream a
repeatProcess Process c
dequeue)
queueProcessorLoopSeq :: (a -> Process ())
-> Process c
-> Processor c (Either e b)
-> Processor e a
-> Processor a b
queueProcessorLoopSeq :: forall a c e b.
(a -> Process ())
-> Process c
-> Processor c (Either e b)
-> Processor e a
-> Processor a b
queueProcessorLoopSeq =
forall a d e c f b.
(Stream a -> Stream d -> Stream e)
-> (e -> Process ())
-> Process c
-> Processor c (Either f b)
-> Processor f d
-> Processor a b
queueProcessorLoopMerging forall a. Stream a -> Stream a -> Stream a
mergeStreams
queueProcessorLoopParallel :: (a -> Process ())
-> Process c
-> Processor c (Either e b)
-> Processor e a
-> Processor a b
queueProcessorLoopParallel :: forall a c e b.
(a -> Process ())
-> Process c
-> Processor c (Either e b)
-> Processor e a
-> Processor a b
queueProcessorLoopParallel a -> Process ()
enqueue Process c
dequeue =
forall a c d e b.
(Stream a -> Stream c -> Process ())
-> Stream d
-> Processor d (Either e b)
-> Processor e c
-> Processor a b
bufferProcessorLoop
(\Stream a
bs Stream a
cs ->
do Process () -> Process ()
spawnProcess forall a b. (a -> b) -> a -> b
$
forall a. (a -> Process ()) -> Stream a -> Process ()
consumeStream a -> Process ()
enqueue Stream a
bs
Process () -> Process ()
spawnProcess forall a b. (a -> b) -> a -> b
$
forall a. (a -> Process ()) -> Stream a -> Process ()
consumeStream a -> Process ()
enqueue Stream a
cs)
(forall a. Process a -> Stream a
repeatProcess Process c
dequeue)
prefetchProcessor :: Processor a a
prefetchProcessor :: forall a. Processor a a
prefetchProcessor = forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a. Stream a -> Stream a
prefetchStream
channelProcessor :: Channel a b -> Processor a b
channelProcessor :: forall a b. Channel a b -> Processor a b
channelProcessor Channel a b
f =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream a
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do let composite :: Composite (Stream b)
composite =
do Signal a
sa <- forall a. Stream a -> Composite (Signal a)
streamSignal Stream a
xs
Signal b
sb <- forall a b. Channel a b -> Signal a -> Composite (Signal b)
runChannel Channel a b
f Signal a
sa
forall a. Signal a -> Composite (Stream a)
signalStream Signal b
sb
(Stream b
ys, DisposableEvent
h) <- forall (m :: * -> *) a. EventLift m => Event a -> m a
liftEvent forall a b. (a -> b) -> a -> b
$
forall a.
Composite a -> DisposableEvent -> Event (a, DisposableEvent)
runComposite Composite (Stream b)
composite forall a. Monoid a => a
mempty
Event () -> Process ()
whenCancellingProcess forall a b. (a -> b) -> a -> b
$
DisposableEvent -> Event ()
disposeEvent DisposableEvent
h
forall a. Stream a -> Process (a, Stream a)
runStream Stream b
ys
processorChannel :: Processor a b -> Channel a b
processorChannel :: forall a b. Processor a b -> Channel a b
processorChannel (Processor Stream a -> Stream b
f) =
forall a b. (Signal a -> Composite (Signal b)) -> Channel a b
Channel forall a b. (a -> b) -> a -> b
$ \Signal a
sa ->
do Stream a
xs <- forall a. Signal a -> Composite (Stream a)
signalStream Signal a
sa
let ys :: Stream b
ys = Stream a -> Stream b
f Stream a
xs
forall a. Stream a -> Composite (Signal a)
streamSignal Stream b
ys
queuedChannelProcessor :: (b -> Event ())
-> Process b
-> Channel a b
-> Processor a b
queuedChannelProcessor :: forall b a.
(b -> Event ()) -> Process b -> Channel a b -> Processor a b
queuedChannelProcessor b -> Event ()
enqueue Process b
dequeue Channel a b
f =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream a
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do let composite :: Composite (Stream b)
composite =
do Signal a
sa <- forall a. Stream a -> Composite (Signal a)
streamSignal Stream a
xs
Signal b
sb <- forall a b. Channel a b -> Signal a -> Composite (Signal b)
runChannel Channel a b
f Signal a
sa
forall a.
(a -> Event ()) -> Process a -> Signal a -> Composite (Stream a)
queuedSignalStream b -> Event ()
enqueue Process b
dequeue Signal b
sb
(Stream b
ys, DisposableEvent
h) <- forall (m :: * -> *) a. EventLift m => Event a -> m a
liftEvent forall a b. (a -> b) -> a -> b
$
forall a.
Composite a -> DisposableEvent -> Event (a, DisposableEvent)
runComposite Composite (Stream b)
composite forall a. Monoid a => a
mempty
Event () -> Process ()
whenCancellingProcess forall a b. (a -> b) -> a -> b
$
DisposableEvent -> Event ()
disposeEvent DisposableEvent
h
forall a. Stream a -> Process (a, Stream a)
runStream Stream b
ys
queuedProcessorChannel :: (a -> Event ())
-> (Process a)
-> Processor a b
-> Channel a b
queuedProcessorChannel :: forall a b.
(a -> Event ()) -> Process a -> Processor a b -> Channel a b
queuedProcessorChannel a -> Event ()
enqueue Process a
dequeue (Processor Stream a -> Stream b
f) =
forall a b. (Signal a -> Composite (Signal b)) -> Channel a b
Channel forall a b. (a -> b) -> a -> b
$ \Signal a
sa ->
do Stream a
xs <- forall a.
(a -> Event ()) -> Process a -> Signal a -> Composite (Stream a)
queuedSignalStream a -> Event ()
enqueue Process a
dequeue Signal a
sa
let ys :: Stream b
ys = Stream a -> Stream b
f Stream a
xs
forall a. Stream a -> Composite (Signal a)
streamSignal Stream b
ys
arrivalProcessor :: Processor a (Arrival a)
arrivalProcessor :: forall a. Processor a (Arrival a)
arrivalProcessor = forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a. Stream a -> Stream (Arrival a)
arrivalStream
delayProcessor :: a -> Processor a a
delayProcessor :: forall a. a -> Processor a a
delayProcessor a
a0 = forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ forall a. a -> Stream a -> Stream a
delayStream a
a0
joinProcessor :: Process (Processor a b) -> Processor a b
joinProcessor :: forall a b. Process (Processor a b) -> Processor a b
joinProcessor Process (Processor a b)
m =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream a
xs ->
forall a. Process (a, Stream a) -> Stream a
Cons forall a b. (a -> b) -> a -> b
$
do Processor Stream a -> Stream b
f <- Process (Processor a b)
m
forall a. Stream a -> Process (a, Stream a)
runStream forall a b. (a -> b) -> a -> b
$ Stream a -> Stream b
f Stream a
xs
failoverProcessor :: [Processor a b] -> Processor a b
failoverProcessor :: forall a b. [Processor a b] -> Processor a b
failoverProcessor [Processor a b]
ps =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ \Stream a
xs -> forall a. [Stream a] -> Stream a
failoverStream [forall a b. Processor a b -> Stream a -> Stream b
runProcessor Processor a b
p Stream a
xs | Processor a b
p <- [Processor a b]
ps]
traceProcessor :: Maybe String
-> Maybe String
-> Processor a b
-> Processor a b
traceProcessor :: forall a b.
Maybe String -> Maybe String -> Processor a b -> Processor a b
traceProcessor Maybe String
request Maybe String
response (Processor Stream a -> Stream b
f) =
forall a b. (Stream a -> Stream b) -> Processor a b
Processor forall a b. (a -> b) -> a -> b
$ forall a. Maybe String -> Maybe String -> Stream a -> Stream a
traceStream Maybe String
request Maybe String
response forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stream a -> Stream b
f