aivika-distributed 0.3.1 → 0.5

raw patch · 44 files changed

+1426/−300 lines, 44 filesdep ~aivikadep ~aivika-transformersdep ~basePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: aivika, aivika-transformers, base

API changes (from Hackage documentation)

Files

• CHANGELOG.md +9/−0
• LICENSE +1/−1
• Simulation/Aivika/Distributed.hs +1/−1
• Simulation/Aivika/Distributed/Optimistic.hs +1/−1
• Simulation/Aivika/Distributed/Optimistic/DIO.hs +12/−3
• Simulation/Aivika/Distributed/Optimistic/Generator.hs +13/−3
• Simulation/Aivika/Distributed/Optimistic/Internal/Channel.hs +3/−3
• Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +369/−32
• Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +92/−33
• Simulation/Aivika/Distributed/Optimistic/Internal/IO.hs +1/−1
• Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs +27/−4
• Simulation/Aivika/Distributed/Optimistic/Internal/InputMessageQueue.hs-boot +5/−2
• Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs +107/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs +58/−7
• Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs +4/−7
• Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs-boot +1/−1
• Simulation/Aivika/Distributed/Optimistic/Internal/Priority.hs +1/−1
• Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs +4/−62
• Simulation/Aivika/Distributed/Optimistic/Internal/Ref.hs-boot +0/−29
• Simulation/Aivika/Distributed/Optimistic/Internal/Ref/Lazy.hs +73/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/Ref/Strict.hs +73/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/Ref/Strict.hs-boot +29/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/SignalHelper.hs +26/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/SignalHelper.hs-boot +22/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs +217/−30
• Simulation/Aivika/Distributed/Optimistic/Internal/TimeWarp.hs +1/−1
• Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs +28/−20
• Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs +1/−1
• Simulation/Aivika/Distributed/Optimistic/Message.hs +1/−1
• Simulation/Aivika/Distributed/Optimistic/Priority.hs +1/−1
• Simulation/Aivika/Distributed/Optimistic/QueueStrategy.hs +1/−1
• Simulation/Aivika/Distributed/Optimistic/Ref/Base.hs +5/−32
• Simulation/Aivika/Distributed/Optimistic/Ref/Base/Lazy.hs +48/−0
• Simulation/Aivika/Distributed/Optimistic/Ref/Base/Strict.hs +48/−0
• Simulation/Aivika/Distributed/Optimistic/TimeServer.hs +1/−1
• aivika-distributed.cabal +49/−9
• tests/MachRep1Simple.hs +0/−2
• tests/MachRep2.hs +2/−2
• tests/MachRep2Distributed.hs +2/−2
• tests/MachRep2Reproducible.hs +2/−2
• tests/MachRep2Sync.hs +2/−2
• tests/MachRep2SyncIO.hs +2/−2
• tests/SimpleLocalnetHelper.hs +70/−0
• tests/cluster.conf +13/−0

@@ -1,4 +1,13 @@ +Version 0.5+-----++* Added an ability to restore the distributed simulation after temporary connection errors.++* Better finalisation of the distributed simulation.++* Implemented lazy references.+ Version 0.3 ----- 

