aivika-distributed 0.8 → 1.0

raw patch · 7 files changed

+156/−46 lines, 7 filesdep +arrayPVP ok

version bump matches the API change (PVP)

Dependencies added: array

API changes (from Hackage documentation)

Files

• CHANGELOG.md +9/−0
• Simulation/Aivika/Distributed/Optimistic/Guard.hs +1/−1
• Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +5/−2
• Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +5/−3
• Simulation/Aivika/Distributed/Optimistic/Internal/UndoableLog.hs +124/−32
• Simulation/Aivika/Distributed/Optimistic/Message.hs +1/−0
• aivika-distributed.cabal +11/−8

@@ -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 ----- 

@@ -1,5 +1,5 @@ -{-# LANGUAGE DeriveGeneric, DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric, DeriveDataTypeable, MonoLocalBinds #-}  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Guard

@@ -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,

@@ -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"

@@ -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

@@ -1,6 +1,7 @@  {-# LANGUAGE RankNTypes #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MonoLocalBinds #-}  -- | -- Module     : Simulation.Aivika.Distributed.Optimistic.Message

@@ -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