aivika-distributed 1.3 → 1.4

raw patch · 9 files changed

+177/−64 lines, 9 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +12/−0
• Simulation/Aivika/Distributed/Optimistic.hs +2/−0
• Simulation/Aivika/Distributed/Optimistic/Event.hs +16/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +9/−3
• Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +112/−46
• Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs +2/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs +5/−7
• Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs +8/−1
• aivika-distributed.cabal +11/−7

@@ -1,4 +1,16 @@ +Version 1.4+-----++* Added the leaveSimulation function that allows leaving the simulation prematurely.+  It was made in addition to an ability for new logical processes to enter the already +  running simulation.+  +* Added the dioProcessDisconnectingEnabled flag to allow logical processes to disconnect+  when the monitoring is enabled, but the time server strategy implies the unregistering of+  disconnected logical processes. But then none process can continue sending messages+  to the disconnected process.+ Version 1.3 ----- 

@@ -13,6 +13,7 @@ module Simulation.Aivika.Distributed.Optimistic        (-- * Modules         module Simulation.Aivika.Distributed.Optimistic.DIO,+        module Simulation.Aivika.Distributed.Optimistic.Event,         module Simulation.Aivika.Distributed.Optimistic.Guard,         module Simulation.Aivika.Distributed.Optimistic.Generator,         module Simulation.Aivika.Distributed.Optimistic.Message,@@ -23,6 +24,7 @@         module Simulation.Aivika.Distributed.Optimistic.TimeServer) where  import Simulation.Aivika.Distributed.Optimistic.DIO+import Simulation.Aivika.Distributed.Optimistic.Event import Simulation.Aivika.Distributed.Optimistic.Guard import Simulation.Aivika.Distributed.Optimistic.Generator import Simulation.Aivika.Distributed.Optimistic.Message

@@ -0,0 +1,16 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Event+-- Copyright  : Copyright (c) 2015-2018, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This module contains some 'Event' computations.+--+module Simulation.Aivika.Distributed.Optimistic.Event+       (leaveSimulation) where++import Simulation.Aivika.Trans.Event+import Simulation.Aivika.Distributed.Optimistic.Internal.Event

@@ -92,7 +92,7 @@               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+              -- ^ Whether the automatic reconnecting to processes is enabled when the monitoring is enabled               dioProcessReconnectingDelay :: Int,               -- ^ The delay in microseconds before reconnecting to the remote process               dioKeepAliveInterval :: Int,@@ -103,8 +103,10 @@               -- ^ The interval in microseconds between sending the simulation monitoring messages               dioSimulationMonitoringTimeout :: Int,               -- ^ The timeout in microseconds when processing the monitoring messages-              dioStrategy :: DIOStrategy+              dioStrategy :: DIOStrategy,               -- ^ The logical process strategy+              dioProcessDisconnectingEnabled :: Bool+              -- ^ Whether the process disconnecting is enabled when the monitoring is enabled but the reconnecting is disabled             } deriving (Eq, Ord, Show, Typeable, Generic)  instance Binary DIOParams@@ -210,7 +212,8 @@               dioTimeServerAcknowledgementTimeout = 5000000,               dioSimulationMonitoringInterval = 30000000,               dioSimulationMonitoringTimeout = 100000,-              dioStrategy = TerminateDueToTimeServerTimeout 300000000+              dioStrategy = TerminateDueToTimeServerTimeout 300000000,+              dioProcessDisconnectingEnabled = False             }  -- | The default environment parameters for the 'DIO' computation@@ -523,6 +526,9 @@                ---                liftIO $                  writeChannel ch (ReconnectProcessMessage pid)+     when (dioProcessDisconnectingEnabled ps && not (dioProcessReconnectingEnabled ps)) $+       liftIO $+       writeChannel ch (DisconnectProcessMessage pid0)  -- | Handle the general message. handleGeneralMessage :: GeneralMessage