@@ -1,4 +1,4 @@-Copyright (c) 2015-2016 David Sorokin <david.sorokin@gmail.com>+Copyright (c) 2015-2017 David Sorokin <david.sorokin@gmail.com>  All rights reserved. 

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -3,7 +3,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.DIO--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -22,7 +22,10 @@         logDIO,         terminateDIO,         registerDIO,-        unregisterDIO) where+        unregisterDIO,+        monitorProcessDIO,+        InboxProcessMessage(MonitorProcessMessage),+        processMonitorSignal) where  import Control.Monad import Control.Monad.Trans@@ -32,16 +35,19 @@ import Simulation.Aivika.Trans.Exception import Simulation.Aivika.Trans.Generator import Simulation.Aivika.Trans.Event+import Simulation.Aivika.Trans.Composite import Simulation.Aivika.Trans.Process import Simulation.Aivika.Trans.Ref.Base import Simulation.Aivika.Trans.QueueStrategy +import Simulation.Aivika.Distributed.Optimistic.Internal.Message import Simulation.Aivika.Distributed.Optimistic.Internal.DIO import Simulation.Aivika.Distributed.Optimistic.Internal.Event import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue import Simulation.Aivika.Distributed.Optimistic.Generator-import Simulation.Aivika.Distributed.Optimistic.Ref.Base+import Simulation.Aivika.Distributed.Optimistic.Ref.Base.Lazy+import Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict import Simulation.Aivika.Distributed.Optimistic.QueueStrategy  instance MonadDES DIO@@ -49,4 +55,7 @@ instance MonadComp DIO  instance {-# OVERLAPPING #-} MonadIO (Process DIO) where+  liftIO = liftEvent . liftIO++instance {-# OVERLAPPING #-} MonadIO (Composite DIO) where   liftIO = liftEvent . liftIO

@@ -3,7 +3,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Generator--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -32,8 +32,10 @@   data Generator DIO =     Generator { generator01 :: DIO Double,                 -- ^ the generator of uniform numbers from 0 to 1-                generatorNormal01 :: DIO Double+                generatorNormal01 :: DIO Double,                 -- ^ the generator of normal numbers with mean 0 and variance 1+                generatorSequenceNo :: DIO Int+                -- ^ the generator of sequence numbers               }    generateUniform = generateUniform01 . generator01@@ -62,6 +64,8 @@    generateDiscrete = generateDiscrete01 . generator01 +  generateSequenceNo = generatorSequenceNo+   newGenerator tp =     case tp of       SimpleGenerator ->@@ -83,8 +87,14 @@    newRandomGenerator01 g01 =     do gNormal01 <- newNormalGenerator01 g01+       gSeqNoRef <- liftIOUnsafe $ newIORef 0+       let gSeqNo =+             do x <- liftIOUnsafe $ readIORef gSeqNoRef+                liftIOUnsafe $ modifyIORef' gSeqNoRef (+1)+                return x        return Generator { generator01 = g01,-                          generatorNormal01 = gNormal01 }+                          generatorNormal01 = gNormal01,+                          generatorSequenceNo = gSeqNo }  -- | Create a normal random number generator with mean 0 and variance 1 -- by the specified generator of uniform random numbers from 0 to 1.

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Channel--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -51,7 +51,7 @@   do empty <- readIORef (channelListEmptyIO ch)      if empty        then return []-       else do writeIORef (channelListEmptyIO ch) True+       else do atomicWriteIORef (channelListEmptyIO ch) True                xs <- atomically $                      do xs <- readTVar (channelList ch)                         writeTVar (channelList ch) []@@ -66,7 +66,7 @@        do xs <- readTVar (channelList ch)           writeTVar (channelList ch) (a : xs)           writeTVar (channelListEmpty ch) False-     writeIORef (channelListEmptyIO ch) False+     atomicWriteIORef (channelListEmptyIO ch) False  -- | Wait for data in the channel. awaitChannel :: Channel a -> IO ()

@@ -3,7 +3,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.DIO--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -20,15 +20,23 @@         terminateDIO,         registerDIO,         unregisterDIO,+        monitorProcessDIO,         dioParams,         messageChannel,         messageInboxId,         timeServerId,+        sendMessageDIO,+        sendMessagesDIO,+        sendAcknowledgmentMessageDIO,+        sendAcknowledgmentMessagesDIO,+        sendLocalTimeDIO,+        sendRequestGlobalTimeDIO,         logDIO,         liftDistributedUnsafe) where  import Data.Typeable import Data.Binary+import Data.IORef  import GHC.Generics @@ -38,7 +46,11 @@ import Control.Exception (throw) import Control.Monad.Catch as C import qualified Control.Distributed.Process as DP+import Control.Concurrent+import Control.Concurrent.STM +import System.Timeout+ import Simulation.Aivika.Trans.Exception import Simulation.Aivika.Trans.Internal.Types @@ -46,6 +58,7 @@ import Simulation.Aivika.Distributed.Optimistic.Internal.Message import Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer import Simulation.Aivika.Distributed.Optimistic.Internal.Priority+import Simulation.Aivika.Distributed.Optimistic.Internal.KeepAliveManager  -- | The parameters for the 'DIO' computation. data DIOParams =@@ -55,12 +68,27 @@               -- ^ The undoable log size threshold used for detecting an overflow               dioOutputMessageQueueSizeThreshold :: Int,               -- ^ The output message queue size threshold used for detecting an overflow+              dioTransientMessageQueueSizeThreshold :: Int,+              -- ^ The transient message queue size threshold used for detecting an overflow               dioSyncTimeout :: Int,-              -- ^ The timeout in microseconds used for synchronising the operations.+              -- ^ The timeout in microseconds used for synchronising the operations               dioAllowPrematureIO :: Bool,               -- ^ Whether to allow performing the premature IO action; otherwise, raise an error-              dioAllowProcessingOutdatedMessage :: Bool-              -- ^ Whether to allow processing an outdated message with the receive time less than the global time+              dioAllowSkippingOutdatedMessage :: Bool,+              -- ^ Whether to allow skipping an outdated message with the receive time less than the global time,+              -- which is possible after reconnection+              dioProcessMonitoringEnabled :: Bool,+              -- ^ Whether the process monitoring is enabled+              dioProcessMonitoringDelay :: Int,+              -- ^ The delay in microseconds which must be applied for monitoring every remote process.+              dioProcessReconnectingEnabled :: Bool,+              -- ^ Whether the automatic reconnecting to processes is enabled when enabled monitoring+              dioProcessReconnectingDelay :: Int,+              -- ^ The timeout in microseconds before reconnecting to the remote process+              dioKeepAliveInterval :: Int,+              -- ^ The interval in microseconds for sending keep-alive messages+              dioTimeServerAcknowledgmentTimeout :: Int+              -- ^ The timeout in microseconds for receiving an acknowledgment message from the time server             } deriving (Eq, Ord, Show, Typeable, Generic)  instance Binary DIOParams@@ -78,8 +106,14 @@                -- ^ The inbox process identifier.                dioTimeServerId :: DP.ProcessId,                -- ^ The time server process-               dioParams0 :: DIOParams+               dioParams0 :: DIOParams,                -- ^ The parameters of the computation.+               dioRegisteredInTimeServer :: TVar Bool,+               -- ^ Whether the compution is registered in the time server.+               dioUnregisteredFromTimeServer :: TVar Bool,+               -- ^ Whether the compution is unregistered from the time server.+               dioTimeServerTerminating :: TVar Bool+               -- ^ Whether the compution asked to terminate the time server.              }  instance Monad DIO where@@ -131,11 +165,22 @@   DIOParams { dioLoggingPriority = WARNING,               dioUndoableLogSizeThreshold = 1000000,               dioOutputMessageQueueSizeThreshold = 10000,+              dioTransientMessageQueueSizeThreshold = 10000,               dioSyncTimeout = 60000000,               dioAllowPrematureIO = False,-              dioAllowProcessingOutdatedMessage = False+              dioAllowSkippingOutdatedMessage = True,+              dioProcessMonitoringEnabled = False,+              dioProcessMonitoringDelay = 5000000,+              dioProcessReconnectingEnabled = False,+              dioProcessReconnectingDelay = 5000000,+              dioKeepAliveInterval = 5000000,+              dioTimeServerAcknowledgmentTimeout = 5000000             } +-- | Return the computation context.+dioContext :: DIO DIOContext+dioContext = DIO return+ -- | Return the parameters of the current computation. dioParams :: DIO DIOParams dioParams = DIO $ return . dioParams0@@ -156,53 +201,345 @@ -- all nodes connected to the time server. terminateDIO :: DIO () terminateDIO =-  do logDIO INFO "Terminating the simulation..."-     sender   <- messageInboxId-     receiver <- timeServerId-     liftDistributedUnsafe $-       DP.send receiver (TerminateTimeServerMessage sender)+  DIO $ \ctx ->+  do let ps = dioParams0 ctx+     logProcess ps INFO "Terminating the simulation..."+     sender   <- invokeDIO ctx messageInboxId+     receiver <- invokeDIO ctx timeServerId+     let inbox = sender+     if dioProcessMonitoringEnabled ps+       then DP.send inbox (SendTerminateTimeServerMessage receiver sender)+       else DP.send receiver (TerminateTimeServerMessage sender)+     liftIO $+       timeout (dioTimeServerAcknowledgmentTimeout ps) $+       atomically $+       do f <- readTVar (dioTimeServerTerminating ctx)+          unless f retry+     DP.send inbox TerminateInboxProcessMessage+     return ()  -- | Register the simulation process in the time server, which -- requires some initial quorum to start synchronizing the global time. registerDIO :: DIO () registerDIO =-  do logDIO INFO "Registering the simulation process..."-     sender   <- messageInboxId-     receiver <- timeServerId-     liftDistributedUnsafe $-       DP.send receiver (RegisterLocalProcessMessage sender)+  DIO $ \ctx ->+  do let ps = dioParams0 ctx+     logProcess ps INFO "Registering the simulation process..."+     sender   <- invokeDIO ctx messageInboxId+     receiver <- invokeDIO ctx timeServerId+     let inbox = sender+     if dioProcessMonitoringEnabled ps+       then DP.send inbox (SendRegisterLocalProcessMessage receiver sender)+       else DP.send receiver (RegisterLocalProcessMessage sender)+     liftIO $+       timeout (dioTimeServerAcknowledgmentTimeout ps) $+       atomically $+       do f <- readTVar (dioRegisteredInTimeServer ctx)+          unless f retry+     return ()  -- | Unregister the simulation process from the time server -- without affecting the processes in other nodes connected to -- the corresponding time server. unregisterDIO :: DIO () unregisterDIO =-  do logDIO INFO "Unregistering the simulation process..."-     sender   <- messageInboxId-     receiver <- timeServerId-     liftDistributedUnsafe $-       DP.send receiver (UnregisterLocalProcessMessage sender)+  DIO $ \ctx ->+  do let ps = dioParams0 ctx+     logProcess ps INFO "Unregistering the simulation process..."+     sender   <- invokeDIO ctx messageInboxId+     receiver <- invokeDIO ctx timeServerId+     let inbox = sender+     if dioProcessMonitoringEnabled ps+       then DP.send inbox (SendUnregisterLocalProcessMessage receiver sender)+       else DP.send receiver (UnregisterLocalProcessMessage sender)+     liftIO $+       timeout (dioTimeServerAcknowledgmentTimeout ps) $+       atomically $+       do f <- readTVar (dioUnregisteredFromTimeServer ctx)+          unless f retry+     DP.send inbox TerminateInboxProcessMessage+     return () +-- | The internal local process message.+data InternalLocalProcessMessage = InternalLocalProcessMessage LocalProcessMessage+                                   -- ^ the local process message+                                 | InternalProcessMonitorNotification DP.ProcessMonitorNotification+                                   -- ^ the process monitor notification+                                 | InternalInboxProcessMessage InboxProcessMessage+                                   -- ^ the inbox process message+                                 | InternalKeepAliveMessage KeepAliveMessage+                                   -- ^ the keep alive message++-- | Handle the specified exception.+handleException :: DIOParams -> SomeException -> DP.Process ()+handleException ps e =+  do ---+     logProcess ps ERROR $ "Exception occured: " ++ show e+     ---+     C.throwM e+ -- | Run the computation using the specified parameters along with time server process -- identifier and return the inbox process identifier and a new simulation process. runDIO :: DIO a -> DIOParams -> DP.ProcessId -> DP.Process (DP.ProcessId, DP.Process a) runDIO m ps serverId =   do ch <- liftIO newChannel+     let keepAliveParams =+           KeepAliveParams { keepAliveLoggingPriority = dioLoggingPriority ps,+                             keepAliveInterval = dioKeepAliveInterval ps }+     keepAliveManager <- liftIO $ newKeepAliveManager keepAliveParams+     terminated <- liftIO $ newIORef False+     registeredInTimeServer <- liftIO $ newTVarIO False+     unregisteredFromTimeServer <- liftIO $ newTVarIO False+     timeServerTerminating <- liftIO $ newTVarIO False+     let loop =+           forever $+           do let f1 :: LocalProcessMessage -> DP.Process InternalLocalProcessMessage+                  f1 x = return (InternalLocalProcessMessage x)+                  f2 :: DP.ProcessMonitorNotification -> DP.Process InternalLocalProcessMessage+                  f2 x = return (InternalProcessMonitorNotification x)+                  f3 :: InboxProcessMessage -> DP.Process InternalLocalProcessMessage+                  f3 x = return (InternalInboxProcessMessage x)+                  f4 :: KeepAliveMessage -> DP.Process InternalLocalProcessMessage+                  f4 x = return (InternalKeepAliveMessage x)+              x <- fmap Just $ DP.receiveWait [DP.match f1, DP.match f2, DP.match f3, DP.match f4]+              case x of+                Nothing -> return ()+                Just (InternalLocalProcessMessage m) ->+                  liftIO $+                  writeChannel ch m+                Just (InternalProcessMonitorNotification m@(DP.ProcessMonitorNotification _ _ _)) ->+                  handleProcessMonitorNotification m ps ch+                Just (InternalInboxProcessMessage m) ->+                  case m of+                    SendQueueMessage pid m ->+                      DP.send pid (QueueMessage m)+                    SendQueueMessageBulk pid ms ->+                      forM_ ms $ \m ->+                      DP.send pid (QueueMessage m)+                    SendAcknowledgmentQueueMessage pid m ->+                      DP.send pid (AcknowledgmentQueueMessage m)+                    SendAcknowledgmentQueueMessageBulk pid ms ->+                      forM_ ms $ \m ->+                      DP.send pid (AcknowledgmentQueueMessage m)+                    SendLocalTimeMessage receiver sender t ->+                      DP.send receiver (LocalTimeMessage sender t)+                    SendRequestGlobalTimeMessage receiver sender ->+                      DP.send receiver (RequestGlobalTimeMessage sender)+                    SendRegisterLocalProcessMessage receiver sender ->+                      DP.send receiver (RegisterLocalProcessMessage sender)+                    SendUnregisterLocalProcessMessage receiver sender ->+                      DP.send receiver (UnregisterLocalProcessMessage sender)+                    SendTerminateTimeServerMessage receiver sender ->+                      DP.send receiver (TerminateTimeServerMessage sender)+                    MonitorProcessMessage pid ->+                      do ---+                         logProcess ps INFO $ "Monitoring process " ++ show pid+                         ---+                         DP.monitor pid+                         liftIO $+                           addKeepAliveReceiver keepAliveManager pid+                         return ()+                    ReMonitorProcessMessage pids ->+                      handleReMonitorProcessMessage pids ps ch+                    TrySendKeepAliveMessage ->+                      trySendKeepAlive keepAliveManager+                    RegisterLocalProcessAcknowledgmentMessage pid ->+                      do ---+                         logProcess ps INFO "Registered the local process in the time server"+                         ---+                         liftIO $+                           atomically $+                           writeTVar registeredInTimeServer True+                    UnregisterLocalProcessAcknowledgmentMessage pid ->+                      do ---+                         logProcess ps INFO "Unregistered the local process from the time server"+                         ---+                         liftIO $+                           atomically $+                           writeTVar unregisteredFromTimeServer True+                    TerminateTimeServerAcknowledgmentMessage pid ->+                      do ---+                         logProcess ps INFO "Started terminating the time server"+                         ---+                         liftIO $+                           atomically $+                           writeTVar timeServerTerminating True+                    TerminateInboxProcessMessage ->+                      do ---+                         logProcess ps INFO "Terminating the inbox and keep-alive processes..."+                         ---+                         liftIO $+                           atomicWriteIORef terminated True+                         DP.terminate+                Just (InternalKeepAliveMessage m) ->+                  do ---+                     logProcess ps DEBUG $ "Received " ++ show m+                     ---+                     liftIO $+                       writeChannel ch (KeepAliveLocalProcessMessage m)      inboxId <-        DP.spawnLocal $-       forever $-       do m <- DP.expect :: DP.Process LocalProcessMessage-          liftIO $-            writeChannel ch m-          when (m == TerminateLocalProcessMessage) $+       C.catch loop (handleException ps)+     DP.spawnLocal $+       let loop =+             do f <- liftIO $ readIORef terminated+                unless f $+                  do liftIO $+                       threadDelay (dioKeepAliveInterval ps)+                     DP.send inboxId TrySendKeepAliveMessage+                     loop+       in C.catch loop (handleException ps)+     let simulation =+           unDIO m DIOContext { dioChannel = ch,+                                dioInboxId = inboxId,+                                dioTimeServerId = serverId,+                                dioParams0 = ps,+                                dioRegisteredInTimeServer = registeredInTimeServer,+                                dioUnregisteredFromTimeServer = unregisteredFromTimeServer,+                                dioTimeServerTerminating = timeServerTerminating }+     return (inboxId, simulation)++-- | Handle the process monitor notification.+handleProcessMonitorNotification :: DP.ProcessMonitorNotification -> DIOParams -> Channel LocalProcessMessage -> DP.Process ()+handleProcessMonitorNotification m@(DP.ProcessMonitorNotification _ pid0 reason) ps ch =+  do let recv m@(DP.ProcessMonitorNotification _ _ _) = +           do ---+              logProcess ps WARNING $ "Received a process monitor notification " ++ show m+              ---+              liftIO $+                writeChannel ch (ProcessMonitorNotificationMessage m)+              return m+     recv m+     when (dioProcessReconnectingEnabled ps && (reason == DP.DiedDisconnect)) $+       do liftIO $+            threadDelay (dioProcessReconnectingDelay ps)+          let pred m@(DP.ProcessMonitorNotification _ _ reason) = reason == DP.DiedDisconnect+              loop :: [DP.ProcessId] -> DP.Process [DP.ProcessId]+              loop acc =+                do y <- DP.receiveTimeout 0 [DP.matchIf pred recv]+                   case y of+                     Nothing -> return $ reverse acc+                     Just m@(DP.ProcessMonitorNotification _ pid _) -> loop (pid : acc)+          pids <- loop [pid0]+          ---+          logProcess ps NOTICE "Begin reconnecting..."+          ---+          forM_ pids $ \pid ->             do ----               logProcess ps INFO "Terminating the inbox process..."+               logProcess ps NOTICE $ "Direct reconnecting to " ++ show pid                ----               DP.terminate-     return (inboxId, unDIO m DIOContext { dioChannel = ch,-                                           dioInboxId = inboxId,-                                           dioTimeServerId = serverId,-                                           dioParams0 = ps })+               DP.reconnect pid+          inboxId <- DP.getSelfPid+          DP.spawnLocal $+            let action =+                  do liftIO $+                       threadDelay (dioProcessMonitoringDelay ps)+                     ---+                     logProcess ps NOTICE $ "Proceed to the re-monitoring"+                     ---+                     DP.send inboxId (ReMonitorProcessMessage pids)+            in C.catch action (handleException ps)+          return ()++-- | Re-monitor the specified remote processes.+handleReMonitorProcessMessage :: [DP.ProcessId] -> DIOParams -> Channel LocalProcessMessage -> DP.Process ()+handleReMonitorProcessMessage pids ps ch =+  forM_ pids $ \pid ->+  do ---+     logProcess ps NOTICE $ "Re-monitoring " ++ show pid+     ---+     DP.monitor pid+     ---+     logProcess ps NOTICE $ "Writing to the channel about reconnecting to " ++ show pid+     ---+     liftIO $+       writeChannel ch (ReconnectProcessMessage pid)++-- | Monitor the specified process.+monitorProcessDIO :: DP.ProcessId -> DIO ()+monitorProcessDIO pid =+  do ps <- dioParams+     if dioProcessMonitoringEnabled ps+       then do inbox <- messageInboxId+               liftDistributedUnsafe $+                 DP.send inbox $+                 MonitorProcessMessage pid+       else liftDistributedUnsafe $+            logProcess ps WARNING "Ignored the process monitoring as it was disabled in the DIO computation parameters"++-- | Send the message.+sendMessageDIO :: DP.ProcessId -> Message -> DIO ()+{-# INLINABLE sendMessageDIO #-}+sendMessageDIO pid m =+  do ps <- dioParams+     if dioProcessMonitoringEnabled ps+       then do inbox <- messageInboxId+               liftDistributedUnsafe $+                 DP.send inbox (SendQueueMessage pid m)+       else liftDistributedUnsafe $+            DP.send pid (QueueMessage m)++-- | Send the bulk of messages.+sendMessagesDIO :: DP.ProcessId -> [Message] -> DIO ()+{-# INLINABLE sendMessagesDIO #-}+sendMessagesDIO pid ms =+  do ps <- dioParams+     if dioProcessMonitoringEnabled ps+       then do inbox <- messageInboxId+               liftDistributedUnsafe $+                 DP.send inbox (SendQueueMessageBulk pid ms)+       else do forM_ ms $ \m ->+                 liftDistributedUnsafe $+                 DP.send pid (QueueMessage m)++-- | Send the acknowledgment message.+sendAcknowledgmentMessageDIO :: DP.ProcessId -> AcknowledgmentMessage -> DIO ()+{-# INLINABLE sendAcknowledgmentMessageDIO #-}+sendAcknowledgmentMessageDIO pid m =+  do ps <- dioParams+     if dioProcessMonitoringEnabled ps+       then do inbox <- messageInboxId+               liftDistributedUnsafe $+                 DP.send inbox (SendAcknowledgmentQueueMessage pid m)+       else liftDistributedUnsafe $+            DP.send pid (AcknowledgmentQueueMessage m)++-- | Send the bulk of acknowledgment messages.+sendAcknowledgmentMessagesDIO :: DP.ProcessId -> [AcknowledgmentMessage] -> DIO ()+{-# INLINABLE sendAcknowledgmentMessagesDIO #-}+sendAcknowledgmentMessagesDIO pid ms =+  do ps <- dioParams+     if dioProcessMonitoringEnabled ps+       then do inbox <- messageInboxId+               liftDistributedUnsafe $+                 DP.send inbox (SendAcknowledgmentQueueMessageBulk pid ms)+       else do forM_ ms $ \m ->+                 liftDistributedUnsafe $+                 DP.send pid (AcknowledgmentQueueMessage m)++-- | Send the local time to the time server.+sendLocalTimeDIO :: DP.ProcessId -> DP.ProcessId -> Double -> DIO ()+{-# INLINABLE sendLocalTimeDIO #-}+sendLocalTimeDIO receiver sender t =+  do ps <- dioParams+     if dioProcessMonitoringEnabled ps+       then do inbox <- messageInboxId+               liftDistributedUnsafe $+                 DP.send inbox (SendLocalTimeMessage receiver sender t)+       else liftDistributedUnsafe $+            DP.send receiver (LocalTimeMessage sender t)++-- | Send the request for the global virtual time to the time server.+sendRequestGlobalTimeDIO :: DP.ProcessId -> DP.ProcessId -> DIO ()+{-# INLINABLE sendRequestGlobalTimeDIO #-}+sendRequestGlobalTimeDIO receiver sender =+  do ps <- dioParams+     if dioProcessMonitoringEnabled ps+       then do inbox <- messageInboxId+               liftDistributedUnsafe $+                 DP.send inbox (SendRequestGlobalTimeMessage receiver sender)+       else liftDistributedUnsafe $+            DP.send receiver (RequestGlobalTimeMessage sender)  -- | Log the message with the specified priority. logDIO :: Priority -> String -> DIO ()

@@ -3,7 +3,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Event--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -14,7 +14,8 @@ module Simulation.Aivika.Distributed.Optimistic.Internal.Event        (queueInputMessages,         queueOutputMessages,-        queueLog) where+        queueLog,+        processMonitorSignal) where  import Data.Maybe import Data.IORef@@ -39,11 +40,12 @@ import Simulation.Aivika.Distributed.Optimistic.Internal.Message import Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer import Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp+import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue import Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog-import {-# SOURCE #-} qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref as R+import {-# SOURCE #-} qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict as R  -- | Convert microseconds to seconds. microsecondsToSeconds :: Int -> Rational@@ -70,8 +72,10 @@                  -- ^ the actual time of the event queue                  queueGlobalTime :: IORef Double,                  -- ^ the global time-                 queueInFind :: IORef Bool+                 queueInFind :: IORef Bool,                  -- ^ whether the queue is in find mode+                 queueProcessMonitorNotificationSource :: SignalSource DIO DP.ProcessMonitorNotification+                 -- ^ the source of process monitor notifications                }    newEventQueue specs =@@ -84,6 +88,7 @@        output <- newOutputMessageQueue $ enqueueTransientMessage transient        input <- newInputMessageQueue log rollbackEventPre rollbackEventPost rollbackEventTime        infind <- liftIOUnsafe $ newIORef False+       s <- newDIOSignalSource0        return EventQueue { queueInputMessages = input,                            queueOutputMessages = output,                            queueTransientMessages = transient,@@ -92,7 +97,8 @@                            queueBusy = f,                            queueTime = t,                            queueGlobalTime = gt,-                           queueInFind = infind }+                           queueInFind = infind,+                           queueProcessMonitorNotificationSource = s }    enqueueEvent t (Event m) =     Event $ \p ->@@ -272,10 +278,12 @@   do let q = runEventQueue $ pointRun p      n1 <- liftIOUnsafe $ logSize (queueLog q)      n2 <- liftIOUnsafe $ outputMessageQueueSize (queueOutputMessages q)+     n3 <- liftIOUnsafe $ transientMessageQueueSize (queueTransientMessages q)      ps <- dioParams      let th1 = dioUndoableLogSizeThreshold ps          th2 = dioOutputMessageQueueSizeThreshold ps-     if (n1 >= th1) || (n2 >= th2)+         th3 = dioTransientMessageQueueSizeThreshold ps+     if (n1 >= th1) || (n2 >= th2) || (n3 >= th3)        then do logDIO NOTICE $                  "t = " ++ (show $ pointTime p) ++                  ": detected the event overflow"@@ -286,7 +294,7 @@ throttleMessageChannel :: TimeWarp DIO () throttleMessageChannel =   TimeWarp $ \p ->-  do invokeEvent p requestGlobalTime+  do -- invokeEvent p requestGlobalTime      ch <- messageChannel      dt <- fmap dioSyncTimeout dioParams      liftIOUnsafe $@@ -321,13 +329,14 @@      infind <- liftIOUnsafe $ readIORef (queueInFind q)      deliverAcknowledgmentMessage (acknowledgmentMessage infind m)      t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q)-     when (messageReceiveTime m < t0) $-       do f <- fmap dioAllowProcessingOutdatedMessage dioParams-          if f-            then invokeEvent p logOutdatedMessage-            else error "Received the outdated message: processChannelMessage"-     invokeTimeWarp p $-       enqueueMessage (queueInputMessages q) m+     p' <- invokeEvent p currentEventPoint+     if messageReceiveTime m < t0+       then do f <- fmap dioAllowSkippingOutdatedMessage dioParams+               if f+                 then invokeEvent p' logOutdatedMessage+                 else error "Received the outdated message: processChannelMessage"+       else invokeTimeWarp p' $+            enqueueMessage (queueInputMessages q) m processChannelMessage x@(QueueMessageBulk ms) =   TimeWarp $ \p ->   do let q = runEventQueue $ pointRun p@@ -339,13 +348,14 @@      deliverAcknowledgmentMessages $ map (acknowledgmentMessage infind) ms      t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q)      forM_ ms $ \m ->-       do when (messageReceiveTime m < t0) $-            do f <- fmap dioAllowProcessingOutdatedMessage dioParams-               if f-                 then invokeEvent p logOutdatedMessage-                 else error "Received the outdated message: processChannelMessage"-          invokeTimeWarp p $-            enqueueMessage (queueInputMessages q) m+       do p' <- invokeEvent p currentEventPoint+          if messageReceiveTime m < t0+            then do f <- fmap dioAllowSkippingOutdatedMessage dioParams+                    if f+                      then invokeEvent p' logOutdatedMessage+                      else error "Received the outdated message: processChannelMessage"+            else invokeTimeWarp p' $+                 enqueueMessage (queueInputMessages q) m processChannelMessage x@(AcknowledgmentQueueMessage m) =   TimeWarp $ \p ->   do let q = runEventQueue $ pointRun p@@ -377,8 +387,7 @@      t' <- invokeEvent p getLocalTime      sender   <- messageInboxId      receiver <- timeServerId-     liftDistributedUnsafe $-       DP.send receiver (LocalTimeMessage sender t')+     sendLocalTimeDIO receiver sender t' processChannelMessage x@(GlobalTimeMessage globalTime) =   TimeWarp $ \p ->   do let q = runEventQueue $ pointRun p@@ -391,14 +400,32 @@           resetAcknowledgmentMessageTime (queueTransientMessages q)      invokeEvent p $        updateGlobalTime globalTime-processChannelMessage x@TerminateLocalProcessMessage =+processChannelMessage x@(ProcessMonitorNotificationMessage y@(DP.ProcessMonitorNotification _ pid reason)) =   TimeWarp $ \p ->-  do ---+  do let q = runEventQueue $ pointRun p+     ---      --- invokeEvent p $      ---   logMessage x      ----     liftDistributedUnsafe $-       DP.terminate+     invokeEvent p $+       triggerSignal (queueProcessMonitorNotificationSource q) y+processChannelMessage x@(ReconnectProcessMessage pid) =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+     ---+     --- invokeEvent p $+     ---   logMessage x+     ---+     invokeEvent p $+       reconnectProcess pid+processChannelMessage x@(KeepAliveLocalProcessMessage m) =+  TimeWarp $ \p ->+  do let q = runEventQueue $ pointRun p+     ---+     --- invokeEvent p $+     ---   logMessage x+     ---+     return ()  -- | Return the local minimum time. getLocalTime :: Event DIO Double@@ -447,8 +474,7 @@      ---      sender   <- messageInboxId      receiver <- timeServerId-     liftDistributedUnsafe $-       DP.send receiver (RequestGlobalTimeMessage sender)+     sendRequestGlobalTimeDIO receiver sender  -- | Show the message. showMessage :: Message -> ShowS@@ -529,9 +555,9 @@ logOutdatedMessage :: Event DIO () logOutdatedMessage =   Event $ \p ->-  logDIO ERROR $+  logDIO WARNING $   "t = " ++ (show $ pointTime p) ++-  ": received the outdated message"+  ": skipping the outdated message"  -- | Reduce events till the specified time. reduceEvents :: Double -> Event DIO ()@@ -588,7 +614,7 @@                                 Just _  ->                                   invokeTimeWarp p $ syncLocalTime m                                 Nothing ->-                                  do invokeEvent p requestGlobalTime+                                  do -- invokeEvent p requestGlobalTime                                      invokeTimeWarp p $ syncLocalTime0 m                       else return ()   @@ -662,7 +688,7 @@   do let q = runEventQueue $ pointRun p          t = pointTime p      ----     logDIO NOTICE $+     logDIO INFO $        "t = " ++ show t ++        ": retrying the computations..."      ---@@ -702,3 +728,36 @@                     "Detected a deadlock when retrying the computations: handleEventRetry\n" ++                     "--- the nested exception ---\n" ++ show e       loop++-- | Reconnect to the remote process.+reconnectProcess :: DP.ProcessId -> Event DIO ()+reconnectProcess pid =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+     ---+     logDIO NOTICE $+       "t = " ++ show (pointTime p) +++       ": reconnecting to " ++ show pid ++ "..."+     ---+     infind <- liftIOUnsafe $ readIORef (queueInFind q)+     let ys = queueInputMessages q+     ys' <- liftIOUnsafe $+            fmap (map $ acknowledgmentMessage infind) $+            filterInputMessages (\x -> messageSenderId x == pid) ys+     unless (null ys') $+       sendAcknowledgmentMessagesDIO pid ys'+     xs <- liftIOUnsafe $ transientMessages (queueTransientMessages q)+     let xs' = filter (\x -> messageReceiverId x == pid) xs+     unless (null xs') $+       sendMessagesDIO pid xs'++-- | A signal triggered when coming the process monitor notification from the Cloud Haskell back-end.+processMonitorSignal :: Signal DIO DP.ProcessMonitorNotification+processMonitorSignal =+  Signal { handleSignal = \h ->+            Event $ \p ->+            let q = runEventQueue (pointRun p)+                s = publishSignal (queueProcessMonitorNotificationSource q)+            in invokeEvent p $+               handleSignal s h+         }

@@ -3,7 +3,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.IO--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -17,7 +17,8 @@         enqueueMessage,         messageEnqueued,         retryInputMessages,-        reduceInputMessages) where+        reduceInputMessages,+        filterInputMessages) where  import Data.Maybe import Data.List@@ -121,7 +122,10 @@          t0 = pointTime p      i <- liftIOUnsafe $ findAntiMessage q m      case i of-       Nothing -> return ()+       Nothing ->+         do -- skip the message duplicate+            logSkipInputMessage t0+            return ()        Just i | i >= 0 ->          do -- found the anti-message at the specified index             when (t <= t0) $@@ -155,10 +159,16 @@               else do liftIOUnsafe $ insertMessage q m i                       invokeEvent p $ activateMessage q i +-- | Log the message skip+logSkipInputMessage :: Double -> DIO ()+logSkipInputMessage t0 =+  logDIO NOTICE $+  "Skip the message at t = " ++ (show t0)+ -- | Log the rollback. logRollbackInputMessages :: Double -> Double -> Bool -> DIO () logRollbackInputMessages t0 t including =-  logDIO NOTICE $+  logDIO INFO $   "Rollback at t = " ++ (show t0) ++ " --> " ++ (show t) ++   (if not including then " not including" else "") @@ -385,3 +395,16 @@                               then loop n (i + 1)                               else return i +-- | Filter the input messages using the specified predicate.+filterInputMessages :: (Message -> Bool) -> InputMessageQueue -> IO [Message]+filterInputMessages pred q =+  do count <- vectorCount (inputMessages q)+     loop count 0 []+       where+         loop n i acc+           | i >= n    = return (reverse acc)+           | otherwise = do item <- readVector (inputMessages q) i+                            let m = itemMessage item+                            if pred m+                              then loop n (i + 1) (m : acc)+                              else loop n (i + 1) acc

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -17,7 +17,8 @@         enqueueMessage,         messageEnqueued,         retryInputMessages,-        reduceInputMessages) where+        reduceInputMessages,+        filterInputMessages) where  import Simulation.Aivika.Trans.Simulation import Simulation.Aivika.Trans.Event@@ -47,3 +48,5 @@ retryInputMessages :: InputMessageQueue -> TimeWarp DIO ()  reduceInputMessages :: InputMessageQueue -> Double -> IO ()++filterInputMessages :: (Message -> Bool) -> InputMessageQueue -> IO [Message]

