packages feed

aivika-distributed (empty) → 0.1

raw patch · 31 files changed

+3500/−0 lines, 31 filesdep +aivikadep +aivika-transformersdep +basesetup-changed

Dependencies added: aivika, aivika-transformers, base, binary, bytestring, containers, distributed-process, exceptions, mtl, random, stm, time

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2015-2016 David Sorokin <david.sorokin@gmail.com>++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions+are met:++1. Redistributions of source code must retain the above copyright+   notice, this list of conditions and the following disclaimer.++2. Redistributions in binary form must reproduce the above copyright+   notice, this list of conditions and the following disclaimer in the+   documentation and/or other materials provided with the distribution.++3. Neither the name of the author nor the names of his contributors+   may be used to endorse or promote products derived from this software+   without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE AUTHORS ``AS IS'' AND+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE+ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS+OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)+HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT+LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY+OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF+SUCH DAMAGE.
+ Setup.lhs view
@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain
+ Simulation/Aivika/Distributed.hs view
@@ -0,0 +1,17 @@++-- |+-- Module     : Simulation.Aivika.Distributed+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module re-exports the library functionality related to the optimistic strategy+-- of running distributed simulations.+--+module Simulation.Aivika.Distributed+       (-- * Modules+        module Simulation.Aivika.Distributed.Optimistic) where++import Simulation.Aivika.Distributed.Optimistic
+ Simulation/Aivika/Distributed/Optimistic.hs view
@@ -0,0 +1,31 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module re-exports the library functionality related to the optimistic strategy+-- of running distributed simulations.+--+module Simulation.Aivika.Distributed.Optimistic+       (-- * Modules+        module Simulation.Aivika.Distributed.Optimistic.DIO,+        module Simulation.Aivika.Distributed.Optimistic.Event,+        module Simulation.Aivika.Distributed.Optimistic.Generator,+        module Simulation.Aivika.Distributed.Optimistic.Message,+        module Simulation.Aivika.Distributed.Optimistic.QueueStrategy,+        module Simulation.Aivika.Distributed.Optimistic.Priority,+        module Simulation.Aivika.Distributed.Optimistic.Ref.Base,+        module Simulation.Aivika.Distributed.Optimistic.TimeServer) where++import Simulation.Aivika.Distributed.Optimistic.DIO+import Simulation.Aivika.Distributed.Optimistic.Event+import Simulation.Aivika.Distributed.Optimistic.Generator+import Simulation.Aivika.Distributed.Optimistic.Message+import Simulation.Aivika.Distributed.Optimistic.QueueStrategy+import Simulation.Aivika.Distributed.Optimistic.Priority+import Simulation.Aivika.Distributed.Optimistic.Ref.Base+import Simulation.Aivika.Distributed.Optimistic.TimeServer
+ Simulation/Aivika/Distributed/Optimistic/DIO.hs view
@@ -0,0 +1,52 @@++{-# LANGUAGE MultiParamTypeClasses, FlexibleInstances #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.DIO+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines 'DIO' as an instance of the 'MonadDES' type class.+--+module Simulation.Aivika.Distributed.Optimistic.DIO+       (DIO,+        DIOParams(..),+        runDIO,+        defaultDIOParams,+        dioParams,+        messageInboxId,+        timeServerId,+        logDIO,+        terminateDIO,+        unregisterDIO) where++import Control.Monad+import Control.Monad.Trans++import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.DES+import Simulation.Aivika.Trans.Exception+import Simulation.Aivika.Trans.Generator+import Simulation.Aivika.Trans.Event+import Simulation.Aivika.Trans.Process+import Simulation.Aivika.Trans.Ref.Base+import Simulation.Aivika.Trans.QueueStrategy++import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.Event+import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue+import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue+import Simulation.Aivika.Distributed.Optimistic.Generator+import Simulation.Aivika.Distributed.Optimistic.Ref.Base+import Simulation.Aivika.Distributed.Optimistic.QueueStrategy++instance MonadDES DIO++instance MonadComp DIO++instance {-# OVERLAPPING #-} MonadIO (Process DIO) where+  liftIO = liftEvent . liftIO+
+ Simulation/Aivika/Distributed/Optimistic/Event.hs view
@@ -0,0 +1,34 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Event+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines additional functions for the 'Event' computation.+--+module Simulation.Aivika.Distributed.Optimistic.Event+       (syncEvent,+        syncEventInStopTime) where++import Simulation.Aivika.Trans++import Simulation.Aivika.Distributed.Optimistic.Internal.Event+import Simulation.Aivika.Distributed.Optimistic.DIO++-- | Synchronize the simulation in all nodes and call+-- the specified computation in the stop time.+--+-- The modeling time must be initial when calling this function.+-- Also this call must be last in your part of the model.+--+-- It is rather safe to call 'liftIO' within this function.+syncEventInStopTime :: Event DIO () -> Simulation DIO ()+syncEventInStopTime h =+  do t0 <- liftParameter stoptime+     runEventInStartTime $+       syncEvent t0 h+     runEventInStopTime $+       return ()
+ Simulation/Aivika/Distributed/Optimistic/Generator.hs view
@@ -0,0 +1,127 @@++{-# LANGUAGE TypeFamilies #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Generator+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- Here is defined a random number generator,+-- where 'DIO' is an instance of 'MonadGenerator'.+--+module Simulation.Aivika.Distributed.Optimistic.Generator () where++import Control.Monad+import Control.Monad.Trans++import System.Random++import Data.IORef++import Simulation.Aivika.Trans+import Simulation.Aivika.Trans.Generator.Primitive++import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.IO++instance MonadGenerator DIO where++  data Generator DIO =+    Generator { generator01 :: DIO Double,+                -- ^ the generator of uniform numbers from 0 to 1+                generatorNormal01 :: DIO Double+                -- ^ the generator of normal numbers with mean 0 and variance 1+              }++  generateUniform = generateUniform01 . generator01++  generateUniformInt = generateUniformInt01 . generator01++  generateTriangular = generateTriangular01 . generator01++  generateNormal = generateNormal01 . generatorNormal01++  generateLogNormal = generateLogNormal01 . generatorNormal01++  generateExponential = generateExponential01 . generator01++  generateErlang = generateErlang01 . generator01++  generatePoisson = generatePoisson01 . generator01++  generateBinomial = generateBinomial01 . generator01++  generateGamma g = generateGamma01 (generatorNormal01 g) (generator01 g)++  generateBeta g = generateBeta01 (generatorNormal01 g) (generator01 g)++  generateWeibull = generateWeibull01 . generator01++  generateDiscrete = generateDiscrete01 . generator01++  newGenerator tp =+    case tp of+      SimpleGenerator ->+        liftIOUnsafe newStdGen >>= newRandomGenerator+      SimpleGeneratorWithSeed x ->+        error "Unsupported generator type SimpleGeneratorWithSeed: newGenerator"+      CustomGenerator g ->+        g+      CustomGenerator01 g ->+        newRandomGenerator01 g++  newRandomGenerator g = +    do r <- liftIOUnsafe $ newIORef g+       let g01 = do g <- liftIOUnsafe $ readIORef r+                    let (x, g') = random g+                    liftIOUnsafe $ writeIORef r g'+                    return x+       newRandomGenerator01 g01++  newRandomGenerator01 g01 =+    do gNormal01 <- newNormalGenerator01 g01+       return Generator { generator01 = g01,+                          generatorNormal01 = gNormal01 }++-- | Create a normal random number generator with mean 0 and variance 1+-- by the specified generator of uniform random numbers from 0 to 1.+newNormalGenerator01 :: DIO Double+                        -- ^ the generator+                        -> DIO (DIO Double)+newNormalGenerator01 g =+  do nextRef <- liftIOUnsafe $ newIORef 0.0+     flagRef <- liftIOUnsafe $ newIORef False+     xi1Ref  <- liftIOUnsafe $ newIORef 0.0+     xi2Ref  <- liftIOUnsafe $ newIORef 0.0+     psiRef  <- liftIOUnsafe $ newIORef 0.0+     let loop =+           do psi <- liftIOUnsafe $ readIORef psiRef+              if (psi >= 1.0) || (psi == 0.0)+                then do g1 <- g+                        g2 <- g+                        let xi1 = 2.0 * g1 - 1.0+                            xi2 = 2.0 * g2 - 1.0+                            psi = xi1 * xi1 + xi2 * xi2+                        liftIOUnsafe $ writeIORef xi1Ref xi1+                        liftIOUnsafe $ writeIORef xi2Ref xi2+                        liftIOUnsafe $ writeIORef psiRef psi+                        loop+                else liftIOUnsafe $ writeIORef psiRef $ sqrt (- 2.0 * log psi / psi)+     return $+       do flag <- liftIOUnsafe $ readIORef flagRef+          if flag+            then do liftIOUnsafe $ writeIORef flagRef False+                    liftIOUnsafe $ readIORef nextRef+            else do liftIOUnsafe $ writeIORef xi1Ref 0.0+                    liftIOUnsafe $ writeIORef xi2Ref 0.0+                    liftIOUnsafe $ writeIORef psiRef 0.0+                    loop+                    xi1 <- liftIOUnsafe $ readIORef xi1Ref+                    xi2 <- liftIOUnsafe $ readIORef xi2Ref+                    psi <- liftIOUnsafe $ readIORef psiRef+                    liftIOUnsafe $ writeIORef flagRef True+                    liftIOUnsafe $ writeIORef nextRef $ xi2 * psi+                    return $ xi1 * psi
+ Simulation/Aivika/Distributed/Optimistic/Internal/Channel.hs view
@@ -0,0 +1,76 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Channel+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines a channel with fast checking procedure whether the channel is empty.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.Channel+       (Channel,+        newChannel,+        channelEmpty,+        readChannel,+        writeChannel,+        awaitChannel) where++import Data.List+import Data.IORef++import Control.Concurrent.STM+import Control.Monad++-- | A channel.+data Channel a =+  Channel { channelList :: TVar [a],+            channelListEmpty :: TVar Bool,+            channelListEmptyIO :: IORef Bool+          }++-- | Create a new channel.+newChannel :: IO (Channel a)+newChannel =+  do list <- newTVarIO []+     listEmpty <- newTVarIO True+     listEmptyIO <- newIORef True+     return Channel { channelList = list,+                      channelListEmpty = listEmpty,+                      channelListEmptyIO = listEmptyIO }++-- | Test quickly whether the channel is empty.+channelEmpty :: Channel a -> IO Bool+channelEmpty ch =+  readIORef (channelListEmptyIO ch)++-- | Read all data from the channel. +readChannel :: Channel a -> IO [a]+readChannel ch =+  do empty <- readIORef (channelListEmptyIO ch)+     if empty+       then return []+       else do writeIORef (channelListEmptyIO ch) True+               xs <- atomically $+                     do xs <- readTVar (channelList ch)+                        writeTVar (channelList ch) []+                        writeTVar (channelListEmpty ch) True+                        return xs+               return (reverse xs)++-- | Write the value in the channel.+writeChannel :: Channel a -> a -> IO ()+writeChannel ch a =+  do atomically $+       do xs <- readTVar (channelList ch)+          writeTVar (channelList ch) (a : xs)+          writeTVar (channelListEmpty ch) False+     writeIORef (channelListEmptyIO ch) False++-- | Wait for data in the channel.+awaitChannel :: Channel a -> IO ()+awaitChannel ch =+  atomically $+  do empty <- readTVar (channelListEmpty ch)+     when empty retry
+ Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs view
@@ -0,0 +1,216 @@++{-# LANGUAGE DeriveGeneric #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.DIO+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines a distributed computation based on 'IO'.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.DIO+       (DIO(..),+        DIOParams(..),+        invokeDIO,+        runDIO,+        defaultDIOParams,+        terminateDIO,+        unregisterDIO,+        dioParams,+        messageChannel,+        messageInboxId,+        timeServerId,+        logDIO,+        liftDistributedUnsafe) where++import Data.Typeable+import Data.Binary++import GHC.Generics++import Control.Applicative+import Control.Monad+import Control.Monad.Trans+import Control.Exception (throw)+import Control.Monad.Catch as C+import qualified Control.Distributed.Process as DP++import Simulation.Aivika.Trans.Exception+import Simulation.Aivika.Trans.Internal.Types++import Simulation.Aivika.Distributed.Optimistic.Internal.Channel+import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer+import Simulation.Aivika.Distributed.Optimistic.Internal.Priority++-- | The parameters for the 'DIO' computation.+data DIOParams =+  DIOParams { dioLoggingPriority :: Priority,+              -- ^ The logging priority+              dioUndoableLogSizeThreshold :: Int,+              -- ^ The undoable log size threshold used for detecting an overflow+              dioOutputMessageQueueSizeThreshold :: Int,+              -- ^ The output message queue size threshold used for detecting an overflow+              dioSyncTimeout :: Int,+              -- ^ The timeout in microseconds used for synchronising the operations.+              dioAllowPrematureIO :: Bool,+              -- ^ Whether to allow performing the premature IO action; otherwise, raise an error+              dioAllowProcessingOutdatedMessage :: Bool+              -- ^ Whether to allow processing an outdated message with the receive time less than the global time+            } deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary DIOParams++-- | The distributed computation based on 'IO'.+newtype DIO a = DIO { unDIO :: DIOContext -> DP.Process a+                      -- ^ Unwrap the computation.+                    }++-- | The context of the 'DIO' computation.+data DIOContext =+  DIOContext { dioChannel :: Channel LocalProcessMessage,+               -- ^ The channel of messages.+               dioInboxId :: DP.ProcessId,+               -- ^ The inbox process identifier.+               dioTimeServerId :: DP.ProcessId,+               -- ^ The time server process+               dioParams0 :: DIOParams+               -- ^ The parameters of the computation.+             }++instance Monad DIO where++  {-# INLINE return #-}+  return = DIO . const . return++  {-# INLINE (>>=) #-}+  (DIO m) >>= k = DIO $ \ps ->+    m ps >>= \a ->+    let m' = unDIO (k a) in m' ps++instance Applicative DIO where++  {-# INLINE pure #-}+  pure = return++  {-# INLINE (<*>) #-}+  (<*>) = ap++instance Functor DIO where++  {-# INLINE fmap #-}+  fmap f (DIO m) = DIO $ fmap f . m ++instance MonadException DIO where++  catchComp (DIO m) h = DIO $ \ps ->+    C.catch (m ps) (\e -> unDIO (h e) ps)++  finallyComp (DIO m1) (DIO m2) = DIO $ \ps ->+    C.finally (m1 ps) (m2 ps)+  +  throwComp e = DIO $ \ps ->+    throw e++-- | Invoke the 'DIO' computation.+invokeDIO :: DIOContext -> DIO a -> DP.Process a+{-# INLINE invokeDIO #-}+invokeDIO ps (DIO m) = m ps++-- | Lift the distributed 'Process' computation.+liftDistributedUnsafe :: DP.Process a -> DIO a+liftDistributedUnsafe = DIO . const++-- | The default parameters for the 'DIO' computation+defaultDIOParams :: DIOParams+defaultDIOParams =+  DIOParams { dioLoggingPriority = DEBUG,+              dioUndoableLogSizeThreshold = 500000,+              dioOutputMessageQueueSizeThreshold = 10000,+              dioSyncTimeout = 5000000,+              dioAllowPrematureIO = False,+              dioAllowProcessingOutdatedMessage = False+            }++-- | Return the parameters of the current computation.+dioParams :: DIO DIOParams+dioParams = DIO $ return . dioParams0++-- | Return the chanel of messages.+messageChannel :: DIO (Channel LocalProcessMessage)+messageChannel = DIO $ return . dioChannel++-- | Return the process identifier of the inbox that receives messages.+messageInboxId :: DIO DP.ProcessId+messageInboxId = DIO $ return . dioInboxId++-- | Return the time server process identifier.+timeServerId :: DIO DP.ProcessId+timeServerId = DIO $ return . dioTimeServerId++-- | Terminate the simulation including the processes in+-- all nodes connected to the time server.+terminateDIO :: DIO ()+terminateDIO =+  do logDIO INFO "Terminating the simulation..."+     sender   <- messageInboxId+     receiver <- timeServerId+     liftDistributedUnsafe $+       DP.send receiver (TerminateTimeServerMessage sender)++-- | Unregister the simulation process from the time server+-- without affecting the processes in other nodes connected to+-- the corresponding time server.+unregisterDIO :: DIO ()+unregisterDIO =+  do logDIO INFO "Unregistering the simulation process..."+     sender   <- messageInboxId+     receiver <- timeServerId+     liftDistributedUnsafe $+       DP.send receiver (UnregisterLocalProcessMessage sender)++-- | Run the computation using the specified parameters along with time server process+-- identifier and return the inbox process identifier and a new simulation process.+runDIO :: DIO a -> DIOParams -> DP.ProcessId -> DP.Process (DP.ProcessId, DP.Process a)+runDIO m ps serverId =+  do ch <- liftIO newChannel+     inboxId <-+       DP.spawnLocal $+       forever $+       do m <- DP.expect :: DP.Process LocalProcessMessage+          liftIO $+            writeChannel ch m+          when (m == TerminateLocalProcessMessage) $+            do ---+               logProcess ps INFO "Terminating the inbox process..."+               ---+               DP.terminate+     ---+     logProcess ps INFO "Registering the simulation process..."+     ---+     DP.send serverId (RegisterLocalProcessMessage inboxId)+     return (inboxId, unDIO m DIOContext { dioChannel = ch,+                                           dioInboxId = inboxId,+                                           dioTimeServerId = serverId,+                                           dioParams0 = ps })++-- | Log the message with the specified priority.+logDIO :: Priority -> String -> DIO ()+{-# INLINE logDIO #-}+logDIO p message =+  do ps <- dioParams+     when (dioLoggingPriority ps <= p) $+       liftDistributedUnsafe $+       DP.say $+       embracePriority p ++ " " ++ message++-- | Log the message with the specified priority.+logProcess :: DIOParams -> Priority -> String -> DP.Process ()+{-# INLINE logProcess #-}+logProcess ps p message =+  when (dioLoggingPriority ps <= p) $+  DP.say $+  embracePriority p ++ " " ++ message
+ Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs view
@@ -0,0 +1,709 @@++{-# LANGUAGE TypeFamilies, FlexibleInstances #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Event+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- The module defines an event queue.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.Event+       (queueInputMessages,+        queueOutputMessages,+        queueLog,+        syncEvent) where++import Data.Maybe+import Data.IORef+import Data.Typeable+import Data.Time.Clock++import System.Timeout++import Control.Monad+import Control.Monad.Trans+import Control.Exception+import qualified Control.Distributed.Process as DP++import qualified Simulation.Aivika.PriorityQueue.Pure as PQ++import Simulation.Aivika.Trans+import Simulation.Aivika.Trans.Internal.Types++import Simulation.Aivika.Distributed.Optimistic.Internal.Priority+import Simulation.Aivika.Distributed.Optimistic.Internal.Channel+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.IO+import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer+import Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp+import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue+import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue+import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog+import {-# SOURCE #-} qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref as R++-- | Convert microseconds to seconds.+microsecondsToSeconds :: Int -> Rational+microsecondsToSeconds x = (fromInteger $ toInteger x) / 1000000++-- | An implementation of the 'EventQueueing' type class.+instance EventQueueing DIO where++  -- | The event queue type.+  data EventQueue DIO =+    EventQueue { queueInputMessages :: InputMessageQueue,+                 -- ^ the input message queue+                 queueOutputMessages :: OutputMessageQueue,+                 -- ^ the output message queue+                 queueLog :: UndoableLog,+                 -- ^ the undoable log of operations+                 queuePQ :: R.Ref (PQ.PriorityQueue (Point DIO -> DIO ())),+                 -- ^ the underlying priority queue+                 queueBusy :: IORef Bool,+                 -- ^ whether the queue is currently processing events+                 queueTime :: IORef Double,+                 -- ^ the actual time of the event queue+                 queueGlobalTime :: IORef Double,+                 -- ^ the global time+                 queueLocalTime :: IORef Double,+                 -- ^ the long-term queue local time+                 queueLocalTime0 :: IORef Double,+                 -- ^ the short-term queue local time+                 queueLocalTimeTimestamp :: IORef UTCTime,+                 -- ^ the queue local time timestamp+                 queueLocalTimeInterval :: NominalDiffTime+                 -- ^ the queue local time interval for updating+               }++  newEventQueue specs =+    do f <- liftIOUnsafe $ newIORef False+       t <- liftIOUnsafe $ newIORef $ spcStartTime specs+       gt <- liftIOUnsafe $ newIORef $ spcStartTime specs+       pq <- R.newRef0 PQ.emptyQueue+       log <- newUndoableLog+       output <- newOutputMessageQueue+       input <- newInputMessageQueue log rollbackEventPre rollbackEventPost rollbackEventTime+       loct <- liftIOUnsafe $ newIORef $ spcStartTime specs+       loct0 <- liftIOUnsafe $ newIORef $ spcStartTime specs+       loctstamp <- liftIOUnsafe $ getCurrentTime >>= newIORef+       locdt <- fmap (fromRational . microsecondsToSeconds . dioSyncTimeout) dioParams +       return EventQueue { queueInputMessages = input,+                           queueOutputMessages = output,+                           queueLog  = log,+                           queuePQ   = pq,+                           queueBusy = f,+                           queueTime = t,+                           queueGlobalTime = gt,+                           queueLocalTime = loct,+                           queueLocalTime0 = loct0,+                           queueLocalTimeTimestamp = loctstamp,+                           queueLocalTimeInterval = locdt }++  enqueueEvent t (Event m) =+    Event $ \p ->+    let pq = queuePQ $ runEventQueue $ pointRun p+    in invokeEvent p $+       R.modifyRef pq $ \x -> PQ.enqueue x t m++  runEventWith processing (Event e) =+    Dynamics $ \p ->+    do p0 <- invokeEvent p currentEventPoint+       invokeEvent p0 $ enqueueEvent (pointTime p) (return ())+       invokeEvent p $ syncEvents processing+       e p++  eventQueueCount =+    Event $ \p ->+    let pq = queuePQ $ runEventQueue $ pointRun p+    in invokeEvent p $+       fmap PQ.queueCount $ R.readRef pq++-- | The first stage of rolling the changes back.+rollbackEventPre :: Bool -> TimeWarp DIO ()+rollbackEventPre including =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+     rollbackLog (queueLog q) (pointTime p) including++-- | The post stage of rolling the changes back.+rollbackEventPost :: Bool -> TimeWarp DIO ()+rollbackEventPost including =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+     rollbackOutputMessages (queueOutputMessages q) (pointTime p) including++-- | Rollback the event time.+rollbackEventTime :: TimeWarp DIO ()+rollbackEventTime =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+         t = pointTime p+     ---+     --- logDIO DEBUG $+     ---   "Setting the queue time = " ++ show t+     ---+     liftIOUnsafe $+       do writeIORef (queueTime q) t+          modifyIORef' (queueLocalTime q) (min t)+          modifyIORef' (queueLocalTime0 q) (min t)+     t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q)+     when (t0 > t) $+       do ---+          --- logDIO DEBUG $+          ---   "Setting the global time = " ++ show t+          ---+          liftIOUnsafe $ writeIORef (queueGlobalTime q) t+          invokeEvent p sendLocalTime++-- | Return the current event time.+currentEventTime :: Event DIO Double+{-# INLINE currentEventTime #-}+currentEventTime =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     liftIOUnsafe $ readIORef (queueTime q)++-- | Return the current event point.+currentEventPoint :: Event DIO (Point DIO)+{-# INLINE currentEventPoint #-}+currentEventPoint =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     t' <- liftIOUnsafe $ readIORef (queueTime q)+     if t' == pointTime p+       then return p+       else let sc = pointSpecs p+                t0 = spcStartTime sc+                dt = spcDT sc+                n' = fromIntegral $ floor ((t' - t0) / dt)+            in return p { pointTime = t',+                          pointIteration = n',+                          pointPhase = -1 }++-- | Process the pending events.+processPendingEventsCore :: Bool -> Dynamics DIO ()+processPendingEventsCore includingCurrentEvents = Dynamics r where+  r p =+    do let q = runEventQueue $ pointRun p+           f = queueBusy q+       f' <- liftIOUnsafe $ readIORef f+       if f'+         then error $+              "Detected an event loop, which may indicate to " +++              "a logical error in the model: processPendingEventsCore"+         else do liftIOUnsafe $ writeIORef f True+                 call q p p+                 liftIOUnsafe $ writeIORef f False+  call q p p0 =+    do let pq = queuePQ q+           r  = pointRun p+       -- process external messages+       p1 <- invokeEvent p0 currentEventPoint+       ok <- invokeEvent p1 $ runTimeWarp processChannelMessages+       if not ok+         then call q p p1+         else do -- proceed with processing the events+                 f <- invokeEvent p1 $ fmap PQ.queueNull $ R.readRef pq+                 unless f $+                   do (t2, c2) <- invokeEvent p1 $ fmap PQ.queueFront $ R.readRef pq+                      let t = queueTime q+                      t' <- liftIOUnsafe $ readIORef t+                      when (t2 < t') $ +                        -- error "The time value is too small: processPendingEventsCore"+                        error $+                        "The time value is too small (" ++ show t2 +++                        " < " ++ show t' ++ "): processPendingEventsCore"+                      when ((t2 < pointTime p) ||+                            (includingCurrentEvents && (t2 == pointTime p))) $+                        do let sc = pointSpecs p+                               t0 = spcStartTime sc+                               dt = spcDT sc+                               n2 = fromIntegral $ floor ((t2 - t0) / dt)+                               p2 = p { pointTime = t2,+                                        pointIteration = n2,+                                        pointPhase = -1 }+                           ---+                           --- ps <- dioParams+                           --- when (dioLoggingPriority ps <= DEBUG) $+                           ---   invokeEvent p2 $+                           ---   writeLog (queueLog q) $+                           ---   logDIO DEBUG $+                           ---   "Reverting the queue time " ++ show t2 ++ " --> " ++ show t'+                           ---+                           liftIOUnsafe $ writeIORef t t2+                           invokeEvent p2 $ R.modifyRef pq PQ.dequeue+                           catchComp+                             (c2 p2)+                             (\e@(SimulationRetry _) -> invokeEvent p2 $ handleEventRetry e) +                           call q p p2++-- | Process the pending events synchronously, i.e. without past.+processPendingEvents :: Bool -> Dynamics DIO ()+processPendingEvents includingCurrentEvents = Dynamics r where+  r p =+    do let q = runEventQueue $ pointRun p+           t = queueTime q+       t' <- liftIOUnsafe $ readIORef t+       if pointTime p < t'+         then error $+              "The current time is less than " +++              "the time in the queue: processPendingEvents"+         else invokeDynamics p m+  m = processPendingEventsCore includingCurrentEvents++-- | A memoized value.+processEventsIncludingCurrent :: Dynamics DIO ()+processEventsIncludingCurrent = processPendingEvents True++-- | A memoized value.+processEventsIncludingEarlier :: Dynamics DIO ()+processEventsIncludingEarlier = processPendingEvents False++-- | A memoized value.+processEventsIncludingCurrentCore :: Dynamics DIO ()+processEventsIncludingCurrentCore = processPendingEventsCore True++-- | A memoized value.+processEventsIncludingEarlierCore :: Dynamics DIO ()+processEventsIncludingEarlierCore = processPendingEventsCore True++-- | Process the events.+processEvents :: EventProcessing -> Dynamics DIO ()+processEvents CurrentEvents = processEventsIncludingCurrent+processEvents EarlierEvents = processEventsIncludingEarlier+processEvents CurrentEventsOrFromPast = processEventsIncludingCurrentCore+processEvents EarlierEventsOrFromPast = processEventsIncludingEarlierCore++-- | Whether there is an overflow.+isEventOverflow :: Event DIO Bool+isEventOverflow =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     n1 <- liftIOUnsafe $ logSize (queueLog q)+     n2 <- liftIOUnsafe $ outputMessageQueueSize (queueOutputMessages q)+     ps <- dioParams+     let th1 = dioUndoableLogSizeThreshold ps+         th2 = dioOutputMessageQueueSizeThreshold ps+     if (n1 >= th1) || (n2 >= th2)+       then do logDIO NOTICE $+                 "t = " ++ (show $ pointTime p) +++                 ": detected the event overflow"+               return True+       else return False++-- | Throttle the message channel.+throttleMessageChannel :: TimeWarp DIO ()+throttleMessageChannel =+  TimeWarp $ \p ->+  do invokeEvent p updateLocalTime+     invokeEvent p sendLocalTime+     ch <- messageChannel+     dt <- fmap dioSyncTimeout dioParams+     liftIOUnsafe $+       timeout dt $ awaitChannel ch+     invokeTimeWarp p $ processChannelMessages++-- | Process the channel messages.+processChannelMessages :: TimeWarp DIO ()+processChannelMessages =+  TimeWarp $ \p ->+  do ch <- messageChannel+     f  <- liftIOUnsafe $ channelEmpty ch+     unless f $+       do xs <- liftIOUnsafe $ readChannel ch+          forM_ xs $ \x ->+            do p' <- invokeEvent p currentEventPoint+               invokeTimeWarp p' $ processChannelMessage x+     p' <- invokeEvent p currentEventPoint+     f2 <- invokeEvent p' isEventOverflow+     when f2 $+       invokeTimeWarp p' throttleMessageChannel++-- | Process the channel message.+processChannelMessage :: LocalProcessMessage -> TimeWarp DIO ()+processChannelMessage x@(QueueMessage m) =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+     ---+     --- invokeEvent p $+     ---   logMessage x+     ---+     t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q)+     when (messageReceiveTime m < t0) $+       do f <- fmap dioAllowProcessingOutdatedMessage dioParams+          if f+            then invokeEvent p logOutdatedMessage+            else error "Received the outdated message: processChannelMessage"+     invokeTimeWarp p $+       enqueueMessage (queueInputMessages q) m+processChannelMessage x@(QueueMessageBulk ms) =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+     ---+     --- invokeEvent p $+     ---   logMessage x+     ---+     t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q)+     forM_ ms $ \m ->+       do when (messageReceiveTime m < t0) $+            do f <- fmap dioAllowProcessingOutdatedMessage dioParams+               if f+                 then invokeEvent p logOutdatedMessage+                 else error "Received the outdated message: processChannelMessage"+          invokeTimeWarp p $+            enqueueMessage (queueInputMessages q) m+processChannelMessage x@(GlobalTimeMessage globalTime) =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+     ---+     --- invokeEvent p $+     ---   logMessage x+     ---+     case globalTime of+       Nothing -> return ()+       Just t0 ->+         invokeEvent p $+         updateGlobalTime t0+     invokeEvent p updateLocalTime+     t <- invokeEvent p getLocalTime+     sender   <- messageInboxId+     receiver <- timeServerId+     liftDistributedUnsafe $+       DP.send receiver (GlobalTimeMessageResp sender t)+processChannelMessage x@(LocalTimeMessageResp globalTime) =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+     ---+     --- invokeEvent p $+     ---   logMessage x+     ---+     invokeEvent p $+       updateGlobalTime globalTime+processChannelMessage x@TerminateLocalProcessMessage =+  TimeWarp $ \p ->+  do ---+     --- invokeEvent p $+     ---   logMessage x+     ---+     liftDistributedUnsafe $+       DP.terminate++-- | Update the global time.+updateGlobalTime :: Double -> Event DIO ()+updateGlobalTime t =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     invokeEvent p updateLocalTime+     t' <- invokeEvent p getLocalTime+     if t > t'+       then logDIO WARNING $+            "t = " ++ show t' +++            ": Ignored the global time that is greater than the current local time"+       else do liftIOUnsafe $+                 writeIORef (queueGlobalTime q) t+               invokeEvent p $+                 reduceEvents t++-- | Show the message.+showMessage :: Message -> ShowS+showMessage m =+  showString "{ " .+  showString "sendTime = " .+  shows (messageSendTime m) .+  showString ", receiveTime = " .+  shows (messageReceiveTime m) .+  (if messageAntiToggle m+   then showString ", antiToggle = True"+   else showString "") .+  showString " }"++-- | Log the message at the specified time.+logMessage :: LocalProcessMessage -> Event DIO ()+logMessage (QueueMessage m) =+  Event $ \p ->+  logDIO INFO $+  "t = " ++ (show $ pointTime p) +++  ": QueueMessage " +++  showMessage m []+logMessage (QueueMessageBulk ms) =+  Event $ \p ->+  logDIO INFO $+  "t = " ++ (show $ pointTime p) +++  ": QueueMessageBulk [ " +++  let fs = foldl1 (\a b -> a . showString ", " . b) $ map showMessage ms+  in fs [] ++ " ]" +logMessage m =+  Event $ \p ->+  logDIO DEBUG $+  "t = " ++ (show $ pointTime p) +++  ": " ++ show m++-- | Log that the local time is to be synchronized.+logSyncLocalTime :: Event DIO ()+logSyncLocalTime =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)+     logDIO DEBUG $+       "t = " ++ (show $ pointTime p) +++       ", global t = " ++ (show t') +++       ": synchronizing the local time..."++-- | Log that the local time is to be synchronized in ring 0.+logSyncLocalTime0 :: Event DIO ()+logSyncLocalTime0 =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)+     logDIO DEBUG $+       "t = " ++ (show $ pointTime p) +++       ", global t = " ++ (show t') +++       ": synchronizing the local time in ring 0..."++-- | Log that the local time is sent to the time server.+logSendLocalTime :: Event DIO ()+logSendLocalTime =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)+     logDIO DEBUG $+       "t = " ++ (show $ pointTime p) +++       ", global t = " ++ (show t') +++       ": sending the local time to the time server after delay..."++-- | Log an evidence of the premature IO.+logPrematureIO :: Event DIO ()+logPrematureIO =+  Event $ \p ->+  logDIO ERROR $+  "t = " ++ (show $ pointTime p) +++  ": detected a premature IO action"++-- | Log an evidence of receiving the outdated message.+logOutdatedMessage :: Event DIO ()+logOutdatedMessage =+  Event $ \p ->+  logDIO ERROR $+  "t = " ++ (show $ pointTime p) +++  ": received the outdated message"++-- | Reduce events till the specified time.+reduceEvents :: Double -> Event DIO ()+reduceEvents t =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     liftIOUnsafe $+       do reduceInputMessages (queueInputMessages q) t+          reduceOutputMessages (queueOutputMessages q) t+          reduceLog (queueLog q) t++instance {-# OVERLAPPING #-} MonadIO (Event DIO) where++  liftIO m =+    Event $ \p ->+    do ok <- invokeEvent p $+             runTimeWarp $+             syncLocalTime $+             return ()+       if ok+         then liftIOUnsafe m+         else do f <- fmap dioAllowPrematureIO dioParams+                 if f+                   then do ---+                           --- invokeEvent p $ logPrematureIO+                           ---+                           liftIOUnsafe m+                   else error $+                        "Detected a premature IO action at t = " +++                        (show $ pointTime p) ++ ": liftIO"++-- | Update the local time.+updateLocalTime :: Event DIO Bool+updateLocalTime =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     timestamp0 <- liftIOUnsafe $ readIORef (queueLocalTimeTimestamp q)+     timestamp  <- liftIOUnsafe getCurrentTime+     let dt = queueLocalTimeInterval q+     if timestamp >= addUTCTime dt timestamp0+       then do ---+               --- logDIO DEBUG $ "t = " ++ (show $ pointTime p) ++ ": updating the local time"+               ---+               liftIOUnsafe $+                 do t <- readIORef (queueTime q)+                    loct0 <- readIORef (queueLocalTime0 q)+                    writeIORef (queueLocalTime q) (min t loct0)+                    writeIORef (queueLocalTime0 q) t+                    writeIORef (queueLocalTimeTimestamp q) timestamp+                    return (t /= loct0)+       else return False++-- | Get the local time.+getLocalTime :: Event DIO Double+getLocalTime =+  Event $ \p ->+  let q = runEventQueue $ pointRun p+  in liftIOUnsafe $ readIORef (queueLocalTime q)++-- | Send the local time to the time server.+sendLocalTime :: Event DIO ()+sendLocalTime =+  Event $ \p ->+  do ---+     --- invokeEvent p logSendLocalTime+     ---+     invokeEvent p updateLocalTime+     t <- invokeEvent p getLocalTime+     sender   <- messageInboxId+     receiver <- timeServerId+     liftDistributedUnsafe $+       DP.send receiver (LocalTimeMessage sender t)++-- | Synchronize the local time executing the specified computation.+syncLocalTime :: Dynamics DIO () -> TimeWarp DIO ()+syncLocalTime m =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+         t = pointTime p+     invokeDynamics p m+     t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)+     if t' > t+       then error "Inconsistent time: syncLocalTime"+       else if t == spcStartTime (pointSpecs p)+            then return ()+            else do ---+                    --- invokeEvent p logSyncLocalTime+                    ---+                    ch <- messageChannel+                    dt <- fmap dioSyncTimeout dioParams+                    f  <- liftIOUnsafe $+                          timeout dt $ awaitChannel ch+                    ok <- invokeEvent p $ runTimeWarp processChannelMessages+                    if ok+                      then do case f of+                                Just _  ->+                                  invokeTimeWarp p $ syncLocalTime m+                                Nothing ->+                                  do f <- invokeEvent p updateLocalTime+                                     invokeEvent p sendLocalTime+                                     if f+                                       then invokeTimeWarp p $ syncLocalTime m+                                       else invokeTimeWarp p $ syncLocalTime0 m+                      else return ()+  +-- | Synchronize the local time executing the specified computation in ring 0.+syncLocalTime0 :: Dynamics DIO () -> TimeWarp DIO ()+syncLocalTime0 m =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+         t = pointTime p+     invokeDynamics p m+     t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)+     if t' > t+       then error "Inconsistent time: syncLocalTime0"+       else if t' == pointTime p+            then return ()+            else do ---+                    --- invokeEvent p logSyncLocalTime0+                    ---+                    ch <- messageChannel+                    dt <- fmap dioSyncTimeout dioParams+                    f  <- liftIOUnsafe $+                          timeout dt $ awaitChannel ch+                    ok <- invokeEvent p $ runTimeWarp processChannelMessages+                    if ok+                      then do case f of+                                Just _  ->+                                  invokeTimeWarp p $ syncLocalTime m+                                Nothing ->+                                  error "Detected a deadlock when synchronizing the local time: syncLocalTime0"+                      else return ()++-- | Run the computation and return a flag indicating whether there was no rollback.+runTimeWarp :: TimeWarp DIO () -> Event DIO Bool+runTimeWarp m =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     v0 <- liftIOUnsafe $ inputMessageQueueVersion (queueInputMessages q)+     invokeTimeWarp p m+     v2 <- liftIOUnsafe $ inputMessageQueueVersion (queueInputMessages q)+     return (v0 == v2)++-- | Synchronize the events.+syncEvents :: EventProcessing -> Event DIO ()+syncEvents processing =+  Event $ \p ->+  do ok <- invokeEvent p $+           runTimeWarp $+           syncLocalTime $+           processEvents processing+     unless ok $+       invokeEvent p $+       syncEvents processing++-- | Synchronize the simulation in all nodes and call+-- the specified computation at the given modeling time.+--+-- It is rather safe to put 'liftIO' within this function.+syncEvent :: Double -> Event DIO () -> Event DIO ()+syncEvent t h =+  enqueueEvent t $+  Event $ \p ->+  do ok <- invokeEvent p $+           runTimeWarp $+           syncLocalTime $+           return ()+     when ok $+       invokeEvent p h++-- | Handle the 'Event' retry.+handleEventRetry :: SimulationRetry -> Event DIO ()+handleEventRetry e =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+         t = pointTime p+     ---+     logDIO NOTICE $+       "t = " ++ show t +++       ": retrying the computations..."+     ---+     invokeTimeWarp p $+       retryInputMessages (queueInputMessages q)+     let loop =+           do ---+              --- logDIO DEBUG $+              ---   "t = " ++ show t +++              ---   ": waiting for arriving a message..."+              ---+              ch <- messageChannel+              dt <- fmap dioSyncTimeout dioParams+              f  <- liftIOUnsafe $+                    timeout dt $ awaitChannel ch+              ok <- invokeEvent p $ runTimeWarp processChannelMessages+              when ok $+                case f of+                  Just _  -> loop+                  Nothing -> loop0+         loop0 =+           do ---+              --- logDIO DEBUG $+              ---   "t = " ++ show t +++              ---   ": waiting for arriving a message in ring 0..."+              ---+              ch <- messageChannel+              dt <- fmap dioSyncTimeout dioParams+              f  <- liftIOUnsafe $+                    timeout dt $ awaitChannel ch+              ok <- invokeEvent p $ runTimeWarp processChannelMessages+              when ok $+                case f of+                  Just _  -> loop+                  Nothing ->+                    error $+                    "Detected a deadlock when retrying the computations: handleEventRetry\n" +++                    "--- the nested exception ---\n" ++ show e +     loop
+ Simulation/Aivika/Distributed/Optimistic/Internal/IO.hs view
@@ -0,0 +1,42 @@++{-# LANGUAGE FlexibleInstances #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.IO+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines functions for direct embedding 'IO' computations.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.IO+       (MonadIOUnsafe(..)) where++import Control.Monad+import Control.Monad.Trans++import Simulation.Aivika.Trans+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO++-- | Allows embedding unsafe 'IO' computations.+class MonadIOUnsafe m where++  -- | Lift the computation.+  liftIOUnsafe :: IO a -> m a++instance MonadIOUnsafe DIO where+  liftIOUnsafe = DIO . const . liftIO++instance MonadIOUnsafe (Parameter DIO) where+  liftIOUnsafe = liftComp . DIO . const . liftIO ++instance MonadIOUnsafe (Simulation DIO) where+  liftIOUnsafe = liftComp . DIO . const . liftIO ++instance MonadIOUnsafe (Dynamics DIO) where+  liftIOUnsafe = liftComp . DIO . const . liftIO +  +instance MonadIOUnsafe (Event DIO) where+  liftIOUnsafe = liftComp . DIO . const . liftIO
+ Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs view
@@ -0,0 +1,387 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines an input message queue.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue+       (InputMessageQueue,+        newInputMessageQueue,+        inputMessageQueueSize,+        inputMessageQueueVersion,+        enqueueMessage,+        messageEnqueued,+        retryInputMessages,+        reduceInputMessages) where++import Data.Maybe+import Data.List+import Data.IORef++import Control.Monad+import Control.Monad.Trans+import qualified Control.Distributed.Process as DP++import Simulation.Aivika.Vector+import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.Simulation+import Simulation.Aivika.Trans.Dynamics+import Simulation.Aivika.Trans.Event+import Simulation.Aivika.Trans.Signal+import Simulation.Aivika.Trans.Internal.Types++import Simulation.Aivika.Distributed.Optimistic.Internal.Priority+import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.IO+import Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp+import Simulation.Aivika.Distributed.Optimistic.DIO++-- | Specifies the input message queue.+data InputMessageQueue =+  InputMessageQueue { inputMessageLog :: UndoableLog,+                      -- ^ the Redo/Undo log.+                      inputMessageRollbackPre :: Bool -> TimeWarp DIO (),+                      -- ^ Rollback the operations till the specified time before actual changes either including the time or not.+                      inputMessageRollbackPost :: Bool -> TimeWarp DIO (),+                      -- ^ Rollback the operations till the specified time after actual changes either including the time or not.+                      inputMessageRollbackTime :: TimeWarp DIO (),+                      -- ^ Rollback the event time.+                      inputMessageSource :: SignalSource DIO Message,+                      -- ^ The message source.+                      inputMessages :: Vector InputMessageQueueItem,+                      -- ^ The input messages.+                      inputMessageActions :: IORef [Event DIO ()],+                      -- ^ The list of actions to perform.+                      inputMessageVersionRef :: IORef Int+                      -- ^ The number of reversions.+                    }++-- | Specified the input message queue item.+data InputMessageQueueItem =+  InputMessageQueueItem { itemMessage :: Message,+                          -- ^ The item message.+                          itemAnnihilated :: IORef Bool,+                          -- ^ Whether the item was annihilated.+                          itemProcessed :: IORef Bool+                          -- ^ Whether the item was processed.+                        }++-- | Create a new input message queue.+newInputMessageQueue :: UndoableLog+                        -- ^ the Redo/Undo log+                        -> (Bool -> TimeWarp DIO ())+                        -- ^ rollback operations till the specified time before actual changes either including the time or not+                        -> (Bool -> TimeWarp DIO ())+                        -- ^ rollback operations till the specified time after actual changes either including the time or not+                        -> TimeWarp DIO ()+                        -- ^ rollback the event time+                        -> DIO InputMessageQueue+newInputMessageQueue log rollbackPre rollbackPost rollbackTime =+  do ms <- liftIOUnsafe newVector+     r  <- liftIOUnsafe $ newIORef []+     s  <- newSignalSource0+     v  <- liftIOUnsafe $ newIORef 0+     return InputMessageQueue { inputMessageLog = log,+                                inputMessageRollbackPre = rollbackPre,+                                inputMessageRollbackPost = rollbackPost,+                                inputMessageRollbackTime = rollbackTime,+                                inputMessageSource = s,+                                inputMessages = ms,+                                inputMessageActions = r,+                                inputMessageVersionRef = v }++-- | Return the input message queue size.+inputMessageQueueSize :: InputMessageQueue -> IO Int+inputMessageQueueSize = vectorCount . inputMessages++-- | Return the reversion count.+inputMessageQueueVersion :: InputMessageQueue -> IO Int+inputMessageQueueVersion = readIORef . inputMessageVersionRef++-- | Return a complement.+complement :: Int -> Int+complement x = - x - 1++-- | Raised when the message is enqueued.+messageEnqueued :: InputMessageQueue -> Signal DIO Message+messageEnqueued q = publishSignal (inputMessageSource q)++-- | Enqueue a new message ignoring the duplicated messages.+enqueueMessage :: InputMessageQueue -> Message -> TimeWarp DIO ()+enqueueMessage q m =+  TimeWarp $ \p ->+  do let t  = messageReceiveTime m+         t0 = pointTime p+     i <- liftIOUnsafe $ findAntiMessage q m+     case i of+       Nothing -> return ()+       Just i | i >= 0 ->+         do -- found the anti-message at the specified index+            when (t <= t0) $+              liftIOUnsafe $ modifyIORef' (inputMessageVersionRef q) (+ 1)+            item <- liftIOUnsafe $ readVector (inputMessages q) i+            f <- liftIOUnsafe $ readIORef (itemProcessed item)+            if f+              then do let p' = pastPoint t p+                      logRollbackInputMessages t0 t True+                      invokeTimeWarp p' $+                        rollbackInputMessages q True $+                        liftIOUnsafe $ annihilateMessage q m i+                      invokeTimeWarp p' $+                        inputMessageRollbackTime q+              else liftIOUnsafe $ annihilateMessage q m i+       Just i' | i' < 0 ->+         do -- insert the message at the specified right index+            when (t < t0) $+              liftIOUnsafe $ modifyIORef' (inputMessageVersionRef q) (+ 1)+            let i = complement i'+            if t < t0+              then do let p' = pastPoint t p+                      logRollbackInputMessages t0 t False+                      invokeTimeWarp p' $+                        rollbackInputMessages q False $+                        Event $ \p' ->+                        do liftIOUnsafe $ insertMessage q m i+                           invokeEvent p' $ activateMessage q i+                      invokeTimeWarp p' $+                        inputMessageRollbackTime q+              else do liftIOUnsafe $ insertMessage q m i+                      invokeEvent p $ activateMessage q i++-- | Log the rollback.+logRollbackInputMessages :: Double -> Double -> Bool -> DIO ()+logRollbackInputMessages t0 t including =+  logDIO NOTICE $+  "Rollback at t = " ++ (show t0) ++ " --> " ++ (show t) +++  (if not including then " not including" else "")++-- | Retry the computations.+retryInputMessages :: InputMessageQueue -> TimeWarp DIO ()+retryInputMessages q =+  TimeWarp $ \p ->+  do liftIOUnsafe $+       modifyIORef' (inputMessageVersionRef q) (+ 1)+     invokeTimeWarp p $+       rollbackInputMessages q True $+       return ()+     invokeTimeWarp p $+       inputMessageRollbackTime q++-- | Rollback the input messages till the specified time, either including the time or not, and apply the given computation.+rollbackInputMessages :: InputMessageQueue -> Bool -> Event DIO () -> TimeWarp DIO ()+rollbackInputMessages q including m =+  TimeWarp $ \p ->+  do liftIOUnsafe $+       requireEmptyMessageActions q+     invokeTimeWarp p $+       inputMessageRollbackPre q including+     invokeEvent p m+     invokeEvent p $+       performMessageActions q+     invokeTimeWarp p $+       inputMessageRollbackPost q including++-- | Return the point in the past.+pastPoint :: Double -> Point DIO -> Point DIO+pastPoint t p = p'+  where sc = pointSpecs p+        t0 = spcStartTime sc+        dt = spcDT sc+        n  = fromIntegral $ floor ((t - t0) / dt)+        p' = p { pointTime = t,+                 pointIteration = n,+                 pointPhase = -1 }++-- | Require that the there are no message actions.+requireEmptyMessageActions :: InputMessageQueue -> IO ()+requireEmptyMessageActions q =+  do xs <- readIORef (inputMessageActions q)+     unless (null xs) $+       error "There are incomplete message actions: requireEmptyMessageActions"++-- | Perform the message actions.+performMessageActions :: InputMessageQueue -> Event DIO ()+performMessageActions q =+  do xs <- liftIOUnsafe $ readIORef (inputMessageActions q)+     liftIOUnsafe $ writeIORef (inputMessageActions q) []+     sequence_ xs++-- | Return the leftmost index for the current time.+leftMessageIndex :: InputMessageQueue -> Double -> Int -> IO Int+leftMessageIndex q t i+  | i == 0    = return 0+  | otherwise = do let i' = i - 1+                   item' <- readVector (inputMessages q) i'+                   let m' = itemMessage item'+                       t' = messageReceiveTime m'+                   if t' > t+                     then error "Incorrect index: leftMessageIndex"+                     else if t' < t+                          then return i+                          else leftMessageIndex q t i'++-- | Find an anti-message and return the index; otherwise, return a complement to+-- the insertion index with the current receive time. The result is 'Nothing' if+-- the message is duplicated.+findAntiMessage :: InputMessageQueue -> Message -> IO (Maybe Int)+findAntiMessage q m =+  do right <- lookupRightMessageIndex q (messageReceiveTime m)+     if right < 0+       then return (Just right)+       else let loop i+                  | i < 0     = return (Just $ complement (right + 1))+                  | otherwise =+                    do item <- readVector (inputMessages q) i+                       let m' = itemMessage item+                           t  = messageReceiveTime m+                           t' = messageReceiveTime m'+                       if t' > t+                         then error "Incorrect index: findAntiMessage"+                         else if t' < t+                              then return (Just $ complement (right + 1))+                              else if antiMessages m m'+                                   then return (Just i)+                                   else if m == m'+                                        then return Nothing+                                        else loop (i - 1)+            in loop right       ++-- | Annihilate a message at the specified index.+annihilateMessage :: InputMessageQueue -> Message -> Int -> IO ()+annihilateMessage q m i =+  do item <- readVector (inputMessages q) i+     let m' = itemMessage item+     unless (antiMessages m m') $+       error "Cannot annihilate another message: annihilateMessage"+     f <- readIORef (itemProcessed item)+     when f $+       error "Cannot annihilate the processed message: annihilateMessage"+     vectorDeleteAt (inputMessages q) i+     writeIORef (itemAnnihilated item) True++-- | Activate a message at the specified index.+activateMessage :: InputMessageQueue -> Int -> Event DIO ()+activateMessage q i =+  do item <- liftIOUnsafe $ readVector (inputMessages q) i+     let m    = itemMessage item+         loop =+           do f <- liftIOUnsafe $ readIORef (itemAnnihilated item)+              unless f $+                do writeLog (inputMessageLog q) $+                     liftIOUnsafe $+                     modifyIORef (inputMessageActions q) (loop :)+                   enqueueEvent (messageReceiveTime m) $+                     do f <- liftIOUnsafe $ readIORef (itemAnnihilated item)+                        unless f $+                          do writeLog (inputMessageLog q) $+                               liftIOUnsafe $+                               writeIORef (itemProcessed item) False+                             liftIOUnsafe $+                               writeIORef (itemProcessed item) True+                             unless (messageAntiToggle m) $+                               triggerSignal (inputMessageSource q) m+     loop++-- | Insert a new message.+insertMessage :: InputMessageQueue -> Message -> Int -> IO ()+-- insertMessage q m i =+--   do r1 <- newIORef False+--      r2 <- newIORef False+--      let item = InputMessageQueueItem m r1 r2+--      vectorInsert (inputMessages q) i item+insertMessage q m i =+  do n <- vectorCount (inputMessages q)+     when (i < n) $+       do item0 <- readVector (inputMessages q) i+          let m0 = itemMessage item0+          unless (messageReceiveTime m < messageReceiveTime m0) $+            error "Error inserting a new input message (check before): insertMessage"+     when (i > 0) $+       do item0 <- readVector (inputMessages q) (i - 1)+          let m0 = itemMessage item0+          unless (messageReceiveTime m >= messageReceiveTime m0) $+            error "Error inserting a new input message (check after): insertMessage"+     r1 <- newIORef False+     r2 <- newIORef False+     let item = InputMessageQueueItem m r1 r2+     vectorInsert (inputMessages q) i item++-- | Search for the rightmost message index.+lookupRightMessageIndex' :: InputMessageQueue -> Double -> Int -> Int -> IO Int+lookupRightMessageIndex' q t left right =+  if left > right+  then return $ complement (right + 1)+  else  +    do let index = ((left + 1) + right) `div` 2+       item <- readVector (inputMessages q) index+       let m' = itemMessage item+           t' = messageReceiveTime m'+       if left == right+         then if t' > t+              then return $ complement right+              else if t' < t+                   then return $ complement (right + 1)+                   else return right+         else if t' > t+              then lookupRightMessageIndex' q t left (index - 1)+              else if t' < t+                   then lookupRightMessageIndex' q t (index + 1) right+                   else lookupRightMessageIndex' q t index right++-- | Search for the leftmost message index.+lookupLeftMessageIndex' :: InputMessageQueue -> Double -> Int -> Int -> IO Int+lookupLeftMessageIndex' q t left right =+  if left > right+  then return $ complement left+  else  +    do let index = (left + right) `div` 2+       item <- readVector (inputMessages q) index+       let m' = itemMessage item+           t' = messageReceiveTime m'+       if left == right+         then if t' > t+              then return $ complement left+              else if t' < t+                   then return $ complement (left + 1)+                   else return left+         else if t' > t+              then lookupLeftMessageIndex' q t left (index - 1)+              else if t' < t+                   then lookupLeftMessageIndex' q t (index + 1) right+                   else lookupLeftMessageIndex' q t left index+ +-- | Search for the rightmost message index.+lookupRightMessageIndex :: InputMessageQueue -> Double -> IO Int+lookupRightMessageIndex q t =+  do n <- vectorCount (inputMessages q)+     lookupRightMessageIndex' q t 0 (n - 1)+ +-- | Search for the leftmost message index.+lookupLeftMessageIndex :: InputMessageQueue -> Double -> IO Int+lookupLeftMessageIndex q t =+  do n <- vectorCount (inputMessages q)+     lookupLeftMessageIndex' q t 0 (n - 1)++-- | Reduce the input messages till the specified time.+reduceInputMessages :: InputMessageQueue -> Double -> IO ()+reduceInputMessages q t =+  do count <- vectorCount (inputMessages q)+     len   <- loop count 0+     when (len > 0) $+       vectorDeleteRange (inputMessages q) 0 len+       where+         loop n i+           | i >= n    = return i+           | otherwise = do item <- readVector (inputMessages q) i+                            let m = itemMessage item+                            if messageReceiveTime m < t+                              then loop n (i + 1)+                              else return i+
+ Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs-boot view
@@ -0,0 +1,49 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This is an hs-boot file.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue+       (InputMessageQueue,+        newInputMessageQueue,+        inputMessageQueueSize,+        inputMessageQueueVersion,+        enqueueMessage,+        messageEnqueued,+        retryInputMessages,+        reduceInputMessages) where++import Simulation.Aivika.Trans.Simulation+import Simulation.Aivika.Trans.Event+import Simulation.Aivika.Trans.Signal++import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp++data InputMessageQueue++newInputMessageQueue :: UndoableLog+                        -> (Bool -> TimeWarp DIO ())+                        -> (Bool -> TimeWarp DIO ())+                        -> TimeWarp DIO ()+                        -> DIO InputMessageQueue++inputMessageQueueSize :: InputMessageQueue -> IO Int++inputMessageQueueVersion :: InputMessageQueue -> IO Int++enqueueMessage :: InputMessageQueue -> Message -> TimeWarp DIO ()++messageEnqueued :: InputMessageQueue -> Signal DIO Message++retryInputMessages :: InputMessageQueue -> TimeWarp DIO ()++reduceInputMessages :: InputMessageQueue -> Double -> IO ()
+ Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs view
@@ -0,0 +1,102 @@++{-# LANGUAGE RankNTypes, DeriveGeneric #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Message+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines a message.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.Message+       (Message(..),+        antiMessage,+        antiMessages,+        LocalProcessMessage(..),+        TimeServerMessage(..)) where++import GHC.Generics++import Data.Typeable+import Data.Binary++import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Serializable++-- | Represents a message.+data Message =+  Message { messageSequenceNo :: Int,+            -- ^ The sequence number.+            messageSendTime :: Double,+            -- ^ The send time.+            messageReceiveTime :: Double,+            -- ^ The receive time.+            messageSenderId :: DP.ProcessId,+            -- ^ The sender of the message.+            messageReceiverId :: DP.ProcessId,+            -- ^ The receiver of the message.+            messageAntiToggle :: Bool,+            -- ^ Whether this is an anti-message.+            messageData :: DP.Message+            -- ^ The message data.+          } deriving (Show, Typeable, Generic)++instance Binary Message++-- | Return an anti-message.+antiMessage :: Message -> Message+antiMessage x = x { messageAntiToggle = not (messageAntiToggle x) }++-- | Whether two messages are anti-messages.+antiMessages :: Message -> Message -> Bool+antiMessages x y =+  (messageSequenceNo x == messageSequenceNo y) &&+  (messageSendTime x == messageSendTime y) &&+  (messageReceiveTime x == messageReceiveTime y) &&+  (messageSenderId x == messageSenderId y) &&+  (messageReceiverId x == messageReceiverId y) &&+  (messageAntiToggle x /= messageAntiToggle y)++instance Eq Message where++  x == y =+    (messageSequenceNo x == messageSequenceNo y) &&+    (messageSendTime x == messageSendTime y) &&+    (messageReceiveTime x == messageReceiveTime y) &&+    (messageSenderId x == messageSenderId y) &&+    (messageReceiverId x == messageReceiverId y) &&+    (messageAntiToggle x == messageAntiToggle y)++-- | The message sent to the local process.+data LocalProcessMessage = QueueMessage Message+                           -- ^ the message has come from the remote process+                         | QueueMessageBulk [Message]+                           -- ^ a bulk of messages that have come from the remote process+                         | GlobalTimeMessage (Maybe Double)+                           -- ^ the time server sent a global time+                         | LocalTimeMessageResp Double+                           -- ^ the time server replied to 'LocalTimeMessage' sending its global time in response+                         | TerminateLocalProcessMessage+                           -- ^ the time server asked to terminate the process+                         deriving (Eq, Show, Typeable, Generic)++instance Binary LocalProcessMessage++-- | The time server message.+data TimeServerMessage = RegisterLocalProcessMessage DP.ProcessId+                         -- ^ register the local process in the time server+                       | UnregisterLocalProcessMessage DP.ProcessId+                         -- ^ unregister the local process from the time server+                       | GlobalTimeMessageResp DP.ProcessId Double+                         -- ^ the local process replied to 'GlobalTimeMessage' sending its local time in response+                       | LocalTimeMessage DP.ProcessId Double+                         -- ^ the local process sent its local time+                       | TerminateTimeServerMessage DP.ProcessId+                         -- ^ the local process asked to terminate the time server+                       deriving (Eq, Show, Typeable, Generic)++instance Binary TimeServerMessage+
+ Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs view
@@ -0,0 +1,135 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines an output message queue.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue+       (OutputMessageQueue,+        newOutputMessageQueue,+        outputMessageQueueSize,+        sendMessage,+        rollbackOutputMessages,+        reduceOutputMessages,+        generateMessageSequenceNo) where++import Data.List+import Data.IORef++import Control.Monad+import Control.Monad.Trans+import qualified Control.Distributed.Process as DP++import qualified Simulation.Aivika.DoubleLinkedList as DLL++import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.Simulation+import Simulation.Aivika.Trans.Event++import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.IO+import Simulation.Aivika.Distributed.Optimistic.DIO++-- | Specifies the output message queue.+data OutputMessageQueue =+  OutputMessageQueue { outputMessages :: DLL.DoubleLinkedList Message,+                       -- ^ The output messages.+                       outputMessageSequenceNo :: IORef Int+                       -- ^ The next sequence number.+                     }++-- | Create a new output message queue.+newOutputMessageQueue :: DIO OutputMessageQueue+newOutputMessageQueue =+  do ms <- liftIOUnsafe DLL.newList+     rn <- liftIOUnsafe $ newIORef 0+     return OutputMessageQueue { outputMessages = ms,+                                 outputMessageSequenceNo = rn }++-- | Return the output message queue size.+outputMessageQueueSize :: OutputMessageQueue -> IO Int+outputMessageQueueSize = DLL.listCount . outputMessages++-- | Send the message.+sendMessage :: OutputMessageQueue -> Message -> DIO ()+sendMessage q m =+  do when (messageSendTime m > messageReceiveTime m) $+       error "The Send time cannot be greater than the Receive message time: sendMessage"+     when (messageAntiToggle m) $+       error "Cannot directly send the anti-message: sendMessage"+     f <- liftIOUnsafe $ DLL.listNull (outputMessages q)+     unless f $+       do m' <- liftIOUnsafe $ DLL.listLast (outputMessages q)+          when (messageSendTime m' > messageSendTime m) $+            error "A new output message comes from the past: sendMessage."+     deliverMessage m+     liftIOUnsafe $ DLL.listAddLast (outputMessages q) m++-- | Rollback the messages till the specified time either including that one or not.+rollbackOutputMessages :: OutputMessageQueue -> Double -> Bool -> DIO ()+rollbackOutputMessages q t including =+  do ms <- liftIOUnsafe $ extractMessagesToRollback q t including+     deliverAntiMessages $ map antiMessage ms+     -- forM_ ms $ deliverAntiMessage . antiMessage+                 +-- | Return the messages to roolback by the specified time.+extractMessagesToRollback :: OutputMessageQueue -> Double -> Bool -> IO [Message]+extractMessagesToRollback q t including = loop []+  where+    loop acc =+      do f <- DLL.listNull (outputMessages q)+         if f+           then return acc+           else do m <- DLL.listLast (outputMessages q)+                   if (not including) && (messageSendTime m == t)+                     then return acc+                     else if messageSendTime m < t+                          then return acc+                          else do DLL.listRemoveLast (outputMessages q)+                                  loop (m : acc)++-- | Reduce the output messages till the specified time.+reduceOutputMessages :: OutputMessageQueue -> Double -> IO ()+reduceOutputMessages q t = loop+  where+    loop =+      do f <- DLL.listNull (outputMessages q)+         unless f $+           do m <- DLL.listFirst (outputMessages q)+              when (messageSendTime m < t) $+                do DLL.listRemoveFirst (outputMessages q)+                   loop++-- | Generate a next message sequence number.+generateMessageSequenceNo :: OutputMessageQueue -> IO Int+generateMessageSequenceNo q =+  atomicModifyIORef (outputMessageSequenceNo q) $ \n ->+  let n' = n + 1 in n' `seq` (n', n)++-- | Deliver the message on low level.+deliverMessage :: Message -> DIO ()+deliverMessage x =+  liftDistributedUnsafe $+  DP.send (messageReceiverId x) (QueueMessage x)++-- | Deliver the anti-message on low level.+deliverAntiMessage :: Message -> DIO ()+deliverAntiMessage x =+  liftDistributedUnsafe $+  DP.send (messageReceiverId x) (QueueMessage x)++-- | Deliver the anti-messages on low level.+deliverAntiMessages :: [Message] -> DIO ()+deliverAntiMessages xs =+  let ys = groupBy (\a b -> messageReceiverId a == messageReceiverId b) xs+      dlv []         = return ()+      dlv zs@(z : _) =+        liftDistributedUnsafe $+        DP.send (messageReceiverId z) (QueueMessageBulk zs)+  in forM_ ys dlv
+ Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs-boot view
@@ -0,0 +1,37 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This is an hs-boot file.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue+       (OutputMessageQueue,+        newOutputMessageQueue,+        outputMessageQueueSize,+        sendMessage,+        rollbackOutputMessages,+        reduceOutputMessages) where++import Simulation.Aivika.Trans.Simulation+import Simulation.Aivika.Trans.Event+import Simulation.Aivika.Trans.Signal++import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO++data OutputMessageQueue++newOutputMessageQueue :: DIO OutputMessageQueue++outputMessageQueueSize :: OutputMessageQueue -> IO Int++sendMessage :: OutputMessageQueue -> Message -> DIO ()++rollbackOutputMessages :: OutputMessageQueue -> Double -> Bool -> DIO ()++reduceOutputMessages :: OutputMessageQueue -> Double -> IO ()
+ Simulation/Aivika/Distributed/Optimistic/Internal/Priority.hs view
@@ -0,0 +1,40 @@++{-# LANGUAGE DeriveGeneric #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Priority+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines the logging 'Priority'.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.Priority+       (Priority(..),+        embracePriority) where++import Data.Typeable+import Data.Binary++import GHC.Generics++-- | The logging priority.+data Priority = DEBUG+                -- ^ Debug messages+              | INFO+                -- ^ Information+              | NOTICE+                -- ^ Normal runtime conditions+              | WARNING+                -- ^ Warnings+              | ERROR+                -- ^ Errors+              deriving (Eq, Ord, Show, Read, Typeable, Generic)++instance Binary Priority++-- | Embrace the priority in brackets.+embracePriority :: Priority -> String+embracePriority p = "[" ++ show p ++ "]"
+ Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs view
@@ -0,0 +1,73 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Ref+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- The implementation of mutable references.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.Ref+       (Ref,+        newRef,+        newRef0,+        readRef,+        writeRef,+        modifyRef) where++import Data.IORef++import Control.Monad+import Control.Monad.Trans++import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.Internal.Types++import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.Event+import Simulation.Aivika.Distributed.Optimistic.Internal.IO+import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog++-- | A mutable reference.+newtype Ref a = Ref { refValue :: IORef a }++instance Eq (Ref a) where+  (Ref r1) == (Ref r2) = r1 == r2++-- | Create a new reference.+newRef :: a -> Simulation DIO (Ref a)+newRef = liftComp . newRef0++-- | Create a new reference.+newRef0 :: a -> DIO (Ref a)+newRef0 a =+  do x <- liftIOUnsafe $ newIORef a+     return Ref { refValue = x }+     +-- | Read the value of a reference.+readRef :: Ref a -> Event DIO a+readRef r = Event $ \p ->+  liftIOUnsafe $ readIORef (refValue r)++-- | Write a new value into the reference.+writeRef :: Ref a -> a -> Event DIO ()+writeRef r a = Event $ \p ->+  do let log = queueLog $ runEventQueue (pointRun p)+     a0 <- liftIOUnsafe $ readIORef (refValue r)+     invokeEvent p $+       writeLog log $+       liftIOUnsafe $ writeIORef (refValue r) a0+     a `seq` liftIOUnsafe $ writeIORef (refValue r) a++-- | Mutate the contents of the reference.+modifyRef :: Ref a -> (a -> a) -> Event DIO ()+modifyRef r f = Event $ \p ->+  do let log = queueLog $ runEventQueue (pointRun p)+     a <- liftIOUnsafe $ readIORef (refValue r)+     let b = f a+     invokeEvent p $+       writeLog log $+       liftIOUnsafe $ writeIORef (refValue r) a+     b `seq` liftIOUnsafe $ writeIORef (refValue r) b
+ Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs-boot view
@@ -0,0 +1,29 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Ref+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This is an hs-boot file.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.Ref where++import Simulation.Aivika.Trans.Internal.Types+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO++data Ref a++instance Eq (Ref a)++newRef :: a -> Simulation DIO (Ref a)++newRef0 :: a -> DIO (Ref a)+     +readRef :: Ref a -> Event DIO a++writeRef :: Ref a -> a -> Event DIO ()++modifyRef :: Ref a -> (a -> a) -> Event DIO ()
+ Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs view
@@ -0,0 +1,247 @@++{-# LANGUAGE DeriveGeneric #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module allows running the time server that coordinates the global simulation time.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer+       (TimeServerParams(..),+        defaultTimeServerParams,+        timeServer) where++import qualified Data.Map as M+import Data.Maybe+import Data.IORef+import Data.Typeable+import Data.Binary++import GHC.Generics++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Closure (remotable, mkClosure)++import Simulation.Aivika.Distributed.Optimistic.Internal.Priority+import Simulation.Aivika.Distributed.Optimistic.Internal.Message++-- | The time server parameters.+data TimeServerParams =+  TimeServerParams { tsLoggingPriority :: Priority,+                     -- ^ the logging priority+                     tsExpectTimeout :: Int,+                     -- ^ the timeout in microseconds within which a new message is expected+                     tsTimeSyncDelay :: Int+                     -- ^ the further delay in microseconds before the time synchronization+                   } deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary TimeServerParams++-- | The time server.+data TimeServer =+  TimeServer { tsParams :: TimeServerParams,+               -- ^ the time server parameters+               tsProcesses :: IORef (M.Map DP.ProcessId LocalProcessInfo),+               -- ^ the information about local processes+               tsGlobalTime :: IORef (Maybe Double),+               -- ^ the global time of the model+               tsGlobalTimeInvalid :: IORef Bool+               -- ^ whether the global time is invalid+             }++-- | The information about the local process.+data LocalProcessInfo =+  LocalProcessInfo { lpLocalTime :: IORef (Maybe Double),+                     -- ^ the local time of the process+                     lpSentGlobalTime :: IORef (Maybe Double)+                     -- ^ the global time sent to the process for the last time+                   }++-- | The default time server parameters.+defaultTimeServerParams :: TimeServerParams+defaultTimeServerParams =+  TimeServerParams { tsLoggingPriority = DEBUG,+                     tsExpectTimeout = 1000,+                     tsTimeSyncDelay = 1000+                   }++-- | Create a new time server.+newTimeServer :: TimeServerParams -> IO TimeServer+newTimeServer ps =+  do m  <- newIORef M.empty+     t0 <- newIORef Nothing+     f  <- newIORef False+     return TimeServer { tsParams = ps,+                         tsProcesses = m,+                         tsGlobalTime = t0,+                         tsGlobalTimeInvalid = f+                       }++-- | Process the time server message.+processTimeServerMessage :: TimeServer -> TimeServerMessage -> DP.Process ()+processTimeServerMessage server (RegisterLocalProcessMessage pid) =+  liftIO $+  do t1 <- newIORef Nothing+     t2 <- newIORef Nothing+     writeIORef (tsGlobalTimeInvalid server) True+     modifyIORef (tsProcesses server) $+       M.insert pid LocalProcessInfo { lpLocalTime = t1, lpSentGlobalTime = t2 }+processTimeServerMessage server (UnregisterLocalProcessMessage pid) =+  join $ liftIO $+  do m <- readIORef (tsProcesses server)+     case M.lookup pid m of+       Nothing ->+         return $+         logTimeServer server WARNING $+         "Time Server: unknown process identifier " ++ show pid+       Just x  ->+         do t0 <- readIORef (tsGlobalTime server) +            t  <- readIORef (lpLocalTime x)+            when (t0 .>=. t) $+              writeIORef (tsGlobalTimeInvalid server) True+            writeIORef (tsProcesses server) $+              M.delete pid m+            return $ return ()+processTimeServerMessage server (TerminateTimeServerMessage pid) =+  do pids <-+       liftIO $+       do m <- readIORef (tsProcesses server)+          writeIORef (tsProcesses server) M.empty+          writeIORef (tsGlobalTime server) Nothing+          writeIORef (tsGlobalTimeInvalid server) True+          return $ filter (/= pid) (M.keys m)+     forM_ pids $ \pid ->+       DP.send pid TerminateLocalProcessMessage+     logTimeServer server INFO "Time Server: terminating..."+     DP.terminate+processTimeServerMessage server (GlobalTimeMessageResp pid t') =+  join $ liftIO $+  do m <- readIORef (tsProcesses server)+     case M.lookup pid m of+       Nothing ->+         return $+         do logTimeServer server WARNING $+              "Time Server: unknown process identifier " ++ show pid+            processTimeServerMessage server (RegisterLocalProcessMessage pid)+            processTimeServerMessage server (GlobalTimeMessageResp pid t')+       Just x  ->+         do t0 <- readIORef (tsGlobalTime server)+            t  <- readIORef (lpLocalTime x)+            when (t /= Just t') $+              do writeIORef (lpLocalTime x) (Just t')+                 when ((t0 .>=. t) || (t0 .>=. Just t')) $+                   writeIORef (tsGlobalTimeInvalid server) True+            return $ return ()+processTimeServerMessage server (LocalTimeMessage pid t') =+  join $ liftIO $+  do m <- readIORef (tsProcesses server)+     case M.lookup pid m of+       Nothing ->+         return $+         do logTimeServer server WARNING $+              "Time Server: unknown process identifier " ++ show pid+            processTimeServerMessage server (RegisterLocalProcessMessage pid)+            processTimeServerMessage server (LocalTimeMessage pid t')+       Just x  ->+         do t0 <- readIORef (tsGlobalTime server)+            t  <- readIORef (lpLocalTime x)+            if t == Just t'+              then return $ return ()+              else do writeIORef (lpLocalTime x) (Just t')+                      when ((t0 .>=. t) || (t0 .>=. Just t')) $+                        writeIORef (tsGlobalTimeInvalid server) True+                      t0' <- readIORef (lpSentGlobalTime x)+                      if (t0 == t0') || (isNothing t0)+                        then return $ return ()+                        else do writeIORef (lpSentGlobalTime x) t0+                                return $+                                  DP.send pid (LocalTimeMessageResp $ fromJust t0)++-- | Whether the both values are defined and the first is greater than or equaled to the second.+(.>=.) :: Maybe Double -> Maybe Double -> Bool+(.>=.) (Just x) (Just y) = x >= y+(.>=.) _ _ = False++-- | Whether the both values are defined and the first is greater than the second.+(.>.) :: Maybe Double -> Maybe Double -> Bool+(.>.) (Just x) (Just y) = x > y+(.>.) _ _ = False++-- | Validate the time server.+validateTimeServer :: TimeServer -> DP.Process ()+validateTimeServer server =+  do f <- liftIO $ readIORef (tsGlobalTimeInvalid server)+     when f $+       do t0 <- timeServerGlobalTime server+          case t0 of+            Nothing -> return ()+            Just t0 ->+              do t' <- liftIO $ readIORef (tsGlobalTime server)+                 when (t' .>. Just t0) $+                   logTimeServer server NOTICE+                   "Time Server: the global time has decreased"+                 liftIO $+                   do writeIORef (tsGlobalTime server) (Just t0)+                      writeIORef (tsGlobalTimeInvalid server) False+                 m <- liftIO $ readIORef (tsProcesses server)+                 forM_ (M.assocs m) $ \(pid, x) ->+                   do t0' <- liftIO $ readIORef (lpSentGlobalTime x)+                      when (t0' /= Just t0) $+                        do liftIO $ writeIORef (lpSentGlobalTime x) (Just t0)+                           DP.send pid (GlobalTimeMessage $ Just t0) ++-- | Return the time server global time.+timeServerGlobalTime :: TimeServer -> DP.Process (Maybe Double)+timeServerGlobalTime server =+  do t0 <- liftIO $ readIORef (tsGlobalTime server)+     zs <- liftIO $ fmap M.assocs $ readIORef (tsProcesses server)+     case zs of+       [] -> return Nothing+       ((pid, x) : zs') ->+         do t <- liftIO $ readIORef (lpLocalTime x)+            loop zs t+              where loop [] acc = return acc+                    loop ((pid, x) : zs') acc =+                      do t <- liftIO $ readIORef (lpLocalTime x)+                         case t of+                           Nothing ->+                             do DP.send pid (GlobalTimeMessage Nothing)+                                loop zs' Nothing+                           Just _  ->+                             loop zs' (liftM2 min t acc)++-- | Start the time server.+timeServer :: TimeServerParams -> DP.Process ()+timeServer ps =+  do server <- liftIO $ newTimeServer ps+     logTimeServer server INFO "Time Server: starting..."+     forever $+       do m <- DP.expectTimeout (tsExpectTimeout ps) :: DP.Process (Maybe TimeServerMessage)+          case m of+            Nothing -> return ()+            Just m  ->+              do ---+                 logTimeServer server DEBUG $+                   "Time Server: " ++ show m+                 ---+                 processTimeServerMessage server m+          liftIO $+            threadDelay (tsTimeSyncDelay ps)+          validateTimeServer server++-- | Log the message with the specified priority.+logTimeServer :: TimeServer -> Priority -> String -> DP.Process ()+{-# INLINE logTimeServer #-}+logTimeServer server p message =+  when (tsLoggingPriority (tsParams server) <= p) $+  DP.say $+  embracePriority p ++ " " ++ message+
+ Simulation/Aivika/Distributed/Optimistic/Internal/TimeWarp.hs view
@@ -0,0 +1,25 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines a computation in the course of which the time may warp.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp+       (TimeWarp(..),+        invokeTimeWarp) where++import Simulation.Aivika.Trans+import Simulation.Aivika.Trans.Internal.Types++-- | The time warp computation.+newtype TimeWarp m a = TimeWarp (Point m -> m a)++-- | Invoke the 'TimeWarp' computation.+invokeTimeWarp :: Point m -> TimeWarp m a -> m a+{-# INLINE invokeTimeWarp #-}+invokeTimeWarp p (TimeWarp m) = m p
+ Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs view
@@ -0,0 +1,102 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines an output message queue.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog+       (UndoableLog,+        newUndoableLog,+        writeLog,+        rollbackLog,+        reduceLog,+        logSize) where++import Data.IORef++import Control.Monad+import Control.Monad.Trans++import qualified Simulation.Aivika.DoubleLinkedList as DLL+import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.Internal.Types++import Simulation.Aivika.Distributed.Optimistic.Internal.Priority+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.IO++-- | Specified an undoable log with ability to rollback the operations.+data UndoableLog =+  UndoableLog { logItems :: DLL.DoubleLinkedList UndoableItem+                -- ^ The items that can be undone.+              }++data UndoableItem =+  UndoableItem { itemTime :: Double,+                 -- ^ The time at which the operation had occured.+                 itemUndo :: DIO ()+                 -- ^ Undo the operation+               }++-- | Create an undoable log.+newUndoableLog :: DIO UndoableLog+newUndoableLog =+  do xs <- liftIOUnsafe DLL.newList+     return UndoableLog { logItems = xs }++-- | Write a new undoable operation.+writeLog :: UndoableLog -> DIO () -> Event DIO ()+{-# INLINE writeLog #-}+writeLog log h =+  Event $ \p ->+  do let x = UndoableItem { itemTime = pointTime p, itemUndo = h }+     ---+     --- logDIO DEBUG $ "Writing the log at t = " ++ show (itemTime x)+     ---+     liftIOUnsafe $+       do f <- DLL.listNull (logItems log)+          if f+            then DLL.listAddLast (logItems log) x+            else do x0 <- DLL.listLast (logItems log)+                    when (itemTime x < itemTime x0) $+                      error $+                      "The logging data are not sorted by time (" +++                      (show $ itemTime x) ++ " < " +++                      (show $ itemTime x0) ++ "): writeLog"+                    DLL.listAddLast (logItems log) x++-- | Rollback the log till the specified time either including that one or not.+rollbackLog :: UndoableLog -> Double -> Bool -> DIO ()+rollbackLog log t including =+  do ---+     --- logDIO DEBUG $ "Rolling the log back to t = " ++ show t+     ---+     loop+       where+         loop =+           do f <- liftIOUnsafe $ DLL.listNull (logItems log)+              unless f $+                do x <- liftIOUnsafe $ DLL.listLast (logItems log)+                   when ((t < itemTime x) || (including && t == itemTime x)) $+                     do liftIOUnsafe $ DLL.listRemoveLast (logItems log)+                        itemUndo x+                        loop++-- | Reduce the log removing all items older than the specified time.+reduceLog :: UndoableLog -> Double -> IO ()+reduceLog log t =+  do f <- DLL.listNull (logItems log)+     unless f $+       do x <- DLL.listFirst (logItems log)+          when (itemTime x < t) $+            do DLL.listRemoveFirst (logItems log)+               reduceLog log t++-- | Return the log size.+logSize :: UndoableLog -> IO Int+logSize log = DLL.listCount (logItems log)
+ Simulation/Aivika/Distributed/Optimistic/Message.hs view
@@ -0,0 +1,82 @@++{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE FlexibleContexts #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Message+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines functions for working with messages.+--+module Simulation.Aivika.Distributed.Optimistic.Message+       (sendMessage,+        enqueueMessage,+        messageReceived) where++import Data.Time+import Data.Monoid++import Control.Monad+import Control.Distributed.Process (ProcessId, getSelfPid, wrapMessage, unwrapMessage)+import Control.Distributed.Process.Serializable++import Simulation.Aivika.Trans hiding (ProcessId)+import Simulation.Aivika.Trans.Internal.Types++import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.IO+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.Event+import qualified Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue as IMQ+import qualified Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue as OMQ+import Simulation.Aivika.Distributed.Optimistic.DIO+import Simulation.Aivika.Distributed.Optimistic.Ref.Base++-- | Send a message to the specified remote process with the current receive time.+sendMessage :: forall a. Serializable a => ProcessId -> a -> Event DIO ()+sendMessage pid a =+  do t <- liftDynamics time+     enqueueMessage pid t a ++-- | Send a message to the specified remote process with the given receive time.+enqueueMessage :: forall a. Serializable a => ProcessId -> Double -> a -> Event DIO ()+enqueueMessage pid t a =+  Event $ \p ->+  do let queue       = queueOutputMessages $+                       runEventQueue (pointRun p)+         sendTime    = pointTime p+         receiveTime = t+     sequenceNo <- liftIOUnsafe $ OMQ.generateMessageSequenceNo queue+     sender <- messageInboxId+     let receiver = pid+         antiToggle = False+         binaryData = wrapMessage a+         message = Message { messageSequenceNo = sequenceNo,+                             messageSendTime = sendTime,+                             messageReceiveTime = receiveTime,+                             messageSenderId = sender,+                             messageReceiverId = receiver,+                             messageAntiToggle = antiToggle,+                             messageData = binaryData+                           }+     OMQ.sendMessage queue message+          +-- | The signal triggered when the remote message of the specified type has come.+messageReceived :: forall a. Serializable a => Signal DIO a+messageReceived =+  Signal { handleSignal = \h ->+            Event $ \p ->+            let queue = queueInputMessages $+                        runEventQueue (pointRun p)+                signal = IMQ.messageEnqueued queue+            in invokeEvent p $+               handleSignal signal $ \x ->+               do y <- unwrapMessage (messageData x)+                  case y of+                    Nothing -> return ()+                    Just a  -> h a+         }
+ Simulation/Aivika/Distributed/Optimistic/Priority.hs view
@@ -0,0 +1,15 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Priority+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines the logging 'Priority'.+--+module Simulation.Aivika.Distributed.Optimistic.Priority+       (Priority(..)) where++import Simulation.Aivika.Distributed.Optimistic.Internal.Priority
+ Simulation/Aivika/Distributed/Optimistic/QueueStrategy.hs view
@@ -0,0 +1,67 @@++{-# LANGUAGE TypeFamilies, MultiParamTypeClasses #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.QueueStrategy+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines queue strategies 'FCFS' and 'LCFS' for the 'DIO' computation.+--+module Simulation.Aivika.Distributed.Optimistic.QueueStrategy () where++import Control.Monad.Trans++import Simulation.Aivika.Trans+import qualified Simulation.Aivika.Trans.DoubleLinkedList as LL+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Ref.Base++-- | An implementation of the 'FCFS' queue strategy.+instance QueueStrategy DIO FCFS where++  -- | A queue used by the 'FCFS' strategy.+  newtype StrategyQueue DIO FCFS a = FCFSQueue (LL.DoubleLinkedList DIO a)++  newStrategyQueue s = fmap FCFSQueue LL.newList++  strategyQueueNull (FCFSQueue q) = LL.listNull q++-- | An implementation of the 'FCFS' queue strategy.+instance DequeueStrategy DIO FCFS where++  strategyDequeue (FCFSQueue q) =+    do i <- LL.listFirst q+       LL.listRemoveFirst q+       return i++-- | An implementation of the 'FCFS' queue strategy.+instance EnqueueStrategy DIO FCFS where++  strategyEnqueue (FCFSQueue q) i = LL.listAddLast q i++-- | An implementation of the 'LCFS' queue strategy.+instance QueueStrategy DIO LCFS where++  -- | A queue used by the 'LCFS' strategy.+  newtype StrategyQueue DIO LCFS a = LCFSQueue (LL.DoubleLinkedList DIO a)++  newStrategyQueue s = fmap LCFSQueue LL.newList+       +  strategyQueueNull (LCFSQueue q) = LL.listNull q++-- | An implementation of the 'LCFS' queue strategy.+instance DequeueStrategy DIO LCFS where++  strategyDequeue (LCFSQueue q) =+    do i <- LL.listFirst q+       LL.listRemoveFirst q+       return i++-- | An implementation of the 'LCFS' queue strategy.+instance EnqueueStrategy DIO LCFS where++  strategyEnqueue (LCFSQueue q) i = LL.listInsertFirst q i
+ Simulation/Aivika/Distributed/Optimistic/Ref/Base.hs view
@@ -0,0 +1,48 @@++{-# LANGUAGE TypeFamilies #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Ref.Base+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- Here is an implementation of mutable references, where+-- 'DIO' is an instance of 'MonadRef' and 'MonadRef0'.+--+module Simulation.Aivika.Distributed.Optimistic.Ref.Base () where++import Simulation.Aivika.Trans.Internal.Types+import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.Ref.Base++import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref as R++-- | The implementation of mutable references.+instance MonadRef DIO where++  -- | The mutable reference.+  newtype Ref DIO a = Ref { refValue :: R.Ref a }++  {-# INLINE newRef #-}+  newRef = fmap Ref . R.newRef ++  {-# INLINE readRef #-}+  readRef (Ref r) = R.readRef r++  {-# INLINE writeRef #-}+  writeRef (Ref r) = R.writeRef r++  {-# INLINE modifyRef #-}+  modifyRef (Ref r) = R.modifyRef r++  {-# INLINE equalRef #-}+  equalRef (Ref r1) (Ref r2) = (r1 == r2)++instance MonadRef0 DIO where++  {-# INLINE newRef0 #-}+  newRef0 = fmap Ref . R.newRef0
+ Simulation/Aivika/Distributed/Optimistic/TimeServer.hs view
@@ -0,0 +1,17 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.TimeServer+-- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module allows running the time server that coordinates the global simulation time.+--+module Simulation.Aivika.Distributed.Optimistic.TimeServer+       (TimeServerParams(..),+        defaultTimeServerParams,+        timeServer) where++import Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer
+ aivika-distributed.cabal view
@@ -0,0 +1,74 @@+name:            aivika-distributed+version:         0.1+synopsis:        Parallel distributed simulation library+description:+    This package extends the Aivika library with facilities for running parallel distributed simulations.+    It uses an optimistic strategy known as the Time Warp method.+    .+category:        Simulation+license:         BSD3+license-file:    LICENSE+copyright:       (c) 2015-2016. David Sorokin <david.sorokin@gmail.com>+author:          David Sorokin+maintainer:      David Sorokin <david.sorokin@gmail.com>+homepage:        http://www.aivikasoft.com/en/products/aivika.html+cabal-version:   >= 1.6+build-type:      Simple+tested-with:     GHC == 7.10.1++extra-source-files:  tests/MachRep1.hs+                     tests/MachRep2.hs+                     tests/MachRep2Reproducible.hs++library++    exposed-modules: Simulation.Aivika.Distributed+                     Simulation.Aivika.Distributed.Optimistic+                     Simulation.Aivika.Distributed.Optimistic.DIO+                     Simulation.Aivika.Distributed.Optimistic.Event+                     Simulation.Aivika.Distributed.Optimistic.Generator+                     Simulation.Aivika.Distributed.Optimistic.QueueStrategy+                     Simulation.Aivika.Distributed.Optimistic.Message+                     Simulation.Aivika.Distributed.Optimistic.Priority+                     Simulation.Aivika.Distributed.Optimistic.Ref.Base+                     Simulation.Aivika.Distributed.Optimistic.TimeServer++    other-modules:   Simulation.Aivika.Distributed.Optimistic.Internal.Channel+                     Simulation.Aivika.Distributed.Optimistic.Internal.DIO+                     Simulation.Aivika.Distributed.Optimistic.Internal.Event+                     Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue+                     Simulation.Aivika.Distributed.Optimistic.Internal.IO+                     Simulation.Aivika.Distributed.Optimistic.Internal.Message+                     Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue+                     Simulation.Aivika.Distributed.Optimistic.Internal.Priority+                     Simulation.Aivika.Distributed.Optimistic.Internal.Ref+                     Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer+                     Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp+                     Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog++    build-depends:   base >= 3 && < 6,+                     mtl >= 2.1.1,+                     stm >= 2.4.2,+                     random >= 1.0.0.3,+                     bytestring >= 0.10.4.0,+                     binary >= 0.6.4.0,+                     time >= 1.5.0.1,+                     containers >= 0.4.0.0,+                     exceptions >= 0.8.0.2,+                     distributed-process >= 0.6.1,+                     aivika >= 4.3.4,+                     aivika-transformers >= 4.3.4++    extensions:      MultiParamTypeClasses,+                     FlexibleInstances,+                     FlexibleContexts,+                     TypeFamilies,+                     RankNTypes,+                     DeriveGeneric++    ghc-options:     -O2++source-repository head++    type:     git+    location: https://github.com/dsorokin/aivika-distributed
+ tests/MachRep1.hs view
@@ -0,0 +1,158 @@++{--# LANGUAGE TemplateHaskell #--}+{-# LANGUAGE DeriveGeneric #-}++-- It corresponds to model MachRep1 described in document +-- Introduction to Discrete-Event Simulation and the SimPy Language+-- [http://heather.cs.ucdavis.edu/~matloff/156/PLN/DESimIntro.pdf]. +-- SimPy is available on [http://simpy.sourceforge.net/].+--   +-- The model description is as follows.+--+-- Two machines, which sometimes break down.+-- Up time is exponentially distributed with mean 1.0, and repair time is+-- exponentially distributed with mean 0.5. There are two repairpersons,+-- so the two machines can be repaired simultaneously if they are down+-- at the same time.+--+-- Output is long-run proportion of up time. Should get value of about+-- 0.66.++{-+ghc -threaded MachRep1Distributed.hs++Fire up some slave nodes (for the example, we run them on a single machine):++./MachRep1Distributed slave localhost 8080 &+./MachRep1Distributed slave localhost 8081 &++And start the master node:++./MachRep1Distributed master localhost 8084+-}++import System.Environment (getArgs)++import Data.Typeable+import Data.Binary++import GHC.Generics++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Distributed.Process.Backend.SimpleLocalnet++import Simulation.Aivika.Trans+import Simulation.Aivika.Distributed++meanUpTime = 1.0+meanRepairTime = 0.5++specs = Specs { spcStartTime = 0.0,+                spcStopTime = 10000.0,+                spcDT = 1.0,+                spcMethod = RungeKutta4,+                spcGeneratorType = SimpleGenerator }++newtype TotalUpTimeChange = TotalUpTimeChange { runTotalUpTimeChange :: Double }+                          deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary TotalUpTimeChange++-- | A sub-model.+slaveModel :: DP.ProcessId -> Simulation DIO ()+slaveModel masterId =+  do let machine =+           do upTime <-+                liftParameter $+                randomExponential meanUpTime+              holdProcess upTime+              ---+              liftEvent $+                sendMessage masterId (TotalUpTimeChange upTime)+              ---+              repairTime <-+                liftParameter $+                randomExponential meanRepairTime+              holdProcess repairTime+              machine++     runProcessInStartTime machine++     syncEventInStopTime $+       liftIO $ putStrLn "The sub-model finished"++-- | The main model.       +masterModel :: Int -> Simulation DIO Double+masterModel n =+  do totalUpTime <- newRef 0.0++     ---+     let totalUpTimeChanged :: Signal DIO TotalUpTimeChange+         totalUpTimeChanged = messageReceived++     runEventInStartTime $+       handleSignal totalUpTimeChanged $ \x ->+       modifyRef totalUpTime (+ runTotalUpTimeChange x)+     ---+     +     let upTimeProp =+           do x <- readRef totalUpTime+              y <- liftDynamics time+              return $ x / (fromIntegral n * y)++     runEventInStopTime upTimeProp++runSlaveModel :: (DP.ProcessId, DP.ProcessId) -> DP.Process (DP.ProcessId, DP.Process ())+runSlaveModel (timeServerId, masterId) =+  runDIO m ps timeServerId+  where+    ps = defaultDIOParams { dioLoggingPriority = NOTICE }+    m  = do runSimulation (slaveModel masterId) specs+            unregisterDIO++-- remotable ['runSlaveModel, 'timeServer]++runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process Double)+runMasterModel timeServerId n =+  runDIO m ps timeServerId+  where+    ps = defaultDIOParams { dioLoggingPriority = NOTICE }+    m  = do a <- runSimulation (masterModel n) specs+            terminateDIO+            return a++master = \backend nodes ->+  do liftIO . putStrLn $ "Slaves: " ++ show nodes+     timeServerId <- DP.spawnLocal $ timeServer defaultTimeServerParams+     -- timeServerId <- DP.spawn node1 ($(mkClosure 'timeServer) defaultTimeServerParams)+     (masterId, masterProcess) <- runMasterModel timeServerId 2+     -- (masterId, masterProcess) <- runMasterModel timeServerId (length nodes)+     -- forM_ nodes $ \node ->+     --   DP.spawn node ($(mkClosure 'runSlaveModel) (timeServerId, masterId))+     forM_ [1..2] $ \i ->+       do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId)+          DP.spawnLocal slaveProcess+     a <- masterProcess+     DP.say $+       "The result is " ++ show a+  +main :: IO ()+main = do+  backend <- initializeBackend "localhost" "8080" initRemoteTable+  startMaster backend (master backend)++-- main :: IO ()+-- main = do+--   args <- getArgs+--   case args of+--     ["master", host, port] -> do+--       backend <- initializeBackend host port initRemoteTable+--       startMaster backend (master backend)+--     ["slave", host, port] -> do+--       backend <- initializeBackend host port initRemoteTable+--       startSlave backend
+ tests/MachRep2.hs view
@@ -0,0 +1,230 @@++{--# LANGUAGE TemplateHaskell #--}+{-# LANGUAGE DeriveGeneric #-}++-- It corresponds to model MachRep2 described in document +-- Introduction to Discrete-Event Simulation and the SimPy Language+-- [http://heather.cs.ucdavis.edu/~matloff/156/PLN/DESimIntro.pdf]. +-- SimPy is available on [http://simpy.sourceforge.net/].+--   +-- The model description is as follows.+--   +-- Two machines, but sometimes break down. Up time is exponentially +-- distributed with mean 1.0, and repair time is exponentially distributed +-- with mean 0.5. In this example, there is only one repairperson, so +-- the two machines cannot be repaired simultaneously if they are down +-- at the same time.+--+-- In addition to finding the long-run proportion of up time as in+-- model MachRep1, let’s also find the long-run proportion of the time +-- that a given machine does not have immediate access to the repairperson +-- when the machine breaks down. Output values should be about 0.6 and 0.67. ++-- import System.Environment (getArgs)++import Data.Typeable+import Data.Binary++import GHC.Generics++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Distributed.Process.Backend.SimpleLocalnet++import Simulation.Aivika.Trans+import Simulation.Aivika.Distributed++meanUpTime = 1.0+meanRepairTime = 0.5++specs = Specs { spcStartTime = 0.0,+                spcStopTime = 1000.0,+                spcDT = 1.0,+                spcMethod = RungeKutta4,+                spcGeneratorType = SimpleGenerator }++-- | The time shift when replying the messages.+delta = 0.01++data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic)+data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NRepChange = NRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NRepChangeResp = NRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NImmedRepChange = NImmedRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NImmedRepChangeResp = NImmedRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data RepairPersonCount = RepairPersonCount DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RepairPersonCountResp = RepairPersonCountResp (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)++data RequestRepairPerson = RequestRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RequestRepairPersonResp = RequestRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data ReleaseRepairPerson = ReleaseRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data ReleaseRepairPersonResp = ReleaseRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary TotalUpTimeChange+instance Binary TotalUpTimeChangeResp++instance Binary NRepChange+instance Binary NRepChangeResp++instance Binary NImmedRepChange+instance Binary NImmedRepChangeResp++instance Binary RepairPersonCount+instance Binary RepairPersonCountResp++instance Binary RequestRepairPerson+instance Binary RequestRepairPersonResp++instance Binary ReleaseRepairPerson+instance Binary ReleaseRepairPersonResp++-- | A sub-model.+slaveModel :: DP.ProcessId -> Simulation DIO ()+slaveModel masterId =+  do inboxId <- liftComp messageInboxId++     let machine =+           do t <- liftDynamics time+              upTime <-+                liftParameter $ randomExponential meanUpTime+              enqueueMessage masterId (t + delta + upTime) (TotalUpTimeChange (inboxId, upTime))+              enqueueMessage masterId (t + delta + upTime) (NRepChange (inboxId, 1))+              enqueueMessage masterId (t + delta + upTime) (RepairPersonCount inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCountResp (senderId, n)) ->+       do t <- liftDynamics time+          when (n == 1) $+            enqueueMessage masterId (t + delta) (NImmedRepChange (inboxId, 1))+          enqueueMessage masterId (t + delta) (RequestRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPersonResp senderId) ->+       do t <- liftDynamics time+          repairTime <-+            liftParameter $ randomExponential meanRepairTime+          enqueueMessage masterId (t + delta + repairTime) (ReleaseRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPersonResp senderId) ->+       machine+  +     runEventInStartTime machine++     syncEventInStopTime $+       liftIO $ putStrLn "The sub-model finished"++-- | The main model.       +masterModel :: Int -> Simulation DIO (Double, Double)+masterModel count =+  do totalUpTime <- newRef 0.0+     nRep <- newRef 0+     nImmedRep <- newRef 0++     let maxRepairPersonCount = 1+     repairPerson <- newFCFSResource maxRepairPersonCount++     inboxId <- liftComp messageInboxId++     runEventInStartTime $+       handleSignal messageReceived $ \(TotalUpTimeChange (senderId, x)) ->+       do modifyRef totalUpTime (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (TotalUpTimeChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NRepChange (senderId, x)) ->+       do modifyRef nRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NImmedRepChange (senderId, x)) ->+       do modifyRef nImmedRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NImmedRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCount senderId) ->+       do n <- resourceCount repairPerson+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (RepairPersonCountResp (inboxId, n))++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPerson senderId) ->+       runProcess $+       do requestResource repairPerson+          t <- liftDynamics time+          liftEvent $+            enqueueMessage senderId (t + delta) (RequestRepairPersonResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPerson senderId) ->+       do t <- liftDynamics time+          releaseResourceWithinEvent repairPerson+          enqueueMessage senderId (t + delta) (ReleaseRepairPersonResp inboxId)+          +     let upTimeProp =+           do x <- readRef totalUpTime+              y <- liftDynamics time+              return $ x / (fromIntegral count * y)++         immedProp =+           do n <- readRef nRep+              nImmed <- readRef nImmedRep+              return $+                fromIntegral nImmed /+                fromIntegral n++     runEventInStopTime $+       do x <- upTimeProp+          y <- immedProp+          return (x, y)++runSlaveModel :: (DP.ProcessId, DP.ProcessId) -> DP.Process (DP.ProcessId, DP.Process ())+runSlaveModel (timeServerId, masterId) =+  runDIO m ps timeServerId+  where+    ps = defaultDIOParams { dioLoggingPriority = WARNING }+    m  = do runSimulation (slaveModel masterId) specs+            unregisterDIO++-- remotable ['runSlaveModel, 'timeServer]++runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process (Double, Double))+runMasterModel timeServerId n =+  runDIO m ps timeServerId+  where+    ps = defaultDIOParams { dioLoggingPriority = WARNING }+    m  = do a <- runSimulation (masterModel n) specs+            terminateDIO+            return a++master = \backend nodes ->+  do liftIO . putStrLn $ "Slaves: " ++ show nodes+     let n = 2+     timeServerId <- DP.spawnLocal $ timeServer defaultTimeServerParams+     -- timeServerId <- DP.spawn node1 ($(mkClosure 'timeServer) defaultTimeServerParams)+     (masterId, masterProcess) <- runMasterModel timeServerId n+     -- (masterId, masterProcess) <- runMasterModel timeServerId (length nodes)+     -- forM_ nodes $ \node ->+     --   DP.spawn node ($(mkClosure 'runSlaveModel) (timeServerId, masterId))+     forM_ [1..n] $ \i ->+       do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId)+          DP.spawnLocal slaveProcess+     a <- masterProcess+     DP.say $+       "The result is " ++ show a+  +main :: IO ()+main = do+  backend <- initializeBackend "localhost" "8080" initRemoteTable+  startMaster backend (master backend)
+ tests/MachRep2Reproducible.hs view
@@ -0,0 +1,246 @@++{--# LANGUAGE TemplateHaskell #--}+{-# LANGUAGE DeriveGeneric #-}++-- It corresponds to model MachRep2 described in document +-- Introduction to Discrete-Event Simulation and the SimPy Language+-- [http://heather.cs.ucdavis.edu/~matloff/156/PLN/DESimIntro.pdf]. +-- SimPy is available on [http://simpy.sourceforge.net/].+--   +-- The model description is as follows.+--   +-- Two machines, but sometimes break down. Up time is exponentially +-- distributed with mean 1.0, and repair time is exponentially distributed +-- with mean 0.5. In this example, there is only one repairperson, so +-- the two machines cannot be repaired simultaneously if they are down +-- at the same time.+--+-- In addition to finding the long-run proportion of up time as in+-- model MachRep1, let’s also find the long-run proportion of the time +-- that a given machine does not have immediate access to the repairperson +-- when the machine breaks down. Output values should be about 0.6 and 0.67. ++-- import System.Environment (getArgs)++import Data.Typeable+import Data.Binary++import System.Random++import GHC.Generics++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Distributed.Process.Backend.SimpleLocalnet++import Simulation.Aivika.Trans+import Simulation.Aivika.Distributed++meanUpTime = 1.0+meanRepairTime = 0.5++specs = Specs { spcStartTime = 0.0,+                spcStopTime = 1000.0,+                spcDT = 1.0,+                spcMethod = RungeKutta4,+                spcGeneratorType = SimpleGenerator }++-- | The time shift when replying the messages.+delta = 0.01++-- | The initial seeds.+seeds = [456, 789]++-- | Create a new random generator by the specified seed.+newRandomRef :: Int -> Simulation DIO (Ref DIO StdGen)+newRandomRef = newRef . mkStdGen++-- | Generate a random number with the specified mean+randomRefExponential :: RandomGen g => Ref DIO g -> Double -> Event DIO Double+randomRefExponential r mu =+  do g <- readRef r+     let (x, g') = random g+     writeRef r g'+     return (- log x * mu)++data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic)+data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NRepChange = NRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NRepChangeResp = NRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NImmedRepChange = NImmedRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NImmedRepChangeResp = NImmedRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data RepairPersonCount = RepairPersonCount DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RepairPersonCountResp = RepairPersonCountResp (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)++data RequestRepairPerson = RequestRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RequestRepairPersonResp = RequestRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data ReleaseRepairPerson = ReleaseRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data ReleaseRepairPersonResp = ReleaseRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary TotalUpTimeChange+instance Binary TotalUpTimeChangeResp++instance Binary NRepChange+instance Binary NRepChangeResp++instance Binary NImmedRepChange+instance Binary NImmedRepChangeResp++instance Binary RepairPersonCount+instance Binary RepairPersonCountResp++instance Binary RequestRepairPerson+instance Binary RequestRepairPersonResp++instance Binary ReleaseRepairPerson+instance Binary ReleaseRepairPersonResp++-- | A sub-model.+slaveModel :: DP.ProcessId -> Int -> Simulation DIO ()+slaveModel masterId i =+  do inboxId <- liftComp messageInboxId+     g <- newRandomRef $ seeds !! (i - 1)++     let machine =+           do t <- liftDynamics time+              upTime <- randomRefExponential g meanUpTime+              enqueueMessage masterId (t + delta + upTime) (TotalUpTimeChange (inboxId, upTime))+              enqueueMessage masterId (t + delta + upTime) (NRepChange (inboxId, 1))+              enqueueMessage masterId (t + delta + upTime) (RepairPersonCount inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCountResp (senderId, n)) ->+       do t <- liftDynamics time+          when (n == 1) $+            enqueueMessage masterId (t + delta) (NImmedRepChange (inboxId, 1))+          enqueueMessage masterId (t + delta) (RequestRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPersonResp senderId) ->+       do t <- liftDynamics time+          repairTime <- randomRefExponential g meanRepairTime+          enqueueMessage masterId (t + delta + repairTime) (ReleaseRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPersonResp senderId) ->+       machine+  +     runEventInStartTime machine++     syncEventInStopTime $+       liftIO $ putStrLn "The sub-model finished"++-- | The main model.       +masterModel :: Int -> Simulation DIO (Double, Double)+masterModel count =+  do totalUpTime <- newRef 0.0+     nRep <- newRef 0+     nImmedRep <- newRef 0++     let maxRepairPersonCount = 1+     repairPerson <- newFCFSResource maxRepairPersonCount++     inboxId <- liftComp messageInboxId++     runEventInStartTime $+       handleSignal messageReceived $ \(TotalUpTimeChange (senderId, x)) ->+       do modifyRef totalUpTime (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (TotalUpTimeChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NRepChange (senderId, x)) ->+       do modifyRef nRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NImmedRepChange (senderId, x)) ->+       do modifyRef nImmedRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NImmedRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCount senderId) ->+       do n <- resourceCount repairPerson+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (RepairPersonCountResp (inboxId, n))++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPerson senderId) ->+       runProcess $+       do requestResource repairPerson+          t <- liftDynamics time+          liftEvent $+            enqueueMessage senderId (t + delta) (RequestRepairPersonResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPerson senderId) ->+       do t <- liftDynamics time+          releaseResourceWithinEvent repairPerson+          enqueueMessage senderId (t + delta) (ReleaseRepairPersonResp inboxId)+          +     let upTimeProp =+           do x <- readRef totalUpTime+              y <- liftDynamics time+              return $ x / (fromIntegral count * y)++         immedProp =+           do n <- readRef nRep+              nImmed <- readRef nImmedRep+              return $+                fromIntegral nImmed /+                fromIntegral n++     runEventInStopTime $+       do x <- upTimeProp+          y <- immedProp+          return (x, y)++runSlaveModel :: (DP.ProcessId, DP.ProcessId, Int) -> DP.Process (DP.ProcessId, DP.Process ())+runSlaveModel (timeServerId, masterId, i) =+  runDIO m ps timeServerId+  where+    ps = defaultDIOParams { dioLoggingPriority = WARNING }+    m  = do runSimulation (slaveModel masterId i) specs+            unregisterDIO++-- remotable ['runSlaveModel, 'timeServer]++runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process (Double, Double))+runMasterModel timeServerId n =+  runDIO m ps timeServerId+  where+    ps = defaultDIOParams { dioLoggingPriority = WARNING }+    m  = do a <- runSimulation (masterModel n) specs+            terminateDIO+            return a++master = \backend nodes ->+  do liftIO . putStrLn $ "Slaves: " ++ show nodes+     let n = length seeds+     timeServerId <- DP.spawnLocal $ timeServer defaultTimeServerParams+     -- timeServerId <- DP.spawn node1 ($(mkClosure 'timeServer) defaultTimeServerParams)+     (masterId, masterProcess) <- runMasterModel timeServerId n+     -- (masterId, masterProcess) <- runMasterModel timeServerId (length nodes)+     -- forM_ nodes $ \node ->+     --   DP.spawn node ($(mkClosure 'runSlaveModel) (timeServerId, masterId))+     forM_ [1..n] $ \i ->+       do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId, i)+          DP.spawnLocal slaveProcess+     a <- masterProcess+     DP.say $+       "The result is " ++ show a+  +main :: IO ()+main = do+  backend <- initializeBackend "localhost" "8080" initRemoteTable+  startMaster backend (master backend)