diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,9 @@ +Version 1.2 +----- + +* Fixed a leak of monitor references that are used in the fault tolerant mode only. + Version 1.1.2 ----- diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2015-2017 David Sorokin +Copyright (c) 2015-2018 David Sorokin All rights reserved. diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs @@ -379,17 +379,12 @@ SendTerminateTimeServerMessage receiver sender -> DP.send receiver (TerminateTimeServerMessage sender) MonitorProcessMessage pid -> - do --- - logProcess ps INFO $ "Monitoring process " ++ show pid - --- - f <- liftIO $ + do f <- liftIO $ existsKeepAliveReceiver keepAliveManager pid unless f $ - do DP.monitor pid - liftIO $ - addKeepAliveReceiver keepAliveManager pid + addKeepAliveReceiver keepAliveManager pid ReMonitorProcessMessage pids -> - handleReMonitorProcessMessage pids ps ch + handleReMonitorProcessMessage pids ps ch keepAliveManager TrySendKeepAliveMessage -> trySendKeepAlive keepAliveManager RegisterLogicalProcessAcknowledgementMessage pid -> @@ -540,13 +535,10 @@ return () -- | Re-monitor the specified remote processes. -handleReMonitorProcessMessage :: [DP.ProcessId] -> DIOParams -> Channel LogicalProcessMessage -> DP.Process () -handleReMonitorProcessMessage pids ps ch = +handleReMonitorProcessMessage :: [DP.ProcessId] -> DIOParams -> Channel LogicalProcessMessage -> KeepAliveManager -> DP.Process () +handleReMonitorProcessMessage pids ps ch keepAliveManager = forM_ pids $ \pid -> - do --- - logProcess ps NOTICE $ "Re-monitoring " ++ show pid - --- - DP.monitor pid + do remonitorKeepAliveReceiver keepAliveManager pid --- logProcess ps NOTICE $ "Writing to the channel about reconnecting to " ++ show pid --- diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs @@ -14,11 +14,12 @@ KeepAliveParams(..), newKeepAliveManager, addKeepAliveReceiver, + remonitorKeepAliveReceiver, existsKeepAliveReceiver, trySendKeepAlive, trySendKeepAliveUTC) where -import qualified Data.Set as S +import qualified Data.Map as M import Data.Maybe import Data.IORef import Data.Time.Clock @@ -44,35 +45,67 @@ -- ^ the manager parameter keepAliveTimestamp :: IORef UTCTime, -- ^ the keep alive timestamp - keepAliveReceivers :: IORef (S.Set DP.ProcessId) + 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 S.empty + receivers <- newIORef M.empty return KeepAliveManager { keepAliveParams = ps, keepAliveTimestamp = timestamp, keepAliveReceivers = receivers } -- | Add the keep-alive message receiver. -addKeepAliveReceiver :: KeepAliveManager -> DP.ProcessId -> IO () +addKeepAliveReceiver :: KeepAliveManager -> DP.ProcessId -> DP.Process () addKeepAliveReceiver manager pid = - modifyIORef (keepAliveReceivers manager) $ - S.insert 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 . S.member pid + return . M.member pid -- | Try to send a keep-alive message. trySendKeepAlive :: KeepAliveManager -> DP.Process () trySendKeepAlive manager = - do empty <- liftIO $ fmap S.null $ readIORef (keepAliveReceivers manager) + do empty <- liftIO $ fmap M.null $ readIORef (keepAliveReceivers manager) unless empty $ do utc <- liftIO getCurrentTime trySendKeepAliveUTC manager utc @@ -80,7 +113,7 @@ -- | 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) + do empty <- liftIO $ fmap M.null $ readIORef (keepAliveReceivers manager) unless empty $ do f <- liftIO $ shouldSendKeepAlive manager utc when f $ @@ -89,9 +122,10 @@ "Sending a keep-alive message" --- liftIO $ writeIORef (keepAliveTimestamp manager) utc - pids <- liftIO $ readIORef (keepAliveReceivers manager) - forM_ pids $ \pid -> - DP.send pid KeepAliveMessage + 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs @@ -299,13 +299,26 @@ DP.send pid msg 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 - --- + 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 -- | Whether the both values are defined and the first is greater than or equaled to the second. diff --git a/aivika-distributed.cabal b/aivika-distributed.cabal --- a/aivika-distributed.cabal +++ b/aivika-distributed.cabal @@ -1,5 +1,5 @@ name: aivika-distributed -version: 1.1.2 +version: 1.2 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. @@ -32,8 +32,8 @@ library using the equivalent sequential models. The lower estimation in 8 times is likely to correspond to complex models. The upper estimation in 15 times will probably correspond to quite simple event-oriented and process-oriented models, where the sequential module can be exceptionally fast. At the same time, the message passing between - the logical processes can dramatically decrease the speed of distributed simulation, especially if they cause rollbacks. - Thus, much depends on the distributed model itself. + the logical processes can dramatically decrease the speed of distributed simulation, especially if the messages cause + rollbacks. Thus, much depends on the distributed model itself. . When residing the logical processes in a computer with multi-core processor, you should follow the next recommendations. You should reserve at least 1 core for each logical process, or even reserve 2 cores if the logical process extensively @@ -53,7 +53,7 @@ category: Simulation license: BSD3 license-file: LICENSE -copyright: (c) 2015-2017. David Sorokin +copyright: (c) 2015-2018. David Sorokin author: David Sorokin maintainer: David Sorokin homepage: http://www.aivikasoft.com