diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,23 @@ +Version 5.0 +----- + +* Added the Composite monad. + +* Added the Channel computation. + +* Breaking change: modified signatures of functions signalStream and streamSignal. + +* Breaking change: the signalProcessor function is replaced with channelProcessor. + +* Breaking change: the processorSignaling function is replaced with processorChannel. + +* Added module Signal.Random. + +* Added functions arrivalTimerSignal and arrivalTimerChannel. + +* Added functions queuedSignalStream, queuedProcessorChannel and queuedChannelProcessor. + Version 4.6 ----- diff --git a/Simulation/Aivika.hs b/Simulation/Aivika.hs --- a/Simulation/Aivika.hs +++ b/Simulation/Aivika.hs @@ -16,7 +16,9 @@ module Simulation.Aivika.Activity.Random, module Simulation.Aivika.Agent, module Simulation.Aivika.Arrival, + module Simulation.Aivika.Channel, module Simulation.Aivika.Circuit, + module Simulation.Aivika.Composite, module Simulation.Aivika.Cont, module Simulation.Aivika.Dynamics, module Simulation.Aivika.Dynamics.Extra, @@ -45,6 +47,7 @@ module Simulation.Aivika.Server, module Simulation.Aivika.Server.Random, module Simulation.Aivika.Signal, + module Simulation.Aivika.Signal.Random, module Simulation.Aivika.Simulation, module Simulation.Aivika.Specs, module Simulation.Aivika.Statistics, @@ -61,7 +64,9 @@ import Simulation.Aivika.Activity.Random import Simulation.Aivika.Agent import Simulation.Aivika.Arrival +import Simulation.Aivika.Channel import Simulation.Aivika.Circuit +import Simulation.Aivika.Composite import Simulation.Aivika.Cont import Simulation.Aivika.Dynamics import Simulation.Aivika.Dynamics.Extra @@ -90,6 +95,7 @@ import Simulation.Aivika.Server import Simulation.Aivika.Server.Random import Simulation.Aivika.Signal +import Simulation.Aivika.Signal.Random import Simulation.Aivika.Simulation import Simulation.Aivika.Specs import Simulation.Aivika.Statistics diff --git a/Simulation/Aivika/Arrival.hs b/Simulation/Aivika/Arrival.hs --- a/Simulation/Aivika/Arrival.hs +++ b/Simulation/Aivika/Arrival.hs @@ -18,6 +18,8 @@ ArrivalTimer, newArrivalTimer, arrivalTimerProcessor, + arrivalTimerSignal, + arrivalTimerChannel, arrivalProcessingTime, arrivalProcessingTimeChanged, arrivalProcessingTimeChanged_) where @@ -28,11 +30,13 @@ import Simulation.Aivika.Simulation import Simulation.Aivika.Dynamics import Simulation.Aivika.Event +import Simulation.Aivika.Composite import Simulation.Aivika.Processor import Simulation.Aivika.Stream import Simulation.Aivika.Statistics import Simulation.Aivika.Ref import Simulation.Aivika.Signal +import Simulation.Aivika.Channel import Simulation.Aivika.Internal.Arrival -- | Accumulates the statistics about that how long the arrived events are processed. @@ -75,3 +79,26 @@ addSamplingStats (t - arrivalTime a) triggerSignal (arrivalProcessingTimeChangedSource timer) () return (a, Cons $ loop xs) + +-- | Return a signal that actually measures how much time has passed from +-- the time of arriving the events. +-- +-- Note that the statistics is counted each time you subscribe to the output signal. +-- For example, if you subscribe twice then the statistics counting is duplicated. +-- Ideally, you should subscribe to the output signal only once. +arrivalTimerSignal :: ArrivalTimer -> Signal (Arrival a) -> Signal (Arrival a) +arrivalTimerSignal timer sa = + Signal { handleSignal = \h -> + handleSignal sa $ \a -> + do t <- liftDynamics time + modifyRef (arrivalProcessingTimeRef timer) $ + addSamplingStats (t - arrivalTime a) + h a + } + +-- | Like 'arrivalTimerSignal' but measures how much time has passed from +-- the time of arriving the events in the channel. +arrivalTimerChannel :: ArrivalTimer -> Channel (Arrival a) (Arrival a) +arrivalTimerChannel timer = + Channel $ \sa -> + return $ arrivalTimerSignal timer sa diff --git a/Simulation/Aivika/Channel.hs b/Simulation/Aivika/Channel.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Channel.hs @@ -0,0 +1,80 @@ + +-- | +-- Module : Simulation.Aivika.Channel +-- Copyright : Copyright (c) 2009-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 8.0.1 +-- +-- The module defines a channel that transforms one 'Signal' to another +-- within the 'Composite' computation. +-- +module Simulation.Aivika.Channel + (-- * Channel Computation + Channel(..), + -- * Delay Channel + delayChannel, + delayChannelM, + -- * Sinking Signal + sinkSignal, + -- * Debugging + traceChannel) where + +import qualified Control.Category as C +import Control.Monad + +import Simulation.Aivika.Simulation +import Simulation.Aivika.Dynamics +import Simulation.Aivika.Event +import Simulation.Aivika.Signal +import Simulation.Aivika.Composite + +-- | It allows representing a signal transformation. +newtype Channel a b = + Channel { runChannel :: Signal a -> Composite (Signal b) + -- ^ Run the channel transform. + } + +instance C.Category Channel where + + id = Channel return + + (Channel g) . (Channel f) = + Channel $ \a -> f a >>= g + +-- | Return a delayed signal. +-- +-- This is actually the 'delaySignal' function wrapped in the 'Channel' type. +delayChannel :: Double -- ^ the delay + -> Channel a a -- ^ the delay channel +delayChannel delay = + Channel $ \a -> return $ delaySignal delay a + +-- | Like 'delayChannel', but it re-computes the delay each time. +-- +-- This is actually the 'delaySignalM' function wrapped in the 'Channel' type. +delayChannelM :: Event Double -- ^ the delay + -> Channel a a -- ^ the delay channel +delayChannelM delay = + Channel $ \a -> return $ delaySignalM delay a + +-- | Sink the signal. It returns a computation that subscribes to +-- the signal and then ignores the received data. The resulting +-- computation can be a moving force to simulate the whole system of +-- the interconnected signals and channels. +sinkSignal :: Signal a -> Composite () +sinkSignal a = + do h <- liftEvent $ + handleSignal a $ + const $ return () + disposableComposite h + +-- | Show the debug message with the current simulation time, +-- when emitting the output signal. +traceChannel :: String -> Channel a b -> Channel a b +traceChannel message (Channel f) = + Channel $ \a -> + do b <- f a + return $ + traceSignal message b diff --git a/Simulation/Aivika/Composite.hs b/Simulation/Aivika/Composite.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Composite.hs @@ -0,0 +1,127 @@ + +{-# LANGUAGE MultiParamTypeClasses, RecursiveDo #-} + +-- | +-- Module : Simulation.Aivika.Composite +-- Copyright : Copyright (c) 2009-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 8.0.1 +-- +-- It defines the 'Composite' monad that allows constructing components which +-- can be then destroyed in case of need. +-- +module Simulation.Aivika.Composite + (-- * Composite Monad + Composite, + runComposite, + runComposite_, + runCompositeInStartTime_, + runCompositeInStopTime_, + disposableComposite) where + +import Data.Monoid + +import Control.Exception +import Control.Monad +import Control.Monad.Trans +import Control.Monad.Fix +import Control.Applicative + +import Simulation.Aivika.Parameter +import Simulation.Aivika.Simulation +import Simulation.Aivika.Dynamics +import Simulation.Aivika.Event + +-- | It represents a composite which can be then destroyed in case of need. +newtype Composite a = Composite { runComposite :: DisposableEvent -> Event (a, DisposableEvent) + -- ^ Run the computation returning the result + -- and some 'DisposableEvent' that being applied + -- destroys the composite, for example, unsubscribes + -- from signals or cancels the processes. + -- + } + +-- | Like 'runComposite' but retains the composite parts during the simulation. +runComposite_ :: Composite a -> Event a +runComposite_ m = + do (a, _) <- runComposite m mempty + return a + +-- | Like 'runComposite_' but runs the computation in the start time. +runCompositeInStartTime_ :: Composite a -> Simulation a +runCompositeInStartTime_ = runEventInStartTime . runComposite_ + +-- | Like 'runComposite_' but runs the computation in the stop time. +runCompositeInStopTime_ :: Composite a -> Simulation a +runCompositeInStopTime_ = runEventInStopTime . runComposite_ + +-- | When destroying the composite, the specified action will be applied. +disposableComposite :: DisposableEvent -> Composite () +disposableComposite h = Composite $ \h0 -> return ((), h0 <> h) + +instance Functor Composite where + + fmap f (Composite m) = + Composite $ \h0 -> + do (a, h) <- m h0 + return (f a, h) + +instance Applicative Composite where + + pure = return + (<*>) = ap + +instance Monad Composite where + + return a = Composite $ \h0 -> return (a, h0) + + (Composite m) >>= k = + Composite $ \h0 -> + do (a, h) <- m h0 + let Composite m' = k a + (b, h') <- m' h + return (b, h') + +instance MonadIO Composite where + + liftIO m = + Composite $ \h0 -> + do a <- liftIO m + return (a, h0) + +instance MonadFix Composite where + + mfix f = + Composite $ \h0 -> + do rec (a, h) <- runComposite (f a) h0 + return (a, h) + +instance ParameterLift Composite where + + liftParameter m = + Composite $ \h0 -> + do a <- liftParameter m + return (a, h0) + +instance SimulationLift Composite where + + liftSimulation m = + Composite $ \h0 -> + do a <- liftSimulation m + return (a, h0) + +instance DynamicsLift Composite where + + liftDynamics m = + Composite $ \h0 -> + do a <- liftDynamics m + return (a, h0) + +instance EventLift Composite where + + liftEvent m = + Composite $ \h0 -> + do a <- liftEvent m + return (a, h0) diff --git a/Simulation/Aivika/Processor.hs b/Simulation/Aivika/Processor.hs --- a/Simulation/Aivika/Processor.hs +++ b/Simulation/Aivika/Processor.hs @@ -44,23 +44,29 @@ joinProcessor, -- * Failover failoverProcessor, - -- * Integrating with Signals - signalProcessor, - processorSignaling, + -- * Integrating with Signals and Channels + channelProcessor, + processorChannel, + queuedChannelProcessor, + queuedProcessorChannel, -- * Debugging 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 -- | Represents a processor of simulation data. @@ -424,39 +430,89 @@ prefetchProcessor :: Processor a a prefetchProcessor = Processor prefetchStream --- | Convert the specified signal transform to a processor. +-- | Convert the specified signal transform, i.e. the channel, to a processor. -- -- The processor may return data with delay as the values are requested by demand. -- Consider using the 'arrivalSignal' function to provide with the information -- about the time points at which the signal was actually triggered. -- -- The point is that the 'Stream' used in the 'Processor' is requested outside, --- while the 'Signal' is triggered inside. They are different by nature. +-- while the 'Signal' used in the 'Channel' is triggered inside. They are different by nature. -- The former is passive, while the latter is active. -- --- Cancel the processor's process to unsubscribe from the signals provided. -signalProcessor :: (Signal a -> Signal b) -> Processor a b -signalProcessor f = +-- The resulting processor may be a root of space leak as it uses an internal queue to store +-- the values received from the input signal. Consider using 'queuedChannelProcessor' that +-- allows specifying the bounded queue in case of need. +channelProcessor :: Channel a b -> Processor a b +channelProcessor f = Processor $ \xs -> Cons $ - do sa <- streamSignal xs - sb <- signalStream (f sa) - runStream sb + do let composite = + do sa <- streamSignal xs + sb <- runChannel f sa + signalStream sb + (ys, h) <- liftEvent $ + runComposite composite mempty + whenCancellingProcess $ + disposeEvent h + runStream ys --- | Convert the specified processor to a signal transform. +-- | Convert the specified processor to a signal transform, i.e. the channel. -- -- The processor may return data with delay as the values are requested by demand. -- Consider using the 'arrivalSignal' function to provide with the information -- about the time points at which the signal was actually triggered. -- -- The point is that the 'Stream' used in the 'Processor' is requested outside, --- while the 'Signal' is triggered inside. They are different by nature. +-- while the 'Signal' used in the 'Channel' is triggered inside. They are different by nature. -- The former is passive, while the latter is active. -- --- Cancel the returned process to unsubscribe from the signal specified. -processorSignaling :: Processor a b -> Signal a -> Process (Signal b) -processorSignaling (Processor f) sa = +-- The resulting channel may be a root of space leak as it uses an internal queue to store +-- the values received from the input stream. Consider using 'queuedProcessorChannel' that +-- allows specifying the bounded queue in case of need. +processorChannel :: Processor a b -> Channel a b +processorChannel (Processor f) = + Channel $ \sa -> do xs <- signalStream sa + let ys = f xs + streamSignal ys + +-- | Like 'channelProcessor' but allows specifying an arbitrary queue for storing the signal values, +-- for example, the bounded queue. +queuedChannelProcessor :: (b -> Event ()) + -- ^ enqueue + -> Process b + -- ^ dequeue + -> Channel a b + -- ^ the channel + -> Processor a b + -- ^ the processor +queuedChannelProcessor enqueue dequeue f = + Processor $ \xs -> + Cons $ + do let composite = + do sa <- streamSignal xs + sb <- runChannel f sa + queuedSignalStream enqueue dequeue sb + (ys, h) <- liftEvent $ + runComposite composite mempty + whenCancellingProcess $ + disposeEvent h + runStream ys + +-- | Like 'processorChannel' but allows specifying an arbitrary queue for storing the signal values, +-- for example, the bounded queue. +queuedProcessorChannel :: (a -> Event ()) + -- ^ enqueue + -> (Process a) + -- ^ dequeue + -> Processor a b + -- ^ the processor + -> Channel a b + -- ^ the channel +queuedProcessorChannel enqueue dequeue (Processor f) = + Channel $ \sa -> + do xs <- queuedSignalStream enqueue dequeue sa let ys = f xs streamSignal ys diff --git a/Simulation/Aivika/Signal/Random.hs b/Simulation/Aivika/Signal/Random.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Signal/Random.hs @@ -0,0 +1,235 @@ + +-- | +-- Module : Simulation.Aivika.Signal.Random +-- Copyright : Copyright (c) 2009-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 8.0.1 +-- +-- This module defines random signals of events, which are useful +-- for describing the input of the model. +-- + +module Simulation.Aivika.Signal.Random + (-- * Signal of Random Events + newRandomSignal, + newRandomUniformSignal, + newRandomUniformIntSignal, + newRandomTriangularSignal, + newRandomNormalSignal, + newRandomLogNormalSignal, + newRandomExponentialSignal, + newRandomErlangSignal, + newRandomPoissonSignal, + newRandomBinomialSignal, + newRandomGammaSignal, + newRandomBetaSignal, + newRandomWeibullSignal, + newRandomDiscreteSignal) where + +import Control.Monad +import Control.Monad.Trans + +import Simulation.Aivika.Generator +import Simulation.Aivika.Parameter +import Simulation.Aivika.Parameter.Random +import Simulation.Aivika.Simulation +import Simulation.Aivika.Dynamics +import Simulation.Aivika.Event +import Simulation.Aivika.Composite +import Simulation.Aivika.Process +import Simulation.Aivika.Signal +import Simulation.Aivika.Statistics +import Simulation.Aivika.Arrival + +-- | Return a signal of random events that arrive with the specified delay. +newRandomSignal :: Parameter (Double, a) + -- ^ compute a pair of the delay and event of type @a@ + -> Composite (Signal (Arrival a)) + -- ^ the computation that returns a signal emitting the delayed events +newRandomSignal delay = + do source <- liftSimulation newSignalSource + let loop t0 = + do (delay, a) <- liftParameter delay + when (delay > 0) $ + holdProcess delay + t2 <- liftDynamics time + let arrival = Arrival { arrivalValue = a, + arrivalTime = t2, + arrivalDelay = + case t0 of + Nothing -> Nothing + Just t0 -> Just delay } + liftEvent $ + triggerSignal source arrival + loop (Just t2) + pid <- liftSimulation newProcessId + liftEvent $ + runProcessUsingId pid $ + loop Nothing + disposableComposite $ + DisposableEvent $ + cancelProcessWithId pid + return $ publishSignal source + +-- | Create a new signal with random delays distributed uniformly. +newRandomUniformSignal :: Double + -- ^ the minimum delay + -> Double + -- ^ the maximum delay + -> Composite (Signal (Arrival Double)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomUniformSignal min max = + newRandomSignal $ + randomUniform min max >>= \x -> + return (x, x) + +-- | Create a new signal with integer random delays distributed uniformly. +newRandomUniformIntSignal :: Int + -- ^ the minimum delay + -> Int + -- ^ the maximum delay + -> Composite (Signal (Arrival Int)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomUniformIntSignal min max = + newRandomSignal $ + randomUniformInt min max >>= \x -> + return (fromIntegral x, x) + +-- | Create a new signal with random delays having the triangular distribution. +newRandomTriangularSignal :: Double + -- ^ the minimum delay + -> Double + -- ^ the median of the delay + -> Double + -- ^ the maximum delay + -> Composite (Signal (Arrival Double)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomTriangularSignal min median max = + newRandomSignal $ + randomTriangular min median max >>= \x -> + return (x, x) + +-- | Create a new signal with random delays distributed normally. +newRandomNormalSignal :: Double + -- ^ the mean delay + -> Double + -- ^ the delay deviation + -> Composite (Signal (Arrival Double)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomNormalSignal mu nu = + newRandomSignal $ + randomNormal mu nu >>= \x -> + return (x, x) + +-- | Create a new signal with random delays having the lognormal distribution. +newRandomLogNormalSignal :: Double + -- ^ the mean of a normal distribution which + -- this distribution is derived from + -> Double + -- ^ the deviation of a normal distribution which + -- this distribution is derived from + -> Composite (Signal (Arrival Double)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomLogNormalSignal mu nu = + newRandomSignal $ + randomLogNormal mu nu >>= \x -> + return (x, x) + +-- | Return a new signal with random delays distibuted exponentially with the specified mean +-- (the reciprocal of the rate). +newRandomExponentialSignal :: Double + -- ^ the mean delay (the reciprocal of the rate) + -> Composite (Signal (Arrival Double)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomExponentialSignal mu = + newRandomSignal $ + randomExponential mu >>= \x -> + return (x, x) + +-- | Return a new signal with random delays having the Erlang distribution with the specified +-- scale (the reciprocal of the rate) and shape parameters. +newRandomErlangSignal :: Double + -- ^ the scale (the reciprocal of the rate) + -> Int + -- ^ the shape + -> Composite (Signal (Arrival Double)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomErlangSignal beta m = + newRandomSignal $ + randomErlang beta m >>= \x -> + return (x, x) + +-- | Return a new signal with random delays having the Poisson distribution with +-- the specified mean. +newRandomPoissonSignal :: Double + -- ^ the mean delay + -> Composite (Signal (Arrival Int)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomPoissonSignal mu = + newRandomSignal $ + randomPoisson mu >>= \x -> + return (fromIntegral x, x) + +-- | Return a new signal with random delays having the binomial distribution with the specified +-- probability and trials. +newRandomBinomialSignal :: Double + -- ^ the probability + -> Int + -- ^ the number of trials + -> Composite (Signal (Arrival Int)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomBinomialSignal prob trials = + newRandomSignal $ + randomBinomial prob trials >>= \x -> + return (fromIntegral x, x) + +-- | Return a new signal with random delays having the Gamma distribution by the specified +-- shape and scale. +newRandomGammaSignal :: Double + -- ^ the shape + -> Double + -- ^ the scale (a reciprocal of the rate) + -> Composite (Signal (Arrival Double)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomGammaSignal kappa theta = + newRandomSignal $ + randomGamma kappa theta >>= \x -> + return (x, x) + +-- | Return a new signal with random delays having the Beta distribution by the specified +-- shape parameters (alpha and beta). +newRandomBetaSignal :: Double + -- ^ the shape (alpha) + -> Double + -- ^ the shape (beta) + -> Composite (Signal (Arrival Double)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomBetaSignal alpha beta = + newRandomSignal $ + randomBeta alpha beta >>= \x -> + return (x, x) + +-- | Return a new signal with random delays having the Weibull distribution by the specified +-- shape and scale. +newRandomWeibullSignal :: Double + -- ^ shape + -> Double + -- ^ scale + -> Composite (Signal (Arrival Double)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomWeibullSignal alpha beta = + newRandomSignal $ + randomWeibull alpha beta >>= \x -> + return (x, x) + +-- | Return a new signal with random delays having the specified discrete distribution. +newRandomDiscreteSignal :: DiscretePDF Double + -- ^ the discrete probability density function + -> Composite (Signal (Arrival Double)) + -- ^ the computation of signal emitting random events with the delays generated +newRandomDiscreteSignal dpdf = + newRandomSignal $ + randomDiscrete dpdf >>= \x -> + return (x, x) diff --git a/Simulation/Aivika/Stream.hs b/Simulation/Aivika/Stream.hs --- a/Simulation/Aivika/Stream.hs +++ b/Simulation/Aivika/Stream.hs @@ -66,6 +66,7 @@ -- * Integrating with Signals signalStream, streamSignal, + queuedSignalStream, -- * Utilities leftStream, rightStream, @@ -91,12 +92,13 @@ 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.Signal import Simulation.Aivika.Resource.Base import Simulation.Aivika.QueueStrategy -import Simulation.Aivika.Queue.Infinite.Base +import qualified Simulation.Aivika.Queue.Infinite.Base as IQ import Simulation.Aivika.Internal.Arrival -- | Represents an infinite stream of data in time, @@ -551,6 +553,21 @@ spawnProcess $ writer s runStream $ repeatProcess reader +-- | Like 'signalStream' but allows specifying an arbitrary queue instead of the unbounded queue. +queuedSignalStream :: (a -> Event ()) + -- ^ enqueue + -> Process a + -- ^ dequeue + -> Signal a + -- ^ the input signal + -> Composite (Stream a) + -- ^ the output stream +queuedSignalStream enqueue dequeue s = + do h <- liftEvent $ + handleSignal s enqueue + disposableComposite h + return $ repeatProcess dequeue + -- | Return a stream of values triggered by the specified signal. -- -- Since the time at which the values of the stream are requested for may differ from @@ -561,32 +578,30 @@ -- The point is that the 'Stream' is requested outside, while the 'Signal' is triggered -- inside. They are different by nature. The former is passive, while the latter is active. -- --- The resulting stream may be a root of space leak as it uses an internal queue to store +-- The resulting stream may be a root of space leak as it uses an internal unbounded queue to store -- the values received from the signal. The oldest value is dequeued each time we request --- the stream and it is returned within the computation. --- --- Cancel the stream's process to unsubscribe from the specified signal. -signalStream :: Signal a -> Process (Stream a) +-- the stream and it is returned within the computation. Consider using 'queuedSignalStream' that +-- allows specifying the bounded queue in case of need. +signalStream :: Signal a -> Composite (Stream a) signalStream s = - do q <- liftSimulation newFCFSQueue - h <- liftEvent $ - handleSignal s $ - enqueue q - whenCancellingProcess $ disposeEvent h - return $ repeatProcess $ dequeue q + do q <- liftSimulation IQ.newFCFSQueue + queuedSignalStream (IQ.enqueue q) (IQ.dequeue q) s --- | Return a computation of the signal that triggers values from the specified stream, +-- | Return a computation of the disposable signal that triggers values from the specified stream, -- each time the next value of the stream is received within the underlying 'Process' -- computation. --- --- Cancel the returned process to stop reading from the specified stream. -streamSignal :: Stream a -> Process (Signal a) +streamSignal :: Stream a -> Composite (Signal a) streamSignal z = do s <- liftSimulation newSignalSource - spawnProcess $ + pid <- liftSimulation newProcessId + liftEvent $ + runProcessUsingId pid $ consumeStream (liftEvent . triggerSignal s) z + disposableComposite $ + DisposableEvent $ + cancelProcessWithId pid return $ publishSignal s - + -- | Transform a stream so that the resulting stream returns a sequence of arrivals -- saving the information about the time points at which the original stream items -- were received by demand. @@ -716,16 +731,16 @@ -- | Create the specified number of equivalent clones of the input stream. cloneStream :: Int -> Stream a -> Simulation [Stream a] cloneStream n s = - do qs <- forM [1..n] $ \i -> newFCFSQueue + do qs <- forM [1..n] $ \i -> IQ.newFCFSQueue rs <- newFCFSResource 1 ref <- liftIO $ newIORef s let reader m q = - do a <- liftEvent $ tryDequeue q + do a <- liftEvent $ IQ.tryDequeue q case a of Just a -> return a Nothing -> usingResource rs $ - do a <- liftEvent $ tryDequeue q + do a <- liftEvent $ IQ.tryDequeue q case a of Just a -> return a Nothing -> @@ -734,7 +749,7 @@ liftIO $ writeIORef ref xs forM_ (zip [1..] qs) $ \(i, q) -> unless (i == m) $ - liftEvent $ enqueue q a + liftEvent $ IQ.enqueue q a return a forM (zip [1..] qs) $ \(i, q) -> return $ repeatProcess $ reader i q diff --git a/Simulation/Aivika/Stream/Random.hs b/Simulation/Aivika/Stream/Random.hs --- a/Simulation/Aivika/Stream/Random.hs +++ b/Simulation/Aivika/Stream/Random.hs @@ -63,7 +63,8 @@ "At least, they can be lost, for example, when trying to enqueue them, but " ++ "the random stream itself must always work: randomStream." (delay, a) <- liftParameter delay - holdProcess delay + when (delay > 0) $ + holdProcess delay t2 <- liftDynamics time let arrival = Arrival { arrivalValue = a, arrivalTime = t2, diff --git a/aivika.cabal b/aivika.cabal --- a/aivika.cabal +++ b/aivika.cabal @@ -1,5 +1,5 @@ name: aivika -version: 4.6 +version: 5.0.1 synopsis: A multi-method simulation library description: Aivika is a multi-method simulation library focused on @@ -148,7 +148,9 @@ Simulation.Aivika.Activity.Random Simulation.Aivika.Agent Simulation.Aivika.Arrival + Simulation.Aivika.Channel Simulation.Aivika.Circuit + Simulation.Aivika.Composite Simulation.Aivika.Cont Simulation.Aivika.DoubleLinkedList Simulation.Aivika.Dynamics @@ -190,6 +192,7 @@ Simulation.Aivika.Server Simulation.Aivika.Server.Random Simulation.Aivika.Signal + Simulation.Aivika.Signal.Random Simulation.Aivika.Simulation Simulation.Aivika.Specs Simulation.Aivika.Statistics