aivika-distributed 0.5 → 0.6

raw patch · 9 files changed

+456/−7 lines, 9 filesdep +mwc-randomdep ~aivikadep ~aivika-transformersPVP ok

version bump matches the API change (PVP)

Dependencies added: mwc-random

Dependency ranges changed: aivika, aivika-transformers

API changes (from Hackage documentation)

Files

• CHANGELOG.md +12/−0
• Simulation/Aivika/Distributed/Optimistic.hs +2/−0
• Simulation/Aivika/Distributed/Optimistic/DIO.hs +1/−0
• Simulation/Aivika/Distributed/Optimistic/Generator.hs +5/−1
• Simulation/Aivika/Distributed/Optimistic/Guard.hs +166/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +67/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/Expect.hs +43/−0
• aivika-distributed.cabal +10/−6
• tests/Guard1.hs +150/−0

@@ -1,4 +1,16 @@ +Version 0.6+-----++* Using the mwc-random package for generating random numbers by default.++Version 0.5.1+-----++* Added functions expectEvent and expectProcess.++* Added the Guard module.+ Version 0.5 ----- 

@@ -13,6 +13,7 @@ module Simulation.Aivika.Distributed.Optimistic        (-- * Modules         module Simulation.Aivika.Distributed.Optimistic.DIO,+        module Simulation.Aivika.Distributed.Optimistic.Guard,         module Simulation.Aivika.Distributed.Optimistic.Generator,         module Simulation.Aivika.Distributed.Optimistic.Message,         module Simulation.Aivika.Distributed.Optimistic.QueueStrategy,@@ -21,6 +22,7 @@         module Simulation.Aivika.Distributed.Optimistic.TimeServer) where  import Simulation.Aivika.Distributed.Optimistic.DIO+import Simulation.Aivika.Distributed.Optimistic.Guard import Simulation.Aivika.Distributed.Optimistic.Generator import Simulation.Aivika.Distributed.Optimistic.Message import Simulation.Aivika.Distributed.Optimistic.QueueStrategy

@@ -39,6 +39,7 @@ import Simulation.Aivika.Trans.Process import Simulation.Aivika.Trans.Ref.Base import Simulation.Aivika.Trans.QueueStrategy+import Simulation.Aivika.Trans.Internal.Types  import Simulation.Aivika.Distributed.Optimistic.Internal.Message import Simulation.Aivika.Distributed.Optimistic.Internal.DIO

@@ -18,6 +18,7 @@ import Control.Monad.Trans  import System.Random+import qualified System.Random.MWC as MWC  import Data.IORef @@ -69,7 +70,10 @@   newGenerator tp =     case tp of       SimpleGenerator ->-        liftIOUnsafe newStdGen >>= newRandomGenerator+        do let g = MWC.uniform <$>+                   MWC.withSystemRandom (return :: MWC.GenIO -> IO MWC.GenIO)+           g' <- liftIOUnsafe g+           newRandomGenerator01 (liftIOUnsafe g')       SimpleGeneratorWithSeed x ->         error "Unsupported generator type SimpleGeneratorWithSeed: newGenerator"       CustomGenerator g ->

