aivika-distributed 0.7 → 0.7.1.1

raw patch · 9 files changed

+350/−160 lines, 9 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +7/−1
• Simulation/Aivika/Distributed/Optimistic/DIO.hs +1/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +113/−34
• Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +10/−18
• Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs +43/−43
• Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs +133/−22
• Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs +41/−41
• Simulation/Aivika/Distributed/Optimistic/TimeServer.hs +1/−0
• aivika-distributed.cabal +1/−1

@@ -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. 

@@ -14,6 +14,7 @@ module Simulation.Aivika.Distributed.Optimistic.DIO        (DIO,         DIOParams(..),+        DIOStrategy(..),         runDIO,         defaultDIOParams,         dioParams,

@@ -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 ()

@@ -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') $

@@ -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

@@ -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 ()

@@ -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

@@ -11,6 +11,7 @@ -- module Simulation.Aivika.Distributed.Optimistic.TimeServer        (TimeServerParams(..),+        TimeServerStrategy(..),         defaultTimeServerParams,         timeServer,         curryTimeServer) where

@@ -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.