@@ -16,7 +16,8 @@         queueOutputMessages,         queueLog,         expectEvent,-        processMonitorSignal) where+        processMonitorSignal,+        leaveSimulation) where  import Data.Maybe import Data.IORef@@ -82,8 +83,10 @@                  -- ^ the global time                  queueInFind :: IORef Bool,                  -- ^ whether the queue is in find mode-                 queueProcessMonitorNotificationSource :: SignalSource DIO DP.ProcessMonitorNotification+                 queueProcessMonitorNotificationSource :: SignalSource DIO DP.ProcessMonitorNotification,                  -- ^ the source of process monitor notifications+                 queueIsLeaving :: IORef Bool+                 -- ^ whether the logical process tries to leave the simulation                }    newEventQueue specs =@@ -98,6 +101,7 @@        ack <- newAcknowledgementMessageQueue        infind <- liftIOUnsafe $ newIORef False        s <- newDIOSignalSource0+       leaving <- liftIOUnsafe $ newIORef False        return EventQueue { queueInputMessages = input,                            queueOutputMessages = output,                            queueTransientMessages = transient,@@ -108,7 +112,8 @@                            queueTime = t,                            queueGlobalTime = gt,                            queueInFind = infind,-                           queueProcessMonitorNotificationSource = s }+                           queueProcessMonitorNotificationSource = s,+                           queueIsLeaving = leaving }    enqueueEvent t (Event m) =     Event $ \p ->@@ -477,6 +482,15 @@      invokeEvent p $        throwEvent $        SimulationAbort "Aborted by the outer process."+processChannelMessage x@(DisconnectProcessMessage pid) =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+     ---+     --- invokeEvent p $+     ---   logMessage x+     ---+     invokeEvent p $+       disconnectProcess pid  -- | Return the local minimum time. getLocalTime :: Event DIO Double@@ -499,6 +513,17 @@      ---      return t' +-- | Test whether the local time is ending.+isLocalTimeEnding :: Event DIO Bool+isLocalTimeEnding =+  Event $ \p ->+  do let q  = runEventQueue $ pointRun p+         t0 = spcStopTime $ pointSpecs p +     t1 <- liftIOUnsafe $ readIORef (queueTime q)+     t2 <- liftIOUnsafe $ transientMessageQueueTime (queueTransientMessages q)+     t3 <- liftIOUnsafe $ acknowledgementMessageTime (queueTransientMessages q)+     return $ (t1 == t0) && (t2 > t0) && (t3 > t0)+ -- | Update the global time. updateGlobalTime :: Double -> Event DIO () updateGlobalTime t =@@ -647,58 +672,62 @@ syncLocalTime :: Dynamics DIO () -> TimeWarp DIO () syncLocalTime m =   TimeWarp $ \p ->-  do let q = runEventQueue $ pointRun p-         t = pointTime p-     invokeDynamics p m-     t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)-     if t' > t-       then error "Inconsistent time: syncLocalTime"-       else if (t == spcStartTime (pointSpecs p)) || (t' == pointTime p)-            then return ()-            else do ----                    --- invokeEvent p logSyncLocalTime-                    ----                    ch <- messageChannel-                    dt <- fmap dioSyncTimeout dioParams-                    f  <- liftIOUnsafe $-                          timeout dt $ awaitChannel ch-                    ok <- invokeEvent p $ runTimeWarp processChannelMessages-                    if ok-                      then do case f of-                                Just _  ->-                                  invokeTimeWarp p $ syncLocalTime m-                                Nothing ->-                                  do -- invokeEvent p requestGlobalTime-                                     invokeTimeWarp p $ syncLocalTime0 m-                      else return ()+  do invokeDynamics p m+     f <- invokeEvent p isLocalTimeSynchronized+     unless f $+       do ---+          --- invokeEvent p logSyncLocalTime+          ---+          ch <- messageChannel+          dt <- fmap dioSyncTimeout dioParams+          f  <- liftIOUnsafe $+                timeout dt $ awaitChannel ch+          ok <- invokeEvent p $ runTimeWarp processChannelMessages+          when ok $+            case f of+              Just _  ->+                invokeTimeWarp p $ syncLocalTime m+              Nothing ->+                do -- invokeEvent p requestGlobalTime+                   invokeTimeWarp p $ syncLocalTime0 m    -- | Synchronize the local time executing the specified computation in ring 0. syncLocalTime0 :: Dynamics DIO () -> TimeWarp DIO () syncLocalTime0 m =   TimeWarp $ \p ->+  do invokeDynamics p m+     f <- invokeEvent p isLocalTimeSynchronized+     unless f $+       do ---+          --- invokeEvent p logSyncLocalTime0+          ---+          ch <- messageChannel+          dt <- fmap dioSyncTimeout dioParams+          f  <- liftIOUnsafe $+                timeout dt $ awaitChannel ch+          ok <- invokeEvent p $ runTimeWarp processChannelMessages+          when ok $+            case f of+              Just _  ->+                invokeTimeWarp p $ syncLocalTime m+              Nothing ->+                error "Exceeded the timeout when synchronizing the local time: syncLocalTime0"++-- | Test whether the local time is synchronized.+isLocalTimeSynchronized :: Event DIO Bool+isLocalTimeSynchronized =+  Event $ \p ->   do let q = runEventQueue $ pointRun p          t = pointTime p-     invokeDynamics p m      t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)      if t' > t-       then error "Inconsistent time: syncLocalTime0"-       else if t' == pointTime p-            then return ()-            else do ----                    --- invokeEvent p logSyncLocalTime0-                    ----                    ch <- messageChannel-                    dt <- fmap dioSyncTimeout dioParams-                    f  <- liftIOUnsafe $-                          timeout dt $ awaitChannel ch-                    ok <- invokeEvent p $ runTimeWarp processChannelMessages-                    if ok-                      then do case f of-                                Just _  ->-                                  invokeTimeWarp p $ syncLocalTime m-                                Nothing ->-                                  error "Detected a deadlock when synchronizing the local time: syncLocalTime0"-                      else return ()+       then error "Inconsistent time: isLocalTimeSynchronized"+       else if (t == spcStartTime (pointSpecs p)) || (t' == t)+            then return True+            else do leaving <- liftIOUnsafe $ readIORef (queueIsLeaving q)+                    if (leaving && (t == spcStopTime (pointSpecs p)))+                      then invokeEvent p isLocalTimeEnding+                      else return False  -- | Run the computation and return a flag indicating whether there was no rollback. runTimeWarp :: TimeWarp DIO () -> Event DIO Bool@@ -911,3 +940,40 @@                              lpStateOutputMessageCount = n3,                              lpStateTransientMessageCount = n4,                              lpStateRollbackCount = n5 }++-- | Try to leave the simulation.+--+-- It enqueues a new event with obligatory synchronization with the global virtual time.+-- If that event will not be reverted during rollback, then the logical process will+-- leave the simulation immediately after approaching the final modeling time without+-- synchronization with the global virtual time anymore if it has no unacknowledged+-- sent messages.+--+-- It makes sense to use this action if logical processes can enter and leave your long-running+-- distributed simulation by demand.+--+leaveSimulation :: Event DIO ()+leaveSimulation =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+         t = pointTime p+     invokeEvent p $+       enqueueEventIO t $+       Event $ \p ->+       liftIOUnsafe $ writeIORef (queueIsLeaving q) True++-- | Disconnect to the remote process.+disconnectProcess :: DP.ProcessId -> Event DIO ()+disconnectProcess pid =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     ---+     logDIO NOTICE $+       "t = " ++ show (pointTime p) +++       ": disconnecting from " ++ show pid ++ "..."+     ---+     ps <- dioParams+     when (dioProcessReconnectingEnabled ps) $+       error "The logical process cannot be in the reconnecting state: disconnectProcess"+     liftIOUnsafe $+       dequeueTransientMessages (queueTransientMessages q) pid