@@ -0,0 +1,166 @@++{-# LANGUAGE DeriveGeneric, DeriveDataTypeable #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Guard+-- 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 8.0.2+--+-- This module defines guards that allow correct finishing the distributed simulation.+--+module Simulation.Aivika.Distributed.Optimistic.Guard+       (-- * Guards With Message Passing+        runMasterGuard,+        runSlaveGuard,+        -- * Guards Without Message Passing+        runMasterGuard_,+        runSlaveGuard_)  where++import GHC.Generics++import Data.Typeable+import Data.Binary+import qualified Data.Map as M++import Control.Monad++import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Serializable++import Simulation.Aivika.Trans+import Simulation.Aivika.Trans.Internal.Types+import Simulation.Aivika.Distributed.Optimistic.Internal.Expect+import Simulation.Aivika.Distributed.Optimistic.DIO+import Simulation.Aivika.Distributed.Optimistic.Message++-- | Represent the master message.+data MasterMessage a = MasterMessage DP.ProcessId (Maybe a)+                     deriving (Show, Typeable, Generic)++-- | Represent the slave message.+data SlaveMessage a = SlaveMessage DP.ProcessId a+                    deriving (Show, Typeable, Generic)+                   +instance Binary a => Binary (MasterMessage a)+instance Binary a => Binary (SlaveMessage a)++-- | Represents the master guard that waits for all slaves to finish.+data MasterGuard a = MasterGuard { masterGuardSlaveMessages :: Ref DIO (M.Map DP.ProcessId a)+                                   -- ^ the messages of slaves connected to the master+                                 }++-- | Represents the slave guard that waits for the master's acknowledgement.+data SlaveGuard a = SlaveGuard { slaveGuardAcknowledgedMessage :: Ref DIO (Maybe (Maybe a))+                                 -- ^ whether the slave process was acknowledged by the master+                               }++-- | Create a new master guard.+newMasterGuard :: Serializable a => Event DIO (MasterGuard a)+newMasterGuard =+  do r <- liftSimulation $ newRef M.empty+     handleSignal messageReceived $ \(SlaveMessage slaveId a) ->+       modifyRef r $ M.insert slaveId a+     return MasterGuard { masterGuardSlaveMessages = r }++-- | Create a new slave guard.+newSlaveGuard :: Serializable a => Event DIO (SlaveGuard a)+newSlaveGuard =+  do r <- liftSimulation $ newRef Nothing+     handleSignal messageReceived $ \(MasterMessage masterId a) ->+       writeRef r (Just a)+     return SlaveGuard { slaveGuardAcknowledgedMessage = r }++-- | Await until the specified number of slaves are connected to the master.+awaitMasterGuard :: Serializable b+                    => MasterGuard a+                    -- ^ the master guard+                    -> Int+                    -- ^ the number of slaves to wait+                    -> (M.Map DP.ProcessId a -> Event DIO (M.Map DP.ProcessId b))+                    -- ^ process the messages sent by slaves+                    -> Process DIO (M.Map DP.ProcessId b)+awaitMasterGuard guard n transform =+  expectProcess $+  do m <- readRef $ masterGuardSlaveMessages guard+     if M.size m < n+       then return Nothing+       else do m' <- transform m+               inboxId <- liftComp messageInboxId+               forM_ (M.keys m) $ \slaveId ->+                 sendMessage slaveId (MasterMessage inboxId $ M.lookup slaveId m')+               return $ Just m'++-- | Await until the specified master receives notifications from the slave processes.+awaitSlaveGuard :: (Serializable a,+                    Serializable b)+                   => SlaveGuard a+                   -- ^ the slave guard+                   -> DP.ProcessId+                   -- ^ the master process identifier+                   -> Event DIO b+                   -- ^ the message generator+                   -> Process DIO (Maybe a)+                   -- ^ the master's reply+awaitSlaveGuard guard masterId generator =+  do liftEvent $+       do b <- generator+          inboxId <- liftComp messageInboxId+          sendMessage masterId (SlaveMessage inboxId b)+     expectProcess $+       readRef $ slaveGuardAcknowledgedMessage guard++-- | Run the master guard by the specified number of slaves and transform function.+runMasterGuard :: (Serializable a,+                   Serializable b)+                  => Int+                  -- ^ the number of slaves to wait+                  -> (M.Map DP.ProcessId a -> Event DIO (M.Map DP.ProcessId b))+                  -- ^ how to transform the messages from the slave processes in the stop time+                  -> Process DIO (M.Map DP.ProcessId b)+runMasterGuard n transform =+  do source <- liftSimulation newSignalSource+     liftEvent $+       do guard <- newMasterGuard+          enqueueEventWithStopTime $+            runProcess $+            do b <- awaitMasterGuard guard n transform+               liftEvent $+                 triggerSignal source b+     processAwait $ publishSignal source++-- | Run the slave guard by the specified master process identifier and message generator.+runSlaveGuard :: (Serializable a,+                  Serializable b)+                 => DP.ProcessId+                 -- ^ the master process identifier+                 -> Event DIO a+                 -- ^ in the stop time generate a message to pass to the master process+                 -> Process DIO (Maybe b)+                 -- ^ the message returned by the master process+runSlaveGuard masterId generator =+  do source <- liftSimulation newSignalSource+     liftEvent $+       do guard <- newSlaveGuard+          enqueueEventWithStopTime $+            runProcess $+            do b <- awaitSlaveGuard guard masterId generator+               liftEvent $+                 triggerSignal source b+     processAwait $ publishSignal source++-- | Run the master guard by the specified number of slaves when there is no message passing.+runMasterGuard_ :: Int -> Process DIO ()+runMasterGuard_ n =+  do _ <- runMasterGuard n transform :: Process DIO (M.Map DP.ProcessId ())+     return ()+       where transform m = return m++-- | Run the slave guard by the specified master process identifier when there is no message passing.+runSlaveGuard_ :: DP.ProcessId -> Process DIO ()+runSlaveGuard_ masterId =+  do _ <- runSlaveGuard masterId generator :: Process DIO (Maybe ())+     return ()+       where generator = return ()

@@ -15,6 +15,7 @@        (queueInputMessages,         queueOutputMessages,         queueLog,+        expectEvent,         processMonitorSignal) where  import Data.Maybe@@ -32,6 +33,9 @@  import Simulation.Aivika.Trans import Simulation.Aivika.Trans.Internal.Types+import Simulation.Aivika.Trans.Internal.Event+import Simulation.Aivika.Trans.Internal.Cont+import Simulation.Aivika.Trans.Internal.Process  import Simulation.Aivika.Distributed.Optimistic.Internal.Priority import Simulation.Aivika.Distributed.Optimistic.Internal.Channel@@ -761,3 +765,66 @@             in invokeEvent p $                handleSignal s h          }+++-- | Suspend the 'Event' computation until the specified computation is determined.+--+-- The tested computation should depend on messages that come from other local processes.+-- Moreover, the event must be initiated through the event queue.+expectEvent :: Event DIO (Maybe a) -> (a -> Event DIO ()) -> Event DIO ()+expectEvent m cont =+  Event $ \p ->+  do let q = runEventQueue $ pointRun p+         t = pointTime p+     ---+     logDIO INFO $+       "t = " ++ show (pointTime p) +++       ": expecting the computation result: expectEvent"+     ---+     let loop =+           do ---+              --- logDIO DEBUG $+              ---   "t = " ++ show (pointTime p) +++              ---   ": testing the predicate: expectEvent"+              ---+              x <- invokeEvent p m+              case x of+                Just a  -> invokeEvent p $ cont a+                Nothing -> next loop0+         loop0 =+           do ---+              --- logDIO DEBUG $+              ---   "t = " ++ show (pointTime p) +++              ---   ": testing the predicate in ring 0: expectEvent"+              ---+              x <- invokeEvent p m+              case x of+                Just a  -> invokeEvent p $ cont a+                Nothing -> next $ error "Detected a deadlock: expectEvent"+         next loop' =+           do pq <- invokeEvent p $ R.readRef $ queuePQ q+              let f = PQ.queueNull pq+              if f+                then await loop'+                else do let (t2, _) = PQ.queueFront pq+                        if t < t2+                          then await loop'+                          else invokeEvent p $+                               enqueueEvent t $+                               Event $ \p -> loop+         await loop' =+           do ---+              --- logDIO DEBUG $+              ---   "t = " ++ show t +++              ---   ": waiting for arriving a message: expectEvent"+              ---+              ch <- messageChannel+              dt <- fmap dioSyncTimeout dioParams+              f  <- liftIOUnsafe $+                    timeout dt $ awaitChannel ch+              ok <- invokeEvent p $ runTimeWarp processChannelMessages+              when ok $+                case f of+                  Just _  -> loop+                  Nothing -> loop'+     loop

