diff --git a/Simulation/Aivika.hs b/Simulation/Aivika.hs --- a/Simulation/Aivika.hs +++ b/Simulation/Aivika.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika --- Copyright : Copyright (c) 2009-2013, David Sorokin +-- Copyright : Copyright (c) 2009-2014, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -13,6 +13,7 @@ module Simulation.Aivika (-- * Modules module Simulation.Aivika.Agent, + module Simulation.Aivika.Arrival, module Simulation.Aivika.Cont, module Simulation.Aivika.Dynamics, module Simulation.Aivika.Dynamics.Interpolate, @@ -33,12 +34,14 @@ module Simulation.Aivika.Simulation, module Simulation.Aivika.Specs, module Simulation.Aivika.Statistics, + module Simulation.Aivika.Statistics.Accumulator, module Simulation.Aivika.Stream, module Simulation.Aivika.Stream.Random, module Simulation.Aivika.Task, module Simulation.Aivika.Var.Unboxed) where import Simulation.Aivika.Agent +import Simulation.Aivika.Arrival import Simulation.Aivika.Cont import Simulation.Aivika.Dynamics import Simulation.Aivika.Dynamics.Interpolate @@ -59,8 +62,8 @@ import Simulation.Aivika.Simulation import Simulation.Aivika.Specs import Simulation.Aivika.Statistics +import Simulation.Aivika.Statistics.Accumulator import Simulation.Aivika.Stream import Simulation.Aivika.Stream.Random import Simulation.Aivika.Task import Simulation.Aivika.Var.Unboxed - diff --git a/Simulation/Aivika/Arrival.hs b/Simulation/Aivika/Arrival.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Arrival.hs @@ -0,0 +1,72 @@ + +-- | +-- Module : Simulation.Aivika.Arrival +-- Copyright : Copyright (c) 2009-2014, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 7.6.3 +-- +-- This module defines the types and functions for working with +-- the external events that usually arrive from outside the model. + +module Simulation.Aivika.Arrival + (Arrival(..), + ArrivalTimer, + newArrivalTimer, + arrivalTimerProcessor, + arrivalProcessingTime) where + +import Control.Monad +import Control.Monad.Trans + +import Simulation.Aivika.Simulation +import Simulation.Aivika.Dynamics +import Simulation.Aivika.Event +import Simulation.Aivika.Processor +import Simulation.Aivika.Stream +import Simulation.Aivika.Statistics +import Simulation.Aivika.Ref + +-- | It defines when an external event has arrived, usually generated by +-- some random 'Stream'. +-- +-- Such events should arrive one by one without time lag in the following sense +-- that the model should start awaiting the next event exactly in that time +-- when the previous event has arrived. It usually happens automatically when +-- using a 'queueProcessor' with an ability to lost the arrived event if the queue +-- is full. +data Arrival = + Arrival { arrivalTime :: Double, + -- ^ the simulation time at which the event has arrived + arrivalDelay :: Double + -- ^ the delay time which has passed from the time of + -- arriving the previous event + } deriving (Eq, Ord, Show) + +-- | Accumulates the statistics about that how long the arrived events are processed. +data ArrivalTimer = + ArrivalTimer { arrivalProcessingTimeRef :: Ref (SamplingStats Double) } + +-- | Create a new timer that measures how long the arrived events are processed. +newArrivalTimer :: Simulation ArrivalTimer +newArrivalTimer = + do r <- newRef emptySamplingStats + return ArrivalTimer { arrivalProcessingTimeRef = r } + +-- | Return the statistics about that how long the arrived events were processed. +arrivalProcessingTime :: ArrivalTimer -> Event (SamplingStats Double) +arrivalProcessingTime = readRef . arrivalProcessingTimeRef + +-- | Return a processor that actually measures how much time has passed from +-- the time of arriving the events. +arrivalTimerProcessor :: ArrivalTimer -> Processor Arrival Arrival +arrivalTimerProcessor timer = + Processor $ \xs -> Cons $ loop xs where + loop xs = + do (a, xs) <- runStream xs + liftEvent $ + do t <- liftDynamics time + modifyRef (arrivalProcessingTimeRef timer) $ + addSamplingStats (t - arrivalTime a) + return (a, Cons $ loop xs) diff --git a/Simulation/Aivika/Dynamics/Memo.hs b/Simulation/Aivika/Dynamics/Memo.hs --- a/Simulation/Aivika/Dynamics/Memo.hs +++ b/Simulation/Aivika/Dynamics/Memo.hs @@ -23,6 +23,7 @@ import Control.Monad import Simulation.Aivika.Internal.Specs +import Simulation.Aivika.Internal.Parameter import Simulation.Aivika.Internal.Simulation import Simulation.Aivika.Internal.Dynamics import Simulation.Aivika.Dynamics.Interpolate @@ -32,7 +33,8 @@ newBoxedArray_ = newArray_ -- | Memoize and order the computation in the integration time points using --- the interpolation that knows of the Runge-Kutta method. +-- the interpolation that knows of the Runge-Kutta method. The values are +-- calculated sequentially starting from 'starttime'. memoDynamics :: Dynamics e -> Simulation (Dynamics e) {-# INLINE memoDynamics #-} memoDynamics (Dynamics m) = diff --git a/Simulation/Aivika/Dynamics/Memo/Unboxed.hs b/Simulation/Aivika/Dynamics/Memo/Unboxed.hs --- a/Simulation/Aivika/Dynamics/Memo/Unboxed.hs +++ b/Simulation/Aivika/Dynamics/Memo/Unboxed.hs @@ -24,13 +24,15 @@ import Control.Monad import Simulation.Aivika.Internal.Specs +import Simulation.Aivika.Internal.Parameter import Simulation.Aivika.Internal.Simulation import Simulation.Aivika.Internal.Dynamics import Simulation.Aivika.Dynamics.Interpolate import Simulation.Aivika.Unboxed -- | Memoize and order the computation in the integration time points using --- the interpolation that knows of the Runge-Kutta method. +-- the interpolation that knows of the Runge-Kutta method. The values are +-- calculated sequentially starting from 'starttime'. memoDynamics :: Unboxed e => Dynamics e -> Simulation (Dynamics e) {-# INLINE memoDynamics #-} memoDynamics (Dynamics m) = diff --git a/Simulation/Aivika/Dynamics/Random.hs b/Simulation/Aivika/Dynamics/Random.hs --- a/Simulation/Aivika/Dynamics/Random.hs +++ b/Simulation/Aivika/Dynamics/Random.hs @@ -7,7 +7,16 @@ -- Stability : experimental -- Tested with: GHC 7.6.3 -- --- This module defines the random parameters of simulation experiments. +-- This module defines the random functions that always return the same values +-- in the integration time points within a single simulation run. The values +-- for another simulation run will be regenerated anew. +-- +-- For example, the computations returned by these functions can be used in +-- the equations of System Dynamics. +-- +-- Also it is worth noting that the values are generated in a strong order starting +-- from 'starttime' with step 'dt'. This is how the 'memo0Dynamics' function +-- actually works. -- module Simulation.Aivika.Dynamics.Random diff --git a/Simulation/Aivika/Event.hs b/Simulation/Aivika/Event.hs --- a/Simulation/Aivika/Event.hs +++ b/Simulation/Aivika/Event.hs @@ -16,6 +16,7 @@ EventLift(..), EventProcessing(..), runEvent, + runEventWith, runEventInStartTime, runEventInStopTime, -- * Event Queue diff --git a/Simulation/Aivika/Internal/Event.hs b/Simulation/Aivika/Internal/Event.hs --- a/Simulation/Aivika/Internal/Event.hs +++ b/Simulation/Aivika/Internal/Event.hs @@ -19,6 +19,7 @@ EventProcessing(..), invokeEvent, runEvent, + runEventWith, runEventInStartTime, runEventInStopTime, -- * Event Queue @@ -148,21 +149,23 @@ do { rec { a <- invokeEvent p (f a) }; return a } -- | Defines how the events are processed. -data EventProcessing = IncludingCurrentEvents +data EventProcessing = CurrentEvents -- ^ either process all earlier and then current events, -- or raise an error if the current simulation time is less - -- than the actual time of the event queue - | IncludingEarlierEvents + -- than the actual time of the event queue (safe within + -- the 'Event' computation as this is protected by the type system) + | EarlierEvents -- ^ either process all earlier events not affecting -- the events at the current simulation time, -- or raise an error if the current simulation time is less - -- than the actual time of the event queue - | IncludingCurrentEventsOrFromPast + -- than the actual time of the event queue (safe within + -- the 'Event' computation as this is protected by the type system) + | CurrentEventsOrFromPast -- ^ either process all earlier and then current events, -- or do nothing if the current simulation time is less -- than the actual time of the event queue -- (do not use unless the documentation states the opposite) - | IncludingEarlierEventsOrFromPast + | EarlierEventsOrFromPast -- ^ either process all earlier events, -- or do nothing if the current simulation time is less -- than the actual time of the event queue @@ -248,28 +251,35 @@ -- | Process the events. processEvents :: EventProcessing -> Dynamics () -processEvents IncludingCurrentEvents = processEventsIncludingCurrent -processEvents IncludingEarlierEvents = processEventsIncludingEarlier -processEvents IncludingCurrentEventsOrFromPast = processEventsIncludingCurrentCore -processEvents IncludingEarlierEventsOrFromPast = processEventsIncludingEarlierCore +processEvents CurrentEvents = processEventsIncludingCurrent +processEvents EarlierEvents = processEventsIncludingEarlier +processEvents CurrentEventsOrFromPast = processEventsIncludingCurrentCore +processEvents EarlierEventsOrFromPast = processEventsIncludingEarlierCore -- | Run the 'Event' computation in the current simulation time --- within the 'Dynamics' computation. -runEvent :: EventProcessing -> Event a -> Dynamics a -runEvent processing (Event e) = +-- within the 'Dynamics' computation involving all pending +-- 'CurrentEvents' in the processing too. +runEvent :: Event a -> Dynamics a +runEvent = runEventWith CurrentEvents + +-- | Run the 'Event' computation in the current simulation time +-- within the 'Dynamics' computation specifying what pending events +-- should be involved in the processing. +runEventWith :: EventProcessing -> Event a -> Dynamics a +runEventWith processing (Event e) = Dynamics $ \p -> do invokeDynamics p $ processEvents processing e p --- | Run the 'Event' computation in the start time. -runEventInStartTime :: EventProcessing -> Event a -> Simulation a -runEventInStartTime processing e = - runDynamicsInStartTime $ runEvent processing e +-- | Run the 'Event' computation in the start time involving all +-- pending 'CurrentEvents' in the processing too. +runEventInStartTime :: Event a -> Simulation a +runEventInStartTime = runDynamicsInStartTime . runEvent --- | Run the 'Event' computation in the stop time. -runEventInStopTime :: EventProcessing -> Event a -> Simulation a -runEventInStopTime processing e = - runDynamicsInStopTime $ runEvent processing e +-- | Run the 'Event' computation in the stop time involving all +-- pending 'CurrentEvents' in the processing too. +runEventInStopTime :: Event a -> Simulation a +runEventInStopTime = runDynamicsInStopTime . runEvent -- | Return the number of pending events that should -- be yet actuated. diff --git a/Simulation/Aivika/Internal/Parameter.hs b/Simulation/Aivika/Internal/Parameter.hs --- a/Simulation/Aivika/Internal/Parameter.hs +++ b/Simulation/Aivika/Internal/Parameter.hs @@ -12,6 +12,9 @@ -- The module defines the 'Parameter' monad that allows representing the model -- parameters. For example, they can be used when running the Monte-Carlo simulation. -- +-- In general, this monad is very useful for representing a computation which is external +-- relative to the model itself. +-- module Simulation.Aivika.Internal.Parameter (-- * Parameter Parameter(..), @@ -52,7 +55,10 @@ import Simulation.Aivika.Internal.Specs -- | The 'Parameter' monad that allows specifying the model parameters. --- For example, they can be used when running the Monte-Carlo simulation. +-- For example, they can be used when running the Monte-Carlo simulation. +-- +-- In general, this monad is very useful for representing a computation which is external +-- relative to the model itself. newtype Parameter a = Parameter (Run -> IO a) instance Monad Parameter where diff --git a/Simulation/Aivika/Internal/Process.hs b/Simulation/Aivika/Internal/Process.hs --- a/Simulation/Aivika/Internal/Process.hs +++ b/Simulation/Aivika/Internal/Process.hs @@ -47,7 +47,7 @@ passivateProcess, processPassive, reactivateProcess, - cancelProcessUsingId, + cancelProcessWithId, cancelProcess, processCancelled, -- * Awaiting Signal @@ -225,25 +225,28 @@ ccont = return m = invokeProcess pid p --- | Run the process in the start time immediately. -runProcessInStartTime :: EventProcessing -> Process () -> Simulation () -runProcessInStartTime processing p = - runEventInStartTime processing $ runProcess p +-- | Run the process in the start time immediately involving all pending +-- 'CurrentEvents' in the computation too. +runProcessInStartTime :: Process () -> Simulation () +runProcessInStartTime = runEventInStartTime . runProcess --- | Run the process in the start time immediately using the specified identifier. -runProcessInStartTimeUsingId :: EventProcessing -> ProcessId -> Process () -> Simulation () -runProcessInStartTimeUsingId processing pid p = - runEventInStartTime processing $ runProcessUsingId pid p +-- | Run the process in the start time immediately using the specified identifier +-- and involving all pending 'CurrentEvents' in the computation too. +runProcessInStartTimeUsingId :: ProcessId -> Process () -> Simulation () +runProcessInStartTimeUsingId pid p = + runEventInStartTime $ runProcessUsingId pid p --- | Run the process in the final simulation time immediately. -runProcessInStopTime :: EventProcessing -> Process () -> Simulation () -runProcessInStopTime processing p = - runEventInStopTime processing $ runProcess p +-- | Run the process in the final simulation time immediately involving all +-- pending 'CurrentEvents' in the computation too. +runProcessInStopTime :: Process () -> Simulation () +runProcessInStopTime = runEventInStopTime . runProcess --- | Run the process in the final simulation time immediately using the specified identifier. -runProcessInStopTimeUsingId :: EventProcessing -> ProcessId -> Process () -> Simulation () -runProcessInStopTimeUsingId processing pid p = - runEventInStopTime processing $ runProcessUsingId pid p +-- | Run the process in the final simulation time immediately using +-- the specified identifier and involving all pending 'CurrentEvents' +-- in the computation too. +runProcessInStopTimeUsingId :: ProcessId -> Process () -> Simulation () +runProcessInStopTimeUsingId pid p = + runEventInStopTime $ runProcessUsingId pid p -- | Enqueue the process that will be then started at the specified time -- from the event queue. @@ -278,15 +281,15 @@ processInterruptVersion = v } -- | Cancel a process with the specified identifier, interrupting it if needed. -cancelProcessUsingId :: ProcessId -> Event () -cancelProcessUsingId pid = contCancellationInitiate (processCancelSource pid) +cancelProcessWithId :: ProcessId -> Event () +cancelProcessWithId pid = contCancellationInitiate (processCancelSource pid) -- | The process cancels itself. cancelProcess :: Process a cancelProcess = do pid <- processId - liftEvent $ cancelProcessUsingId pid - throwProcess $ error "The process must be cancelled already: cancelProcessItself." + liftEvent $ cancelProcessWithId pid + throwProcess $ error "The process must be cancelled already: cancelProcess." -- | Test whether the process with the specified identifier was cancelled. processCancelled :: ProcessId -> Event Bool @@ -558,7 +561,7 @@ finallyProcess (holdProcess timeout) (liftEvent $ - cancelProcessUsingId pid) + cancelProcessWithId pid) spawnProcessUsingId CancelChildAfterParent pid $ do r <- liftIO $ newIORef Nothing finallyProcess diff --git a/Simulation/Aivika/Internal/Signal.hs b/Simulation/Aivika/Internal/Signal.hs --- a/Simulation/Aivika/Internal/Signal.hs +++ b/Simulation/Aivika/Internal/Signal.hs @@ -41,6 +41,7 @@ SignalHistory, signalHistorySignal, newSignalHistory, + newSignalHistoryStartingWith, readSignalHistory, -- * Signalable Computations Signalable(..), @@ -251,13 +252,26 @@ -- | Create a history of the signal values. newSignalHistory :: Signal a -> Event (SignalHistory a) -newSignalHistory signal = - do ts <- liftIO UV.newVector - xs <- liftIO V.newVector - handleSignal_ signal $ \a -> +newSignalHistory = + newSignalHistoryStartingWith Nothing + +-- | Create a history of the signal values starting with +-- the optional initial value. +newSignalHistoryStartingWith :: Maybe a -> Signal a -> Event (SignalHistory a) +newSignalHistoryStartingWith init signal = + Event $ \p -> + do ts <- UV.newVector + xs <- V.newVector + case init of + Nothing -> return () + Just a -> + do UV.appendVector ts (pointTime p) + V.appendVector xs a + invokeEvent p $ + handleSignal_ signal $ \a -> Event $ \p -> - do liftIO $ UV.appendVector ts (pointTime p) - liftIO $ V.appendVector xs a + do UV.appendVector ts (pointTime p) + V.appendVector xs a return SignalHistory { signalHistorySignal = signal, signalHistoryTimes = ts, signalHistoryValues = xs } diff --git a/Simulation/Aivika/Internal/Specs.hs b/Simulation/Aivika/Internal/Specs.hs --- a/Simulation/Aivika/Internal/Specs.hs +++ b/Simulation/Aivika/Internal/Specs.hs @@ -86,14 +86,14 @@ -- | Returns the integration iterations starting from zero. integIterations :: Specs -> [Int] integIterations sc = [i1 .. i2] where - i1 = 0 - i2 = round ((spcStopTime sc - - spcStartTime sc) / spcDT sc) + i1 = integIterationLoBnd sc + i2 = integIterationHiBnd sc -- | Returns the first and last integration iterations. integIterationBnds :: Specs -> (Int, Int) -integIterationBnds sc = (0, round ((spcStopTime sc - - spcStartTime sc) / spcDT sc)) +integIterationBnds sc = (i1, i2) where + i1 = integIterationLoBnd sc + i2 = integIterationHiBnd sc -- | Returns the first integration iteration, i.e. zero. integIterationLoBnd :: Specs -> Int @@ -101,8 +101,28 @@ -- | Returns the last integration iteration. integIterationHiBnd :: Specs -> Int -integIterationHiBnd sc = round ((spcStopTime sc - - spcStartTime sc) / spcDT sc) +integIterationHiBnd sc = + let n = round ((spcStopTime sc - + spcStartTime sc) / spcDT sc) + in if n < 0 + then + error $ + "The iteration number in the stop time has a negative value. " ++ + "Either the simulation specs are incorrect, " ++ + "or a floating point overflow occurred, " ++ + "for example, when using a too small integration time step. " ++ + "You have to define this time step regardless of " ++ + "whether you actually use it or not, " ++ + "for Aivika allows combining the ordinary differential equations " ++ + "with the discrete event simulation within one model. " ++ + "So, if you are still using the 32-bit architecture and " ++ + "you do need a small integration time step " ++ + "for integrating the equations " ++ + "then you might think of using the 64-bit architecture. " ++ + "Although you could probably just forget " ++ + "to increase the time step " ++ + "after increasing the stop time: integIterationHiBnd" + else n -- | Returns the phases for the specified simulation specs starting from zero. integPhases :: Specs -> [Int] diff --git a/Simulation/Aivika/Parameter.hs b/Simulation/Aivika/Parameter.hs --- a/Simulation/Aivika/Parameter.hs +++ b/Simulation/Aivika/Parameter.hs @@ -8,6 +8,9 @@ -- -- The module defines the 'Parameter' monad that allows representing the model -- parameters. For example, they can be used when running the Monte-Carlo simulation. +-- +-- In general, this monad is very useful for representing a computation which is external +-- relative to the model itself. -- module Simulation.Aivika.Parameter (-- * Parameter diff --git a/Simulation/Aivika/Parameter/Random.hs b/Simulation/Aivika/Parameter/Random.hs --- a/Simulation/Aivika/Parameter/Random.hs +++ b/Simulation/Aivika/Parameter/Random.hs @@ -9,6 +9,15 @@ -- -- This module defines the random parameters of simulation experiments. -- +-- To create a parameter that would return the same value within the simulation run, +-- you should memoize the computation with help of 'memoParameter', which is important +-- for the Monte-Carlo simulation. +-- +-- To create a random function that would return the same values in the integration +-- time points within the simulation run, you should either lift the computation to +-- the 'Dynamics' computation and then memoize it too but using the 'memo0Dynamics' +-- function for that computation, or just take the predefined function that does +-- namely this. module Simulation.Aivika.Parameter.Random (randomUniform, @@ -16,7 +25,9 @@ randomExponential, randomErlang, randomPoisson, - randomBinomial) where + randomBinomial, + randomTrue, + randomFalse) where import System.Random @@ -25,17 +36,10 @@ import Simulation.Aivika.Generator import Simulation.Aivika.Internal.Specs import Simulation.Aivika.Internal.Parameter +import Simulation.Aivika.Dynamics +import Simulation.Aivika.Dynamics.Memo.Unboxed -- | Computation that generates a new random number distributed uniformly. --- --- To create a parameter that would return the same value within the simulation run, --- you should memoize the computation, which is important for the Monte-Carlo simulation. --- --- To create a random function that would return the same values in the integration --- time points within the simulation run, you should either lift the computation to --- the @Dynamics@ computation and then memoize it too but using the corresponded --- function for that computation, or just take the predefined function that does --- namely this. randomUniform :: Double -- ^ minimum -> Double -- ^ maximum -> Parameter Double @@ -45,15 +49,6 @@ in generatorUniform g min max -- | Computation that generates a new random number distributed normally. --- --- To create a parameter that would return the same value within the simulation run, --- you should memoize the computation, which is important for the Monte-Carlo simulation. --- --- To create a random function that would return the same values in the integration --- time points within the simulation run, you should either lift the computation to --- the @Dynamics@ computation and then memoize it too but using the corresponded --- function for that computation, or just take the predefined function that does --- namely this. randomNormal :: Double -- ^ mean -> Double -- ^ deviation -> Parameter Double @@ -64,15 +59,6 @@ -- | Computation that returns a new exponential random number with the specified mean -- (the reciprocal of the rate). --- --- To create a parameter that would return the same value within the simulation run, --- you should memoize the computation, which is important for the Monte-Carlo simulation. --- --- To create a random function that would return the same values in the integration --- time points within the simulation run, you should either lift the computation to --- the @Dynamics@ computation and then memoize it too but using the corresponded --- function for that computation, or just take the predefined function that does --- namely this. randomExponential :: Double -- ^ the mean (the reciprocal of the rate) -> Parameter Double @@ -83,15 +69,6 @@ -- | Computation that returns a new Erlang random number with the specified scale -- (the reciprocal of the rate) and integer shape. --- --- To create a parameter that would return the same value within the simulation run, --- you should memoize the computation, which is important for the Monte-Carlo simulation. --- --- To create a random function that would return the same values in the integration --- time points within the simulation run, you should either lift the computation to --- the @Dynamics@ computation and then memoize it too but using the corresponded --- function for that computation, or just take the predefined function that does --- namely this. randomErlang :: Double -- ^ the scale (the reciprocal of the rate) -> Int @@ -103,15 +80,6 @@ in generatorErlang g beta m -- | Computation that returns a new Poisson random number with the specified mean. --- --- To create a parameter that would return the same value within the simulation run, --- you should memoize the computation, which is important for the Monte-Carlo simulation. --- --- To create a random function that would return the same values in the integration --- time points within the simulation run, you should either lift the computation to --- the @Dynamics@ computation and then memoize it too but using the corresponded --- function for that computation, or just take the predefined function that does --- namely this. randomPoisson :: Double -- ^ the mean -> Parameter Int @@ -122,15 +90,6 @@ -- | Computation that returns a new binomial random number with the specified -- probability and trials. --- --- To create a parameter that would return the same value within the simulation run, --- you should memoize the computation, which is important for the Monte-Carlo simulation. --- --- To create a random function that would return the same values in the integration --- time points within the simulation run, you should either lift the computation to --- the @Dynamics@ computation and then memoize it too but using the corresponded --- function for that computation, or just take the predefined function that does --- namely this. randomBinomial :: Double -- ^ the probability -> Int -- ^ the number of trials -> Parameter Int @@ -138,3 +97,17 @@ Parameter $ \r -> let g = runGenerator r in generatorBinomial g prob trials + +-- | Computation that returns 'True' in case of success. +randomTrue :: Double -- ^ the probability of the success + -> Parameter Bool +randomTrue p = + do x <- randomUniform 0 1 + return (x <= p) + +-- | Computation that returns 'False' in case of success. +randomFalse :: Double -- ^ the probability of the success + -> Parameter Bool +randomFalse p = + do x <- randomUniform 0 1 + return (x > p) diff --git a/Simulation/Aivika/Process.hs b/Simulation/Aivika/Process.hs --- a/Simulation/Aivika/Process.hs +++ b/Simulation/Aivika/Process.hs @@ -35,7 +35,7 @@ -- * Spawning Processes spawnProcess, spawnProcessUsingId, - -- * Enqueuing Process + -- * Enqueueing Process enqueueProcess, enqueueProcessUsingId, -- * Creating Process Identifier @@ -49,7 +49,7 @@ passivateProcess, processPassive, reactivateProcess, - cancelProcessUsingId, + cancelProcessWithId, cancelProcess, processCancelled, -- * Awaiting Signal diff --git a/Simulation/Aivika/Processor.hs b/Simulation/Aivika/Processor.hs --- a/Simulation/Aivika/Processor.hs +++ b/Simulation/Aivika/Processor.hs @@ -17,6 +17,8 @@ statefulProcessor, -- * Specifying Identifier processorUsingId, + -- * Prefetch Processor + prefetchProcessor, -- * Buffer Processor bufferProcessor, bufferProcessorLoop, @@ -291,25 +293,29 @@ runStream output -- | Like 'bufferProcessor' but allows creating a loop when some items --- can be returned for processing them again. It is very useful for --- modeling the processors with queues and loop-backs. +-- can be processed repeatedly. It is very useful for modeling the processors +-- with queues and loop-backs. bufferProcessorLoop :: (Stream a -> Stream c -> Process ()) -- ^ consume two streams: the input values of type @a@ -- and the values of type @c@ returned by the loop -> Stream d -- ^ the stream of data that may become results - -> Processor d (Either c b) - -- ^ process and then decide what values of type @c@ - -- should be processed again + -> Processor d (Either e b) + -- ^ process and then decide what values of type @e@ + -- should be processed in the loop (this is a condition) + -> Processor e c + -- ^ process in the loop and then return a value + -- of type @c@ to the input again (this is a loop body) -> Processor a b -bufferProcessorLoop consume preoutput filter = +bufferProcessorLoop consume preoutput cond body = Processor $ \xs -> Cons $ do (reverted, output) <- liftSimulation $ partitionEitherStream $ - runProcessor filter preoutput - spawnProcess CancelTogether (consume xs reverted) + runProcessor cond preoutput + spawnProcess CancelTogether + (consume xs $ runProcessor body reverted) runStream output -- | Return a processor with help of which we can model the queue. @@ -335,7 +341,7 @@ queueProcessor :: (a -> Process ()) -- ^ enqueue the input item and wait -- while the queue is full if required - -- so that there was no hanging items + -- so that there were no hanging items -> Process b -- ^ dequeue an output item -> Processor a b @@ -345,21 +351,24 @@ (consumeStream enqueue) (repeatProcess dequeue) --- | Like 'queueProcessor' creates a queue processor but allows creating --- a loop when some items can be returned and added to the queue again. --- Also it allows specifying how two input streams of data can be merged. +-- | Like 'queueProcessor' creates a queue processor but with a loop when some items +-- can be processed and then added to the queue again. Also it allows specifying +-- how two input streams of data can be merged. queueProcessorLoopMerging :: (Stream a -> Stream d -> Stream e) -- ^ merge two streams: the input values of type @a@ -- and the values of type @d@ returned by the loop -> (e -> Process ()) -- ^ enqueue the input item and wait -- while the queue is full if required - -- so that there was no hanging items + -- so that there were no hanging items -> Process c -- ^ dequeue an item for the further processing - -> Processor c (Either d b) - -- ^ process and then decide what values of type @d@ - -- should be processed again + -> Processor c (Either f b) + -- ^ process and then decide what values of type @f@ + -- should be processed in the loop (this is a condition) + -> Processor f d + -- ^ process in the loop and then return a value + -- of type @d@ to the queue again (this is a loop body) -> Processor a b -- ^ the buffering processor queueProcessorLoopMerging merge enqueue dequeue = @@ -369,39 +378,44 @@ merge bs cs) (repeatProcess dequeue) --- | Like 'queueProcessorLoopMerging' creates a queue processor and allows --- creating a loop when some items can be returned and added to the queue again. --- Only it sequentially merges two input streams of data: one stream --- that come from the external source and another stream of data returned --- by the loop. The first stream has a priority over the second one. +-- | Like 'queueProcessorLoopMerging' creates a queue processor with a loop when +-- some items can be processed and then added to the queue again. Only it sequentially +-- merges two input streams of data: one stream that come from the external source and +-- another stream of data returned by the loop. The first stream has a priority over +-- the second one. queueProcessorLoopSeq :: (a -> Process ()) -- ^ enqueue the input item and wait -- while the queue is full if required - -- so that there was no hanging items + -- so that there were no hanging items -> Process c -- ^ dequeue an item for the further processing - -> Processor c (Either a b) - -- ^ process and then decide what values of type @a@ - -- should be processed again + -> Processor c (Either e b) + -- ^ process and then decide what values of type @e@ + -- should be processed in the loop (this is a condition) + -> Processor e a + -- ^ process in the loop and then return a value + -- of type @a@ to the queue again (this is a loop body) -> Processor a b -- ^ the buffering processor queueProcessorLoopSeq = queueProcessorLoopMerging mergeStreams --- | Like 'queueProcessorLoopMerging' creates a queue processor and allows --- creating a loop when some items can be returned and added to the queue again. --- Only it runs two simultaneous processes to enqueue the input streams of data: --- one stream that come from the external source and another stream of data returned --- by the loop. +-- | Like 'queueProcessorLoopMerging' creates a queue processor with a loop when +-- some items can be processed and then added to the queue again. Only it runs two +-- simultaneous processes to enqueue the input streams of data: one stream that come +-- from the external source and another stream of data returned by the loop. queueProcessorLoopParallel :: (a -> Process ()) -- ^ enqueue the input item and wait -- while the queue is full if required - -- so that there was no hanging items + -- so that there were no hanging items -> Process c -- ^ dequeue an item for the further processing - -> Processor c (Either a b) - -- ^ process and then decide what values of type @a@ - -- should be processed again + -> Processor c (Either e b) + -- ^ process and then decide what values of type @e@ + -- should be processed in the loop (this is a condition) + -> Processor e a + -- ^ process in the loop and then return a value + -- of type @a@ to the queue again (this is a loop body) -> Processor a b -- ^ the buffering processor queueProcessorLoopParallel enqueue dequeue = @@ -412,3 +426,13 @@ spawnProcess CancelTogether $ consumeStream enqueue cs) (repeatProcess dequeue) + +-- | This is a prefetch processor that requests for one more data item from +-- the input in advance while the latest item is not yet fully processed in +-- the chain of streams, usually by other processors. +-- +-- You can think of this as the prefetched processor could place its latest +-- data item in some temporary space for later use, which is very useful +-- for modeling a sequence of separate and independent work places. +prefetchProcessor :: Processor a a +prefetchProcessor = Processor prefetchStream diff --git a/Simulation/Aivika/Queue.hs b/Simulation/Aivika/Queue.hs --- a/Simulation/Aivika/Queue.hs +++ b/Simulation/Aivika/Queue.hs @@ -26,27 +26,27 @@ newPriorityQueue, newQueue, -- * Queue Properties and Activities - queueInputStrategy, - queueStoringStrategy, - queueOutputStrategy, + enqueueStrategy, + enqueueStoringStrategy, + dequeueStrategy, queueNull, queueFull, queueMaxCount, queueCount, - queueLostCount, - queueInputCount, - queueStoreCount, - queueOutputRequestCount, - queueOutputCount, + enqueueCount, + enqueueLostCount, + enqueueStoreCount, + dequeueCount, + dequeueExtractCount, queueLoadFactor, - queueInputRate, - queueStoreRate, - queueOutputRequestRate, - queueOutputRate, + enqueueRate, + enqueueStoreRate, + dequeueRate, + dequeueExtractRate, queueWaitTime, queueTotalWaitTime, - queueInputWaitTime, - queueOutputWaitTime, + enqueueWaitTime, + dequeueWaitTime, -- * Dequeuing and Enqueuing dequeue, dequeueWithOutputPriority, @@ -72,26 +72,26 @@ queueFullChanged_, queueCountChanged, queueCountChanged_, - queueLostCountChanged, - queueLostCountChanged_, - queueInputCountChanged, - queueInputCountChanged_, - queueStoreCountChanged, - queueStoreCountChanged_, - queueOutputRequestCountChanged, - queueOutputRequestCountChanged_, - queueOutputCountChanged, - queueOutputCountChanged_, + enqueueCountChanged, + enqueueCountChanged_, + enqueueLostCountChanged, + enqueueLostCountChanged_, + enqueueStoreCountChanged, + enqueueStoreCountChanged_, + dequeueCountChanged, + dequeueCountChanged_, + dequeueExtractCountChanged, + dequeueExtractCountChanged_, queueLoadFactorChanged, queueLoadFactorChanged_, queueWaitTimeChanged, queueWaitTimeChanged_, queueTotalWaitTimeChanged, queueTotalWaitTimeChanged_, - queueInputWaitTimeChanged, - queueInputWaitTimeChanged_, - queueOutputWaitTimeChanged, - queueOutputWaitTimeChanged_, + enqueueWaitTimeChanged, + enqueueWaitTimeChanged_, + dequeueWaitTimeChanged, + dequeueWaitTimeChanged_, -- * Basic Signals enqueueInitiated, enqueueStored, @@ -143,33 +143,33 @@ type PriorityQueue a = Queue FCFS DLL.DoubleLinkedList StaticPriorities PQ.PriorityQueue FCFS DLL.DoubleLinkedList a --- | Represents the queue using the specified strategies for input @si@, --- internal storing (in memory) @sm@ and output @so@, where @a@ denotes +-- | Represents a queue using the specified strategies for enqueueing (input), @si@, +-- internal storing (in memory), @sm@, and dequeueing (output), @so@, where @a@ denotes -- the type of items stored in the queue. Types @qi@, @qm@ and @qo@ are -- determined automatically and you should not care about them - they -- are dependent types. data Queue si qi sm qm so qo a = Queue { queueMaxCount :: Int, -- ^ The queue capacity. - queueInputStrategy :: si, - -- ^ The strategy applied to the input (enqueuing) process. - queueStoringStrategy :: sm, + enqueueStrategy :: si, + -- ^ The strategy applied to the enqueueing (input) processes when the queue is full. + enqueueStoringStrategy :: sm, -- ^ The strategy applied when storing (in memory) items in the queue. - queueOutputStrategy :: so, - -- ^ The strategy applied to the output (dequeuing) process. - queueInputRes :: Resource si qi, + dequeueStrategy :: so, + -- ^ The strategy applied to the dequeueing (output) processes when the queue is empty. + enqueueRes :: Resource si qi, queueStore :: qm (QueueItem a), - queueOutputRes :: Resource so qo, + dequeueRes :: Resource so qo, queueCountRef :: IORef Int, - queueLostCountRef :: IORef Int, - queueInputCountRef :: IORef Int, - queueStoreCountRef :: IORef Int, - queueOutputRequestCountRef :: IORef Int, - queueOutputCountRef :: IORef Int, + enqueueCountRef :: IORef Int, + enqueueLostCountRef :: IORef Int, + enqueueStoreCountRef :: IORef Int, + dequeueCountRef :: IORef Int, + dequeueExtractCountRef :: IORef Int, queueWaitTimeRef :: IORef (SamplingStats Double), queueTotalWaitTimeRef :: IORef (SamplingStats Double), - queueInputWaitTimeRef :: IORef (SamplingStats Double), - queueOutputWaitTimeRef :: IORef (SamplingStats Double), + enqueueWaitTimeRef :: IORef (SamplingStats Double), + dequeueWaitTimeRef :: IORef (SamplingStats Double), enqueueInitiatedSource :: SignalSource a, enqueueLostSource :: SignalSource a, enqueueStoredSource :: SignalSource a, @@ -209,18 +209,18 @@ QueueStrategy sm qm, QueueStrategy so qo) => si - -- ^ the strategy applied to the input (enqueuing) process + -- ^ the strategy applied to the enqueueing (input) processes when the queue is full -> sm -- ^ the strategy applied when storing items in the queue -> so - -- ^ the strategy applied to the output (dequeuing) process + -- ^ the strategy applied to the dequeueing (output) processes when the queue is empty -> Int -- ^ the queue capacity -> Simulation (Queue si qi sm qm so qo a) newQueue si sm so count = do i <- liftIO $ newIORef 0 - l <- liftIO $ newIORef 0 ci <- liftIO $ newIORef 0 + cl <- liftIO $ newIORef 0 cm <- liftIO $ newIORef 0 cr <- liftIO $ newIORef 0 co <- liftIO $ newIORef 0 @@ -237,22 +237,22 @@ s4 <- newSignalSource s5 <- newSignalSource return Queue { queueMaxCount = count, - queueInputStrategy = si, - queueStoringStrategy = sm, - queueOutputStrategy = so, - queueInputRes = ri, + enqueueStrategy = si, + enqueueStoringStrategy = sm, + dequeueStrategy = so, + enqueueRes = ri, queueStore = qm, - queueOutputRes = ro, + dequeueRes = ro, queueCountRef = i, - queueLostCountRef = l, - queueInputCountRef = ci, - queueStoreCountRef = cm, - queueOutputRequestCountRef = cr, - queueOutputCountRef = co, + enqueueCountRef = ci, + enqueueLostCountRef = cl, + enqueueStoreCountRef = cm, + dequeueCountRef = cr, + dequeueExtractCountRef = co, queueWaitTimeRef = w, queueTotalWaitTimeRef = wt, - queueInputWaitTimeRef = wi, - queueOutputWaitTimeRef = wo, + enqueueWaitTimeRef = wi, + dequeueWaitTimeRef = wo, enqueueInitiatedSource = s1, enqueueLostSource = s2, enqueueStoredSource = s3, @@ -312,92 +312,92 @@ queueCountChanged_ q = mapSignal (const ()) (enqueueStored q) <> mapSignal (const ()) (dequeueExtracted q) - --- | Return the number of lost items. --- --- See also 'queueLostCountChanged' and 'queueLostCountChanged_'. -queueLostCount :: Queue si qi sm qm so qo a -> Event Int -queueLostCount q = - Event $ \p -> readIORef (queueLostCountRef q) - --- | Signal when the 'queueLostCount' property value has changed. -queueLostCountChanged :: Queue si qi sm qm so qo a -> Signal Int -queueLostCountChanged q = - mapSignalM (const $ queueLostCount q) (queueLostCountChanged_ q) - --- | Signal when the 'queueLostCount' property value has changed. -queueLostCountChanged_ :: Queue si qi sm qm so qo a -> Signal () -queueLostCountChanged_ q = - mapSignal (const ()) (enqueueLost q) -- | Return the total number of input items that were enqueued. -- --- See also 'queueInputCountChanged' and 'queueInputCountChanged_'. -queueInputCount :: Queue si qi sm qm so qo a -> Event Int -queueInputCount q = - Event $ \p -> readIORef (queueInputCountRef q) +-- See also 'enqueueCountChanged' and 'enqueueCountChanged_'. +enqueueCount :: Queue si qi sm qm so qo a -> Event Int +enqueueCount q = + Event $ \p -> readIORef (enqueueCountRef q) --- | Signal when the 'queueInputCount' property value has changed. -queueInputCountChanged :: Queue si qi sm qm so qo a -> Signal Int -queueInputCountChanged q = - mapSignalM (const $ queueInputCount q) (queueInputCountChanged_ q) +-- | Signal when the 'enqueueCount' property value has changed. +enqueueCountChanged :: Queue si qi sm qm so qo a -> Signal Int +enqueueCountChanged q = + mapSignalM (const $ enqueueCount q) (enqueueCountChanged_ q) --- | Signal when the 'queueInputCount' property value has changed. -queueInputCountChanged_ :: Queue si qi sm qm so qo a -> Signal () -queueInputCountChanged_ q = +-- | Signal when the 'enqueueCount' property value has changed. +enqueueCountChanged_ :: Queue si qi sm qm so qo a -> Signal () +enqueueCountChanged_ q = mapSignal (const ()) (enqueueInitiated q) + +-- | Return the number of lost items. +-- +-- See also 'enqueueLostCountChanged' and 'enqueueLostCountChanged_'. +enqueueLostCount :: Queue si qi sm qm so qo a -> Event Int +enqueueLostCount q = + Event $ \p -> readIORef (enqueueLostCountRef q) + +-- | Signal when the 'enqueueLostCount' property value has changed. +enqueueLostCountChanged :: Queue si qi sm qm so qo a -> Signal Int +enqueueLostCountChanged q = + mapSignalM (const $ enqueueLostCount q) (enqueueLostCountChanged_ q) + +-- | Signal when the 'enqueueLostCount' property value has changed. +enqueueLostCountChanged_ :: Queue si qi sm qm so qo a -> Signal () +enqueueLostCountChanged_ q = + mapSignal (const ()) (enqueueLost q) -- | Return the total number of input items that were stored. -- --- See also 'queueStoreCountChanged' and 'queueStoreCountChanged_'. -queueStoreCount :: Queue si qi sm qm so qo a -> Event Int -queueStoreCount q = - Event $ \p -> readIORef (queueStoreCountRef q) +-- See also 'enqueueStoreCountChanged' and 'enqueueStoreCountChanged_'. +enqueueStoreCount :: Queue si qi sm qm so qo a -> Event Int +enqueueStoreCount q = + Event $ \p -> readIORef (enqueueStoreCountRef q) --- | Signal when the 'queueStoreCount' property value has changed. -queueStoreCountChanged :: Queue si qi sm qm so qo a -> Signal Int -queueStoreCountChanged q = - mapSignalM (const $ queueStoreCount q) (queueStoreCountChanged_ q) +-- | Signal when the 'enqueueStoreCount' property value has changed. +enqueueStoreCountChanged :: Queue si qi sm qm so qo a -> Signal Int +enqueueStoreCountChanged q = + mapSignalM (const $ enqueueStoreCount q) (enqueueStoreCountChanged_ q) --- | Signal when the 'queueStoreCount' property value has changed. -queueStoreCountChanged_ :: Queue si qi sm qm so qo a -> Signal () -queueStoreCountChanged_ q = +-- | Signal when the 'enqueueStoreCount' property value has changed. +enqueueStoreCountChanged_ :: Queue si qi sm qm so qo a -> Signal () +enqueueStoreCountChanged_ q = mapSignal (const ()) (enqueueStored q) -- | Return the total number of requests for dequeueing the items, --- not taking into account the attempts to dequeue immediately +-- not taking into account the failed attempts to dequeue immediately -- without suspension. -- --- See also 'queueOutputRequestCountChanged' and 'queueOutputRequestCountChanged_'. -queueOutputRequestCount :: Queue si qi sm qm so qo a -> Event Int -queueOutputRequestCount q = - Event $ \p -> readIORef (queueOutputRequestCountRef q) +-- See also 'dequeueCountChanged' and 'dequeueCountChanged_'. +dequeueCount :: Queue si qi sm qm so qo a -> Event Int +dequeueCount q = + Event $ \p -> readIORef (dequeueCountRef q) --- | Signal when the 'queueOutputRequestCount' property value has changed. -queueOutputRequestCountChanged :: Queue si qi sm qm so qo a -> Signal Int -queueOutputRequestCountChanged q = - mapSignalM (const $ queueOutputRequestCount q) (queueOutputRequestCountChanged_ q) +-- | Signal when the 'dequeueCount' property value has changed. +dequeueCountChanged :: Queue si qi sm qm so qo a -> Signal Int +dequeueCountChanged q = + mapSignalM (const $ dequeueCount q) (dequeueCountChanged_ q) --- | Signal when the 'queueOutputRequestCount' property value has changed. -queueOutputRequestCountChanged_ :: Queue si qi sm qm so qo a -> Signal () -queueOutputRequestCountChanged_ q = +-- | Signal when the 'dequeueCount' property value has changed. +dequeueCountChanged_ :: Queue si qi sm qm so qo a -> Signal () +dequeueCountChanged_ q = mapSignal (const ()) (dequeueRequested q) --- | Return the total number of output items that were dequeued. +-- | Return the total number of output items that were actually dequeued. -- --- See also 'queueOutputCountChanged' and 'queueOutputCountChanged_'. -queueOutputCount :: Queue si qi sm qm so qo a -> Event Int -queueOutputCount q = - Event $ \p -> readIORef (queueOutputCountRef q) +-- See also 'dequeueExtractCountChanged' and 'dequeueExtractCountChanged_'. +dequeueExtractCount :: Queue si qi sm qm so qo a -> Event Int +dequeueExtractCount q = + Event $ \p -> readIORef (dequeueExtractCountRef q) --- | Signal when the 'queueOutputCount' property value has changed. -queueOutputCountChanged :: Queue si qi sm qm so qo a -> Signal Int -queueOutputCountChanged q = - mapSignalM (const $ queueOutputCount q) (queueOutputCountChanged_ q) +-- | Signal when the 'dequeueExtractCount' property value has changed. +dequeueExtractCountChanged :: Queue si qi sm qm so qo a -> Signal Int +dequeueExtractCountChanged q = + mapSignalM (const $ dequeueExtractCount q) (dequeueExtractCountChanged_ q) --- | Signal when the 'queueOutputCount' property value has changed. -queueOutputCountChanged_ :: Queue si qi sm qm so qo a -> Signal () -queueOutputCountChanged_ q = +-- | Signal when the 'dequeueExtractCount' property value has changed. +dequeueExtractCountChanged_ :: Queue si qi sm qm so qo a -> Signal () +dequeueExtractCountChanged_ q = mapSignal (const ()) (dequeueExtracted q) -- | Return the load factor: the queue size divided by its maximum size. @@ -423,41 +423,41 @@ -- | Return the rate of the input items that were enqueued: how many items -- per time. -queueInputRate :: Queue si qi sm qm so qo a -> Event Double -queueInputRate q = +enqueueRate :: Queue si qi sm qm so qo a -> Event Double +enqueueRate q = Event $ \p -> - do x <- readIORef (queueInputCountRef q) + do x <- readIORef (enqueueCountRef q) let t0 = spcStartTime $ pointSpecs p t = pointTime p return (fromIntegral x / (t - t0)) -- | Return the rate of the items that were stored: how many items -- per time. -queueStoreRate :: Queue si qi sm qm so qo a -> Event Double -queueStoreRate q = +enqueueStoreRate :: Queue si qi sm qm so qo a -> Event Double +enqueueStoreRate q = Event $ \p -> - do x <- readIORef (queueStoreCountRef q) + do x <- readIORef (enqueueStoreCountRef q) let t0 = spcStartTime $ pointSpecs p t = pointTime p return (fromIntegral x / (t - t0)) -- | Return the rate of the requests for dequeueing the items: how many requests --- per time. It does not include the attempts to dequeue immediately +-- per time. It does not include the failed attempts to dequeue immediately -- without suspension. -queueOutputRequestRate :: Queue si qi sm qm so qo a -> Event Double -queueOutputRequestRate q = +dequeueRate :: Queue si qi sm qm so qo a -> Event Double +dequeueRate q = Event $ \p -> - do x <- readIORef (queueOutputRequestCountRef q) + do x <- readIORef (dequeueCountRef q) let t0 = spcStartTime $ pointSpecs p t = pointTime p return (fromIntegral x / (t - t0)) --- | Return the rate of the output items that were dequeued: how many items +-- | Return the rate of the output items that were actually dequeued: how many items -- per time. -queueOutputRate :: Queue si qi sm qm so qo a -> Event Double -queueOutputRate q = +dequeueExtractRate :: Queue si qi sm qm so qo a -> Event Double +dequeueExtractRate q = Event $ \p -> - do x <- readIORef (queueOutputCountRef q) + do x <- readIORef (dequeueExtractCountRef q) let t0 = spcStartTime $ pointSpecs p t = pointTime p return (fromIntegral x / (t - t0)) @@ -500,40 +500,40 @@ queueTotalWaitTimeChanged_ q = mapSignal (const ()) (dequeueExtracted q) --- | Return the input wait time from the time at which the enqueueing operation +-- | Return the enqueue wait time from the time at which the enqueueing operation -- was initiated to the time at which the item was stored in the queue. -- --- See also 'queueInputWaitTimeChanged' and 'queueInputWaitTimeChanged_'. -queueInputWaitTime :: Queue si qi sm qm so qo a -> Event (SamplingStats Double) -queueInputWaitTime q = - Event $ \p -> readIORef (queueInputWaitTimeRef q) +-- See also 'enqueueWaitTimeChanged' and 'enqueueWaitTimeChanged_'. +enqueueWaitTime :: Queue si qi sm qm so qo a -> Event (SamplingStats Double) +enqueueWaitTime q = + Event $ \p -> readIORef (enqueueWaitTimeRef q) --- | Signal when the 'queueInputWaitTime' property value has changed. -queueInputWaitTimeChanged :: Queue si qi sm qm so qo a -> Signal (SamplingStats Double) -queueInputWaitTimeChanged q = - mapSignalM (const $ queueInputWaitTime q) (queueInputWaitTimeChanged_ q) +-- | Signal when the 'enqueueWaitTime' property value has changed. +enqueueWaitTimeChanged :: Queue si qi sm qm so qo a -> Signal (SamplingStats Double) +enqueueWaitTimeChanged q = + mapSignalM (const $ enqueueWaitTime q) (enqueueWaitTimeChanged_ q) --- | Signal when the 'queueInputWaitTime' property value has changed. -queueInputWaitTimeChanged_ :: Queue si qi sm qm so qo a -> Signal () -queueInputWaitTimeChanged_ q = +-- | Signal when the 'enqueueWaitTime' property value has changed. +enqueueWaitTimeChanged_ :: Queue si qi sm qm so qo a -> Signal () +enqueueWaitTimeChanged_ q = mapSignal (const ()) (enqueueStored q) --- | Return the output wait time from the time at which the item was requested +-- | Return the dequeue wait time from the time at which the item was requested -- for dequeueing to the time at which it was actually dequeued. -- --- See also 'queueOutputWaitTimeChanged' and 'queueOutputWaitTimeChanged_'. -queueOutputWaitTime :: Queue si qi sm qm so qo a -> Event (SamplingStats Double) -queueOutputWaitTime q = - Event $ \p -> readIORef (queueOutputWaitTimeRef q) +-- See also 'dequeueWaitTimeChanged' and 'dequeueWaitTimeChanged_'. +dequeueWaitTime :: Queue si qi sm qm so qo a -> Event (SamplingStats Double) +dequeueWaitTime q = + Event $ \p -> readIORef (dequeueWaitTimeRef q) --- | Signal when the 'queueOutputWaitTime' property value has changed. -queueOutputWaitTimeChanged :: Queue si qi sm qm so qo a -> Signal (SamplingStats Double) -queueOutputWaitTimeChanged q = - mapSignalM (const $ queueOutputWaitTime q) (queueOutputWaitTimeChanged_ q) +-- | Signal when the 'dequeueWaitTime' property value has changed. +dequeueWaitTimeChanged :: Queue si qi sm qm so qo a -> Signal (SamplingStats Double) +dequeueWaitTimeChanged q = + mapSignalM (const $ dequeueWaitTime q) (dequeueWaitTimeChanged_ q) --- | Signal when the 'queueOutputWaitTime' property value has changed. -queueOutputWaitTimeChanged_ :: Queue si qi sm qm so qo a -> Signal () -queueOutputWaitTimeChanged_ q = +-- | Signal when the 'dequeueWaitTime' property value has changed. +dequeueWaitTimeChanged_ :: Queue si qi sm qm so qo a -> Signal () +dequeueWaitTimeChanged_ q = mapSignal (const ()) (dequeueExtracted q) -- | Dequeue suspending the process if the queue is empty. @@ -546,7 +546,7 @@ -- ^ the dequeued value dequeue q = do t <- liftEvent $ dequeueRequest q - requestResource (queueOutputRes q) + requestResource (dequeueRes q) liftEvent $ dequeueExtract q t -- | Dequeue with the output priority suspending the process if the queue is empty. @@ -561,7 +561,7 @@ -- ^ the dequeued value dequeueWithOutputPriority q po = do t <- liftEvent $ dequeueRequest q - requestResourceWithPriority (queueOutputRes q) po + requestResourceWithPriority (dequeueRes q) po liftEvent $ dequeueExtract q t -- | Try to dequeue immediately. @@ -572,7 +572,7 @@ -> Event (Maybe a) -- ^ the dequeued value of 'Nothing' tryDequeue q = - do x <- tryRequestResourceWithinEvent (queueOutputRes q) + do x <- tryRequestResourceWithinEvent (dequeueRes q) if x then do t <- dequeueRequest q fmap Just $ dequeueExtract q t @@ -589,7 +589,7 @@ -> Process () enqueue q a = do i <- liftEvent $ enqueueInitiate q a - requestResource (queueInputRes q) + requestResource (enqueueRes q) liftEvent $ enqueueStore q i -- | Enqueue with the input priority the item suspending the process if the queue is full. @@ -605,7 +605,7 @@ -> Process () enqueueWithInputPriority q pi a = do i <- liftEvent $ enqueueInitiate q a - requestResourceWithPriority (queueInputRes q) pi + requestResourceWithPriority (enqueueRes q) pi liftEvent $ enqueueStore q i -- | Enqueue with the storing priority the item suspending the process if the queue is full. @@ -621,7 +621,7 @@ -> Process () enqueueWithStoringPriority q pm a = do i <- liftEvent $ enqueueInitiate q a - requestResource (queueInputRes q) + requestResource (enqueueRes q) liftEvent $ enqueueStoreWithPriority q pm i -- | Enqueue with the input and storing priorities the item suspending the process if the queue is full. @@ -639,7 +639,7 @@ -> Process () enqueueWithInputStoringPriorities q pi pm a = do i <- liftEvent $ enqueueInitiate q a - requestResourceWithPriority (queueInputRes q) pi + requestResourceWithPriority (enqueueRes q) pi liftEvent $ enqueueStoreWithPriority q pm i -- | Try to enqueue the item. Return 'False' in the monad if the queue is full. @@ -651,7 +651,7 @@ -- ^ the item which we try to enqueue -> Event Bool tryEnqueue q a = - do x <- tryRequestResourceWithinEvent (queueInputRes q) + do x <- tryRequestResourceWithinEvent (enqueueRes q) if x then do enqueueInitiate q a >>= enqueueStore q return True @@ -669,7 +669,7 @@ -- ^ the item which we try to enqueue -> Event Bool tryEnqueueWithStoringPriority q pm a = - do x <- tryRequestResourceWithinEvent (queueInputRes q) + do x <- tryRequestResourceWithinEvent (enqueueRes q) if x then do enqueueInitiate q a >>= enqueueStoreWithPriority q pm return True @@ -685,7 +685,7 @@ -- ^ the item which we try to enqueue -> Event Bool enqueueOrLost q a = - do x <- tryRequestResourceWithinEvent (queueInputRes q) + do x <- tryRequestResourceWithinEvent (enqueueRes q) if x then do enqueueInitiate q a >>= enqueueStore q return True @@ -704,7 +704,7 @@ -- ^ the item which we try to enqueue -> Event Bool enqueueWithStoringPriorityOrLost q pm a = - do x <- tryRequestResourceWithinEvent (queueInputRes q) + do x <- tryRequestResourceWithinEvent (enqueueRes q) if x then do enqueueInitiate q a >>= enqueueStoreWithPriority q pm return True @@ -779,7 +779,7 @@ enqueueInitiate q a = Event $ \p -> do let t = pointTime p - modifyIORef (queueInputCountRef q) (+ 1) + modifyIORef' (enqueueCountRef q) (+ 1) invokeEvent p $ triggerSignal (enqueueInitiatedSource q) a return QueueItem { itemValue = a, @@ -799,13 +799,13 @@ Event $ \p -> do let i' = i { itemStoringTime = pointTime p } -- now we have the actual time of storing invokeEvent p $ - strategyEnqueue (queueStoringStrategy q) (queueStore q) i' - modifyIORef (queueCountRef q) (+ 1) - modifyIORef (queueStoreCountRef q) (+ 1) + strategyEnqueue (enqueueStoringStrategy q) (queueStore q) i' + modifyIORef' (queueCountRef q) (+ 1) + modifyIORef' (enqueueStoreCountRef q) (+ 1) invokeEvent p $ enqueueStat q i' invokeEvent p $ - releaseResourceWithinEvent (queueOutputRes q) + releaseResourceWithinEvent (dequeueRes q) invokeEvent p $ triggerSignal (enqueueStoredSource q) (itemValue i') @@ -823,13 +823,13 @@ Event $ \p -> do let i' = i { itemStoringTime = pointTime p } -- now we have the actual time of storing invokeEvent p $ - strategyEnqueueWithPriority (queueStoringStrategy q) (queueStore q) pm i' - modifyIORef (queueCountRef q) (+ 1) - modifyIORef (queueStoreCountRef q) (+ 1) + strategyEnqueueWithPriority (enqueueStoringStrategy q) (queueStore q) pm i' + modifyIORef' (queueCountRef q) (+ 1) + modifyIORef' (enqueueStoreCountRef q) (+ 1) invokeEvent p $ enqueueStat q i' invokeEvent p $ - releaseResourceWithinEvent (queueOutputRes q) + releaseResourceWithinEvent (dequeueRes q) invokeEvent p $ triggerSignal (enqueueStoredSource q) (itemValue i') @@ -841,7 +841,7 @@ -> Event () enqueueDeny q a = Event $ \p -> - do modifyIORef (queueLostCountRef q) $ (+) 1 + do modifyIORef' (enqueueLostCountRef q) $ (+) 1 invokeEvent p $ triggerSignal (enqueueLostSource q) a @@ -856,7 +856,7 @@ Event $ \p -> do let t0 = itemInputTime i t1 = itemStoringTime i - modifyIORef (queueInputWaitTimeRef q) $ + modifyIORef' (enqueueWaitTimeRef q) $ addSamplingStats (t1 - t0) -- | Accept the dequeuing request and return the current simulation time. @@ -866,7 +866,7 @@ -- ^ the current time dequeueRequest q = Event $ \p -> - do modifyIORef (queueOutputRequestCountRef q) (+ 1) + do modifyIORef' (dequeueCountRef q) (+ 1) invokeEvent p $ triggerSignal (dequeueRequestedSource q) () return $ pointTime p @@ -883,13 +883,13 @@ dequeueExtract q t' = Event $ \p -> do i <- invokeEvent p $ - strategyDequeue (queueStoringStrategy q) (queueStore q) - modifyIORef (queueCountRef q) (+ (- 1)) - modifyIORef (queueOutputCountRef q) (+ 1) + strategyDequeue (enqueueStoringStrategy q) (queueStore q) + modifyIORef' (queueCountRef q) (+ (- 1)) + modifyIORef' (dequeueExtractCountRef q) (+ 1) invokeEvent p $ dequeueStat q t' i invokeEvent p $ - releaseResourceWithinEvent (queueInputRes q) + releaseResourceWithinEvent (enqueueRes q) invokeEvent p $ triggerSignal (dequeueExtractedSource q) (itemValue i) return $ itemValue i @@ -909,11 +909,11 @@ do let t0 = itemInputTime i t1 = itemStoringTime i t = pointTime p - modifyIORef (queueOutputWaitTimeRef q) $ + modifyIORef' (dequeueWaitTimeRef q) $ addSamplingStats (t - t') - modifyIORef (queueTotalWaitTimeRef q) $ + modifyIORef' (queueTotalWaitTimeRef q) $ addSamplingStats (t - t0) - modifyIORef (queueWaitTimeRef q) $ + modifyIORef' (queueWaitTimeRef q) $ addSamplingStats (t - t1) -- | Wait while the queue is full. @@ -942,31 +942,31 @@ -- properties and activities using the specified indent. queueSummary :: (Show si, Show sm, Show so) => Queue si qi sm qm so qo a -> Int -> Event ShowS queueSummary q indent = - do let si = queueInputStrategy q - sm = queueStoringStrategy q - so = queueOutputStrategy q + do let si = enqueueStrategy q + sm = enqueueStoringStrategy q + so = dequeueStrategy q null <- queueNull q full <- queueFull q let maxCount = queueMaxCount q count <- queueCount q - lostCount <- queueLostCount q - inputCount <- queueInputCount q - storeCount <- queueStoreCount q - outputRequestCount <- queueOutputRequestCount q - outputCount <- queueOutputCount q + enqueueCount <- enqueueCount q + enqueueLostCount <- enqueueLostCount q + enqueueStoreCount <- enqueueStoreCount q + dequeueCount <- dequeueCount q + dequeueExtractCount <- dequeueExtractCount q loadFactor <- queueLoadFactor q - inputRate <- queueInputRate q - storeRate <- queueStoreRate q - outputRequestRate <- queueOutputRequestRate q - outputRate <- queueOutputRate q + enqueueRate <- enqueueRate q + enqueueStoreRate <- enqueueStoreRate q + dequeueRate <- dequeueRate q + dequeueExtractRate <- dequeueExtractRate q waitTime <- queueWaitTime q totalWaitTime <- queueTotalWaitTime q - inputWaitTime <- queueInputWaitTime q - outputWaitTime <- queueOutputWaitTime q + enqueueWaitTime <- enqueueWaitTime q + dequeueWaitTime <- dequeueWaitTime q let tab = replicate indent ' ' return $ showString tab . - showString "the input (enqueueing) strategy = " . + showString "the enqueueing (input) strategy = " . shows si . showString "\n" . showString tab . @@ -974,7 +974,7 @@ shows sm . showString "\n" . showString tab . - showString "the output (dequeueing) strategy = " . + showString "the dequeueing (output) strategy = " . shows so . showString "\n" . showString tab . @@ -994,44 +994,44 @@ shows count . showString "\n" . showString tab . - showString "the lost count (number of the lost items) = " . - shows lostCount . + showString "the enqueue count (number of the input items that were enqueued) = " . + shows enqueueCount . showString "\n" . showString tab . - showString "the input count (number of the input items that were enqueued) = " . - shows inputCount . + showString "the enqueue lost count (number of the lost items) = " . + shows enqueueLostCount . showString "\n" . showString tab . - showString "the store count (number of the input items that were stored) = " . - shows storeCount . + showString "the enqueue store count (number of the input items that were stored) = " . + shows enqueueStoreCount . showString "\n" . showString tab . - showString "the output request count (number of requests for dequeueing an item) = " . - shows outputRequestCount . + showString "the dequeue count (number of requests for dequeueing an item) = " . + shows dequeueCount . showString "\n" . showString tab . - showString "the output count (number of the output items that were dequeued) = " . - shows outputCount . + showString "the dequeue extract count (number of the output items that were dequeued) = " . + shows dequeueExtractCount . showString "\n" . showString tab . showString "the load factor (size / max. capacity) = " . shows loadFactor . showString "\n" . showString tab . - showString "the input rate (how many input items were enqueued per time) = " . - shows inputRate . + showString "the enqueue rate (how many input items were enqueued per time) = " . + shows enqueueRate . showString "\n" . showString tab . - showString "the store rate (how many input items were stored per time) = " . - shows storeRate . + showString "the enqueue store rate (how many input items were stored per time) = " . + shows enqueueStoreRate . showString "\n" . showString tab . - showString "the output request rate (how many requests for dequeueing per time) = " . - shows outputRequestRate . + showString "the dequeue rate (how many requests for dequeueing per time) = " . + shows dequeueRate . showString "\n" . showString tab . - showString "the output rate (how many output items were dequeued per time) = " . - shows outputRate . + showString "the dequeue extract rate (how many output items were dequeued per time) = " . + shows dequeueExtractRate . showString "\n" . showString tab . showString "the wait time (when was stored -> when was dequeued) = \n\n" . @@ -1042,9 +1042,9 @@ samplingStatsSummary totalWaitTime (2 + indent) . showString "\n\n" . showString tab . - showString "the input wait time (when the enqueueing was initiated -> when was stored) = \n\n" . - samplingStatsSummary inputWaitTime (2 + indent) . + showString "the enqueue wait time (when the enqueueing was initiated -> when was stored) = \n\n" . + samplingStatsSummary enqueueWaitTime (2 + indent) . showString "\n\n" . showString tab . - showString "the output wait time (when was requested for dequeueing -> when was dequeued) = \n\n" . - samplingStatsSummary outputWaitTime (2 + indent) + showString "the dequeue wait time (when was requested for dequeueing -> when was dequeued) = \n\n" . + samplingStatsSummary dequeueWaitTime (2 + indent) diff --git a/Simulation/Aivika/Queue/Infinite.hs b/Simulation/Aivika/Queue/Infinite.hs --- a/Simulation/Aivika/Queue/Infinite.hs +++ b/Simulation/Aivika/Queue/Infinite.hs @@ -23,18 +23,18 @@ newPriorityQueue, newQueue, -- * Queue Properties and Activities - queueStoringStrategy, - queueOutputStrategy, + enqueueStoringStrategy, + dequeueStrategy, queueNull, queueCount, - queueStoreCount, - queueOutputRequestCount, - queueOutputCount, - queueStoreRate, - queueOutputRequestRate, - queueOutputRate, + enqueueStoreCount, + dequeueCount, + dequeueExtractCount, + enqueueStoreRate, + dequeueRate, + dequeueExtractRate, queueWaitTime, - queueOutputWaitTime, + dequeueWaitTime, -- * Dequeuing and Enqueuing dequeue, dequeueWithOutputPriority, @@ -48,16 +48,16 @@ queueNullChanged_, queueCountChanged, queueCountChanged_, - queueStoreCountChanged, - queueStoreCountChanged_, - queueOutputRequestCountChanged, - queueOutputRequestCountChanged_, - queueOutputCountChanged, - queueOutputCountChanged_, + enqueueStoreCountChanged, + enqueueStoreCountChanged_, + dequeueCountChanged, + dequeueCountChanged_, + dequeueExtractCountChanged, + dequeueExtractCountChanged_, queueWaitTimeChanged, queueWaitTimeChanged_, - queueOutputWaitTimeChanged, - queueOutputWaitTimeChanged_, + dequeueWaitTimeChanged, + dequeueWaitTimeChanged_, -- * Basic Signals enqueueStored, dequeueRequested, @@ -107,24 +107,24 @@ type PriorityQueue a = Queue StaticPriorities PQ.PriorityQueue FCFS DLL.DoubleLinkedList a --- | Represents the infinite queue using the specified strategies for --- internal storing (in memory) @sm@ and output @so@, where @a@ denotes +-- | Represents an infinite queue using the specified strategies for +-- internal storing (in memory), @sm@, and dequeueing (output), @so@, where @a@ denotes -- the type of items stored in the queue. Types @qm@ and @qo@ are -- determined automatically and you should not care about them - they -- are dependent types. data Queue sm qm so qo a = - Queue { queueStoringStrategy :: sm, + Queue { enqueueStoringStrategy :: sm, -- ^ The strategy applied when storing (in memory) items in the queue. - queueOutputStrategy :: so, - -- ^ The strategy applied to the output (dequeuing) process. + dequeueStrategy :: so, + -- ^ The strategy applied to the dequeueing (output) processes. queueStore :: qm (QueueItem a), - queueOutputRes :: Resource so qo, + dequeueRes :: Resource so qo, queueCountRef :: IORef Int, - queueStoreCountRef :: IORef Int, - queueOutputRequestCountRef :: IORef Int, - queueOutputCountRef :: IORef Int, + enqueueStoreCountRef :: IORef Int, + dequeueCountRef :: IORef Int, + dequeueExtractCountRef :: IORef Int, queueWaitTimeRef :: IORef (SamplingStats Double), - queueOutputWaitTimeRef :: IORef (SamplingStats Double), + dequeueWaitTimeRef :: IORef (SamplingStats Double), enqueueStoredSource :: SignalSource a, dequeueRequestedSource :: SignalSource (), dequeueExtractedSource :: SignalSource a } @@ -159,7 +159,7 @@ sm -- ^ the strategy applied when storing items in the queue -> so - -- ^ the strategy applied to the output (dequeuing) process + -- ^ the strategy applied to the dequeueing (output) processes when the queue is empty -> Simulation (Queue sm qm so qo a) newQueue sm so = do i <- liftIO $ newIORef 0 @@ -173,16 +173,16 @@ s3 <- newSignalSource s4 <- newSignalSource s5 <- newSignalSource - return Queue { queueStoringStrategy = sm, - queueOutputStrategy = so, + return Queue { enqueueStoringStrategy = sm, + dequeueStrategy = so, queueStore = qm, - queueOutputRes = ro, + dequeueRes = ro, queueCountRef = i, - queueStoreCountRef = cm, - queueOutputRequestCountRef = cr, - queueOutputCountRef = co, + enqueueStoreCountRef = cm, + dequeueCountRef = cr, + dequeueExtractCountRef = co, queueWaitTimeRef = w, - queueOutputWaitTimeRef = wo, + dequeueWaitTimeRef = wo, enqueueStoredSource = s3, dequeueRequestedSource = s4, dequeueExtractedSource = s5 } @@ -225,84 +225,84 @@ -- | Return the total number of input items that were stored. -- --- See also 'queueStoreCountChanged' and 'queueStoreCountChanged_'. -queueStoreCount :: Queue sm qm so qo a -> Event Int -queueStoreCount q = - Event $ \p -> readIORef (queueStoreCountRef q) +-- See also 'enqueueStoreCountChanged' and 'enqueueStoreCountChanged_'. +enqueueStoreCount :: Queue sm qm so qo a -> Event Int +enqueueStoreCount q = + Event $ \p -> readIORef (enqueueStoreCountRef q) --- | Signal when the 'queueStoreCount' property value has changed. -queueStoreCountChanged :: Queue sm qm so qo a -> Signal Int -queueStoreCountChanged q = - mapSignalM (const $ queueStoreCount q) (queueStoreCountChanged_ q) +-- | Signal when the 'enqueueStoreCount' property value has changed. +enqueueStoreCountChanged :: Queue sm qm so qo a -> Signal Int +enqueueStoreCountChanged q = + mapSignalM (const $ enqueueStoreCount q) (enqueueStoreCountChanged_ q) --- | Signal when the 'queueStoreCount' property value has changed. -queueStoreCountChanged_ :: Queue sm qm so qo a -> Signal () -queueStoreCountChanged_ q = +-- | Signal when the 'enqueueStoreCount' property value has changed. +enqueueStoreCountChanged_ :: Queue sm qm so qo a -> Signal () +enqueueStoreCountChanged_ q = mapSignal (const ()) (enqueueStored q) -- | Return the total number of requests for dequeueing the items, --- not taking into account the attempts to dequeue immediately +-- not taking into account the failed attempts to dequeue immediately -- without suspension. -- --- See also 'queueOutputRequestCountChanged' and 'queueOutputRequestCountChanged_'. -queueOutputRequestCount :: Queue sm qm so qo a -> Event Int -queueOutputRequestCount q = - Event $ \p -> readIORef (queueOutputRequestCountRef q) +-- See also 'dequeueCountChanged' and 'dequeueCountChanged_'. +dequeueCount :: Queue sm qm so qo a -> Event Int +dequeueCount q = + Event $ \p -> readIORef (dequeueCountRef q) --- | Signal when the 'queueOutputRequestCount' property value has changed. -queueOutputRequestCountChanged :: Queue sm qm so qo a -> Signal Int -queueOutputRequestCountChanged q = - mapSignalM (const $ queueOutputRequestCount q) (queueOutputRequestCountChanged_ q) +-- | Signal when the 'dequeueCount' property value has changed. +dequeueCountChanged :: Queue sm qm so qo a -> Signal Int +dequeueCountChanged q = + mapSignalM (const $ dequeueCount q) (dequeueCountChanged_ q) --- | Signal when the 'queueOutputRequestCount' property value has changed. -queueOutputRequestCountChanged_ :: Queue sm qm so qo a -> Signal () -queueOutputRequestCountChanged_ q = +-- | Signal when the 'dequeueCount' property value has changed. +dequeueCountChanged_ :: Queue sm qm so qo a -> Signal () +dequeueCountChanged_ q = mapSignal (const ()) (dequeueRequested q) --- | Return the total number of output items that were dequeued. +-- | Return the total number of output items that were actually dequeued. -- --- See also 'queueOutputCountChanged' and 'queueOutputCountChanged_'. -queueOutputCount :: Queue sm qm so qo a -> Event Int -queueOutputCount q = - Event $ \p -> readIORef (queueOutputCountRef q) +-- See also 'dequeueExtractCountChanged' and 'dequeueExtractCountChanged_'. +dequeueExtractCount :: Queue sm qm so qo a -> Event Int +dequeueExtractCount q = + Event $ \p -> readIORef (dequeueExtractCountRef q) --- | Signal when the 'queueOutputCount' property value has changed. -queueOutputCountChanged :: Queue sm qm so qo a -> Signal Int -queueOutputCountChanged q = - mapSignalM (const $ queueOutputCount q) (queueOutputCountChanged_ q) +-- | Signal when the 'dequeueExtractCount' property value has changed. +dequeueExtractCountChanged :: Queue sm qm so qo a -> Signal Int +dequeueExtractCountChanged q = + mapSignalM (const $ dequeueExtractCount q) (dequeueExtractCountChanged_ q) --- | Signal when the 'queueOutputCount' property value has changed. -queueOutputCountChanged_ :: Queue sm qm so qo a -> Signal () -queueOutputCountChanged_ q = +-- | Signal when the 'dequeueExtractCount' property value has changed. +dequeueExtractCountChanged_ :: Queue sm qm so qo a -> Signal () +dequeueExtractCountChanged_ q = mapSignal (const ()) (dequeueExtracted q) -- | Return the rate of the items that were stored: how many items -- per time. -queueStoreRate :: Queue sm qm so qo a -> Event Double -queueStoreRate q = +enqueueStoreRate :: Queue sm qm so qo a -> Event Double +enqueueStoreRate q = Event $ \p -> - do x <- readIORef (queueStoreCountRef q) + do x <- readIORef (enqueueStoreCountRef q) let t0 = spcStartTime $ pointSpecs p t = pointTime p return (fromIntegral x / (t - t0)) -- | Return the rate of the requests for dequeueing the items: how many requests --- per time. It does not include the attempts to dequeue immediately +-- per time. It does not include the failed attempts to dequeue immediately -- without suspension. -queueOutputRequestRate :: Queue sm qm so qo a -> Event Double -queueOutputRequestRate q = +dequeueRate :: Queue sm qm so qo a -> Event Double +dequeueRate q = Event $ \p -> - do x <- readIORef (queueOutputRequestCountRef q) + do x <- readIORef (dequeueCountRef q) let t0 = spcStartTime $ pointSpecs p t = pointTime p return (fromIntegral x / (t - t0)) -- | Return the rate of the output items that were dequeued: how many items -- per time. -queueOutputRate :: Queue sm qm so qo a -> Event Double -queueOutputRate q = +dequeueExtractRate :: Queue sm qm so qo a -> Event Double +dequeueExtractRate q = Event $ \p -> - do x <- readIORef (queueOutputCountRef q) + do x <- readIORef (dequeueExtractCountRef q) let t0 = spcStartTime $ pointSpecs p t = pointTime p return (fromIntegral x / (t - t0)) @@ -325,22 +325,22 @@ queueWaitTimeChanged_ q = mapSignal (const ()) (dequeueExtracted q) --- | Return the output wait time from the time at which the item was requested +-- | Return the dequeue wait time from the time at which the item was requested -- for dequeueing to the time at which it was actually dequeued. -- --- See also 'queueOutputWaitTimeChanged' and 'queueOutputWaitTimeChanged_'. -queueOutputWaitTime :: Queue sm qm so qo a -> Event (SamplingStats Double) -queueOutputWaitTime q = - Event $ \p -> readIORef (queueOutputWaitTimeRef q) +-- See also 'dequeueWaitTimeChanged' and 'dequeueWaitTimeChanged_'. +dequeueWaitTime :: Queue sm qm so qo a -> Event (SamplingStats Double) +dequeueWaitTime q = + Event $ \p -> readIORef (dequeueWaitTimeRef q) --- | Signal when the 'queueOutputWaitTime' property value has changed. -queueOutputWaitTimeChanged :: Queue sm qm so qo a -> Signal (SamplingStats Double) -queueOutputWaitTimeChanged q = - mapSignalM (const $ queueOutputWaitTime q) (queueOutputWaitTimeChanged_ q) +-- | Signal when the 'dequeueWaitTime' property value has changed. +dequeueWaitTimeChanged :: Queue sm qm so qo a -> Signal (SamplingStats Double) +dequeueWaitTimeChanged q = + mapSignalM (const $ dequeueWaitTime q) (dequeueWaitTimeChanged_ q) --- | Signal when the 'queueOutputWaitTime' property value has changed. -queueOutputWaitTimeChanged_ :: Queue sm qm so qo a -> Signal () -queueOutputWaitTimeChanged_ q = +-- | Signal when the 'dequeueWaitTime' property value has changed. +dequeueWaitTimeChanged_ :: Queue sm qm so qo a -> Signal () +dequeueWaitTimeChanged_ q = mapSignal (const ()) (dequeueExtracted q) -- | Dequeue suspending the process if the queue is empty. @@ -352,7 +352,7 @@ -- ^ the dequeued value dequeue q = do t <- liftEvent $ dequeueRequest q - requestResource (queueOutputRes q) + requestResource (dequeueRes q) liftEvent $ dequeueExtract q t -- | Dequeue with the output priority suspending the process if the queue is empty. @@ -366,7 +366,7 @@ -- ^ the dequeued value dequeueWithOutputPriority q po = do t <- liftEvent $ dequeueRequest q - requestResourceWithPriority (queueOutputRes q) po + requestResourceWithPriority (dequeueRes q) po liftEvent $ dequeueExtract q t -- | Try to dequeue immediately. @@ -376,7 +376,7 @@ -> Event (Maybe a) -- ^ the dequeued value of 'Nothing' tryDequeue q = - do x <- tryRequestResourceWithinEvent (queueOutputRes q) + do x <- tryRequestResourceWithinEvent (dequeueRes q) if x then do t <- dequeueRequest q fmap Just $ dequeueExtract q t @@ -431,11 +431,11 @@ do let i = QueueItem { itemValue = a, itemStoringTime = pointTime p } invokeEvent p $ - strategyEnqueue (queueStoringStrategy q) (queueStore q) i - modifyIORef (queueCountRef q) (+ 1) - modifyIORef (queueStoreCountRef q) (+ 1) + strategyEnqueue (enqueueStoringStrategy q) (queueStore q) i + modifyIORef' (queueCountRef q) (+ 1) + modifyIORef' (enqueueStoreCountRef q) (+ 1) invokeEvent p $ - releaseResourceWithinEvent (queueOutputRes q) + releaseResourceWithinEvent (dequeueRes q) invokeEvent p $ triggerSignal (enqueueStoredSource q) (itemValue i) @@ -454,11 +454,11 @@ do let i = QueueItem { itemValue = a, itemStoringTime = pointTime p } invokeEvent p $ - strategyEnqueueWithPriority (queueStoringStrategy q) (queueStore q) pm i - modifyIORef (queueCountRef q) (+ 1) - modifyIORef (queueStoreCountRef q) (+ 1) + strategyEnqueueWithPriority (enqueueStoringStrategy q) (queueStore q) pm i + modifyIORef' (queueCountRef q) (+ 1) + modifyIORef' (enqueueStoreCountRef q) (+ 1) invokeEvent p $ - releaseResourceWithinEvent (queueOutputRes q) + releaseResourceWithinEvent (dequeueRes q) invokeEvent p $ triggerSignal (enqueueStoredSource q) (itemValue i) @@ -469,7 +469,7 @@ -- ^ the current time dequeueRequest q = Event $ \p -> - do modifyIORef (queueOutputRequestCountRef q) (+ 1) + do modifyIORef' (dequeueCountRef q) (+ 1) invokeEvent p $ triggerSignal (dequeueRequestedSource q) () return $ pointTime p @@ -485,9 +485,9 @@ dequeueExtract q t' = Event $ \p -> do i <- invokeEvent p $ - strategyDequeue (queueStoringStrategy q) (queueStore q) - modifyIORef (queueCountRef q) (+ (- 1)) - modifyIORef (queueOutputCountRef q) (+ 1) + strategyDequeue (enqueueStoringStrategy q) (queueStore q) + modifyIORef' (queueCountRef q) (+ (- 1)) + modifyIORef' (dequeueExtractCountRef q) (+ 1) invokeEvent p $ dequeueStat q t' i invokeEvent p $ @@ -508,9 +508,9 @@ Event $ \p -> do let t1 = itemStoringTime i t = pointTime p - modifyIORef (queueOutputWaitTimeRef q) $ + modifyIORef' (dequeueWaitTimeRef q) $ addSamplingStats (t - t') - modifyIORef (queueWaitTimeRef q) $ + modifyIORef' (queueWaitTimeRef q) $ addSamplingStats (t - t1) -- | Signal whenever any property of the queue changes. @@ -529,18 +529,18 @@ -- properties and activities using the specified indent. queueSummary :: (Show sm, Show so) => Queue sm qm so qo a -> Int -> Event ShowS queueSummary q indent = - do let sm = queueStoringStrategy q - so = queueOutputStrategy q + do let sm = enqueueStoringStrategy q + so = dequeueStrategy q null <- queueNull q count <- queueCount q - storeCount <- queueStoreCount q - outputRequestCount <- queueOutputRequestCount q - outputCount <- queueOutputCount q - storeRate <- queueStoreRate q - outputRequestRate <- queueOutputRequestRate q - outputRate <- queueOutputRate q + enqueueStoreCount <- enqueueStoreCount q + dequeueCount <- dequeueCount q + dequeueExtractCount <- dequeueExtractCount q + enqueueStoreRate <- enqueueStoreRate q + dequeueRate <- dequeueRate q + dequeueExtractRate <- dequeueExtractRate q waitTime <- queueWaitTime q - outputWaitTime <- queueOutputWaitTime q + dequeueWaitTime <- dequeueWaitTime q let tab = replicate indent ' ' return $ showString tab . @@ -548,7 +548,7 @@ shows sm . showString "\n" . showString tab . - showString "the output (dequeueing) strategy = " . + showString "the dequeueing (output) strategy = " . shows so . showString "\n" . showString tab . @@ -560,33 +560,33 @@ shows count . showString "\n" . showString tab . - showString "the store count (number of the input items that were stored) = " . - shows storeCount . + showString "the enqueue store count (number of the input items that were stored) = " . + shows enqueueStoreCount . showString "\n" . showString tab . - showString "the output request count (number of requests for dequeueing an item) = " . - shows outputRequestCount . + showString "the dequeue count (number of requests for dequeueing an item) = " . + shows dequeueCount . showString "\n" . showString tab . - showString "the output count (number of the output items that were dequeued) = " . - shows outputCount . + showString "the dequeue extract count (number of the output items that were dequeued) = " . + shows dequeueExtractCount . showString "\n" . showString tab . - showString "the store rate (how many input items were stored per time) = " . - shows storeRate . + showString "the enqueue store rate (how many input items were stored per time) = " . + shows enqueueStoreRate . showString "\n" . showString tab . - showString "the output request rate (how many requests for dequeueing per time) = " . - shows outputRequestRate . + showString "the dequeue rate (how many requests for dequeueing per time) = " . + shows dequeueRate . showString "\n" . showString tab . - showString "the output rate (how many output items were dequeued per time) = " . - shows outputRate . + showString "the dequeue extract rate (how many output items were dequeued per time) = " . + shows dequeueExtractRate . showString "\n" . showString tab . showString "the wait time (when was stored -> when was dequeued) = \n\n" . samplingStatsSummary waitTime (2 + indent) . showString "\n\n" . showString tab . - showString "the output wait time (when was requested for dequeueing -> when was dequeued) = \n\n" . - samplingStatsSummary outputWaitTime (2 + indent) + showString "the dequeue wait time (when was requested for dequeueing -> when was dequeued) = \n\n" . + samplingStatsSummary dequeueWaitTime (2 + indent) diff --git a/Simulation/Aivika/Server.hs b/Simulation/Aivika/Server.hs --- a/Simulation/Aivika/Server.hs +++ b/Simulation/Aivika/Server.hs @@ -18,38 +18,38 @@ -- * Server Properties and Activities serverInitState, serverState, - serverTotalInputTime, + serverTotalInputWaitTime, serverTotalProcessingTime, - serverTotalOutputTime, - serverInputTime, + serverTotalOutputWaitTime, + serverInputWaitTime, serverProcessingTime, - serverOutputTime, - serverInputTimeFactor, - serverProcessingTimeFactor, - serverOutputTimeFactor, + serverOutputWaitTime, + serverInputWaitFactor, + serverProcessingFactor, + serverOutputWaitFactor, -- * Summary serverSummary, -- * Derived Signals for Properties serverStateChanged, serverStateChanged_, - serverTotalInputTimeChanged, - serverTotalInputTimeChanged_, + serverTotalInputWaitTimeChanged, + serverTotalInputWaitTimeChanged_, serverTotalProcessingTimeChanged, serverTotalProcessingTimeChanged_, - serverTotalOutputTimeChanged, - serverTotalOutputTimeChanged_, - serverInputTimeChanged, - serverInputTimeChanged_, + serverTotalOutputWaitTimeChanged, + serverTotalOutputWaitTimeChanged_, + serverInputWaitTimeChanged, + serverInputWaitTimeChanged_, serverProcessingTimeChanged, serverProcessingTimeChanged_, - serverOutputTimeChanged, - serverOutputTimeChanged_, - serverInputTimeFactorChanged, - serverInputTimeFactorChanged_, - serverProcessingTimeFactorChanged, - serverProcessingTimeFactorChanged_, - serverOutputTimeFactorChanged, - serverOutputTimeFactorChanged_, + serverOutputWaitTimeChanged, + serverOutputWaitTimeChanged_, + serverInputWaitFactorChanged, + serverInputWaitFactorChanged_, + serverProcessingFactorChanged, + serverProcessingFactorChanged_, + serverOutputWaitFactorChanged, + serverOutputWaitFactorChanged_, -- * Basic Signals serverInputReceived, serverTaskProcessed, @@ -59,7 +59,9 @@ import Data.IORef import Data.Monoid + import Control.Monad.Trans +import Control.Arrow import Simulation.Aivika.Simulation import Simulation.Aivika.Dynamics @@ -80,17 +82,17 @@ -- ^ The current state of the server. serverProcess :: (s, a) -> Process (s, b), -- ^ Provide @b@ by specified @a@. - serverTotalInputTimeRef :: IORef Double, + serverTotalInputWaitTimeRef :: IORef Double, -- ^ The counted total time spent in awating the input. serverTotalProcessingTimeRef :: IORef Double, -- ^ The counted total time spent to process the input and prepare the output. - serverTotalOutputTimeRef :: IORef Double, + serverTotalOutputWaitTimeRef :: IORef Double, -- ^ The counted total time spent for delivering the output. - serverInputTimeRef :: IORef (SamplingStats Double), + serverInputWaitTimeRef :: IORef (SamplingStats Double), -- ^ The statistics for the time spent in awaiting the input. serverProcessingTimeRef :: IORef (SamplingStats Double), -- ^ The statistics for the time spent to process the input and prepare the output. - serverOutputTimeRef :: IORef (SamplingStats Double), + serverOutputWaitTimeRef :: IORef (SamplingStats Double), -- ^ The statistics for the time spent for delivering the output. serverInputReceivedSource :: SignalSource a, -- ^ A signal raised when the server recieves a new input to process. @@ -135,12 +137,12 @@ let server = Server { serverInitState = state, serverStateRef = r0, serverProcess = provide, - serverTotalInputTimeRef = r1, + serverTotalInputWaitTimeRef = r1, serverTotalProcessingTimeRef = r2, - serverTotalOutputTimeRef = r3, - serverInputTimeRef = r4, + serverTotalOutputWaitTimeRef = r3, + serverInputWaitTimeRef = r4, serverProcessingTimeRef = r5, - serverOutputTimeRef = r6, + serverOutputWaitTimeRef = r6, serverInputReceivedSource = s1, serverTaskProcessedSource = s2, serverOutputProvidedSource = s3 } @@ -150,7 +152,23 @@ -- -- The processor updates the internal state of the server. The usual case is when -- the processor is applied only once in a chain of data processing. Otherwise; --- every time the processor is used, the state of the server changes. +-- every time the processor is used, the state of the server changes. Sometimes +-- it can be indeed useful if you want to aggregate the statistics for different +-- servers simultaneously, but it would be more preferable to avoid this. +-- +-- If you connect different server processors returned by this function in a chain +-- with help of '>>>' or other category combinator then this chain will act as one +-- whole, where the first server will take a new task only after the last server +-- finishes its current task and requests for the next one from the previous processor +-- in the chain. This is not always that thing you might need. +-- +-- To model a sequence of the server processors working independently, you +-- should separate them with help of the 'prefetchProcessor' that plays a role +-- of a small one-place buffer in that case. +-- +-- The queue processors usually have the prefetching capabilities per se, where +-- the items are already stored in the queue. Therefore, the server processor +-- should not be prefetched if it is connected directly with the queue processor. serverProcessor :: Server s a b -> Processor a b serverProcessor server = Processor $ \xs -> loop (serverInitState server) Nothing xs @@ -163,8 +181,8 @@ Nothing -> return () Just (t', a', b') -> do liftIO $ - do modifyIORef (serverTotalOutputTimeRef server) (+ (t0 - t')) - modifyIORef (serverOutputTimeRef server) $ + do modifyIORef' (serverTotalOutputWaitTimeRef server) (+ (t0 - t')) + modifyIORef' (serverOutputWaitTimeRef server) $ addSamplingStats (t0 - t') triggerSignal (serverOutputProvidedSource server) (a', b') -- get input @@ -172,8 +190,8 @@ t1 <- liftDynamics time liftEvent $ do liftIO $ - do modifyIORef (serverTotalInputTimeRef server) (+ (t1 - t0)) - modifyIORef (serverInputTimeRef server) $ + do modifyIORef' (serverTotalInputWaitTimeRef server) (+ (t1 - t0)) + modifyIORef' (serverInputWaitTimeRef server) $ addSamplingStats (t1 - t0) triggerSignal (serverInputReceivedSource server) a -- provide the service @@ -181,9 +199,9 @@ t2 <- liftDynamics time liftEvent $ do liftIO $ - do writeIORef (serverStateRef server) s' - modifyIORef (serverTotalProcessingTimeRef server) (+ (t2 - t1)) - modifyIORef (serverProcessingTimeRef server) $ + do writeIORef (serverStateRef server) $! s' + modifyIORef' (serverTotalProcessingTimeRef server) (+ (t2 - t1)) + modifyIORef' (serverProcessingTimeRef server) $ addSamplingStats (t2 - t1) triggerSignal (serverTaskProcessedSource server) (a, b) return (b, loop s' (Just $ (t2, a, b)) xs') @@ -205,27 +223,27 @@ serverStateChanged_ server = mapSignal (const ()) (serverTaskProcessed server) --- | Return the counted total time spent by the server in awaiting the input. +-- | Return the counted total time when the server was locked while awaiting the input. -- -- The value returned changes discretely and it is usually delayed relative -- to the current simulation time. -- --- See also 'serverTotalInputTimeChanged' and 'serverTotalInputTimeChanged_'. -serverTotalInputTime :: Server s a b -> Event Double -serverTotalInputTime server = - Event $ \p -> readIORef (serverTotalInputTimeRef server) +-- See also 'serverTotalInputWaitTimeChanged' and 'serverTotalInputWaitTimeChanged_'. +serverTotalInputWaitTime :: Server s a b -> Event Double +serverTotalInputWaitTime server = + Event $ \p -> readIORef (serverTotalInputWaitTimeRef server) --- | Signal when the 'serverTotalInputTime' property value has changed. -serverTotalInputTimeChanged :: Server s a b -> Signal Double -serverTotalInputTimeChanged server = - mapSignalM (const $ serverTotalInputTime server) (serverTotalInputTimeChanged_ server) +-- | Signal when the 'serverTotalInputWaitTime' property value has changed. +serverTotalInputWaitTimeChanged :: Server s a b -> Signal Double +serverTotalInputWaitTimeChanged server = + mapSignalM (const $ serverTotalInputWaitTime server) (serverTotalInputWaitTimeChanged_ server) --- | Signal when the 'serverTotalInputTime' property value has changed. -serverTotalInputTimeChanged_ :: Server s a b -> Signal () -serverTotalInputTimeChanged_ server = +-- | Signal when the 'serverTotalInputWaitTime' property value has changed. +serverTotalInputWaitTimeChanged_ :: Server s a b -> Signal () +serverTotalInputWaitTimeChanged_ server = mapSignal (const ()) (serverInputReceived server) --- | Return the counted total time spent by the server to process all tasks. +-- | Return the counted total time spent by the server while processing the tasks. -- -- The value returned changes discretely and it is usually delayed relative -- to the current simulation time. @@ -245,48 +263,48 @@ serverTotalProcessingTimeChanged_ server = mapSignal (const ()) (serverTaskProcessed server) --- | Return the counted total time when the server was in the lock state trying +-- | Return the counted total time when the server was locked while trying -- to deliver the output. -- -- The value returned changes discretely and it is usually delayed relative -- to the current simulation time. -- --- See also 'serverTotalOutputTimeChanged' and 'serverTotalOutputTimeChanged_'. -serverTotalOutputTime :: Server s a b -> Event Double -serverTotalOutputTime server = - Event $ \p -> readIORef (serverTotalOutputTimeRef server) +-- See also 'serverTotalOutputWaitTimeChanged' and 'serverTotalOutputWaitTimeChanged_'. +serverTotalOutputWaitTime :: Server s a b -> Event Double +serverTotalOutputWaitTime server = + Event $ \p -> readIORef (serverTotalOutputWaitTimeRef server) --- | Signal when the 'serverTotalOutputTime' property value has changed. -serverTotalOutputTimeChanged :: Server s a b -> Signal Double -serverTotalOutputTimeChanged server = - mapSignalM (const $ serverTotalOutputTime server) (serverTotalOutputTimeChanged_ server) +-- | Signal when the 'serverTotalOutputWaitTime' property value has changed. +serverTotalOutputWaitTimeChanged :: Server s a b -> Signal Double +serverTotalOutputWaitTimeChanged server = + mapSignalM (const $ serverTotalOutputWaitTime server) (serverTotalOutputWaitTimeChanged_ server) --- | Signal when the 'serverTotalOutputTime' property value has changed. -serverTotalOutputTimeChanged_ :: Server s a b -> Signal () -serverTotalOutputTimeChanged_ server = +-- | Signal when the 'serverTotalOutputWaitTime' property value has changed. +serverTotalOutputWaitTimeChanged_ :: Server s a b -> Signal () +serverTotalOutputWaitTimeChanged_ server = mapSignal (const ()) (serverOutputProvided server) --- | Return the statistics of the time spent by the server in awaiting the input. +-- | Return the statistics of the time when the server was locked while awaiting the input. -- -- The value returned changes discretely and it is usually delayed relative -- to the current simulation time. -- --- See also 'serverInputTimeChanged' and 'serverInputTimeChanged_'. -serverInputTime :: Server s a b -> Event (SamplingStats Double) -serverInputTime server = - Event $ \p -> readIORef (serverInputTimeRef server) +-- See also 'serverInputWaitTimeChanged' and 'serverInputWaitTimeChanged_'. +serverInputWaitTime :: Server s a b -> Event (SamplingStats Double) +serverInputWaitTime server = + Event $ \p -> readIORef (serverInputWaitTimeRef server) --- | Signal when the 'serverInputTime' property value has changed. -serverInputTimeChanged :: Server s a b -> Signal (SamplingStats Double) -serverInputTimeChanged server = - mapSignalM (const $ serverInputTime server) (serverInputTimeChanged_ server) +-- | Signal when the 'serverInputWaitTime' property value has changed. +serverInputWaitTimeChanged :: Server s a b -> Signal (SamplingStats Double) +serverInputWaitTimeChanged server = + mapSignalM (const $ serverInputWaitTime server) (serverInputWaitTimeChanged_ server) --- | Signal when the 'serverInputTime' property value has changed. -serverInputTimeChanged_ :: Server s a b -> Signal () -serverInputTimeChanged_ server = +-- | Signal when the 'serverInputWaitTime' property value has changed. +serverInputWaitTimeChanged_ :: Server s a b -> Signal () +serverInputWaitTimeChanged_ server = mapSignal (const ()) (serverInputReceived server) --- | Return the statistics of the time spent by the server to process the tasks. +-- | Return the statistics of the time spent by the server while processing the tasks. -- -- The value returned changes discretely and it is usually delayed relative -- to the current simulation time. @@ -306,25 +324,25 @@ serverProcessingTimeChanged_ server = mapSignal (const ()) (serverTaskProcessed server) --- | Return the statistics of the time when the server was in the lock state trying +-- | Return the statistics of the time when the server was locked while trying -- to deliver the output. -- -- The value returned changes discretely and it is usually delayed relative -- to the current simulation time. -- --- See also 'serverOutputTimeChanged' and 'serverOutputTimeChanged_'. -serverOutputTime :: Server s a b -> Event (SamplingStats Double) -serverOutputTime server = - Event $ \p -> readIORef (serverOutputTimeRef server) +-- See also 'serverOutputWaitTimeChanged' and 'serverOutputWaitTimeChanged_'. +serverOutputWaitTime :: Server s a b -> Event (SamplingStats Double) +serverOutputWaitTime server = + Event $ \p -> readIORef (serverOutputWaitTimeRef server) --- | Signal when the 'serverOutputTime' property value has changed. -serverOutputTimeChanged :: Server s a b -> Signal (SamplingStats Double) -serverOutputTimeChanged server = - mapSignalM (const $ serverOutputTime server) (serverOutputTimeChanged_ server) +-- | Signal when the 'serverOutputWaitTime' property value has changed. +serverOutputWaitTimeChanged :: Server s a b -> Signal (SamplingStats Double) +serverOutputWaitTimeChanged server = + mapSignalM (const $ serverOutputWaitTime server) (serverOutputWaitTimeChanged_ server) --- | Signal when the 'serverOutputTime' property value has changed. -serverOutputTimeChanged_ :: Server s a b -> Signal () -serverOutputTimeChanged_ server = +-- | Signal when the 'serverOutputWaitTime' property value has changed. +serverOutputWaitTimeChanged_ :: Server s a b -> Signal () +serverOutputWaitTimeChanged_ server = mapSignal (const ()) (serverOutputProvided server) -- | It returns the factor changing from 0 to 1, which estimates how often @@ -333,29 +351,29 @@ -- This factor is calculated as -- -- @ --- totalInputTime \/ (totalInputTime + totalProcessingTime + totalOutputTime) +-- totalInputWaitTime \/ (totalInputWaitTime + totalProcessingTime + totalOutputWaitTime) -- @ -- -- As before in this module, the value returned changes discretely and -- it is usually delayed relative to the current simulation time. -- --- See also 'serverInputTimeFactorChanged' and 'serverInputTimeFactorChanged_'. -serverInputTimeFactor :: Server s a b -> Event Double -serverInputTimeFactor server = +-- See also 'serverInputWaitFactorChanged' and 'serverInputWaitFactorChanged_'. +serverInputWaitFactor :: Server s a b -> Event Double +serverInputWaitFactor server = Event $ \p -> - do x1 <- readIORef (serverTotalInputTimeRef server) + do x1 <- readIORef (serverTotalInputWaitTimeRef server) x2 <- readIORef (serverTotalProcessingTimeRef server) - x3 <- readIORef (serverTotalOutputTimeRef server) + x3 <- readIORef (serverTotalOutputWaitTimeRef server) return (x1 / (x1 + x2 + x3)) --- | Signal when the 'serverInputTimeFactor' property value has changed. -serverInputTimeFactorChanged :: Server s a b -> Signal Double -serverInputTimeFactorChanged server = - mapSignalM (const $ serverInputTimeFactor server) (serverInputTimeFactorChanged_ server) +-- | Signal when the 'serverInputWaitFactor' property value has changed. +serverInputWaitFactorChanged :: Server s a b -> Signal Double +serverInputWaitFactorChanged server = + mapSignalM (const $ serverInputWaitFactor server) (serverInputWaitFactorChanged_ server) --- | Signal when the 'serverInputTimeFactor' property value has changed. -serverInputTimeFactorChanged_ :: Server s a b -> Signal () -serverInputTimeFactorChanged_ server = +-- | Signal when the 'serverInputWaitFactor' property value has changed. +serverInputWaitFactorChanged_ :: Server s a b -> Signal () +serverInputWaitFactorChanged_ server = mapSignal (const ()) (serverInputReceived server) <> mapSignal (const ()) (serverTaskProcessed server) <> mapSignal (const ()) (serverOutputProvided server) @@ -366,29 +384,29 @@ -- This factor is calculated as -- -- @ --- totalProcessingTime \/ (totalInputTime + totalProcessingTime + totalOutputTime) +-- totalProcessingTime \/ (totalInputWaitTime + totalProcessingTime + totalOutputWaitTime) -- @ -- -- As before in this module, the value returned changes discretely and -- it is usually delayed relative to the current simulation time. -- --- See also 'serverProcessingTimeFactorChanged' and 'serverProcessingTimeFactorChanged_'. -serverProcessingTimeFactor :: Server s a b -> Event Double -serverProcessingTimeFactor server = +-- See also 'serverProcessingFactorChanged' and 'serverProcessingFactorChanged_'. +serverProcessingFactor :: Server s a b -> Event Double +serverProcessingFactor server = Event $ \p -> - do x1 <- readIORef (serverTotalInputTimeRef server) + do x1 <- readIORef (serverTotalInputWaitTimeRef server) x2 <- readIORef (serverTotalProcessingTimeRef server) - x3 <- readIORef (serverTotalOutputTimeRef server) + x3 <- readIORef (serverTotalOutputWaitTimeRef server) return (x2 / (x1 + x2 + x3)) --- | Signal when the 'serverProcessingTimeFactor' property value has changed. -serverProcessingTimeFactorChanged :: Server s a b -> Signal Double -serverProcessingTimeFactorChanged server = - mapSignalM (const $ serverProcessingTimeFactor server) (serverProcessingTimeFactorChanged_ server) +-- | Signal when the 'serverProcessingFactor' property value has changed. +serverProcessingFactorChanged :: Server s a b -> Signal Double +serverProcessingFactorChanged server = + mapSignalM (const $ serverProcessingFactor server) (serverProcessingFactorChanged_ server) --- | Signal when the 'serverProcessingTimeFactor' property value has changed. -serverProcessingTimeFactorChanged_ :: Server s a b -> Signal () -serverProcessingTimeFactorChanged_ server = +-- | Signal when the 'serverProcessingFactor' property value has changed. +serverProcessingFactorChanged_ :: Server s a b -> Signal () +serverProcessingFactorChanged_ server = mapSignal (const ()) (serverInputReceived server) <> mapSignal (const ()) (serverTaskProcessed server) <> mapSignal (const ()) (serverOutputProvided server) @@ -399,29 +417,29 @@ -- This factor is calculated as -- -- @ --- totalOutputTime \/ (totalInputTime + totalProcessingTime + totalOutputTime) +-- totalOutputWaitTime \/ (totalInputWaitTime + totalProcessingTime + totalOutputWaitTime) -- @ -- -- As before in this module, the value returned changes discretely and -- it is usually delayed relative to the current simulation time. -- --- See also 'serverOutputTimeFactorChanged' and 'serverOutputTimeFactorChanged_'. -serverOutputTimeFactor :: Server s a b -> Event Double -serverOutputTimeFactor server = +-- See also 'serverOutputWaitFactorChanged' and 'serverOutputWaitFactorChanged_'. +serverOutputWaitFactor :: Server s a b -> Event Double +serverOutputWaitFactor server = Event $ \p -> - do x1 <- readIORef (serverTotalInputTimeRef server) + do x1 <- readIORef (serverTotalInputWaitTimeRef server) x2 <- readIORef (serverTotalProcessingTimeRef server) - x3 <- readIORef (serverTotalOutputTimeRef server) + x3 <- readIORef (serverTotalOutputWaitTimeRef server) return (x3 / (x1 + x2 + x3)) --- | Signal when the 'serverOutputTimeFactor' property value has changed. -serverOutputTimeFactorChanged :: Server s a b -> Signal Double -serverOutputTimeFactorChanged server = - mapSignalM (const $ serverOutputTimeFactor server) (serverOutputTimeFactorChanged_ server) +-- | Signal when the 'serverOutputWaitFactor' property value has changed. +serverOutputWaitFactorChanged :: Server s a b -> Signal Double +serverOutputWaitFactorChanged server = + mapSignalM (const $ serverOutputWaitFactor server) (serverOutputWaitFactorChanged_ server) --- | Signal when the 'serverOutputTimeFactor' property value has changed. -serverOutputTimeFactorChanged_ :: Server s a b -> Signal () -serverOutputTimeFactorChanged_ server = +-- | Signal when the 'serverOutputWaitFactor' property value has changed. +serverOutputWaitFactorChanged_ :: Server s a b -> Signal () +serverOutputWaitFactorChanged_ server = mapSignal (const ()) (serverInputReceived server) <> mapSignal (const ()) (serverTaskProcessed server) <> mapSignal (const ()) (serverOutputProvided server) @@ -450,37 +468,37 @@ serverSummary :: Server s a b -> Int -> Event ShowS serverSummary server indent = Event $ \p -> - do tx1 <- readIORef (serverTotalInputTimeRef server) + do tx1 <- readIORef (serverTotalInputWaitTimeRef server) tx2 <- readIORef (serverTotalProcessingTimeRef server) - tx3 <- readIORef (serverTotalOutputTimeRef server) + tx3 <- readIORef (serverTotalOutputWaitTimeRef server) let xf1 = tx1 / (tx1 + tx2 + tx3) xf2 = tx2 / (tx1 + tx2 + tx3) xf3 = tx3 / (tx1 + tx2 + tx3) - xs1 <- readIORef (serverInputTimeRef server) + xs1 <- readIORef (serverInputWaitTimeRef server) xs2 <- readIORef (serverProcessingTimeRef server) - xs3 <- readIORef (serverOutputTimeRef server) + xs3 <- readIORef (serverOutputWaitTimeRef server) let tab = replicate indent ' ' return $ showString tab . - showString "total input time (in awaiting the input) = " . shows tx1 . + showString "total input wait time (locked while awaiting the input) = " . shows tx1 . showString "\n" . showString tab . showString "total processing time = " . shows tx2 . showString "\n" . showString tab . - showString "total output time (to deliver the output) = " . shows tx3 . + showString "total output wait time (locked while delivering the output) = " . shows tx3 . showString "\n\n" . showString tab . - showString "input time factor (from 0 to 1) = " . shows xf1 . + showString "input wait factor (from 0 to 1) = " . shows xf1 . showString "\n" . showString tab . - showString "processing time factor (from 0 to 1) = " . shows xf2 . + showString "processing factor (from 0 to 1) = " . shows xf2 . showString "\n" . showString tab . - showString "output time factor (from 0 to 1) = " . shows xf3 . + showString "output wait factor (from 0 to 1) = " . shows xf3 . showString "\n\n" . showString tab . - showString "input time:\n\n" . + showString "input wait time (locked while awaiting the input):\n\n" . samplingStatsSummary xs1 (2 + indent) . showString "\n\n" . showString tab . @@ -488,5 +506,5 @@ samplingStatsSummary xs2 (2 + indent) . showString "\n\n" . showString tab . - showString "output time:\n\n" . + showString "output wait time (locked while delivering the output):\n\n" . samplingStatsSummary xs3 (2 + indent) diff --git a/Simulation/Aivika/Signal.hs b/Simulation/Aivika/Signal.hs --- a/Simulation/Aivika/Signal.hs +++ b/Simulation/Aivika/Signal.hs @@ -40,6 +40,7 @@ SignalHistory, signalHistorySignal, newSignalHistory, + newSignalHistoryStartingWith, readSignalHistory, -- * Signalable Computations Signalable(..), diff --git a/Simulation/Aivika/Statistics/Accumulator.hs b/Simulation/Aivika/Statistics/Accumulator.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Statistics/Accumulator.hs @@ -0,0 +1,44 @@ + +-- | +-- Module : Simulation.Aivika.Statistics.Accumulator +-- Copyright : Copyright (c) 2009-2014, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 7.6.3 +-- +-- This small utility module allows accumulating the timing statistics based on 'Signalable' data +-- such as the queue size or the number of lost items in the queue. +-- + +module Simulation.Aivika.Statistics.Accumulator + (-- * Timing Statistics Accumulator + TimingStatsAccumulator, + newTimingStatsAccumulator, + timingStatsAccumulated) where + +import Simulation.Aivika.Simulation +import Simulation.Aivika.Dynamics +import Simulation.Aivika.Event +import Simulation.Aivika.Ref +import Simulation.Aivika.Statistics +import Simulation.Aivika.Signal + +-- | Represents an accumulator for the timing statistics. +newtype TimingStatsAccumulator a = + TimingStatsAccumulator { timingStatsAccumulatedRef :: Ref (TimingStats a) } + +-- | Return the accumulated statistics. +timingStatsAccumulated :: TimingStatsAccumulator a -> Event (TimingStats a) +timingStatsAccumulated = readRef . timingStatsAccumulatedRef + +-- | Start gathering the timing statistics from the current simulation time. +newTimingStatsAccumulator :: TimingData a => Signalable a -> Event (TimingStatsAccumulator a) +newTimingStatsAccumulator x = + do t0 <- liftDynamics time + a0 <- readSignalable x + r <- liftSimulation $ newRef (returnTimingStats t0 a0) + handleSignal_ (signalableChanged x) $ \a -> + do t <- liftDynamics time + modifyRef r $ addTimingStats t a + return TimingStatsAccumulator { timingStatsAccumulatedRef = r } diff --git a/Simulation/Aivika/Stream.hs b/Simulation/Aivika/Stream.hs --- a/Simulation/Aivika/Stream.hs +++ b/Simulation/Aivika/Stream.hs @@ -25,6 +25,8 @@ splitStreamPrioritising, -- * Specifying Identifier streamUsingId, + -- * Prefetching Stream + prefetchStream, -- * Memoizing, Zipping and Uzipping Stream memoStream, zipStreamSeq, @@ -232,7 +234,7 @@ -- | Replace the 'Left' values. replaceLeftStream :: Stream (Either a b) -> Stream c -> Stream (Either c b) -replaceLeftStream (Cons sab) (ys0 @ (Cons sc)) = Cons z where +replaceLeftStream (Cons sab) (ys0 @ ~(Cons sc)) = Cons z where z = do (a, xs) <- sab case a of Left _ -> @@ -243,7 +245,7 @@ -- | Replace the 'Right' values. replaceRightStream :: Stream (Either a b) -> Stream c -> Stream (Either a c) -replaceRightStream (Cons sab) (ys0 @ (Cons sc)) = Cons z where +replaceRightStream (Cons sab) (ys0 @ ~(Cons sc)) = Cons z where z = do (a, xs) <- sab case a of Right _ -> @@ -332,12 +334,14 @@ concatQueuedStreams s streams = Cons z where z = do reading <- liftSimulation $ newResourceWithMaxCount FCFS 0 (Just 1) writing <- liftSimulation $ newResourceWithMaxCount s 1 (Just 1) + conting <- liftSimulation $ newResourceWithMaxCount FCFS 0 (Just 1) ref <- liftIO $ newIORef Nothing let writer p = do (a, xs) <- runStream p requestResource writing liftIO $ writeIORef ref (Just a) releaseResource reading + requestResource conting writer xs reader = do requestResource reading @@ -346,7 +350,9 @@ releaseResource writing return a forM_ streams $ spawnProcess CancelTogether . writer - runStream $ repeatProcess reader + a <- reader + let xs = repeatProcess (releaseResource conting >> reader) + return (a, xs) -- | Concatenate the input priority streams producing one output stream. concatPriorityStreams :: PriorityQueueStrategy s q p @@ -359,12 +365,14 @@ concatPriorityStreams s streams = Cons z where z = do reading <- liftSimulation $ newResourceWithMaxCount FCFS 0 (Just 1) writing <- liftSimulation $ newResourceWithMaxCount s 1 (Just 1) + conting <- liftSimulation $ newResourceWithMaxCount FCFS 0 (Just 1) ref <- liftIO $ newIORef Nothing let writer p = do ((priority, a), xs) <- runStream p requestResourceWithPriority writing priority liftIO $ writeIORef ref (Just a) releaseResource reading + requestResource conting writer xs reader = do requestResource reading @@ -373,7 +381,9 @@ releaseResource writing return a forM_ streams $ spawnProcess CancelTogether . writer - runStream $ repeatProcess reader + a <- reader + let xs = repeatProcess (releaseResource conting >> reader) + return (a, xs) -- | Merge two streams applying the 'FCFS' strategy for enqueuing the input data. mergeStreams :: Stream a -> Stream a -> Stream a @@ -419,7 +429,7 @@ -- | Consume the stream. It returns a process that infinitely reads data -- from the stream and then redirects them to the provided function. --- It is useful for modeling the process of enqueuing data in the queue +-- It is useful for modeling the process of enqueueing data in the queue -- from the input stream. consumeStream :: (a -> Process ()) -> Stream a -> Process () consumeStream f s = p s where @@ -436,3 +446,29 @@ p (Cons s) = do (a, xs) <- s p xs +-- | Prefetch the input stream requesting for one more data item in advance +-- while the last received item is not yet fully processed in the chain of +-- streams, usually by the processors. +-- +-- You can think of this as the prefetched stream could place its latest +-- data item in some temporary space for later use, which is very useful +-- for modeling a sequence of separate and independent work places. +prefetchStream :: Stream a -> Stream a +prefetchStream s = Cons z where + z = do reading <- liftSimulation $ newResourceWithMaxCount FCFS 0 (Just 1) + writing <- liftSimulation $ newResourceWithMaxCount FCFS 1 (Just 1) + ref <- liftIO $ newIORef Nothing + let writer p = + do (a, xs) <- runStream p + requestResource writing + liftIO $ writeIORef ref (Just a) + releaseResource reading + writer xs + reader = + do requestResource reading + Just a <- liftIO $ readIORef ref + liftIO $ writeIORef ref Nothing + releaseResource writing + return a + spawnProcess CancelTogether $ writer s + runStream $ repeatProcess reader 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 @@ -7,12 +7,14 @@ -- Stability : experimental -- Tested with: GHC 7.6.3 -- --- This module defines random streams of data, which are useful +-- This module defines random streams of events, which are useful -- for describing the input of the model. -- module Simulation.Aivika.Stream.Random - (randomUniformStream, + (-- * Stream of Random Events + randomStream, + randomUniformStream, randomNormalStream, randomExponentialStream, randomErlangStream, @@ -21,69 +23,99 @@ import System.Random +import Control.Monad import Control.Monad.Trans import Simulation.Aivika.Parameter import Simulation.Aivika.Parameter.Random import Simulation.Aivika.Simulation import Simulation.Aivika.Dynamics +import Simulation.Aivika.Event import Simulation.Aivika.Process +import Simulation.Aivika.Processor import Simulation.Aivika.Stream +import Simulation.Aivika.Statistics +import Simulation.Aivika.Ref +import Simulation.Aivika.Arrival +-- | Return a sream of random events that arrive with the specified delay. +randomStream :: Parameter Double -> Stream Arrival +randomStream delay = Cons z0 where + z0 = + do t0 <- liftDynamics time + loop t0 + loop t0 = + do t1 <- liftDynamics time + when (t1 /= t0) $ + error $ + "The time of requesting for a new random event is different from " ++ + "the time when the previous event has arrived. Probably, your model " ++ + "contains a logical error. The random events should be requested permanently. " ++ + "At least, they can be lost, for example, when trying to enqueue them, but " ++ + "the random stream itself must always work: randomStream." + delay <- liftParameter delay + holdProcess delay + t2 <- liftDynamics time + let arrival = Arrival { arrivalTime = t2, + arrivalDelay = delay } + return (arrival, Cons $ loop t2) + -- | Create a new stream with delays distributed uniformly. -randomUniformStream :: Double -- ^ the minimum delay - -> Double -- ^ the maximum delay - -> Stream Double -- ^ the stream of delays -randomUniformStream min max = Cons z where - z = do delay <- liftParameter $ randomUniform min max - holdProcess delay - return (delay, randomUniformStream min max) +randomUniformStream :: Double + -- ^ the minimum delay + -> Double + -- ^ the maximum delay + -> Stream Arrival + -- ^ the stream of random events +randomUniformStream min max = + randomStream $ randomUniform min max -- | Create a new stream with delays distributed normally. -randomNormalStream :: Double -- ^ the mean delay - -> Double -- ^ the delay deviation - -> Stream Double -- ^ the stream of delays -randomNormalStream mu nu = Cons z where - z = do delay <- liftParameter $ randomNormal mu nu - holdProcess delay - return (delay, randomNormalStream mu nu) +randomNormalStream :: Double + -- ^ the mean delay + -> Double + -- ^ the delay deviation + -> Stream Arrival + -- ^ the stream of random events +randomNormalStream mu nu = + randomStream $ randomNormal mu nu -- | Return a new stream with delays distibuted exponentially with the specified mean -- (the reciprocal of the rate). randomExponentialStream :: Double -- ^ the mean delay (the reciprocal of the rate) - -> Stream Double - -- ^ the stream of delays -randomExponentialStream mu = Cons z where - z = do delay <- liftParameter $ randomExponential mu - holdProcess delay - return (delay, randomExponentialStream mu) + -> Stream Arrival + -- ^ the stream of random events +randomExponentialStream mu = + randomStream $ randomExponential mu -- | Return a new stream with delays having the Erlang distribution with the specified -- scale (the reciprocal of the rate) and shape parameters. -randomErlangStream :: Double -- ^ the scale (the reciprocal of the rate) - -> Int -- ^ the shape - -> Stream Double -- ^ the stream of delays -randomErlangStream beta m = Cons z where - z = do delay <- liftParameter $ randomErlang beta m - holdProcess delay - return (delay, randomErlangStream beta m) +randomErlangStream :: Double + -- ^ the scale (the reciprocal of the rate) + -> Int + -- ^ the shape + -> Stream Arrival + -- ^ the stream of random events +randomErlangStream beta m = + randomStream $ randomErlang beta m -- | Return a new stream with delays having the Poisson distribution with -- the specified mean. -randomPoissonStream :: Double -- ^ the mean delay - -> Stream Int -- ^ the stream of delays -randomPoissonStream mu = Cons z where - z = do delay <- liftParameter $ randomPoisson mu - holdProcess $ fromIntegral delay - return (delay, randomPoissonStream mu) +randomPoissonStream :: Double + -- ^ the mean delay + -> Stream Arrival + -- ^ the stream of random events +randomPoissonStream mu = + randomStream $ fmap fromIntegral $ randomPoisson mu -- | Return a new stream with delays having the binomial distribution with the specified -- probability and trials. -randomBinomialStream :: Double -- ^ the probability - -> Int -- ^ the number of trials - -> Stream Int -- ^ the stream of delays -randomBinomialStream prob trials = Cons z where - z = do delay <- liftParameter $ randomBinomial prob trials - holdProcess $ fromIntegral delay - return (delay, randomBinomialStream prob trials) +randomBinomialStream :: Double + -- ^ the probability + -> Int + -- ^ the number of trials + -> Stream Arrival + -- ^ the stream of random events +randomBinomialStream prob trials = + randomStream $ fmap fromIntegral $ randomBinomial prob trials diff --git a/Simulation/Aivika/Task.hs b/Simulation/Aivika/Task.hs --- a/Simulation/Aivika/Task.hs +++ b/Simulation/Aivika/Task.hs @@ -87,7 +87,7 @@ -- | Cancel the task. cancelTask :: Task a -> Event () cancelTask t = - cancelProcessUsingId (taskId t) + cancelProcessWithId (taskId t) -- | Test whether the task was cancelled. taskCancelled :: Task a -> Event Bool diff --git a/Simulation/Aivika/Var.hs b/Simulation/Aivika/Var.hs --- a/Simulation/Aivika/Var.hs +++ b/Simulation/Aivika/Var.hs @@ -61,11 +61,10 @@ -- | Read the value of a variable. -- --- It is safe to run the resulting computation with help of the 'runEvent' --- function using modes 'IncludingCurrentEventsOrFromPast' and --- 'IncludingEarlierEventsOrFromPast', which is necessary if you are going --- to use the variable in the differential or difference equations. Only --- it is preferrable if the variable is not updated twice +-- It is safe to run the resulting computation with help of the 'runEventWith' +-- function using modes 'CurrentEventsOrFromPast' and 'EarlierEventsOrFromPast', +-- which is necessary if you are going to use the variable in the differential +-- or difference equations. Only it is preferrable if the variable is not updated twice -- in the same integration time point; otherwise, different values can be returned -- for the same point. readVar :: Var a -> Event a diff --git a/Simulation/Aivika/Var/Unboxed.hs b/Simulation/Aivika/Var/Unboxed.hs --- a/Simulation/Aivika/Var/Unboxed.hs +++ b/Simulation/Aivika/Var/Unboxed.hs @@ -61,11 +61,10 @@ -- | Read the value of a variable. -- --- It is safe to run the resulting computation with help of the 'runEvent' --- function using modes 'IncludingCurrentEventsOrFromPast' and --- 'IncludingEarlierEventsOrFromPast', which is necessary if you are going --- to use the variable in the differential or difference equations. Only --- it is preferrable if the variable is not updated twice +-- It is safe to run the resulting computation with help of the 'runEventWith' +-- function using modes 'CurrentEventsOrFromPast' and 'EarlierEventsOrFromPast', +-- which is necessary if you are going to use the variable in the differential +-- or difference equations. Only it is preferrable if the variable is not updated twice -- in the same integration time point; otherwise, different values can be returned -- for the same point. readVar :: Unboxed a => Var a -> Event a diff --git a/aivika.cabal b/aivika.cabal --- a/aivika.cabal +++ b/aivika.cabal @@ -1,17 +1,20 @@ name: aivika -version: 1.0 +version: 1.1 synopsis: A multi-paradigm simulation library description: - Aivika is a multi-paradigm simulation library which has - the following features: + Aivika is a multi-paradigm simulation library with a strong emphasis + on the Discrete Event Simulation (DES) in the first order and + System Dynamics (SD) in the second one. . + The library has the following features: + . * allows defining recursive stochastic differential equations of System Dynamics (unordered as in maths via the recursive do-notation); . - * has a basic support of the event-driven paradigm of - the Discrete Event Simulation (DES); + * supports the event-driven paradigm of DES as a basic core for + implementing other paradigms; . - * has a basic support of the process-oriented paradigm of DES + * supports extensively the process-oriented paradigm of DES with an ability to resume, suspend and cancel the discontinuous processes; . @@ -22,18 +25,21 @@ or use the predefined queue strategies); . * allows defining an infinite stream of data based on the - process-oriented computation (designed but not tested in - anyway - please be very careful when using it); + process-oriented computation, where we can define a complex enough + behaviour just in a few lines of code; . * allows defining processors (actually, the Haskell arrows) that - operate on infinite streams of data (designed but not tested - in anyway - please be very careful when using them); + operate on the infinite streams of data, because of which some models + can remind of their high-level graphical representation on the + diagram used by visual simulation software tools; . * supports the activity-oriented paradigm of DES; . * supports the basic constructs for the agent-based modeling; . - * allows creating combined discrete-continuous models; + * allows creating combined discrete-continuous models as all parts + of the library are very well integrated and this is reflected + directly in the type system; . * the arrays of simulation variables are inherently supported (this is mostly a feature of Haskell itself); @@ -48,7 +54,8 @@ * allows gathering statistics in time points; . * hides the technical details in high-level simulation monads - (three of them support the recursive do-notation). + and even one arrow (some of these monads support the recursive + do-notation). . Aivika itself is a light-weight engine with minimal dependencies. However, it has additional packages Aivika Experiment [1] and @@ -70,25 +77,9 @@ All three libraries were tested on Linux, Windows and OS X. . Please read the PDF document An Introduction to - Aivika Simulation Library [3] for more details (a little outdated). - This document is included in the distributive of Aivika but - you can usually find a more recent version by the link provided. - . - P.S. - . - Two items, streams and processors, are not yet tested. This is a - goal for the future version of Aivika. The main reason why I ever uploaded - my three packages is that the Aivika Experiment Chart package - was broken in its dependencies, namely, when using the charting - library. So, I decided to provide the compilable packages again. - . - Although I would like to say that the mentioned streams and processors - will be the main improvement in the future version as they actually - allow defining some DES models on a very high level as you would define - them with help of diagrams. - . - Also the queues and server are not tested carefully. Use at your own - risk. At least, the infinite queue seems to be working. + Aivika Simulation Library [3] for more details, although it is + outdated and contains a very basic description only. The most powerful + features of Aivika are not yet described in this PDF document. . \[1] . @@ -116,8 +107,11 @@ examples/MachRep2.hs examples/MachRep3.hs examples/Furnace.hs + examples/SimpleWorkflow.hs examples/TimeOut.hs examples/TimeOutInt.hs + examples/TimeOutWait.hs + examples/WorkflowLoop.hs examples/README data-files: doc/aivika.pdf @@ -126,6 +120,7 @@ exposed-modules: Simulation.Aivika Simulation.Aivika.Agent + Simulation.Aivika.Arrival Simulation.Aivika.Cont Simulation.Aivika.DoubleLinkedList Simulation.Aivika.Dynamics @@ -152,6 +147,7 @@ Simulation.Aivika.Simulation Simulation.Aivika.Specs Simulation.Aivika.Statistics + Simulation.Aivika.Statistics.Accumulator Simulation.Aivika.Stream Simulation.Aivika.Stream.Random Simulation.Aivika.SystemDynamics diff --git a/examples/BassDiffusion.hs b/examples/BassDiffusion.hs --- a/examples/BassDiffusion.hs +++ b/examples/BassDiffusion.hs @@ -18,11 +18,6 @@ spcMethod = RungeKutta4, spcGeneratorType = SimpleGenerator } -randomTrue :: Double -> Parameter Bool -randomTrue p = - do x <- randomUniform 0 1 - return (x <= p) - data Person = Person { personAgent :: Agent, personPotentialAdopter :: AgentState, personAdopter :: AgentState } @@ -89,10 +84,10 @@ adopters <- newRef 0 ps <- createPersons definePersons ps potentialAdopters adopters - runEventInStartTime IncludingCurrentEvents $ + runEventInStartTime $ activatePersons ps runDynamicsInIntegTimes $ - runEvent IncludingCurrentEvents $ + runEvent $ do i1 <- readRef potentialAdopters i2 <- readRef adopters return [i1, i2] diff --git a/examples/Furnace.hs b/examples/Furnace.hs --- a/examples/Furnace.hs +++ b/examples/Furnace.hs @@ -273,23 +273,23 @@ do furnace <- newFurnace -- initialize the furnace and start its iterating in start time - runEventInStartTime IncludingCurrentEvents $ + runEventInStartTime $ do initializeFurnace furnace startIteratingFurnace furnace -- generate randomly new input ingots - runProcessInStartTime IncludingCurrentEvents $ + runProcessInStartTime $ inputProcess furnace -- load permanently the input ingots in the furnace - runProcessInStartTime IncludingCurrentEvents $ + runProcessInStartTime $ loadingProcess furnace -- run the model in the final time point - runEventInStopTime IncludingCurrentEvents $ + runEventInStopTime $ do -- the ingots - c0 <- queueStoreCount (furnaceQueue furnace) - c1 <- queueOutputCount (furnaceQueue furnace) + c0 <- enqueueStoreCount (furnaceQueue furnace) + c1 <- dequeueCount (furnaceQueue furnace) c2 <- readRef (furnaceReadyCount furnace) liftIO $ do diff --git a/examples/MachRep1.hs b/examples/MachRep1.hs --- a/examples/MachRep1.hs +++ b/examples/MachRep1.hs @@ -46,10 +46,10 @@ holdProcess repairTime machine - runProcessInStartTime IncludingCurrentEvents machine - runProcessInStartTime IncludingCurrentEvents machine + runProcessInStartTime machine + runProcessInStartTime machine - runEventInStopTime IncludingCurrentEvents $ + runEventInStopTime $ do x <- readRef totalUpTime y <- liftParameter stoptime return $ x / (2 * y) diff --git a/examples/MachRep1EventDriven.hs b/examples/MachRep1EventDriven.hs --- a/examples/MachRep1EventDriven.hs +++ b/examples/MachRep1EventDriven.hs @@ -57,13 +57,13 @@ let t = startUpTime + upTime enqueueEvent t $ machineBroken startUpTime - runEventInStartTime IncludingCurrentEvents $ + runEventInStartTime $ do -- start the first machine machineRepaired -- start the second machine machineRepaired - runEventInStopTime IncludingCurrentEvents $ + runEventInStopTime $ do x <- readRef totalUpTime y <- liftParameter stoptime return $ x / (2 * y) diff --git a/examples/MachRep1TimeDriven.hs b/examples/MachRep1TimeDriven.hs --- a/examples/MachRep1TimeDriven.hs +++ b/examples/MachRep1TimeDriven.hs @@ -94,14 +94,14 @@ m2 <- machine -- start the time-driven simulation of the machines - runEventInStartTime IncludingCurrentEvents $ + runEventInStartTime $ -- in the integration time points enqueueEventWithIntegTimes $ do m1 m2 -- return the result in the stop time - runEventInStopTime IncludingCurrentEvents $ + runEventInStopTime $ do x <- readRef totalUpTime y <- liftParameter stoptime return $ x / (2 * y) diff --git a/examples/MachRep2.hs b/examples/MachRep2.hs --- a/examples/MachRep2.hs +++ b/examples/MachRep2.hs @@ -70,10 +70,10 @@ machine - runProcessInStartTime IncludingCurrentEvents machine - runProcessInStartTime IncludingCurrentEvents machine + runProcessInStartTime machine + runProcessInStartTime machine - runEventInStopTime IncludingCurrentEvents $ + runEventInStopTime $ do x <- readRef totalUpTime y <- liftParameter stoptime n <- readRef nRep diff --git a/examples/MachRep3.hs b/examples/MachRep3.hs --- a/examples/MachRep3.hs +++ b/examples/MachRep3.hs @@ -70,13 +70,13 @@ machine pid - runProcessInStartTimeUsingId IncludingCurrentEvents + runProcessInStartTimeUsingId pid1 (machine pid2) - runProcessInStartTimeUsingId IncludingCurrentEvents + runProcessInStartTimeUsingId pid2 (machine pid1) - runEventInStopTime IncludingCurrentEvents $ + runEventInStopTime $ do x <- readRef totalUpTime y <- liftDynamics time return $ x / (2 * y) diff --git a/examples/SimpleWorkflow.hs b/examples/SimpleWorkflow.hs new file mode 100644 --- /dev/null +++ b/examples/SimpleWorkflow.hs @@ -0,0 +1,157 @@ + +-- This is a model of the workflow with originally two work places. Also there are two finite queues. +-- +-- It is described in different sources [1, 2]. So, this is chapter 7 of [2]. +-- +-- [1] { add a foreign source in English } +-- +-- [2] Труб И.И., Объектно-ориентированное моделирование на C++: Учебный курс. - СПб.: Питер, 2006 + +import Prelude hiding (id, (.)) + +import Control.Monad +import Control.Monad.Trans +import Control.Arrow +import Control.Category (id, (.)) + +import Simulation.Aivika +import Simulation.Aivika.Queue + +-- | The simulation specs. +specs = Specs { spcStartTime = 0.0, + spcStopTime = 300.0, + spcDT = 0.1, + spcMethod = RungeKutta4, + spcGeneratorType = SimpleGenerator } + +-- the mean delay of the input arrivals distributed exponentially +meanOrderDelay = 0.4 + +-- the capacity of the queue before the first work places +queueMaxCount1 = 4 + +-- the capacity of the queue before the second work places +queueMaxCount2 = 2 + +-- the mean processing time distributed exponentially in +-- the first work places +meanProcessingTime1 = 0.25 + +-- the mean processing time distributed exponentially in +-- the second work places +meanProcessingTime2 = 0.5 + +-- the number of the first work places +-- (in parallel but the commented code allocates them sequentially) +workplaceCount1 = 1 + +-- the number of the second work places +-- (in parallel but the commented code allocates them sequentially) +workplaceCount2 = 1 + +-- create an accumulator to gather the queue size statistics +newQueueSizeAccumulator queue = + newTimingStatsAccumulator $ + Signalable (queueCount queue) (queueCountChanged_ queue) + +-- create a workflow with the exponential processing time +newWorkplaceExponential meanTime = + newServer $ \a -> + do holdProcess =<< + (liftParameter $ + randomExponential meanTime) + return a + +-- interpose the prefetch processor between two processors +interposePrefetchProcessor x y = + x >>> prefetchProcessor >>> y + +model :: Simulation () +model = do + -- it will gather the statistics of the processing time + arrivalTimer <- newArrivalTimer + -- define a stream of input events + let inputStream = randomExponentialStream meanOrderDelay + -- create a queue before the first work place + queue1 <- newFCFSQueue queueMaxCount1 + -- create a queue before the second work place + queue2 <- newFCFSQueue queueMaxCount2 + -- the first queue size statistics + queueSizeAcc1 <- + runEventInStartTime $ + newQueueSizeAccumulator queue1 + -- the second queue size statistics + queueSizeAcc2 <- + runEventInStartTime $ + newQueueSizeAccumulator queue2 + -- create the first work places, i.e. the "servers" + workplace1s <- forM [1 .. workplaceCount1] $ \_ -> + newWorkplaceExponential meanProcessingTime1 + -- create the second work places, i.e. the "servers" + workplace2s <- forM [1 .. workplaceCount2] $ \_ -> + newWorkplaceExponential meanProcessingTime2 + -- processor for the queue before the first work place + let queueProcessor1 = + queueProcessor + (\a -> liftEvent $ enqueueOrLost_ queue1 a) + (dequeue queue1) + -- processor for the queue before the second work place + let queueProcessor2 = + queueProcessor + (enqueue queue2) + (dequeue queue2) + -- the entire processor from input to output + let entireProcessor = + queueProcessor1 >>> + processorParallel (map serverProcessor workplace1s) >>> + -- foldr1 interposePrefetchProcessor (map serverProcessor workplace1s) >>> + queueProcessor2 >>> + processorParallel (map serverProcessor workplace2s) >>> + -- foldr1 interposePrefetchProcessor (map serverProcessor workplace2s) >>> + arrivalTimerProcessor arrivalTimer + -- start simulating the model + runProcessInStartTime $ + sinkStream $ runProcessor entireProcessor inputStream + -- show the results in the final time + runEventInStopTime $ + do queueSum1 <- queueSummary queue1 2 + queueSum2 <- queueSummary queue2 2 + workplaceSum1s <- forM workplace1s $ \x -> serverSummary x 2 + workplaceSum2s <- forM workplace2s $ \x -> serverSummary x 2 + processingTime <- arrivalProcessingTime arrivalTimer + queueSize1 <- timingStatsAccumulated queueSizeAcc1 + queueSize2 <- timingStatsAccumulated queueSizeAcc2 + liftIO $ + do putStrLn "" + putStrLn "--- the first queue summary (in the final time) ---" + putStrLn "" + putStrLn $ queueSum1 [] + putStrLn "" + forM_ (zip [1..] workplaceSum1s) $ \(i, x) -> + do putStrLn $ "--- the first work place no." ++ show i ++ " (in the final time) ---" + putStrLn "" + putStrLn $ x [] + putStrLn "" + putStrLn "--- the second queue summary (in the final time) ---" + putStrLn "" + putStrLn $ queueSum2 [] + putStrLn "" + forM_ (zip [1..] workplaceSum2s) $ \(i, x) -> + do putStrLn $ "--- the second work place no. " ++ show i ++ " (in the final time) ---" + putStrLn "" + putStrLn $ x [] + putStrLn "" + putStrLn "--- the processing time summary ---" + putStrLn "" + putStrLn $ samplingStatsSummary processingTime 2 [] + putStrLn "" + putStrLn "--- the first queue size summary ---" + putStrLn "" + putStrLn $ timingStatsSummary queueSize1 2 [] + putStrLn "" + putStrLn "--- the second queue size summary ---" + putStrLn "" + putStrLn $ timingStatsSummary queueSize2 2 [] + putStrLn "" + +main = runSimulation model specs diff --git a/examples/TimeOut.hs b/examples/TimeOut.hs --- a/examples/TimeOut.hs +++ b/examples/TimeOut.hs @@ -70,7 +70,7 @@ liftEvent $ do writeRef reactivatedCode 1 reactivateProcess nodePid - cancelProcessUsingId ackPid + cancelProcessWithId ackPid acknowledge :: ProcessId -> Process () acknowledge timeoutPid = @@ -81,12 +81,12 @@ liftEvent $ do writeRef reactivatedCode 2 reactivateProcess nodePid - cancelProcessUsingId timeoutPid + cancelProcessWithId timeoutPid - runProcessInStartTimeUsingId IncludingCurrentEvents + runProcessInStartTimeUsingId nodePid node - runEventInStopTime IncludingCurrentEvents $ + runEventInStopTime $ do x <- readRef nTimeOuts y <- readRef nMsgs return $ x / y diff --git a/examples/TimeOutInt.hs b/examples/TimeOutInt.hs --- a/examples/TimeOutInt.hs +++ b/examples/TimeOutInt.hs @@ -56,7 +56,7 @@ do interrupted <- processInterrupted nodePid if interrupted then modifyRef nTimeOuts $ (+) 1 - else cancelProcessUsingId timeoutPid + else cancelProcessWithId timeoutPid node timeout :: Process () @@ -64,10 +64,10 @@ do holdProcess toPeriod liftEvent $ interruptProcess nodePid - runProcessInStartTimeUsingId IncludingCurrentEvents + runProcessInStartTimeUsingId nodePid node - runEventInStopTime IncludingCurrentEvents $ + runEventInStopTime $ do x <- readRef nTimeOuts y <- readRef nMsgs return $ x / y diff --git a/examples/TimeOutWait.hs b/examples/TimeOutWait.hs new file mode 100644 --- /dev/null +++ b/examples/TimeOutWait.hs @@ -0,0 +1,70 @@ + +-- It corresponds to model TimeOut described in document +-- Advanced Features of the SimPy Language +-- [http://heather.cs.ucdavis.edu/~matloff/156/PLN/AdvancedSimPy.pdf]. +-- SimPy is available on [http://simpy.sourceforge.net/]. +-- +-- The model description is as follows. +-- +-- Introductory example to illustrate the modeling of "competing +-- events" such as timeouts, especially using the timeoutProcess +-- function. A network node starts a process within the specified +-- timeout and receives a signal that notifies whether the process +-- has finished successfully within the timeout; if the node +-- times out, it assumes the message it had sent was lost, and it +-- will send again. The time to get an acknowledgement for a message is +-- exponentially distributed with mean 1.0, and the timeout period is +-- 0.5. Immediately after receiving an acknowledgement, the node sends +-- out a new message. +-- +-- We find the proportion of messages which timeout. The output should +-- be about 0.61. + +import Control.Monad +import Control.Monad.Trans + +import Data.Maybe + +import Simulation.Aivika + +ackRate = 1.0 / 1.0 -- reciprocal of the acknowledge mean time +toPeriod = 0.5 -- timeout period + +specs = Specs { spcStartTime = 0.0, + spcStopTime = 10000.0, + spcDT = 1.0, + spcMethod = RungeKutta4, + spcGeneratorType = SimpleGenerator } + +model :: Simulation Double +model = + do -- number of messages sent + nMsgs <- newRef 0 + + -- number of timeouts which have occured + nTimeOuts <- newRef 0 + + let node :: Process () + node = + do liftEvent $ modifyRef nMsgs $ (+) 1 + result <- + timeoutProcess toPeriod $ + do ackTime <- + liftParameter $ + randomExponential (1 / ackRate) + holdProcess ackTime + liftEvent $ + when (isNothing result) $ + modifyRef nTimeOuts $ (+) 1 + node + + runProcessInStartTime node + + runEventInStopTime $ + do x <- readRef nTimeOuts + y <- readRef nMsgs + return $ x / y + +main = + do putStr "The percentage of timeout was " + runSimulation model specs >>= print diff --git a/examples/WorkflowLoop.hs b/examples/WorkflowLoop.hs new file mode 100644 --- /dev/null +++ b/examples/WorkflowLoop.hs @@ -0,0 +1,191 @@ + +{-# LANGUAGE RecursiveDo, Arrows #-} + +-- This is a model of the workflow with a loop. Also there are two infinite queues. +-- +-- It is described in different sources [1, 2]. So, this is chapter 8 of [2]. +-- +-- [1] { add a foreign source in English } +-- +-- [2] Труб И.И., Объектно-ориентированное моделирование на C++: Учебный курс. - СПб.: Питер, 2006 + +-- CAUTION: this is model is not yet fully tested and it may contain logical errors. + +import Prelude hiding (id, (.)) + +import Control.Monad +import Control.Monad.Trans +import Control.Arrow +import Control.Category (id, (.)) + +import Simulation.Aivika +import Simulation.Aivika.Queue.Infinite + +-- | The simulation specs. +specs = Specs { spcStartTime = 0.0, + spcStopTime = 480.0, + spcDT = 0.1, + spcMethod = RungeKutta4, + spcGeneratorType = SimpleGenerator } + +-- the minimum delay for arriving the next TV set +minArrivalDelay = 3.5 + +-- the maximum delay for arriving the next TV set +maxArrivalDelay = 7.5 + +-- the minimum test time +minTestTime = 6 + +-- the maximum test time +maxTestTime = 12 + +-- the probability of passing the test +testPassingProb = 0.85 + +-- how many testers are there? +testerWorkplaceCount = 2 + +-- the minimum time of tuning the TV set +-- that has not passed the test +minTuningTime = 20 + +-- the maximum time of tuning the TV set +-- that has not passed the test +maxTuningTime = 40 + +-- how many persons perform a tuning of TV sets? +tunerWorkplaceCount = 1 + +-- create an accumulator to gather the queue size statistics +newQueueSizeAccumulator queue = + newTimingStatsAccumulator $ + Signalable (queueCount queue) (queueCountChanged_ queue) + +-- create a tester's workplace +newTesterWorkplace = + newServer $ \a -> + do holdProcess =<< + (liftParameter $ + randomUniform minTestTime maxTestTime) + passed <- + liftParameter $ + randomTrue testPassingProb + if passed + then return $ Right a + else return $ Left a + +-- create a tuner's workplace +newTunerWorkplace = + newServer $ \a -> + do holdProcess =<< + (liftParameter $ + randomUniform minTuningTime maxTuningTime) + return a + +model :: Simulation () +model = mdo + -- it will gather the statistics of the processing time + inputArrivalTimer <- newArrivalTimer + outputArrivalTimer <- newArrivalTimer + -- define a stream of input events + let inputStream = + randomUniformStream minArrivalDelay maxArrivalDelay + -- create a queue before the tester's work place + testerQueue <- newFCFSQueue + -- create a queue before the tuner's work place + tunerQueue <- newFCFSQueue + -- the tester's queue size statistics + testerQueueSizeAcc <- + runEventInStartTime $ + newQueueSizeAccumulator testerQueue + -- the tuner's queue size statistics + tunerQueueSizeAcc <- + runEventInStartTime $ + newQueueSizeAccumulator tunerQueue + -- create the tester's work places, i.e. the "servers" + testerWorkplaces <- + forM [1 .. testerWorkplaceCount] $ \_ -> + newTesterWorkplace + -- create the tuner's work places, i.e. the "servers" + tunerWorkplaces <- + forM [1 .. tunerWorkplaceCount] $ \_ -> + newTunerWorkplace + -- a processor loop for the tester's queue + let testerQueueProcessorLoop = + queueProcessorLoopSeq + (liftEvent . enqueue testerQueue) + (dequeue testerQueue) + testerProcessor + (tunerQueueProcessor >>> tunerProcessor) + -- a processor for the tuner's queue + let tunerQueueProcessor = + queueProcessor + (liftEvent . enqueue tunerQueue) + (dequeue tunerQueue) + -- the parallel work of all the testers + let testerProcessor = + processorParallel (map serverProcessor testerWorkplaces) + -- the parallel work of all the tuners + let tunerProcessor = + processorParallel (map serverProcessor tunerWorkplaces) + -- the entire processor from input to output + let entireProcessor = + arrivalTimerProcessor inputArrivalTimer >>> + testerQueueProcessorLoop >>> + arrivalTimerProcessor outputArrivalTimer + -- start simulating the model + runProcessInStartTime $ + sinkStream $ runProcessor entireProcessor inputStream + -- show the results in the final time + runEventInStopTime $ + do testerQueueSum <- queueSummary testerQueue 2 + tunerQueueSum <- queueSummary tunerQueue 2 + testerWorkplaceSums <- + forM testerWorkplaces $ \x -> serverSummary x 2 + tunerWorkplaceSums <- + forM tunerWorkplaces $ \x -> serverSummary x 2 + inputProcessingTime <- arrivalProcessingTime inputArrivalTimer + outputProcessingTime <- arrivalProcessingTime outputArrivalTimer + testerQueueSize <- timingStatsAccumulated testerQueueSizeAcc + tunerQueueSize <- timingStatsAccumulated tunerQueueSizeAcc + liftIO $ + do putStrLn "" + putStrLn "--- the tester's queue summary (in the final time) ---" + putStrLn "" + putStrLn $ testerQueueSum [] + putStrLn "" + forM_ (zip [1..] testerWorkplaceSums) $ \(i, x) -> + do putStrLn $ "--- the tester's work place no." + ++ show i ++ " (in the final time) ---" + putStrLn "" + putStrLn $ x [] + putStrLn "" + putStrLn "--- the tuner's queue summary (in the final time) ---" + putStrLn "" + putStrLn $ tunerQueueSum [] + putStrLn "" + forM_ (zip [1..] tunerWorkplaceSums) $ \(i, x) -> + do putStrLn $ "--- the tuner's work place no. " + ++ show i ++ " (in the final time) ---" + putStrLn "" + putStrLn $ x [] + putStrLn "" + putStrLn "--- the arrival receiving time summary (we are interested in their count) ---" + putStrLn "" + putStrLn $ samplingStatsSummary inputProcessingTime 2 [] + putStrLn "" + putStrLn "--- the arrival processing time summary ---" + putStrLn "" + putStrLn $ samplingStatsSummary outputProcessingTime 2 [] + putStrLn "" + putStrLn "--- the tester's queue size summary (updated when enqueueing and dequeueing) ---" + putStrLn "" + putStrLn $ timingStatsSummary testerQueueSize 2 [] + putStrLn "" + putStrLn "--- the tuner's queue size summary (updated when enqueueing and dequeueing) ---" + putStrLn "" + putStrLn $ timingStatsSummary tunerQueueSize 2 [] + putStrLn "" + +main = runSimulation model specs