@@ -0,0 +1,107 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.KeepAliveManager+-- Copyright  : Copyright (c) 2015-2017, 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 is responsible for delivering keep-alive messages.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.KeepAliveManager+       (KeepAliveManager,+        KeepAliveParams(..),+        newKeepAliveManager,+        addKeepAliveReceiver,+        trySendKeepAlive,+        trySendKeepAliveUTC) where++import qualified Data.Set as S+import Data.Maybe+import Data.IORef+import Data.Time.Clock++import Control.Monad+import Control.Monad.Trans+import qualified Control.Distributed.Process as DP++import Simulation.Aivika.Distributed.Optimistic.Internal.Priority+import Simulation.Aivika.Distributed.Optimistic.Internal.Message++-- | The keep-alive parameters.+data KeepAliveParams =+  KeepAliveParams { keepAliveLoggingPriority :: Priority,+                    -- ^ the logging priority+                    keepAliveInterval :: Int+                    -- ^ the interval in microseconds to send keep-alive messages+                  }++-- | The keep-alive manager.+data KeepAliveManager =+  KeepAliveManager { keepAliveParams :: KeepAliveParams,+                     -- ^ the manager parameter+                     keepAliveTimestamp :: IORef UTCTime,+                     -- ^ the keep alive timestamp+                     keepAliveReceivers :: IORef (S.Set DP.ProcessId)+                     -- ^ the receivers of the keep-alive messages+                   }++-- | Create a new keep-alive manager.+newKeepAliveManager :: KeepAliveParams -> IO KeepAliveManager+newKeepAliveManager ps =+  do timestamp <- getCurrentTime >>= newIORef+     receivers <- newIORef S.empty+     return KeepAliveManager { keepAliveParams = ps,+                               keepAliveTimestamp = timestamp,+                               keepAliveReceivers = receivers }++-- | Add the keep-alive message receiver.+addKeepAliveReceiver :: KeepAliveManager -> DP.ProcessId -> IO ()+addKeepAliveReceiver manager pid =+  modifyIORef (keepAliveReceivers manager) $+  S.insert pid++-- | Try to send a keep-alive message.+trySendKeepAlive :: KeepAliveManager -> DP.Process ()+trySendKeepAlive manager =+  do empty <- liftIO $ fmap S.null $ readIORef (keepAliveReceivers manager)+     unless empty $ +       do utc <- liftIO getCurrentTime+          trySendKeepAliveUTC manager utc++-- | Try to send a keep-alive message by the specified current time.+trySendKeepAliveUTC :: KeepAliveManager -> UTCTime -> DP.Process ()+trySendKeepAliveUTC manager utc =+  do empty <- liftIO $ fmap S.null $ readIORef (keepAliveReceivers manager)+     unless empty $ +       do f <- liftIO $ shouldSendKeepAlive manager utc+          when f $+            do ---+               logKeepAliveManager manager INFO $+                 "Sending a keep-alive message"+               ---+               liftIO $ writeIORef (keepAliveTimestamp manager) utc+               pids <- liftIO $ readIORef (keepAliveReceivers manager)+               forM_ pids $ \pid ->+                 DP.send pid KeepAliveMessage++-- | Whether should send a keep-alive message.+shouldSendKeepAlive :: KeepAliveManager -> UTCTime -> IO Bool+shouldSendKeepAlive manager utc =+  do utc0 <- readIORef (keepAliveTimestamp manager)+     let dt = fromRational $ toRational (diffUTCTime utc utc0)+     return $+       secondsToMicroseconds dt > (keepAliveInterval $ keepAliveParams manager)++-- | Convert seconds to microseconds.+secondsToMicroseconds :: Double -> Int+secondsToMicroseconds x = fromInteger $ toInteger $ round (1000000 * x)+  +-- | Log the message with the specified priority.+logKeepAliveManager :: KeepAliveManager -> Priority -> String -> DP.Process ()+{-# INLINE logKeepAliveManager #-}+logKeepAliveManager manager p message =+  when (keepAliveLoggingPriority (keepAliveParams manager) <= p) $+  DP.say $+  embracePriority p ++ " " ++ message

@@ -3,7 +3,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Message--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -18,7 +18,9 @@         AcknowledgmentMessage(..),         acknowledgmentMessage,         LocalProcessMessage(..),-        TimeServerMessage(..)) where+        TimeServerMessage(..),+        InboxProcessMessage(..),+        KeepAliveMessage(..)) where  import GHC.Generics @@ -117,9 +119,13 @@                            -- ^ the time server requests for a local minimum time                          | GlobalTimeMessage Double                            -- ^ the time server sent a global time-                         | TerminateLocalProcessMessage-                           -- ^ the time server asked to terminate the process-                         deriving (Eq, Show, Typeable, Generic)+                         | ProcessMonitorNotificationMessage DP.ProcessMonitorNotification+                           -- ^ the process monitor notification+                         | ReconnectProcessMessage DP.ProcessId+                           -- ^ finish reconnecting to the specified process+                         | KeepAliveLocalProcessMessage KeepAliveMessage+                           -- ^ the keep alive message for the local process+                         deriving (Show, Typeable, Generic)  instance Binary LocalProcessMessage @@ -128,13 +134,58 @@                          -- ^ register the local process in the time server                        | UnregisterLocalProcessMessage DP.ProcessId                          -- ^ unregister the local process from the time server+                       | TerminateTimeServerMessage DP.ProcessId+                         -- ^ the local process asked to terminate the time server                        | RequestGlobalTimeMessage DP.ProcessId                          -- ^ the local process requested for the global minimum time                        | LocalTimeMessage DP.ProcessId Double                          -- ^ the local process sent its local minimum time-                       | TerminateTimeServerMessage DP.ProcessId-                         -- ^ the local process asked to terminate the time server+                       | ReMonitorTimeServerMessage [DP.ProcessId]+                         -- ^ re-monitor the local processes by their identifiers                        deriving (Eq, Show, Typeable, Generic)  instance Binary TimeServerMessage +-- | The message destined directly for the inbox process.+data InboxProcessMessage = MonitorProcessMessage DP.ProcessId+                           -- ^ monitor the local process by its inbox process identifier+                         | ReMonitorProcessMessage [DP.ProcessId]+                           -- ^ re-monitor the local processes by their identifiers+                         | TrySendKeepAliveMessage+                           -- ^ try to send a keep alive message+                         | SendQueueMessage DP.ProcessId Message+                           -- ^ send a queue message via the inbox process+                         | SendQueueMessageBulk DP.ProcessId [Message]+                           -- ^ send a bulk of queue messages via the inbox process+                         | SendAcknowledgmentQueueMessage DP.ProcessId AcknowledgmentMessage+                           -- ^ send an acknowledgment message via the inbox process+                         | SendAcknowledgmentQueueMessageBulk DP.ProcessId [AcknowledgmentMessage]+                           -- ^ send a bulk of acknowledgment messages via the inbox process+                         | SendLocalTimeMessage DP.ProcessId DP.ProcessId Double+                           -- ^ send the local time message to the time server+                         | SendRequestGlobalTimeMessage DP.ProcessId DP.ProcessId+                           -- ^ send the request for the global virtual time+                         | SendRegisterLocalProcessMessage DP.ProcessId DP.ProcessId+                           -- ^ register the local process in the time server+                         | SendUnregisterLocalProcessMessage DP.ProcessId DP.ProcessId+                           -- ^ unregister the local process from the time server+                         | SendTerminateTimeServerMessage DP.ProcessId DP.ProcessId+                           -- ^ the local process asked to terminate the time server+                         | RegisterLocalProcessAcknowledgmentMessage DP.ProcessId+                           -- ^ after registering the local process in the time server+                         | UnregisterLocalProcessAcknowledgmentMessage DP.ProcessId+                           -- ^ after unregistering the local process from the time server+                         | TerminateTimeServerAcknowledgmentMessage DP.ProcessId+                           -- ^ after started terminating the time server+                         | TerminateInboxProcessMessage+                           -- ^ terminate the inbox process+                         deriving (Eq, Show, Typeable, Generic)++instance Binary InboxProcessMessage++-- | The keep-alive message type.+data KeepAliveMessage = KeepAliveMessage+                        -- ^ the keel-alive message.+                        deriving (Eq, Show, Typeable, Generic)++instance Binary KeepAliveMessage

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -122,14 +122,12 @@ -- | Deliver the message on low level. deliverMessage :: Message -> DIO () deliverMessage x =-  liftDistributedUnsafe $-  DP.send (messageReceiverId x) (QueueMessage x)+  sendMessageDIO (messageReceiverId x) x  -- | Deliver the anti-message on low level. deliverAntiMessage :: Message -> DIO () deliverAntiMessage x =-  liftDistributedUnsafe $-  DP.send (messageReceiverId x) (QueueMessage x)+  sendMessageDIO (messageReceiverId x) x  -- | Deliver the anti-messages on low level. deliverAntiMessages :: [Message] -> DIO ()@@ -137,6 +135,5 @@   let ys = groupBy (\a b -> messageReceiverId a == messageReceiverId b) xs       dlv []         = return ()       dlv zs@(z : _) =-        liftDistributedUnsafe $-        DP.send (messageReceiverId z) (QueueMessageBulk zs)+        sendMessagesDIO (messageReceiverId z) zs   in forM_ ys dlv

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -3,7 +3,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Priority--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -1,73 +1,15 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Ref--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental -- Tested with: GHC 7.10.3 ----- The implementation of mutable references.+-- The implementation of mutable strict references. -- module Simulation.Aivika.Distributed.Optimistic.Internal.Ref-       (Ref,-        newRef,-        newRef0,-        readRef,-        writeRef,-        modifyRef) where--import Data.IORef--import Control.Monad-import Control.Monad.Trans--import Simulation.Aivika.Trans.Comp-import Simulation.Aivika.Trans.Internal.Types--import Simulation.Aivika.Distributed.Optimistic.Internal.DIO-import Simulation.Aivika.Distributed.Optimistic.Internal.Event-import Simulation.Aivika.Distributed.Optimistic.Internal.IO-import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog---- | A mutable reference.-newtype Ref a = Ref { refValue :: IORef a }--instance Eq (Ref a) where-  (Ref r1) == (Ref r2) = r1 == r2---- | Create a new reference.-newRef :: a -> Simulation DIO (Ref a)-newRef = liftComp . newRef0---- | Create a new reference.-newRef0 :: a -> DIO (Ref a)-newRef0 a =-  do x <- liftIOUnsafe $ newIORef a-     return Ref { refValue = x }-     --- | Read the value of a reference.-readRef :: Ref a -> Event DIO a-readRef r = Event $ \p ->-  liftIOUnsafe $ readIORef (refValue r)---- | Write a new value into the reference.-writeRef :: Ref a -> a -> Event DIO ()-writeRef r a = Event $ \p ->-  do let log = queueLog $ runEventQueue (pointRun p)-     a0 <- liftIOUnsafe $ readIORef (refValue r)-     invokeEvent p $-       writeLog log $-       liftIOUnsafe $ writeIORef (refValue r) a0-     a `seq` liftIOUnsafe $ writeIORef (refValue r) a+       (module Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict) where --- | Mutate the contents of the reference.-modifyRef :: Ref a -> (a -> a) -> Event DIO ()-modifyRef r f = Event $ \p ->-  do let log = queueLog $ runEventQueue (pointRun p)-     a <- liftIOUnsafe $ readIORef (refValue r)-     let b = f a-     invokeEvent p $-       writeLog log $-       liftIOUnsafe $ writeIORef (refValue r) a-     b `seq` liftIOUnsafe $ writeIORef (refValue r) b+import Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict

@@ -1,29 +0,0 @@---- |--- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Ref--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>--- License    : BSD3--- Maintainer : David Sorokin <david.sorokin@gmail.com>--- Stability  : experimental--- Tested with: GHC 7.10.3------ This is an hs-boot file.----module Simulation.Aivika.Distributed.Optimistic.Internal.Ref where--import Simulation.Aivika.Trans.Internal.Types-import Simulation.Aivika.Distributed.Optimistic.Internal.DIO--data Ref a--instance Eq (Ref a)--newRef :: a -> Simulation DIO (Ref a)--newRef0 :: a -> DIO (Ref a)-     -readRef :: Ref a -> Event DIO a--writeRef :: Ref a -> a -> Event DIO ()--modifyRef :: Ref a -> (a -> a) -> Event DIO ()

@@ -0,0 +1,73 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Lazy+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- The implementation of mutable lazy references.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Lazy+       (Ref,+        newRef,+        newRef0,+        readRef,+        writeRef,+        modifyRef) where++import Data.IORef++import Control.Monad+import Control.Monad.Trans++import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.Internal.Types++import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.Event+import Simulation.Aivika.Distributed.Optimistic.Internal.IO+import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog++-- | A mutable lazy reference.+newtype Ref a = Ref { refValue :: IORef a }++instance Eq (Ref a) where+  (Ref r1) == (Ref r2) = r1 == r2++-- | Create a new reference.+newRef :: a -> Simulation DIO (Ref a)+newRef = liftComp . newRef0++-- | Create a new reference.+newRef0 :: a -> DIO (Ref a)+newRef0 a =+  do x <- liftIOUnsafe $ newIORef a+     return Ref { refValue = x }+     +-- | Read the value of a reference.+readRef :: Ref a -> Event DIO a+readRef r = Event $ \p ->+  liftIOUnsafe $ readIORef (refValue r)++-- | Write a new value into the reference.+writeRef :: Ref a -> a -> Event DIO ()+writeRef r a = Event $ \p ->+  do let log = queueLog $ runEventQueue (pointRun p)+     a0 <- liftIOUnsafe $ readIORef (refValue r)+     invokeEvent p $+       writeLog log $+       liftIOUnsafe $ writeIORef (refValue r) a0+     liftIOUnsafe $ writeIORef (refValue r) a++-- | Mutate the contents of the reference.+modifyRef :: Ref a -> (a -> a) -> Event DIO ()+modifyRef r f = Event $ \p ->+  do let log = queueLog $ runEventQueue (pointRun p)+     a <- liftIOUnsafe $ readIORef (refValue r)+     let b = f a+     invokeEvent p $+       writeLog log $+       liftIOUnsafe $ writeIORef (refValue r) a+     liftIOUnsafe $ writeIORef (refValue r) b