@@ -130,6 +130,8 @@                              -- ^ provide the logical process state                            | AbortSimulationMessage                              -- ^ abort the simulation+                           | DisconnectProcessMessage DP.ProcessId+                             -- ^ the specified process has been disconnected                            deriving (Show, Typeable, Generic)  instance Binary LogicalProcessMessage

@@ -81,17 +81,15 @@  -- | The time server strategy. data TimeServerStrategy = WaitIndefinitelyForLogicalProcess-                          -- ^ wait for the logical process forever+                          -- ^ wait for the logical processes 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'+                          -- but not less than 'tsTimeSyncTimeout', which should be applied if+                          -- the process reconnecting is enabled                         | 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+                          -- but not less than 'tsTimeSyncTimeout', which can be applied only if+                          -- the process disconnecting is enabled                         deriving (Eq, Ord, Show, Typeable, Generic)  instance Binary TimeServerStrategy

@@ -20,7 +20,8 @@         acknowledgementMessageTime,         resetAcknowledgementMessageTime,         deliverAcknowledgementMessage,-        deliverAcknowledgementMessages) where+        deliverAcknowledgementMessages,+        dequeueTransientMessages) where  import qualified Data.Map as M import Data.List@@ -170,3 +171,9 @@       dlv zs@(z : _) =         sendAcknowledgementMessagesDIO (acknowledgementSenderId z) zs   in forM_ ys dlv++-- | Dequeue the transient messages associated with the specified logical process.+dequeueTransientMessages :: TransientMessageQueue -> DP.ProcessId -> IO ()+dequeueTransientMessages q pid =+  modifyIORef (queueTransientMessages q) $+  M.filter (\m -> messageReceiverId m /= pid)

