DSTM (empty) → 0.1
raw patch · 24 files changed
+3301/−0 lines, 24 filesdep +basedep +containersdep +haskell98setup-changedbinary-added
Dependencies added: base, containers, haskell98, network, process, unix
Files
- Bomberman/Bomberman.hs +818/−0
- Bomberman/field.txt +10/−0
- Bomberman/input1.txt +1/−0
- Bomberman/input2.txt +1/−0
- Bomberman/input3.txt +1/−0
- Chat/ChatClient.hs +45/−0
- Chat/ChatServer.hs +46/−0
- Control/Distributed/STM/DSTM.hs +684/−0
- Control/Distributed/STM/DebugBase.hs +180/−0
- Control/Distributed/STM/DebugDSTM.hs +44/−0
- Control/Distributed/STM/Dist.hs +108/−0
- Control/Distributed/STM/EnvAddr.hs +375/−0
- Control/Distributed/STM/NameService.hs +117/−0
- Control/Distributed/STM/RetryVar.hs +123/−0
- Control/Distributed/STM/STM.hs +44/−0
- Control/Distributed/STM/TVar.hs +400/−0
- Control/Distributed/STM/Utils.hs +39/−0
- DSTM.cabal +82/−0
- DSTMManual.pdf binary
- LICENSE +30/−0
- Nameserver/NameServer.hs +6/−0
- Phil/Phil.hs +84/−0
- README +53/−0
- Setup.hs +10/−0
+ Bomberman/Bomberman.hs view
@@ -0,0 +1,818 @@+{-# OPTIONS_GHC -XScopedTypeVariables -XDeriveDataTypeable #-}++module Main where++import Char+import Control.Concurrent+import Control.Exception as CE hiding (catch)+import qualified Control.Exception as CE (catch)+import Control.Distributed.STM.DebugDSTM+import Control.Distributed.STM.DSTM+import Maybe+import Network.BSD+import Prelude hiding (catch)+import System hiding (system)+import System.IO hiding (hPutStrLn)+import System.IO.Unsafe+import System.Posix++data Point = Point Int Int+ deriving (Eq, Show, Read)++instance Num Point where+ (Point x1 y1) + (Point x2 y2) = Point (x1 + x2) (y1 + y2)+ (Point x1 y1) - (Point x2 y2) = Point (x1 - x2) (y1 - y2)+ (Point x1 y1) * (Point x2 y2) = Point (x1 * x2) (y1 * y2)+ abs (Point x y) = Point (abs x) (abs y)+ signum (Point x y) = Point (signum x) (signum y)+ fromInteger x = Point (fromInteger x) (fromInteger x)++data Key = QKey + | AKey | MKey | SKey+ | UpKey | DownKey | RightKey | LeftKey+ | SpaceKey+ | DebugKey+ deriving (Eq, Show, Read)++point :: Int -> Int -> Point+point x y = Point x y+ +cWallSymbol :: Char; cWallSymbol = 'x'+cFSize :: Int; cFSize = 10+cDelay :: Int; cDelay = 2000000 -- 2 s; 50 ms granularity+cInitLifes :: Int; cInitLifes = 0+cInitPlayer :: Point; cInitPlayer = point 2 2+pI :: Point; pI = point 0 0+pR :: Point; pR = point 1 0+pL :: Point; pL = point (-1) 0+pU :: Point; pU = point 0 (-1)+pD :: Point; pD = point 0 1+fieldFile :: String; fieldFile = "./field.txt"+emptyImage :: String; emptyImage = " "+wallImage :: String; wallImage = "W "+playerImage :: String; playerImage = "o "+slaveImage :: String; slaveImage = "@ "+bombImage :: String; bombImage = ". "+blastImage :: String; blastImage = "X "+esc :: Char; esc = chr 27 +csi :: Char; csi = chr 91+upKey :: Char; upKey = chr 65+downKey :: Char; downKey = chr 66+rightKey :: Char; rightKey = chr 67+leftKey :: Char; leftKey = chr 68+aKeys :: [Char]; aKeys = ['a','A']+mKeys :: [Char]; mKeys = ['m','M']+sKeys :: [Char]; sKeys = ['s','S']+qKeys :: [Char]; qKeys = ['q','Q']+dKeys :: [Char]; dKeys = ['d','D']+spaceKey :: Char; spaceKey = ' '+initKeys :: [Char]; initKeys = aKeys++mKeys++sKeys+ctrlKeys :: [Char]; ctrlKeys = (esc:spaceKey:qKeys)++dKeys+cursKeys :: [Char]; cursKeys = [upKey,downKey,rightKey,leftKey]+clearScreen :: String; clearScreen = (esc:csi:"2J")+home :: String; home = (esc:csi:"H")+saveCursor :: String; saveCursor = (esc:csi:"s")+restoreCursor :: String; restoreCursor = (esc:csi:"u")+getCursor :: String; getCursor = (esc:csi:"6n")+midScreen :: String; midScreen = (esc:csi:"12;1H")++setCursor :: Point -> String+setCursor (Point x y) = (esc:csi:show (y+1))++(';':show ((x+1)*2))++"H"++line :: Int -> String+line l = (esc:csi:show l)++(";1H")++data Mode = AutonomM | MasterM String | SlaveM String+ deriving (Show, Read)++type Input = (Int, Handle)++data Element = Empty | Wall | Player | Slave | Bomb | Blast+ deriving (Show, Read, Eq)+type Field = [[Element]]+type Canvas = Field++type Bombs = [Point]+type Blasts = [Bombs]++data GameState = GameState {move :: TVar (Maybe Move),+ repaint :: TVar Bool,+ field :: TVar Field,+ lifes :: TVar Int,+ player :: TVar Point,+ slaves :: TVar [TVar Point],+ repaints :: TVar [TVar Bool],+ plBombs :: TVar Bombs,+ plBlasts :: TVar Blasts,+ plBCount :: TVar Int,+ bombs :: TVar [TVar Bombs],+ blasts :: TVar [TVar Blasts],+ bCounts :: TVar [TVar Int],+ quit :: TVar Bool,+ quits :: TVar [TVar Bool]}+ deriving (Show, Read)++data Move = MoveLeft | MoveRight | MoveUp | MoveDown+ | DropBomb | Dead+ deriving (Eq, Show, Read)++data View = View Field Point [Point] Bombs Blasts+ deriving Show++instance Dist Move where+ finTVars _ = return ()+ regTVars _ _ = return ()++instance Dist Point where+ finTVars _ = return ()+ regTVars _ _ = return ()++instance Dist Element where+ finTVars _ = return ()+ regTVars _ _ = return ()++gKeyCount :: MVar (Int, Int)+gKeyCount = unsafePerformIO (newMVar (0,0))++main :: IO ()+main = CE.catch (do+ debugStrLn6 "### bomberman main"+ putStr clearScreen+ hSetEcho stdin False+ hSetBuffering stdin NoBuffering+ args <- getArgs+ debugStrLn6 (show args)+ (mode, input) <- gameMode args+ debugStrLn6 (show mode)+ debugStrLn6 (show input)+ startDist (startGame mode input benchmark)+ --showTCPStat (midScreen++"quit")+ )(\(e::SomeException) -> error ("error in Bomberman: " ++ show e))++gameMode :: [String] -> IO (Mode, Input)+gameMode ((mode:_):[]) | elem mode aKeys = -- autonom + interactive+ return (AutonomM, (1, stdin))+gameMode ((mode:_):input:[]) | elem mode aKeys = -- autonom + file+ (openFile input ReadMode) >>= \h -> return (AutonomM, (1, h))+gameMode ((mode:_):[]) | elem mode mKeys = -- master + interactive+ return (MasterM gDefaultNameServer, (1, stdin))+gameMode ((mode:_):input:[]) | elem mode mKeys = -- master + file+ (openFile input ReadMode) >>= \h -> return (MasterM gDefaultNameServer, (1,h))+gameMode ((mode:_):delay:input:[]) | elem mode mKeys = do -- master+delay+file+ h <- (openFile input ReadMode)+ return (MasterM gDefaultNameServer, (read delay, h))+gameMode ((mode:_):[]) | elem mode sKeys = -- slave+interactive+ return (SlaveM gDefaultNameServer, (1, stdin))+gameMode ((mode:_):input:[]) | elem mode sKeys = -- slave + file+ (openFile input ReadMode) >>= \h -> return (SlaveM gDefaultNameServer, (1, h))+gameMode ((mode:_):delay:input:[]) | elem mode sKeys = do -- slave+delay+file+ h <- (openFile input ReadMode)+ return (SlaveM gDefaultNameServer, (read delay, h))+gameMode ((mode:_):input:nameServer:[]) | elem mode sKeys = -- s+f+nameserver+ (openFile input ReadMode) >>= \h -> return (SlaveM nameServer, (1, h))+gameMode ((mode:_):delay:input:nameServer:[]) | elem mode sKeys = do --s+d+f+n+ h <- (openFile input ReadMode)+ return (SlaveM nameServer, (read delay, h))+gameMode _ = getInitKey >>= \mode -> return (mode, (1, stdin)) -- interactive++benchmark :: IO Int -> Mode -> Input -> IO ()+benchmark action mode (delayStep, file) = do+ debugStrLn6 "### bomberman benchmark1"+ debugStrLn1 "### bomberman benchmark1"+ start <- getProcessTimes+ debugStrLn1 "### bomberman benchmark2"+ lostLifes <- action+ debugStrLn1 "### bomberman benchmark3"+ end <- getProcessTimes+ debugStrLn1 "### bomberman benchmark4"+ debugStrLn6 "### bomberman benchmark2"+ clockTick <- getSysVar ClockTick+ hostname <- getHostName+ (keysPressed, keysProcessed) <- readMVar gKeyCount+ let elapsed = toInteger (fromEnum ((elapsedTime end) - (elapsedTime start))) + user = toInteger (fromEnum ((userTime end) - (userTime start)))+ system = toInteger (fromEnum ((systemTime end) - (systemTime start))) + delay = toInteger delayStep * clockTick * toInteger keysPressed `div` 20+ filler = user + system + delay - elapsed+ putStrLn ("\nHost: " ++ hostname)+ putStrLn ("Input file: " ++ show file ++ "\n")+ putStr (" elapsed: " ++ showJustify elapsed 5 ++ " ticks ")+ putStrLn ("-> elapsed time total: "++ ticks2Sec elapsed clockTick)+ putStr ("= user: " ++ showJustify user 5 ++ " ticks ")+ putStrLn (" (" ++ show clockTick ++ " ticks/sec)")+ putStr ("+ system: " ++ showJustify system 5 ++ " ticks ")+ putStrLn ("-> user + system time: "++ ticks2Sec (user + system) clockTick)+ putStr ("+ delay: " ++ showJustify delay 5 ++ " ticks ") + putStrLn ("<- "++show keysPressed ++ "/" ++ show keysProcessed ++ " keys read/processed")+ putStr ("- filler: " ++ showJustify filler 5 ++ " ticks ") + putStrLn (" ("++show (delayStep*50)++" ms/key stroke)")+ putStr (showLifes lostLifes)+ putStrLn (" Mode: " ++ show mode ++ " ")++ticks2Sec :: Integer -> Integer -> String+ticks2Sec t ticksPerSec =+ let (sec, ticks) = divMod t ticksPerSec+ ms = ticks * 1000 `div` ticksPerSec+ in (show sec ++ "." ++ (concat [show ((ms `mod` (x * 10)) `div` x) | x <- [100,10,1]]) ++ " s")++showLifes :: Int -> String+showLifes l | l == 1 = " 1 life lost "+ | otherwise = showJustify (toInteger l) 3 ++ " lifes lost"++showJustify :: Show a => a -> Int -> String+showJustify i len = let s = show i in (replicate (len - length s) ' ') ++ s++launchGame :: Input -> GameState -> IO Int+launchGame input gameState = do+ debugStrLn6 ("### bomberman launchGame")+ showGameState gameState+ forkIO (view gameState) >> return ()+ forkIO (playerCtrl gameState) >> return ()+ debugStrLn1 ("### bomberman launchGame 0")+ processKeyboard gameState input+ debugStrLn1 ("### bomberman launchGame 1")+ atomic $ readTVar (lifes gameState)++startGame :: Mode -> Input -> (IO Int -> Mode -> Input -> IO ()) -> IO ()+startGame mode input bench = do+ debugStrLn6 "### bomberman startGame"+ debugStrLn6 (show mode)+ debugStrLn6 (show input)+ case mode of+ AutonomM -> bench (initAuto >>= launchGame input) mode input+ MasterM server -> do+ gameState <- initMaster server+ threadDelay cDelay+ bench (launchGame input gameState) mode input+ termMaster server+ threadDelay cDelay+ SlaveM server -> do+ gameState <- initSlave server+ threadDelay cDelay+ bench (launchGame input gameState) mode input+ threadDelay cDelay++initMaster :: String -> IO GameState+initMaster server = do + debugStrLn6 ("### bomberman initMaster")+ game <- initAuto+ atomic $ do + writeTVar (slaves game) [player game]+ registerTVar server (quits game) "QUITS"+ registerTVar server (field game) "FIELD"+ registerTVar server (slaves game) "SLAVES"+ registerTVar server (repaints game) "REPAINTS"+ registerTVar server (bombs game) "BOMBS"+ registerTVar server (blasts game) "BLASTS"+ registerTVar server (bCounts game) "BCOUNTS"+ return game++termMaster :: String -> IO ()+termMaster server = do+ debugStrLn6 ("### bomberman terminateMaster")+ deregisterTVar server "QUITS"+ deregisterTVar server "FIELD"+ deregisterTVar server "SLAVES"+ deregisterTVar server "REPAINTS"+ deregisterTVar server "BOMBS"+ deregisterTVar server "BLASTS"+ deregisterTVar server "BCOUNTS"++initSlave :: String -> IO GameState+initSlave server = CE.catch (do+ debugStrLn6 ("### bomberman initSlave")+ mQuits <- lookupTVar server "QUITS"+ let sQuits = fromJust mQuits+ mField <- lookupTVar server "FIELD"+ let sField = fromJust mField+ mSlaves <- lookupTVar server "SLAVES"+ let sSlaves = fromJust mSlaves+ mRepaints <- lookupTVar server "REPAINTS"+ let sRepaints = fromJust mRepaints+ mBombs <- lookupTVar server "BOMBS"+ let sBombs = fromJust mBombs+ mBlasts <- lookupTVar server "BLASTS"+ let sBlasts = fromJust mBlasts+ mBCounts <- lookupTVar server "BCOUNTS"+ let sBCounts = fromJust mBCounts+ myLives <- atomic $ newTVar cInitLifes+ newSlave <- atomic $ newTVar cInitPlayer+ sMove <- atomic $ newTVar Nothing+ myBombs <- atomic $ newTVar []+ myBlasts <- atomic $ newTVar []+ myBCount <- atomic $ newTVar 0+ myRepaint <- atomic $ newTVar True+ myQuit <- atomic $ newTVar False+ atomic $ do+ oldSlaves <- readTVar sSlaves+ writeTVar sSlaves (newSlave:oldSlaves)+ oldBombs <- readTVar sBombs+ writeTVar sBombs (myBombs:oldBombs)+ oldBlasts <- readTVar sBlasts+ writeTVar sBlasts (myBlasts:oldBlasts)+ oldBCounts <- readTVar sBCounts+ writeTVar sBCounts (myBCount:oldBCounts)+ otherRepaints <- readTVar sRepaints+ writeTVar sRepaints (myRepaint:otherRepaints)+ otherQuits <- readTVar sQuits+ writeTVar sQuits (myQuit:otherQuits)+ return (GameState sMove myRepaint sField myLives newSlave sSlaves sRepaints myBombs myBlasts myBCount sBombs sBlasts sBCounts myQuit sQuits)+ )(\(e::SomeException) -> putStrLn (midScreen ++ show e +++ " in initSlave from "++ show server +++ ". Starting auto game.") >> initAuto)++initAuto :: IO GameState+initAuto = do+ debugStrLn6 ("### bomberman initAuto")+ initField <- readField+ atomic $ do+ aMove <- newTVar Nothing+ aRepaint <- newTVar True+ aField <- newTVar initField+ aLives <- newTVar cInitLifes+ aPlayer <- newTVar cInitPlayer+ aSlaves <- newTVar []+ aRepaints <- newTVar [aRepaint]+ aPlBombs <- newTVar []+ aPlBlasts <- newTVar []+ aPLBCount <- newTVar 0+ aBombs <- newTVar [aPlBombs]+ aBlasts <- newTVar [aPlBlasts]+ aBCounts <- newTVar [aPLBCount]+ aQuit <- newTVar False+ aQuits <- newTVar [aQuit]+ return (GameState aMove aRepaint aField aLives aPlayer aSlaves aRepaints aPlBombs aPlBlasts aPLBCount aBombs aBlasts aBCounts aQuit aQuits)++readField :: IO Field+readField = do+ fieldStr <- readFile fieldFile+ return (map (map readEl) (lines fieldStr))++readEl :: Char -> Element+readEl c = if (c == cWallSymbol) then Wall else Empty++view :: GameState -> IO ()+view game = CE.catch (do+ debugStrLn6 ("### bomberman view")+ initField <- readField -- nur test+ debugStrLn6 ("### bomberman view 1")+ newView <- atomic $ do + b <- readTVar (repaint game)+ proc $ debugStrLn5 ("view 0")+ if b+ then do + writeTVar (repaint game) False++ f1 <- readTVar (field game)+ writeTVar (field game) initField+ writeTVar (field game) initField+ writeTVar (field game) f1+--+ p1 <- readTVar (player game)+ writeTVar (player game) cInitPlayer+ p2 <- readTVar (player game)+ writeTVar (player game) cInitPlayer+ writeTVar (player game) p1+ --}+ proc $ debugStrLn6 ("view 0")+ f <- readTVar (field game)+ proc $ debugStrLn6 ("view 1")+ p <- readTVar (player game)+ proc $ debugStrLn6 ("view 2")+ bsTVars <- readTVar (bombs game)+ allBombs <- mapM readTVar bsTVars+ proc $ debugStrLn6 ("view 3")+ blsTVars <- readTVar (blasts game)+ allBlasts <- mapM readTVar blsTVars+ proc $ debugStrLn6 ("view 4")+ slavesTVars <- readTVar (slaves game)+ proc $ debugStrLn6 ("view 5")+ slavesPts <- mapM readTVar slavesTVars+ proc $ debugStrLn6 ("view 6")+ return (View f p slavesPts (concat allBombs) (concat allBlasts))+ else retry+ debugStrLn6 ("### bomberman view 2")+ draw (makeCanvas newView)+ debugStrLn6 ("### bomberman view 3")+ printStatistics (midScreen)+ debugStrLn6 ("### bomberman view 4")+ view game+ )(\(e::SomeDistTVarException) -> do+ debugStrLn1 ("view catch "++show e)+ atomic $ do+ proc $ debugStrLn1 ("view catch atomic")+ cleanupSTM e game+ debugStrLn1 ("view catch dyn: <- ")+ view game+ )++cleanupSTM :: SomeDistTVarException -> GameState -> STM ()+cleanupSTM err game = do+ --proc $ debugStrLn1 ("--> cleanupSTM bsTVars")+ bsTVars <- readTVar (bombs game)+ let new = (removeErrTVars err bsTVars)+ --proc $ debugStrLn1 ("--> bsTVars "++ show bsTVars ++ show new)+ writeTVar (bombs game) (removeErrTVars err bsTVars)+ blsTVars <- readTVar (blasts game)+ --proc $ debugStrLn1 ("--> blsTVars "++ show blsTVars)+ writeTVar (blasts game) (removeErrTVars err blsTVars)+ bCtTVars <- readTVar (bCounts game)+ --proc $ debugStrLn1 ("--> bCtTVars "++ show bCtTVars)+ writeTVar (bCounts game) (removeErrTVars err bCtTVars)+ slavesTVars <- readTVar (slaves game)+ --proc $ debugStrLn1 ("--> slavesTVars "++ show slavesTVars)+ writeTVar (slaves game) (removeErrTVars err slavesTVars)+ repaintsTVars <- readTVar (repaints game)+ --proc $ debugStrLn1 ("--> repaintsTVars "++ show repaintsTVars)+ writeTVar (repaints game) (removeErrTVars err repaintsTVars)+ --proc $ debugStrLn1 ("<-- cleanupSTM bsTVars")++removeErrTVars :: SomeDistTVarException -> [TVar a] -> [TVar a]+removeErrTVars ex tVars = [tVar | tVar <- tVars, not (isDistErrTVar ex tVar)]++draw :: Canvas -> IO ()+draw canvas = do+ putStr saveCursor+ putStr home+ mapM_ (putStrLn.concatMap elToStr) canvas+ putStr restoreCursor++elToStr :: Element -> String+elToStr e = case e of+ Empty -> emptyImage+ Wall -> wallImage+ Player -> playerImage+ Slave -> slaveImage+ Bomb -> bombImage+ Blast -> blastImage++processKeyboard :: GameState -> Input -> IO ()+processKeyboard game input@(delayStep, file) = CE.catch (do+ debugStrLn6 ("### bomberman processKeyboard")+ end <- processEnd game+ if end+ then hClose file+ else do+ key <- if delayStep /= 0+ then do -- pause between key strokes in micros+ threadDelay (delayStep*50000) + oldMove <- atomic (readTVar (move game))+ let keyOk = if oldMove == Nothing then 1 else 0+ modifyMVar_ gKeyCount (\(key, ok) -> return (key + 1, ok+keyOk))+ getGameKey file+ else do+ atomic $ do+ oldMove <- readTVar (move game)+ if oldMove /= Nothing then retry else return ()+ modifyMVar_ gKeyCount (\(key, ok) -> return (key + 1, ok + 1))+ getGameKey file+ debugStrLn6 ("### bomberman Key= "++show key)+ case key of+ QKey -> atomic (writeTVar (quit game) True)+ >> debugStrLn1 ("### bomberman Key= "++show key)+ >> processKeyboard game input+ LeftKey -> atomic (writeTVar (move game) (Just MoveLeft)) + >> processKeyboard game input+ RightKey -> atomic (writeTVar (move game) (Just MoveRight))+ >> processKeyboard game input+ UpKey -> atomic (writeTVar (move game) (Just MoveUp))+ >> processKeyboard game input+ DownKey -> atomic (writeTVar (move game) (Just MoveDown))+ >> processKeyboard game input+ SpaceKey -> atomic (writeTVar (move game) (Just DropBomb))+ >> processKeyboard game input+ DebugKey -> startGDebug >> processKeyboard game input + _ -> processKeyboard game input + )(\(e::SomeDistTVarException) -> do+ debugStrLn1 ("processKeyboard catch "++show e)+ hClose file+ )++processEnd :: GameState -> IO Bool+processEnd game = CE.catch (do+ debugStrLn6 ("### bomberman processEnd")+ atomic $ do -- eof + 0 bombs + all nodes done -> end w/o busy waiting+ eof <- readTVar (quit game)+ if eof+ then retry+ else return False+ `orElse` do+ --proc $ debugStrLn1 ("processEnd 1 ")+ count <- getSTMBombCount game -- # of active bombs+ --proc $ debugStrLn1 ("processEnd 2 ")+ qTVars <- readTVar (quits game) -- list of quit status of all nodes+ proc $ debugStrLn1 ("processEnd 3 "++ show count)+ qs <- mapM readTVar qTVars+ proc $ debugStrLn1 ("processEnd 4 " ++ show qs)+ if (count == 0) && (and qs)+ then return True+ else retry+ )(\(e::SomeDistTVarException) -> do+ debugStrLn1 ("processEnd catch "++show e)+ atomic $ do+ qTVars <- readTVar (quits game)+ writeTVar (quits game) (removeErrTVars e qTVars)+ cTVars <- readTVar (bCounts game)+ writeTVar (bCounts game) (removeErrTVars e cTVars)+ processEnd game+ )++getSTMBombCount :: GameState -> STM Int+getSTMBombCount game = do -- # of active bombs+ countTVars <- readTVar (bCounts game)+ counts <- mapM readTVar countTVars+ return (foldr (+) 0 counts)++getGameKey :: Handle -> IO Key+getGameKey input = do+ c <- hGetChar input+ if (elem c ctrlKeys)+ then if (c == esc)+ then do+ c1 <- hGetChar input+ if (c1 == csi)+ then do+ c2 <- hGetChar input+ if (elem c2 cursKeys)+ then return (cursorKey c2)+ else getGameKey input+ else getGameKey input+ else return (controlKey c)+ else getGameKey input++cursorKey :: Char -> Key+cursorKey c = + if (c == upKey) then UpKey+ else if (c == downKey) then DownKey+ else if (c == leftKey) then LeftKey+ else if (c == rightKey) then RightKey+ else error ("error wrong char '"++(c:"' in cursorKey"))++controlKey :: Char -> Key+controlKey c = + if (elem c qKeys) then QKey+ else if (c == spaceKey) then SpaceKey+ else if (elem c dKeys) then DebugKey+ else error ("error wrong char '"++(c:"' in controlKey"))++getInitKey :: IO Mode+getInitKey = do+ putStr clearScreen+ putStr (line 15)+ putStrLn"Bitte waehlen: Autonom (A), Master (M), Slave (S)"+ hSetEcho stdin False+ hSetBuffering stdin NoBuffering+ c <- getChar+ if (elem c initKeys)+ then initMode c+ else return AutonomM++initMode :: Char -> IO Mode+initMode c = + if (elem c aKeys) then return AutonomM+ else if (elem c mKeys) then do+ server <- getNameServer+ return (MasterM server) + else if (elem c sKeys) then do + server <- getNameServer+ return (SlaveM server) + else error ("error wrong char '"++(c:"' in initMode"))++getNameServer :: IO String+getNameServer = do+ hSetEcho stdin True+ putStrLn "Masterserver [default: localhost]: "+ s <- getLine+ hSetEcho stdin False+ return (if (s == []) then gDefaultNameServer else s)++playerCtrl :: GameState -> IO ()+playerCtrl game = CE.catch (do+ debugStrLn6 ("### bomberman playerCtrl")+ -- get new move if available orElse check if dead on explosion+ m <- atomic $ getSTMCommand (move game) -- get move or retry+ `orElse` do -- check active explosions+ expls <- getSTMBombCount game -- # of active bombs+ if expls > 0+ then do+ p <- readTVar (player game)+ blsTVars <- readTVar (blasts game)+ allBlasts <- mapM readTVar blsTVars+ -- check player explosion+ if elem p ((concat.concat) allBlasts)+ then return Dead -- player exploded+ else retry -- wait for next move+ else retry -- wait for next move+ debugStrLn6 ("# case m "++show m)+ case m of -- make players move+ DropBomb -> forkIO (dropBomb game) >> return ()+ Dead -> atomic $ do+ p <- readTVar (player game)+ blsTVars <- readTVar (blasts game)+ allBlasts <- mapM readTVar blsTVars+ if elem p ((concat.concat) allBlasts) + then retry+ else modifyTVar (lifes game) (+1)+ _ -> atomic $ do+ f <- readTVar (field game)+ p <- readTVar (player game)+ let newPlayer = case m of+ MoveLeft -> legalPos f p pL+ MoveRight -> legalPos f p pR+ MoveUp -> legalPos f p pU+ MoveDown -> legalPos f p pD+ _ -> legalPos f p pI+ writeTVar (player game) newPlayer + setSTMRepaintAll (repaints game) + debugStrLn6 ("# end loop")+ playerCtrl game+ )(\(e::SomeDistTVarException) -> do+ debugStrLn1 ("playerCtrl catch "++show e)+ atomic $ do+ proc $ debugStrLn1 ("playerCtrl catch atomic: ")+ cleanupSTM e game+ debugStrLn1 ("playerCtrl catch dyn ---: ")+ --bs <- atomic $ showSTMRepaintAll (repaints game)+ --debugStrLn1 ("reps " ++ show bs)+ atomic $ setSTMRepaintAll (repaints game) + debugStrLn1 ("playerCtrl catch dyn <-: ")+ playerCtrl game+ )++getSTMCommand :: TVar (Maybe Move) -> STM Move+getSTMCommand mTVar = do+ m <- readTVar mTVar+ proc $ debugStrLn2 ("getSTMCommand: "++(show m))+ if (m == Nothing)+ then retry + else do+ writeTVar mTVar Nothing+ return (fromJust m)++legalPos :: Field -> Point -> Point -> Point+legalPos f p moveP = if (isLegal newPos f) then newPos else p+ where newPos = p + moveP++inBoundsPos :: Point -> Point -> Point+inBoundsPos p moveP = if (isInBounds newPos) then newPos else p+ where newPos = p + moveP++isLegal :: Point -> Field -> Bool+isLegal p f = isInBounds p && ((f `at` p) /= Wall)++isInBounds :: Point -> Bool+isInBounds (Point x y) = x >= 0 && x < cFSize && y >= 0 && y < cFSize+ +dropBomb :: GameState -> IO ()+dropBomb game = do+ debugStrLn6 ("### bomberman dropBomb")+ debugStrLn6 "# add new bomb@currPos to list"+ pos <- atomic $ do -- add new bomb@currPos to player bombs list+ p <- readTVar (player game)+ modifyTVar (plBombs game) (p:)+ return p+ debugStrLn6 "# set repaint"+ saveSetRepaintAll game -- mark changes to show+ timeBomb game pos++timeBomb :: GameState -> Point -> IO ()+timeBomb game pos = CE.catch (do -- catch bombs from dropped players+ debugStrLn6 "# init sema wake"+ wake <- atomic $ newTVar False+ debugStrLn6 "# start sleep counter wake"+ forkIO (threadDelay cDelay >> atomic (writeTVar wake True)) -- start timer+ debugStrLn6 "# wait for wake counter orelse active explosions"+ atomic $ do+ w <- readTVar wake+ if w + then return () -- timeout+ else retry -- check active explosion @ pos -> new bomb explodes also+ `orElse` do+ expls <- getSTMBombCount game -- # of active bombs+ if expls > 0 + then do -- sync with other expls+ blsTVars <- readTVar (blasts game)+ allBlasts <- mapM readTVar blsTVars+ if elem pos ((concat.concat) allBlasts) + then return () -- dont wait for timeout+ else retry -- wait for timeout+ else retry -- wait for timeout+ debugStrLn6 "# end bomb"+ blastBomb game pos+ )(\(e::SomeDistTVarException) -> do -- clean up and dont wait for timeout+ debugStrLn1 ("timeBomb catch "++show e)+ atomic $ do+ proc $ debugStrLn1 ("timeBomb catch atomic: ")+ cleanupSTM e game+ debugStrLn1 ("timeBomb catch dyn ---: ")+ blastBomb game pos+ )++blastBomb :: GameState -> Point -> IO ()+blastBomb game pos = do+ debugStrLn6 ("### bomberman blastBomb")+ debugStrLn6 "# remove bomb@pos, add new blasts, clear field@blast, inc #blast"+ newBls <- atomic $ do+ modifyTVar (plBombs game) (filter (/=pos)) --remove new bomb from bombs list+ let newBls = neighbours pos+ modifyTVar (plBlasts game) (newBls:) -- add new blasts to blasts lists+ modifyTVar (field game) (updateField Empty newBls) -- remove bombed walls+ modifyTVar (plBCount game) (+1)+ return newBls+ saveSetRepaintAll game -- mark changes to show+ debugStrLn6 "# delay"+ threadDelay cDelay+ debugStrLn6 "# remove new blast, dec #blast"+ atomic $ do+ modifyTVar (plBlasts game) (filter (/=newBls)) -- remove new blast+ modifyTVar (plBCount game) (+ (-1))+ saveSetRepaintAll game -- mark changes to show+ debugStrLn6 "# end bomb"++makeCanvas :: View -> Canvas+makeCanvas v =+ [ [canvasEl v (Point x y) | x <- [0..cFSize-1] ] | y <- [0..cFSize-1] ]++canvasEl :: View -> Point -> Element+canvasEl (View vField vPlayer vSlaves vBombs vBlasts) p =+ if elem p (concat vBlasts)+ then Blast+ else if (vPlayer == p) + then Player+ else if elem p vSlaves+ then Slave+ else if elem p vBombs+ then Bomb+ else vField `at` p++updateField :: Element -> [Point] -> Field -> Field+updateField el ps fd =+ [ [ f x y | x <- [0..cFSize-1] ] | y <- [0..cFSize-1] ]+ where f x y = if (elem (Point x y) ps) + then el else fd `at` (Point x y)+{- [bVec | y <- [0..cFSize-1],+ let bVec = [bEl | x <- [0..cFSize-1],+ let bEl = if (elem (Point x y) ps) + then el + else fd `at` (Point x y)]]+-}++neighbours :: Point -> [Point]+neighbours p = p:map (inBoundsPos p) [pL,pR,pU,pD]++at :: Field -> Point -> Element+canvas `at` (Point x y) = (canvas !! y) !! x++modifyTVar :: Dist a => TVar a -> (a -> a) -> STM ()+modifyTVar t f = do+ v <- readTVar t+ writeTVar t (f v)++setSTMRepaintAll :: TVar [TVar Bool] -> STM ()+setSTMRepaintAll allRepaints = do+ repaintsTVars <- readTVar allRepaints+ mapM_ (flip writeTVar True) repaintsTVars++saveSetRepaintAll :: GameState -> IO ()+saveSetRepaintAll game = CE.catch (atomic $ do+ repaintsTVars <- readTVar (repaints game)+ mapM_ (flip writeTVar True) repaintsTVars+ )(\(e::SomeDistTVarException) -> do+ debugStrLn1 ("saveSetRepaintAll catch "++show e)+ atomic $ do+ proc $ debugStrLn1 ("saveSetRepaintAll catch atomic")+ cleanupSTM e game+ debugStrLn1 ("saveSetRepaintAll catch dyn: --- ")+ saveSetRepaintAll game+ debugStrLn1 ("saveSetRepaintAll catch dyn: <- ")+ )++showSTMRepaintAll :: TVar [TVar Bool] -> STM [Bool]+showSTMRepaintAll allRepaints = do+ proc $ debugStrLn6 ("showSTMRepaintAll")+ repaintsTVars <- readTVar allRepaints+ proc $ debugStrLn6 ("showSTMRepaintAll "++show repaintsTVars)+ bs <- mapM readTVar repaintsTVars+ proc $ debugStrLn6 ("showSTMRepaintAll end "++show bs)+ return bs++showGameState :: GameState -> IO ()+showGameState g = atomic $ do + mv <- readTVar (move g)+ proc $ debugStrLn6 ("TVar move= "++show (move g)++" move= "++show mv)+ rp <- readTVar (repaint g)+ proc $ debugStrLn6 ("TVar repaint= "++show (repaint g)++" repaint= "++show rp)+ fd <- readTVar (field g)+ proc $ debugStrLn6 ("TVar field= "++show (field g)++" field= "++show fd)+ pl <- readTVar (player g)+ proc $ debugStrLn6 ("TVar player= "++show (player g)++" player= "++show pl)+ sl <- readTVar (slaves g)+ proc $ debugStrLn6 ("TVar slaves= "++show (slaves g)++" slaves= "++show sl)+ rs <- readTVar (repaints g)+ proc $ debugStrLn6 ("TVar repaints= "++show (repaints g)++" repaints= "++show rs)+ bs <- readTVar (bombs g)+ proc $ debugStrLn6 ("TVar bombs= "++show (bombs g)++" bombs= "++show bs)+ bl <- readTVar (blasts g)+ proc $ debugStrLn6 ("TVar blasts= "++show (blasts g)++" blasts= "++show bl)+ bc <- readTVar (bCounts g)+ proc $ debugStrLn6 ("TVar bCounts= "++show (bCounts g)++" bCount= "++show bc)
+ Bomberman/field.txt view
@@ -0,0 +1,10 @@+xxxxxxxxxx+xoooooooox+xooxxxxoox+xooxxxxoox+xooxxxxoox+xooxxxoxox+xoooxoooox+xoooxoooox+xoooooooox+xxxooxxxxx
+ Bomberman/input1.txt view
@@ -0,0 +1,1 @@+ [B [B [B [B [B [B [C [C[C[C[C[C[C[A[A[A[A[A[A[A[A[B[B[B[B[B[B[B[B[A[A[A[A[A[A[A[A[A[B[B[B[B[B[B[B[B[B[A[A[A[A[A[A[A[A[B[B[B[B[B[B[B[B[A[A[A[A[A[A[A[A[A[B[B[B[B[B[B[B[B[B[A[A[A[A[A[A[AQ
+ Bomberman/input2.txt view
@@ -0,0 +1,1 @@+[B[B[B[B[B[B[C[C[C[C[C[C[C[C [A [A [A [A [A [A [A[D[D[D[D[D[D[D[D[D[C[C[C[C[C[C[D[D[D[D[D[D[D[D[C[C[C[C[C[C[D[D[D[D[D[D[D[D[D[D[C[C[C[C[C[C[D[D[D[D[D[D[D[D[C[C[C[C[C[C[D[D[D[D[D[D[D[DQ
+ Bomberman/input3.txt view
@@ -0,0 +1,1 @@+[B[B[B[B[B[B[B[B [C [C [C [C [C [C [A[A[A[A[A[A[A[A[A[D[D[D[D[D[D[D[D[C[C[C[C[D[D[D[D[D[C[C[C[C[C[C[D[D[D[D[D[D[D[D[C[C[C[C[C[D[D[D[D[D[D[C[C[C[C[C[C[D[D[D[D[D[D[D[D[C[C[C[C[C[D[D[D[D[D[D[D[D[C[C[C[C[C[C[D[D[D[D[D[D[D[D[C[C[C[C[C[D[D[D[D[D[D[D[C[C[C[C[C[D[D[D[D[D[D[D[C[C[C[C[C[C[D[D[D[D[D[D[D[C[C[C[C[C[C[D[D[D[D[D[D[D[D[D[C[C[C[C[C[D[D[D[D[D[D[D[C[C[C[C[C[D[D[D[D[D[D[C[C[C[C[C[D[D[D[D[D[D[DQ
+ Chat/ChatClient.hs view
@@ -0,0 +1,45 @@+module Main where++import ChatData+import Control.Concurrent+import Control.Distributed.STM.DSTM+import Maybe+import System.IO++main :: IO ()+main = startDist $ do+ putStrLn "Your Name: "+ name <- getLine+ serverTVar <- lookupTVar "localhost" "Chat"+ case serverTVar of + Nothing -> putStrLn "Chatserver not reachable"+ Just cmdTVar -> do+ myTVar <- atomic $ do+ new <- newTVar Nothing+ writeTVar cmdTVar (Just (Join name new))+ return new+ forkIO (serverClient myTVar)+ stdinClient name cmdTVar++stdinClient :: String -> CmdTVar -> IO ()+stdinClient name cmdTVar = do+ putStrLn (name ++ " >")+ msg <- getLine+ if msg == "bye"+ then atomic $ writeTVar cmdTVar (Just (Leave name))+ else do+ atomic $ writeTVar cmdTVar (Just (Msg name msg))+ stdinClient name cmdTVar++serverClient :: CmdTVar -> IO ()+serverClient myTVar = do+ s <- atomic $ do+ cmd <- readTVar myTVar+ case cmd of+ Nothing -> retry+ Just (Msg name msg) -> do+ writeTVar myTVar Nothing+ return (name ++ ": " ++ msg)+ _ -> return ""+ putStrLn s+ serverClient myTVar
+ Chat/ChatServer.hs view
@@ -0,0 +1,46 @@+module Main where++import ChatData+import Control.Distributed.STM.DebugDSTM+import Control.Distributed.STM.DSTM+import Control.Exception as CE+import Maybe++main :: IO ()+main = startDist $ do+ inVar <- atomic $ newTVar Nothing+ registerTVar "localhost" inVar "Chat"+ chatServer inVar []++chatServer :: CmdTVar -> [(String, CmdTVar)] -> IO ()+chatServer inCmd dict = CE.catch (do+ newDict <- atomic $ do+ cmd <- readTVar inCmd+ case cmd of+ Nothing -> do+ mapM_ (readTVar.snd) dict+ retry+ Just serverCmd -> do+ writeTVar inCmd Nothing+ case serverCmd of+ Join name msgVar -> do+ proc $ putStrLn (name++" joint the chat")+ mapM_ (flip writeTVar msg.snd) dict+ return ((name,msgVar):dict)+ where msg = Just (Msg name " joint")+ Msg _ _ -> do+ proc $ putStrLn "Message forwarded"+ mapM_ (flip writeTVar cmd.snd) dict+ return dict+ Leave name -> do+ proc $ putStrLn (name++" left the chat")+ mapM_ (flip writeTVar msg.snd) dic+ return dic+ where msg = Just (Msg name " left")+ dic = filter ((/=name).fst) dict+ chatServer inCmd newDict+ )(\e -> print "Dropped Client" >> chatServer inCmd (removeErrDict e dict))++removeErrDict :: SomeDistTVarException -> [(String, CmdTVar)] + -> [(String, CmdTVar)]+removeErrDict e dict = [d | d <- dict, not (isDistErrTVar e (snd d))]
+ Control/Distributed/STM/DSTM.hs view
@@ -0,0 +1,684 @@+{-# OPTIONS_GHC -XScopedTypeVariables #-}+{-# OPTIONS_GHC -fno-warn-unused-do-bind #-}++module Control.Distributed.STM.DSTM (TVar, STM, newTVar, readTVar, writeTVar,+ atomic, retry, orElse, throw, catch, + module Control.Distributed.STM.Dist,+ module Control.Distributed.STM.NameService,+ startDist, SomeDistTVarException, isDistErrTVar) where++import Control.Concurrent+import Control.Distributed.STM.DebugDSTM+import Control.Distributed.STM.Dist+import Control.Distributed.STM.EnvAddr+import Control.Distributed.STM.NameService -- just for export+import Control.Distributed.STM.RetryVar+import Control.Distributed.STM.STM+import Control.Distributed.STM.TVar+import Control.Distributed.STM.Utils+import Control.Exception hiding (catch, throw)+import qualified Control.Exception as CE (catch, throw)+import Control.Monad+import qualified Data.List+import Network+import Network.Socket hiding (accept)+import Prelude as P hiding (catch)+import System.IO+import System.IO.Unsafe+import System.Mem+import System.Process (ProcessHandle, runCommand, terminateProcess)+import System.Posix++infixl 2 `orElse`++------------------+-- TVar actions --+------------------++gInitVersionID :: ID+gInitVersionID = 1+ +-- | Create a new TVar holding a value supplied+newTVar :: Dist a => a -> STM (TVar a)+newTVar v = STM (\stmState -> do+ newID <- uniqueId+ newRef <- newMVar (v, gInitVersionID)+ newLock <- newMVar ()+ waitQ <- newMVar []+-- newDebugMVar enables later MVar inspection+-- newRef <- newDebugMVar ("Ref: TVar "++show newID) (v, gInitVersionID)+-- newLock <- newDebugMVar ("Lock: TVar "++show newID) ()+-- waitQ <- newDebugMVar ("WaitQ: TVar "++show newID) []+ let tVar = (TVar newRef newID newLock waitQ)+ return (Success stmState tVar))+ +-- |Return the current value stored in a TVar+readTVar :: Dist a => TVar a -> STM a+readTVar tVar = STM (\stmState -> do+ (val, vId) <- readIntraTransTVar tVar (stmCommit stmState)+ let newState = stmState+ {stmValid = bundledValidLogs tVar+ (Just (vId, (stmRetryVar stmState)))+ (stmValid stmState)+ }+ return (Success newState val))++-- check if uncommitted transaction-local writes precede this read+readIntraTransTVar :: Dist a => TVar a -> [CommitLog] -> IO (a, VersionID)+readIntraTransTVar tVar commitLog =+ case P.lookup env commitLog of+ Just (Left commitActs) ->+ case P.lookup tId commitActs of+ Just acts -> readHost tVar $ commitAct acts+ Nothing -> coreReadTVar tVar+ Just (Right (strVals, _)) ->+ case P.lookup tId strVals of+ Just v -> coreReadTVar tVar >>= return . ((,) (read v)) . snd+ Nothing -> coreReadTVar tVar+ Nothing -> coreReadTVar tVar+ where VarLink env tId = tVarToLink tVar++coreReadTVar :: Dist a => TVar a -> IO (a,VersionID)+coreReadTVar (TVar tVarRef _ lock _) = do+ takeMVar lock -- read lock to prevent read before finishing remote commit+ v <- readMVar tVarRef + putMVar lock ()+ debugStrLn8 ("coreReadTVar Loc "++show v)+ return v+coreReadTVar (LinkTVar (VarLink tEnv tVarId)) = CE.catch (do + answer <- remGetMsg tEnv (RemReadTVar tVarId gMyEnv)+ let vv@(v,_) = read answer+ finTVars v+ debugStrLn8 ("coreReadTVar Rem "++show vv)+ return vv+ )(propagateEx "coreReadTVar")++-- read uncommitted transaction-local writes by keeping original version id+readHost :: forall a . Read a => TVar a -> IO () -> IO (a, VersionID)+ -- explicit forall a extends scope of a into function body+readHost (TVar tVarRef _ lock _) commit = do+ takeMVar lock+ orig@(_, origVersion) <- readMVar tVarRef+ commit+ (modV, _) <- swapMVar tVarRef orig+ putMVar lock ()+ return (modV, origVersion)+readHost (LinkTVar (VarLink _ tId)) commit = do+ lockTVarFromId tId+ origStr <- readTVarValFromId tId+ let (_::a, origVersion) = read origStr+ commit+ (modV, _::VersionID) <- liftM read (swapTVarValFromId tId origStr)+ unLockTVarFromId tId+ return (modV, origVersion)++-- |Write the supplied value into a TVar+writeTVar :: Dist a => TVar a -> a -> STM ()+writeTVar tVar v = STM (\stmState -> + let newState = stmState{+ stmValid = bundledValidLogs tVar Nothing (stmValid stmState),+ stmCommit = bundledCommitLogs tVar v (stmCommit stmState)+ }+ in debugStrLn8 ("writeTVar "++show (stmCommit newState))>>return (Success newState ()))++----------------------+-- STM computations --+----------------------++-- |Perform a series of STM actions atomically+atomic :: Show a => STM a -> IO a+atomic stmAction = do+ aStat ATM+ debugStrLn7 ("A")+ iState <- initialState+ debugStrLn7 ("I")+ stmResult <- runSTM stmAction iState+ debugStrLn7 ("M")+ case stmResult of+ Exception newState e -> do + debugStrLn7 ("E")+ let gState = gatherStmState newState+ valid <- startTrans gState+ if valid+ then do+ endTrans gState+ CE.throw e -- propagate Exception into IO world+ else do+ endTrans gState+ atomic stmAction+ Retry newState -> do+ debugStrLn7 ("R")+ let gState = gatherStmState newState+ valid <- startTrans gState+ if valid+ then do+ debugStrLn7 ("v")+ retryTrans gState -- including set auto-link+ endTrans gState+ let retryVar = stmRetryVar gState+ insertRetryVarAction retryVar+ debugStrLn7 ("SUSPEND "++show (stmId gState))+ suspend retryVar+ debugStrLn7 ("RESUME "++show (stmId gState))+ deleteRetryVarAction retryVar+ unRetryTrans gState -- reset auto-link+ else do+ debugStrLn7 ("i")+ endTrans gState+ aStat INV+ atomic stmAction+ Success newState res -> do+ debugStrLn7 ("S")+ let gState = gatherStmState newState+ valid <- startTrans gState+ if valid+ then do+ debugStrLn7 ("v")+ commitTrans gState+ endTrans gState+ return res+ else do+ debugStrLn7 ("i")+ endTrans gState+ aStat INV+ atomic stmAction++initialState :: IO STMState+initialState = do+ atomicId <- uniqueId+ retryVar <- newRetryVar+ return (STMState {stmId = (gMyEnv, atomicId),+ stmRetryVar = retryVar,+ stmValid = [],+ stmCommit = []})++-- |Retry execution of the current memory transaction because it has seen +-- values in TVars which mean that it should not continue (e.g. the TVars+-- represent a shared buffer that is now empty). The implementation may block+-- the thread until one of the TVars that it has read from has been udpated.+retry :: STM a+retry = STM (\state -> return (Retry state))++-- |Compose two alternative STM actions. If the first action completes without+-- retrying then it forms the result of the orElse. Otherwise, if the first+-- action retries, then the second action is tried in its place. If both+-- actions retry then the orElse as a whole retries+orElse :: STM a -> STM a -> STM a+orElse (STM stm1) (STM stm2) =+ STM (\(stmState@STMState{stmCommit = savedCont}) -> do+ stm1Res <- stm1 stmState+ case stm1Res of+ Retry newState -> stm2 newState{stmCommit = savedCont}+ _ -> return stm1Res)++----------------+-- Exceptions --+----------------++-- |Throw an exception within an STM action+throw :: SomeException -> STM a+throw e = STM (\state -> return (Exception state e))++-- |Exception handling within STM actions+catch :: STM a -> (SomeException -> STM a) -> STM a+catch (STM stm) eHandler = STM (\stmState -> do+ res <- stm stmState+ case res of+ Exception _ e -> do+ let (STM stmEx) = eHandler e+ stmEx stmState+ _ -> return res)+++-------------------------+-- Perform STM Actions --+-------------------------++-- copy WriteVals from CommitVals to ValidVals at end of transaction because of+-- possible nested orElse-scopes where locks and validations are preserved+-- while commits are reset using simple push/pop in orElse stack +-- finally all info is needed in ValidVals to be transmitted to all trans+-- to ensure proper recovery in 3-phase model++gatherStmState :: STMState -> STMState+gatherStmState state = state {stmValid =+ map (gatherValidRemVal (stmCommit state)) (stmValid state)}++gatherValidRemVal :: [CommitLog] -> ValidLog -> ValidLog+gatherValidRemVal _ (env, Left stActs) = (env, Left stActs)+gatherValidRemVal comLogs vLog@(env, Right validRemVals) =+ case P.lookup env comLogs of+ Nothing -> vLog+ Just (Right (v, _)) -> (env, Right(map (addCommitRemVals v) validRemVals))+ _ -> vLog -- internal error++addCommitRemVals :: [CommitRemVal] -> ValidRemVal -> ValidRemVal+addCommitRemVals comVals (tId,(rVal, _)) = (tId, (rVal, P.lookup tId comVals))++-------------------+startTrans :: STMState -> IO Bool+startTrans state = do+ robustFoldValidAct (stmId state) (fst.unzip $ valLogs) True valLogs+ where valLogs = stmValid state++robustFoldValidAct :: TransID -> [EnvAddr] -> Bool -> [ValidLog] -> IO Bool+robustFoldValidAct _ _ isValid [] = return isValid+robustFoldValidAct transId envs isValid (vLog:vLogs) = CE.catch (do+ debugStrLn5 ("robustFoldValidAct: x= "++show vLog++"xs= "++show vLogs)+ b <- doValidAction isValid vLog transId envs+ robustFoldValidAct transId envs b vLogs+ )(\(e::SomeDistTVarException) -> do+ debugStrLn1 (">>> robustFoldValidAct -> error: " ++ show e)+ debugStrLn1 (">>> robustFoldValidAct dyn: "++ show (distTVarExEnv e))+ CE.catch (doEndAction transId vLog) + (\(_::SomeDistTVarException) -> return ())+ propagateEx "robustFoldValidAct _" e+ )++doValidAction :: Bool -> ValidLog -> TransID -> [EnvAddr] -> IO Bool+doValidAction isValid (_, Left (lockActs, readAct)) _ _ = do+ mapM_ (lockAct.snd) lockActs+ isValid +>> validAct readAct+doValidAction isValid (env, (Right idVRs)) transId transEnvs =+ remGetMsg env (RemStartTrans transId idVRs transEnvs)+ >>= return . (isValid &&) . read+ -- possible opt: lazy val. msg -> requires skipping recovery on lazy val.++commitTrans :: STMState -> IO ()+commitTrans state = robustMapM_ (doCommitAction (stmId state)) (stmCommit state)++doCommitAction :: TransID -> CommitLog -> IO ()+doCommitAction _ (env, Left idCommits) = do+ debugStrLn5 ("doCommitAction: env= "++show env++"ids= "++show idCommits)+ mapM_ (commitAct.snd) idCommits+ mapM_ (notifyAct.snd) idCommits+doCommitAction transId (env, Right (_, regAct)) = do+ regAct -- register dest env on TVars within values+ remPutMsg env (RemContTrans transId Com)++retryTrans :: STMState -> IO ()+retryTrans state = robustMapM_ (doRetryAction (stmId state)) (stmValid state)++doRetryAction :: TransID -> ValidLog -> IO ()+doRetryAction _ (env, Left (_, valLog)) = do+ debugStrLn5 ("doRetryAction: env= "++show env)+ extWaitQAct valLog+doRetryAction transId (env, Right idrs) = do+ remPutMsg env (RemContTrans transId Ret)+ -- insert auto link or update existing (retry or auto) link with retryVar+ insertRetryLinks env idrs++unRetryTrans :: STMState -> IO ()+unRetryTrans state = mapM_ doUnRetryAction (stmValid state)++doUnRetryAction :: ValidLog -> IO ()+doUnRetryAction (_, Left _) = return ()+doUnRetryAction (env, Right idrs) = do+ deleteRetryLinks env idrs+ debugStrLn5 ("makeUnWaitQAction: env= "++show env++"idrs= "++show idrs)++endTrans :: STMState -> IO ()+endTrans state = robustMapM_ (doEndAction (stmId state)) (stmValid state)++doEndAction :: TransID -> ValidLog -> IO ()+doEndAction _ (_, Left stLog) = do+ mapM_ (unLockAct.snd) (fst stLog)+doEndAction transId (env, Right _) = do+ remPutMsg env (RemContTrans transId End)++robustMapM_ :: Show a => (a -> IO ()) -> [a] -> IO ()+robustMapM_ _ [] = return ()+robustMapM_ io (x:xs) = CE.catch (do+ debugStrLn5 ("robustMapM_: x= "++show x++"xs= "++show xs)+ io x+ robustMapM_ io xs+ )(\(ex::SomeDistTVarException) -> do+ debugStrLn1 (">>> robustMapM_ -> error: " ++ show ex)+ debugStrLn1 (">>> robustMapM_ dyn: "++ show (distTVarExEnv ex))+ robustMapM_ io xs -- ignore action x and continue w/actions xs, w/o prop.!+ )++---------------------------------+-- Remote Transactional Status --+---------------------------------++type DistTransCont = (TransID, Chan RemCont)++-- state of received remote transactions needed for possible recovery+gDistTransCont :: MVar [DistTransCont]+{-# NOINLINE gDistTransCont #-}+gDistTransCont = unsafePerformIO (newMVar [])++startRemTrans :: TransID -> [ValidRemVal] -> + [EnvAddr] -> (Bool -> IO ()) -> IO ()+startRemTrans transId idVRVars transEnvs notifyCaller = do+ updateAutoTrans (+1) transId+ msgChan <- newChan+ modifyMVar_ gDistTransCont (return . ((transId, msgChan):))+ forkIO (CE.catch+ (ctrlTrans msgChan transId idVRVars transEnvs notifyCaller)+ (\(e::SomeException) -> debugStrLn1 ("startRemTrans " ++ show e)))+ return ()++contRemTrans :: TransID -> RemCont -> IO ()+contRemTrans transId msg = do+ debugStrLn8 ("<<< contRemTrans transId "++show transId)+ conts <- readMVar gDistTransCont+ case P.lookup transId conts of+ Just msgChan -> do+ debugStrLn8 ("<<< contRemTrans : transId = "++show transId)+ writeChan msgChan msg+ debugStrLn8 ("<<< contRemTrans : writeChan msg = "++show msg)+ Nothing -> -- return () -- no error, possible duplicate recovery msgs+ debugStrLn8 ("<<< contRemTrans N RemCont= "++show msg)++ctrlTrans :: Chan RemCont -> TransID -> [ValidRemVal] -> + [EnvAddr] -> (Bool -> IO ()) -> IO ()+ctrlTrans msgChan transId idVRVars transEnvs notifyCaller = do+ -- start remote transaction (phase 1)+ printValidFromId "ctrlTrans" transId idVRVars+ debugStrLn8 ("<<< ctrlTrans lock transId "++show transId ++ " idVRVars "++ show idVRVars)+ mapM_ (lockTVarFromId.fst) idVRVars+ debugStrLn8 ("### ctrlTrans locked transId= "++show transId)+ isValid <- foldr validateValidIds (return True) idVRVars+ CE.catch (notifyCaller isValid) + (\(e::SomeException) -> do+ debugStrLn1 $ "<<< ctrlTransFromIds !!! catch= " ++ show e+ writeChan msgChan Err+ )+ ctrlContTrans msgChan transId idVRVars transEnvs++ctrlContTrans :: Chan RemCont -> TransID -> [ValidRemVal] -> + [EnvAddr] -> IO ()+ctrlContTrans msgChan transId idVRVars transEnvs = do+ -- continue remote transaction (phase 2)+ debugStrLn8 ("> ctrlContTrans readChan ")+ msg <- readChan msgChan+ debugStrLn8 ("< ctrlContTrans readChan ")+ case msg of+ Com -> do -- decision for Com+ debugStrLn8 ("<<< ctrlContTrans: Com transId = "++show transId)+ mapM_ contWriteTVar idVRVars+ mapM_ (notifyFromId.fst) idVRVars+ ctrlEndTrans msgChan transId Com idVRVars transEnvs + Ret -> do -- decision for Ret+ debugStrLn8 ("<<< ctrlContTrans: Ret transId = "++show transId)+ mapM_ contExtWaitQ idVRVars+ ctrlEndTrans msgChan transId Ret idVRVars transEnvs+ End -> do -- decision for rerun (invalid) trans+ debugStrLn8 ("<<< ctrlContTrans: End/Invalid transId = "++show transId)+ finishTrans transId idVRVars+ Err -> do -- error before decision+ debugStrLn8 ("<<< ctrlContTrans: Err transId = "++show transId)+ electNewTransCoordinator transId transEnvs End+ ctrlContTrans msgChan transId idVRVars transEnvs++ctrlEndTrans :: Chan RemCont -> TransID -> RemCont -> + [ValidRemVal] -> [EnvAddr] -> IO ()+ctrlEndTrans msgChan transId remCont idVRVars transEnvs = do+ -- finish remote transaction (phase 3)+ debugStrLn8 ("> ctrlEndTrans readChan ")+ msg <- readChan msgChan+ debugStrLn8 ("< ctrlEndTrans readChan ")+ case msg of+ End -> do -- finish trans ok+ debugStrLn8 ("<<< ctrlEndTrans: End transId = "++show transId ++ " idVRVars "++ show idVRVars)+ finishTrans transId idVRVars+ debugStrLn8 ("<<< ctrlEndTrans: End unlocked transId = "++show transId)+ Err -> do -- error after decision+ debugStrLn8 ("<<< ctrlEndTrans: Err transId = "++show transId)+ debugStrLn8 ("### ctrlEndTrans phase III cont trans w/new initiator")+ electNewTransCoordinator transId transEnvs remCont+ ctrlEndTrans msgChan transId remCont idVRVars transEnvs+ debugStrLn8 ("### ctrlEndTrans ok")+ m -> if m == remCont -- no error, possible duplicate recovery msgs+ then ctrlEndTrans msgChan transId m idVRVars transEnvs + else error "error ctrlEndTrans"++finishTrans :: TransID -> [ValidRemVal] -> IO ()+finishTrans trId idVRs = do+ mapM_ (unLockTVarFromId.fst) idVRs+ updateAutoTrans (+ (-1)) trId+ modifyMVar_ gDistTransCont (return . P.filter ((/= trId).fst))++electNewTransCoordinator :: TransID -> [EnvAddr] -> RemCont -> IO ()+electNewTransCoordinator transId@(env, _) transEnvs remCont = do+ let upEnvs = P.filter (/= env) transEnvs+ case minimum upEnvs of+ newC | newC == gMyEnv -> contTransWNewCoord transId remCont upEnvs+ | otherwise -> remPutMsg newC + (RemElectedNewCoord transId remCont upEnvs)+ `CE.catch` -- newCoordinator unavailable itself (SomeDistTVarException)+ \ex -> let eEnv = (/= distTVarExEnv ex)+ upEnvs = P.filter eEnv . P.filter (/= env) $ transEnvs+ in electNewTransCoordinator transId upEnvs remCont+ `CE.catch` -- other exception+ \(e1::SomeException) -> debugStrLn1 ("electNewTransCoordinator other ex"++show e1)++contTransWNewCoord :: TransID -> RemCont -> [EnvAddr] -> IO ()+contTransWNewCoord transId remCont upEnvs = do -- recovery+ debugStrLn1 (">>><<< contTransWNewCoord 1 : "++show remCont)+ let remEnvs = P.filter (/= gMyEnv) upEnvs+ case remCont of+ End -> return ()+ _ -> do -- continue with trans coordination+ debugStrLn1 (">>><<< contTransWNewCoord 2 : "++show remCont)+ robustMapM_ (flip remPutMsg (RemContTrans transId remCont)) remEnvs+ robustMapM_ (flip remPutMsg (RemContTrans transId End)) remEnvs+ contRemTrans transId End -- end local++validateValidIds :: ValidRemVal -> IO Bool -> IO Bool+validateValidIds (_, (Nothing, _)) isValid = isValid+validateValidIds (tId, (Just (versionId, _), _)) isValid =+ validateTVarFromId (tId, versionId) >>+ isValid++contWriteTVar :: ValidRemVal -> IO ()+contWriteTVar (_, (_, Nothing)) = return ()+contWriteTVar (tId, (_, Just str)) = writeTVarFromId (tId, str)++contExtWaitQ :: ValidRemVal -> IO ()+contExtWaitQ (_, (Nothing, _)) = return ()+contExtWaitQ (tId, (Just (_, rVar), _)) = extWaitQFromId (tId, rVar)++-----------------------+-- Distributed Stuff --+-----------------------++nodeReceiver :: Socket -> IO () -- new socket for new sender process+nodeReceiver sock = CE.catch (do+ setSocketOption sock NoDelay 1+ time <- getProcessTimes+ debugStrLn1 ("accept: "++show (elapsedTime time))+ (h, hn, p) <- accept sock -- wait for new process msg handle on socket+ debugStrLn4 ("### new connection: " ++ show h ++ " -> name: " ++ show hn ++ " -> port: " ++ show p)+ hSetBuffering h LineBuffering+ forkIO (readMsg h)+ nodeReceiver sock+ )(\(e::SomeException) -> debugStrLn1 ("nodeReceiver "++show e))++readMsg :: Handle -> IO ()+readMsg h = CE.catch (do+ str <- hGetLine h -- wait for new message on handle+ handleMsg h (read str)+ readMsg h+ )(\(e::SomeException) -> debugStrLn1 ("readMsg dropLostSocket "++show e))++handleMsg :: Handle -> STMMessage -> IO ()+handleMsg h msg =+ case msg of+ RemResume retryVarIds ->+ mapM_ resumeFromId retryVarIds >> aStat RES+ RemAddEnvToAction tVarId env ->+ addEnvToTVarActions tVarId env >> aStat AEA+ RemDelEnvFromAction tVarId env -> + delEnvFromTVarActions tVarId env >> aStat DEA+ RemLifeCheck -> + debugStrLn2 ("RemLifeCheck: I'm alive") >> aStat LFC+ RemReadTVar tId destEnv -> -- async+ readTVarFromId tId destEnv (hPutStrLn h) >> aStat RDT+ RemStartTrans transId idVRVars transEnvs -> -- async+ startRemTrans transId idVRVars transEnvs (hPutStrLn h . show)+ >> aStat STT+ RemContTrans transId remCont -> + contRemTrans transId remCont >> aStat CTT+ RemElectedNewCoord transId remCont oprtlEnvs -> + contTransWNewCoord transId remCont oprtlEnvs >> aStat ENC++--------------------+-- TVar Lifecheck --+--------------------++data AutoLink = AutoLink {autoTrans :: Int, -- stack of open remote trans+ autoRetry :: [RetryVar]} -- active retry vars+ deriving Show++gDefaultLink :: AutoLink+gDefaultLink = AutoLink {autoTrans = 0, autoRetry = []}++type Link = (EnvAddr, AutoLink)++gLinks :: MVar [Link]+{-# NOINLINE gLinks #-}+gLinks = unsafePerformIO (newMVar [])++gTransChkIntv, gRetryChkIntv :: Int+gTransChkIntv = 1+gRetryChkIntv = 3++gMegaMuSec :: Int+gMegaMuSec = 1000000+++updateAutoTrans :: (Int -> Int) -> TransID -> IO ()+updateAutoTrans f (env, _) = do + links <- takeMVar gLinks+ case Data.List.partition ((== env).fst) links of+ ([], otherLinks) -> do -- no link to env yet+ putMVar gLinks ((env, gDefaultLink{autoTrans = f (autoTrans gDefaultLink)}):otherLinks)+ forkIO (lifeCheck env) -- start new life check loop+ return ()+ ([(e, link)], otherLinks) -> do -- existing link, modify link only+ putMVar gLinks ((e, link{autoTrans = f (autoTrans link)}):otherLinks)+ _ -> putMVar gLinks links -- internal error++insertRetryLinks :: EnvAddr -> [ValidRemVal] -> IO ()+insertRetryLinks env idVRs = do -- env mapped over other processes in trans+ --printGLinkMap "insertRetryLinks"+ links <- takeMVar gLinks+ case Data.List.partition ((== env).fst) links of+ ([], otherLinks) -> do -- no link to env yet+ putMVar gLinks ((env,retrylink gDefaultLink):otherLinks)+ forkIO (lifeCheck env) -- start new life check loop+ return ()+ ([(e, link)], otherLinks) -> do -- existing link, modify link only+ putMVar gLinks ((e, retrylink link):otherLinks)+ _ -> putMVar gLinks links -- internal error+ where retrylink l = foldr insertRetryLink l idVRs++insertRetryLink :: ValidRemVal -> AutoLink -> AutoLink+insertRetryLink (_, (Just (_, rVar), _)) link@AutoLink{autoRetry=rVars} = + -- duplicate retryVars in different TVars in same env possible+ link{autoRetry=rVar:rVars}+insertRetryLink _ link = link++deleteRetryLinks :: EnvAddr -> [ValidRemVal] -> IO ()+deleteRetryLinks env idVRs = do -- env mapped over other processes in trans+ links <- takeMVar gLinks+ case Data.List.partition ((== env).fst) links of+ ([], otherLinks) -> do -- no link to env+ putMVar gLinks otherLinks -- internal error+ ([(e, link)], otherLinks) -> -- existing link, modify link only+ putMVar gLinks ((e, foldr deleteRetryLink link idVRs):otherLinks)+ _ -> putMVar gLinks links -- internal error++deleteRetryLink :: ValidRemVal -> AutoLink -> AutoLink+deleteRetryLink (_, (Just (_, rVar), _)) link@AutoLink{autoRetry=rVars} = + -- filter also possible duplicate retryVars in different TVars in same env+ link{autoRetry = P.filter (/= rVar) rVars}+deleteRetryLink _ link = link++lifeCheck :: EnvAddr -> IO ()+lifeCheck env = CE.catch (do+ links <- readMVar gLinks+ case P.lookup env links of+ Nothing -> do+ debugStrLn9 ("lifeCheck Nothing: "++show env)+ return () -- no ping for env+ Just link | autoTrans link > 0 -> do+ debugStrLn9 ("lifeCheck trans "++show env)+ ping+ threadDelay (gMegaMuSec * gTransChkIntv) + lifeCheck env+ | autoRetry link /= [] -> do+ debugStrLn9 ("lifeCheck retry "++show env)+ ping+ threadDelay (gMegaMuSec * gRetryChkIntv) + lifeCheck env+ | otherwise -> do -- last ping+ ping+ debugStrLn9 ("lifeCheck -------: "++show env)+ deleteLink+ )(\(e::SomeException) -> do+ debugStrLn9 ("!!! lifeCheck recovery" ++ show e)+ aStat DRP+ recoverBrokenReactiveTrans env+ recoverBrokenInactiveTrans env+ deleteLink+ )+ where + ping = remPutMsg env RemLifeCheck+ deleteLink = modifyMVar_ gLinks (return.P.filter ((/= env).fst))++recoverBrokenReactiveTrans :: EnvAddr -> IO ()+recoverBrokenReactiveTrans env = do+ debugStrLn1 ("<<< !!! recoverBrokenReactiveTrans: "++show env)+ conts <- readMVar gDistTransCont+ let brokenRemTrans = P.filter ((== env) . fst . fst) conts+ mapM_ ((flip writeChan Err).snd) brokenRemTrans+ debugStrLn1 ("<<< recoverBrokenReactiveTrans "++show brokenRemTrans)++recoverBrokenInactiveTrans :: EnvAddr -> IO ()+recoverBrokenInactiveTrans env = do+ debugStrLn1 ("<<< !!! recoverBrokenInactiveTrans env: "++show env)+ links <- readMVar gLinks+ case P.lookup env links of+ Nothing -> return ()+ Just AutoLink{autoRetry=retryVars} -> do+ debugStrLn1 ("<<< recoverBrokenInactiveTrans retryVars: "++show retryVars)+ mapM_ coreResume retryVars -- resume retryVars (deadlock from broken link)+coreResume :: RetryVar -> IO ()+coreResume (RetryVar retryMVar _) = resumeRetryVarAct retryMVar+coreResume (LinkRetryVar (VarLink env retVarId)) = CE.catch (+ remPutMsg env (RemResume [retVarId])+-- ignore errors in coreResume, dead process proxy waking up other dead process+ ) (\(_::SomeException) -> return ())++--------------------+-- Initialization --+--------------------++-- |'startDist' enables inter process communication and exception handling and+-- then executes the given main function +startDist :: IO () -- ^application main function to be executed. Each main + -- function in the distributed system has to be wrapped+ -- in a 'startDist' call + -> IO ()+startDist nodeMain = CE.catch (do+ serverPid <- startNameService -- may be removed+ threadDelay gMegaMuSec+ installHandler sigPIPE Ignore Nothing -- no UNIX signal on closed sockets+ time <- getProcessTimes+ debugStrLn1 ("listenOn: "++show (elapsedTime time))+ seq gMyEnv (forkIO (nodeReceiver gMySocket))+ -- >> forkIO timedDebugger+ nodeMain+ swapMVar gLinks [] -- delete all links to enable finalizer GC+ performGC -- activate all pending finalizers+ terminateProcess serverPid -- may be removed+ debugStrLn1 ("terminate NameService")+ )(propagateEx "startDist")++startNameService :: IO ProcessHandle+startNameService = do+ debugStrLn1 ("startNameService")+ runCommand "NameServer"
+ Control/Distributed/STM/DebugBase.hs view
@@ -0,0 +1,180 @@+{-# OPTIONS_HADDOCK hide #-}++module Control.Distributed.STM.DebugBase+ (debugStrLn0,debugStrLn1,debugStrLn2,debugStrLn3,debugStrLn4,+ debugStrLn5,debugStrLn6,debugStrLn7,debugStrLn8,debugStrLn9,+ gDebugLock, startGDebug, stopGDebug, gDebugStrLn,+ newDebugMVar, inspectMVars, timedInspect) where++import Control.Concurrent+import Prelude+import System.IO+import System.IO.Unsafe++---------------------+-- Debugging Tools --+---------------------++debug0,debug1,debug2,debug3,debug4,debug5,debug6,debug7,debug8,debug9 :: Bool+debug0 = False -- name server+debug1 = True -- catch error+debug2 = False -- robustness+debug3 = False -- robust <-+debug4 = False -- tcp connection+debug5 = False -- robust ->+debug6 = False -- bomberman+debug7 = False -- atomic+debug8 = False -- 3 phase commit+debug9 = True -- life check++-- located in module EnvAddr+--debugErrStrLn1 :: String -> Exception -> IO ()+--debugErrStrLn1 str e = debugStrLn1 (str ++ (showError e))++debugStrLn0 :: String -> IO ()+debugStrLn0 str = if debug0 then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return () ++debugStrLn1 :: String -> IO ()+debugStrLn1 str = if debug1 then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return () ++debugStrLn2 :: String -> IO ()+debugStrLn2 str = if debug2 then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return () ++debugStrLn3 :: String -> IO ()+debugStrLn3 str = if debug3 then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return () ++debugStrLn4 :: String -> IO ()+debugStrLn4 str = if debug4 then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return () +debugStrLn5 :: String -> IO ()+debugStrLn5 str = if debug5 then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return () ++debugStrLn6 :: String -> IO ()+debugStrLn6 str = if debug6 then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return () ++debugStrLn7 :: String -> IO ()+debugStrLn7 str = if debug7 then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return () ++debugStrLn8 :: String -> IO ()+debugStrLn8 str = if debug8 then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return () ++debugStrLn9 :: String -> IO ()+debugStrLn9 str = if debug9 then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return () ++gDebug :: MVar Bool+{-# NOINLINE gDebug #-}+gDebug = unsafePerformIO (newMVar False)++gDebugLock :: MVar ()+{-# NOINLINE gDebugLock #-}+gDebugLock = unsafePerformIO (newMVar ())++gDebugStrLn :: String -> IO ()+gDebugStrLn str = do+ isDebug <- readMVar gDebug + if isDebug then do+ myPid <- myThreadId+ takeMVar gDebugLock+ hPutStrLn stderr (show myPid++": "++str) + putMVar gDebugLock ()+ else return ()++startGDebug :: IO ()+startGDebug = swapMVar gDebug True >> return ()+ +stopGDebug :: IO ()+stopGDebug = swapMVar gDebug False >> return ()++gMVarStates :: MVar (IO ())+{-# NOINLINE gMVarStates #-}+gMVarStates = unsafePerformIO (newMVar (return ()))++newDebugMVar :: String -> a -> IO (MVar a)+newDebugMVar s var = do+ mVar <- newMVar var+ mVarStates <- takeMVar gMVarStates+ putMVar gMVarStates (do+ hPutStr stderr (s++" ")+ b <- isEmptyMVar mVar+ hPutStr stderr (if b then "empty !; " else "full; ")+ mVarStates)+ return mVar++inspectMVars :: String -> IO ()+inspectMVars s = do+ takeMVar gDebugLock+ hPutStrLn stderr s+ myPid <- myThreadId+ hPutStr stderr (show myPid++": ### MVar states >>>")+ mVarStates <- readMVar gMVarStates+ mVarStates+ hPutStrLn stderr ("<<< MVar states ###")+ putMVar gDebugLock ()++timedInspect :: IO ()+timedInspect = do+ if debug6 + then do+ inspectMVars "### Timed Debugger ###"+ threadDelay (5 * 1000000)+ timedInspect+ else return ()++instance Show (IO a) where+ show _ = show "IO "++instance Show (MVar a) where+ show _ = show "MVar "++instance Show (Chan a) where+ show _ = show "Chan "+
+ Control/Distributed/STM/DebugDSTM.hs view
@@ -0,0 +1,44 @@+{-# OPTIONS_HADDOCK hide #-}++module Control.Distributed.STM.DebugDSTM+ (module Control.Distributed.STM.DebugBase, + printValid, printValidFromId, proc, printStatistics) where++import Control.Distributed.STM.DebugBase+import Control.Distributed.STM.EnvAddr+import Control.Distributed.STM.STM+import Control.Distributed.STM.TVar+import Prelude as P hiding (catch, putStr, putStrLn)++------------------------+-- Debugging Routines --+------------------------++printValid :: String -> TransID -> ValidLog -> IO ()+printValid str transId (env, startVals) = do+ debugStrLn8 (">>> printValid "++str++" <<<")+ debugStrLn8 ("TransID = "++show transId)+ case startVals of+ Left idActsRVal -> + debugStrLn8 ("loc env = "++show env++show ((fst.unzip.fst) idActsRVal))+ Right startRemVals -> do+ debugStrLn8 ("rem env = "++show env)+ mapM_ printRemVal startRemVals+ debugStrLn8 ("<<< printValid")++printRemVal :: ValidRemVal -> IO ()+printRemVal (vId, (rVal, wVal)) = do+ debugStrLn8 (" id = "++show vId++" rVal = "++show rVal++" wVal = "++show wVal)++printValidFromId :: String -> TransID -> [ValidRemVal] -> IO ()+printValidFromId str transId startRemVals = do+ debugStrLn8 (">>> printValidFromId "++str++" <<<")+ debugStrLn8 ("TransID = "++show transId)+ mapM_ printRemVal startRemVals+ debugStrLn8 ("<<< printValidFromId")++proc :: IO a -> STM a+proc io = STM (\stmState -> io >>= return . Success stmState)++printStatistics :: String -> IO ()+printStatistics = printTCPStat
+ Control/Distributed/STM/Dist.hs view
@@ -0,0 +1,108 @@+{-# OPTIONS_HADDOCK hide #-}++module Control.Distributed.STM.Dist (Dist, regTVars, finTVars) where++import Control.Distributed.STM.EnvAddr+import Prelude as P++---------------------------+-- Class Definition Dist --+---------------------------++-- |The class 'Dist' defines the distribution property of 'TVar' values. Any +-- TVar value must implement class 'Dist'. All basic data types exported by+-- the Prelude are instances of 'Dist', and 'Dist' may be derived for any data+-- type whose constituents are also instances of 'Dist'. Any custom-typed TVar+-- value type should implement 'finTVars' and 'regTVars' to do nothing and +-- return '()'.+--+-- Note that 'finTVars' and 'regTVars' should never be called by the application+-- itself!+class (Show a, Read a) => Dist a where+ -- |Do not call regTVars yourself!+ --+ -- regTVars registers all TVars within @a@ with a host TVar link count before + -- the TVars in @a@ are sent to remote nodes+ regTVars :: EnvAddr -> a -> IO () -- create link in env to remote TVars in a+ -- |Do not call finTVars yourself!+ --+ -- finTVars installs finalizers at all link TVars in @a@ which send messages+ -- to their host TVars to remove them from the host TVar link count after the+ -- link TVars have been garbage collected+ finTVars :: a -> IO () -- finalizer to delete link to remote TVars in a+++reg2 :: (Dist a, Dist b) => EnvAddr -> a -> b -> IO ()+reg2 env x y = regTVars env x >> regTVars env y++reg3 :: (Dist a, Dist b, Dist c) => EnvAddr -> a -> b -> c -> IO ()+reg3 env x y z = regTVars env x >> regTVars env y >> regTVars env z++reg4 :: (Dist a, Dist b, Dist c, Dist d) => EnvAddr ->a -> b -> c -> d -> IO ()+reg4 e w x y z = regTVars e w >> regTVars e x >> regTVars e y >> regTVars e z++fin2 :: (Dist a, Dist b) => a -> b -> IO ()+fin2 x y = finTVars x >> finTVars y++fin3 :: (Dist a, Dist b, Dist c) => a -> b -> c -> IO ()+fin3 x y z = finTVars x >> finTVars y >> finTVars z++fin4 :: (Dist a, Dist b, Dist c, Dist d) => a -> b -> c -> d -> IO ()+fin4 w x y z = finTVars w >> finTVars x >> finTVars y >> finTVars z++instance Dist a => Dist [a] where+ finTVars [] = return ()+ finTVars (x:xs) = fin2 x xs++ regTVars _ [] = return ()+ regTVars env (x:xs) = reg2 env x xs++instance Dist a => Dist (Maybe a) where+ finTVars (Nothing) = return ()+ finTVars (Just x) = finTVars x++ regTVars _ (Nothing) = return ()+ regTVars env (Just x) = regTVars env x++instance (Dist a, Dist b) => Dist (Either a b) where+ finTVars (Left x) = finTVars x+ finTVars (Right x) = finTVars x++ regTVars env (Left x) = regTVars env x+ regTVars env (Right x) = regTVars env x++instance Dist Bool where+ finTVars _ = return ()+ regTVars _ _ = return ()++instance Dist Int where+ finTVars _ = return ()+ regTVars _ _ = return ()++instance Dist Integer where+ finTVars _ = return ()+ regTVars _ _ = return ()++instance Dist Char where+ finTVars _ = return ()+ regTVars _ _ = return ()++instance Dist Float where+ finTVars _ = return ()+ regTVars _ _ = return ()++instance Dist () where+ finTVars () = return ()+ regTVars _ () = return ()++instance (Dist a, Dist b) => Dist (a,b) where+ finTVars (x, y) = fin2 x y+ regTVars env (x, y) = reg2 env x y++instance (Dist a, Dist b, Dist c) => Dist (a,b,c) where+ finTVars (x, y, z) = fin3 x y z+ regTVars env (x, y, z) = reg3 env x y z++instance (Dist a, Dist b, Dist c, Dist d) => Dist (a,b,c,d) where+ finTVars (w, x, y, z) = fin4 w x y z+ regTVars env (w, x, y, z) = reg4 env w x y z
+ Control/Distributed/STM/EnvAddr.hs view
@@ -0,0 +1,375 @@+{-# OPTIONS_GHC -XScopedTypeVariables -XDeriveDataTypeable #-}+{-# OPTIONS_HADDOCK hide #-}++module Control.Distributed.STM.EnvAddr (ID, STMID, VarID, VersionID, uniqueId,+ EnvAddr, gMyEnv, gMySocket, VarLink (VarLink),+ connectToRetryEnv, connectToEnvHold, connectAtomicToEnv,+ SomeDistTVarException (PropagateDistTVarFail, CommunicationFail,+ NodeConnectionFail), propagateEx, distTVarExEnv,+ sendTCP, recvTCP, aStat, printTCPStat,+ RemMsgIdx (ATM,INV,STT,CTT,ENC,RES,AEA,DEA,LFC,DRP,RDT),+ gNameServerPort, connectToNameServer) where++import Control.Concurrent+import Control.Exception+import Data.Dynamic+import Control.Distributed.STM.DebugBase+import Control.Distributed.STM.Utils+import Network+import Network.BSD+import Network.Socket+import Maybe+import Prelude as P hiding (catch, putStr, putStrLn)+import System.IO+import System.IO.Unsafe++-----------------+-- Identifiers --+-----------------++type ID = Integer+type STMID = ID+type VarID = ID+type VersionID = ID++---------------------------------------+-- Unique IDs from common pool for --+-- STM, TVar, TVar-Version, RetryVar --+---------------------------------------++gInitID :: ID+gInitID = 1++gFreshIds :: MVar [ID]+{-# NOINLINE gFreshIds #-}+gFreshIds = unsafePerformIO (newMVar [gInitID..])++uniqueId :: IO ID+uniqueId = modifyMVar gFreshIds (\(h:t) -> return (t,h))++freeGlobalId :: ID -> IO ()+freeGlobalId n = modifyMVar_ gFreshIds (return . (n:))++-------------------------+-- Network Environment --+-------------------------++gNameServerPort :: PortNumber+gNameServerPort = 60000++gServerPort :: PortNumber+gServerPort = 60001++freshPort :: PortNumber -> IO (PortID, Socket)+freshPort p = catch (do+ s <- listenOn (PortNumber p)+ return (PortNumber p,s)+ )(\(_::SomeException) -> freshPort (p+1))++type IPAddr = String++data EnvAddr = EnvAddr PortID IPAddr+ deriving (Show, Read, Eq, Ord)++data VarLink = VarLink EnvAddr -- host TVar node+ VarID -- TVar Id on host TVar node+ deriving (Eq, Show, Read)++instance Show PortID where+ show (PortNumber p) = show p+ show _ = "Show PortID: otherPort" -- internal error++instance Read PortID where+ readsPrec _ str0 =+ case reads str0 of+ ((p,str2):_) -> [(PortNumber (fromInteger p),str2)]+ o -> error ("error readsPrec PortID" ++ show o)++instance Eq PortID where+ (PortNumber p1)==(PortNumber p2) = p1==p2+ _ == _ = False -- error++instance Ord PortID where+ (PortNumber p1)<=(PortNumber p2) = p1<=p2+ _ <= _ = False -- error++gMyIpAddr :: IPAddr+{-# NOINLINE gMyIpAddr #-}+gMyIpAddr = unsafePerformIO $ do+ hName <- getHostName+ hEntry <- getHostByName hName+ inet_ntoa (hostAddress hEntry)++gMyPort :: PortID+{-# NOINLINE gMyPort #-}+gMySocket :: Socket+{-# NOINLINE gMySocket #-}+(gMyPort,gMySocket) = unsafePerformIO $ do+ (port,sock) <- freshPort gServerPort+ return (port,sock)++gMyEnv :: EnvAddr -- initialize environment+gMyEnv = (EnvAddr $! gMyPort) $! gMyIpAddr++nameServerEnv :: String -> EnvAddr+nameServerEnv = EnvAddr (PortNumber gNameServerPort)++-----------------+-- Post Office --+-----------------++type TCPSndConn = (EnvAddr, Handle)++gTCPSndConn :: MVar [TCPSndConn]+{-# NOINLINE gTCPSndConn #-}+gTCPSndConn = unsafePerformIO (newMVar [])++lookupTCPSndEnv :: Handle -> IO EnvAddr+lookupTCPSndEnv h = do+ conns <- readMVar gTCPSndConn+ debugStrLn5 ("lookupTCPSndEnv: "++show conns++"\n"++show h++"\n"++show (lookupSndEnv h conns))+ return (fromJust (lookupSndEnv h conns))++lookupSndEnv :: Handle -> [TCPSndConn] -> Maybe EnvAddr+lookupSndEnv _ [] = Nothing+lookupSndEnv key ((env, h):conns) | key == h = Just env+ | otherwise = lookupSndEnv key conns++connectToEnvHold' :: EnvAddr -> IO Handle+connectToEnvHold' env = catch (do+ sndConns <- readMVar gTCPSndConn+ case P.lookup env sndConns of+ Just h -> return h+ Nothing -> do -- no tcp yet, establish (one and only) unidir. conn.+ h <- connectToEnv' env+ modifyMVar_ gTCPSndConn (return . ((env, h):))+ return h+ )(propagateEx "connectToEnvHold'")++type TCPRecConn = (EnvAddr, [Handle])++gTCPRecConn :: MVar [TCPRecConn]+{-# NOINLINE gTCPRecConn #-}+gTCPRecConn = unsafePerformIO (newMVar [])++gTCPTotalRecConn :: MVar [TCPRecConn]+{-# NOINLINE gTCPTotalRecConn #-}+gTCPTotalRecConn = unsafePerformIO (newMVar [])++-- all open connections to determine env in case of broken connection+lookupTCPRecEnv :: Handle -> IO EnvAddr+lookupTCPRecEnv h = do+ conns <- readMVar gTCPTotalRecConn+ debugStrLn5 ("lookupTCPRecEnv: "++show conns++"\n"++show h++"\n"++show (lookupRecEnv h conns))+ case lookupRecEnv h conns of+ Nothing -> return gMyEnv -- internal error+ Just env -> return env++lookupRecEnv :: Handle -> [TCPRecConn] -> Maybe EnvAddr+lookupRecEnv _ [] = Nothing+lookupRecEnv key ((_, []):conns) = lookupRecEnv key conns+lookupRecEnv key ((env, (h:hs)):conns) + | key == h = Just env+ | otherwise = lookupRecEnv key ((env, hs):conns)++connectAtomicToEnv' :: EnvAddr -> (Handle -> IO String) -> IO String+connectAtomicToEnv' env tcp = do+ debugStrLn3 ("connectAtomicToEnv ->: " ++ show env)+ h <- allocTcpHandle env+ debugStrLn3 ("connectAtomicToEnv -: " ++ show h)+ answer <- tcp h -- execute 2-way tcp comm. atomically w/rspct to handle+ debugStrLn3 ("connectAtomicToEnv --: " ++ show answer)+ freeTcpHandle h env+ debugStrLn3 ("connectAtomicToEnv <-: ")+ return answer++freeTcpHandle :: Handle -> EnvAddr -> IO ()+freeTcpHandle h env = do+ tcpConns <- takeMVar gTCPRecConn+ case P.lookup env tcpConns of+ Just hs -> putMVar gTCPRecConn (replaceListElems (h:hs) env tcpConns)+ Nothing -> putMVar gTCPRecConn ((env,[h]):tcpConns)+++allocTcpHandle :: EnvAddr -> IO Handle+allocTcpHandle env = catch (do+ conns <- takeMVar gTCPRecConn+ case P.lookup env conns of+ Just [] -> do -- no tcp available -> get new tcp+ putMVar gTCPRecConn conns+ h <- connectToEnv' env+ modifyMVar_ gTCPTotalRecConn (return . (updateListElems (h:) [h] env))+ return h+ Just (h:hs) -> do -- get head of available tcp+ putMVar gTCPRecConn (replaceListElems hs env conns)+ return h+ Nothing -> do -- no tcp yet -> get first new tcp+ putMVar gTCPRecConn conns+ h <- connectToEnv' env+ modifyMVar_ gTCPTotalRecConn (return . ((env,[h]):))+ return h+ )(propagateEx "allocTcpHandle")++connectToEnv' :: EnvAddr -> IO Handle+connectToEnv' env@(EnvAddr pid ip) = catch (do + h <- connectTo ip pid+ hSetBuffering h LineBuffering+ return h+ )(\e -> debugStrLn1 ("connectToEnv' " ++ show e) >> throw (NodeConnectionFail "connectToEnv' " env e))++connectToNameServer' :: String -> IO Handle+connectToNameServer' = connectToEnv'.nameServerEnv++--------------------+-- TCP Statistics --+--------------------++data RemMsgIdx = ATM|INV|STT|CTT|ENC|RES|AEA|DEA|LFC|DRP|RDT+ deriving (Eq, Ord, Show)++type STMMsgStat = (RemMsgIdx, Int)++gSTMMsgStat :: MVar [STMMsgStat]+{-# NOINLINE gSTMMsgStat #-}+gSTMMsgStat = unsafePerformIO (newMVar ((ATM, 0):(INV,0):[]))++aStat :: RemMsgIdx -> IO ()+aStat msgIdx = modifyMVar_ gSTMMsgStat (return . (insertWith (\_ x ->x+1)) msgIdx 1)++data TCPStat = TCPStat {+ conEnv :: Int,+ conExclEnv :: Int,+ conRetryVar :: Int,+ conNS :: Int+ }++gTCPStat :: MVar TCPStat+{-# NOINLINE gTCPStat #-}+gTCPStat = unsafePerformIO (newMVar (TCPStat{conEnv = 0, + conExclEnv = 0,+ conRetryVar = 0,+ conNS = 0}))++connectToEnvHold :: EnvAddr -> IO Handle+connectToEnvHold env = do+ modifyMVar_ gTCPStat (\ips -> return ips{conEnv=(conEnv ips)+1})+ connectToEnvHold' env++connectAtomicToEnv :: EnvAddr -> (Handle -> IO String) -> IO String+connectAtomicToEnv env tcp = do+ modifyMVar_ gTCPStat (\ips -> return ips{conExclEnv=(conExclEnv ips)+1})+ connectAtomicToEnv' env tcp++connectToRetryEnv :: EnvAddr -> IO Handle+connectToRetryEnv env = do+ modifyMVar_ gTCPStat (\ips -> return ips{conRetryVar=(conRetryVar ips)+1})+ connectToEnvHold' env++connectToNameServer :: String -> IO Handle+connectToNameServer ns = do+ modifyMVar_ gTCPStat (\ips -> return ips{conNS=(conNS ips)+1})+ connectToNameServer' ns++printTCPStat :: String -> IO ()+printTCPStat s = do+ ips <- readMVar gTCPStat+ sndConns <- readMVar gTCPSndConn+ recConns <- readMVar gTCPRecConn+ msgs <- readMVar gSTMMsgStat+ printStat s ips sndConns recConns msgs++printStat :: String -> TCPStat -> [TCPSndConn] -> [TCPRecConn] -> [STMMsgStat] + -> IO ()+printStat s ips sndConns recConns msgs = do+ let l = foldr (+) 1 (map (length.snd) recConns)+ takeMVar gDebugLock+ putStrLn s+ putStrLn ("--> TCP Verbindungen: " ++ showJustify (length sndConns) 5)+ putStrLn ("<-> TCP Verbindungen: " ++ showJustify l 5)+ putStrLn ("--> TCP Nachrichten: " ++ showJustify (conEnv ips) 5)+ putStrLn ("<-> TCP Nachrichten: " ++ showJustify (conExclEnv ips) 5)+ putStrLn ("|-> TCP Nachrichten: " ++ showJustify (conRetryVar ips) 5)+ printMsgs msgs+ putMVar gDebugLock ()++printMsgs :: [STMMsgStat] -> IO ()+printMsgs [] = return ()+printMsgs msgs = do+ putStrLn (">-> Transaktionen: " ++ (showJustify.snd.head) msgs 5)+ putStrLn ("-X- Transaktionen: " ++ (showJustify.snd.head.tail) msgs 5)+ putStrLn ("<-- TCP Nachrichten: " ++ showJustify (foldr (+) 0 (map snd ((tail.tail) msgs))) 5 ++ "\n")+ mapM_ printMsg ((tail.tail) msgs)++printMsg :: (RemMsgIdx, Int) -> IO ()+printMsg (msg, count) = putStrLn (show msg ++ ": " ++ showJustify count 5)++----------------+-- Robustness --+----------------++-- |'SomeDistTVarException' is the abstract exception type which is thrown+-- by the DSTM library when either 'readTVar' or 'writeTVar' is called on an+-- unreachable TVar.+-- A TVar becomes unreachable when the process hosting the TVar becomes+-- unreachable. +-- An atomic transaction using a TVar which becomes unreachable during the +-- execution of 'atomic' may either execute completely (without the unreachable+-- TVar(s)) or execute not at all depending on transaction states. In either+-- case an exception of type 'SomeDistTVarException' is raised.+data SomeDistTVarException = PropagateDistTVarFail String SomeDistTVarException+ | CommunicationFail String EnvAddr SomeException+ | NodeConnectionFail String EnvAddr SomeException+ deriving Typeable++instance Exception SomeDistTVarException++instance Show SomeDistTVarException where+ show (PropagateDistTVarFail loc ex) = + "PropagateDistTVarFail " ++ loc ++ "\n" ++ show ex+ show (CommunicationFail loc env ex) = formEx "CommunicationFail" loc env ex+ show (NodeConnectionFail loc env ex) = formEx "NodeConnectionFail" loc env ex++formEx :: String -> String -> EnvAddr -> SomeException -> String+formEx err loc env cause = "Message : " ++ err+ ++ "\nLocation: " ++ loc+ ++ "\nProcess : " ++ show env+ ++ "\nSysError: " ++ show cause ++ "\n"++propagateEx :: String -> SomeDistTVarException -> IO a+propagateEx loc e = throw (PropagateDistTVarFail (loc ++ " -> ") e)++distTVarExEnv :: SomeDistTVarException -> EnvAddr+distTVarExEnv eDist = case eDist of+ (PropagateDistTVarFail _ e) -> distTVarExEnv e+ (CommunicationFail _ env _) -> env+ (NodeConnectionFail _ env _) -> env++--------------------------+-- Robust Remote Access --+--------------------------++gSendLock :: MVar ()+{-# NOINLINE gSendLock #-}+gSendLock = unsafePerformIO (newMVar ())++sendTCP :: Show a => a -> Handle -> IO ()+sendTCP msg h = catch (do+ takeMVar gSendLock+ hPutStrLn h (show msg) + hFlush h+ putMVar gSendLock ()+ )(\e -> do+ debugStrLn1 $ "sendTCP " ++ show e+ putMVar gSendLock ()+ env <- lookupTCPSndEnv h+ throw (CommunicationFail ("sendTCP: " ++ show msg ++ show h) env e))++recvTCP :: Show a => a -> Handle -> IO String+recvTCP msg h = catch (do + hPutStrLn h (show msg) + hFlush h + hGetLine h+ )(\e -> do+ debugStrLn1 $ "recvTCP " ++ show e+ env <- lookupTCPRecEnv h+ throw (CommunicationFail ("recvTCP: " ++ show msg ++ show h) env e))
+ Control/Distributed/STM/NameService.hs view
@@ -0,0 +1,117 @@+{-# OPTIONS_GHC -XScopedTypeVariables #-}+{-# OPTIONS_HADDOCK hide #-}++module Control.Distributed.STM.NameService + (nameService, gDefaultNameServer,+ registerTVar, deregisterTVar, lookupTVar) where++import qualified Control.Exception as CE+import Control.Distributed.STM.DebugBase+import Control.Distributed.STM.Dist+import Control.Distributed.STM.EnvAddr+import Control.Distributed.STM.TVar+import Network+import Prelude as P hiding (catch, putStr, putStrLn)+import System.IO++---------------+-- Messaging --+---------------++data NameServerMsg = NameServerPing+ | NameServerReg String VarLink+ | NameServerUnReg String+ | NameServerLookup String+ deriving (Show, Read)++type TVarDict = [(String, VarLink)]++-----------------+-- Post Office --+-----------------++putNameServerLn :: Show a => String -> a -> IO ()+putNameServerLn nameServer msg = do+ debugStrLn0 ("--> putNameServerLn: "++nameServer++" msg: "++(show msg))+ h <- connectTo nameServer (PortNumber gNameServerPort) + sendTCP msg h+ hClose h++getNameServerLn :: Show a => String -> a -> IO String+getNameServerLn nameServer msg = do+ debugStrLn0 ("<-- getNameServerLn: "++nameServer++" msg: "++(show msg))+ h <- connectTo nameServer (PortNumber gNameServerPort) + answer <- recvTCP msg h+ hClose h+ return answer++------------------+-- Name Service --+------------------++-- |The default name server for the process running the main function. +-- Usually it is @localhost@.+gDefaultNameServer :: String+gDefaultNameServer = "localhost"++nameService :: String -> IO ()+nameService server = CE.catch (putNameServerLn server NameServerPing)+ (\(e::CE.SomeException) -> do -- no name server yet -> start one per node+ debugStrLn0 ("nameService: start name server: "++ show e)+ listenOn (PortNumber gNameServerPort) >>= readNameServerMsg [])++readNameServerMsg :: TVarDict -> Socket -> IO ()+readNameServerMsg tVarDict s = do+ debugStrLn0 ("nameService: readNameServerMsg: "++ show s)+ (h, _, _) <- accept s+ str <- hGetLine h+ newTable <- case reads str of+ ((msg,_):_) -> handleNameServerMsg h msg tVarDict+ _ -> return tVarDict -- internal error+ hClose h+ readNameServerMsg newTable s++handleNameServerMsg :: Handle -> NameServerMsg -> TVarDict -> IO TVarDict+handleNameServerMsg h msg tVarDict =+ case msg of+ NameServerPing -> return tVarDict+ NameServerReg name tVar -> do+ debugStrLn0 ("Registered: "++(show name))+ return $ (name, tVar):filter ((name/=).fst) tVarDict+ NameServerUnReg name -> do+ debugStrLn0 ("Unregistered: "++(show name))+ return $ filter ((name/=) . fst) tVarDict+ NameServerLookup name -> do+ debugStrLn0 ("Lookup: "++(show name))+ hPutStrLn h (show (lookup name tVarDict))+ return tVarDict++-- |'registerTVar' @server tVar name@ registers @tVar@ with @name@ onto @server@+registerTVar :: Dist a => String -> TVar a -> String -> IO ()+registerTVar server tVar name = CE.catch (do+ debugStrLn0 ("registerTVar: " ++ show tVar ++ " " ++ name)+ putNameServerLn server (NameServerReg name (tVarToLink tVar))+ regTVars gMyEnv tVar -- generate actions for exported tVar+ )(propagateEx "registerTVar")++-- |'deregisterTVar' @server name@ removes @name@ from @server@+deregisterTVar :: String -> String -> IO ()+deregisterTVar server name = CE.catch (do+ debugStrLn0 ("deregisterTVar: " ++ show name)+ putNameServerLn server (NameServerUnReg name)+ )(propagateEx "deregisterTVar")++-- |'lookupTVar' @server name@ returns ('Just' @tVar@) if a @tVar@ registration+-- of @name@ exists on @server@, 'Nothing' otherwise.+lookupTVar :: forall a . Dist a => String -> String -> IO (Maybe (TVar a))+lookupTVar server name = CE.catch (do+ debugStrLn0 ("lookupTVar: " ++ server ++ " , " ++ name)+ answer <- getNameServerLn server (NameServerLookup name)+ debugStrLn0 ("lookupTVar: " ++ show answer)+ case reads answer of+ ((Just link,_):_) -> do+ let tVar::TVar a = LinkTVar link+ finTVars tVar+ return $ Just tVar+ _ -> return Nothing+ )(propagateEx "lookupTVar")
+ Control/Distributed/STM/RetryVar.hs view
@@ -0,0 +1,123 @@+{-# OPTIONS_HADDOCK hide #-}++module Control.Distributed.STM.RetryVar + (RetryLog, RetryVar (RetryVar, LinkRetryVar), newRetryVar,+ suspend, resumeRetryVarAct, resumeFromId, bundledRetryLogs,+ insertRetryVarAction, deleteRetryVarAction,+ printGRetryVarActionMap, remPutRetryEnvLine) where++import Control.Concurrent+import qualified Control.Exception as CE (catch)+import qualified Data.Map as M hiding (map)+import Control.Distributed.STM.DebugBase+import Control.Distributed.STM.EnvAddr+import qualified Control.Distributed.STM.Utils as U+import Prelude as P+import System.IO.Unsafe++--------------------+-- Retry Variable --+--------------------++data RetryVar = RetryVar (MVar ()) -- suspend var for retry+ VarID -- retry var id+ | LinkRetryVar VarLink -- link to remote RetryVar+ deriving Eq++instance Show RetryVar where+ show (RetryVar _ tId) = show (LinkRetryVar (VarLink gMyEnv tId))+ show (LinkRetryVar v) = "(" ++ show v ++ ")"++instance Read RetryVar where+ readsPrec i str = map (\(x,s) -> (LinkRetryVar x,s)) (readsPrec i str)++---------------------+-- RetryVar Access --+---------------------++newRetryVar :: IO RetryVar+newRetryVar = do+ retryMVar <- newEmptyMVar+ newID <- uniqueId+ return (RetryVar retryMVar newID)++suspend :: RetryVar -> IO ()+suspend (RetryVar retryMVar _) = takeMVar retryMVar+suspend (LinkRetryVar _) = return () -- internal error++resumeRetryVarAct :: MVar () -> IO ()+resumeRetryVarAct retryMVar = tryPutMVar retryMVar () >> return ()++------------------------+-- Collect STM Action --+------------------------++type RetryLog = (EnvAddr, RetryLogBundle)+type RetryLogBundle = Either (IO ()) [VarID]++bundledRetryLogs :: RetryVar -> [RetryLog] -> [RetryLog]+bundledRetryLogs retryVar queue = + U.insertWith updRetryLog (retryVarEnv retryVar) singletonRLog queue+ where singletonRLog = singletonRetryLog retryVar++singletonRetryLog :: RetryVar -> RetryLogBundle+singletonRetryLog (RetryVar retryMVar _) = Left (resumeRetryVarAct retryMVar)+singletonRetryLog (LinkRetryVar (VarLink env retryVarId))+ | env == gMyEnv = Left (resumeFromId retryVarId)+ | otherwise = Right [retryVarId]++updRetryLog :: RetryLogBundle -> RetryLogBundle -> RetryLogBundle+updRetryLog (Left resume) (Left resumes) = Left (resumes >> resume)+updRetryLog (Right [rVarId]) (Right rVarIds) = Right (rVarId:rVarIds)+updRetryLog _ y = y -- internal error, either local or remote envs, not mixed++retryVarEnv :: RetryVar -> EnvAddr+retryVarEnv (RetryVar _ _) = gMyEnv+retryVarEnv (LinkRetryVar (VarLink env _)) = env++--------------------------------+-- RetryVar Remote Access Map --+--------------------------------++type RetryVarAction = IO ()++gRetryVarActionMap :: MVar (M.Map VarID RetryVarAction)+gRetryVarActionMap = unsafePerformIO (newMVar M.empty)++printGRetryVarActionMap :: VarID -> IO ()+printGRetryVarActionMap tVarId = do+ gMap <- readMVar gRetryVarActionMap+ case M.lookup tVarId gMap of+ Nothing -> debugStrLn2 ("tVarID non existing: " ++ show tVarId)+ Just _ -> debugStrLn2 ("tVarID existing " ++ show tVarId)+ debugStrLn2 ("---")+++insertRetryVarAction :: RetryVar -> IO ()+insertRetryVarAction (RetryVar retryMVar retryVarId) =+ modifyMVar_ gRetryVarActionMap $+ return . M.insertWith (flip const) retryVarId (resumeRetryVarAct retryMVar)+insertRetryVarAction (LinkRetryVar _) = return () -- internal error+++deleteRetryVarAction :: RetryVar -> IO ()+deleteRetryVarAction (RetryVar _ retryVarId) =+ modifyMVar_ gRetryVarActionMap $ return . M.delete retryVarId+deleteRetryVarAction (LinkRetryVar _) = return () -- internal error++resumeFromId :: VarID -> IO ()+resumeFromId retryVarId = do+ gMap <- readMVar gRetryVarActionMap+ case M.lookup retryVarId gMap of+ Just act -> act+ Nothing -> return () -- act already deleted because trans already resumed++-----------------+-- Post Office --+-----------------+-- like remPutMsg, statistics is only difference+remPutRetryEnvLine :: Show a => EnvAddr -> a -> IO ()+remPutRetryEnvLine env msg = CE.catch (do+ h <- connectToRetryEnv env+ sendTCP msg h+ )(propagateEx "remPutRetryEnvLine")
+ Control/Distributed/STM/STM.hs view
@@ -0,0 +1,44 @@+{-# OPTIONS_HADDOCK hide #-}++module Control.Distributed.STM.STM (STM (STM), STMState (STMState), + STMResult (Retry, Success, Exception), + runSTM, stmId, stmValid, stmCommit, stmRetryVar) where++import Control.Distributed.STM.RetryVar+import Control.Distributed.STM.TVar+import Control.Exception as CE (SomeException)+++-------------------+-- The STM monad --+-------------------++-- |A monad supporting atomic memory transactions+newtype STM a = STM (STMState -> IO (STMResult a))++instance Monad STM where+ -- (>>=) :: STM a -> (a -> STM b) -> STM b+ (STM tr1) >>= k = STM (\state -> do+ stmRes <- tr1 state+ case stmRes of+ Success newState v ->+ let (STM tr2) = k v in+ tr2 newState+ Retry newState -> return (Retry newState)+ Exception newState e -> return (Exception newState e)+ )+ -- return :: a -> STM a+ return x = STM (\state -> return (Success state x))++data STMState = STMState {stmId :: TransID,+ stmRetryVar :: RetryVar,+ stmValid :: [ValidLog],+ stmCommit :: [CommitLog]}++data STMResult a = Retry STMState+ | Success STMState a+ | Exception STMState CE.SomeException++runSTM :: STM a -> STMState -> IO (STMResult a)+runSTM (STM stm) state = stm state+
+ Control/Distributed/STM/TVar.hs view
@@ -0,0 +1,400 @@+{-# OPTIONS_GHC -XScopedTypeVariables #-}+{-# OPTIONS_GHC -fno-warn-unused-do-bind #-}+{-# OPTIONS_HADDOCK hide #-}++module Control.Distributed.STM.TVar (TVar (TVar, LinkTVar), + tVarToLink, tVarEnv, isDistErrTVar, + ValidLog, CommitLog, + lockAct, unLockAct, validAct, extWaitQAct, commitAct, notifyAct,+ CommitRemVal, ValidRemVal, MaybeRead,+ TransID, RemCont (Com, Ret, End, Err),+ STMMessage (RemResume,RemAddEnvToAction,RemDelEnvFromAction+ ,RemLifeCheck,RemReadTVar,RemStartTrans,RemContTrans+ ,RemElectedNewCoord),+ remGetMsg, remPutMsg, bundledValidLogs, bundledCommitLogs,+ addEnvToTVarActions, delEnvFromTVarActions,+ lockTVarFromId, unLockTVarFromId, readTVarFromId, writeTVarFromId,+ validateTVarFromId, notifyFromId, extWaitQFromId,+ gTVarActionMap,remLock,remValidate,+ readTVarValFromId, swapTVarValFromId) where++import Control.Concurrent+import qualified Control.Exception as CE (catch,throw,SomeException)+import Control.Monad+import qualified Data.List+import qualified Data.Map as DM hiding (map) +import Control.Distributed.STM.DebugBase+import Control.Distributed.STM.Dist+import Control.Distributed.STM.EnvAddr+import Control.Distributed.STM.Utils hiding (insertWith)+import qualified Control.Distributed.STM.Utils as U+import Prelude as P hiding (catch, putStr, putStrLn)+import Control.Distributed.STM.RetryVar+import System.Mem.Weak+import System.IO+import System.IO.Unsafe++----------------------------+-- Transactional Variable --+----------------------------++-- |Shared memory locations that support atomic memory transactions. Between+-- different nodes memory is shared using transparent process communication.+-- (TVars are called @host TVars@ when they reside on the process where they+-- have been created by calling 'newTVar'. They are called @link TVars@ on +-- other processes)+data TVar a = TVar (MVar (a,VersionID)) -- host process local TVar value+ VarID -- VersionID used for validity check+ (MVar ()) -- TVar lock for exclusive verify/commit+ (MVar [RetryLog]) -- wait queue of suspended transactions+ | LinkTVar VarLink -- link TVar++instance Show (TVar a) where+ show (TVar _ tId _ _) = show (LinkTVar (VarLink gMyEnv tId))+ show (LinkTVar v) = "(" ++ show v ++ ")" -- precedence ( ) level of read++instance Read (TVar a) where+ readsPrec i str = map (\(x,s) -> (LinkTVar x,s)) (readsPrec i str)++-----------------+-- TVar Access --+-----------------++tVarEnv :: TVar a -> EnvAddr+tVarEnv (TVar _ _ _ _) = gMyEnv+tVarEnv (LinkTVar (VarLink env _)) = env++tVarToLink :: TVar a -> VarLink+tVarToLink (TVar _ tId _ _) = VarLink gMyEnv tId+tVarToLink (LinkTVar link) = link++----------------------------------------------------------+-- Collect STM Actions -- lock, validate and make waitQ --+----------------------------------------------------------++type ValidLog = (EnvAddr, ValidLogBundle)+type ValidLogBundle = Either ([(VarID, LockActs)], ValidActs)+ [ValidRemVal]++data LockActs = LockActs {+ lockAct :: IO (), + unLockAct :: IO ()+ } deriving Show+data ValidActs = ValidActs {+ validAct :: IO Bool, + extWaitQAct :: IO ()+ } deriving Show++type ValidRemVal = (VarID, MaybeReadWrite)+type MaybeReadWrite = (MaybeRead, Maybe String)+type MaybeRead = Maybe (VersionID, RetryVar)+++bundledValidLogs :: TVar a -> MaybeRead -> [ValidLog] -> [ValidLog]+bundledValidLogs tVar versIdRetryVar validLog =+ -- tVarEnv == primary key+ U.insertWith updValidLog tVEnv singletonVLog validLog+ where singletonVLog = singletonValidLog tVar versIdRetryVar+ tVEnv = tVarEnv tVar++singletonValidLog :: TVar a -> MaybeRead -> ValidLogBundle+-- orig TVar: read TVar -> validate action, curr transaction -> xtd waitQ action+singletonValidLog (TVar ref tId lock waitQ) (Just (versionId, retryVar)) = Left+ ([(tId, LockActs {lockAct = takeMVar lock,+ unLockAct = putMVar lock () })],+ ValidActs {validAct = readMVar ref >>= return . (== versionId) . snd,+ extWaitQAct = modifyMVar_ waitQ insertRetry })+ where insertRetry = return . (bundledRetryLogs retryVar)+-- orig TVar: write TVar -> no "start"-actions other than locking+singletonValidLog (TVar _ tId lock _) Nothing = Left+ ([(tId, LockActs {lockAct = takeMVar lock,+ unLockAct = putMVar lock () })],+ ValidActs {validAct = return True, extWaitQAct = return ()})+-- rem TVar: read TVar -> validate action, curr transaction -> xtd waitQ action+singletonValidLog (LinkTVar (VarLink env tId)) (Just (versionId, retryVar))+ | env == gMyEnv = Left + ([(tId, LockActs{lockAct = lockTVarFromId tId,+ unLockAct = unLockTVarFromId tId })],+ ValidActs {validAct = validateTVarFromId (tId, versionId),+ extWaitQAct = extWaitQFromId (tId, retryVar)})+ -- rem TVar: read TVar -> validate data, curr transaction -> xtd waitQ data+ | otherwise = Right [(tId, (Just (versionId, retryVar), Nothing))]+-- rem TVar: write TVar -> no "start"-data other than needed for locking+singletonValidLog (LinkTVar (VarLink env tId)) Nothing+ | env == gMyEnv = Left+ ([(tId, LockActs {lockAct = lockTVarFromId tId,+ unLockAct = unLockTVarFromId tId })],+ ValidActs {validAct = return True, extWaitQAct = return ()})+ | otherwise = Right [(tId, (Nothing, Nothing))]++updValidLog :: ValidLogBundle -> ValidLogBundle -> ValidLogBundle+updValidLog (Left ([(tId, lock)], val)) (Left (locks, vals)) = Left+ -- tId == secondary key => total order on locks+ (U.insertWith const tId lock locks, + ValidActs {validAct = validAct val >>+ validAct vals,+ extWaitQAct = extWaitQAct val >> extWaitQAct vals })+updValidLog (Right [(tId, rwVal)]) (Right rwVals) = Right+ (U.insertWith mergeRVal tId rwVal rwVals)+updValidLog _ y = y -- internal error++mergeRVal :: MaybeReadWrite -> MaybeReadWrite -> MaybeReadWrite+mergeRVal (Nothing, _) v = v+mergeRVal (Just r, w) _ = (Just r, w)++----------------------------------------------+-- Collect STM Actions -- commit and notify --+----------------------------------------------++type CommitLog = (EnvAddr, CommitLogBundle)+type CommitLogBundle = Either [(VarID, CommitActs)] ([CommitRemVal], RegIO)++data CommitActs = CommitActs {+ commitAct :: IO (), + notifyAct :: IO ()+ } deriving Show++type CommitRemVal = (VarID, String)+type RegIO = IO ()++bundledCommitLogs :: Dist a => TVar a -> a -> [CommitLog] -> [CommitLog]+bundledCommitLogs tVar value commitLog =+ U.insertWith updCommit tVEnv singletonCLog commitLog+ where singletonCLog = singletonCommitLog tVar value+ tVEnv = tVarEnv tVar++singletonCommitLog :: Dist a => TVar a -> a -> CommitLogBundle+singletonCommitLog (TVar ref tId _ waitQ) value = Left+ [(tId, CommitActs{commitAct= modifyMVar_ ref incVersId,+ notifyAct= swapMVar waitQ [] >>= mapM_ coreNotify })]+ where incVersId =+ return . (,)value . (+1) . snd+singletonCommitLog (LinkTVar (VarLink env tId)) value+ | env == gMyEnv = + Left [(tId, CommitActs{commitAct = writeTVarFromId (tId, show value),+ notifyAct = notifyFromId tId+ })]+ | otherwise = Right ([(tId, show value)], regTVars env value)++updCommit :: CommitLogBundle -> CommitLogBundle -> CommitLogBundle+updCommit (Left [(tId, comNfyAct)]) (Left comNfys) = + Left (U.insertWith const tId comNfyAct comNfys)+updCommit (Right ([(tId, val)], reg)) (Right (idStrs, regs)) = + Right (U.insertWith const tId val idStrs, regs >> reg)+updCommit _ y = y -- internal error++----------------------------+-- TVar Remote Access Map --+----------------------------++data TVarActions = TVarActions {+ remRead :: EnvAddr -> IO String,+ remWrite :: String -> IO (),+ remReadVal :: IO String,+ remSwapVal :: String -> IO String,+ remValidate :: VersionID -> IO String,+ remLock :: IO (),+ remUnLock :: IO (),+ remExtWaitQ :: RetryVar -> IO (),+ remNotify :: IO ()+ }++gTVarActionMap :: MVar (DM.Map VarID (TVarActions, [EnvAddr]))+{-# NOINLINE gTVarActionMap #-}+gTVarActionMap = unsafePerformIO (newMVar DM.empty)++insertTVarActions :: Dist a => TVar a -> EnvAddr -> IO ()+insertTVarActions (TVar tVarRef tVarId tVarLock waitQ) fstDestEnv =+ let tVarActions = TVarActions {+ -- regular read of host TVar value and register possibly exported TVars+ remRead = \destEnv -> do + -- read lock to prevent read before finishing remote commit+ takeMVar tVarLock+ v'@(v, _) <- readMVar tVarRef+ putMVar tVarLock () -- read lock+ regTVars destEnv v+ debugStrLn8 ("###>>> remRead "++show v')+ return (show v'),+ -- regular write to host TVar, incr. version and finalize received TVars+ remWrite = \str -> do+ debugStrLn8 ("###>>> remWrite "++show str)+ let v = read str+ modifyMVar_ tVarRef (return .((,) v) .((+1) .snd))+ finTVars v,+ -- "non-destructive" read of uncommitted writes -> no TVar register+ remReadVal = readMVar tVarRef >>= return . show,+ -- "nd" swap of uncommitted writes -> no version increment and finalize+ remSwapVal = liftM show . swapMVar tVarRef . read,+ remValidate = \versionId -> do + (_, vId) <- readMVar tVarRef+ return (show (versionId == vId)),+ remLock = takeMVar tVarLock,+ remUnLock = putMVar tVarLock (),+ remExtWaitQ = \retryVar -> modifyMVar_ waitQ + (return . (bundledRetryLogs retryVar)),+ remNotify = do queue <- swapMVar waitQ []+ mapM_ coreNotify queue+ } in+ modifyMVar_ gTVarActionMap $+ return . DM.insertWith (\_ (acts, envs) -> (acts, (fstDestEnv:envs))) + tVarId (tVarActions, [fstDestEnv])+insertTVarActions (LinkTVar _) _ = return () -- internal error++coreNotify :: RetryLog -> IO ()+coreNotify (_, Left resumeAct) = resumeAct+coreNotify (env, Right retryVarIds) = do+ forkIO $ CE.catch (remPutRetryEnvLine env (RemResume retryVarIds)+ )(\(e::CE.SomeException) -> debugStrLn1 ("coreNotify "++show e))+ return ()-- ignore notification of lost nodes++addEnvToTVarActions :: VarID -> EnvAddr -> IO ()+addEnvToTVarActions tVarId env = do+ modifyMVar_ gTVarActionMap + (return . (DM.insertWith (\_ (as, envs) -> (as, (env:envs))) + tVarId (error ("error addEnvToTVarActions: TVarId "+++ show tVarId ++" not listed"))))++delEnvFromTVarActions :: VarID -> EnvAddr -> IO ()+delEnvFromTVarActions tVarId env = do+ oldMap <- takeMVar gTVarActionMap+ case DM.lookup tVarId oldMap of+ Nothing -> error ("error trying to remove non-existing TVar: "++show tVarId)+ Just (acts, envs) ->+ putMVar gTVarActionMap + (case Data.List.delete env envs of+ [] -> DM.delete tVarId oldMap -- no more refs to tVar+ newEnvs -> DM.insert tVarId (acts, newEnvs) oldMap)++readTVarFromId :: VarID -> EnvAddr -> (String -> IO ()) -> IO ()+readTVarFromId tVarId destEnv sendReply = CE.catch (do+ aMap <- readMVar gTVarActionMap+ remRead (fst (aMap DM.! tVarId)) destEnv >>= sendReply+ )(distInstanceError "readTVarFromId")++writeTVarFromId :: (VarID, String) -> IO ()+writeTVarFromId (tVarId, str) = CE.catch (do+ aMap <- readMVar gTVarActionMap+ remWrite (fst (aMap DM.! tVarId)) str+ )(distInstanceError "writeTVarFromId")++readTVarValFromId :: VarID -> IO String+readTVarValFromId tVarId = CE.catch (do+ aMap <- readMVar gTVarActionMap+ remReadVal (fst (aMap DM.! tVarId))+ )(\e -> (distInstanceError "readTVarValFromId" e) >> return "")++swapTVarValFromId :: VarID -> String -> IO String+swapTVarValFromId tVarId v' = CE.catch (do+ aMap <- readMVar gTVarActionMap+ remSwapVal (fst (aMap DM.! tVarId)) v'+ )(\e -> (distInstanceError "swapTVarValFromId" e) >> return "")++lockTVarFromId :: VarID -> IO ()+lockTVarFromId tVarId = CE.catch (do+ aMap <- readMVar gTVarActionMap+ remLock (fst (aMap DM.! tVarId))+ return ()+ )(distInstanceError "lockTVarFromId")++unLockTVarFromId :: VarID -> IO ()+unLockTVarFromId tVarId = CE.catch (do+ aMap <- readMVar gTVarActionMap+ remUnLock (fst (aMap DM.! tVarId))+ )(distInstanceError "unLockTVarFromId")++validateTVarFromId :: (VarID, VersionID) -> IO Bool+validateTVarFromId (tVarId, versionId) = CE.catch (do+ aMap <- readMVar gTVarActionMap+ answer <- remValidate (fst (aMap DM.! tVarId)) versionId+ return (read answer)+ )(\e -> (distInstanceError "validateTVarFromId" e) >> return False)++extWaitQFromId :: (VarID, RetryVar) -> IO ()+extWaitQFromId (tVarId, retryVar) = CE.catch (do+ aMap <- readMVar gTVarActionMap+ remExtWaitQ (fst (aMap DM.! tVarId)) retryVar+ )(distInstanceError "extWaitQFromId")++notifyFromId :: VarID -> IO ()+notifyFromId tVarId = CE.catch (do+ aMap <- readMVar gTVarActionMap+ forkIO (CE.catch (remNotify (fst (aMap DM.! tVarId))) + (\(e::CE.SomeException) -> debugStrLn1 ("notifyFromId " ++ show e)))+ return ()+ )(distInstanceError "notifyFromId")++distInstanceError :: String -> CE.SomeException -> IO ()+distInstanceError s e = do+ putStrLn ("\nDSTM Program Error in " ++ s ++ + "\nPossibly missing Dist instance for TVar data type")+ CE.throw e+ ++-----------------+-- Post Office --+-----------------++remPutMsg :: Show a => EnvAddr -> a -> IO ()+remPutMsg env msg = CE.catch (do+ h <- connectToEnvHold env+ sendTCP msg h+ )(propagateEx "remPutMsg")++remGetMsg :: Show a => EnvAddr -> a -> IO String+remGetMsg env msg = CE.catch (+ connectAtomicToEnv env (recvTCP msg)+ )(propagateEx "remGetMsg")++----------------------+-- Remote Messaging --+----------------------++type TransID = (EnvAddr, STMID) ++data RemCont = Com | Ret | End | Err + deriving (Show, Read, Eq)++data STMMessage = RemResume [VarID] -- RetryVarIDs+ | RemAddEnvToAction VarID EnvAddr+ | RemDelEnvFromAction VarID EnvAddr+ | RemLifeCheck+ | RemReadTVar VarID EnvAddr+ | RemStartTrans TransID [ValidRemVal] [EnvAddr]+ | RemContTrans TransID RemCont+ | RemElectedNewCoord TransID RemCont [EnvAddr]+ deriving (Show, Read)++-------------------------------+-- Class Dist TVar instances --+-------------------------------++instance Dist a => Dist (TVar a) where+ finTVars (TVar _ _ _ _) = return () -- internal error++ finTVars remTVar@(LinkTVar linkTVar) = CE.catch (do+ addFinalizer remTVar (finalizeTVar linkTVar)+ )(\(e::CE.SomeException) -> debugStrLn1 ("### Error finTVars: " ++ show linkTVar ++ show e)) --return ()) -- ignore errors in dead process TVars++-- remember locally which (1.) node(s) are using my host TVar+ regTVars destEnv tVar@(TVar _ _ _ _) = insertTVarActions tVar destEnv++-- make linkTVar node remember locally that dest node is using it+ regTVars destEnv (LinkTVar (VarLink tEnv tVarId)) = CE.catch (+ remPutMsg tEnv (RemAddEnvToAction tVarId destEnv)+ )(\(e::CE.SomeException) -> debugStrLn1 ("### Error regTVars: " ++ show tEnv ++ " -> destEnv: " ++ show destEnv ++ show e) )--return ()) -- ignore errors in regTVar (to not prohibit remRead)++-- make linkTVar node forget locally that my node was using it (enable GC)+finalizeTVar :: VarLink -> IO ()+finalizeTVar (VarLink tEnv tVarId) = do+ remPutMsg tEnv (RemDelEnvFromAction tVarId gMyEnv)++---------------------------+-- TVar Robustness Check --+---------------------------++-- |'isDistErrTVar' @e@ @tVar@ checks whether @tVar@ is unreachable when +-- exception @e@ had been raised. It returns 'True' if the exception raised +-- denotes @tVar@ as unreachable, 'False' otherwise. A TVar returning 'True' +-- once will never return a 'False' check result.+isDistErrTVar :: SomeDistTVarException -> TVar a -> Bool+isDistErrTVar e tVar = distTVarExEnv e == tVarEnv tVar+
+ Control/Distributed/STM/Utils.hs view
@@ -0,0 +1,39 @@+{-# OPTIONS_HADDOCK hide #-}++module Control.Distributed.STM.Utils ((>>+), (+>>), insertWith, showJustify,+ replaceListElems, updateListElems) where++(>>+) :: IO Bool -> IO Bool -> IO Bool+a1 >>+ a2 = do+ b <- a1+ if b then a2+ else return False++(+>>) :: Bool -> IO Bool -> IO Bool+b +>> a = if b then a else return False++{- insertWith key new update keyvalues+inserts sorted (key, new) into keyvalues+if key exists calls (update new old) at key -}++insertWith :: Ord a => (b -> b -> b) -> a -> b -> [(a, b)] -> [(a, b)]+insertWith _ key new [] = [(key, new)]+insertWith upd key new ((k, old) : kvs)+ | key == k = (k, upd new old) : kvs+ | key > k = (k, old) : insertWith upd key new kvs+ | otherwise = (key, new) : (k, old) : kvs++replaceListElems :: Eq a => b -> a -> [(a,b)] -> [(a,b)]+replaceListElems val key [] = [(key, val)]+replaceListElems val key ((k,v):kvs) | key == k = (k, val) : kvs+ | otherwise = (k,v)+ :replaceListElems val key kvs++updateListElems :: Eq a => (b -> b) -> b -> a -> [(a,b)] -> [(a,b)]+updateListElems _ val key [] = [(key, val)]+updateListElems f v0 key ((k,v):kvs) | key == k = (k, f v) : kvs+ | otherwise = (k,v)+ :updateListElems f v0 key kvs++showJustify :: Show a => a -> Int -> String+showJustify i len = let s = show i in (replicate (len - length s) ' ') ++ s
+ DSTM.cabal view
@@ -0,0 +1,82 @@+Name: DSTM+Version: 0.1+Copyright: (c) 2010, Frank Kupke+License: LGPL+License-File: LICENSE+Author: Frank Kupke+Maintainer: frk@informatik.uni-kiel.de+Cabal-Version: >= 1.2.3+Stability: provisional+Synopsis: A framework for using STM within distributed systems+Description:+{-+The DSTM package consists of the DSTM library, a name server application, and+three sample distributed programs using the library. DSTM is a framework+enabling the use of the STM interface, known from concurrent programming, to be+used for distributed Haskell applications as well. Provided are a simple Dining+Philosophers, a Chat, and a soft real-time Bomberman game application.++Distributed communication is transparent to the application programmer. The+application designer uses a very simple nameserver mechanism to set up the +system. The DSTM library includes the management of unavailable process nodes+and provides the application with abstract error information thus facilitating +the implementation of robust distributed application programs.++For usage please look into the included file: DSTMManual.pdf.+-}+Category: Distributed Computing+Build-Type: Simple+Tested-With: GHC ==6.10++Data-Files: field.txt, input1.txt, input2.txt, input3.txt+Data-Dir: Bomberman+Extra-source-files: README LICENSE DSTMManual.pdf++Library+ Build-Depends: base >=4 && <5, unix, process, network, haskell98, containers++ GHC-Options: -Wall++ Exposed-Modules:+ Control.Distributed.STM.DSTM++ Other-Modules:+ Control.Distributed.STM.DebugBase,+ Control.Distributed.STM.DebugDSTM,+ Control.Distributed.STM.Dist+ Control.Distributed.STM.EnvAddr,+ Control.Distributed.STM.NameService,+ Control.Distributed.STM.RetryVar,+ Control.Distributed.STM.TVar,+ Control.Distributed.STM.STM,+ Control.Distributed.STM.Utils,+ Paths_DSTM++ Extensions: DeriveDataTypeable,+ ScopedTypeVariables++Executable NameServer+ Main-Is: NameServer.hs+ Other modules: DSTM, NameService+ Build-Depends: base >=4 && <5, unix, process, network, haskell98, containers+ Hs-Source-Dirs: Nameserver, .++Executable Phil+ Main-Is: Phil.hs+ Build-Depends: base >=4 && <5, unix, process, network, haskell98, containers+ Hs-Source-Dirs: Phil, .++Executable ChatClient+ Main-Is: ChatClient.hs+ Build-Depends: base >=4 && <5, unix, process, network, haskell98, containers+ Hs-Source-Dirs: Chat, .++Executable ChatServer+ Main-Is: ChatServer.hs+ Build-Depends: base >=4 && <5, unix, process, network, haskell98, containers+ Hs-Source-Dirs: Chat, .++Executable Bomberman+ Main-Is: Bomberman.hs+ Build-Depends: base >=4 && <5, unix, process, network, haskell98, containers+ Hs-Source-Dirs: Bomberman, .
+ DSTMManual.pdf view
binary file changed (absent → 192685 bytes)
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2010, Frank Kupke+All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are+met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Frank Kupke nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Nameserver/NameServer.hs view
@@ -0,0 +1,6 @@+module Main (main) where++import Control.Distributed.STM.NameService++main :: IO ()+main = nameService gDefaultNameServer
+ Phil/Phil.hs view
@@ -0,0 +1,84 @@+module Main where++import Control.Distributed.STM.DebugDSTM+import Control.Distributed.STM.DSTM+import Prelude hiding (catch)+import System hiding (system)+import System.IO hiding (hPutStrLn)++type Stick = TVar Bool++eatings = 200++takeStick :: Stick -> Int -> STM ()+takeStick s n = do+ b <- readTVar s+ if b + then writeTVar s False+ else retry++putStick :: Stick -> STM ()+putStick s = do+ b <- readTVar s+ if False+ then error "Stick already released?"+ else writeTVar s True++phil :: Int -> Int -> Stick -> Stick -> IO ()+phil m n l r = do+ putStrLn ("*** "++show n ++ ". Phil thinks")+ atomic $ do+ takeStick l n+ takeStick r n+ putStrLn ("+++ "++show n ++ ". Phil eats")+ atomic $ do+ putStick l+ putStick r+ if m>0+ then print ("run: "++show m) >>phil (m-1) n l r+ else print ("Ready: "++show n) >>+ getLine >> return ()++{-+startPhils :: Int -> IO ()+startPhils n = do+ sync <- newEmptyMVar+ ioSticks <- atomically $ do+ sticks <- mapM (const (newTVar True)) [1..n]+ return sticks+ mapM_ (\(l,r,i)->forkIO (phil eatings sync i l r)) + (zip3 ioSticks (tail ioSticks) [1..n-1])+ forkIO (phil eatings sync n (last ioSticks) (head ioSticks))+ takeMVar sync+ takeMVar sync+ takeMVar sync+ takeMVar sync+ takeMVar sync-}++ +main :: IO ()+--main = startPhils 5+main = do+ startDist goPhil++goPhil = do+ (arg:_) <- getArgs+ let n= read arg+ myStick <- atomic $ newTVar True+ registerTVar gDefaultNameServer myStick arg+ putStrLn ("registered stick nr: "++ arg)+ getLine+ otherStick <- lookupLoop ((n `mod` 3) + 1)+ putStrLn ("looked-up other stick nr: " ++ (show ((n `mod` 3)+1)))+ getLine+ phil eatings n myStick otherStick++lookupLoop :: Int -> IO Stick+lookupLoop m = do+ putStrLn ("lookup stick nr: "++ show m)+ maybeStick <- lookupTVar gDefaultNameServer (show m)+ print maybeStick+ --getLine+ case maybeStick of+ Nothing -> lookupLoop m+ Just stick -> return stick
+ README view
@@ -0,0 +1,53 @@+The Cabal library package+=========================++Installation instructions for the DSTM library+===============================================++Installing as a user (no root or administer access)+---------------------------------------------------++ ghc --make Setup+ ./Setup configure --user+ ./Setup build+ ./Setup install++Note the use of the `--user` flag at the configure step.++Compiling Setup rather than using `runghc Setup` is much faster and works on+Windows. For all packages other than Cabal itself it is fine to use `runghc`.++This will install into `$HOME/.cabal/` on unix and into+`$Documents and Settings\$User\Application Data\cabal\` on Windows+If you want to install elsewhere use the `--prefix=` flag at the+configure step.+++Installing as root / Administrator+----------------------------------++ ghc --make Setup+ ./Setup configure+ ./Setup build+ sudo ./Setup install++Compiling Setup rather than using `runghc Setup` is much faster and works on+Windows. For all packages other than Cabal itself it is fine to use `runghc`.++This will install into `/usr/local` on unix and on Windows it will+install into `$ProgramFiles/Haskell`. If you want to install+elsewhere use the `--prefix=` flag at the configure step.++Bugs+=======++Please report bugs and wish-list items to:++frk@informatik.uni-kiel.de+++Your Help+---------++To help us in the next round of development work it would be+enormously helpful to know from our users what their user experience is.
+ Setup.hs view
@@ -0,0 +1,10 @@+import Distribution.Simple+main :: IO ()+main = defaultMain++-- Although this looks like the Simple build type, it is in fact vital that+-- we use this Setup.hs because it'll get compiled against the local copy+-- of the Cabal lib, thus enabling Cabal to bootstrap itself without relying+-- on any previous installation. This also means we can use any new features+-- immediately because we never have to worry about building Cabal with an+-- older version of itself.