diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,10 @@ +Version 0.7.1 +----- + +* Added the time server and logical process strategies to shutdown the cluster + in case of failure by the specified timeout intervals. + Version 0.7 ----- @@ -30,7 +36,7 @@ * Started using Samadi's algorithm to synchronize the global virtual time. -* The local processes must call registerDIO to connect to the time server. +* The logical processes must call registerDIO to connect to the time server. * Increased the default synchronization time-out and delay. 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 @@ -14,6 +14,7 @@ module Simulation.Aivika.Distributed.Optimistic.DIO (DIO, DIOParams(..), + DIOStrategy(..), runDIO, defaultDIOParams, dioParams, 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 @@ -14,6 +14,7 @@ module Simulation.Aivika.Distributed.Optimistic.Internal.DIO (DIO(..), DIOParams(..), + DIOStrategy(..), invokeDIO, runDIO, defaultDIOParams, @@ -27,8 +28,8 @@ timeServerId, sendMessageDIO, sendMessagesDIO, - sendAcknowledgmentMessageDIO, - sendAcknowledgmentMessagesDIO, + sendAcknowledgementMessageDIO, + sendAcknowledgementMessagesDIO, sendLocalTimeDIO, sendRequestGlobalTimeDIO, logDIO, @@ -37,6 +38,7 @@ import Data.Typeable import Data.Binary import Data.IORef +import Data.Time.Clock import GHC.Generics @@ -87,12 +89,23 @@ -- ^ 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 + dioTimeServerAcknowledgementTimeout :: Int, + -- ^ The timeout in microseconds for receiving an acknowledgement message from the time server + dioStrategy :: DIOStrategy + -- ^ The logical process strategy } deriving (Eq, Ord, Show, Typeable, Generic) instance Binary DIOParams +-- | The logical process strategy. +data DIOStrategy = WaitIndefinitelyForTimeServer + -- ^ Wait for the time server forever + | TerminateDueToTimeServerTimeout Int + -- ^ Terminate due to the exceeded time server timeout in microseconds, but not less than 'dioSyncTimeout' + deriving (Eq, Ord, Show, Typeable, Generic) + +instance Binary DIOStrategy + -- | The distributed computation based on 'IO'. newtype DIO a = DIO { unDIO :: DIOContext -> DP.Process a -- ^ Unwrap the computation. @@ -174,7 +187,8 @@ dioProcessReconnectingEnabled = False, dioProcessReconnectingDelay = 5000000, dioKeepAliveInterval = 5000000, - dioTimeServerAcknowledgmentTimeout = 5000000 + dioTimeServerAcknowledgementTimeout = 5000000, + dioStrategy = TerminateDueToTimeServerTimeout 300000000 } -- | Return the computation context. @@ -211,7 +225,7 @@ then DP.send inbox (SendTerminateTimeServerMessage receiver sender) else DP.send receiver (TerminateTimeServerMessage sender) liftIO $ - timeout (dioTimeServerAcknowledgmentTimeout ps) $ + timeout (dioTimeServerAcknowledgementTimeout ps) $ atomically $ do f <- readTVar (dioTimeServerTerminating ctx) unless f retry @@ -232,7 +246,7 @@ then DP.send inbox (SendRegisterLogicalProcessMessage receiver sender) else DP.send receiver (RegisterLogicalProcessMessage sender) liftIO $ - timeout (dioTimeServerAcknowledgmentTimeout ps) $ + timeout (dioTimeServerAcknowledgementTimeout ps) $ atomically $ do f <- readTVar (dioRegisteredInTimeServer ctx) unless f retry @@ -253,7 +267,7 @@ then DP.send inbox (SendUnregisterLogicalProcessMessage receiver sender) else DP.send receiver (UnregisterLogicalProcessMessage sender) liftIO $ - timeout (dioTimeServerAcknowledgmentTimeout ps) $ + timeout (dioTimeServerAcknowledgementTimeout ps) $ atomically $ do f <- readTVar (dioUnregisteredFromTimeServer ctx) unless f retry @@ -291,6 +305,7 @@ registeredInTimeServer <- liftIO $ newTVarIO False unregisteredFromTimeServer <- liftIO $ newTVarIO False timeServerTerminating <- liftIO $ newTVarIO False + timeServerTimestamp <- liftIO $ getCurrentTime >>= newIORef let loop = forever $ do let f1 :: LogicalProcessMessage -> DP.Process InternalLogicalProcessMessage @@ -305,10 +320,11 @@ case x of Nothing -> return () Just (InternalLogicalProcessMessage m) -> - liftIO $ - writeChannel ch m + do processTimeServerMessage m serverId timeServerTimestamp + liftIO $ + writeChannel ch m Just (InternalProcessMonitorNotification m@(DP.ProcessMonitorNotification _ _ _)) -> - handleProcessMonitorNotification m ps ch + handleProcessMonitorNotification m ps ch serverId Just (InternalInboxProcessMessage m) -> case m of SendQueueMessage pid m -> @@ -316,11 +332,11 @@ SendQueueMessageBulk pid ms -> forM_ ms $ \m -> DP.send pid (QueueMessage m) - SendAcknowledgmentQueueMessage pid m -> - DP.send pid (AcknowledgmentQueueMessage m) - SendAcknowledgmentQueueMessageBulk pid ms -> + SendAcknowledgementQueueMessage pid m -> + DP.send pid (AcknowledgementQueueMessage m) + SendAcknowledgementQueueMessageBulk pid ms -> forM_ ms $ \m -> - DP.send pid (AcknowledgmentQueueMessage m) + DP.send pid (AcknowledgementQueueMessage m) SendLocalTimeMessage receiver sender t -> DP.send receiver (LocalTimeMessage sender t) SendRequestGlobalTimeMessage receiver sender -> @@ -343,21 +359,21 @@ handleReMonitorProcessMessage pids ps ch TrySendKeepAliveMessage -> trySendKeepAlive keepAliveManager - RegisterLogicalProcessAcknowledgmentMessage pid -> + RegisterLogicalProcessAcknowledgementMessage pid -> do --- logProcess ps INFO "Registered the logical process in the time server" --- liftIO $ atomically $ writeTVar registeredInTimeServer True - UnregisterLogicalProcessAcknowledgmentMessage pid -> + UnregisterLogicalProcessAcknowledgementMessage pid -> do --- logProcess ps INFO "Unregistered the logical process from the time server" --- liftIO $ atomically $ writeTVar unregisteredFromTimeServer True - TerminateTimeServerAcknowledgmentMessage pid -> + TerminateTimeServerAcknowledgementMessage pid -> do --- logProcess ps INFO "Started terminating the time server" --- @@ -375,8 +391,7 @@ do --- logProcess ps DEBUG $ "Received " ++ show m --- - liftIO $ - writeChannel ch (KeepAliveLogicalProcessMessage m) + return () inboxId <- DP.spawnLocal $ C.catch loop (handleException ps) @@ -389,6 +404,20 @@ DP.send inboxId TrySendKeepAliveMessage loop in C.catch loop (handleException ps) + DP.spawnLocal $ + let stop = + do f <- liftIO $ readIORef terminated + return (f || dioStrategy ps == WaitIndefinitelyForTimeServer) + loop = + do f <- stop + unless f $ + do liftIO $ + threadDelay (dioSyncTimeout ps) + f <- stop + unless f $ + do validateTimeServer ps inboxId timeServerTimestamp + loop + in C.catch loop (handleException ps) let simulation = unDIO m DIOContext { dioChannel = ch, dioInboxId = inboxId, @@ -400,8 +429,8 @@ return (inboxId, simulation) -- | Handle the process monitor notification. -handleProcessMonitorNotification :: DP.ProcessMonitorNotification -> DIOParams -> Channel LogicalProcessMessage -> DP.Process () -handleProcessMonitorNotification m@(DP.ProcessMonitorNotification _ pid0 reason) ps ch = +handleProcessMonitorNotification :: DP.ProcessMonitorNotification -> DIOParams -> Channel LogicalProcessMessage -> DP.ProcessId -> DP.Process () +handleProcessMonitorNotification m@(DP.ProcessMonitorNotification _ pid0 reason) ps ch serverId = do let recv m@(DP.ProcessMonitorNotification _ _ _) = do --- logProcess ps WARNING $ "Received a process monitor notification " ++ show m @@ -410,6 +439,12 @@ writeChannel ch (ProcessMonitorNotificationMessage m) return m recv m + when (pid0 == serverId) $ + case reason of + DP.DiedNormal -> processTimeServerTerminated ps serverId + DP.DiedException _ -> processTimeServerTerminated ps serverId + DP.DiedNodeDown -> processTimeServerTerminated ps serverId + _ -> return () when (dioProcessReconnectingEnabled ps && (reason == DP.DiedDisconnect)) $ do liftIO $ threadDelay (dioProcessReconnectingDelay ps) @@ -467,6 +502,50 @@ else liftDistributedUnsafe $ logProcess ps WARNING "Ignored the process monitoring as it was disabled in the DIO computation parameters" +-- | Process the time server message in a stream of messages destined for the logical process. +processTimeServerMessage :: LogicalProcessMessage -> DP.ProcessId -> IORef UTCTime -> DP.Process () +processTimeServerMessage ComputeLocalTimeMessage serverId r = + do liftIO $ + getCurrentTime >>= writeIORef r + inboxId <- DP.getSelfPid + DP.send serverId (ComputeLocalTimeAcknowledgementMessage inboxId) +processTimeServerMessage (GlobalTimeMessage _) serverId r = + liftIO $ + getCurrentTime >>= writeIORef r +processTimeServerMessage _ serverId r = + return () + +-- | Validate the time server by the specified inbox and recent timestamp. +validateTimeServer :: DIOParams -> DP.ProcessId -> IORef UTCTime -> DP.Process () +validateTimeServer ps inboxId r = + do --- + logProcess ps NOTICE "Validating the time server" + --- + case dioStrategy ps of + WaitIndefinitelyForTimeServer -> + return () + TerminateDueToTimeServerTimeout timeout -> + do utc0 <- liftIO $ readIORef r + utc <- liftIO getCurrentTime + let dt = fromRational $ toRational (diffUTCTime utc utc0) + when (secondsToMicroseconds dt > timeout) $ + do --- + logProcess ps WARNING "Terminating due to the exceeded time server timeout" + --- + DP.send inboxId TerminateInboxProcessMessage + +-- | Process the situation when the time server has suddenly been terminated. +processTimeServerTerminated :: DIOParams -> DP.ProcessId -> DP.Process () +processTimeServerTerminated ps inboxId = + do --- + logProcess ps NOTICE "Terminating due to sudden termination of the time server" + --- + DP.send inboxId TerminateInboxProcessMessage + +-- | Convert seconds to microseconds. +secondsToMicroseconds :: Double -> Int +secondsToMicroseconds x = fromInteger $ toInteger $ round (1000000 * x) + -- | Send the message. sendMessageDIO :: DP.ProcessId -> Message -> DIO () {-# INLINABLE sendMessageDIO #-} @@ -492,30 +571,30 @@ liftDistributedUnsafe $ DP.send pid (QueueMessage m) --- | Send the acknowledgment message. -sendAcknowledgmentMessageDIO :: DP.ProcessId -> AcknowledgmentMessage -> DIO () -{-# INLINABLE sendAcknowledgmentMessageDIO #-} -sendAcknowledgmentMessageDIO pid m = +-- | Send the acknowledgement message. +sendAcknowledgementMessageDIO :: DP.ProcessId -> AcknowledgementMessage -> DIO () +{-# INLINABLE sendAcknowledgementMessageDIO #-} +sendAcknowledgementMessageDIO pid m = do ps <- dioParams if dioProcessMonitoringEnabled ps then do inbox <- messageInboxId liftDistributedUnsafe $ - DP.send inbox (SendAcknowledgmentQueueMessage pid m) + DP.send inbox (SendAcknowledgementQueueMessage pid m) else liftDistributedUnsafe $ - DP.send pid (AcknowledgmentQueueMessage m) + DP.send pid (AcknowledgementQueueMessage m) --- | Send the bulk of acknowledgment messages. -sendAcknowledgmentMessagesDIO :: DP.ProcessId -> [AcknowledgmentMessage] -> DIO () -{-# INLINABLE sendAcknowledgmentMessagesDIO #-} -sendAcknowledgmentMessagesDIO pid ms = +-- | Send the bulk of acknowledgement messages. +sendAcknowledgementMessagesDIO :: DP.ProcessId -> [AcknowledgementMessage] -> DIO () +{-# INLINABLE sendAcknowledgementMessagesDIO #-} +sendAcknowledgementMessagesDIO pid ms = do ps <- dioParams if dioProcessMonitoringEnabled ps then do inbox <- messageInboxId liftDistributedUnsafe $ - DP.send inbox (SendAcknowledgmentQueueMessageBulk pid ms) + DP.send inbox (SendAcknowledgementQueueMessageBulk pid ms) else do forM_ ms $ \m -> liftDistributedUnsafe $ - DP.send pid (AcknowledgmentQueueMessage m) + DP.send pid (AcknowledgementQueueMessage m) -- | Send the local time to the time server. sendLocalTimeDIO :: DP.ProcessId -> DP.ProcessId -> Double -> 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 @@ -331,7 +331,7 @@ --- logMessage x --- infind <- liftIOUnsafe $ readIORef (queueInFind q) - deliverAcknowledgmentMessage (acknowledgmentMessage infind m) + deliverAcknowledgementMessage (acknowledgementMessage infind m) t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q) p' <- invokeEvent p currentEventPoint if messageReceiveTime m < t0 @@ -349,7 +349,7 @@ --- logMessage x --- infind <- liftIOUnsafe $ readIORef (queueInFind q) - deliverAcknowledgmentMessages $ map (acknowledgmentMessage infind) ms + deliverAcknowledgementMessages $ map (acknowledgementMessage infind) ms t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q) forM_ ms $ \m -> do p' <- invokeEvent p currentEventPoint @@ -360,7 +360,7 @@ else error "Received the outdated message: processChannelMessage" else invokeTimeWarp p' $ enqueueMessage (queueInputMessages q) m -processChannelMessage x@(AcknowledgmentQueueMessage m) = +processChannelMessage x@(AcknowledgementQueueMessage m) = TimeWarp $ \p -> do let q = runEventQueue $ pointRun p --- @@ -368,8 +368,8 @@ --- logMessage x --- liftIOUnsafe $ - processAcknowledgmentMessage (queueTransientMessages q) m -processChannelMessage x@(AcknowledgmentQueueMessageBulk ms) = + processAcknowledgementMessage (queueTransientMessages q) m +processChannelMessage x@(AcknowledgementQueueMessageBulk ms) = TimeWarp $ \p -> do let q = runEventQueue $ pointRun p --- @@ -378,7 +378,7 @@ --- liftIOUnsafe $ forM_ ms $ - processAcknowledgmentMessage (queueTransientMessages q) + processAcknowledgementMessage (queueTransientMessages q) processChannelMessage x@ComputeLocalTimeMessage = TimeWarp $ \p -> do let q = runEventQueue $ pointRun p @@ -401,7 +401,7 @@ --- liftIOUnsafe $ do writeIORef (queueInFind q) False - resetAcknowledgmentMessageTime (queueTransientMessages q) + resetAcknowledgementMessageTime (queueTransientMessages q) invokeEvent p $ updateGlobalTime globalTime processChannelMessage x@(ProcessMonitorNotificationMessage y@(DP.ProcessMonitorNotification _ pid reason)) = @@ -422,14 +422,6 @@ --- invokeEvent p $ reconnectProcess pid -processChannelMessage x@(KeepAliveLogicalProcessMessage m) = - TimeWarp $ \p -> - do let q = runEventQueue $ pointRun p - --- - --- invokeEvent p $ - --- logMessage x - --- - return () -- | Return the local minimum time. getLocalTime :: Event DIO Double @@ -438,7 +430,7 @@ do let q = runEventQueue $ pointRun p t1 <- liftIOUnsafe $ readIORef (queueTime q) t2 <- liftIOUnsafe $ transientMessageQueueTime (queueTransientMessages q) - t3 <- liftIOUnsafe $ acknowledgmentMessageTime (queueTransientMessages q) + t3 <- liftIOUnsafe $ acknowledgementMessageTime (queueTransientMessages q) let t' = t1 `min` t2 `min` t3 --- --- n <- liftIOUnsafe $ transientMessageQueueSize (queueTransientMessages q) @@ -746,10 +738,10 @@ infind <- liftIOUnsafe $ readIORef (queueInFind q) let ys = queueInputMessages q ys' <- liftIOUnsafe $ - fmap (map $ acknowledgmentMessage infind) $ + fmap (map $ acknowledgementMessage infind) $ filterInputMessages (\x -> messageSenderId x == pid) ys unless (null ys') $ - sendAcknowledgmentMessagesDIO pid ys' + sendAcknowledgementMessagesDIO pid ys' xs <- liftIOUnsafe $ transientMessages (queueTransientMessages q) let xs' = filter (\x -> messageReceiverId x == pid) xs unless (null xs') $ 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 @@ -15,8 +15,8 @@ (Message(..), antiMessage, antiMessages, - AcknowledgmentMessage(..), - acknowledgmentMessage, + AcknowledgementMessage(..), + acknowledgementMessage, LogicalProcessMessage(..), TimeServerMessage(..), InboxProcessMessage(..), @@ -75,46 +75,46 @@ (messageAntiToggle x == messageAntiToggle y) -- | Represents an acknowledgement message. -data AcknowledgmentMessage = - AcknowledgmentMessage { acknowledgmentSequenceNo :: Int, - -- ^ The sequence number. - acknowledgmentSendTime :: Double, - -- ^ The send time. - acknowledgmentReceiveTime :: Double, - -- ^ The receive time. - acknowledgmentSenderId :: DP.ProcessId, - -- ^ The sender of the source message. - acknowledgmentReceiverId :: DP.ProcessId, - -- ^ The receiver of the source message. - acknowledgmentAntiToggle :: Bool, - -- ^ Whether this is an anti-message acknowledgment. - acknowledgmentMarked :: Bool - -- ^ Whether the acknowledgment is marked. - } deriving (Eq, Ord, Show, Typeable, Generic) +data AcknowledgementMessage = + AcknowledgementMessage { acknowledgementSequenceNo :: Int, + -- ^ The sequence number. + acknowledgementSendTime :: Double, + -- ^ The send time. + acknowledgementReceiveTime :: Double, + -- ^ The receive time. + acknowledgementSenderId :: DP.ProcessId, + -- ^ The sender of the source message. + acknowledgementReceiverId :: DP.ProcessId, + -- ^ The receiver of the source message. + acknowledgementAntiToggle :: Bool, + -- ^ Whether this is an anti-message acknowledgement. + acknowledgementMarked :: Bool + -- ^ Whether the acknowledgement is marked. + } deriving (Eq, Ord, Show, Typeable, Generic) -instance Binary AcknowledgmentMessage +instance Binary AcknowledgementMessage --- | Create an acknowledgment message specifying whether it will be marked. -acknowledgmentMessage :: Bool -> Message -> AcknowledgmentMessage -acknowledgmentMessage marked x = - AcknowledgmentMessage { acknowledgmentSequenceNo = messageSequenceNo x, - acknowledgmentSendTime = messageSendTime x, - acknowledgmentReceiveTime = messageReceiveTime x, - acknowledgmentSenderId = messageSenderId x, - acknowledgmentReceiverId = messageReceiverId x, - acknowledgmentAntiToggle = messageAntiToggle x, - acknowledgmentMarked = marked - } +-- | Create an acknowledgement message specifying whether it will be marked. +acknowledgementMessage :: Bool -> Message -> AcknowledgementMessage +acknowledgementMessage marked x = + AcknowledgementMessage { acknowledgementSequenceNo = messageSequenceNo x, + acknowledgementSendTime = messageSendTime x, + acknowledgementReceiveTime = messageReceiveTime x, + acknowledgementSenderId = messageSenderId x, + acknowledgementReceiverId = messageReceiverId x, + acknowledgementAntiToggle = messageAntiToggle x, + acknowledgementMarked = marked + } -- | The message sent to the logical process. data LogicalProcessMessage = QueueMessage Message -- ^ the message has come from the remote process | QueueMessageBulk [Message] -- ^ a bulk of messages that have come from the remote process - | AcknowledgmentQueueMessage AcknowledgmentMessage - -- ^ the acknowledgment message has come from the remote process - | AcknowledgmentQueueMessageBulk [AcknowledgmentMessage] - -- ^ a bulk of acknowledgment messages that have come from the remote process + | AcknowledgementQueueMessage AcknowledgementMessage + -- ^ the acknowledgement message has come from the remote process + | AcknowledgementQueueMessageBulk [AcknowledgementMessage] + -- ^ a bulk of acknowledgement messages that have come from the remote process | ComputeLocalTimeMessage -- ^ the time server requests for a local minimum time | GlobalTimeMessage Double @@ -123,8 +123,6 @@ -- ^ the process monitor notification | ReconnectProcessMessage DP.ProcessId -- ^ finish reconnecting to the specified process - | KeepAliveLogicalProcessMessage KeepAliveMessage - -- ^ the keep alive message for the logical process deriving (Show, Typeable, Generic) instance Binary LogicalProcessMessage @@ -136,6 +134,8 @@ -- ^ unregister the logical process from the time server | TerminateTimeServerMessage DP.ProcessId -- ^ the logical process asked to terminate the time server + | ComputeLocalTimeAcknowledgementMessage DP.ProcessId + -- ^ the acknowledgement message after the server asked to compute the local time | RequestGlobalTimeMessage DP.ProcessId -- ^ the logical process requested for the global minimum time | LocalTimeMessage DP.ProcessId Double @@ -157,10 +157,10 @@ -- ^ 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 + | SendAcknowledgementQueueMessage DP.ProcessId AcknowledgementMessage + -- ^ send an acknowledgement message via the inbox process + | SendAcknowledgementQueueMessageBulk DP.ProcessId [AcknowledgementMessage] + -- ^ send a bulk of acknowledgement 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 @@ -171,11 +171,11 @@ -- ^ unregister the logical process from the time server | SendTerminateTimeServerMessage DP.ProcessId DP.ProcessId -- ^ the logical process asked to terminate the time server - | RegisterLogicalProcessAcknowledgmentMessage DP.ProcessId + | RegisterLogicalProcessAcknowledgementMessage DP.ProcessId -- ^ after registering the logical process in the time server - | UnregisterLogicalProcessAcknowledgmentMessage DP.ProcessId + | UnregisterLogicalProcessAcknowledgementMessage DP.ProcessId -- ^ after unregistering the logical process from the time server - | TerminateTimeServerAcknowledgmentMessage DP.ProcessId + | TerminateTimeServerAcknowledgementMessage DP.ProcessId -- ^ after started terminating the time server | TerminateInboxProcessMessage -- ^ terminate the inbox process 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 @@ -13,6 +13,7 @@ -- module Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer (TimeServerParams(..), + TimeServerStrategy(..), defaultTimeServerParams, timeServer, curryTimeServer) where @@ -48,17 +49,36 @@ tsTimeSyncDelay :: Int, -- ^ the delay in microseconds between the time synchronization sessions tsProcessMonitoringEnabled :: Bool, - -- ^ Whether the process monitoring is enabled + -- ^ whether the process monitoring is enabled tsProcessMonitoringDelay :: Int, - -- ^ The delay in microseconds which must be applied for monitoring every remote process. + -- ^ 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 + -- ^ whether the automatic reconnecting to processes is enabled when enabled monitoring + tsProcessReconnectingDelay :: Int, -- ^ the delay in microseconds before reconnecting + tsStrategy :: TimeServerStrategy + -- ^ the time server strategy } deriving (Eq, Ord, Show, Typeable, Generic) instance Binary TimeServerParams +-- | The time server strategy. +data TimeServerStrategy = WaitIndefinitelyForLogicalProcess + -- ^ wait for the logical process forever + | TerminateDueToLogicalProcessTimeout Int + -- ^ terminate the server due to the exceeded logical process timeout in microseconds, + -- but not less than 'tsTimeSyncTimeout', which is a much more preferable option than + -- significantly more risky 'UnregisterLogicalProcessDueToTimeout' + | UnregisterLogicalProcessDueToTimeout Int + -- ^ unregister the logical process due to the exceeded timeout in microseconds, + -- but not less than 'tsTimeSyncTimeout', which is a very risky option as there can be + -- un-acknowledged messages by the just unregistered logical process that might shutdown, + -- which would keep the global virtual time on the same value even if the existent + -- logical processes had another local time + deriving (Eq, Ord, Show, Typeable, Generic) + +instance Binary TimeServerStrategy + -- | The time server. data TimeServer = TimeServer { tsParams :: TimeServerParams, @@ -75,14 +95,20 @@ -- ^ the processed used in the current finding of the global time tsGlobalTime :: IORef (Maybe Double), -- ^ the global time of the model - tsGlobalTimeTimestamp :: IORef (Maybe UTCTime) + tsGlobalTimeTimestamp :: IORef (Maybe UTCTime), -- ^ the global time timestamp + tsLogicalProcessValidationTimestamp :: IORef UTCTime + -- ^ the logical process validation timestamp } -- | The information about the logical process. data LogicalProcessInfo = - LogicalProcessInfo { lpLocalTime :: IORef (Maybe Double), + LogicalProcessInfo { lpId :: DP.ProcessId, + -- ^ the logical process identifier + lpLocalTime :: IORef (Maybe Double), -- ^ the local time of the process + lpTimestamp :: IORef UTCTime, + -- ^ the logical process timestamp lpMonitorRef :: Maybe DP.MonitorRef -- ^ the logical process monitor reference } @@ -97,7 +123,8 @@ tsProcessMonitoringEnabled = False, tsProcessMonitoringDelay = 3000000, tsProcessReconnectingEnabled = False, - tsProcessReconnectingDelay = 5000000 + tsProcessReconnectingDelay = 5000000, + tsStrategy = TerminateDueToLogicalProcessTimeout 300000000 } -- | Create a new time server by the specified initial quorum and parameters. @@ -109,6 +136,7 @@ s <- newIORef S.empty t0 <- newIORef Nothing t' <- newIORef Nothing + t2 <- getCurrentTime >>= newIORef return TimeServer { tsParams = ps, tsInitQuorum = n, tsInInit = f, @@ -116,7 +144,8 @@ tsProcesses = m, tsProcessesInFind = s, tsGlobalTime = t0, - tsGlobalTimeTimestamp = t' + tsGlobalTimeTimestamp = t', + tsLogicalProcessValidationTimestamp = t2 } -- | Process the time server message. @@ -131,8 +160,9 @@ "Time Server: already registered process identifier " ++ show pid Nothing -> do t <- newIORef Nothing + utc <- getCurrentTime >>= newIORef modifyIORef (tsProcesses server) $ - M.insert pid LogicalProcessInfo { lpLocalTime = t, lpMonitorRef = Nothing } + M.insert pid LogicalProcessInfo { lpId = pid, lpLocalTime = t, lpTimestamp = utc, lpMonitorRef = Nothing } return $ do when (tsProcessMonitoringEnabled $ tsParams server) $ do logTimeServer server INFO $ @@ -142,7 +172,7 @@ modifyIORef (tsProcesses server) $ M.update (\x -> Just x { lpMonitorRef = Just r }) pid serverId <- DP.getSelfPid - DP.send pid (RegisterLogicalProcessAcknowledgmentMessage serverId) + DP.send pid (RegisterLogicalProcessAcknowledgementMessage serverId) tryStartTimeServer server processTimeServerMessage server (UnregisterLogicalProcessMessage pid) = join $ liftIO $ @@ -166,7 +196,7 @@ "Time Server: unmonitoring the process by identifier " ++ show pid DP.unmonitor r serverId <- DP.getSelfPid - DP.send pid (UnregisterLogicalProcessAcknowledgmentMessage serverId) + DP.send pid (UnregisterLogicalProcessAcknowledgementMessage serverId) tryProvideTimeServerGlobalTime server tryTerminateTimeServer server processTimeServerMessage server (TerminateTimeServerMessage pid) = @@ -191,7 +221,7 @@ "Time Server: unmonitoring the process by identifier " ++ show pid DP.unmonitor r serverId <- DP.getSelfPid - DP.send pid (TerminateTimeServerAcknowledgmentMessage serverId) + DP.send pid (TerminateTimeServerAcknowledgementMessage serverId) startTerminatingTimeServer server processTimeServerMessage server (RequestGlobalTimeMessage pid) = tryComputeTimeServerGlobalTime server @@ -206,11 +236,28 @@ processTimeServerMessage server (RegisterLogicalProcessMessage pid) processTimeServerMessage server (LocalTimeMessage pid t') Just x -> - do writeIORef (lpLocalTime x) (Just t') + do utc <- getCurrentTime + writeIORef (lpLocalTime x) (Just t') + writeIORef (lpTimestamp x) utc modifyIORef (tsProcessesInFind server) $ S.delete pid return $ tryProvideTimeServerGlobalTime server +processTimeServerMessage server (ComputeLocalTimeAcknowledgementMessage pid) = + join $ liftIO $ + do m <- readIORef (tsProcesses server) + case M.lookup pid m of + Nothing -> + return $ + do logTimeServer server WARNING $ + "Time Server: unknown process identifier " ++ show pid + processTimeServerMessage server (RegisterLogicalProcessMessage pid) + processTimeServerMessage server (ComputeLocalTimeAcknowledgementMessage pid) + Just x -> + do utc <- getCurrentTime + writeIORef (lpTimestamp x) utc + return $ + return () processTimeServerMessage server (ReMonitorTimeServerMessage pids) = do forM_ pids $ \pid -> do --- @@ -341,6 +388,27 @@ Just _ -> loop zs' (liftM2 min t acc) +-- | Return a logical process with the minimal timestamp. +minTimestampLogicalProcess :: TimeServer -> IO (Maybe LogicalProcessInfo) +minTimestampLogicalProcess server = + do zs <- fmap M.assocs $ readIORef (tsProcesses server) + case zs of + [] -> return Nothing + ((pid, x) : zs') -> loop zs x + where loop [] acc = return (Just acc) + loop ((pid, x) : zs') acc = + do t0 <- readIORef (lpTimestamp acc) + t <- readIORef (lpTimestamp x) + if t0 <= t + then loop zs' acc + else loop zs' x + +-- | Filter the logical processes. +filterLogicalProcesses :: TimeServer -> [DP.ProcessId] -> IO [DP.ProcessId] +filterLogicalProcesses server pids = + do xs <- readIORef (tsProcesses server) + return $ filter (\pid -> M.member pid xs) pids + -- | Start terminating the time server. startTerminatingTimeServer :: TimeServer -> DP.Process () startTerminatingTimeServer server = @@ -411,12 +479,15 @@ --- return () utc <- liftIO getCurrentTime + validation <- liftIO $ readIORef (tsLogicalProcessValidationTimestamp server) timestamp <- liftIO $ readIORef (tsGlobalTimeTimestamp server) + when (timeSyncTimeoutExceeded server validation utc) $ + validateLogicalProcesses server utc case timestamp of - Just x | shouldResetComputingTimeServerGlobalTime server x utc -> + Just x | timeSyncTimeoutExceeded server x utc -> resetComputingTimeServerGlobalTime server _ -> return () - if shouldComputeTimeServerGlobalTime server utc0 utc + if timeSyncDelayExceeded server utc0 utc then do tryComputeTimeServerGlobalTime server loop utc else loop utc0 @@ -443,7 +514,7 @@ case y of Nothing -> return $ reverse acc Just m@(DP.ProcessMonitorNotification _ pid _) -> loop (pid : acc) - pids <- loop [pid0] + pids <- loop [pid0] >>= (liftIO . filterLogicalProcesses server) --- logTimeServer server NOTICE "Begin reconnecting..." --- @@ -465,17 +536,57 @@ 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 = +-- | Test whether the sychronization delay has been exceeded. +timeSyncDelayExceeded :: TimeServer -> UTCTime -> UTCTime -> Bool +timeSyncDelayExceeded 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 = +-- | Test whether the synchronization timeout has been exceeded. +timeSyncTimeoutExceeded :: TimeServer -> UTCTime -> UTCTime -> Bool +timeSyncTimeoutExceeded server utc0 utc = let dt = fromRational $ toRational (diffUTCTime utc utc0) in secondsToMicroseconds dt > (tsTimeSyncTimeout $ tsParams server) + +-- | Get the difference between the specified time and the logical process timestamp. +diffLogicalProcessTimestamp :: UTCTime -> LogicalProcessInfo -> IO Int +diffLogicalProcessTimestamp utc lp = + do utc0 <- readIORef (lpTimestamp lp) + let dt = fromRational $ toRational (diffUTCTime utc utc0) + return $ secondsToMicroseconds dt + +-- | Validate the logical processes. +validateLogicalProcesses :: TimeServer -> UTCTime -> DP.Process () +validateLogicalProcesses server utc = + do logTimeServer server NOTICE $ + "Time Server: validating the logical processes" + liftIO $ + writeIORef (tsLogicalProcessValidationTimestamp server) utc + case tsStrategy (tsParams server) of + WaitIndefinitelyForLogicalProcess -> + return () + TerminateDueToLogicalProcessTimeout timeout -> + do x <- liftIO $ minTimestampLogicalProcess server + case x of + Just lp -> + do diff <- liftIO $ diffLogicalProcessTimestamp utc lp + when (diff > timeout) $ + do logTimeServer server WARNING $ + "Time Server: terminating due to the exceeded logical process timeout" + DP.terminate + Nothing -> + return () + UnregisterLogicalProcessDueToTimeout timeout -> + do x <- liftIO $ minTimestampLogicalProcess server + case x of + Just lp -> + do diff <- liftIO $ diffLogicalProcessTimestamp utc lp + when (diff > timeout) $ + do logTimeServer server WARNING $ + "Time Server: unregistering the logical process due to the exceeded timeout" + processTimeServerMessage server (UnregisterLogicalProcessMessage $ lpId lp) + Nothing -> + return () -- | 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/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 @@ -16,11 +16,11 @@ transientMessageQueueTime, transientMessages, enqueueTransientMessage, - processAcknowledgmentMessage, - acknowledgmentMessageTime, - resetAcknowledgmentMessageTime, - deliverAcknowledgmentMessage, - deliverAcknowledgmentMessages) where + processAcknowledgementMessage, + acknowledgementMessageTime, + resetAcknowledgementMessageTime, + deliverAcknowledgementMessage, + deliverAcknowledgementMessages) where import qualified Data.Map as M import Data.List @@ -39,7 +39,7 @@ TransientMessageQueue { queueTransientMessages :: IORef (M.Map TransientMessageQueueItem Message), -- ^ The transient messages. queueMarkedMessageTime :: IORef Double - -- ^ The marked acknowledgment message time. + -- ^ The marked acknowledgement message time. } -- | Represents an acknowledgement message representative. @@ -55,7 +55,7 @@ itemReceiverId :: DP.ProcessId, -- ^ The receiver of the source message. itemAntiToggle :: Bool - -- ^ Whether it was the anti-message acknowledgment. + -- ^ Whether it was the anti-message acknowledgement. } deriving (Eq, Show) instance Ord TransientMessageQueueItem where @@ -85,15 +85,15 @@ itemReceiverId = messageReceiverId m, itemAntiToggle = messageAntiToggle m } --- | Return a queue item by the specified acknowledgment message. -acknowledgmentMessageQueueItem :: AcknowledgmentMessage -> TransientMessageQueueItem -acknowledgmentMessageQueueItem m = - TransientMessageQueueItem { itemSequenceNo = acknowledgmentSequenceNo m, - itemSendTime = acknowledgmentSendTime m, - itemReceiveTime = acknowledgmentReceiveTime m, - itemSenderId = acknowledgmentSenderId m, - itemReceiverId = acknowledgmentReceiverId m, - itemAntiToggle = acknowledgmentAntiToggle m } +-- | Return a queue item by the specified acknowledgement message. +acknowledgementMessageQueueItem :: AcknowledgementMessage -> TransientMessageQueueItem +acknowledgementMessageQueueItem m = + TransientMessageQueueItem { itemSequenceNo = acknowledgementSequenceNo m, + itemSendTime = acknowledgementSendTime m, + itemReceiveTime = acknowledgementReceiveTime m, + itemSenderId = acknowledgementSenderId m, + itemReceiverId = acknowledgementReceiverId m, + itemAntiToggle = acknowledgementAntiToggle m } -- | Create a new transient message queue. newTransientMessageQueue :: DIO TransientMessageQueue @@ -129,43 +129,43 @@ modifyIORef (queueTransientMessages q) $ M.insert (transientMessageQueueItem m) m --- | Enqueue the acknowledgment message. -enqueueAcknowledgmentMessage :: TransientMessageQueue -> AcknowledgmentMessage -> IO () -enqueueAcknowledgmentMessage q m = +-- | Enqueue the acknowledgement message. +enqueueAcknowledgementMessage :: TransientMessageQueue -> AcknowledgementMessage -> IO () +enqueueAcknowledgementMessage q m = modifyIORef' (queueMarkedMessageTime q) $ - min (acknowledgmentReceiveTime m) + min (acknowledgementReceiveTime m) --- | Process the acknowledgment message. -processAcknowledgmentMessage :: TransientMessageQueue -> AcknowledgmentMessage -> IO () -processAcknowledgmentMessage q m = +-- | Process the acknowledgement message. +processAcknowledgementMessage :: TransientMessageQueue -> AcknowledgementMessage -> IO () +processAcknowledgementMessage q m = do ms <- readIORef (queueTransientMessages q) - let k = acknowledgmentMessageQueueItem m + let k = acknowledgementMessageQueueItem m when (M.member k ms) $ do modifyIORef (queueTransientMessages q) $ M.delete k - when (acknowledgmentMarked m) $ - enqueueAcknowledgmentMessage q m + when (acknowledgementMarked m) $ + enqueueAcknowledgementMessage q m --- | Get the minimal marked acknowledgment message time. -acknowledgmentMessageTime :: TransientMessageQueue -> IO Double -acknowledgmentMessageTime q = +-- | Get the minimal marked acknowledgement message time. +acknowledgementMessageTime :: TransientMessageQueue -> IO Double +acknowledgementMessageTime q = readIORef (queueMarkedMessageTime q) --- | Reset the marked acknowledgment message time. -resetAcknowledgmentMessageTime :: TransientMessageQueue -> IO () -resetAcknowledgmentMessageTime q = +-- | Reset the marked acknowledgement message time. +resetAcknowledgementMessageTime :: TransientMessageQueue -> IO () +resetAcknowledgementMessageTime q = writeIORef (queueMarkedMessageTime q) (1 / 0) --- | Deliver the acknowledgment message on low level. -deliverAcknowledgmentMessage :: AcknowledgmentMessage -> DIO () -deliverAcknowledgmentMessage x = - sendAcknowledgmentMessageDIO (acknowledgmentSenderId x) x +-- | Deliver the acknowledgement message on low level. +deliverAcknowledgementMessage :: AcknowledgementMessage -> DIO () +deliverAcknowledgementMessage x = + sendAcknowledgementMessageDIO (acknowledgementSenderId x) x --- | Deliver the acknowledgment messages on low level. -deliverAcknowledgmentMessages :: [AcknowledgmentMessage] -> DIO () -deliverAcknowledgmentMessages xs = - let ys = groupBy (\a b -> acknowledgmentSenderId a == acknowledgmentSenderId b) xs +-- | Deliver the acknowledgement messages on low level. +deliverAcknowledgementMessages :: [AcknowledgementMessage] -> DIO () +deliverAcknowledgementMessages xs = + let ys = groupBy (\a b -> acknowledgementSenderId a == acknowledgementSenderId b) xs dlv [] = return () dlv zs@(z : _) = - sendAcknowledgmentMessagesDIO (acknowledgmentSenderId z) zs + sendAcknowledgementMessagesDIO (acknowledgementSenderId z) zs in forM_ ys dlv 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 @@ -11,6 +11,7 @@ -- module Simulation.Aivika.Distributed.Optimistic.TimeServer (TimeServerParams(..), + TimeServerStrategy(..), defaultTimeServerParams, timeServer, curryTimeServer) where diff --git a/aivika-distributed.cabal b/aivika-distributed.cabal --- a/aivika-distributed.cabal +++ b/aivika-distributed.cabal @@ -1,5 +1,5 @@ name: aivika-distributed -version: 0.7 +version: 0.7.1.1 synopsis: Parallel distributed discrete event simulation module for the Aivika library description: This package extends the aivika-transformers [1] package and allows running parallel distributed simulations.