@@ -1,5 +1,5 @@ name:            aivika-distributed-version:         1.3+version:         1.4 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.@@ -28,12 +28,15 @@     Regarding the speed of simulation, the recent rough estimation is as follows. This estimation may change from      version to version. For example, in version 1.0 the rollback log was rewritten, which had a significant effect.     .-    The distributed simulation module is slower up to 8-15 times in comparison with the sequential aivika [2] simulation -    library using the equivalent sequential models. The lower estimation in 8 times is likely to correspond to complex -    models. The upper estimation in 15 times will probably correspond to quite simple event-oriented and process-oriented -    models, where the sequential module can be exceptionally fast. At the same time, the message passing between -    the logical processes can dramatically decrease the speed of distributed simulation, especially if the messages cause -    rollbacks. Thus, much depends on the distributed model itself.+    When simulating sequential models, the speed of single logical process of the distributed module in comparison with +    the sequential aivika [2] module varies and depends essentially on the number of +    simultaneously processed discrete events, or the number of simultaneously running discontinuous processes, +    which is very close. If there are many simultaneous events, then the distributed module can be slower in 4-5 times only. +    The more simultaneous events are defined in the model, the less is a gap in the speed between modules.+    But if the simultaneous events are rare, then the distributed module can be slower even in 15 times, +    where the sequential module can be exceptionally fast. At the same time, the message passing between the logical +    processes can dramatically decrease the speed of distributed simulation, especially if the messages cause rollbacks. +    Then it makes sense to define the time horizon parameter. Thus, much depends on the distributed model itself.     .     When residing the logical processes in a computer with multi-core processor, you should follow the next recommendations.      You should reserve at least 1 core for each logical process, or even reserve 2 cores if the logical process extensively @@ -83,6 +86,7 @@     exposed-modules: Simulation.Aivika.Distributed                      Simulation.Aivika.Distributed.Optimistic                      Simulation.Aivika.Distributed.Optimistic.DIO+                     Simulation.Aivika.Distributed.Optimistic.Event                      Simulation.Aivika.Distributed.Optimistic.Guard                      Simulation.Aivika.Distributed.Optimistic.Generator                      Simulation.Aivika.Distributed.Optimistic.QueueStrategy