diff --git a/Bomberman/Bomberman.hs b/Bomberman/Bomberman.hs new file mode 100644 --- /dev/null +++ b/Bomberman/Bomberman.hs @@ -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) diff --git a/Bomberman/field.txt b/Bomberman/field.txt new file mode 100644 --- /dev/null +++ b/Bomberman/field.txt @@ -0,0 +1,10 @@ +xxxxxxxxxx +xoooooooox +xooxxxxoox +xooxxxxoox +xooxxxxoox +xooxxxoxox +xoooxoooox +xoooxoooox +xoooooooox +xxxooxxxxx diff --git a/Bomberman/input1.txt b/Bomberman/input1.txt new file mode 100644 --- /dev/null +++ b/Bomberman/input1.txt @@ -0,0 +1,1 @@ +        Q diff --git a/Bomberman/input2.txt b/Bomberman/input2.txt new file mode 100644 --- /dev/null +++ b/Bomberman/input2.txt @@ -0,0 +1,1 @@ +       Q diff --git a/Bomberman/input3.txt b/Bomberman/input3.txt new file mode 100644 --- /dev/null +++ b/Bomberman/input3.txt @@ -0,0 +1,1 @@ +       Q diff --git a/Chat/ChatClient.hs b/Chat/ChatClient.hs new file mode 100644 --- /dev/null +++ b/Chat/ChatClient.hs @@ -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 diff --git a/Chat/ChatServer.hs b/Chat/ChatServer.hs new file mode 100644 --- /dev/null +++ b/Chat/ChatServer.hs @@ -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))] diff --git a/Control/Distributed/STM/DSTM.hs b/Control/Distributed/STM/DSTM.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/STM/DSTM.hs @@ -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" diff --git a/Control/Distributed/STM/DebugBase.hs b/Control/Distributed/STM/DebugBase.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/STM/DebugBase.hs @@ -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 " + diff --git a/Control/Distributed/STM/DebugDSTM.hs b/Control/Distributed/STM/DebugDSTM.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/STM/DebugDSTM.hs @@ -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 diff --git a/Control/Distributed/STM/Dist.hs b/Control/Distributed/STM/Dist.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/STM/Dist.hs @@ -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 diff --git a/Control/Distributed/STM/EnvAddr.hs b/Control/Distributed/STM/EnvAddr.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/STM/EnvAddr.hs @@ -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)) diff --git a/Control/Distributed/STM/NameService.hs b/Control/Distributed/STM/NameService.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/STM/NameService.hs @@ -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") diff --git a/Control/Distributed/STM/RetryVar.hs b/Control/Distributed/STM/RetryVar.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/STM/RetryVar.hs @@ -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") diff --git a/Control/Distributed/STM/STM.hs b/Control/Distributed/STM/STM.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/STM/STM.hs @@ -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 + diff --git a/Control/Distributed/STM/TVar.hs b/Control/Distributed/STM/TVar.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/STM/TVar.hs @@ -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 + diff --git a/Control/Distributed/STM/Utils.hs b/Control/Distributed/STM/Utils.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/STM/Utils.hs @@ -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 diff --git a/DSTM.cabal b/DSTM.cabal new file mode 100644 --- /dev/null +++ b/DSTM.cabal @@ -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, . diff --git a/DSTMManual.pdf b/DSTMManual.pdf new file mode 100644 Binary files /dev/null and b/DSTMManual.pdf differ diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Nameserver/NameServer.hs b/Nameserver/NameServer.hs new file mode 100644 --- /dev/null +++ b/Nameserver/NameServer.hs @@ -0,0 +1,6 @@ +module Main (main) where + +import Control.Distributed.STM.NameService + +main :: IO () +main = nameService gDefaultNameServer diff --git a/Phil/Phil.hs b/Phil/Phil.hs new file mode 100644 --- /dev/null +++ b/Phil/Phil.hs @@ -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 diff --git a/README b/README new file mode 100644 --- /dev/null +++ b/README @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -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.