@@ -0,0 +1,43 @@++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.Internal.Expect+-- 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+--+-- Implements the 'expectProcess' function that allows waiting for some computation to have a determined result.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.Expect+       (expectProcess) where++import Control.Monad+import Control.Monad.Trans++import Simulation.Aivika.Trans+import Simulation.Aivika.Trans.Internal.Types+import Simulation.Aivika.Trans.Internal.Simulation+import Simulation.Aivika.Trans.Internal.Event+import Simulation.Aivika.Trans.Internal.Cont+import Simulation.Aivika.Trans.Internal.Process++import Simulation.Aivika.Distributed.Optimistic.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.Event++-- | Suspend the 'Process' until the specified computation is determined.+--  +-- The tested computation should depend on messages that come from other local processes.+-- Moreover, the process must be initiated through the event queue.+--+-- In the current implementation there is a limitation that this function can be used only+-- once for the entire local process simulation; otherwise, a race condition may arise.+expectProcess :: Event DIO (Maybe a) -> Process DIO a+expectProcess m =+  Process $ \pid ->+  Cont $ \c ->+  Event $ \p ->+  do let t = pointTime p+     invokeEvent p $+       expectEvent m $+       resumeCont c

@@ -1,5 +1,5 @@ name:            aivika-distributed-version:         0.5+version:         0.6 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.@@ -17,7 +17,7 @@     is that this is the optimistic distributed simulation with possible rollbacks. It is assumed that optimistic methods      tend to better support the parallelism inherited in the models.      .-    You may test the distributed simulation using your own notebook only, although the package is still destined to be +    You may test the distributed simulation using your own laptop only, although the package is still destined to be      used with a multi-core computer, or computers connected in the distributed cluster.     .     There are additional packages that allow you to run the distributed simulation experiments by @@ -25,7 +25,7 @@     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 estimations are as follows. The simulation is slower up to+    Regarding the speed of simulation, the rough estimations are as follows. The distributed simulation module is slower up to     6-9 times in comparison with the sequential aivika [2] simulation library using the equivalent sequential models.      Note that you can run up to 7 parallel local 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 local processes simultaneously.@@ -51,7 +51,8 @@ build-type:      Simple tested-with:     GHC == 7.10.1 -extra-source-files:  tests/MachRep1.hs+extra-source-files:  tests/Guard1.hs+                     tests/MachRep1.hs                      tests/MachRep1Simple.hs                      tests/MachRep2.hs                      tests/MachRep2Distributed.hs@@ -67,6 +68,7 @@     exposed-modules: Simulation.Aivika.Distributed                      Simulation.Aivika.Distributed.Optimistic                      Simulation.Aivika.Distributed.Optimistic.DIO+                     Simulation.Aivika.Distributed.Optimistic.Guard                      Simulation.Aivika.Distributed.Optimistic.Generator                      Simulation.Aivika.Distributed.Optimistic.QueueStrategy                      Simulation.Aivika.Distributed.Optimistic.Message@@ -79,6 +81,7 @@     other-modules:   Simulation.Aivika.Distributed.Optimistic.Internal.Channel                      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@@ -98,13 +101,14 @@                      mtl >= 2.1.1,                      stm >= 2.4.2,                      random >= 1.0.0.3,+                     mwc-random >= 0.13.0.0,                      binary >= 0.6.4.0,                      time >= 1.5.0.1,                      containers >= 0.4.0.0,                      exceptions >= 0.8.0.2,                      distributed-process >= 0.6.1,-                     aivika >= 5.1,-                     aivika-transformers >= 5.1+                     aivika >= 5.2,+                     aivika-transformers >= 5.2      extensions:      MultiParamTypeClasses,                      FlexibleInstances,

