diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2015-2016 David Sorokin + +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. diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,3 @@ +#!/usr/bin/env runhaskell +> import Distribution.Simple +> main = defaultMain diff --git a/Simulation/Aivika/Distributed.hs b/Simulation/Aivika/Distributed.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed.hs @@ -0,0 +1,17 @@ + +-- | +-- Module : Simulation.Aivika.Distributed +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic.hs b/Simulation/Aivika/Distributed/Optimistic.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic.hs @@ -0,0 +1,31 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/DIO.hs b/Simulation/Aivika/Distributed/Optimistic/DIO.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/DIO.hs @@ -0,0 +1,52 @@ + +{-# LANGUAGE MultiParamTypeClasses, FlexibleInstances #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.DIO +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 + diff --git a/Simulation/Aivika/Distributed/Optimistic/Event.hs b/Simulation/Aivika/Distributed/Optimistic/Event.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Event.hs @@ -0,0 +1,34 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Event +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 () diff --git a/Simulation/Aivika/Distributed/Optimistic/Generator.hs b/Simulation/Aivika/Distributed/Optimistic/Generator.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Generator.hs @@ -0,0 +1,127 @@ + +{-# LANGUAGE TypeFamilies #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Generator +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Channel.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Channel.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Channel.hs @@ -0,0 +1,76 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Channel +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs @@ -0,0 +1,216 @@ + +{-# LANGUAGE DeriveGeneric #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.DIO +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs @@ -0,0 +1,709 @@ + +{-# LANGUAGE TypeFamilies, FlexibleInstances #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Event +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/IO.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/IO.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/IO.hs @@ -0,0 +1,42 @@ + +{-# LANGUAGE FlexibleInstances #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.IO +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs @@ -0,0 +1,387 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 + diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs-boot b/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs-boot new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs-boot @@ -0,0 +1,49 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 () diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs @@ -0,0 +1,102 @@ + +{-# LANGUAGE RankNTypes, DeriveGeneric #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Message +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 + diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs @@ -0,0 +1,135 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs-boot b/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs-boot new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs-boot @@ -0,0 +1,37 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 () diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Priority.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Priority.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Priority.hs @@ -0,0 +1,40 @@ + +{-# LANGUAGE DeriveGeneric #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Priority +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 ++ "]" diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs @@ -0,0 +1,73 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Ref +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs-boot b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs-boot new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs-boot @@ -0,0 +1,29 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Ref +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 () diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs @@ -0,0 +1,247 @@ + +{-# LANGUAGE DeriveGeneric #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 + diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/TimeWarp.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeWarp.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeWarp.hs @@ -0,0 +1,25 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs @@ -0,0 +1,102 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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) diff --git a/Simulation/Aivika/Distributed/Optimistic/Message.hs b/Simulation/Aivika/Distributed/Optimistic/Message.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Message.hs @@ -0,0 +1,82 @@ + +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE FlexibleContexts #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Message +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 + } diff --git a/Simulation/Aivika/Distributed/Optimistic/Priority.hs b/Simulation/Aivika/Distributed/Optimistic/Priority.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Priority.hs @@ -0,0 +1,15 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Priority +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/QueueStrategy.hs b/Simulation/Aivika/Distributed/Optimistic/QueueStrategy.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/QueueStrategy.hs @@ -0,0 +1,67 @@ + +{-# LANGUAGE TypeFamilies, MultiParamTypeClasses #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.QueueStrategy +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/Ref/Base.hs b/Simulation/Aivika/Distributed/Optimistic/Ref/Base.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Ref/Base.hs @@ -0,0 +1,48 @@ + +{-# LANGUAGE TypeFamilies #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Ref.Base +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/TimeServer.hs b/Simulation/Aivika/Distributed/Optimistic/TimeServer.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/TimeServer.hs @@ -0,0 +1,17 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.TimeServer +-- Copyright : Copyright (c) 2015-2016, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/aivika-distributed.cabal b/aivika-distributed.cabal new file mode 100644 --- /dev/null +++ b/aivika-distributed.cabal @@ -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 +author: David Sorokin +maintainer: David Sorokin +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 diff --git a/tests/MachRep1.hs b/tests/MachRep1.hs new file mode 100644 --- /dev/null +++ b/tests/MachRep1.hs @@ -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 diff --git a/tests/MachRep2.hs b/tests/MachRep2.hs new file mode 100644 --- /dev/null +++ b/tests/MachRep2.hs @@ -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) diff --git a/tests/MachRep2Reproducible.hs b/tests/MachRep2Reproducible.hs new file mode 100644 --- /dev/null +++ b/tests/MachRep2Reproducible.hs @@ -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)