@@ -0,0 +1,73 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- The implementation of mutable strict references.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict+       (Ref,+        newRef,+        newRef0,+        readRef,+        writeRef,+        modifyRef) where++import Data.IORef++import Control.Monad+import Control.Monad.Trans++import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.Internal.Types++import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.Event+import Simulation.Aivika.Distributed.Optimistic.Internal.IO+import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog++-- | A mutable reference.+newtype Ref a = Ref { refValue :: IORef a }++instance Eq (Ref a) where+  (Ref r1) == (Ref r2) = r1 == r2++-- | Create a new reference.+newRef :: a -> Simulation DIO (Ref a)+newRef = liftComp . newRef0++-- | Create a new reference.+newRef0 :: a -> DIO (Ref a)+newRef0 a =+  do x <- liftIOUnsafe $ newIORef a+     return Ref { refValue = x }+     +-- | Read the value of a reference.+readRef :: Ref a -> Event DIO a+readRef r = Event $ \p ->+  liftIOUnsafe $ readIORef (refValue r)++-- | Write a new value into the reference.+writeRef :: Ref a -> a -> Event DIO ()+writeRef r a = Event $ \p ->+  do let log = queueLog $ runEventQueue (pointRun p)+     a0 <- liftIOUnsafe $ readIORef (refValue r)+     invokeEvent p $+       writeLog log $+       liftIOUnsafe $ writeIORef (refValue r) a0+     a `seq` liftIOUnsafe $ writeIORef (refValue r) a++-- | Mutate the contents of the reference.+modifyRef :: Ref a -> (a -> a) -> Event DIO ()+modifyRef r f = Event $ \p ->+  do let log = queueLog $ runEventQueue (pointRun p)+     a <- liftIOUnsafe $ readIORef (refValue r)+     let b = f a+     invokeEvent p $+       writeLog log $+       liftIOUnsafe $ writeIORef (refValue r) a+     b `seq` liftIOUnsafe $ writeIORef (refValue r) b

@@ -0,0 +1,29 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This is an hs-boot file.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict where++import Simulation.Aivika.Trans.Internal.Types+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO++data Ref a++instance Eq (Ref a)++newRef :: a -> Simulation DIO (Ref a)++newRef0 :: a -> DIO (Ref a)+     +readRef :: Ref a -> Event DIO a++writeRef :: Ref a -> a -> Event DIO ()++modifyRef :: Ref a -> (a -> a) -> Event DIO ()

@@ -0,0 +1,26 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper+-- Copyright  : Copyright (c) 2015-2017, 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 defines a signal helper.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper+       (newDIOSignalSource0,+        handleDIOSignalComposite) where++import Simulation.Aivika.Trans++import Simulation.Aivika.Distributed.Optimistic.DIO++-- | Create a new signal source in safe manner without constraints.+newDIOSignalSource0 :: DIO (SignalSource DIO a)+newDIOSignalSource0 = newSignalSource0++-- | Handle the signal in safe manner without constraints.+handleDIOSignalComposite :: Signal DIO a -> (a -> Event DIO ()) -> Composite DIO ()+handleDIOSignalComposite = handleSignalComposite

@@ -0,0 +1,22 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- This is an hs-boot file.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper+       (newDIOSignalSource0,+        handleDIOSignalComposite) where++import Simulation.Aivika.Trans++import Simulation.Aivika.Distributed.Optimistic.Internal.DIO++newDIOSignalSource0 :: DIO (SignalSource DIO a)++handleDIOSignalComposite :: Signal DIO a -> (a -> Event DIO ()) -> Composite DIO ()

