aivika-distributed 1.2 → 1.3
raw patch · 12 files changed
+629/−338 lines, 12 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- CHANGELOG.md +5/−0
- Simulation/Aivika/Distributed/Optimistic/Internal/AcknowledgementMessageQueue.hs +117/−0
- Simulation/Aivika/Distributed/Optimistic/Internal/AcknowledgementMessageQueue.hs-boot +34/−0
- Simulation/Aivika/Distributed/Optimistic/Internal/ConnectionManager.hs +300/−0
- Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +67/−73
- Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +24/−8
- Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs +0/−148
- Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs +10/−13
- Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs +3/−2
- Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs +63/−90
- Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs +2/−1
- aivika-distributed.cabal +4/−3
CHANGELOG.md view
@@ -1,4 +1,9 @@ +Version 1.3+-----++* Fixed the implementation of the fault-tolerant mode.+ Version 1.2 -----
+ Simulation/Aivika/Distributed/Optimistic/Internal/AcknowledgementMessageQueue.hs view
@@ -0,0 +1,117 @@++-- |+-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.AcknowledgementMessageQueue+-- Copyright : Copyright (c) 2015-2018, David Sorokin <david.sorokin@gmail.com>+-- License : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability : experimental+-- Tested with: GHC 7.10.3+--+-- This module defines an acknowledegment message queue.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.AcknowledgementMessageQueue+ (AcknowledgementMessageQueue,+ newAcknowledgementMessageQueue,+ acknowledgementMessageQueueSize,+ enqueueAcknowledgementMessage,+ reduceAcknowledgementMessages,+ filterAcknowledgementMessages) where++import Data.Maybe+import Data.List+import Data.IORef++import Control.Monad+import Control.Monad.Trans++import Simulation.Aivika.Vector+import Simulation.Aivika.Trans.Comp+import Simulation.Aivika.Trans.Simulation+import Simulation.Aivika.Trans.Dynamics+import Simulation.Aivika.Trans.Event+import Simulation.Aivika.Trans.Signal+import Simulation.Aivika.Trans.Internal.Types++import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.IO+import Simulation.Aivika.Distributed.Optimistic.DIO++-- | Specifies the acknowledgement message queue.+data AcknowledgementMessageQueue =+ AcknowledgementMessageQueue { acknowledgementMessages :: Vector AcknowledgementMessage+ -- ^ the acknowedgement messages+ }++-- | Create a new acknowledgement message queue.+newAcknowledgementMessageQueue :: DIO AcknowledgementMessageQueue+newAcknowledgementMessageQueue =+ do ms <- liftIOUnsafe newVector+ return AcknowledgementMessageQueue { acknowledgementMessages = ms }++-- | Return the acknowledgement message queue size.+acknowledgementMessageQueueSize :: AcknowledgementMessageQueue -> IO Int+acknowledgementMessageQueueSize = vectorCount . acknowledgementMessages++-- | Return a complement.+complement :: Int -> Int+complement x = - x - 1++-- | Enqueue a new acknowledement message ignoring the duplicated messages.+enqueueAcknowledgementMessage :: AcknowledgementMessageQueue -> AcknowledgementMessage -> IO ()+enqueueAcknowledgementMessage q m =+ do i <- lookupAcknowledgementMessageIndex q m+ when (i < 0) $+ do -- insert the message at the specified index+ let i' = complement i+ vectorInsert (acknowledgementMessages q) i' m++-- | Search for the message index.+lookupAcknowledgementMessageIndex' :: AcknowledgementMessageQueue -> AcknowledgementMessage -> Int -> Int -> IO Int+lookupAcknowledgementMessageIndex' q m left right =+ if left > right+ then return $ complement left+ else + do let index = (left + right) `div` 2+ m' <- readVector (acknowledgementMessages q) index+ let t' = acknowledgementReceiveTime m'+ t = acknowledgementReceiveTime m+ if t' > t || (t' == t && m' > m)+ then lookupAcknowledgementMessageIndex' q m left (index - 1)+ else if t' < t || (t' == t && m' < m)+ then lookupAcknowledgementMessageIndex' q m (index + 1) right+ else return index + +-- | Search for the message index.+lookupAcknowledgementMessageIndex :: AcknowledgementMessageQueue -> AcknowledgementMessage -> IO Int+lookupAcknowledgementMessageIndex q m =+ do n <- vectorCount (acknowledgementMessages q)+ lookupAcknowledgementMessageIndex' q m 0 (n - 1)++-- | Reduce the acknowledgement messages till the specified time.+reduceAcknowledgementMessages :: AcknowledgementMessageQueue -> Double -> IO ()+reduceAcknowledgementMessages q t =+ do count <- vectorCount (acknowledgementMessages q)+ len <- loop count 0+ when (len > 0) $+ vectorDeleteRange (acknowledgementMessages q) 0 len+ where+ loop n i+ | i >= n = return i+ | otherwise = do m <- readVector (acknowledgementMessages q) i+ if acknowledgementReceiveTime m < t+ then loop n (i + 1)+ else return i++-- | Filter the acknowledgement messages using the specified predicate.+filterAcknowledgementMessages :: (AcknowledgementMessage -> Bool) -> AcknowledgementMessageQueue -> IO [AcknowledgementMessage]+filterAcknowledgementMessages pred q =+ do count <- vectorCount (acknowledgementMessages q)+ loop count 0 []+ where+ loop n i acc+ | i >= n = return (reverse acc)+ | otherwise = do m <- readVector (acknowledgementMessages q) i+ if pred m+ then loop n (i + 1) (m : acc)+ else loop n (i + 1) acc
+ Simulation/Aivika/Distributed/Optimistic/Internal/AcknowledgementMessageQueue.hs-boot view
@@ -0,0 +1,34 @@++-- |+-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.AcknowledgementMessageQueue+-- Copyright : Copyright (c) 2015-2018, David Sorokin <david.sorokin@gmail.com>+-- License : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability : experimental+-- Tested with: GHC 7.10.3+--+-- This is an hs-boot file.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.AcknowledgementMessageQueue+ (AcknowledgementMessageQueue,+ newAcknowledgementMessageQueue,+ acknowledgementMessageQueueSize,+ enqueueAcknowledgementMessage,+ reduceAcknowledgementMessages,+ filterAcknowledgementMessages) where++import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.IO++data AcknowledgementMessageQueue++newAcknowledgementMessageQueue :: DIO AcknowledgementMessageQueue++acknowledgementMessageQueueSize :: AcknowledgementMessageQueue -> IO Int++enqueueAcknowledgementMessage :: AcknowledgementMessageQueue -> AcknowledgementMessage -> IO ()++reduceAcknowledgementMessages :: AcknowledgementMessageQueue -> Double -> IO ()++filterAcknowledgementMessages :: (AcknowledgementMessage -> Bool) -> AcknowledgementMessageQueue -> IO [AcknowledgementMessage]
+ Simulation/Aivika/Distributed/Optimistic/Internal/ConnectionManager.hs view
@@ -0,0 +1,300 @@++-- |+-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.ConnectionManager+-- Copyright : Copyright (c) 2015-2018, David Sorokin <david.sorokin@gmail.com>+-- License : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability : experimental+-- Tested with: GHC 7.10.3+--+-- This module is responsible for managing the connections.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.ConnectionManager+ (ConnectionManager,+ ConnectionParams(..),+ newConnectionManager,+ tryAddMessageReceiver,+ addMessageReceiver,+ removeMessageReceiver,+ clearMessageReceivers,+ reconnectMessageReceivers,+ filterMessageReceivers,+ existsMessageReceiver,+ trySendKeepAlive,+ trySendKeepAliveUTC) where++import qualified Data.Map as M+import qualified Data.Set as S+import Data.Maybe+import Data.Either+import Data.IORef+import Data.Time.Clock+import Data.Word++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP++import Simulation.Aivika.Distributed.Optimistic.Internal.Priority+import Simulation.Aivika.Distributed.Optimistic.Internal.Message++-- | The connection parameters.+data ConnectionParams =+ ConnectionParams { connLoggingPriority :: Priority,+ -- ^ the logging priority+ connKeepAliveInterval :: Int,+ -- ^ the interval in microseconds to send keep-alive messages+ connReconnectingDelay :: Int,+ -- ^ the reconnecting delay in microseconds+ connMonitoringDelay :: Int+ -- ^ the monitoring delay in microseconds+ }++-- | The connection manager.+data ConnectionManager =+ ConnectionManager { connParams :: ConnectionParams,+ -- ^ the manager parameter+ connKeepAliveTimestamp :: IORef UTCTime,+ -- ^ the keep alive timestamp+ connReceivers :: IORef (M.Map DP.ProcessId ConnectionMessageReceiver)+ -- ^ the receivers of messages+ }++-- | The connection message receiver.+data ConnectionMessageReceiver =+ ConnectionMessageReceiver { connReceiverProcess :: DP.ProcessId,+ -- ^ the receiver of messages+ connReceiverNodeMonitor :: IORef (Maybe DP.MonitorRef),+ -- ^ a monitor of the message receiver node+ connReceiverMonitor :: IORef (Maybe DP.MonitorRef)+ -- ^ a monitor of the message receiver+ }++-- | Create a new connection manager.+newConnectionManager :: ConnectionParams -> IO ConnectionManager+newConnectionManager ps =+ do timestamp <- getCurrentTime >>= newIORef+ receivers <- newIORef M.empty+ return ConnectionManager { connParams = ps,+ connKeepAliveTimestamp = timestamp,+ connReceivers = receivers }++-- | Try to add the connection message receiver.+tryAddMessageReceiver :: ConnectionManager -> DP.ProcessId -> DP.Process Bool+tryAddMessageReceiver manager pid =+ do f <- liftIO $+ existsMessageReceiver manager pid+ if f+ then return False+ else do addMessageReceiver manager pid+ return True++-- | Add the connection message receiver.+addMessageReceiver :: ConnectionManager -> DP.ProcessId -> DP.Process ()+addMessageReceiver manager pid =+ do r1 <- liftIO $ newIORef Nothing+ r2 <- liftIO $ newIORef Nothing+ let x = ConnectionMessageReceiver { connReceiverProcess = pid,+ connReceiverNodeMonitor = r1,+ connReceiverMonitor = r2 }+ monitorMessageReceiver manager x+ liftIO $+ modifyIORef (connReceivers manager) $+ M.insert pid x++-- | Remove the connection message receiver.+removeMessageReceiver :: ConnectionManager -> DP.ProcessId -> DP.Process ()+removeMessageReceiver manager pid =+ do rs <- liftIO $ readIORef (connReceivers manager)+ case M.lookup pid rs of+ Nothing ->+ logConnectionManager manager WARNING $ "Could not find the monitored process " ++ show pid+ Just r ->+ do unmonitorMessageReceiver manager r+ liftIO $+ modifyIORef (connReceivers manager) $+ M.delete pid++-- | Clear the connection message receivers.+clearMessageReceivers :: ConnectionManager -> DP.Process ()+clearMessageReceivers manager =+ do rs <- liftIO $ readIORef (connReceivers manager)+ forM_ (M.elems rs) $ \r ->+ do let pid = connReceiverProcess r+ removeMessageReceiver manager pid++-- | Reconnect to the message receivers.+reconnectMessageReceivers :: ConnectionManager -> [DP.ProcessId] -> DP.Process ()+reconnectMessageReceivers manager pids =+ do rs <- messageReceivers manager pids+ unless (null rs) $+ do forM_ rs $+ unmonitorMessageReceiver manager+ liftIO $+ threadDelay (connReconnectingDelay $ connParams manager)+ logConnectionManager manager NOTICE "Begin reconnecting..."+ forM_ rs $+ reconnectToMessageReceiver manager+ liftIO $+ threadDelay (connMonitoringDelay $ connParams manager)+ logConnectionManager manager NOTICE "Begin remonitoring..."+ forM_ rs $+ monitorMessageReceiver manager++-- | Unmonitor the message receiver.+unmonitorMessageReceiver :: ConnectionManager -> ConnectionMessageReceiver -> DP.Process ()+unmonitorMessageReceiver manager r =+ do unmonitorMessageReceiverProcess manager r+ unmonitorMessageReceiverNode manager r++-- | Monitor the message receiver.+monitorMessageReceiver :: ConnectionManager -> ConnectionMessageReceiver -> DP.Process ()+monitorMessageReceiver manager r =+ do monitorMessageReceiverNode manager r+ monitorMessageReceiverProcess manager r++-- | Unmonitor the message receiver process.+unmonitorMessageReceiverProcess :: ConnectionManager -> ConnectionMessageReceiver -> DP.Process ()+unmonitorMessageReceiverProcess manager r =+ do let pid = connReceiverProcess r+ ref <- liftIO $ readIORef (connReceiverMonitor r)+ case ref of+ Just m ->+ do logConnectionManager manager NOTICE $ "Unmonitoring process " ++ show pid+ DP.unmonitor m+ liftIO $ writeIORef (connReceiverMonitor r) Nothing+ Nothing ->+ logConnectionManager manager WARNING $ "Could not find the monitor reference for process " ++ show pid++-- | Monitor the message receiver process.+monitorMessageReceiverProcess :: ConnectionManager -> ConnectionMessageReceiver -> DP.Process ()+monitorMessageReceiverProcess manager r =+ do let pid = connReceiverProcess r+ ref <- liftIO $ readIORef (connReceiverMonitor r)+ case ref of+ Nothing ->+ do logConnectionManager manager NOTICE $ "Monitoring process " ++ show pid+ x <- DP.monitor pid+ liftIO $ writeIORef (connReceiverMonitor r) (Just x)+ Just x0 ->+ do logConnectionManager manager WARNING $ "Re-monitoring process " ++ show pid+ x <- DP.monitor pid+ DP.unmonitor x0+ liftIO $ writeIORef (connReceiverMonitor r) (Just x)++-- | Unmonitor the message receiver node.+unmonitorMessageReceiverNode :: ConnectionManager -> ConnectionMessageReceiver -> DP.Process ()+unmonitorMessageReceiverNode manager r =+ do let nid = DP.processNodeId $ connReceiverProcess r+ ref <- liftIO $ readIORef (connReceiverNodeMonitor r)+ case ref of+ Just m ->+ do logConnectionManager manager NOTICE $ "Unmonitoring node " ++ show nid+ DP.unmonitor m+ liftIO $ writeIORef (connReceiverNodeMonitor r) Nothing+ Nothing ->+ logConnectionManager manager WARNING $ "Could not find the monitor reference for node " ++ show nid++-- | Monitor the message receiver node.+monitorMessageReceiverNode :: ConnectionManager -> ConnectionMessageReceiver -> DP.Process ()+monitorMessageReceiverNode manager r =+ do let nid = DP.processNodeId $ connReceiverProcess r+ ref <- liftIO $ readIORef (connReceiverNodeMonitor r)+ case ref of+ Nothing ->+ do logConnectionManager manager NOTICE $ "Monitoring node " ++ show nid+ x <- DP.monitorNode nid+ liftIO $ writeIORef (connReceiverNodeMonitor r) (Just x)+ Just x0 ->+ do logConnectionManager manager WARNING $ "Re-monitoring node " ++ show nid+ x <- DP.monitorNode nid+ DP.unmonitor x0+ liftIO $ writeIORef (connReceiverNodeMonitor r) (Just x)++-- | Reconnect to the message receiver.+reconnectToMessageReceiver :: ConnectionManager -> ConnectionMessageReceiver -> DP.Process ()+reconnectToMessageReceiver manager r =+ do let pid = connReceiverProcess r+ logConnectionManager manager NOTICE $ "Direct reconnecting to " ++ show pid+ DP.reconnect pid++-- | Whether the connection message receiver exists.+existsMessageReceiver :: ConnectionManager -> DP.ProcessId -> IO Bool+existsMessageReceiver manager pid =+ readIORef (connReceivers manager) >>=+ return . M.member pid++-- | Try to send keep-alive messages.+trySendKeepAlive :: ConnectionManager -> DP.Process ()+trySendKeepAlive manager =+ do empty <- liftIO $ fmap M.null $ readIORef (connReceivers manager)+ unless empty $ + do utc <- liftIO getCurrentTime+ trySendKeepAliveUTC manager utc++-- | Try to send keep-alive messages by the specified current time.+trySendKeepAliveUTC :: ConnectionManager -> UTCTime -> DP.Process ()+trySendKeepAliveUTC manager utc =+ do empty <- liftIO $ fmap M.null $ readIORef (connReceivers manager)+ unless empty $ + do f <- liftIO $ shouldSendKeepAlive manager utc+ when f $+ do ---+ logConnectionManager manager INFO $+ "Sending keep-alive messages"+ ---+ liftIO $ writeIORef (connKeepAliveTimestamp manager) utc+ rs <- liftIO $ readIORef (connReceivers manager)+ forM_ rs $ \r ->+ do let pid = connReceiverProcess r+ DP.usend pid KeepAliveMessage++-- | Whether should send a keep-alive message.+shouldSendKeepAlive :: ConnectionManager -> UTCTime -> IO Bool+shouldSendKeepAlive manager utc =+ do utc0 <- readIORef (connKeepAliveTimestamp manager)+ let dt = fromRational $ toRational (diffUTCTime utc utc0)+ return $+ secondsToMicroseconds dt > (connKeepAliveInterval $ connParams manager)++-- | Convert seconds to microseconds.+secondsToMicroseconds :: Double -> Int+secondsToMicroseconds x = fromInteger $ toInteger $ round (1000000 * x)++-- | Get the connection message receivers.+messageReceivers :: ConnectionManager -> [DP.ProcessId] -> DP.Process [ConnectionMessageReceiver]+messageReceivers manager pids =+ do rs <- liftIO $ readIORef (connReceivers manager)+ fmap mconcat $+ forM pids $ \pid ->+ case M.lookup pid rs of+ Just x -> return [x]+ Nothing ->+ do logConnectionManager manager WARNING $ "Could not find the monitored process " ++ show pid+ return []++-- | Filter the message receivers.+filterMessageReceivers :: ConnectionManager -> [DP.ProcessMonitorNotification] -> DP.Process [DP.ProcessId]+filterMessageReceivers manager ms =+ do rs <- liftIO $ readIORef (connReceivers manager)+ fmap (S.toList . S.fromList . mconcat) $+ forM ms $ \(DP.ProcessMonitorNotification ref pid _) ->+ case M.lookup pid rs of+ Nothing ->+ do logConnectionManager manager WARNING $ "Could not find the monitored process " ++ show pid+ return []+ Just x ->+ do ref0 <- liftIO $ readIORef (connReceiverMonitor x)+ if ref0 == Just ref+ then return [pid]+ else do logConnectionManager manager NOTICE $ "Received the old monitor reference for process " ++ show pid+ return []++-- | Log the message with the specified priority.+logConnectionManager :: ConnectionManager -> Priority -> String -> DP.Process ()+{-# INLINE logConnectionManager #-}+logConnectionManager manager p message =+ when (connLoggingPriority (connParams manager) <= p) $+ DP.say $+ embracePriority p ++ " " ++ message
Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs view
@@ -63,7 +63,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+import Simulation.Aivika.Distributed.Optimistic.Internal.ConnectionManager import Simulation.Aivika.Distributed.Optimistic.State -- | The parameters for the 'DIO' computation.@@ -94,7 +94,7 @@ 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+ -- ^ The delay in microseconds before reconnecting to the remote process dioKeepAliveInterval :: Int, -- ^ The interval in microseconds for sending keep-alive messages dioTimeServerAcknowledgementTimeout :: Int,@@ -308,8 +308,8 @@ -- ^ the process monitor notification | InternalInboxProcessMessage InboxProcessMessage -- ^ the inbox process message- | InternalKeepAliveMessage KeepAliveMessage- -- ^ the keep alive message+ | InternalGeneralMessage GeneralMessage+ -- ^ the general message -- | Handle the specified exception. handleException :: DIOParams -> SomeException -> DP.Process ()@@ -328,10 +328,12 @@ runDIOWithEnv :: DIO a -> DIOParams -> DIOEnv -> DP.ProcessId -> DP.Process (DP.ProcessId, DP.Process a) runDIOWithEnv m ps env serverId = do ch <- liftIO newChannel- let keepAliveParams =- KeepAliveParams { keepAliveLoggingPriority = dioLoggingPriority ps,- keepAliveInterval = dioKeepAliveInterval ps }- keepAliveManager <- liftIO $ newKeepAliveManager keepAliveParams+ let connParams =+ ConnectionParams { connLoggingPriority = dioLoggingPriority ps,+ connKeepAliveInterval = dioKeepAliveInterval ps,+ connReconnectingDelay = dioProcessReconnectingDelay ps,+ connMonitoringDelay = dioProcessMonitoringDelay ps }+ connManager <- liftIO $ newConnectionManager connParams terminated <- liftIO $ newIORef False registeredInTimeServer <- liftIO $ newTVarIO False unregisteredFromTimeServer <- liftIO $ newTVarIO False@@ -345,48 +347,43 @@ f2 x = return (InternalProcessMonitorNotification x) f3 :: InboxProcessMessage -> DP.Process InternalLogicalProcessMessage f3 x = return (InternalInboxProcessMessage x)- f4 :: KeepAliveMessage -> DP.Process InternalLogicalProcessMessage- f4 x = return (InternalKeepAliveMessage x)+ f4 :: GeneralMessage -> DP.Process InternalLogicalProcessMessage+ f4 x = return (InternalGeneralMessage x) x <- fmap Just $ DP.receiveWait [DP.match f1, DP.match f2, DP.match f3, DP.match f4] case x of Nothing -> return () Just (InternalLogicalProcessMessage m) ->- do processTimeServerMessage m serverId timeServerTimestamp+ do processTimeServerMessage m ps serverId timeServerTimestamp liftIO $ writeChannel ch m Just (InternalProcessMonitorNotification m@(DP.ProcessMonitorNotification _ _ _)) ->- handleProcessMonitorNotification m ps ch serverId+ handleProcessMonitorNotification m ps ch connManager serverId Just (InternalInboxProcessMessage m) -> case m of SendQueueMessage pid m ->- DP.send pid (QueueMessage m)+ DP.usend pid (QueueMessage m) SendQueueMessageBulk pid ms -> forM_ ms $ \m ->- DP.send pid (QueueMessage m)+ DP.usend pid (QueueMessage m) SendAcknowledgementQueueMessage pid m ->- DP.send pid (AcknowledgementQueueMessage m)+ DP.usend pid (AcknowledgementQueueMessage m) SendAcknowledgementQueueMessageBulk pid ms -> forM_ ms $ \m ->- DP.send pid (AcknowledgementQueueMessage m)+ DP.usend pid (AcknowledgementQueueMessage m) SendLocalTimeMessage receiver sender t ->- DP.send receiver (LocalTimeMessage sender t)+ DP.usend receiver (LocalTimeMessage sender t) SendRequestGlobalTimeMessage receiver sender ->- DP.send receiver (RequestGlobalTimeMessage sender)+ DP.usend receiver (RequestGlobalTimeMessage sender) SendRegisterLogicalProcessMessage receiver sender ->- DP.send receiver (RegisterLogicalProcessMessage sender)+ DP.usend receiver (RegisterLogicalProcessMessage sender) SendUnregisterLogicalProcessMessage receiver sender ->- DP.send receiver (UnregisterLogicalProcessMessage sender)+ DP.usend receiver (UnregisterLogicalProcessMessage sender) SendTerminateTimeServerMessage receiver sender ->- DP.send receiver (TerminateTimeServerMessage sender)+ DP.usend receiver (TerminateTimeServerMessage sender) MonitorProcessMessage pid ->- do f <- liftIO $- existsKeepAliveReceiver keepAliveManager pid- unless f $- addKeepAliveReceiver keepAliveManager pid- ReMonitorProcessMessage pids ->- handleReMonitorProcessMessage pids ps ch keepAliveManager- TrySendKeepAliveMessage ->- trySendKeepAlive keepAliveManager+ tryAddMessageReceiver connManager pid >> return ()+ TrySendProcessKeepAliveMessage ->+ trySendKeepAlive connManager RegisterLogicalProcessAcknowledgementMessage pid -> do --- logProcess ps INFO "Registered the logical process in the time server"@@ -416,16 +413,14 @@ do atomicWriteIORef terminated True writeChannel ch AbortSimulationMessage DP.terminate- Just (InternalKeepAliveMessage m) ->- do ---- logProcess ps DEBUG $ "Received " ++ show m- ---- return ()+ Just (InternalGeneralMessage m) ->+ handleGeneralMessage m ps ch connManager loop = C.finally loop0- (liftIO $- do atomicWriteIORef terminated True- writeChannel ch AbortSimulationMessage)+ (do liftIO $+ do atomicWriteIORef terminated True+ writeChannel ch AbortSimulationMessage+ clearMessageReceivers connManager) inboxId <- DP.spawnLocal $ C.catch loop (handleException ps)@@ -435,7 +430,7 @@ unless f $ do liftIO $ threadDelay (dioKeepAliveInterval ps)- DP.send inboxId TrySendKeepAliveMessage+ DP.send inboxId TrySendProcessKeepAliveMessage loop in C.catch loop (handleException ps) DP.spawnLocal $@@ -487,8 +482,13 @@ return (inboxId, simulation) -- | Handle the process monitor notification.-handleProcessMonitorNotification :: DP.ProcessMonitorNotification -> DIOParams -> Channel LogicalProcessMessage -> DP.ProcessId -> DP.Process ()-handleProcessMonitorNotification m@(DP.ProcessMonitorNotification _ pid0 reason) ps ch serverId =+handleProcessMonitorNotification :: DP.ProcessMonitorNotification+ -> DIOParams+ -> Channel LogicalProcessMessage+ -> ConnectionManager+ -> DP.ProcessId+ -> DP.Process ()+handleProcessMonitorNotification m@(DP.ProcessMonitorNotification _ pid0 reason) ps ch connManager serverId = do let recv m@(DP.ProcessMonitorNotification _ _ _) = do --- logProcess ps WARNING $ "Received a process monitor notification " ++ show m@@ -507,43 +507,35 @@ do liftIO $ threadDelay (dioProcessReconnectingDelay ps) let pred m@(DP.ProcessMonitorNotification _ _ reason) = reason == DP.DiedDisconnect- loop :: [DP.ProcessId] -> DP.Process [DP.ProcessId]+ loop :: [DP.ProcessMonitorNotification] -> DP.Process [DP.ProcessMonitorNotification] 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..."- ---+ Just m@(DP.ProcessMonitorNotification _ _ _) -> loop (m : acc)+ ms <- loop [m]+ pids <- filterMessageReceivers connManager ms+ reconnectMessageReceivers connManager pids forM_ pids $ \pid -> do ---- logProcess ps NOTICE $ "Direct reconnecting to " ++ show pid+ logProcess ps NOTICE $+ "Writing to the channel about reconnecting to " ++ show pid ---- 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 ()+ liftIO $+ writeChannel ch (ReconnectProcessMessage pid) --- | Re-monitor the specified remote processes.-handleReMonitorProcessMessage :: [DP.ProcessId] -> DIOParams -> Channel LogicalProcessMessage -> KeepAliveManager -> DP.Process ()-handleReMonitorProcessMessage pids ps ch keepAliveManager =- forM_ pids $ \pid ->- do remonitorKeepAliveReceiver keepAliveManager pid- ---- logProcess ps NOTICE $ "Writing to the channel about reconnecting to " ++ show pid+-- | Handle the general message.+handleGeneralMessage :: GeneralMessage+ -> DIOParams+ -> Channel LogicalProcessMessage+ -> ConnectionManager+ -> DP.Process ()+handleGeneralMessage m@KeepAliveMessage ps ch connManager =+ do ---+ logProcess ps DEBUG $+ "Received " ++ show m ---- liftIO $- writeChannel ch (ReconnectProcessMessage pid)+ return () -- | Monitor the specified process. monitorProcessDIO :: DP.ProcessId -> DIO ()@@ -558,16 +550,18 @@ logProcess ps WARNING "Ignored the process monitoring as it was disabled in the DIO computation parameters" -- | Process the time server message in a stream of messages destined for the logical process.-processTimeServerMessage :: LogicalProcessMessage -> DP.ProcessId -> IORef UTCTime -> DP.Process ()-processTimeServerMessage ComputeLocalTimeMessage serverId r =+processTimeServerMessage :: LogicalProcessMessage -> DIOParams -> DP.ProcessId -> IORef UTCTime -> DP.Process ()+processTimeServerMessage ComputeLocalTimeMessage ps serverId r = do liftIO $ getCurrentTime >>= writeIORef r inboxId <- DP.getSelfPid- DP.send serverId (ComputeLocalTimeAcknowledgementMessage inboxId)-processTimeServerMessage (GlobalTimeMessage _) serverId r =+ if dioProcessMonitoringEnabled ps+ then DP.usend serverId (ComputeLocalTimeAcknowledgementMessage inboxId)+ else DP.send serverId (ComputeLocalTimeAcknowledgementMessage inboxId)+processTimeServerMessage (GlobalTimeMessage _) ps serverId r = liftIO $ getCurrentTime >>= writeIORef r-processTimeServerMessage _ serverId r =+processTimeServerMessage _ ps serverId r = return () -- | Validate the time server by the specified inbox and recent timestamp.
Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs view
@@ -49,6 +49,7 @@ import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue import Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog+import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.AcknowledgementMessageQueue import {-# SOURCE #-} qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict as R import Simulation.Aivika.Distributed.Optimistic.State @@ -67,6 +68,8 @@ -- ^ the output message queue queueTransientMessages :: TransientMessageQueue, -- ^ the transient message queue+ queueAcknowledgementMessages :: AcknowledgementMessageQueue,+ -- ^ the acknowledgement message queue queueLog :: UndoableLog, -- ^ the undoable log of operations queuePQ :: R.Ref (PQ.PriorityQueue (Point DIO -> DIO ())),@@ -92,11 +95,13 @@ transient <- newTransientMessageQueue output <- newOutputMessageQueue $ enqueueTransientMessage transient input <- newInputMessageQueue log rollbackEventPre rollbackEventPost rollbackEventTime+ ack <- newAcknowledgementMessageQueue infind <- liftIOUnsafe $ newIORef False s <- newDIOSignalSource0 return EventQueue { queueInputMessages = input, queueOutputMessages = output, queueTransientMessages = transient,+ queueAcknowledgementMessages = ack, queueLog = log, queuePQ = pq, queueBusy = f,@@ -362,8 +367,13 @@ if f then invokeEvent p' logOutdatedMessage else error "Received the outdated message: processChannelMessage"- else invokeTimeWarp p' $- enqueueMessage (queueInputMessages q) m+ else do ps <- dioParams+ when (dioProcessReconnectingEnabled ps) $+ liftIOUnsafe $+ enqueueAcknowledgementMessage (queueAcknowledgementMessages q) $+ acknowledgementMessage infind m+ invokeTimeWarp p' $+ enqueueMessage (queueInputMessages q) m processChannelMessage x@(QueueMessageBulk ms) = TimeWarp $ \p -> do let q = runEventQueue $ pointRun p@@ -381,8 +391,13 @@ if f then invokeEvent p' logOutdatedMessage else error "Received the outdated message: processChannelMessage"- else invokeTimeWarp p' $- enqueueMessage (queueInputMessages q) m+ else do ps <- dioParams+ when (dioProcessReconnectingEnabled ps) $+ liftIOUnsafe $+ enqueueAcknowledgementMessage (queueAcknowledgementMessages q) $+ acknowledgementMessage infind m+ invokeTimeWarp p' $+ enqueueMessage (queueInputMessages q) m processChannelMessage x@(AcknowledgementQueueMessage m) = TimeWarp $ \p -> do let q = runEventQueue $ pointRun p@@ -600,10 +615,13 @@ reduceEvents t = Event $ \p -> do let q = runEventQueue $ pointRun p+ ps <- dioParams liftIOUnsafe $ do reduceInputMessages (queueInputMessages q) t reduceOutputMessages (queueOutputMessages q) t reduceLog (queueLog q) t+ when (dioProcessReconnectingEnabled ps) $+ reduceAcknowledgementMessages (queueAcknowledgementMessages q) t instance {-# OVERLAPPING #-} MonadIO (Event DIO) where @@ -775,11 +793,9 @@ "t = " ++ show (pointTime p) ++ ": reconnecting to " ++ show pid ++ "..." ---- infind <- liftIOUnsafe $ readIORef (queueInFind q)- let ys = queueInputMessages q+ let ys = queueAcknowledgementMessages q ys' <- liftIOUnsafe $- fmap (map $ acknowledgementMessage infind) $- filterInputMessages (\x -> messageSenderId x == pid) ys+ filterAcknowledgementMessages (\x -> acknowledgementSenderId x == pid) ys unless (null ys') $ sendAcknowledgementMessagesDIO pid ys' xs <- liftIOUnsafe $ transientMessages (queueTransientMessages q)
− Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs
@@ -1,148 +0,0 @@---- |--- 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,- remonitorKeepAliveReceiver,- existsKeepAliveReceiver,- trySendKeepAlive,- trySendKeepAliveUTC) where--import qualified Data.Map as M-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 (M.Map DP.ProcessId KeepAliveReceiver)- -- ^ the receivers of the keep-alive messages- }--data KeepAliveReceiver =- KeepAliveReceiver { keepAliveReceiverProcess :: DP.ProcessId,- -- ^ the receiver of keep-alive messages- keepAliveReceiverMonitor :: IORef DP.MonitorRef- -- ^ a monitor of the keep-alive receiver- }---- | Create a new keep-alive manager.-newKeepAliveManager :: KeepAliveParams -> IO KeepAliveManager-newKeepAliveManager ps =- do timestamp <- getCurrentTime >>= newIORef- receivers <- newIORef M.empty- return KeepAliveManager { keepAliveParams = ps,- keepAliveTimestamp = timestamp,- keepAliveReceivers = receivers }---- | Add the keep-alive message receiver.-addKeepAliveReceiver :: KeepAliveManager -> DP.ProcessId -> DP.Process ()-addKeepAliveReceiver manager pid =- do ---- logKeepAliveManager manager INFO $ "Monitoring process " ++ show pid- ---- r <- DP.monitor pid- r2 <- liftIO $ newIORef r- let x = KeepAliveReceiver pid r2- liftIO $- modifyIORef (keepAliveReceivers manager) $- M.insert pid x---- | Remonitor the keep-alive message receiver.-remonitorKeepAliveReceiver :: KeepAliveManager -> DP.ProcessId -> DP.Process ()-remonitorKeepAliveReceiver manager pid =- do rs <- liftIO $ readIORef (keepAliveReceivers manager)- case M.lookup pid rs of- Nothing ->- do ---- logKeepAliveManager manager WARNING $ "Could not find the monitored process " ++ show pid- ---- Just x ->- do ---- logKeepAliveManager manager NOTICE $ "Re-monitoring process " ++ show pid- ---- r <- liftIO $ readIORef (keepAliveReceiverMonitor x) - r' <- DP.monitor pid- DP.unmonitor r- liftIO $ writeIORef (keepAliveReceiverMonitor x) r'---- | Whether the keep-alive message receiver exists.-existsKeepAliveReceiver :: KeepAliveManager -> DP.ProcessId -> IO Bool-existsKeepAliveReceiver manager pid =- readIORef (keepAliveReceivers manager) >>=- return . M.member pid---- | Try to send a keep-alive message.-trySendKeepAlive :: KeepAliveManager -> DP.Process ()-trySendKeepAlive manager =- do empty <- liftIO $ fmap M.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 M.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- rs <- liftIO $ readIORef (keepAliveReceivers manager)- forM_ rs $ \r ->- let pid = keepAliveReceiverProcess r- in 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
Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs view
@@ -20,12 +20,13 @@ LogicalProcessMessage(..), TimeServerMessage(..), InboxProcessMessage(..),- KeepAliveMessage(..)) where+ GeneralMessage(..)) where import GHC.Generics import Data.Typeable import Data.Binary+import Data.Word import qualified Control.Distributed.Process as DP import Control.Distributed.Process.Serializable@@ -34,7 +35,7 @@ -- | Represents a message. data Message =- Message { messageSequenceNo :: Int,+ Message { messageSequenceNo :: Word64, -- ^ The sequence number. messageSendTime :: Double, -- ^ The send time.@@ -78,7 +79,7 @@ -- | Represents an acknowledgement message. data AcknowledgementMessage =- AcknowledgementMessage { acknowledgementSequenceNo :: Int,+ AcknowledgementMessage { acknowledgementSequenceNo :: Word64, -- ^ The sequence number. acknowledgementSendTime :: Double, -- ^ The send time.@@ -148,8 +149,6 @@ -- ^ the logical process sent its local minimum time | ProvideTimeServerStateMessage DP.ProcessId -- ^ send the time server monitoring state message- | ReMonitorTimeServerMessage [DP.ProcessId]- -- ^ re-monitor the logical processes by their identifiers deriving (Eq, Show, Typeable, Generic) instance Binary TimeServerMessage@@ -157,9 +156,7 @@ -- | The message destined directly for the inbox process. data InboxProcessMessage = MonitorProcessMessage DP.ProcessId -- ^ monitor the logical process by its inbox process identifier- | ReMonitorProcessMessage [DP.ProcessId]- -- ^ re-monitor the logical processes by their identifiers- | TrySendKeepAliveMessage+ | TrySendProcessKeepAliveMessage -- ^ try to send a keep alive message | SendQueueMessage DP.ProcessId Message -- ^ send a queue message via the inbox process@@ -191,9 +188,9 @@ instance Binary InboxProcessMessage --- | The keep-alive message type.-data KeepAliveMessage = KeepAliveMessage- -- ^ the keel-alive message.- deriving (Eq, Show, Typeable, Generic)+-- | The general message type.+data GeneralMessage = KeepAliveMessage+ -- ^ the keel-alive message.+ deriving (Eq, Show, Typeable, Generic) -instance Binary KeepAliveMessage+instance Binary GeneralMessage
Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs view
@@ -20,6 +20,7 @@ import Data.List import Data.IORef+import Data.Word import Control.Monad import Control.Monad.Trans@@ -42,7 +43,7 @@ -- ^ Enqueue the transient message. outputMessages :: DLL.DoubleLinkedList Message, -- ^ The output messages.- outputMessageSequenceNo :: IORef Int+ outputMessageSequenceNo :: IORef Word64 -- ^ The next sequence number. } @@ -114,7 +115,7 @@ loop -- | Generate a next message sequence number.-generateMessageSequenceNo :: OutputMessageQueue -> IO Int+generateMessageSequenceNo :: OutputMessageQueue -> IO Word64 generateMessageSequenceNo q = atomicModifyIORef (outputMessageSequenceNo q) $ \n -> let n' = n + 1 in n' `seq` (n', n)
Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs view
@@ -40,6 +40,7 @@ import Simulation.Aivika.Distributed.Optimistic.Internal.Priority import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.ConnectionManager import Simulation.Aivika.Distributed.Optimistic.State -- | The time server parameters.@@ -115,8 +116,10 @@ -- ^ the global time of the model tsGlobalTimeTimestamp :: IORef (Maybe UTCTime), -- ^ the global time timestamp- tsLogicalProcessValidationTimestamp :: IORef UTCTime+ tsLogicalProcessValidationTimestamp :: IORef UTCTime, -- ^ the logical process validation timestamp+ tsConnectionManager :: ConnectionManager+ -- ^ the connection manager } -- | The information about the logical process.@@ -125,10 +128,8 @@ -- ^ the logical process identifier lpLocalTime :: IORef (Maybe Double), -- ^ the local time of the process- lpTimestamp :: IORef UTCTime,+ lpTimestamp :: IORef UTCTime -- ^ the logical process timestamp- lpMonitorRef :: Maybe DP.MonitorRef- -- ^ the logical process monitor reference } -- | The default time server parameters.@@ -164,6 +165,11 @@ t0 <- newIORef Nothing t' <- newIORef Nothing t2 <- getCurrentTime >>= newIORef+ connManager <- newConnectionManager $+ ConnectionParams { connLoggingPriority = tsLoggingPriority ps,+ connKeepAliveInterval = 0, -- not used+ connReconnectingDelay = tsProcessReconnectingDelay ps,+ connMonitoringDelay = tsProcessMonitoringDelay ps } return TimeServer { tsParams = ps, tsInitQuorum = n, tsInInit = f,@@ -173,7 +179,8 @@ tsProcessesInFind = s, tsGlobalTime = t0, tsGlobalTimeTimestamp = t',- tsLogicalProcessValidationTimestamp = t2+ tsLogicalProcessValidationTimestamp = t2,+ tsConnectionManager = connManager } -- | Process the time server message.@@ -190,17 +197,15 @@ do t <- newIORef Nothing utc <- getCurrentTime >>= newIORef modifyIORef (tsProcesses server) $- M.insert pid LogicalProcessInfo { lpId = pid, lpLocalTime = t, lpTimestamp = utc, lpMonitorRef = Nothing }+ M.insert pid LogicalProcessInfo { lpId = pid, lpLocalTime = t, lpTimestamp = utc } return $ 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+ do tryAddMessageReceiver (tsConnectionManager server) pid+ return () serverId <- DP.getSelfPid- DP.send pid (RegisterLogicalProcessAcknowledgementMessage serverId)+ if tsProcessMonitoringEnabled (tsParams server)+ then DP.usend pid (RegisterLogicalProcessAcknowledgementMessage serverId)+ else DP.send pid (RegisterLogicalProcessAcknowledgementMessage serverId) tryStartTimeServer server processTimeServerMessage server (UnregisterLogicalProcessMessage pid) = join $ liftIO $@@ -217,14 +222,11 @@ 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+ removeMessageReceiver (tsConnectionManager server) pid serverId <- DP.getSelfPid- DP.send pid (UnregisterLogicalProcessAcknowledgementMessage serverId)+ if tsProcessMonitoringEnabled (tsParams server)+ then DP.usend pid (UnregisterLogicalProcessAcknowledgementMessage serverId)+ else DP.send pid (UnregisterLogicalProcessAcknowledgementMessage serverId) tryProvideTimeServerGlobalTime server tryTerminateTimeServer server processTimeServerMessage server (TerminateTimeServerMessage pid) =@@ -242,14 +244,11 @@ 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+ removeMessageReceiver (tsConnectionManager server) pid serverId <- DP.getSelfPid- DP.send pid (TerminateTimeServerAcknowledgementMessage serverId)+ if tsProcessMonitoringEnabled (tsParams server)+ then DP.usend pid (TerminateTimeServerAcknowledgementMessage serverId)+ else DP.send pid (TerminateTimeServerAcknowledgementMessage serverId) startTerminatingTimeServer server processTimeServerMessage server (RequestGlobalTimeMessage pid) = tryComputeTimeServerGlobalTime server@@ -296,30 +295,9 @@ tsStateName = name, tsStateGlobalVirtualTime = t, tsStateLogicalProcesses = M.keys m }- DP.send pid msg-processTimeServerMessage server (ReMonitorTimeServerMessage pids) =- do forM_ pids $ \pid ->- do m <- liftIO $ readIORef (tsProcesses server)- case M.lookup pid m of- Nothing ->- logTimeServer server WARNING $- "Time Server: unknown process identifier " ++ show pid- Just x ->- case lpMonitorRef x of- Nothing ->- logTimeServer server WARNING $- "Time Server: there is no monitor reference to " ++ show pid- Just r ->- do logTimeServer server INFO $- "Time Server: re-monitoring the process by identifier " ++ show pid- r' <- DP.monitor pid- DP.unmonitor r- liftIO $- modifyIORef (tsProcesses server) $- M.update (\x -> Just x { lpMonitorRef = Just r' }) pid- logTimeServer server NOTICE $- "Time Server: started re-monitoring " ++ show pid- resetComputingTimeServerGlobalTime server+ if tsProcessMonitoringEnabled (tsParams server)+ then DP.usend pid msg+ else DP.send pid msg -- | Whether the both values are defined and the first is greater than or equaled to the second. (.>=.) :: Maybe Double -> Maybe Double -> Bool@@ -400,7 +378,9 @@ modifyIORef (tsProcessesInFind server) $ S.insert pid forM_ zs $ \(pid, x) ->- DP.send pid ComputeLocalTimeMessage+ if tsProcessMonitoringEnabled (tsParams server)+ then DP.usend pid ComputeLocalTimeMessage+ else DP.send pid ComputeLocalTimeMessage -- | Provide the logical processes with the global time. provideTimeServerGlobalTime :: TimeServer -> DP.Process ()@@ -420,7 +400,9 @@ liftIO $ writeIORef (tsGlobalTimeTimestamp server) (Just timestamp) zs <- liftIO $ fmap M.assocs $ readIORef (tsProcesses server) forM_ zs $ \(pid, x) ->- DP.send pid (GlobalTimeMessage t0)+ if tsProcessMonitoringEnabled (tsParams server)+ then DP.usend pid (GlobalTimeMessage t0)+ else DP.send pid (GlobalTimeMessage t0) -- | Return the time server global time. timeServerGlobalTime :: TimeServer -> IO (Maybe Double)@@ -488,8 +470,8 @@ -- ^ the time server message | InternalProcessMonitorNotification DP.ProcessMonitorNotification -- ^ the process monitor notification- | InternalKeepAliveMessage KeepAliveMessage- -- ^ the keep alive message+ | InternalGeneralMessage GeneralMessage+ -- ^ the general message -- | Handle the time server exception handleTimeServerException :: TimeServer -> SomeException -> DP.Process ()@@ -509,14 +491,15 @@ timeServerWithEnv :: Int -> TimeServerParams -> TimeServerEnv -> DP.Process () timeServerWithEnv n ps env = do server <- liftIO $ newTimeServer n ps+ serverId <- DP.getSelfPid logTimeServer server INFO "Time Server: starting..." let loop utc0 = 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)+ f3 :: GeneralMessage -> DP.Process InternalTimeServerMessage+ f3 x = return (InternalGeneralMessage x) a <- DP.receiveTimeout (tsReceiveTimeout ps) [DP.match f1, DP.match f2, DP.match f3] case a of Nothing -> return ()@@ -528,12 +511,8 @@ processTimeServerMessage server m Just (InternalProcessMonitorNotification m) -> handleProcessMonitorNotification m server- Just (InternalKeepAliveMessage m) ->- do ---- logTimeServer server DEBUG $- "Time Server: " ++ show m- ---- return ()+ Just (InternalGeneralMessage m) ->+ handleGeneralMessage m server utc <- liftIO getCurrentTime validation <- liftIO $ readIORef (tsLogicalProcessValidationTimestamp server) timestamp <- liftIO $ readIORef (tsGlobalTimeTimestamp server)@@ -550,13 +529,13 @@ loop' utc0 = C.finally (loop utc0)- (liftIO $- atomicWriteIORef (tsTerminated server) True)+ (do liftIO $+ atomicWriteIORef (tsTerminated server) True+ clearMessageReceivers (tsConnectionManager server)) case tsSimulationMonitoringAction env of Nothing -> return () Just act ->- do serverId <- DP.getSelfPid- monitorId <-+ do monitorId <- DP.spawnLocal $ let loop = do f <- liftIO $ readIORef (tsTerminated server)@@ -594,33 +573,27 @@ do liftIO $ threadDelay (tsProcessReconnectingDelay ps) let pred m@(DP.ProcessMonitorNotification _ _ reason) = reason == DP.DiedDisconnect- loop :: [DP.ProcessId] -> DP.Process [DP.ProcessId]+ loop :: [DP.ProcessMonitorNotification] -> DP.Process [DP.ProcessMonitorNotification] 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] >>= (liftIO . filterLogicalProcesses server)- ---- 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 ()+ Just m@(DP.ProcessMonitorNotification _ _ _) -> loop (m : acc)+ ms <- loop [m]+ pids <- filterMessageReceivers (tsConnectionManager server) ms >>=+ (liftIO . filterLogicalProcesses server)+ reconnectMessageReceivers (tsConnectionManager server) pids+ resetComputingTimeServerGlobalTime server+ tryComputeTimeServerGlobalTime server++-- | Handle the general message.+handleGeneralMessage :: GeneralMessage -> TimeServer -> DP.Process ()+handleGeneralMessage m@KeepAliveMessage server =+ do ---+ logTimeServer server DEBUG $+ "Time Server: " ++ show m+ ---+ return () -- | Test whether the sychronization delay has been exceeded. timeSyncDelayExceeded :: TimeServer -> UTCTime -> UTCTime -> Bool
Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs view
@@ -25,6 +25,7 @@ import qualified Data.Map as M import Data.List import Data.IORef+import Data.Word import Control.Monad import Control.Monad.Trans@@ -44,7 +45,7 @@ -- | Represents an acknowledgement message representative. data TransientMessageQueueItem =- TransientMessageQueueItem { itemSequenceNo :: Int,+ TransientMessageQueueItem { itemSequenceNo :: Word64, -- ^ The sequence number. itemSendTime :: Double, -- ^ The send time.
aivika-distributed.cabal view
@@ -1,5 +1,5 @@ name: aivika-distributed-version: 1.2+version: 1.3 synopsis: Parallel distributed discrete event simulation module for the Aivika library description: This package extends the aivika-transformers [1] package and allows running parallel distributed simulations.@@ -94,13 +94,14 @@ Simulation.Aivika.Distributed.Optimistic.State Simulation.Aivika.Distributed.Optimistic.TimeServer - other-modules: Simulation.Aivika.Distributed.Optimistic.Internal.Channel+ other-modules: Simulation.Aivika.Distributed.Optimistic.Internal.AcknowledgementMessageQueue+ Simulation.Aivika.Distributed.Optimistic.Internal.Channel+ Simulation.Aivika.Distributed.Optimistic.Internal.ConnectionManager Simulation.Aivika.Distributed.Optimistic.Internal.DIO Simulation.Aivika.Distributed.Optimistic.Internal.Event Simulation.Aivika.Distributed.Optimistic.Internal.Expect 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