aivika-distributed 1.1.2 → 1.2
raw patch · 6 files changed
+82/−38 lines, 6 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- CHANGELOG.md +5/−0
- LICENSE +1/−1
- Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +6/−14
- Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs +46/−12
- Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs +20/−7
- aivika-distributed.cabal +4/−4
CHANGELOG.md view
@@ -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 -----
LICENSE view
@@ -1,4 +1,4 @@-Copyright (c) 2015-2017 David Sorokin <david.sorokin@gmail.com>+Copyright (c) 2015-2018 David Sorokin <david.sorokin@gmail.com> All rights reserved.
Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs view
@@ -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 ---
Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs view
@@ -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
Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs view
@@ -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.
aivika-distributed.cabal view
@@ -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 <david.sorokin@gmail.com>+copyright: (c) 2015-2018. David Sorokin <david.sorokin@gmail.com> author: David Sorokin maintainer: David Sorokin <david.sorokin@gmail.com> homepage: http://www.aivikasoft.com