@@ -3,7 +3,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -29,6 +29,8 @@  import Control.Monad import Control.Monad.Trans+import Control.Exception+import qualified Control.Monad.Catch as C import Control.Concurrent import qualified Control.Distributed.Process as DP @@ -39,10 +41,20 @@ data TimeServerParams =   TimeServerParams { tsLoggingPriority :: Priority,                      -- ^ the logging priority-                     tsExpectTimeout :: Int,-                     -- ^ the timeout in microseconds within which a new message is expected-                     tsTimeSyncDelay :: Int+                     tsReceiveTimeout :: Int,+                     -- ^ the timeout in microseconds used when receiving messages+                     tsTimeSyncTimeout :: Int,+                     -- ^ the timeout in microseconds used for the time synchronization sessions+                     tsTimeSyncDelay :: Int,                      -- ^ the delay in microseconds between the time synchronization sessions+                     tsProcessMonitoringEnabled :: Bool,+                     -- ^ Whether the process monitoring is enabled+                     tsProcessMonitoringDelay :: Int,+                     -- ^ The delay in microseconds which must be applied for monitoring every remote process.+                     tsProcessReconnectingEnabled :: Bool,+                     -- ^ Whether the automatic reconnecting to processes is enabled when enabled monitoring+                     tsProcessReconnectingDelay :: Int+                     -- ^ the delay in microseconds before reconnecting                    } deriving (Eq, Ord, Show, Typeable, Generic)  instance Binary TimeServerParams@@ -55,41 +67,56 @@                -- ^ the initial quorum of registered local processes to start the simulation                tsInInit :: IORef Bool,                -- ^ whether the time server is in the initial mode+               tsTerminating :: IORef Bool,+               -- ^ whether the time server is in the terminating mode                tsProcesses :: IORef (M.Map DP.ProcessId LocalProcessInfo),                -- ^ the information about local processes                tsProcessesInFind :: IORef (S.Set DP.ProcessId),                -- ^ the processed used in the current finding of the global time-               tsGlobalTime :: IORef (Maybe Double)+               tsGlobalTime :: IORef (Maybe Double),                -- ^ the global time of the model+               tsGlobalTimeTimestamp :: IORef (Maybe UTCTime)+               -- ^ the global time timestamp              }  -- | The information about the local process. data LocalProcessInfo =-  LocalProcessInfo { lpLocalTime :: IORef (Maybe Double)+  LocalProcessInfo { lpLocalTime :: IORef (Maybe Double),                      -- ^ the local time of the process+                     lpMonitorRef :: Maybe DP.MonitorRef+                     -- ^ the local process monitor reference                    }  -- | The default time server parameters. defaultTimeServerParams :: TimeServerParams defaultTimeServerParams =   TimeServerParams { tsLoggingPriority = WARNING,-                     tsExpectTimeout = 100000,-                     tsTimeSyncDelay = 1000000+                     tsReceiveTimeout = 100000,+                     tsTimeSyncTimeout = 60000000,+                     tsTimeSyncDelay = 1000000,+                     tsProcessMonitoringEnabled = False,+                     tsProcessMonitoringDelay = 3000000,+                     tsProcessReconnectingEnabled = False,+                     tsProcessReconnectingDelay = 5000000                    }  -- | Create a new time server by the specified initial quorum and parameters. newTimeServer :: Int -> TimeServerParams -> IO TimeServer newTimeServer n ps =   do f  <- newIORef True+     ft <- newIORef False      m  <- newIORef M.empty      s  <- newIORef S.empty      t0 <- newIORef Nothing+     t' <- newIORef Nothing      return TimeServer { tsParams = ps,                          tsInitQuorum = n,                          tsInInit = f,+                         tsTerminating = ft,                          tsProcesses = m,                          tsProcessesInFind = s,-                         tsGlobalTime = t0+                         tsGlobalTime = t0,+                         tsGlobalTimeTimestamp = t'                        }  -- | Process the time server message.@@ -105,9 +132,18 @@        Nothing  ->          do t <- newIORef Nothing             modifyIORef (tsProcesses server) $-              M.insert pid LocalProcessInfo { lpLocalTime = t }+              M.insert pid LocalProcessInfo { lpLocalTime = t, lpMonitorRef = Nothing }             return $-              tryStartTimeServer server+              do when (tsProcessMonitoringEnabled $ tsParams server) $+                   do logTimeServer server INFO $+                        "Time Server: monitoring the process by identifier " ++ show pid+                      r <- DP.monitor pid+                      liftIO $+                        modifyIORef (tsProcesses server) $+                        M.update (\x -> Just x { lpMonitorRef = Just r }) pid+                 serverId <- DP.getSelfPid+                 DP.send pid (RegisterLocalProcessAcknowledgmentMessage serverId)+                 tryStartTimeServer server processTimeServerMessage server (UnregisterLocalProcessMessage pid) =   join $ liftIO $   do m <- readIORef (tsProcesses server)@@ -122,19 +158,41 @@             modifyIORef (tsProcessesInFind server) $               S.delete pid             return $-              tryProvideTimeServerGlobalTime server+              do when (tsProcessMonitoringEnabled $ tsParams server) $+                   case lpMonitorRef x of+                     Nothing -> return ()+                     Just r  ->+                       do logTimeServer server INFO $+                            "Time Server: unmonitoring the process by identifier " ++ show pid+                          DP.unmonitor r+                 serverId <- DP.getSelfPid+                 DP.send pid (UnregisterLocalProcessAcknowledgmentMessage serverId)+                 tryProvideTimeServerGlobalTime server+                 tryTerminateTimeServer server processTimeServerMessage server (TerminateTimeServerMessage pid) =-  do pids <--       liftIO $-       do m <- readIORef (tsProcesses server)-          writeIORef (tsProcesses server) M.empty-          writeIORef (tsProcessesInFind server) S.empty-          writeIORef (tsGlobalTime server) Nothing-          return $ filter (/= pid) (M.keys m)-     forM_ pids $ \pid ->-       DP.send pid TerminateLocalProcessMessage-     logTimeServer server INFO "Time Server: terminating..."-     DP.terminate+  join $ liftIO $+  do m <- readIORef (tsProcesses server)+     case M.lookup pid m of+       Nothing ->+         return $+         logTimeServer server WARNING $+         "Time Server: unknown process identifier " ++ show pid+       Just x  ->+         do modifyIORef (tsProcesses server) $+              M.delete pid+            modifyIORef (tsProcessesInFind server) $+              S.delete pid+            return $+              do when (tsProcessMonitoringEnabled $ tsParams server) $+                   case lpMonitorRef x of+                     Nothing -> return ()+                     Just r  ->+                       do logTimeServer server INFO $+                            "Time Server: unmonitoring the process by identifier " ++ show pid+                          DP.unmonitor r+                 serverId <- DP.getSelfPid+                 DP.send pid (TerminateTimeServerAcknowledgmentMessage serverId)+                 startTerminatingTimeServer server processTimeServerMessage server (RequestGlobalTimeMessage pid) =   tryComputeTimeServerGlobalTime server processTimeServerMessage server (LocalTimeMessage pid t') =@@ -153,6 +211,16 @@               S.delete pid             return $               tryProvideTimeServerGlobalTime server+processTimeServerMessage server (ReMonitorTimeServerMessage pids) =+  do forM_ pids $ \pid ->+       do ---+          logTimeServer server NOTICE $ "Time Server: re-monitoring " ++ show pid+          ---+          DP.monitor pid+          ---+          logTimeServer server NOTICE $ "Time Server: started re-monitoring " ++ show pid+          ---+     resetComputingTimeServerGlobalTime server  -- | Whether the both values are defined and the first is greater than or equaled to the second. (.>=.) :: Maybe Double -> Maybe Double -> Bool@@ -197,6 +265,16 @@                  else return $                       computeTimeServerGlobalTime server +-- | Reset computing the time server global time.+resetComputingTimeServerGlobalTime :: TimeServer -> DP.Process ()+resetComputingTimeServerGlobalTime server =+  do logTimeServer server NOTICE $+       "Time Server: reset computing the global time"+     liftIO $+       do utc <- getCurrentTime+          writeIORef (tsProcessesInFind server) S.empty+          writeIORef (tsGlobalTimeTimestamp server) (Just utc)+ -- | Try to provide the local processes wth the global time.  tryProvideTimeServerGlobalTime :: TimeServer -> DP.Process () tryProvideTimeServerGlobalTime server =@@ -229,7 +307,7 @@ provideTimeServerGlobalTime :: TimeServer -> DP.Process () provideTimeServerGlobalTime server =   do t0 <- liftIO $ timeServerGlobalTime server-     logTimeServer server DEBUG $+     logTimeServer server INFO $        "Time Server: providing the global time = " ++ show t0      case t0 of        Nothing -> return ()@@ -238,7 +316,9 @@             when (t' .>. Just t0) $               logTimeServer server NOTICE               "Time Server: the global time has decreased"+            timestamp <- liftIO getCurrentTime             liftIO $ writeIORef (tsGlobalTime server) (Just t0)+            liftIO $ writeIORef (tsGlobalTimeTimestamp server) (Just timestamp)             zs <- liftIO $ fmap M.assocs $ readIORef (tsProcesses server)             forM_ zs $ \(pid, x) ->               DP.send pid (GlobalTimeMessage t0)@@ -261,10 +341,44 @@                            Just _  ->                              loop zs' (liftM2 min t acc) +-- | Start terminating the time server.+startTerminatingTimeServer :: TimeServer -> DP.Process ()+startTerminatingTimeServer server =+  do logTimeServer server INFO "Time Server: start terminating..."+     liftIO $+       writeIORef (tsTerminating server) True+     tryTerminateTimeServer server++-- | Try to terminate the time server.+tryTerminateTimeServer :: TimeServer -> DP.Process ()+tryTerminateTimeServer server =+  do f <- liftIO $ readIORef (tsTerminating server)+     when f $+       do m <- liftIO $ readIORef (tsProcesses server)+          when (M.null m) $+            do logTimeServer server INFO "Time Server: terminate"+               DP.terminate+ -- | Convert seconds to microseconds. secondsToMicroseconds :: Double -> Int secondsToMicroseconds x = fromInteger $ toInteger $ round (1000000 * x) +-- | The internal time server message.+data InternalTimeServerMessage = InternalTimeServerMessage TimeServerMessage+                                 -- ^ the time server message+                               | InternalProcessMonitorNotification DP.ProcessMonitorNotification+                                 -- ^ the process monitor notification+                               | InternalKeepAliveMessage KeepAliveMessage+                                 -- ^ the keep alive message++-- | Handle the time server exception+handleTimeServerException :: TimeServer -> SomeException -> DP.Process ()+handleTimeServerException server e =+  do ---+     logTimeServer server ERROR $ "Exception occured: " ++ show e+     ---+     C.throwM e+ -- | Start the time server by the specified initial quorum and parameters. -- The quorum defines the number of local processes that must be registered in -- the time server before the global time synchronization is started.@@ -273,22 +387,95 @@   do server <- liftIO $ newTimeServer n ps      logTimeServer server INFO "Time Server: starting..."      let loop utc0 =-           do m <- DP.expectTimeout (tsExpectTimeout ps) :: DP.Process (Maybe TimeServerMessage)-              case m of+           do let f1 :: TimeServerMessage -> DP.Process InternalTimeServerMessage+                  f1 x = return (InternalTimeServerMessage x)+                  f2 :: DP.ProcessMonitorNotification -> DP.Process InternalTimeServerMessage+                  f2 x = return (InternalProcessMonitorNotification x)+                  f3 :: KeepAliveMessage -> DP.Process InternalTimeServerMessage+                  f3 x = return (InternalKeepAliveMessage x)+              a <- DP.receiveTimeout (tsReceiveTimeout ps) [DP.match f1, DP.match f2, DP.match f3]+              case a of                 Nothing -> return ()-                Just m  ->+                Just (InternalTimeServerMessage m) ->                   do ---                      logTimeServer server DEBUG $                        "Time Server: " ++ show m                      ---                      processTimeServerMessage server m+                Just (InternalProcessMonitorNotification m) ->+                  handleProcessMonitorNotification m server+                Just (InternalKeepAliveMessage m) ->+                  do ---+                     logTimeServer server DEBUG $+                       "Time Server: " ++ show m+                     ---+                     return ()               utc <- liftIO getCurrentTime-              let dt = fromRational $ toRational (diffUTCTime utc utc0)-              if secondsToMicroseconds dt > tsTimeSyncDelay ps+              timestamp <- liftIO $ readIORef (tsGlobalTimeTimestamp server)+              case timestamp of+                Just x | shouldResetComputingTimeServerGlobalTime server x utc ->+                  resetComputingTimeServerGlobalTime server+                _ -> return ()+              if shouldComputeTimeServerGlobalTime server utc0 utc                 then do tryComputeTimeServerGlobalTime server                         loop utc                 else loop utc0-     liftIO getCurrentTime >>= loop +     C.catch (liftIO getCurrentTime >>= loop) (handleTimeServerException server) ++-- | Handle the process monitor notification.+handleProcessMonitorNotification :: DP.ProcessMonitorNotification -> TimeServer -> DP.Process ()+handleProcessMonitorNotification m@(DP.ProcessMonitorNotification _ pid0 reason) server =+  do let ps = tsParams server+         recv m@(DP.ProcessMonitorNotification _ _ _) = +           do ---+              logTimeServer server WARNING $+                "Time Server: received a process monitor notification " ++ show m+              ---+              return m+     recv m+     when (tsProcessReconnectingEnabled ps && reason == DP.DiedDisconnect) $+       do liftIO $+            threadDelay (tsProcessReconnectingDelay ps)+          let pred m@(DP.ProcessMonitorNotification _ _ reason) = reason == DP.DiedDisconnect+              loop :: [DP.ProcessId] -> DP.Process [DP.ProcessId]+              loop acc =+                do y <- DP.receiveTimeout 0 [DP.matchIf pred recv]+                   case y of+                     Nothing -> return $ reverse acc+                     Just m@(DP.ProcessMonitorNotification _ pid _) -> loop (pid : acc)+          pids <- loop [pid0]+          ---+          logTimeServer server NOTICE "Begin reconnecting..."+          ---+          forM_ pids $ \pid ->+            do ---+               logTimeServer server NOTICE $+                 "Time Server: reconnecting to " ++ show pid+               ---+               DP.reconnect pid+          serverId <- DP.getSelfPid+          DP.spawnLocal $+            let action =+                  do liftIO $+                       threadDelay (tsProcessMonitoringDelay ps)+                     ---+                     logTimeServer server NOTICE $ "Time Server: proceed to the re-monitoring"+                     ---+                     DP.send serverId (ReMonitorTimeServerMessage pids)+            in C.catch action (handleTimeServerException server)+          return ()++-- | Test whether should compute the time server global time.+shouldComputeTimeServerGlobalTime :: TimeServer -> UTCTime -> UTCTime -> Bool+shouldComputeTimeServerGlobalTime server utc0 utc =+  let dt = fromRational $ toRational (diffUTCTime utc utc0)+  in secondsToMicroseconds dt > (tsTimeSyncDelay $ tsParams server)++-- | Test whether should reset computing the time server global time.+shouldResetComputingTimeServerGlobalTime :: TimeServer -> UTCTime -> UTCTime -> Bool+shouldResetComputingTimeServerGlobalTime server utc0 utc =+  let dt = fromRational $ toRational (diffUTCTime utc utc0)+  in secondsToMicroseconds dt > (tsTimeSyncTimeout $ tsParams server)  -- | A curried version of 'timeServer' for starting the time server on remote node. curryTimeServer :: (Int, TimeServerParams) -> DP.Process ()

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental@@ -14,6 +14,7 @@         newTransientMessageQueue,         transientMessageQueueSize,         transientMessageQueueTime,+        transientMessages,         enqueueTransientMessage,         processAcknowledgmentMessage,         acknowledgmentMessageTime,@@ -21,7 +22,7 @@         deliverAcknowledgmentMessage,         deliverAcknowledgmentMessages) where -import qualified Data.Set as S+import qualified Data.Map as M import Data.List import Data.IORef @@ -35,8 +36,8 @@  -- | Specifies the transient message queue. data TransientMessageQueue =-  TransientMessageQueue { queuePrototypeMessages :: IORef (S.Set TransientMessageQueueItem),-                          -- ^ The prototype messages.+  TransientMessageQueue { queueTransientMessages :: IORef (M.Map TransientMessageQueueItem Message),+                          -- ^ The transient messages.                           queueMarkedMessageTime :: IORef Double                           -- ^ The marked acknowledgment message time.                         }@@ -97,30 +98,36 @@ -- | Create a new transient message queue. newTransientMessageQueue :: DIO TransientMessageQueue newTransientMessageQueue =-  do ms <- liftIOUnsafe $ newIORef S.empty+  do ms <- liftIOUnsafe $ newIORef M.empty      r  <- liftIOUnsafe $ newIORef (1 / 0)-     return TransientMessageQueue { queuePrototypeMessages = ms,+     return TransientMessageQueue { queueTransientMessages = ms,                                     queueMarkedMessageTime = r }  -- | Get the transient message queue size. transientMessageQueueSize :: TransientMessageQueue -> IO Int transientMessageQueueSize q =-  fmap S.size $ readIORef (queuePrototypeMessages q)+  fmap M.size $ readIORef (queueTransientMessages q)  -- | Get the virtual time of the transient message queue. transientMessageQueueTime :: TransientMessageQueue -> IO Double transientMessageQueueTime q =-  do s <- readIORef (queuePrototypeMessages q)-     if S.null s+  do ms <- readIORef (queueTransientMessages q)+     if M.null ms        then return (1 / 0)-       else let m = S.findMin s+       else let (m, _) = M.findMin ms             in return (itemReceiveTime m) +-- | Get the transient messages.+transientMessages :: TransientMessageQueue -> IO [Message]+transientMessages q =+  do ms <- readIORef (queueTransientMessages q)+     return (M.elems ms)+ -- | Enqueue the transient message. enqueueTransientMessage :: TransientMessageQueue -> Message -> IO () enqueueTransientMessage q m =-  modifyIORef (queuePrototypeMessages q) $-  S.insert (transientMessageQueueItem m)+  modifyIORef (queueTransientMessages q) $+  M.insert (transientMessageQueueItem m) m  -- | Enqueue the acknowledgment message. enqueueAcknowledgmentMessage :: TransientMessageQueue -> AcknowledgmentMessage -> IO ()@@ -131,10 +138,13 @@ -- | Process the acknowledgment message. processAcknowledgmentMessage :: TransientMessageQueue -> AcknowledgmentMessage -> IO ()  processAcknowledgmentMessage q m =-  do modifyIORef (queuePrototypeMessages q) $-       S.delete (acknowledgmentMessageQueueItem m)-     when (acknowledgmentMarked m) $-       enqueueAcknowledgmentMessage q m+  do ms <- readIORef (queueTransientMessages q)+     let k = acknowledgmentMessageQueueItem m+     when (M.member k ms) $+       do modifyIORef (queueTransientMessages q) $+            M.delete k+          when (acknowledgmentMarked m) $+            enqueueAcknowledgmentMessage q m  -- | Get the minimal marked acknowledgment message time. acknowledgmentMessageTime :: TransientMessageQueue -> IO Double@@ -149,8 +159,7 @@ -- | Deliver the acknowledgment message on low level. deliverAcknowledgmentMessage :: AcknowledgmentMessage -> DIO () deliverAcknowledgmentMessage x =-  liftDistributedUnsafe $-  DP.send (acknowledgmentSenderId x) (AcknowledgmentQueueMessage x)+  sendAcknowledgmentMessageDIO (acknowledgmentSenderId x) x  -- | Deliver the acknowledgment messages on low level. deliverAcknowledgmentMessages :: [AcknowledgmentMessage] -> DIO ()@@ -158,6 +167,5 @@   let ys = groupBy (\a b -> acknowledgmentSenderId a == acknowledgmentSenderId b) xs       dlv []         = return ()       dlv zs@(z : _) =-        liftDistributedUnsafe $-        DP.send (acknowledgmentSenderId z) (AcknowledgmentQueueMessageBulk zs)+        sendAcknowledgmentMessagesDIO (acknowledgmentSenderId z) zs   in forM_ ys dlv

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -4,7 +4,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Message--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Priority--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -3,7 +3,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.QueueStrategy--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -1,48 +1,21 @@ -{-# LANGUAGE TypeFamilies #-}- -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Ref.Base--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental -- Tested with: GHC 7.10.3 ----- Here is an implementation of mutable references, where+-- Here is an implementation of strict mutable references, where -- 'DIO' is an instance of 'MonadRef' and 'MonadRef0'. ---module Simulation.Aivika.Distributed.Optimistic.Ref.Base () where+module Simulation.Aivika.Distributed.Optimistic.Ref.Base +       (module Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict) where  import Simulation.Aivika.Trans.Internal.Types import Simulation.Aivika.Trans.Comp import Simulation.Aivika.Trans.Ref.Base  import Simulation.Aivika.Distributed.Optimistic.Internal.DIO-import qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref as R---- | The implementation of mutable references.-instance MonadRef DIO where--  -- | The mutable reference.-  newtype Ref DIO a = Ref { refValue :: R.Ref a }--  {-# INLINE newRef #-}-  newRef = fmap Ref . R.newRef --  {-# INLINE readRef #-}-  readRef (Ref r) = R.readRef r--  {-# INLINE writeRef #-}-  writeRef (Ref r) = R.writeRef r--  {-# INLINE modifyRef #-}-  modifyRef (Ref r) = R.modifyRef r--  {-# INLINE equalRef #-}-  equalRef (Ref r1) (Ref r2) = (r1 == r2)--instance MonadRef0 DIO where--  {-# INLINE newRef0 #-}-  newRef0 = fmap Ref . R.newRef0+import Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict

@@ -0,0 +1,48 @@++{-# LANGUAGE TypeFamilies #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Ref.Base.Lazy+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- Here is an implementation of lazy mutable references, where+-- 'DIO' is an instance of 'MonadRef' and 'MonadRef0'.+--+module Simulation.Aivika.Distributed.Optimistic.Ref.Base.Lazy () where++import Simulation.Aivika.Trans.Internal.Types+import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.Ref.Base.Lazy++import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Lazy as R++-- | The implementation of lazy mutable references.+instance MonadRef DIO where++  -- | The lazy mutable reference.+  newtype Ref DIO a = Ref { refValue :: R.Ref a }++  {-# INLINE newRef #-}+  newRef = fmap Ref . R.newRef ++  {-# INLINE readRef #-}+  readRef (Ref r) = R.readRef r++  {-# INLINE writeRef #-}+  writeRef (Ref r) = R.writeRef r++  {-# INLINE modifyRef #-}+  modifyRef (Ref r) = R.modifyRef r++  {-# INLINE equalRef #-}+  equalRef (Ref r1) (Ref r2) = (r1 == r2)++instance MonadRef0 DIO where++  {-# INLINE newRef0 #-}+  newRef0 = fmap Ref . R.newRef0

@@ -0,0 +1,48 @@++{-# LANGUAGE TypeFamilies #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com>+-- License    : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability  : experimental+-- Tested with: GHC 7.10.3+--+-- Here is an implementation of strict mutable references, where+-- 'DIO' is an instance of 'MonadRef' and 'MonadRef0'.+--+module Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict () where++import Simulation.Aivika.Trans.Internal.Types+import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.Ref.Base.Strict++import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict as R++-- | The implementation of strict mutable references.+instance MonadRef DIO where++  -- | The strict mutable reference.+  newtype Ref DIO a = Ref { refValue :: R.Ref a }++  {-# INLINE newRef #-}+  newRef = fmap Ref . R.newRef ++  {-# INLINE readRef #-}+  readRef (Ref r) = R.readRef r++  {-# INLINE writeRef #-}+  writeRef (Ref r) = R.writeRef r++  {-# INLINE modifyRef #-}+  modifyRef (Ref r) = R.modifyRef r++  {-# INLINE equalRef #-}+  equalRef (Ref r1) (Ref r2) = (r1 == r2)++instance MonadRef0 DIO where++  {-# INLINE newRef0 #-}+  newRef0 = fmap Ref . R.newRef0

@@ -1,7 +1,7 @@  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.TimeServer--- Copyright  : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- Copyright  : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com> -- License    : BSD3 -- Maintainer : David Sorokin <david.sorokin@gmail.com> -- Stability  : experimental

@@ -1,20 +1,52 @@ name:            aivika-distributed-version:         0.3.1+version:         0.5 synopsis:        Parallel distributed discrete event simulation module for the Aivika library description:-    This package extends the aivika-transformers [1] package with facilities for running parallel distributed simulations.-    It uses an optimistic strategy known as the Time Warp method. To synchronize the global virtual time, since version 0.3-    it uses Samadi's algorithm.+    This package extends the aivika-transformers [1] package and allows running parallel distributed simulations.+    It uses an optimistic strategy known as the Time Warp method. To synchronize the global virtual time, +    it uses Samadi's algorithm.      .+    Moreover, this package uses the author's modification that allows recovering the distributed+    simulation after temporary connection errors whenever possible. For that, you have to enable explicitly +    the recovering mode and enable monitoring all local processes including the specialized Time Server process +    as it is shown in one of the test examples included in the distributive.+    .+    With the recovering mode enabled, you can try to build a distributed simulation using ordinary computers connected+    via the ordinary net. For example, such a distributed model could even consist of computers located in different +    continents of the Earth, where the computers could be connected through the Internet. Here the most exciting thing +    is that this is the optimistic distributed simulation with possible rollbacks. It is assumed that optimistic methods +    tend to better support the parallelism inherited in the models. +    .+    You may test the distributed simulation using your own notebook only, although the package is still destined to be +    used with a multi-core computer, or computers connected in the distributed cluster.+    .+    There are additional packages that allow you to run the distributed simulation experiments by +    the Monte-Carlo method. They allow you to save the simulation results in SQL databases and then generate a report +    or a set of reports consisting of HTML pages with charts, histograms, links to CSV tables, summary statistics etc.+    Please consult the AivikaSoft [3] website for more details.+    .+    Regarding the speed of simulation, the rough estimations are as follows. The simulation is slower up to+    6-9 times in comparison with the sequential aivika [2] simulation library using the equivalent sequential models. +    Note that you can run up to 7 parallel local processes on a single 8-core processor computer and run the Time Server +    process too. On a 36-core processor, you can launch up to 35 local processes simultaneously.+    . +    So, this estimation seems to be quite good. At the same time, the message passing between the local processes can +    dramatically decrease the speed of distributed simulation, especially if they cause rollbacks. Thus, much depends on +    the distributed model itself.+    .     \[1] <http://hackage.haskell.org/package/aivika-transformers>     .+    \[2] <http://hackage.haskell.org/package/aivika>+    .+    \[3] <http://www.aivikasoft.com>+    . category:        Simulation license:         BSD3 license-file:    LICENSE-copyright:       (c) 2015-2016. David Sorokin <david.sorokin@gmail.com>+copyright:       (c) 2015-2017. David Sorokin <david.sorokin@gmail.com> author:          David Sorokin maintainer:      David Sorokin <david.sorokin@gmail.com>-homepage:        http://www.aivikasoft.com/en/products/aivika.html+homepage:        http://www.aivikasoft.com cabal-version:   >= 1.6 build-type:      Simple tested-with:     GHC == 7.10.1@@ -26,6 +58,8 @@                      tests/MachRep2Reproducible.hs                      tests/MachRep2Sync.hs                      tests/MachRep2SyncIO.hs+                     tests/SimpleLocalnetHelper.hs+                     tests/cluster.conf                      CHANGELOG.md  library@@ -38,6 +72,8 @@                      Simulation.Aivika.Distributed.Optimistic.Message                      Simulation.Aivika.Distributed.Optimistic.Priority                      Simulation.Aivika.Distributed.Optimistic.Ref.Base+                     Simulation.Aivika.Distributed.Optimistic.Ref.Base.Lazy+                     Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict                      Simulation.Aivika.Distributed.Optimistic.TimeServer      other-modules:   Simulation.Aivika.Distributed.Optimistic.Internal.Channel@@ -45,16 +81,20 @@                      Simulation.Aivika.Distributed.Optimistic.Internal.Event                      Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue                      Simulation.Aivika.Distributed.Optimistic.Internal.IO+                     Simulation.Aivika.Distributed.Optimistic.Internal.KeepAliveManager                      Simulation.Aivika.Distributed.Optimistic.Internal.Message                      Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue                      Simulation.Aivika.Distributed.Optimistic.Internal.Priority                      Simulation.Aivika.Distributed.Optimistic.Internal.Ref+                     Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Lazy+                     Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict+                     Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper                      Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer                      Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp                      Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue                      Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog -    build-depends:   base >= 3 && < 6,+    build-depends:   base >= 4.6.0.0 && < 6,                      mtl >= 2.1.1,                      stm >= 2.4.2,                      random >= 1.0.0.3,@@ -63,8 +103,8 @@                      containers >= 0.4.0.0,                      exceptions >= 0.8.0.2,                      distributed-process >= 0.6.1,-                     aivika >= 4.5,-                     aivika-transformers >= 4.5+                     aivika >= 5.1,+                     aivika-transformers >= 5.1      extensions:      MultiParamTypeClasses,                      FlexibleInstances,

@@ -84,8 +84,6 @@      let timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG }      timeServerId  <- DP.spawnLocal $ timeServer 1 timeServerParams      runModel timeServerId-     liftIO $-       threadDelay 1000000    main :: IO () main = do

@@ -46,8 +46,8 @@                 spcMethod = RungeKutta4,                 spcGeneratorType = SimpleGenerator } --- | The time shift when replying the messages.-delta = 0.01+-- | The time shift when replying to the messages.+delta = 1e-6  data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic) data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)

@@ -65,8 +65,8 @@                 spcMethod = RungeKutta4,                 spcGeneratorType = SimpleGenerator } --- | The time shift when replying the messages.-delta = 0.01+-- | The time shift when replying to the messages.+delta = 1e-6  data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic) data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)