@@ -0,0 +1,150 @@++{--# LANGUAGE TemplateHaskell #--}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveDataTypeable #-}++-- It corresponds to model MachRep1 described in document +-- Introduction to Discrete-Event Simulation and the SimPy Language+-- [http://heather.cs.ucdavis.edu/~matloff/156/PLN/DESimIntro.pdf]. +-- SimPy is available on [http://simpy.sourceforge.net/].+--   +-- The model description is as follows.+--+-- Two machines, which sometimes break down.+-- Up time is exponentially distributed with mean 1.0, and repair time is+-- exponentially distributed with mean 0.5. There are two repairpersons,+-- so the two machines can be repaired simultaneously if they are down+-- at the same time.+--+-- Output is long-run proportion of up time. Should get value of about+-- 0.66.++import System.Environment (getArgs)++import Data.Typeable+import Data.Binary++import GHC.Generics++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Distributed.Process.Backend.SimpleLocalnet++import Simulation.Aivika.Trans+import Simulation.Aivika.Distributed++meanUpTime = 1.0+meanRepairTime = 0.5++specs = Specs { spcStartTime = 0.0,+                spcStopTime = 10000.0,+                spcDT = 1.0,+                spcMethod = RungeKutta4,+                spcGeneratorType = SimpleGenerator }++newtype TotalUpTimeChange = TotalUpTimeChange { runTotalUpTimeChange :: Double }+                          deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary TotalUpTimeChange++-- | A sub-model.+slaveModel :: DP.ProcessId -> Simulation DIO ()+slaveModel masterId =+  do let machine =+           do upTime <-+                liftParameter $+                randomExponential meanUpTime+              holdProcess upTime+              ---+              liftEvent $+                sendMessage masterId (TotalUpTimeChange upTime)+              ---+              repairTime <-+                liftParameter $+                randomExponential meanRepairTime+              holdProcess repairTime+              machine++     runProcessInStartTime machine++     runProcessInStartTime $+       runSlaveGuard_ masterId++     runEventInStartTime $+       enqueueEventIOWithStopTime $+       do liftIO $+            putStrLn "The sub-model finished"++     runEventInStopTime $+       return ()++-- | The main model.       +masterModel :: Int -> Simulation DIO Double+masterModel n =+  do totalUpTime <- newRef 0.0++     ---+     let totalUpTimeChanged :: Signal DIO TotalUpTimeChange+         totalUpTimeChanged = messageReceived++     runEventInStartTime $+       handleSignal totalUpTimeChanged $ \x ->+       modifyRef totalUpTime (+ runTotalUpTimeChange x)+     ---++     runProcessInStartTime $+       runMasterGuard_ n++     let upTimeProp =+           do x <- readRef totalUpTime+              y <- liftDynamics time+              return $ x / (fromIntegral n * y)++     runEventInStopTime $+       do liftIO $+            putStrLn "The main model finished"+          upTimeProp++runSlaveModel :: (DP.ProcessId, DP.ProcessId) -> DP.Process (DP.ProcessId, DP.Process ())+runSlaveModel (timeServerId, masterId) =+  runDIO m ps timeServerId+  where+    ps = defaultDIOParams { dioLoggingPriority = NOTICE }+    m  = do registerDIO+            runSimulation (slaveModel masterId) specs+            unregisterDIO++runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process Double)+runMasterModel timeServerId n =+  runDIO m ps timeServerId+  where+    ps = defaultDIOParams { dioLoggingPriority = NOTICE }+    m  = do registerDIO+            a <- runSimulation (masterModel n) specs+            terminateDIO+            return a++master = \backend nodes ->+  do liftIO . putStrLn $ "Slaves: " ++ show nodes+     let timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG }+     timeServerId <- DP.spawnLocal $ timeServer 3 timeServerParams+     (masterId, masterProcess) <- runMasterModel timeServerId 2+     forM_ [1..2] $ \i ->+       do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId)+          DP.spawnLocal slaveProcess+     a <- masterProcess+     DP.say $+       "The result is " ++ show a+  +main :: IO ()+main = do+  backend <- initializeBackend "localhost" "8080" rtable+  startMaster backend (master backend)+    where+      rtable :: DP.RemoteTable+      -- rtable = __remoteTable initRemoteTable+      rtable = initRemoteTable