diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,13 @@ +Version 1.0 +----- + +* Optimized the rollback log. + +* Increased the default rollback log threshold. + +* Returned the size threshold for the output message queue. + Version 0.8 ----- diff --git a/Simulation/Aivika/Distributed/Optimistic/Guard.hs b/Simulation/Aivika/Distributed/Optimistic/Guard.hs --- a/Simulation/Aivika/Distributed/Optimistic/Guard.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Guard.hs @@ -1,5 +1,5 @@ -{-# LANGUAGE DeriveGeneric, DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric, DeriveDataTypeable, MonoLocalBinds #-} -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Guard 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 @@ -74,6 +74,8 @@ -- ^ The name of the logical process. dioUndoableLogSizeThreshold :: Int, -- ^ 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, @@ -191,8 +193,9 @@ defaultDIOParams = DIOParams { dioLoggingPriority = WARNING, dioName = "LP", - dioUndoableLogSizeThreshold = 1000000, - dioTransientMessageQueueSizeThreshold = 10000, + dioUndoableLogSizeThreshold = 10000000, + dioOutputMessageQueueSizeThreshold = 10000, + dioTransientMessageQueueSizeThreshold = 5000, dioSyncTimeout = 60000000, dioAllowPrematureIO = False, dioAllowSkippingOutdatedMessage = True, diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs @@ -282,11 +282,13 @@ Event $ \p -> do let q = runEventQueue $ pointRun p n1 <- liftIOUnsafe $ logSize (queueLog q) - n2 <- liftIOUnsafe $ transientMessageQueueSize (queueTransientMessages q) + n2 <- liftIOUnsafe $ outputMessageQueueSize (queueOutputMessages q) + n3 <- liftIOUnsafe $ transientMessageQueueSize (queueTransientMessages q) ps <- dioParams let th1 = dioUndoableLogSizeThreshold ps - th2 = dioTransientMessageQueueSizeThreshold ps - if (n1 >= th1) || (n2 >= th2) + th2 = dioOutputMessageQueueSizeThreshold ps + th3 = dioTransientMessageQueueSizeThreshold ps + if (n1 >= th1) || (n2 >= th2) || (n3 >= th3) then do logDIO NOTICE $ "t = " ++ (show $ pointTime p) ++ ": detected the event overflow" diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs @@ -18,6 +18,7 @@ logSize) where import Data.IORef +import Data.Array.IO.Safe import Control.Monad import Control.Monad.Trans @@ -30,45 +31,62 @@ import Simulation.Aivika.Distributed.Optimistic.Internal.DIO import Simulation.Aivika.Distributed.Optimistic.Internal.IO --- | Specified an undoable log with ability to rollback the operations. +-- | Specifies an undoable log with ability to rollback the operations. data UndoableLog = - UndoableLog { logItems :: DLL.DoubleLinkedList UndoableItem - -- ^ The items that can be undone. + UndoableLog { logItemChunks :: DLL.DoubleLinkedList UndoableItemChunk + -- ^ The chunk of items that can be undone. } -data UndoableItem = - UndoableItem { itemTime :: Double, - -- ^ The time at which the operation had occured. - itemUndo :: DIO () - -- ^ Undo the operation - } +-- | A chunk of undoable operation items. +data UndoableItemChunk = + UndoableItemChunk { chunkItemStart :: IORef Int, + -- ^ the start item index in the chunk. + chunkItemEnd :: IORef Int, + -- ^ the end item index in the chunk. + chunkItemTimes :: IOUArray Int Double, + -- ^ the times at which the operations had occurred. + chunkItemUndo :: IOArray Int (DIO ()) + -- ^ the undo operations. + } +-- | The chunk item capacity. +chunkItemCapacity = 512 + -- | Create an undoable log. newUndoableLog :: DIO UndoableLog newUndoableLog = do xs <- liftIOUnsafe DLL.newList - return UndoableLog { logItems = xs } + return UndoableLog { logItemChunks = xs } -- | Write a new undoable operation. writeLog :: UndoableLog -> DIO () -> Event DIO () -{-# INLINE writeLog #-} +{-# INLINABLE writeLog #-} writeLog log h = Event $ \p -> - do let x = UndoableItem { itemTime = pointTime p, itemUndo = h } + do let t = pointTime p --- - --- logDIO DEBUG $ "Writing the log at t = " ++ show (itemTime x) + --- logDIO DEBUG $ "Writing the log at t = " ++ show t --- liftIOUnsafe $ - do f <- DLL.listNull (logItems log) + do f <- DLL.listNull (logItemChunks log) if f - then DLL.listAddLast (logItems log) x - else do x0 <- DLL.listLast (logItems log) - when (itemTime x < itemTime x0) $ + then do ch <- newItemChunk t h + DLL.listAddLast (logItemChunks log) ch + else do ch <- DLL.listLast (logItemChunks log) + e <- readIORef (chunkItemEnd ch) + t0 <- readArray (chunkItemTimes ch) (e - 1) + when (t < t0) $ error $ "The logging data are not sorted by time (" ++ - (show $ itemTime x) ++ " < " ++ - (show $ itemTime x0) ++ "): writeLog" - DLL.listAddLast (logItems log) x + show t ++ " < " ++ + show t0 ++ "): writeLog" + if e == chunkItemCapacity + then do ch <- newItemChunk t h + DLL.listAddLast (logItemChunks log) ch + else do let e' = e + 1 + writeArray (chunkItemTimes ch) e t + writeArray (chunkItemUndo ch) e h + e' `seq` writeIORef (chunkItemEnd ch) e' -- | Rollback the log till the specified time either including that one or not. rollbackLog :: UndoableLog -> Double -> Bool -> DIO () @@ -79,24 +97,98 @@ loop where loop = - do f <- liftIOUnsafe $ DLL.listNull (logItems log) + do f <- liftIOUnsafe $ DLL.listNull (logItemChunks log) unless f $ - do x <- liftIOUnsafe $ DLL.listLast (logItems log) - when ((t < itemTime x) || (including && t == itemTime x)) $ - do liftIOUnsafe $ DLL.listRemoveLast (logItems log) - itemUndo x - loop + do ch <- liftIOUnsafe $ DLL.listLast (logItemChunks log) + s <- liftIOUnsafe $ readIORef (chunkItemStart ch) + let inner e = + if e == s + then do liftIOUnsafe $ DLL.listRemoveLast (logItemChunks log) + loop + else do let e' = e - 1 + t0 <- e' `seq` liftIOUnsafe $ readArray (chunkItemTimes ch) e' + when ((t < t0) || (including && t == t0)) $ + do h <- liftIOUnsafe $ readArray (chunkItemUndo ch) e' + liftIOUnsafe $ writeArray (chunkItemUndo ch) e' undefined + liftIOUnsafe $ writeIORef (chunkItemEnd ch) e' + h + inner e' + e <- liftIOUnsafe $ readIORef (chunkItemEnd ch) + inner e -- | Reduce the log removing all items older than the specified time. reduceLog :: UndoableLog -> Double -> IO () reduceLog log t = - do f <- DLL.listNull (logItems log) + do f <- DLL.listNull (logItemChunks log) unless f $ - do x <- DLL.listFirst (logItems log) - when (itemTime x < t) $ - do DLL.listRemoveFirst (logItems log) - reduceLog log t + do ch <- DLL.listFirst (logItemChunks log) + e <- readIORef (chunkItemEnd ch) + t0 <- readArray (chunkItemTimes ch) (e - 1) + if t0 < t + then do DLL.listRemoveFirst (logItemChunks log) + reduceLog log t + else do let loop s = + if s == e + then error "The log is corrupted: reduceLog" + else do t0 <- readArray (chunkItemTimes ch) s + when (t0 < t) $ + do let s' = s + 1 + writeArray (chunkItemUndo ch) s undefined + s' `seq` writeIORef (chunkItemStart ch) s' + loop s' + s <- readIORef (chunkItemStart ch) + loop s -- | Return the log size. logSize :: UndoableLog -> IO Int -logSize log = DLL.listCount (logItems log) +logSize log = + do n <- DLL.listCount (logItemChunks log) + if n == 0 + then return 0 + else do n1 <- firstItemChunkSize log + n2 <- lastItemChunkSize log + if n == 1 + then if n1 == n2 + then return n1 + else error "The log is corrupted: logSize" + else return (n1 + (n - 2) * chunkItemCapacity + n2) + +-- | Create a new item chunk. +newItemChunk :: Double -> DIO () -> IO UndoableItemChunk +{-# INLINABLE newItemChunk #-} +newItemChunk t h = + do times <- newArray_ (0, chunkItemCapacity - 1) + undo <- newArray_ (0, chunkItemCapacity - 1) + start <- newIORef 0 + end <- newIORef 1 + writeArray times 0 t + writeArray undo 0 h + return UndoableItemChunk { chunkItemStart = start, + chunkItemEnd = end, + chunkItemTimes = times, + chunkItemUndo = undo } + +-- | Get the item chunk size. +itemChunkSize :: UndoableItemChunk -> IO Int +itemChunkSize ch = + do s <- readIORef (chunkItemStart ch) + e <- readIORef (chunkItemEnd ch) + return (e - s) + +-- | Get the first item chunk size; otherwise, return zero. +firstItemChunkSize :: UndoableLog -> IO Int +firstItemChunkSize log = + do f <- DLL.listNull (logItemChunks log) + if f + then return 0 + else do ch <- DLL.listFirst (logItemChunks log) + itemChunkSize ch + +-- | Get the last item chunk size; otherwise, return zero. +lastItemChunkSize :: UndoableLog -> IO Int +lastItemChunkSize log = + do f <- DLL.listNull (logItemChunks log) + if f + then return 0 + else do ch <- DLL.listLast (logItemChunks log) + itemChunkSize ch diff --git a/Simulation/Aivika/Distributed/Optimistic/Message.hs b/Simulation/Aivika/Distributed/Optimistic/Message.hs --- a/Simulation/Aivika/Distributed/Optimistic/Message.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Message.hs @@ -1,6 +1,7 @@ {-# LANGUAGE RankNTypes #-} {-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE MonoLocalBinds #-} -- | -- Module : Simulation.Aivika.Distributed.Optimistic.Message 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: 0.8 +version: 1.0 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. @@ -25,13 +25,14 @@ 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 estimation is as follows. The distributed simulation module is slower up to - 12-30 times in comparison with the sequential aivika [2] simulation library using the equivalent sequential models. - The estimation has dramatically changed after started using another more fast pseudo-random number generator by default, - which made the sequential module even more fast. The lower estimation in 12 times is likely to correspond to complex models. - The upper estimation in 30 times will probably correspond to quite simple event-oriented and process-oriented models, where - the sequential module can be exceptionally fast. + Regarding the speed of simulation, the recent rough estimation is as follows. This estimation may change from + version to version. For example, in version 1.0 the rollback log was rewritten, which had a significant effect. . + The distributed simulation module is slower up to 8-15 times in comparison with the sequential aivika [2] simulation + library using the equivalent sequential models. The lower estimation in 8 times is likely to correspond to complex + models. The upper estimation in 15 times will probably correspond to quite simple event-oriented and process-oriented + models, where the sequential module can be exceptionally fast. + . Note that you can run up to 7 parallel logical 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 logical processes simultaneously. . @@ -113,6 +114,7 @@ build-depends: base >= 4.6.0.0 && < 6, mtl >= 2.1.1, + array >= 0.3.0.0, stm >= 2.4.2, random >= 1.0.0.3, mwc-random >= 0.13.0.0, @@ -131,7 +133,8 @@ RankNTypes, DeriveGeneric, DeriveDataTypeable, - OverlappingInstances + OverlappingInstances, + MonoLocalBinds ghc-options: -O2