@@ -48,8 +48,8 @@                 spcMethod = RungeKutta4,                 spcGeneratorType = SimpleGenerator } --- | The time shift when replying the messages.-delta = 0.01+-- | The time shift when replying to the messages.+delta = 1e-6  -- | The initial seeds. seeds = [456, 789]

@@ -46,8 +46,8 @@                 spcMethod = RungeKutta4,                 spcGeneratorType = SimpleGenerator } --- | The time shift when replying the messages.-delta = 0.01+-- | The time shift when replying to the messages.+delta = 1e-6  data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic) data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)

@@ -46,8 +46,8 @@                 spcMethod = RungeKutta4,                 spcGeneratorType = SimpleGenerator } --- | The time shift when replying the messages.-delta = 0.01+-- | The time shift when replying to the messages.+delta = 1e-6  data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic) data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)

@@ -0,0 +1,70 @@++{-# LANGUAGE OverloadedStrings #-}++module SimpleLocalnetHelper+       (getSlaveBackend,+        getSlaveConfBackend,+        getMasterBackend,+        getMasterConfBackend,+        startMasterProcess) where++import System.IO+import qualified Control.Distributed.Process as DP+import qualified Control.Distributed.Process.Node as Node+import Control.Distributed.Process.Backend.SimpleLocalnet+import qualified Data.ByteString.Char8 as BS+import Network.Transport (EndPointAddress(..))++-- | Make a NodeId from "host:port" string.+makeNodeId :: String -> DP.NodeId+makeNodeId addr = DP.NodeId . EndPointAddress . BS.concat $ [BS.pack addr, ":0"]++splitAddr :: String -> (String, String)+splitAddr addr = (host, port)+  where+    (host, rest) = break (== ':') addr+    port = tail rest++deleteAt :: Int -> [String] -> [String]+deleteAt i xs = xs1 ++ xs2+  where (xs1, rest) = splitAt i xs+        xs2 = tail rest++getSlaveBackend :: Int -> [String] -> DP.RemoteTable -> IO Backend+getSlaveBackend i addrs rtable =+  do let addr = addrs !! i+         (host, port) = splitAddr addr+     backend <- initializeBackend host port rtable+     return backend++getMasterBackend :: Int -> [String] -> DP.RemoteTable -> IO (Backend, [DP.NodeId])+getMasterBackend i addrs rtable =+  do let addr = addrs !! i+         (host, port) = splitAddr addr+         slaves = map makeNodeId $ deleteAt i addrs +     backend <- initializeBackend host port rtable+     return (backend, slaves)++getClusterConf :: IO [String]+getClusterConf = fmap addresses $ readFile "cluster.conf"++addresses :: String -> [String]+addresses =+  filter (\x -> head x /= '#') .+  filter (\x -> not $ null $ filter (\y -> y /= ' ') x) .+  lines++getSlaveConfBackend :: Int -> DP.RemoteTable -> IO Backend+getSlaveConfBackend i rtable =+  do addrs <- getClusterConf+     getSlaveBackend i addrs rtable++getMasterConfBackend :: Int -> DP.RemoteTable -> IO (Backend, [DP.NodeId])+getMasterConfBackend i rtable =+  do addrs <- getClusterConf+     getMasterBackend i addrs rtable++startMasterProcess :: Backend -> [DP.NodeId] -> (Backend -> [DP.NodeId] -> DP.Process ()) -> IO ()+startMasterProcess backend slaves process =+  do node <- newLocalNode backend+     Node.runProcess node (process backend slaves)

@@ -0,0 +1,13 @@+# The Simulation Cluster Configuration++# 0: the master simulation node, which must be run last+localhost:8088++# 1: the time server+localhost:8080++# 2: the first slave simulation node+localhost:8081++# 3: the second slave simulation node+localhost:8082