diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,13 @@ +Version 0.5 +----- + +* Added an ability to restore the distributed simulation after temporary connection errors. + +* Better finalisation of the distributed simulation. + +* Implemented lazy references. + Version 0.3 ----- diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2015-2016 David Sorokin +Copyright (c) 2015-2017 David Sorokin All rights reserved. diff --git a/Simulation/Aivika/Distributed.hs b/Simulation/Aivika/Distributed.hs --- a/Simulation/Aivika/Distributed.hs +++ b/Simulation/Aivika/Distributed.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/Simulation/Aivika/Distributed/Optimistic.hs b/Simulation/Aivika/Distributed/Optimistic.hs --- a/Simulation/Aivika/Distributed/Optimistic.hs +++ b/Simulation/Aivika/Distributed/Optimistic.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/Simulation/Aivika/Distributed/Optimistic/DIO.hs b/Simulation/Aivika/Distributed/Optimistic/DIO.hs --- a/Simulation/Aivika/Distributed/Optimistic/DIO.hs +++ b/Simulation/Aivika/Distributed/Optimistic/DIO.hs @@ -3,7 +3,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.DIO --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -22,7 +22,10 @@ logDIO, terminateDIO, registerDIO, - unregisterDIO) where + unregisterDIO, + monitorProcessDIO, + InboxProcessMessage(MonitorProcessMessage), + processMonitorSignal) where import Control.Monad import Control.Monad.Trans @@ -32,16 +35,19 @@ import Simulation.Aivika.Trans.Exception import Simulation.Aivika.Trans.Generator import Simulation.Aivika.Trans.Event +import Simulation.Aivika.Trans.Composite import Simulation.Aivika.Trans.Process import Simulation.Aivika.Trans.Ref.Base import Simulation.Aivika.Trans.QueueStrategy +import Simulation.Aivika.Distributed.Optimistic.Internal.Message 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.Ref.Base.Lazy +import Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict import Simulation.Aivika.Distributed.Optimistic.QueueStrategy instance MonadDES DIO @@ -49,4 +55,7 @@ instance MonadComp DIO instance {-# OVERLAPPING #-} MonadIO (Process DIO) where + liftIO = liftEvent . liftIO + +instance {-# OVERLAPPING #-} MonadIO (Composite DIO) where liftIO = liftEvent . liftIO diff --git a/Simulation/Aivika/Distributed/Optimistic/Generator.hs b/Simulation/Aivika/Distributed/Optimistic/Generator.hs --- a/Simulation/Aivika/Distributed/Optimistic/Generator.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Generator.hs @@ -3,7 +3,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Generator --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -32,8 +32,10 @@ data Generator DIO = Generator { generator01 :: DIO Double, -- ^ the generator of uniform numbers from 0 to 1 - generatorNormal01 :: DIO Double + generatorNormal01 :: DIO Double, -- ^ the generator of normal numbers with mean 0 and variance 1 + generatorSequenceNo :: DIO Int + -- ^ the generator of sequence numbers } generateUniform = generateUniform01 . generator01 @@ -62,6 +64,8 @@ generateDiscrete = generateDiscrete01 . generator01 + generateSequenceNo = generatorSequenceNo + newGenerator tp = case tp of SimpleGenerator -> @@ -83,8 +87,14 @@ newRandomGenerator01 g01 = do gNormal01 <- newNormalGenerator01 g01 + gSeqNoRef <- liftIOUnsafe $ newIORef 0 + let gSeqNo = + do x <- liftIOUnsafe $ readIORef gSeqNoRef + liftIOUnsafe $ modifyIORef' gSeqNoRef (+1) + return x return Generator { generator01 = g01, - generatorNormal01 = gNormal01 } + generatorNormal01 = gNormal01, + generatorSequenceNo = gSeqNo } -- | Create a normal random number generator with mean 0 and variance 1 -- by the specified generator of uniform random numbers from 0 to 1. diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Channel.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Channel.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/Channel.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Channel.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Channel --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -51,7 +51,7 @@ do empty <- readIORef (channelListEmptyIO ch) if empty then return [] - else do writeIORef (channelListEmptyIO ch) True + else do atomicWriteIORef (channelListEmptyIO ch) True xs <- atomically $ do xs <- readTVar (channelList ch) writeTVar (channelList ch) [] @@ -66,7 +66,7 @@ do xs <- readTVar (channelList ch) writeTVar (channelList ch) (a : xs) writeTVar (channelListEmpty ch) False - writeIORef (channelListEmptyIO ch) False + atomicWriteIORef (channelListEmptyIO ch) False -- | Wait for data in the channel. awaitChannel :: Channel a -> IO () diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs @@ -3,7 +3,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.DIO --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -20,15 +20,23 @@ terminateDIO, registerDIO, unregisterDIO, + monitorProcessDIO, dioParams, messageChannel, messageInboxId, timeServerId, + sendMessageDIO, + sendMessagesDIO, + sendAcknowledgmentMessageDIO, + sendAcknowledgmentMessagesDIO, + sendLocalTimeDIO, + sendRequestGlobalTimeDIO, logDIO, liftDistributedUnsafe) where import Data.Typeable import Data.Binary +import Data.IORef import GHC.Generics @@ -38,7 +46,11 @@ import Control.Exception (throw) import Control.Monad.Catch as C import qualified Control.Distributed.Process as DP +import Control.Concurrent +import Control.Concurrent.STM +import System.Timeout + import Simulation.Aivika.Trans.Exception import Simulation.Aivika.Trans.Internal.Types @@ -46,6 +58,7 @@ import Simulation.Aivika.Distributed.Optimistic.Internal.Message import Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer import Simulation.Aivika.Distributed.Optimistic.Internal.Priority +import Simulation.Aivika.Distributed.Optimistic.Internal.KeepAliveManager -- | The parameters for the 'DIO' computation. data DIOParams = @@ -55,12 +68,27 @@ -- ^ The undoable log size threshold used for detecting an overflow dioOutputMessageQueueSizeThreshold :: Int, -- ^ The output message queue size threshold used for detecting an overflow + dioTransientMessageQueueSizeThreshold :: Int, + -- ^ The transient message queue size threshold used for detecting an overflow dioSyncTimeout :: Int, - -- ^ The timeout in microseconds used for synchronising the operations. + -- ^ 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 + dioAllowSkippingOutdatedMessage :: Bool, + -- ^ Whether to allow skipping an outdated message with the receive time less than the global time, + -- which is possible after reconnection + dioProcessMonitoringEnabled :: Bool, + -- ^ Whether the process monitoring is enabled + dioProcessMonitoringDelay :: Int, + -- ^ The delay in microseconds which must be applied for monitoring every remote process. + dioProcessReconnectingEnabled :: Bool, + -- ^ Whether the automatic reconnecting to processes is enabled when enabled monitoring + dioProcessReconnectingDelay :: Int, + -- ^ The timeout in microseconds before reconnecting to the remote process + dioKeepAliveInterval :: Int, + -- ^ The interval in microseconds for sending keep-alive messages + dioTimeServerAcknowledgmentTimeout :: Int + -- ^ The timeout in microseconds for receiving an acknowledgment message from the time server } deriving (Eq, Ord, Show, Typeable, Generic) instance Binary DIOParams @@ -78,8 +106,14 @@ -- ^ The inbox process identifier. dioTimeServerId :: DP.ProcessId, -- ^ The time server process - dioParams0 :: DIOParams + dioParams0 :: DIOParams, -- ^ The parameters of the computation. + dioRegisteredInTimeServer :: TVar Bool, + -- ^ Whether the compution is registered in the time server. + dioUnregisteredFromTimeServer :: TVar Bool, + -- ^ Whether the compution is unregistered from the time server. + dioTimeServerTerminating :: TVar Bool + -- ^ Whether the compution asked to terminate the time server. } instance Monad DIO where @@ -131,11 +165,22 @@ DIOParams { dioLoggingPriority = WARNING, dioUndoableLogSizeThreshold = 1000000, dioOutputMessageQueueSizeThreshold = 10000, + dioTransientMessageQueueSizeThreshold = 10000, dioSyncTimeout = 60000000, dioAllowPrematureIO = False, - dioAllowProcessingOutdatedMessage = False + dioAllowSkippingOutdatedMessage = True, + dioProcessMonitoringEnabled = False, + dioProcessMonitoringDelay = 5000000, + dioProcessReconnectingEnabled = False, + dioProcessReconnectingDelay = 5000000, + dioKeepAliveInterval = 5000000, + dioTimeServerAcknowledgmentTimeout = 5000000 } +-- | Return the computation context. +dioContext :: DIO DIOContext +dioContext = DIO return + -- | Return the parameters of the current computation. dioParams :: DIO DIOParams dioParams = DIO $ return . dioParams0 @@ -156,53 +201,345 @@ -- 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) + DIO $ \ctx -> + do let ps = dioParams0 ctx + logProcess ps INFO "Terminating the simulation..." + sender <- invokeDIO ctx messageInboxId + receiver <- invokeDIO ctx timeServerId + let inbox = sender + if dioProcessMonitoringEnabled ps + then DP.send inbox (SendTerminateTimeServerMessage receiver sender) + else DP.send receiver (TerminateTimeServerMessage sender) + liftIO $ + timeout (dioTimeServerAcknowledgmentTimeout ps) $ + atomically $ + do f <- readTVar (dioTimeServerTerminating ctx) + unless f retry + DP.send inbox TerminateInboxProcessMessage + return () -- | Register the simulation process in the time server, which -- requires some initial quorum to start synchronizing the global time. registerDIO :: DIO () registerDIO = - do logDIO INFO "Registering the simulation process..." - sender <- messageInboxId - receiver <- timeServerId - liftDistributedUnsafe $ - DP.send receiver (RegisterLocalProcessMessage sender) + DIO $ \ctx -> + do let ps = dioParams0 ctx + logProcess ps INFO "Registering the simulation process..." + sender <- invokeDIO ctx messageInboxId + receiver <- invokeDIO ctx timeServerId + let inbox = sender + if dioProcessMonitoringEnabled ps + then DP.send inbox (SendRegisterLocalProcessMessage receiver sender) + else DP.send receiver (RegisterLocalProcessMessage sender) + liftIO $ + timeout (dioTimeServerAcknowledgmentTimeout ps) $ + atomically $ + do f <- readTVar (dioRegisteredInTimeServer ctx) + unless f retry + return () -- | 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) + DIO $ \ctx -> + do let ps = dioParams0 ctx + logProcess ps INFO "Unregistering the simulation process..." + sender <- invokeDIO ctx messageInboxId + receiver <- invokeDIO ctx timeServerId + let inbox = sender + if dioProcessMonitoringEnabled ps + then DP.send inbox (SendUnregisterLocalProcessMessage receiver sender) + else DP.send receiver (UnregisterLocalProcessMessage sender) + liftIO $ + timeout (dioTimeServerAcknowledgmentTimeout ps) $ + atomically $ + do f <- readTVar (dioUnregisteredFromTimeServer ctx) + unless f retry + DP.send inbox TerminateInboxProcessMessage + return () +-- | The internal local process message. +data InternalLocalProcessMessage = InternalLocalProcessMessage LocalProcessMessage + -- ^ the local process message + | InternalProcessMonitorNotification DP.ProcessMonitorNotification + -- ^ the process monitor notification + | InternalInboxProcessMessage InboxProcessMessage + -- ^ the inbox process message + | InternalKeepAliveMessage KeepAliveMessage + -- ^ the keep alive message + +-- | Handle the specified exception. +handleException :: DIOParams -> SomeException -> DP.Process () +handleException ps e = + do --- + logProcess ps ERROR $ "Exception occured: " ++ show e + --- + C.throwM e + -- | 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 + let keepAliveParams = + KeepAliveParams { keepAliveLoggingPriority = dioLoggingPriority ps, + keepAliveInterval = dioKeepAliveInterval ps } + keepAliveManager <- liftIO $ newKeepAliveManager keepAliveParams + terminated <- liftIO $ newIORef False + registeredInTimeServer <- liftIO $ newTVarIO False + unregisteredFromTimeServer <- liftIO $ newTVarIO False + timeServerTerminating <- liftIO $ newTVarIO False + let loop = + forever $ + do let f1 :: LocalProcessMessage -> DP.Process InternalLocalProcessMessage + f1 x = return (InternalLocalProcessMessage x) + f2 :: DP.ProcessMonitorNotification -> DP.Process InternalLocalProcessMessage + f2 x = return (InternalProcessMonitorNotification x) + f3 :: InboxProcessMessage -> DP.Process InternalLocalProcessMessage + f3 x = return (InternalInboxProcessMessage x) + f4 :: KeepAliveMessage -> DP.Process InternalLocalProcessMessage + f4 x = return (InternalKeepAliveMessage x) + x <- fmap Just $ DP.receiveWait [DP.match f1, DP.match f2, DP.match f3, DP.match f4] + case x of + Nothing -> return () + Just (InternalLocalProcessMessage m) -> + liftIO $ + writeChannel ch m + Just (InternalProcessMonitorNotification m@(DP.ProcessMonitorNotification _ _ _)) -> + handleProcessMonitorNotification m ps ch + Just (InternalInboxProcessMessage m) -> + case m of + SendQueueMessage pid m -> + DP.send pid (QueueMessage m) + SendQueueMessageBulk pid ms -> + forM_ ms $ \m -> + DP.send pid (QueueMessage m) + SendAcknowledgmentQueueMessage pid m -> + DP.send pid (AcknowledgmentQueueMessage m) + SendAcknowledgmentQueueMessageBulk pid ms -> + forM_ ms $ \m -> + DP.send pid (AcknowledgmentQueueMessage m) + SendLocalTimeMessage receiver sender t -> + DP.send receiver (LocalTimeMessage sender t) + SendRequestGlobalTimeMessage receiver sender -> + DP.send receiver (RequestGlobalTimeMessage sender) + SendRegisterLocalProcessMessage receiver sender -> + DP.send receiver (RegisterLocalProcessMessage sender) + SendUnregisterLocalProcessMessage receiver sender -> + DP.send receiver (UnregisterLocalProcessMessage sender) + SendTerminateTimeServerMessage receiver sender -> + DP.send receiver (TerminateTimeServerMessage sender) + MonitorProcessMessage pid -> + do --- + logProcess ps INFO $ "Monitoring process " ++ show pid + --- + DP.monitor pid + liftIO $ + addKeepAliveReceiver keepAliveManager pid + return () + ReMonitorProcessMessage pids -> + handleReMonitorProcessMessage pids ps ch + TrySendKeepAliveMessage -> + trySendKeepAlive keepAliveManager + RegisterLocalProcessAcknowledgmentMessage pid -> + do --- + logProcess ps INFO "Registered the local process in the time server" + --- + liftIO $ + atomically $ + writeTVar registeredInTimeServer True + UnregisterLocalProcessAcknowledgmentMessage pid -> + do --- + logProcess ps INFO "Unregistered the local process from the time server" + --- + liftIO $ + atomically $ + writeTVar unregisteredFromTimeServer True + TerminateTimeServerAcknowledgmentMessage pid -> + do --- + logProcess ps INFO "Started terminating the time server" + --- + liftIO $ + atomically $ + writeTVar timeServerTerminating True + TerminateInboxProcessMessage -> + do --- + logProcess ps INFO "Terminating the inbox and keep-alive processes..." + --- + liftIO $ + atomicWriteIORef terminated True + DP.terminate + Just (InternalKeepAliveMessage m) -> + do --- + logProcess ps DEBUG $ "Received " ++ show m + --- + liftIO $ + writeChannel ch (KeepAliveLocalProcessMessage m) inboxId <- DP.spawnLocal $ - forever $ - do m <- DP.expect :: DP.Process LocalProcessMessage - liftIO $ - writeChannel ch m - when (m == TerminateLocalProcessMessage) $ + C.catch loop (handleException ps) + DP.spawnLocal $ + let loop = + do f <- liftIO $ readIORef terminated + unless f $ + do liftIO $ + threadDelay (dioKeepAliveInterval ps) + DP.send inboxId TrySendKeepAliveMessage + loop + in C.catch loop (handleException ps) + let simulation = + unDIO m DIOContext { dioChannel = ch, + dioInboxId = inboxId, + dioTimeServerId = serverId, + dioParams0 = ps, + dioRegisteredInTimeServer = registeredInTimeServer, + dioUnregisteredFromTimeServer = unregisteredFromTimeServer, + dioTimeServerTerminating = timeServerTerminating } + return (inboxId, simulation) + +-- | Handle the process monitor notification. +handleProcessMonitorNotification :: DP.ProcessMonitorNotification -> DIOParams -> Channel LocalProcessMessage -> DP.Process () +handleProcessMonitorNotification m@(DP.ProcessMonitorNotification _ pid0 reason) ps ch = + do let recv m@(DP.ProcessMonitorNotification _ _ _) = + do --- + logProcess ps WARNING $ "Received a process monitor notification " ++ show m + --- + liftIO $ + writeChannel ch (ProcessMonitorNotificationMessage m) + return m + recv m + when (dioProcessReconnectingEnabled ps && (reason == DP.DiedDisconnect)) $ + do liftIO $ + threadDelay (dioProcessReconnectingDelay ps) + let pred m@(DP.ProcessMonitorNotification _ _ reason) = reason == DP.DiedDisconnect + loop :: [DP.ProcessId] -> DP.Process [DP.ProcessId] + loop acc = + do y <- DP.receiveTimeout 0 [DP.matchIf pred recv] + case y of + Nothing -> return $ reverse acc + Just m@(DP.ProcessMonitorNotification _ pid _) -> loop (pid : acc) + pids <- loop [pid0] + --- + logProcess ps NOTICE "Begin reconnecting..." + --- + forM_ pids $ \pid -> do --- - logProcess ps INFO "Terminating the inbox process..." + logProcess ps NOTICE $ "Direct reconnecting to " ++ show pid --- - DP.terminate - return (inboxId, unDIO m DIOContext { dioChannel = ch, - dioInboxId = inboxId, - dioTimeServerId = serverId, - dioParams0 = ps }) + DP.reconnect pid + inboxId <- DP.getSelfPid + DP.spawnLocal $ + let action = + do liftIO $ + threadDelay (dioProcessMonitoringDelay ps) + --- + logProcess ps NOTICE $ "Proceed to the re-monitoring" + --- + DP.send inboxId (ReMonitorProcessMessage pids) + in C.catch action (handleException ps) + return () + +-- | Re-monitor the specified remote processes. +handleReMonitorProcessMessage :: [DP.ProcessId] -> DIOParams -> Channel LocalProcessMessage -> DP.Process () +handleReMonitorProcessMessage pids ps ch = + forM_ pids $ \pid -> + do --- + logProcess ps NOTICE $ "Re-monitoring " ++ show pid + --- + DP.monitor pid + --- + logProcess ps NOTICE $ "Writing to the channel about reconnecting to " ++ show pid + --- + liftIO $ + writeChannel ch (ReconnectProcessMessage pid) + +-- | Monitor the specified process. +monitorProcessDIO :: DP.ProcessId -> DIO () +monitorProcessDIO pid = + do ps <- dioParams + if dioProcessMonitoringEnabled ps + then do inbox <- messageInboxId + liftDistributedUnsafe $ + DP.send inbox $ + MonitorProcessMessage pid + else liftDistributedUnsafe $ + logProcess ps WARNING "Ignored the process monitoring as it was disabled in the DIO computation parameters" + +-- | Send the message. +sendMessageDIO :: DP.ProcessId -> Message -> DIO () +{-# INLINABLE sendMessageDIO #-} +sendMessageDIO pid m = + do ps <- dioParams + if dioProcessMonitoringEnabled ps + then do inbox <- messageInboxId + liftDistributedUnsafe $ + DP.send inbox (SendQueueMessage pid m) + else liftDistributedUnsafe $ + DP.send pid (QueueMessage m) + +-- | Send the bulk of messages. +sendMessagesDIO :: DP.ProcessId -> [Message] -> DIO () +{-# INLINABLE sendMessagesDIO #-} +sendMessagesDIO pid ms = + do ps <- dioParams + if dioProcessMonitoringEnabled ps + then do inbox <- messageInboxId + liftDistributedUnsafe $ + DP.send inbox (SendQueueMessageBulk pid ms) + else do forM_ ms $ \m -> + liftDistributedUnsafe $ + DP.send pid (QueueMessage m) + +-- | Send the acknowledgment message. +sendAcknowledgmentMessageDIO :: DP.ProcessId -> AcknowledgmentMessage -> DIO () +{-# INLINABLE sendAcknowledgmentMessageDIO #-} +sendAcknowledgmentMessageDIO pid m = + do ps <- dioParams + if dioProcessMonitoringEnabled ps + then do inbox <- messageInboxId + liftDistributedUnsafe $ + DP.send inbox (SendAcknowledgmentQueueMessage pid m) + else liftDistributedUnsafe $ + DP.send pid (AcknowledgmentQueueMessage m) + +-- | Send the bulk of acknowledgment messages. +sendAcknowledgmentMessagesDIO :: DP.ProcessId -> [AcknowledgmentMessage] -> DIO () +{-# INLINABLE sendAcknowledgmentMessagesDIO #-} +sendAcknowledgmentMessagesDIO pid ms = + do ps <- dioParams + if dioProcessMonitoringEnabled ps + then do inbox <- messageInboxId + liftDistributedUnsafe $ + DP.send inbox (SendAcknowledgmentQueueMessageBulk pid ms) + else do forM_ ms $ \m -> + liftDistributedUnsafe $ + DP.send pid (AcknowledgmentQueueMessage m) + +-- | Send the local time to the time server. +sendLocalTimeDIO :: DP.ProcessId -> DP.ProcessId -> Double -> DIO () +{-# INLINABLE sendLocalTimeDIO #-} +sendLocalTimeDIO receiver sender t = + do ps <- dioParams + if dioProcessMonitoringEnabled ps + then do inbox <- messageInboxId + liftDistributedUnsafe $ + DP.send inbox (SendLocalTimeMessage receiver sender t) + else liftDistributedUnsafe $ + DP.send receiver (LocalTimeMessage sender t) + +-- | Send the request for the global virtual time to the time server. +sendRequestGlobalTimeDIO :: DP.ProcessId -> DP.ProcessId -> DIO () +{-# INLINABLE sendRequestGlobalTimeDIO #-} +sendRequestGlobalTimeDIO receiver sender = + do ps <- dioParams + if dioProcessMonitoringEnabled ps + then do inbox <- messageInboxId + liftDistributedUnsafe $ + DP.send inbox (SendRequestGlobalTimeMessage receiver sender) + else liftDistributedUnsafe $ + DP.send receiver (RequestGlobalTimeMessage sender) -- | Log the message with the specified priority. logDIO :: Priority -> String -> DIO () diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs @@ -3,7 +3,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Event --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -14,7 +14,8 @@ module Simulation.Aivika.Distributed.Optimistic.Internal.Event (queueInputMessages, queueOutputMessages, - queueLog) where + queueLog, + processMonitorSignal) where import Data.Maybe import Data.IORef @@ -39,11 +40,12 @@ 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.SignalHelper import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue import Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog -import {-# SOURCE #-} qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref as R +import {-# SOURCE #-} qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict as R -- | Convert microseconds to seconds. microsecondsToSeconds :: Int -> Rational @@ -70,8 +72,10 @@ -- ^ the actual time of the event queue queueGlobalTime :: IORef Double, -- ^ the global time - queueInFind :: IORef Bool + queueInFind :: IORef Bool, -- ^ whether the queue is in find mode + queueProcessMonitorNotificationSource :: SignalSource DIO DP.ProcessMonitorNotification + -- ^ the source of process monitor notifications } newEventQueue specs = @@ -84,6 +88,7 @@ output <- newOutputMessageQueue $ enqueueTransientMessage transient input <- newInputMessageQueue log rollbackEventPre rollbackEventPost rollbackEventTime infind <- liftIOUnsafe $ newIORef False + s <- newDIOSignalSource0 return EventQueue { queueInputMessages = input, queueOutputMessages = output, queueTransientMessages = transient, @@ -92,7 +97,8 @@ queueBusy = f, queueTime = t, queueGlobalTime = gt, - queueInFind = infind } + queueInFind = infind, + queueProcessMonitorNotificationSource = s } enqueueEvent t (Event m) = Event $ \p -> @@ -272,10 +278,12 @@ do let q = runEventQueue $ pointRun p n1 <- liftIOUnsafe $ logSize (queueLog q) n2 <- liftIOUnsafe $ outputMessageQueueSize (queueOutputMessages q) + n3 <- liftIOUnsafe $ transientMessageQueueSize (queueTransientMessages q) ps <- dioParams let th1 = dioUndoableLogSizeThreshold ps th2 = dioOutputMessageQueueSizeThreshold ps - if (n1 >= th1) || (n2 >= th2) + th3 = dioTransientMessageQueueSizeThreshold ps + if (n1 >= th1) || (n2 >= th2) || (n3 >= th3) then do logDIO NOTICE $ "t = " ++ (show $ pointTime p) ++ ": detected the event overflow" @@ -286,7 +294,7 @@ throttleMessageChannel :: TimeWarp DIO () throttleMessageChannel = TimeWarp $ \p -> - do invokeEvent p requestGlobalTime + do -- invokeEvent p requestGlobalTime ch <- messageChannel dt <- fmap dioSyncTimeout dioParams liftIOUnsafe $ @@ -321,13 +329,14 @@ infind <- liftIOUnsafe $ readIORef (queueInFind q) deliverAcknowledgmentMessage (acknowledgmentMessage infind m) 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 + p' <- invokeEvent p currentEventPoint + if messageReceiveTime m < t0 + then do f <- fmap dioAllowSkippingOutdatedMessage dioParams + if f + then invokeEvent p' logOutdatedMessage + else error "Received the outdated message: processChannelMessage" + else invokeTimeWarp p' $ + enqueueMessage (queueInputMessages q) m processChannelMessage x@(QueueMessageBulk ms) = TimeWarp $ \p -> do let q = runEventQueue $ pointRun p @@ -339,13 +348,14 @@ deliverAcknowledgmentMessages $ map (acknowledgmentMessage infind) ms 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 + do p' <- invokeEvent p currentEventPoint + if messageReceiveTime m < t0 + then do f <- fmap dioAllowSkippingOutdatedMessage dioParams + if f + then invokeEvent p' logOutdatedMessage + else error "Received the outdated message: processChannelMessage" + else invokeTimeWarp p' $ + enqueueMessage (queueInputMessages q) m processChannelMessage x@(AcknowledgmentQueueMessage m) = TimeWarp $ \p -> do let q = runEventQueue $ pointRun p @@ -377,8 +387,7 @@ t' <- invokeEvent p getLocalTime sender <- messageInboxId receiver <- timeServerId - liftDistributedUnsafe $ - DP.send receiver (LocalTimeMessage sender t') + sendLocalTimeDIO receiver sender t' processChannelMessage x@(GlobalTimeMessage globalTime) = TimeWarp $ \p -> do let q = runEventQueue $ pointRun p @@ -391,14 +400,32 @@ resetAcknowledgmentMessageTime (queueTransientMessages q) invokeEvent p $ updateGlobalTime globalTime -processChannelMessage x@TerminateLocalProcessMessage = +processChannelMessage x@(ProcessMonitorNotificationMessage y@(DP.ProcessMonitorNotification _ pid reason)) = TimeWarp $ \p -> - do --- + do let q = runEventQueue $ pointRun p + --- --- invokeEvent p $ --- logMessage x --- - liftDistributedUnsafe $ - DP.terminate + invokeEvent p $ + triggerSignal (queueProcessMonitorNotificationSource q) y +processChannelMessage x@(ReconnectProcessMessage pid) = + TimeWarp $ \p -> + do let q = runEventQueue $ pointRun p + --- + --- invokeEvent p $ + --- logMessage x + --- + invokeEvent p $ + reconnectProcess pid +processChannelMessage x@(KeepAliveLocalProcessMessage m) = + TimeWarp $ \p -> + do let q = runEventQueue $ pointRun p + --- + --- invokeEvent p $ + --- logMessage x + --- + return () -- | Return the local minimum time. getLocalTime :: Event DIO Double @@ -447,8 +474,7 @@ --- sender <- messageInboxId receiver <- timeServerId - liftDistributedUnsafe $ - DP.send receiver (RequestGlobalTimeMessage sender) + sendRequestGlobalTimeDIO receiver sender -- | Show the message. showMessage :: Message -> ShowS @@ -529,9 +555,9 @@ logOutdatedMessage :: Event DIO () logOutdatedMessage = Event $ \p -> - logDIO ERROR $ + logDIO WARNING $ "t = " ++ (show $ pointTime p) ++ - ": received the outdated message" + ": skipping the outdated message" -- | Reduce events till the specified time. reduceEvents :: Double -> Event DIO () @@ -588,7 +614,7 @@ Just _ -> invokeTimeWarp p $ syncLocalTime m Nothing -> - do invokeEvent p requestGlobalTime + do -- invokeEvent p requestGlobalTime invokeTimeWarp p $ syncLocalTime0 m else return () @@ -662,7 +688,7 @@ do let q = runEventQueue $ pointRun p t = pointTime p --- - logDIO NOTICE $ + logDIO INFO $ "t = " ++ show t ++ ": retrying the computations..." --- @@ -702,3 +728,36 @@ "Detected a deadlock when retrying the computations: handleEventRetry\n" ++ "--- the nested exception ---\n" ++ show e loop + +-- | Reconnect to the remote process. +reconnectProcess :: DP.ProcessId -> Event DIO () +reconnectProcess pid = + Event $ \p -> + do let q = runEventQueue $ pointRun p + --- + logDIO NOTICE $ + "t = " ++ show (pointTime p) ++ + ": reconnecting to " ++ show pid ++ "..." + --- + infind <- liftIOUnsafe $ readIORef (queueInFind q) + let ys = queueInputMessages q + ys' <- liftIOUnsafe $ + fmap (map $ acknowledgmentMessage infind) $ + filterInputMessages (\x -> messageSenderId x == pid) ys + unless (null ys') $ + sendAcknowledgmentMessagesDIO pid ys' + xs <- liftIOUnsafe $ transientMessages (queueTransientMessages q) + let xs' = filter (\x -> messageReceiverId x == pid) xs + unless (null xs') $ + sendMessagesDIO pid xs' + +-- | A signal triggered when coming the process monitor notification from the Cloud Haskell back-end. +processMonitorSignal :: Signal DIO DP.ProcessMonitorNotification +processMonitorSignal = + Signal { handleSignal = \h -> + Event $ \p -> + let q = runEventQueue (pointRun p) + s = publishSignal (queueProcessMonitorNotificationSource q) + in invokeEvent p $ + handleSignal s h + } diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/IO.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/IO.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/IO.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/IO.hs @@ -3,7 +3,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.IO --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -17,7 +17,8 @@ enqueueMessage, messageEnqueued, retryInputMessages, - reduceInputMessages) where + reduceInputMessages, + filterInputMessages) where import Data.Maybe import Data.List @@ -121,7 +122,10 @@ t0 = pointTime p i <- liftIOUnsafe $ findAntiMessage q m case i of - Nothing -> return () + Nothing -> + do -- skip the message duplicate + logSkipInputMessage t0 + return () Just i | i >= 0 -> do -- found the anti-message at the specified index when (t <= t0) $ @@ -155,10 +159,16 @@ else do liftIOUnsafe $ insertMessage q m i invokeEvent p $ activateMessage q i +-- | Log the message skip +logSkipInputMessage :: Double -> DIO () +logSkipInputMessage t0 = + logDIO NOTICE $ + "Skip the message at t = " ++ (show t0) + -- | Log the rollback. logRollbackInputMessages :: Double -> Double -> Bool -> DIO () logRollbackInputMessages t0 t including = - logDIO NOTICE $ + logDIO INFO $ "Rollback at t = " ++ (show t0) ++ " --> " ++ (show t) ++ (if not including then " not including" else "") @@ -385,3 +395,16 @@ then loop n (i + 1) else return i +-- | Filter the input messages using the specified predicate. +filterInputMessages :: (Message -> Bool) -> InputMessageQueue -> IO [Message] +filterInputMessages pred q = + do count <- vectorCount (inputMessages q) + loop count 0 [] + where + loop n i acc + | i >= n = return (reverse acc) + | otherwise = do item <- readVector (inputMessages q) i + let m = itemMessage item + if pred m + then loop n (i + 1) (m : acc) + else loop n (i + 1) acc diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs-boot b/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs-boot --- a/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs-boot +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs-boot @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -17,7 +17,8 @@ enqueueMessage, messageEnqueued, retryInputMessages, - reduceInputMessages) where + reduceInputMessages, + filterInputMessages) where import Simulation.Aivika.Trans.Simulation import Simulation.Aivika.Trans.Event @@ -47,3 +48,5 @@ retryInputMessages :: InputMessageQueue -> TimeWarp DIO () reduceInputMessages :: InputMessageQueue -> Double -> IO () + +filterInputMessages :: (Message -> Bool) -> InputMessageQueue -> IO [Message] diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs @@ -0,0 +1,107 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.KeepAliveManager +-- Copyright : Copyright (c) 2015-2017, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 7.10.3 +-- +-- This module is responsible for delivering keep-alive messages. +-- +module Simulation.Aivika.Distributed.Optimistic.Internal.KeepAliveManager + (KeepAliveManager, + KeepAliveParams(..), + newKeepAliveManager, + addKeepAliveReceiver, + trySendKeepAlive, + trySendKeepAliveUTC) where + +import qualified Data.Set as S +import Data.Maybe +import Data.IORef +import Data.Time.Clock + +import Control.Monad +import Control.Monad.Trans +import qualified Control.Distributed.Process as DP + +import Simulation.Aivika.Distributed.Optimistic.Internal.Priority +import Simulation.Aivika.Distributed.Optimistic.Internal.Message + +-- | The keep-alive parameters. +data KeepAliveParams = + KeepAliveParams { keepAliveLoggingPriority :: Priority, + -- ^ the logging priority + keepAliveInterval :: Int + -- ^ the interval in microseconds to send keep-alive messages + } + +-- | The keep-alive manager. +data KeepAliveManager = + KeepAliveManager { keepAliveParams :: KeepAliveParams, + -- ^ the manager parameter + keepAliveTimestamp :: IORef UTCTime, + -- ^ the keep alive timestamp + keepAliveReceivers :: IORef (S.Set DP.ProcessId) + -- ^ the receivers of the keep-alive messages + } + +-- | Create a new keep-alive manager. +newKeepAliveManager :: KeepAliveParams -> IO KeepAliveManager +newKeepAliveManager ps = + do timestamp <- getCurrentTime >>= newIORef + receivers <- newIORef S.empty + return KeepAliveManager { keepAliveParams = ps, + keepAliveTimestamp = timestamp, + keepAliveReceivers = receivers } + +-- | Add the keep-alive message receiver. +addKeepAliveReceiver :: KeepAliveManager -> DP.ProcessId -> IO () +addKeepAliveReceiver manager pid = + modifyIORef (keepAliveReceivers manager) $ + S.insert pid + +-- | Try to send a keep-alive message. +trySendKeepAlive :: KeepAliveManager -> DP.Process () +trySendKeepAlive manager = + do empty <- liftIO $ fmap S.null $ readIORef (keepAliveReceivers manager) + unless empty $ + do utc <- liftIO getCurrentTime + trySendKeepAliveUTC manager utc + +-- | Try to send a keep-alive message by the specified current time. +trySendKeepAliveUTC :: KeepAliveManager -> UTCTime -> DP.Process () +trySendKeepAliveUTC manager utc = + do empty <- liftIO $ fmap S.null $ readIORef (keepAliveReceivers manager) + unless empty $ + do f <- liftIO $ shouldSendKeepAlive manager utc + when f $ + do --- + logKeepAliveManager manager INFO $ + "Sending a keep-alive message" + --- + liftIO $ writeIORef (keepAliveTimestamp manager) utc + pids <- liftIO $ readIORef (keepAliveReceivers manager) + forM_ pids $ \pid -> + DP.send pid KeepAliveMessage + +-- | Whether should send a keep-alive message. +shouldSendKeepAlive :: KeepAliveManager -> UTCTime -> IO Bool +shouldSendKeepAlive manager utc = + do utc0 <- readIORef (keepAliveTimestamp manager) + let dt = fromRational $ toRational (diffUTCTime utc utc0) + return $ + secondsToMicroseconds dt > (keepAliveInterval $ keepAliveParams manager) + +-- | Convert seconds to microseconds. +secondsToMicroseconds :: Double -> Int +secondsToMicroseconds x = fromInteger $ toInteger $ round (1000000 * x) + +-- | Log the message with the specified priority. +logKeepAliveManager :: KeepAliveManager -> Priority -> String -> DP.Process () +{-# INLINE logKeepAliveManager #-} +logKeepAliveManager manager p message = + when (keepAliveLoggingPriority (keepAliveParams manager) <= p) $ + DP.say $ + embracePriority p ++ " " ++ message diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs @@ -3,7 +3,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Message --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -18,7 +18,9 @@ AcknowledgmentMessage(..), acknowledgmentMessage, LocalProcessMessage(..), - TimeServerMessage(..)) where + TimeServerMessage(..), + InboxProcessMessage(..), + KeepAliveMessage(..)) where import GHC.Generics @@ -117,9 +119,13 @@ -- ^ the time server requests for a local minimum time | GlobalTimeMessage Double -- ^ the time server sent a global time - | TerminateLocalProcessMessage - -- ^ the time server asked to terminate the process - deriving (Eq, Show, Typeable, Generic) + | ProcessMonitorNotificationMessage DP.ProcessMonitorNotification + -- ^ the process monitor notification + | ReconnectProcessMessage DP.ProcessId + -- ^ finish reconnecting to the specified process + | KeepAliveLocalProcessMessage KeepAliveMessage + -- ^ the keep alive message for the local process + deriving (Show, Typeable, Generic) instance Binary LocalProcessMessage @@ -128,13 +134,58 @@ -- ^ register the local process in the time server | UnregisterLocalProcessMessage DP.ProcessId -- ^ unregister the local process from the time server + | TerminateTimeServerMessage DP.ProcessId + -- ^ the local process asked to terminate the time server | RequestGlobalTimeMessage DP.ProcessId -- ^ the local process requested for the global minimum time | LocalTimeMessage DP.ProcessId Double -- ^ the local process sent its local minimum time - | TerminateTimeServerMessage DP.ProcessId - -- ^ the local process asked to terminate the time server + | ReMonitorTimeServerMessage [DP.ProcessId] + -- ^ re-monitor the local processes by their identifiers deriving (Eq, Show, Typeable, Generic) instance Binary TimeServerMessage +-- | The message destined directly for the inbox process. +data InboxProcessMessage = MonitorProcessMessage DP.ProcessId + -- ^ monitor the local process by its inbox process identifier + | ReMonitorProcessMessage [DP.ProcessId] + -- ^ re-monitor the local processes by their identifiers + | TrySendKeepAliveMessage + -- ^ try to send a keep alive message + | SendQueueMessage DP.ProcessId Message + -- ^ send a queue message via the inbox process + | SendQueueMessageBulk DP.ProcessId [Message] + -- ^ send a bulk of queue messages via the inbox process + | SendAcknowledgmentQueueMessage DP.ProcessId AcknowledgmentMessage + -- ^ send an acknowledgment message via the inbox process + | SendAcknowledgmentQueueMessageBulk DP.ProcessId [AcknowledgmentMessage] + -- ^ send a bulk of acknowledgment messages via the inbox process + | SendLocalTimeMessage DP.ProcessId DP.ProcessId Double + -- ^ send the local time message to the time server + | SendRequestGlobalTimeMessage DP.ProcessId DP.ProcessId + -- ^ send the request for the global virtual time + | SendRegisterLocalProcessMessage DP.ProcessId DP.ProcessId + -- ^ register the local process in the time server + | SendUnregisterLocalProcessMessage DP.ProcessId DP.ProcessId + -- ^ unregister the local process from the time server + | SendTerminateTimeServerMessage DP.ProcessId DP.ProcessId + -- ^ the local process asked to terminate the time server + | RegisterLocalProcessAcknowledgmentMessage DP.ProcessId + -- ^ after registering the local process in the time server + | UnregisterLocalProcessAcknowledgmentMessage DP.ProcessId + -- ^ after unregistering the local process from the time server + | TerminateTimeServerAcknowledgmentMessage DP.ProcessId + -- ^ after started terminating the time server + | TerminateInboxProcessMessage + -- ^ terminate the inbox process + deriving (Eq, Show, Typeable, Generic) + +instance Binary InboxProcessMessage + +-- | The keep-alive message type. +data KeepAliveMessage = KeepAliveMessage + -- ^ the keel-alive message. + deriving (Eq, Show, Typeable, Generic) + +instance Binary KeepAliveMessage diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -122,14 +122,12 @@ -- | Deliver the message on low level. deliverMessage :: Message -> DIO () deliverMessage x = - liftDistributedUnsafe $ - DP.send (messageReceiverId x) (QueueMessage x) + sendMessageDIO (messageReceiverId x) x -- | Deliver the anti-message on low level. deliverAntiMessage :: Message -> DIO () deliverAntiMessage x = - liftDistributedUnsafe $ - DP.send (messageReceiverId x) (QueueMessage x) + sendMessageDIO (messageReceiverId x) x -- | Deliver the anti-messages on low level. deliverAntiMessages :: [Message] -> DIO () @@ -137,6 +135,5 @@ let ys = groupBy (\a b -> messageReceiverId a == messageReceiverId b) xs dlv [] = return () dlv zs@(z : _) = - liftDistributedUnsafe $ - DP.send (messageReceiverId z) (QueueMessageBulk zs) + sendMessagesDIO (messageReceiverId z) 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 --- a/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs-boot +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs-boot @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Priority.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Priority.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/Priority.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Priority.hs @@ -3,7 +3,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Priority --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs @@ -1,73 +1,15 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Ref --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental -- Tested with: GHC 7.10.3 -- --- The implementation of mutable references. +-- The implementation of mutable strict 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 + (module Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict) where --- | 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 +import Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs-boot b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs-boot deleted file mode 100644 --- a/Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs-boot +++ /dev/null @@ -1,29 +0,0 @@ - --- | --- 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/Ref/Lazy.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref/Lazy.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref/Lazy.hs @@ -0,0 +1,73 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Lazy +-- Copyright : Copyright (c) 2015-2017, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 7.10.3 +-- +-- The implementation of mutable lazy references. +-- +module Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Lazy + (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 lazy 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 + 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 + liftIOUnsafe $ writeIORef (refValue r) b diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Ref/Strict.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref/Strict.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref/Strict.hs @@ -0,0 +1,73 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict +-- Copyright : Copyright (c) 2015-2017, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 7.10.3 +-- +-- The implementation of mutable strict references. +-- +module Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict + (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/Strict.hs-boot b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref/Strict.hs-boot new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Ref/Strict.hs-boot @@ -0,0 +1,29 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict +-- Copyright : Copyright (c) 2015-2017, 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.Strict 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/SignalHelper.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/SignalHelper.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/SignalHelper.hs @@ -0,0 +1,26 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper +-- Copyright : Copyright (c) 2015-2017, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 7.10.3 +-- +-- This module defines a signal helper. +-- +module Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper + (newDIOSignalSource0, + handleDIOSignalComposite) where + +import Simulation.Aivika.Trans + +import Simulation.Aivika.Distributed.Optimistic.DIO + +-- | Create a new signal source in safe manner without constraints. +newDIOSignalSource0 :: DIO (SignalSource DIO a) +newDIOSignalSource0 = newSignalSource0 + +-- | Handle the signal in safe manner without constraints. +handleDIOSignalComposite :: Signal DIO a -> (a -> Event DIO ()) -> Composite DIO () +handleDIOSignalComposite = handleSignalComposite diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/SignalHelper.hs-boot b/Simulation/Aivika/Distributed/Optimistic/Internal/SignalHelper.hs-boot new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/SignalHelper.hs-boot @@ -0,0 +1,22 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper +-- Copyright : Copyright (c) 2015-2017, 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.SignalHelper + (newDIOSignalSource0, + handleDIOSignalComposite) where + +import Simulation.Aivika.Trans + +import Simulation.Aivika.Distributed.Optimistic.Internal.DIO + +newDIOSignalSource0 :: DIO (SignalSource DIO a) + +handleDIOSignalComposite :: Signal DIO a -> (a -> Event DIO ()) -> Composite DIO () diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs @@ -3,7 +3,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -29,6 +29,8 @@ import Control.Monad import Control.Monad.Trans +import Control.Exception +import qualified Control.Monad.Catch as C import Control.Concurrent import qualified Control.Distributed.Process as DP @@ -39,10 +41,20 @@ data TimeServerParams = TimeServerParams { tsLoggingPriority :: Priority, -- ^ the logging priority - tsExpectTimeout :: Int, - -- ^ the timeout in microseconds within which a new message is expected - tsTimeSyncDelay :: Int + tsReceiveTimeout :: Int, + -- ^ the timeout in microseconds used when receiving messages + tsTimeSyncTimeout :: Int, + -- ^ the timeout in microseconds used for the time synchronization sessions + tsTimeSyncDelay :: Int, -- ^ the delay in microseconds between the time synchronization sessions + tsProcessMonitoringEnabled :: Bool, + -- ^ Whether the process monitoring is enabled + tsProcessMonitoringDelay :: Int, + -- ^ The delay in microseconds which must be applied for monitoring every remote process. + tsProcessReconnectingEnabled :: Bool, + -- ^ Whether the automatic reconnecting to processes is enabled when enabled monitoring + tsProcessReconnectingDelay :: Int + -- ^ the delay in microseconds before reconnecting } deriving (Eq, Ord, Show, Typeable, Generic) instance Binary TimeServerParams @@ -55,41 +67,56 @@ -- ^ the initial quorum of registered local processes to start the simulation tsInInit :: IORef Bool, -- ^ whether the time server is in the initial mode + tsTerminating :: IORef Bool, + -- ^ whether the time server is in the terminating mode tsProcesses :: IORef (M.Map DP.ProcessId LocalProcessInfo), -- ^ the information about local processes tsProcessesInFind :: IORef (S.Set DP.ProcessId), -- ^ the processed used in the current finding of the global time - tsGlobalTime :: IORef (Maybe Double) + tsGlobalTime :: IORef (Maybe Double), -- ^ the global time of the model + tsGlobalTimeTimestamp :: IORef (Maybe UTCTime) + -- ^ the global time timestamp } -- | The information about the local process. data LocalProcessInfo = - LocalProcessInfo { lpLocalTime :: IORef (Maybe Double) + LocalProcessInfo { lpLocalTime :: IORef (Maybe Double), -- ^ the local time of the process + lpMonitorRef :: Maybe DP.MonitorRef + -- ^ the local process monitor reference } -- | The default time server parameters. defaultTimeServerParams :: TimeServerParams defaultTimeServerParams = TimeServerParams { tsLoggingPriority = WARNING, - tsExpectTimeout = 100000, - tsTimeSyncDelay = 1000000 + tsReceiveTimeout = 100000, + tsTimeSyncTimeout = 60000000, + tsTimeSyncDelay = 1000000, + tsProcessMonitoringEnabled = False, + tsProcessMonitoringDelay = 3000000, + tsProcessReconnectingEnabled = False, + tsProcessReconnectingDelay = 5000000 } -- | Create a new time server by the specified initial quorum and parameters. newTimeServer :: Int -> TimeServerParams -> IO TimeServer newTimeServer n ps = do f <- newIORef True + ft <- newIORef False m <- newIORef M.empty s <- newIORef S.empty t0 <- newIORef Nothing + t' <- newIORef Nothing return TimeServer { tsParams = ps, tsInitQuorum = n, tsInInit = f, + tsTerminating = ft, tsProcesses = m, tsProcessesInFind = s, - tsGlobalTime = t0 + tsGlobalTime = t0, + tsGlobalTimeTimestamp = t' } -- | Process the time server message. @@ -105,9 +132,18 @@ Nothing -> do t <- newIORef Nothing modifyIORef (tsProcesses server) $ - M.insert pid LocalProcessInfo { lpLocalTime = t } + M.insert pid LocalProcessInfo { lpLocalTime = t, lpMonitorRef = Nothing } return $ - tryStartTimeServer server + do when (tsProcessMonitoringEnabled $ tsParams server) $ + do logTimeServer server INFO $ + "Time Server: monitoring the process by identifier " ++ show pid + r <- DP.monitor pid + liftIO $ + modifyIORef (tsProcesses server) $ + M.update (\x -> Just x { lpMonitorRef = Just r }) pid + serverId <- DP.getSelfPid + DP.send pid (RegisterLocalProcessAcknowledgmentMessage serverId) + tryStartTimeServer server processTimeServerMessage server (UnregisterLocalProcessMessage pid) = join $ liftIO $ do m <- readIORef (tsProcesses server) @@ -122,19 +158,41 @@ modifyIORef (tsProcessesInFind server) $ S.delete pid return $ - tryProvideTimeServerGlobalTime server + do when (tsProcessMonitoringEnabled $ tsParams server) $ + case lpMonitorRef x of + Nothing -> return () + Just r -> + do logTimeServer server INFO $ + "Time Server: unmonitoring the process by identifier " ++ show pid + DP.unmonitor r + serverId <- DP.getSelfPid + DP.send pid (UnregisterLocalProcessAcknowledgmentMessage serverId) + tryProvideTimeServerGlobalTime server + tryTerminateTimeServer server processTimeServerMessage server (TerminateTimeServerMessage pid) = - do pids <- - liftIO $ - do m <- readIORef (tsProcesses server) - writeIORef (tsProcesses server) M.empty - writeIORef (tsProcessesInFind server) S.empty - writeIORef (tsGlobalTime server) Nothing - return $ filter (/= pid) (M.keys m) - forM_ pids $ \pid -> - DP.send pid TerminateLocalProcessMessage - logTimeServer server INFO "Time Server: terminating..." - DP.terminate + 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 modifyIORef (tsProcesses server) $ + M.delete pid + modifyIORef (tsProcessesInFind server) $ + S.delete pid + return $ + do when (tsProcessMonitoringEnabled $ tsParams server) $ + case lpMonitorRef x of + Nothing -> return () + Just r -> + do logTimeServer server INFO $ + "Time Server: unmonitoring the process by identifier " ++ show pid + DP.unmonitor r + serverId <- DP.getSelfPid + DP.send pid (TerminateTimeServerAcknowledgmentMessage serverId) + startTerminatingTimeServer server processTimeServerMessage server (RequestGlobalTimeMessage pid) = tryComputeTimeServerGlobalTime server processTimeServerMessage server (LocalTimeMessage pid t') = @@ -153,6 +211,16 @@ S.delete pid return $ tryProvideTimeServerGlobalTime server +processTimeServerMessage server (ReMonitorTimeServerMessage pids) = + do forM_ pids $ \pid -> + do --- + logTimeServer server NOTICE $ "Time Server: re-monitoring " ++ show pid + --- + DP.monitor pid + --- + logTimeServer server NOTICE $ "Time Server: started re-monitoring " ++ show pid + --- + resetComputingTimeServerGlobalTime server -- | Whether the both values are defined and the first is greater than or equaled to the second. (.>=.) :: Maybe Double -> Maybe Double -> Bool @@ -197,6 +265,16 @@ else return $ computeTimeServerGlobalTime server +-- | Reset computing the time server global time. +resetComputingTimeServerGlobalTime :: TimeServer -> DP.Process () +resetComputingTimeServerGlobalTime server = + do logTimeServer server NOTICE $ + "Time Server: reset computing the global time" + liftIO $ + do utc <- getCurrentTime + writeIORef (tsProcessesInFind server) S.empty + writeIORef (tsGlobalTimeTimestamp server) (Just utc) + -- | Try to provide the local processes wth the global time. tryProvideTimeServerGlobalTime :: TimeServer -> DP.Process () tryProvideTimeServerGlobalTime server = @@ -229,7 +307,7 @@ provideTimeServerGlobalTime :: TimeServer -> DP.Process () provideTimeServerGlobalTime server = do t0 <- liftIO $ timeServerGlobalTime server - logTimeServer server DEBUG $ + logTimeServer server INFO $ "Time Server: providing the global time = " ++ show t0 case t0 of Nothing -> return () @@ -238,7 +316,9 @@ when (t' .>. Just t0) $ logTimeServer server NOTICE "Time Server: the global time has decreased" + timestamp <- liftIO getCurrentTime liftIO $ writeIORef (tsGlobalTime server) (Just t0) + liftIO $ writeIORef (tsGlobalTimeTimestamp server) (Just timestamp) zs <- liftIO $ fmap M.assocs $ readIORef (tsProcesses server) forM_ zs $ \(pid, x) -> DP.send pid (GlobalTimeMessage t0) @@ -261,10 +341,44 @@ Just _ -> loop zs' (liftM2 min t acc) +-- | Start terminating the time server. +startTerminatingTimeServer :: TimeServer -> DP.Process () +startTerminatingTimeServer server = + do logTimeServer server INFO "Time Server: start terminating..." + liftIO $ + writeIORef (tsTerminating server) True + tryTerminateTimeServer server + +-- | Try to terminate the time server. +tryTerminateTimeServer :: TimeServer -> DP.Process () +tryTerminateTimeServer server = + do f <- liftIO $ readIORef (tsTerminating server) + when f $ + do m <- liftIO $ readIORef (tsProcesses server) + when (M.null m) $ + do logTimeServer server INFO "Time Server: terminate" + DP.terminate + -- | Convert seconds to microseconds. secondsToMicroseconds :: Double -> Int secondsToMicroseconds x = fromInteger $ toInteger $ round (1000000 * x) +-- | The internal time server message. +data InternalTimeServerMessage = InternalTimeServerMessage TimeServerMessage + -- ^ the time server message + | InternalProcessMonitorNotification DP.ProcessMonitorNotification + -- ^ the process monitor notification + | InternalKeepAliveMessage KeepAliveMessage + -- ^ the keep alive message + +-- | Handle the time server exception +handleTimeServerException :: TimeServer -> SomeException -> DP.Process () +handleTimeServerException server e = + do --- + logTimeServer server ERROR $ "Exception occured: " ++ show e + --- + C.throwM e + -- | Start the time server by the specified initial quorum and parameters. -- The quorum defines the number of local processes that must be registered in -- the time server before the global time synchronization is started. @@ -273,22 +387,95 @@ do server <- liftIO $ newTimeServer n ps logTimeServer server INFO "Time Server: starting..." let loop utc0 = - do m <- DP.expectTimeout (tsExpectTimeout ps) :: DP.Process (Maybe TimeServerMessage) - case m of + do let f1 :: TimeServerMessage -> DP.Process InternalTimeServerMessage + f1 x = return (InternalTimeServerMessage x) + f2 :: DP.ProcessMonitorNotification -> DP.Process InternalTimeServerMessage + f2 x = return (InternalProcessMonitorNotification x) + f3 :: KeepAliveMessage -> DP.Process InternalTimeServerMessage + f3 x = return (InternalKeepAliveMessage x) + a <- DP.receiveTimeout (tsReceiveTimeout ps) [DP.match f1, DP.match f2, DP.match f3] + case a of Nothing -> return () - Just m -> + Just (InternalTimeServerMessage m) -> do --- logTimeServer server DEBUG $ "Time Server: " ++ show m --- processTimeServerMessage server m + Just (InternalProcessMonitorNotification m) -> + handleProcessMonitorNotification m server + Just (InternalKeepAliveMessage m) -> + do --- + logTimeServer server DEBUG $ + "Time Server: " ++ show m + --- + return () utc <- liftIO getCurrentTime - let dt = fromRational $ toRational (diffUTCTime utc utc0) - if secondsToMicroseconds dt > tsTimeSyncDelay ps + timestamp <- liftIO $ readIORef (tsGlobalTimeTimestamp server) + case timestamp of + Just x | shouldResetComputingTimeServerGlobalTime server x utc -> + resetComputingTimeServerGlobalTime server + _ -> return () + if shouldComputeTimeServerGlobalTime server utc0 utc then do tryComputeTimeServerGlobalTime server loop utc else loop utc0 - liftIO getCurrentTime >>= loop + C.catch (liftIO getCurrentTime >>= loop) (handleTimeServerException server) + +-- | Handle the process monitor notification. +handleProcessMonitorNotification :: DP.ProcessMonitorNotification -> TimeServer -> DP.Process () +handleProcessMonitorNotification m@(DP.ProcessMonitorNotification _ pid0 reason) server = + do let ps = tsParams server + recv m@(DP.ProcessMonitorNotification _ _ _) = + do --- + logTimeServer server WARNING $ + "Time Server: received a process monitor notification " ++ show m + --- + return m + recv m + when (tsProcessReconnectingEnabled ps && reason == DP.DiedDisconnect) $ + do liftIO $ + threadDelay (tsProcessReconnectingDelay ps) + let pred m@(DP.ProcessMonitorNotification _ _ reason) = reason == DP.DiedDisconnect + loop :: [DP.ProcessId] -> DP.Process [DP.ProcessId] + loop acc = + do y <- DP.receiveTimeout 0 [DP.matchIf pred recv] + case y of + Nothing -> return $ reverse acc + Just m@(DP.ProcessMonitorNotification _ pid _) -> loop (pid : acc) + pids <- loop [pid0] + --- + logTimeServer server NOTICE "Begin reconnecting..." + --- + forM_ pids $ \pid -> + do --- + logTimeServer server NOTICE $ + "Time Server: reconnecting to " ++ show pid + --- + DP.reconnect pid + serverId <- DP.getSelfPid + DP.spawnLocal $ + let action = + do liftIO $ + threadDelay (tsProcessMonitoringDelay ps) + --- + logTimeServer server NOTICE $ "Time Server: proceed to the re-monitoring" + --- + DP.send serverId (ReMonitorTimeServerMessage pids) + in C.catch action (handleTimeServerException server) + return () + +-- | Test whether should compute the time server global time. +shouldComputeTimeServerGlobalTime :: TimeServer -> UTCTime -> UTCTime -> Bool +shouldComputeTimeServerGlobalTime server utc0 utc = + let dt = fromRational $ toRational (diffUTCTime utc utc0) + in secondsToMicroseconds dt > (tsTimeSyncDelay $ tsParams server) + +-- | Test whether should reset computing the time server global time. +shouldResetComputingTimeServerGlobalTime :: TimeServer -> UTCTime -> UTCTime -> Bool +shouldResetComputingTimeServerGlobalTime server utc0 utc = + let dt = fromRational $ toRational (diffUTCTime utc utc0) + in secondsToMicroseconds dt > (tsTimeSyncTimeout $ tsParams server) -- | A curried version of 'timeServer' for starting the time server on remote node. curryTimeServer :: (Int, TimeServerParams) -> DP.Process () diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/TimeWarp.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeWarp.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/TimeWarp.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeWarp.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental @@ -14,6 +14,7 @@ newTransientMessageQueue, transientMessageQueueSize, transientMessageQueueTime, + transientMessages, enqueueTransientMessage, processAcknowledgmentMessage, acknowledgmentMessageTime, @@ -21,7 +22,7 @@ deliverAcknowledgmentMessage, deliverAcknowledgmentMessages) where -import qualified Data.Set as S +import qualified Data.Map as M import Data.List import Data.IORef @@ -35,8 +36,8 @@ -- | Specifies the transient message queue. data TransientMessageQueue = - TransientMessageQueue { queuePrototypeMessages :: IORef (S.Set TransientMessageQueueItem), - -- ^ The prototype messages. + TransientMessageQueue { queueTransientMessages :: IORef (M.Map TransientMessageQueueItem Message), + -- ^ The transient messages. queueMarkedMessageTime :: IORef Double -- ^ The marked acknowledgment message time. } @@ -97,30 +98,36 @@ -- | Create a new transient message queue. newTransientMessageQueue :: DIO TransientMessageQueue newTransientMessageQueue = - do ms <- liftIOUnsafe $ newIORef S.empty + do ms <- liftIOUnsafe $ newIORef M.empty r <- liftIOUnsafe $ newIORef (1 / 0) - return TransientMessageQueue { queuePrototypeMessages = ms, + return TransientMessageQueue { queueTransientMessages = ms, queueMarkedMessageTime = r } -- | Get the transient message queue size. transientMessageQueueSize :: TransientMessageQueue -> IO Int transientMessageQueueSize q = - fmap S.size $ readIORef (queuePrototypeMessages q) + fmap M.size $ readIORef (queueTransientMessages q) -- | Get the virtual time of the transient message queue. transientMessageQueueTime :: TransientMessageQueue -> IO Double transientMessageQueueTime q = - do s <- readIORef (queuePrototypeMessages q) - if S.null s + do ms <- readIORef (queueTransientMessages q) + if M.null ms then return (1 / 0) - else let m = S.findMin s + else let (m, _) = M.findMin ms in return (itemReceiveTime m) +-- | Get the transient messages. +transientMessages :: TransientMessageQueue -> IO [Message] +transientMessages q = + do ms <- readIORef (queueTransientMessages q) + return (M.elems ms) + -- | Enqueue the transient message. enqueueTransientMessage :: TransientMessageQueue -> Message -> IO () enqueueTransientMessage q m = - modifyIORef (queuePrototypeMessages q) $ - S.insert (transientMessageQueueItem m) + modifyIORef (queueTransientMessages q) $ + M.insert (transientMessageQueueItem m) m -- | Enqueue the acknowledgment message. enqueueAcknowledgmentMessage :: TransientMessageQueue -> AcknowledgmentMessage -> IO () @@ -131,10 +138,13 @@ -- | Process the acknowledgment message. processAcknowledgmentMessage :: TransientMessageQueue -> AcknowledgmentMessage -> IO () processAcknowledgmentMessage q m = - do modifyIORef (queuePrototypeMessages q) $ - S.delete (acknowledgmentMessageQueueItem m) - when (acknowledgmentMarked m) $ - enqueueAcknowledgmentMessage q m + do ms <- readIORef (queueTransientMessages q) + let k = acknowledgmentMessageQueueItem m + when (M.member k ms) $ + do modifyIORef (queueTransientMessages q) $ + M.delete k + when (acknowledgmentMarked m) $ + enqueueAcknowledgmentMessage q m -- | Get the minimal marked acknowledgment message time. acknowledgmentMessageTime :: TransientMessageQueue -> IO Double @@ -149,8 +159,7 @@ -- | Deliver the acknowledgment message on low level. deliverAcknowledgmentMessage :: AcknowledgmentMessage -> DIO () deliverAcknowledgmentMessage x = - liftDistributedUnsafe $ - DP.send (acknowledgmentSenderId x) (AcknowledgmentQueueMessage x) + sendAcknowledgmentMessageDIO (acknowledgmentSenderId x) x -- | Deliver the acknowledgment messages on low level. deliverAcknowledgmentMessages :: [AcknowledgmentMessage] -> DIO () @@ -158,6 +167,5 @@ let ys = groupBy (\a b -> acknowledgmentSenderId a == acknowledgmentSenderId b) xs dlv [] = return () dlv zs@(z : _) = - liftDistributedUnsafe $ - DP.send (acknowledgmentSenderId z) (AcknowledgmentQueueMessageBulk zs) + sendAcknowledgmentMessagesDIO (acknowledgmentSenderId z) zs in forM_ ys dlv diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/Simulation/Aivika/Distributed/Optimistic/Message.hs b/Simulation/Aivika/Distributed/Optimistic/Message.hs --- a/Simulation/Aivika/Distributed/Optimistic/Message.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Message.hs @@ -4,7 +4,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Message --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/Simulation/Aivika/Distributed/Optimistic/Priority.hs b/Simulation/Aivika/Distributed/Optimistic/Priority.hs --- a/Simulation/Aivika/Distributed/Optimistic/Priority.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Priority.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Priority --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/Simulation/Aivika/Distributed/Optimistic/QueueStrategy.hs b/Simulation/Aivika/Distributed/Optimistic/QueueStrategy.hs --- a/Simulation/Aivika/Distributed/Optimistic/QueueStrategy.hs +++ b/Simulation/Aivika/Distributed/Optimistic/QueueStrategy.hs @@ -3,7 +3,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.QueueStrategy --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/Simulation/Aivika/Distributed/Optimistic/Ref/Base.hs b/Simulation/Aivika/Distributed/Optimistic/Ref/Base.hs --- a/Simulation/Aivika/Distributed/Optimistic/Ref/Base.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Ref/Base.hs @@ -1,48 +1,21 @@ -{-# LANGUAGE TypeFamilies #-} - -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Ref.Base --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental -- Tested with: GHC 7.10.3 -- --- Here is an implementation of mutable references, where +-- Here is an implementation of strict mutable references, where -- 'DIO' is an instance of 'MonadRef' and 'MonadRef0'. -- -module Simulation.Aivika.Distributed.Optimistic.Ref.Base () where +module Simulation.Aivika.Distributed.Optimistic.Ref.Base + (module Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict) 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 +import Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict diff --git a/Simulation/Aivika/Distributed/Optimistic/Ref/Base/Lazy.hs b/Simulation/Aivika/Distributed/Optimistic/Ref/Base/Lazy.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Ref/Base/Lazy.hs @@ -0,0 +1,48 @@ + +{-# LANGUAGE TypeFamilies #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Ref.Base.Lazy +-- Copyright : Copyright (c) 2015-2017, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 7.10.3 +-- +-- Here is an implementation of lazy mutable references, where +-- 'DIO' is an instance of 'MonadRef' and 'MonadRef0'. +-- +module Simulation.Aivika.Distributed.Optimistic.Ref.Base.Lazy () where + +import Simulation.Aivika.Trans.Internal.Types +import Simulation.Aivika.Trans.Comp +import Simulation.Aivika.Trans.Ref.Base.Lazy + +import Simulation.Aivika.Distributed.Optimistic.Internal.DIO +import qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Lazy as R + +-- | The implementation of lazy mutable references. +instance MonadRef DIO where + + -- | The lazy 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/Ref/Base/Strict.hs b/Simulation/Aivika/Distributed/Optimistic/Ref/Base/Strict.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Ref/Base/Strict.hs @@ -0,0 +1,48 @@ + +{-# LANGUAGE TypeFamilies #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict +-- Copyright : Copyright (c) 2015-2017, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 7.10.3 +-- +-- Here is an implementation of strict mutable references, where +-- 'DIO' is an instance of 'MonadRef' and 'MonadRef0'. +-- +module Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict () where + +import Simulation.Aivika.Trans.Internal.Types +import Simulation.Aivika.Trans.Comp +import Simulation.Aivika.Trans.Ref.Base.Strict + +import Simulation.Aivika.Distributed.Optimistic.Internal.DIO +import qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict as R + +-- | The implementation of strict mutable references. +instance MonadRef DIO where + + -- | The strict 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 --- a/Simulation/Aivika/Distributed/Optimistic/TimeServer.hs +++ b/Simulation/Aivika/Distributed/Optimistic/TimeServer.hs @@ -1,7 +1,7 @@ -- | -- Module : Simulation.Aivika.Distributed.Optimistic.TimeServer --- Copyright : Copyright (c) 2015-2016, David Sorokin +-- Copyright : Copyright (c) 2015-2017, David Sorokin -- License : BSD3 -- Maintainer : David Sorokin -- Stability : experimental diff --git a/aivika-distributed.cabal b/aivika-distributed.cabal --- a/aivika-distributed.cabal +++ b/aivika-distributed.cabal @@ -1,20 +1,52 @@ name: aivika-distributed -version: 0.3.1 +version: 0.5 synopsis: Parallel distributed discrete event simulation module for the Aivika library description: - This package extends the aivika-transformers [1] package with facilities for running parallel distributed simulations. - It uses an optimistic strategy known as the Time Warp method. To synchronize the global virtual time, since version 0.3 - it uses Samadi's algorithm. + This package extends the aivika-transformers [1] package and allows running parallel distributed simulations. + It uses an optimistic strategy known as the Time Warp method. To synchronize the global virtual time, + it uses Samadi's algorithm. . + Moreover, this package uses the author's modification that allows recovering the distributed + simulation after temporary connection errors whenever possible. For that, you have to enable explicitly + the recovering mode and enable monitoring all local processes including the specialized Time Server process + as it is shown in one of the test examples included in the distributive. + . + With the recovering mode enabled, you can try to build a distributed simulation using ordinary computers connected + via the ordinary net. For example, such a distributed model could even consist of computers located in different + continents of the Earth, where the computers could be connected through the Internet. Here the most exciting thing + is that this is the optimistic distributed simulation with possible rollbacks. It is assumed that optimistic methods + tend to better support the parallelism inherited in the models. + . + You may test the distributed simulation using your own notebook only, although the package is still destined to be + used with a multi-core computer, or computers connected in the distributed cluster. + . + There are additional packages that allow you to run the distributed simulation experiments by + the Monte-Carlo method. They allow you to save the simulation results in SQL databases and then generate a report + or a set of reports consisting of HTML pages with charts, histograms, links to CSV tables, summary statistics etc. + Please consult the AivikaSoft [3] website for more details. + . + Regarding the speed of simulation, the rough estimations are as follows. The simulation is slower up to + 6-9 times in comparison with the sequential aivika [2] simulation library using the equivalent sequential models. + Note that you can run up to 7 parallel local processes on a single 8-core processor computer and run the Time Server + process too. On a 36-core processor, you can launch up to 35 local processes simultaneously. + . + So, this estimation seems to be quite good. At the same time, the message passing between the local processes can + dramatically decrease the speed of distributed simulation, especially if they cause rollbacks. Thus, much depends on + the distributed model itself. + . \[1] . + \[2] + . + \[3] + . category: Simulation license: BSD3 license-file: LICENSE -copyright: (c) 2015-2016. David Sorokin +copyright: (c) 2015-2017. David Sorokin author: David Sorokin maintainer: David Sorokin -homepage: http://www.aivikasoft.com/en/products/aivika.html +homepage: http://www.aivikasoft.com cabal-version: >= 1.6 build-type: Simple tested-with: GHC == 7.10.1 @@ -26,6 +58,8 @@ tests/MachRep2Reproducible.hs tests/MachRep2Sync.hs tests/MachRep2SyncIO.hs + tests/SimpleLocalnetHelper.hs + tests/cluster.conf CHANGELOG.md library @@ -38,6 +72,8 @@ Simulation.Aivika.Distributed.Optimistic.Message Simulation.Aivika.Distributed.Optimistic.Priority Simulation.Aivika.Distributed.Optimistic.Ref.Base + Simulation.Aivika.Distributed.Optimistic.Ref.Base.Lazy + Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict Simulation.Aivika.Distributed.Optimistic.TimeServer other-modules: Simulation.Aivika.Distributed.Optimistic.Internal.Channel @@ -45,16 +81,20 @@ Simulation.Aivika.Distributed.Optimistic.Internal.Event Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue Simulation.Aivika.Distributed.Optimistic.Internal.IO + Simulation.Aivika.Distributed.Optimistic.Internal.KeepAliveManager 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.Ref.Lazy + Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict + Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog - build-depends: base >= 3 && < 6, + build-depends: base >= 4.6.0.0 && < 6, mtl >= 2.1.1, stm >= 2.4.2, random >= 1.0.0.3, @@ -63,8 +103,8 @@ containers >= 0.4.0.0, exceptions >= 0.8.0.2, distributed-process >= 0.6.1, - aivika >= 4.5, - aivika-transformers >= 4.5 + aivika >= 5.1, + aivika-transformers >= 5.1 extensions: MultiParamTypeClasses, FlexibleInstances, diff --git a/tests/MachRep1Simple.hs b/tests/MachRep1Simple.hs --- a/tests/MachRep1Simple.hs +++ b/tests/MachRep1Simple.hs @@ -84,8 +84,6 @@ let timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG } timeServerId <- DP.spawnLocal $ timeServer 1 timeServerParams runModel timeServerId - liftIO $ - threadDelay 1000000 main :: IO () main = do diff --git a/tests/MachRep2.hs b/tests/MachRep2.hs --- a/tests/MachRep2.hs +++ b/tests/MachRep2.hs @@ -46,8 +46,8 @@ spcMethod = RungeKutta4, spcGeneratorType = SimpleGenerator } --- | The time shift when replying the messages. -delta = 0.01 +-- | The time shift when replying to the messages. +delta = 1e-6 data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic) data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic) diff --git a/tests/MachRep2Distributed.hs b/tests/MachRep2Distributed.hs --- a/tests/MachRep2Distributed.hs +++ b/tests/MachRep2Distributed.hs @@ -65,8 +65,8 @@ spcMethod = RungeKutta4, spcGeneratorType = SimpleGenerator } --- | The time shift when replying the messages. -delta = 0.01 +-- | The time shift when replying to the messages. +delta = 1e-6 data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic) data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic) diff --git a/tests/MachRep2Reproducible.hs b/tests/MachRep2Reproducible.hs --- a/tests/MachRep2Reproducible.hs +++ b/tests/MachRep2Reproducible.hs @@ -48,8 +48,8 @@ spcMethod = RungeKutta4, spcGeneratorType = SimpleGenerator } --- | The time shift when replying the messages. -delta = 0.01 +-- | The time shift when replying to the messages. +delta = 1e-6 -- | The initial seeds. seeds = [456, 789] diff --git a/tests/MachRep2Sync.hs b/tests/MachRep2Sync.hs --- a/tests/MachRep2Sync.hs +++ b/tests/MachRep2Sync.hs @@ -46,8 +46,8 @@ spcMethod = RungeKutta4, spcGeneratorType = SimpleGenerator } --- | The time shift when replying the messages. -delta = 0.01 +-- | The time shift when replying to the messages. +delta = 1e-6 data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic) data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic) diff --git a/tests/MachRep2SyncIO.hs b/tests/MachRep2SyncIO.hs --- a/tests/MachRep2SyncIO.hs +++ b/tests/MachRep2SyncIO.hs @@ -46,8 +46,8 @@ spcMethod = RungeKutta4, spcGeneratorType = SimpleGenerator } --- | The time shift when replying the messages. -delta = 0.01 +-- | The time shift when replying to the messages. +delta = 1e-6 data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic) data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic) diff --git a/tests/SimpleLocalnetHelper.hs b/tests/SimpleLocalnetHelper.hs new file mode 100644 --- /dev/null +++ b/tests/SimpleLocalnetHelper.hs @@ -0,0 +1,70 @@ + +{-# LANGUAGE OverloadedStrings #-} + +module SimpleLocalnetHelper + (getSlaveBackend, + getSlaveConfBackend, + getMasterBackend, + getMasterConfBackend, + startMasterProcess) where + +import System.IO +import qualified Control.Distributed.Process as DP +import qualified Control.Distributed.Process.Node as Node +import Control.Distributed.Process.Backend.SimpleLocalnet +import qualified Data.ByteString.Char8 as BS +import Network.Transport (EndPointAddress(..)) + +-- | Make a NodeId from "host:port" string. +makeNodeId :: String -> DP.NodeId +makeNodeId addr = DP.NodeId . EndPointAddress . BS.concat $ [BS.pack addr, ":0"] + +splitAddr :: String -> (String, String) +splitAddr addr = (host, port) + where + (host, rest) = break (== ':') addr + port = tail rest + +deleteAt :: Int -> [String] -> [String] +deleteAt i xs = xs1 ++ xs2 + where (xs1, rest) = splitAt i xs + xs2 = tail rest + +getSlaveBackend :: Int -> [String] -> DP.RemoteTable -> IO Backend +getSlaveBackend i addrs rtable = + do let addr = addrs !! i + (host, port) = splitAddr addr + backend <- initializeBackend host port rtable + return backend + +getMasterBackend :: Int -> [String] -> DP.RemoteTable -> IO (Backend, [DP.NodeId]) +getMasterBackend i addrs rtable = + do let addr = addrs !! i + (host, port) = splitAddr addr + slaves = map makeNodeId $ deleteAt i addrs + backend <- initializeBackend host port rtable + return (backend, slaves) + +getClusterConf :: IO [String] +getClusterConf = fmap addresses $ readFile "cluster.conf" + +addresses :: String -> [String] +addresses = + filter (\x -> head x /= '#') . + filter (\x -> not $ null $ filter (\y -> y /= ' ') x) . + lines + +getSlaveConfBackend :: Int -> DP.RemoteTable -> IO Backend +getSlaveConfBackend i rtable = + do addrs <- getClusterConf + getSlaveBackend i addrs rtable + +getMasterConfBackend :: Int -> DP.RemoteTable -> IO (Backend, [DP.NodeId]) +getMasterConfBackend i rtable = + do addrs <- getClusterConf + getMasterBackend i addrs rtable + +startMasterProcess :: Backend -> [DP.NodeId] -> (Backend -> [DP.NodeId] -> DP.Process ()) -> IO () +startMasterProcess backend slaves process = + do node <- newLocalNode backend + Node.runProcess node (process backend slaves) diff --git a/tests/cluster.conf b/tests/cluster.conf new file mode 100644 --- /dev/null +++ b/tests/cluster.conf @@ -0,0 +1,13 @@ +# The Simulation Cluster Configuration + +# 0: the master simulation node, which must be run last +localhost:8088 + +# 1: the time server +localhost:8080 + +# 2: the first slave simulation node +localhost:8081 + +# 3: the second slave simulation node +localhost:8082