aivika 4.6 → 5.0.1

raw patch · 10 files changed

+609/−40 lines, 10 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +19/−0
• Simulation/Aivika.hs +6/−0
• Simulation/Aivika/Arrival.hs +27/−0
• Simulation/Aivika/Channel.hs +80/−0
• Simulation/Aivika/Composite.hs +127/−0
• Simulation/Aivika/Processor.hs +72/−16
• Simulation/Aivika/Signal/Random.hs +235/−0
• Simulation/Aivika/Stream.hs +37/−22
• Simulation/Aivika/Stream/Random.hs +2/−1
• aivika.cabal +4/−1

@@ -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 ----- 

@@ -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

@@ -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

@@ -0,0 +1,80 @@++-- |+-- Module     : Simulation.Aivika.Channel+-- Copyright  : Copyright (c) 2009-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- 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

@@ -0,0 +1,127 @@++{-# LANGUAGE MultiParamTypeClasses, RecursiveDo #-}++-- |+-- Module     : Simulation.Aivika.Composite+-- Copyright  : Copyright (c) 2009-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- 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)

@@ -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 

@@ -0,0 +1,235 @@++-- |+-- Module     : Simulation.Aivika.Signal.Random+-- Copyright  : Copyright (c) 2009-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- 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)

@@ -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

@@ -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,

@@ -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