Holumbus-Distribution 0.0.1.1 → 0.1.0

raw patch · 56 files changed

+4484/−168 lines, 56 filesdep +arraydep +paralleldep +randomdep −editlinedep ~basenew-component:exe:PortRegistryDaemonPVP ok

version bump matches the API change (PVP)

Dependencies added: array, parallel, random, readline, stm

Dependencies removed: editline

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• Examples/ClientServer/Makefile +2/−2
• Examples/Distribution/ChatDChan/ClientDChan.hs +134/−0
• Examples/Distribution/ChatDChan/Makefile +27/−0
• Examples/Distribution/ChatDChan/MessagesDChan.hs +82/−0
• Examples/Distribution/ChatDChan/ServerDChan.hs +143/−0
• Examples/Distribution/ChatDFunction/ClientDFunction.hs +178/−0
• Examples/Distribution/ChatDFunction/InterfacesDFunction.hs +141/−0
• Examples/Distribution/ChatDFunction/Makefile +27/−0
• Examples/Distribution/ChatDFunction/ServerDFunction.hs +129/−0
• Examples/Distribution/CounterDFunction/Makefile +27/−0
• Examples/Distribution/CounterDFunction/MasterCounter.hs +42/−0
• Examples/Distribution/CounterDFunction/SlaveCounter.hs +31/−0
• Examples/Distribution/CounterDMVar/ErrorSlaveCounter.hs +24/−0
• Examples/Distribution/CounterDMVar/Makefile +27/−0
• Examples/Distribution/CounterDMVar/MasterCounter.hs +30/−0
• Examples/Distribution/CounterDMVar/SlaveCounter.hs +33/−0
• Examples/Distribution/CounterDMVar/WaitSlaveCounter.hs +36/−0
• Examples/Distribution/HelloWorld/Makefile +27/−0
• Examples/Distribution/HelloWorld/Receiver.hs +40/−0
• Examples/Distribution/HelloWorld/Sender.hs +41/−0
• Examples/Distribution/Makefile +12/−0
• Examples/Makefile +1/−1
• Examples/Ports/Makefile +0/−4
• Examples/Ports/Single.hs +1/−0
• Holumbus-Distribution.cabal +55/−11
• Programs/PortRegistry/Makefile +0/−4
• Programs/PortRegistryDaemon/Makefile +26/−0
• Programs/PortRegistryDaemon/PortRegistryDaemon.hs +123/−0
• README +86/−76
• source/Holumbus/Common/Debug.hs +1/−0
• source/Holumbus/Common/FileHandling.hs +18/−29
• source/Holumbus/Common/Logging.hs +32/−4
• source/Holumbus/Common/Threading.hs +0/−0
• source/Holumbus/Common/Utils.hs +1/−3
• source/Holumbus/Console/Console.hs +1/−4
• source/Holumbus/Console/ServerConsole.hs +216/−0
• source/Holumbus/Data/AccuMap.hs +2/−2
• source/Holumbus/Data/KeyMap.hs +0/−1
• source/Holumbus/Distribution/DChan.hs +272/−0
• source/Holumbus/Distribution/DFunction.hs +224/−0
• source/Holumbus/Distribution/DMVar.hs +307/−0
• source/Holumbus/Distribution/DNode.hs +46/−0
• source/Holumbus/Distribution/DNode/Base.hs +970/−0
• source/Holumbus/Distribution/DNode/Network.hs +305/−0
• source/Holumbus/Distribution/DStreamPort.hs +194/−0
• source/Holumbus/Distribution/DValue.hs +122/−0
• source/Holumbus/Network/Communication.hs +47/−7
• source/Holumbus/Network/Core.hs +46/−12
• source/Holumbus/Network/DoWithServer.hs +136/−0
• source/Holumbus/Network/Messages.hs +3/−0
• source/Holumbus/Network/Port.hs +9/−7
• source/Holumbus/Network/PortRegistry.hs +1/−0
• source/Holumbus/Network/PortRegistry/Messages.hs +2/−1
• source/Holumbus/Network/PortRegistry/PortRegistryData.hs +1/−0
• source/Holumbus/Network/PortRegistry/PortRegistryPort.hs +1/−0
• source/Holumbus/Network/Site.hs +2/−0

@@ -1,8 +1,8 @@ # Making FileSystem-Standalone Programs -HOME		= ../..+# HOME		= ../.. -SOURCE		= .+# SOURCE		= .  GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT) GHC		= ghc $(GHC_FLAGS)

@@ -0,0 +1,134 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +-}++-- ----------------------------------------------------------------------------+++module Main(main) where++import           Prelude hiding (catch)++import           System.IO+import           Control.Concurrent+import           Control.Exception++import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DChan+import           Holumbus.Common.Logging++import           MessagesDChan++data ClientData = ClientData {+    cd_server :: DChan ChatRequest+  , cd_client :: DChan ChatResponse+  , cd_name   :: String+  , cd_token  :: ClientToken+  }+++main :: IO ()+main+  = do+    initializeLogging []+    initDNode $ defaultDNodeConfig ""+    addForeignDNode $ mkDNodeAddress "ServerDChan" "april" (fromInteger 7999)+    sc <- newRemoteDChan "server" "ServerDChan"+    +    -- addForeignDNodeHandler (mkDNodeId "ServerDChan") listener+    +    cc <- newDChan ""+    cd <- registerClient sc cc+    forkIO $ messageReader cd+    messageWriter cd+    unregisterClient cd+    closeDChan sc+    closeDChan cc+    deinitDNode+    +    +registerClient :: DChan ChatRequest -> DChan ChatResponse -> IO ClientData+registerClient sc cc+  = do+    putStr "please insert login name: "+    hFlush stdout+    cn <- getLine+    handle (\(SomeException e) -> do+      putStr $ show e+      putStrLn "ERROR: server not reachable"+      registerClient sc cc+      ) $+      do +      writeDChan sc (ReqRegister cn cc)+      rsp <- readDChan cc+      case rsp of+        (RspRegister ct) ->+          return $ ClientData sc cc cn ct+        (RspError e) ->+          do+          putStrLn $ "ERROR: " ++ e+          registerClient sc cc+        _ ->+          do+          putStrLn $ "ERROR: invalid server response" +          registerClient sc cc+++unregisterClient :: ClientData -> IO ()+unregisterClient cd+  = do+    handle (\(SomeException _) -> do return ()) $+      writeDChan (cd_server cd) (ReqUnregister (cd_token cd))+++messageReader :: ClientData -> IO ()+messageReader cd+  = do+    msg <- readDChan (cd_client cd)+    case msg of+      (RspMessage cn text) ->+        do+        putStrLn $ cn ++ "> " ++ text+      (RspError e) ->+        do+        putStr $ "ERROR: " ++ e+      _ ->+        do+        putStrLn $ "ERROR: invalid server response" +    messageReader cd+++messageWriter :: ClientData -> IO ()+messageWriter cd+  = do+    putStr $ (cd_name cd) ++ "> "+    hFlush stdout+    msg <- getLine+    case msg of+      "exit" -> +        return ()+      "debug" ->+        do+        nd <- getDNodeData+        putStrLn $ show nd+        messageWriter cd+      "" ->+        do+        messageWriter cd+      text ->+        do+        handle (\(SomeException _) -> do +          putStrLn "ERROR: Server unreachable, message not send"+          ) $ do+          writeDChan (cd_server cd) (ReqMessage (cd_token cd) text)+          messageWriter cd

@@ -0,0 +1,27 @@+# Making FileSystem-Standalone Programs++GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT)+GHC		= ghc $(GHC_FLAGS)++RM_FLAGS	= -rf+RM		= rm $(RM_FLAGS)++PROG		= ClientDChan ServerDChan++OUTPUT		= output++all : $(PROG)++#prof : $(PROG).hs+#	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+#	$(RM) $(OUTPUT)/*+#	$(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+	$(RM) $(OUTPUT)/*+	$(GHC) --make -o $@ $<++clean :+	$(RM) $(PROG) $(OUTPUT) *.port+	

@@ -0,0 +1,82 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +-}++-- ----------------------------------------------------------------------------++module MessagesDChan+(+    ChatRequest(..)+  , ChatResponse(..)+  +  , ClientToken+  , genClientToken+)+where++import           Data.Binary+--import           Holumbus.Common.MRBinary+import           System.Random+import           Holumbus.Distribution.DChan+++data ChatRequest+  = ReqRegister String (DChan ChatResponse)+  | ReqUnregister ClientToken+  | ReqMessage ClientToken String++instance Binary ChatRequest where+  put(ReqRegister cn ch)  = putWord8 1 >> put cn >> put ch+  put(ReqUnregister ct)   = putWord8 2 >> put ct+  put(ReqMessage ct text) = putWord8 3 >> put ct >> put text+  get+    = do+      t <- getWord8+      case t of+        1 -> get >>= \cn -> get >>= \ch -> return (ReqRegister cn ch)+        2 -> get >>= \ct -> return (ReqUnregister ct)+        3 -> get >>= \ct -> get >>= \text -> return (ReqMessage ct text)+        _ -> error "ChatRequest: wrong encoding"++data ChatResponse+  = RspRegister ClientToken+  | RspMessage String String+  | RspError String+  +instance Binary ChatResponse where+  put(RspRegister ct)     = putWord8 1 >> put ct+  put(RspMessage cn text) = putWord8 2 >> put cn >> put text+  put(RspError e)         = putWord8 3 >> put e+  get+    = do+      t <- getWord8+      case t of+        1 -> get >>= \ct -> return (RspRegister ct)+        2 -> get >>= \cn -> get >>= \text -> return (RspMessage cn text)+        3 -> get >>= \e -> return (RspError e)+        _ -> error "ChatResponse: wrong encoding"+++data ClientToken = ClientToken Integer Integer+  deriving (Show, Eq, Ord)++instance Binary ClientToken where+  put(ClientToken i rnd) = put i >> put rnd+  get = get >>= \i -> get >>= \rnd -> return (ClientToken i rnd)++genClientToken :: Integer -> IO ClientToken+genClientToken i+  = do+    rnd <- getStdRandom (randomR (1,1000000000))+    return $ ClientToken i rnd+  

@@ -0,0 +1,143 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +-}++-- ----------------------------------------------------------------------------++module Main(main) where++import           Control.Concurrent+import qualified Data.Map as Map+import           Data.Maybe+import qualified Data.Set as Set++import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DChan+import           Holumbus.Common.Logging++import           MessagesDChan++++data ServerData = ServerData {+    sd_clients   :: Map.Map ClientToken (String, (DChan ChatResponse))+  , sd_names     :: Set.Set String+  , sd_number    :: Integer+  }+++main :: IO ()+main+  = do+    initializeLogging []+    initDNode $ (defaultDNodeConfig "ServerDChan")+      { dnc_MinPort = (fromInteger 7999), dnc_MaxPort = (fromInteger 7999) }+    let serverdata = mkNewServer+    channel <- newDChan "server"+    forkIO $ dispatcher serverdata channel+    repeatTillReturn+    closeDChan channel+    deinitDNode++    ++mkNewServer :: ServerData+mkNewServer = ServerData Map.empty Set.empty 1+++mkNextClientToken :: ServerData -> IO (ServerData, ClientToken)+mkNextClientToken s+  = do+    let i  = sd_number s+        s' = s { sd_number = (i+1)}+    t <- genClientToken i+    return (s', t)++    +addClient :: ServerData -> ClientToken -> String -> DChan ChatResponse -> ServerData+addClient s ct cn cc = s { sd_clients = c' , sd_names = n' }+  where+  c' = Map.insert ct (cn,cc) (sd_clients s)+  n' = Set.insert cn (sd_names s)+++deleteClient :: ServerData -> ClientToken -> ServerData+deleteClient s ct = s { sd_clients = c' , sd_names = n' }+  where+  c' = Map.delete ct (sd_clients s)+  n' = case (Map.lookup ct (sd_clients s)) of+          (Just (cn,_)) -> Set.delete cn (sd_names s)+          (Nothing)     -> (sd_names s)++isTokenValid :: ServerData -> ClientToken -> Bool+isTokenValid s ct = Map.member ct (sd_clients s) ++    +isNameValid :: ServerData -> String -> Bool+isNameValid _ "" = False+isNameValid s cn = (not $ Set.member cn (sd_names s))+++sendMessage :: ServerData -> ClientToken -> String -> IO ()+sendMessage s ct text+  = do+    if (isTokenValid s ct)+      then do+        let (cn, _) = fromJust $ Map.lookup ct (sd_clients s)+            ccs = map (snd) $ Map.elems $ Map.delete ct (sd_clients s)+        mapM (\cc -> writeDChan cc (RspMessage cn text)) ccs+        return () +      else return ()+      ++repeatTillReturn :: IO ()+repeatTillReturn+  = do+    hnd <- getDNodeData+    putStrLn $ show hnd+    l <- getLine+    case l of+      "exit" -> +        return ()+      "debug" ->+        do+        nd <- getDNodeData+        putStrLn $ show nd+      _  ->+        repeatTillReturn++            +dispatcher :: ServerData -> DChan ChatRequest -> IO ()+dispatcher s c+  = do+    msg <- readDChan c+    s' <- case msg of+      (ReqRegister cn cc) ->+        do+        if (isNameValid s cn)+          then do+            (s', ct) <- mkNextClientToken s +            writeDChan cc (RspRegister ct)+            return $ addClient s' ct cn cc+          else do+            writeDChan cc (RspError "name already used")+            return s+      (ReqUnregister ct) -> +        do+        return $ deleteClient s ct +      (ReqMessage ct text) ->+        do+        sendMessage s ct text+        return s+    dispatcher s' c    +      

@@ -0,0 +1,178 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +-}++-- ----------------------------------------------------------------------------++{-# OPTIONS_GHC -XDeriveDataTypeable #-}+module Main(main) where++import           Control.Concurrent+import           Control.Exception+import           Data.Typeable+import           System.IO++import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DFunction+import           Holumbus.Common.Logging++import           InterfacesDFunction+++data ConnectionException = ConnectionException+  deriving(Typeable, Show)+instance Exception ConnectionException+++data ClientData = ClientData {+    cd_DNodeId :: DNodeId+  , cd_id      :: Maybe Int+  , cd_tid     :: ThreadId+  , cd_name    :: String+  , cd_stub    :: ClientInterfaceStub+  , cd_server  :: RemoteServerInterface+  , cd_exit    :: IO ()+  }++type Client = MVar ClientData+++-- what to do when we see the server for the first time after a disconnection+positiveHandlerFunction :: Client -> DHandlerId -> IO ()+positiveHandlerFunction cc _+  = do+    putStrLn "positive Trigger"+    client <- takeMVar cc+    clearStdIn+    hFlush stdout+    -- kill the wait thread+    throwTo (cd_tid client) ConnectionException+    -- start the chat thread+    tid <- forkIO $ chatLoop cc+    putMVar cc $ client { cd_tid = tid }+++-- what to do when the server becomes unreachable+negativeHandlerFunction :: Client -> DHandlerId -> IO ()+negativeHandlerFunction cc _+  = do+    putStrLn "negative Trigger"+    client <- takeMVar cc+    -- kill the chat thread+    throwTo (cd_tid client) ConnectionException+    -- start the wait thread+    tid <- forkIO $ waitLoop+    putMVar cc $ client { cd_id = Nothing, cd_tid = tid }+    ++mkChatClient :: DNodeId -> ThreadId -> ClientInterfaceStub -> RemoteServerInterface -> IO () -> IO Client+mkChatClient myDNodeId tid cifs rsif endfun+  = do+    newMVar $ ClientData myDNodeId Nothing tid "" cifs rsif endfun+++clearStdIn :: IO()+clearStdIn+  = do+    b <- hReady stdin+    if b+      then do+        _ <- getChar+        clearStdIn+      else do +        return ()    +++generateExitFunktions :: IO (IO (), IO ())+generateExitFunktions+  = do+    quitMarker <- newEmptyMVar+    return (waitForTermination quitMarker, terminateProgram quitMarker)+    where+    waitForTermination :: MVar () -> IO ()+    waitForTermination qm+      = do+        _ <- takeMVar qm+        return ()+    terminateProgram :: MVar () -> IO ()+    terminateProgram qm+      = do+        putMVar qm ()+++waitLoop :: IO ()+waitLoop+  = handle (\ConnectionException -> return ()) $+      do+      putStrLn "waiting for server..."+      hFlush stdout+      threadDelay 10000000+      waitLoop+++chatLoop :: Client -> IO ()+chatLoop cc+  = handle (\ConnectionException -> return ()) $+      do+      client <- readMVar cc+      case (cd_id client) of+        (Just cid) -> do+          -- putStr $ (cd_name client) ++ "> "+          hFlush stdout+          msg <- getLine+          case msg of+            "exit" -> do+              (sif_unregister $ cd_server client) cid+              (cd_exit client)+            _ -> (sif_send $ cd_server client) cid msg+        Nothing -> do+          putStr "login with name: "+          hFlush stdout+          cn <- getLine+          mbId <- (sif_register $ cd_server client) cn (cd_DNodeId client) (cd_stub client)+          client' <- takeMVar cc+          putMVar cc $ client' {cd_id = mbId, cd_name = cn}+      chatLoop cc+++receiveChatMessage :: ReceiveChatMessageFunction+receiveChatMessage cn msg+  = do+    putStrLn $ "\n" ++ cn ++ "> " ++ msg++++main :: IO ()+main+  = do+    initializeLogging []+    myDNodeId <- initDNode $ defaultDNodeConfig ""+    addForeignDNode $ mkDNodeAddress "ChatServer" "april" (fromInteger 7999)+    (waitForTermination, terminateProgram) <- generateExitFunktions+    -- start with the wait loop+    tid <- forkIO waitLoop+    -- create chat client data strukture+    cifs <- createLocalClientInterfaceStub receiveChatMessage+    rsif <- createRemoveServerInterface "ChatServer"+    cc <- mkChatClient myDNodeId tid cifs rsif terminateProgram+    -- register the handlers which will controll the existence of the server+    let dni = mkDNodeId "ChatServer"+    addForeignDNodeHandler True dni (positiveHandlerFunction cc)+    addForeignDNodeHandler False dni (negativeHandlerFunction cc)+    -- wait for the chatLoop to stop by the user+    waitForTermination+    -- clean up+    closeLocalClientInterfaceStub cifs+    deinitDNode++

@@ -0,0 +1,141 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +-}++-- ----------------------------------------------------------------------------++module InterfacesDFunction+(+    RegisterClientFunction+  , UnregisterClientFunction+  , SendChatMessageFunction++  , ServerInterfaceStub+  , RemoteServerInterface(..)+  , createLocalServerInterfaceStub+  , closeLocalServerInterfaceStub+  , createRemoveServerInterface++++  , ReceiveChatMessageFunction++  , ClientInterfaceStub+  , RemoteClientInterface(..)+  , createLocalClientInterfaceStub+  , closeLocalClientInterfaceStub+  , createRemoteClientInterface+)+where++import           Data.Binary++import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DFunction+++-- ----------------------------------------------------------------------------+-- the servers' interface+-- ----------------------------------------------------------------------------++-- functions of the chat server+type RegisterClientFunction = String -> DNodeId -> ClientInterfaceStub -> IO (Maybe Int)+type UnregisterClientFunction = Int -> IO ()+type SendChatMessageFunction = Int -> String -> IO ()++-- in theory this could be made binary, but we don't need this+data ServerInterfaceStub = ServerInterfaceStub {+    sifs_register   :: DFunction RegisterClientFunction+  , sifs_unregister :: DFunction UnregisterClientFunction+  , sifs_send       :: DFunction SendChatMessageFunction+  }++-- this cannot be binary, but it is much nicer to call the real functions+data RemoteServerInterface = RemoteServerInterface {+    sif_register   :: RegisterClientFunction+  , sif_unregister :: UnregisterClientFunction+  , sif_send       :: SendChatMessageFunction+  }++createLocalServerInterfaceStub+  :: RegisterClientFunction+  -> UnregisterClientFunction+  -> SendChatMessageFunction+  -> IO ServerInterfaceStub+createLocalServerInterfaceStub f1 f2 f3+  = do+    df1 <- newDFunction "register" f1+    df2 <- newDFunction "unregister" f2+    df3 <- newDFunction "send" f3+    return (ServerInterfaceStub df1 df2 df3)++closeLocalServerInterfaceStub :: ServerInterfaceStub -> IO ()+closeLocalServerInterfaceStub (ServerInterfaceStub df1 df2 df3)+  = do+    closeDFunction df1+    closeDFunction df2+    closeDFunction df3++createRemoveServerInterface :: String -> IO RemoteServerInterface+createRemoveServerInterface sn+  = do+    f1 <- mkNamedServerFunction "register" sn+    f2 <- mkNamedServerFunction "unregister" sn+    f3 <- mkNamedServerFunction "send" sn+    return (RemoteServerInterface f1 f2 f3)+    where+    mkNamedServerFunction :: (BinaryFunction a) => String -> String -> IO a+    mkNamedServerFunction fn sn+      = do+        df <- newRemoteDFunction fn sn+        return (accessDFunction df)+++-- ----------------------------------------------------------------------------+-- the client's interface+-- ----------------------------------------------------------------------------++-- functions of the client+type ReceiveChatMessageFunction = String -> String -> IO ()++-- this needs to be binary, because we send this to the server+data ClientInterfaceStub = ClientInterfaceStub {+    cifs_receive :: DFunction ReceiveChatMessageFunction+  }++instance Binary ClientInterfaceStub where+  put (ClientInterfaceStub f1) = put f1+  get = get >>= \f1 -> return (ClientInterfaceStub f1)++-- this cannot be binary, but it is much nicer to call the real functions+data RemoteClientInterface = RemoteClientInterface {+    cif_receive :: ReceiveChatMessageFunction+  }++createLocalClientInterfaceStub :: ReceiveChatMessageFunction -> IO ClientInterfaceStub+createLocalClientInterfaceStub f1+  = do+    df1 <- newDFunction "receive" f1+    return (ClientInterfaceStub df1)++closeLocalClientInterfaceStub :: ClientInterfaceStub -> IO ()+closeLocalClientInterfaceStub (ClientInterfaceStub df1)+  = do+    closeDFunction df1++createRemoteClientInterface :: ClientInterfaceStub -> RemoteClientInterface+createRemoteClientInterface (ClientInterfaceStub df1)+  = (RemoteClientInterface f1)+    where+    f1 = (accessDFunction df1)+

@@ -0,0 +1,27 @@+# Making FileSystem-Standalone Programs++GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT)+GHC		= ghc $(GHC_FLAGS)++RM_FLAGS	= -rf+RM		= rm $(RM_FLAGS)++PROG		= ClientDFunction ServerDFunction++OUTPUT		= output++all : $(PROG)++#prof : $(PROG).hs+#	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+#	$(RM) $(OUTPUT)/*+#	$(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+	$(RM) $(OUTPUT)/*+	$(GHC) --make -o $@ $<++clean :+	$(RM) $(PROG) $(OUTPUT) *.port+	

@@ -0,0 +1,129 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +-}++-- ----------------------------------------------------------------------------++module Main(main) where++import           Control.Concurrent+import qualified Data.Map as Map+import           Data.Maybe+import qualified Data.Set as Set++import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DFunction+import           Holumbus.Common.Logging++import           InterfacesDFunction++++data ServerData = ServerData {+    sd_clients   :: Map.Map Int (String, RemoteClientInterface)+  , sd_names     :: Set.Set String+  , sd_number    :: Int+  }++type Server = MVar ServerData+++-- what to do when a client becomes unreachable+negativeHandlerFunction :: Server -> Int -> DHandlerId -> IO ()+negativeHandlerFunction cs cId hdlId+  = do+    putStrLn "negative Trigger"+    unregisterClient cs cId+    -- delete this trigger, it won't be called again+    delForeignHandler hdlId+++registerClient :: Server -> RegisterClientFunction+registerClient _ "" _ _ = return Nothing+registerClient cs cn dni stub+  = do+    putStrLn $ "registering: " ++ cn+    server <- takeMVar cs+    if (Set.member cn (sd_names server))+      then do+        putMVar cs server+        return Nothing+      else do+        let sd_clients' = Map.insert (sd_number server) (cn, createRemoteClientInterface stub) (sd_clients server)+            sd_names'   = Set.insert cn (sd_names server)+            sd_number'  = 1 + (sd_number server)+        putMVar cs $ server { sd_clients = sd_clients', sd_names = sd_names', sd_number = sd_number' }+        -- install a handler to remove the client when he becomes unreachable+        addForeignDNodeHandler False dni (negativeHandlerFunction cs (sd_number server))+        return (Just $ sd_number server)+++unregisterClient :: Server -> UnregisterClientFunction+unregisterClient cs cId+  = do+    putStrLn $ "unregistering: " ++ (show cId)+    server <- takeMVar cs+    case (Map.lookup cId (sd_clients server)) of+      Nothing -> do+        putMVar cs server+        return ()+      (Just (cn,_)) -> do+        let sd_clients' = Map.delete cId (sd_clients server)+            sd_names'   = Set.delete cn (sd_names server)+        putMVar cs $ server { sd_clients = sd_clients', sd_names = sd_names' }+++sendChatMessage :: Server -> SendChatMessageFunction+sendChatMessage cs cId msg+  = do+    putStrLn $ "sending: " ++ (show cId) ++ "> " ++ msg+    server <- readMVar cs+    case (Map.lookup cId (sd_clients server)) of+      Nothing -> do+        -- TODO return true / false to tell the client that his request was accepted or not+        putStrLn $ "warning: unnkown client with id " ++ (show cId) ++ " wants to send message: " ++ msg+        return ()+      (Just (cn,_)) -> do+        -- TODO exception handling here+        let funs = map (\t -> (cif_receive $ snd t) cn msg) $ Map.elems $ Map.filterWithKey (\cId' _ -> cId' /= cId) (sd_clients server)+        sequence_ funs+++mkChatServer :: IO Server+mkChatServer = newMVar $ ServerData Map.empty Set.empty 1+++main :: IO ()+main+  = do+    initializeLogging []+    initDNode $ (defaultDNodeConfig "ChatServer")+      { dnc_MinPort = (fromInteger 7999), dnc_MaxPort = (fromInteger 7999) }+    cs <- mkChatServer+    sifs <- createLocalServerInterfaceStub+      (registerClient cs)+      (unregisterClient cs)+      (sendChatMessage cs)+    inputLoop+    closeLocalServerInterfaceStub sifs+    deinitDNode+++inputLoop :: IO ()+inputLoop+  = do+    msg <- getLine+    case msg of+      "exit" -> return ()+      _ -> inputLoop+

@@ -0,0 +1,27 @@+# Making FileSystem-Standalone Programs++GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT)+GHC		= ghc $(GHC_FLAGS)++RM_FLAGS	= -rf+RM		= rm $(RM_FLAGS)++PROG		= MasterCounter SlaveCounter++OUTPUT		= output++all : $(PROG)++#prof : $(PROG).hs+#	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+#	$(RM) $(OUTPUT)/*+#	$(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+	$(RM) $(OUTPUT)/*+	$(GHC) --make -o $@ $<++clean :+	$(RM) $(PROG) $(OUTPUT) *.port+	

@@ -0,0 +1,42 @@++module Main (main) where++import           Control.Concurrent.MVar++import           Holumbus.Common.Logging+import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DFunction+++main :: IO ()+main+  = do+    initializeLogging []+    initDNode $ (defaultDNodeConfig "Master")+      { dnc_MinPort = (fromInteger 7999), dnc_MaxPort = (fromInteger 7999) }+    mv <- newMVar 0+    df <- newDFunction "doCounting" (doCounting mv) +    doWatching mv+    closeDFunction df+    deinitDNode+++doCounting :: MVar Int -> Int -> IO Int+doCounting mv i+  = do+    modifyMVar mv $ \v -> +      do+      let v' = v + i+      return (v',v')+++doWatching :: MVar Int -> IO ()+doWatching mv+  = do+    l <- getLine+    case  l of+      "exit" -> return ()+      _ -> do+        v <- readMVar mv+        putStrLn $ "current Value: " ++ (show v)+        doWatching mv

@@ -0,0 +1,31 @@++module Main(main) where+                 +import           Holumbus.Common.Logging+import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DFunction+++main :: IO ()+main+  = do+    initializeLogging []+    initDNode $ defaultDNodeConfig ""+    addForeignDNode $ mkDNodeAddress "Master" "april" (fromInteger 7999)+    df <- newRemoteDFunction "doCounting" "Master"+    -- addForeignDNodeHandler (mkDNodeId "Master") listener+    doCounting (accessDFunction df)+    closeDFunction df+    deinitDNode+++doCounting :: (Int -> IO Int) -> IO ()+doCounting f+  = do+    l <- getLine+    case  l of+      "exit" -> return ()+      _ -> do+        i <- f 1+        putStrLn $ "current Value: " ++ (show i)+        doCounting f

@@ -0,0 +1,24 @@++module Main(main) where+                 +import           Holumbus.Common.Logging+import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DMVar+++main :: IO ()+main+  = do+    initializeLogging []+    initDNode $ defaultDNodeConfig ""+    addForeignDNode $ mkDNodeAddress "Master" "april" (fromInteger 7999)+    c <- (newRemoteDMVar "counter" "Master")::IO (DMVar Int) +    +    +    -- just take the value and leave... +    -- in a good system, this shouldn't block the other counters+    i <- takeDMVar c+    putStrLn $ "current Value: " ++ show i ++ " -> and leaving!!!" +    -- closeDMVar c+    -- deinitDNode+    return ()

@@ -0,0 +1,27 @@+# Making FileSystem-Standalone Programs++GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT)+GHC		= ghc $(GHC_FLAGS)++RM_FLAGS	= -rf+RM		= rm $(RM_FLAGS)++PROG		= ErrorSlaveCounter MasterCounter SlaveCounter WaitSlaveCounter++OUTPUT		= output++all : $(PROG)++#prof : $(PROG).hs+#	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+#	$(RM) $(OUTPUT)/*+#	$(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+	$(RM) $(OUTPUT)/*+	$(GHC) --make -o $@ $<++clean :+	$(RM) $(PROG) $(OUTPUT) *.port+	

@@ -0,0 +1,30 @@++module Main (main) where++import           Holumbus.Common.Logging+import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DMVar+++main :: IO ()+main+  = do+    initializeLogging []+    initDNode $ (defaultDNodeConfig "Master")+      { dnc_MinPort = (fromInteger 7999), dnc_MaxPort = (fromInteger 7999) }+    c <- newDMVar "counter" 0+    doWatching c+    closeDMVar c+    deinitDNode+++doWatching :: DMVar Int -> IO ()+doWatching c+  = do+    l <- getLine+    case  l of+      "exit" -> return ()+      _ -> do+        i <- readDMVar c+        putStrLn $ "current Value: " ++ (show i)+        doWatching c

@@ -0,0 +1,33 @@++module Main(main) where+                 +import           Holumbus.Common.Logging+import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DMVar+++main :: IO ()+main+  = do+    initializeLogging []+    initDNode $ defaultDNodeConfig ""+    addForeignDNode $ mkDNodeAddress "Master" "april" (fromInteger 7999)+    c <- newRemoteDMVar "counter" "Master"+    -- addForeignDNodeHandler (mkDNodeId "Master") listener+    doCounting c+    closeDMVar c+    deinitDNode+++doCounting :: DMVar Int -> IO ()+doCounting c+  = do+    l <- getLine+    case  l of+      "exit" -> return ()+      _ -> do+        i <- takeDMVar c+        let i' = i+1+        putStrLn $ "old Value: " ++ (show i) ++ " -> new Value: " ++ (show i')+        putDMVar c i'+        doCounting c  

@@ -0,0 +1,36 @@++module Main(main) where+                 +import           Holumbus.Common.Logging+import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DMVar+++main :: IO ()+main+  = do+    initializeLogging []+    initDNode $ defaultDNodeConfig ""+    addForeignDNode $ mkDNodeAddress "Master" "april" (fromInteger 7999)+    c <- newRemoteDMVar "counter" "Master"+    -- addForeignDNodeHandler (mkDNodeId "Master") listener+    doCounting c+    closeDMVar c+    deinitDNode+++doCounting :: DMVar Int -> IO ()+doCounting c+  = do+    l <- getLine+    case  l of+      "exit" -> return ()+      _ -> do+        i <- takeDMVar c+        let i' = i+1+        putStrLn $ "old Value: " ++ (show i) ++ " -> new Value: " ++ (show i')+        putStrLn "waiting in critical section until you are pressing 'enter'..."+        _ <- getLine+        putStrLn "leaving critical section"+        putDMVar c i'+        doCounting c  

@@ -0,0 +1,27 @@+# Making FileSystem-Standalone Programs++GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT)+GHC		= ghc $(GHC_FLAGS)++RM_FLAGS	= -rf+RM		= rm $(RM_FLAGS)++PROG		= Sender Receiver++OUTPUT		= output++all : $(PROG)++#prof : $(PROG).hs+#	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+#	$(RM) $(OUTPUT)/*+#	$(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+	$(RM) $(OUTPUT)/*+	$(GHC) --make -o $@ $<++clean :+	$(RM) $(PROG) $(OUTPUT) *.port+	

@@ -0,0 +1,40 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +-}++-- ----------------------------------------------------------------------------++module Main(main) where++import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DStreamPort+import           Holumbus.Common.Logging++main :: IO ()+main+  = do+    -- we want some Holumbus specific logging+    initializeLogging []+    -- make a Haskell DNode, named "myReceiver" on the Port 7999+    -- this only needs to be called once during the runtime of the program+    _ <- initDNode $ (defaultDNodeConfig "myReceiver")+      { dnc_MinPort = (fromInteger 7999), dnc_MaxPort = (fromInteger 7999) }+    -- make a new DStream, named "myStream"+    -- this stream can be used until you close it with "closeDStream"+    stream <- newDStream "myStream"+    -- wait for the next message, read it print it out to stdout+    msg <- (receive stream)::(IO String)+    putStrLn msg+    -- we are behaving nicely and clean everything up before we leave+    closeDStream stream+    deinitDNode

@@ -0,0 +1,41 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +-}++-- ----------------------------------------------------------------------------++module Main(main) where++import           Holumbus.Distribution.DNode+import           Holumbus.Distribution.DStreamPort+import           Holumbus.Common.Logging++main :: IO ()+main+  = do+    -- we want some Holumbus specific logging+    initializeLogging []+    -- make a Haskell DNode, we don't care about its name, so we leave it+    -- blank. The system will generate a unique random name on its own.+    -- this only needs to be called once during the runtime of the program+    _ <- initDNode $ defaultDNodeConfig ""+    -- we need to know how to address the receiver node, so we have to provide+    -- its address, this only needs to be done once, or when the address of+    -- the receiver changes (which will not happen in most applications)+    addForeignDNode $ mkDNodeAddress "myReceiver" "localhost" (fromInteger 7999)+    -- we make a new port, connected to "myStream" at the node "myReceiver"+    port <- newDPort "myStream" "myReceiver"+    -- send the messages+    send port "Hello World"+    -- we are behaving nicely and clean everything up before we leave+    deinitDNode

@@ -0,0 +1,12 @@+# Making all examples++ALLEXAMPLES       = ChatDChan ChatDFunction CounterDFunction CounterDMVar HelloWorld++all :+	$(foreach i,$(ALLEXAMPLES),$(MAKE) -C $i all ;)++wc :+	@wc -l `find . -wholename './_darcs/*' -prune -o -name "*.hs" -print`++clean :+	$(foreach i,$(ALLEXAMPLES),$(MAKE) -C $i $@ ;)

@@ -1,6 +1,6 @@ # Making all examples -ALLEXAMPLES       = ClientServer Ports+ALLEXAMPLES       = ClientServer Distribution Ports   all :

@@ -1,9 +1,5 @@ # Making FileSystem-Standalone Programs -HOME		= ../..--SOURCE		= .- GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT) GHC		= ghc $(GHC_FLAGS) 

@@ -16,6 +16,7 @@ module Main(main) where  import           Data.Binary+--import           Holumbus.Common.MRBinary import qualified Data.ByteString.Lazy as B  import           Holumbus.Common.Logging

@@ -1,17 +1,18 @@ name:          Holumbus-Distribution-version:       0.0.1.1+version:       0.1.0 license:       OtherLicense license-file:  LICENSE author:        Stefan Schmidt, Uwe Schmidt-copyright:     Copyright (c) 2008 Stefan Schmidt, Uwe Schmidt+copyright:     Copyright (c) 2010 Stefan Schmidt, Uwe Schmidt, Sebastian Reese maintainer:    Stefan Schmidt <sts@holumbus.org> stability:     experimental category:      Distributed Computing, Network synopsis:      intra- and inter-program communication homepage:      http://holumbus.fh-wedel.de-description:   Holumbus-Distribution offers Erlang-Style mailboxes for an easy implementation-               of distributed systems in Haskell. The mailboxes can be used to exchange messages-               between threads inside a single program or between programs over the network.+description:   Holumbus-Distribution offers distributed data structures like Chan, MVar or functions.+               These datatype can be used between different programs on different computers to exchange+               data. With the help of this library it is possible to build Erlang-Style mailboxes for an+               easy implementation of distributed systems in Haskell.  build-type:    Simple cabal-version: >=1.6 @@ -21,6 +22,8 @@     Programs/Makefile     Programs/PortRegistry/Makefile     Programs/PortRegistry/PortRegistry.hs+    Programs/PortRegistryDaemon/Makefile+    Programs/PortRegistryDaemon/PortRegistryDaemon.hs         Examples/Makefile     Examples/ClientServer/Client.hs     Examples/ClientServer/Makefile@@ -29,19 +32,44 @@     Examples/Ports/Receiver.hs     Examples/Ports/Sender.hs     Examples/Ports/Single.hs+    Examples/Distribution/Makefile+    Examples/Distribution/ChatDChan/ClientDChan.hs+    Examples/Distribution/ChatDChan/Makefile+    Examples/Distribution/ChatDChan/ServerDChan.hs+    Examples/Distribution/ChatDChan/MessagesDChan.hs+    Examples/Distribution/ChatDFunction/ClientDFunction.hs+    Examples/Distribution/ChatDFunction/InterfacesDFunction.hs+    Examples/Distribution/ChatDFunction/Makefile+    Examples/Distribution/ChatDFunction/ServerDFunction.hs+    Examples/Distribution/CounterDFunction/Makefile+    Examples/Distribution/CounterDFunction/MasterCounter.hs+    Examples/Distribution/CounterDFunction/SlaveCounter.hs+    Examples/Distribution/CounterDMVar/Makefile+    Examples/Distribution/CounterDMVar/MasterCounter.hs+    Examples/Distribution/CounterDMVar/SlaveCounter.hs+    Examples/Distribution/CounterDMVar/WaitSlaveCounter.hs+    Examples/Distribution/CounterDMVar/ErrorSlaveCounter.hs+    Examples/Distribution/HelloWorld/Makefile+    Examples/Distribution/HelloWorld/Receiver.hs+    Examples/Distribution/HelloWorld/Sender.hs + library-  build-depends:  base       >= 4+  build-depends:  base       >= 4 && < 5                 , binary     >= 0.4                 , bytestring >= 0.9                 , containers >= 0.1-                , editline   >= 0.2                 , haskell98  >= 1                 , hslogger   >= 1.0-                , hxt        >= 8.2+                , hxt        >= 8.2                  , network    >= 2.1                 , time       >= 1.1                 , unix       >= 2.3+                , random     >= 1.0+                , stm        >= 2.1.1.2+                , parallel   >= 1.1.0.0+                , array      >= 0.2.0.0+                , readline   >= 1.0    hs-source-dirs:                    source@@ -53,22 +81,38 @@                  , Holumbus.Network.Messages                  , Holumbus.Network.Port                  , Holumbus.Network.Site+                 , Holumbus.Network.DoWithServer                                   , Holumbus.Network.PortRegistry                  , Holumbus.Network.PortRegistry.Messages                  , Holumbus.Network.PortRegistry.PortRegistryData                  , Holumbus.Network.PortRegistry.PortRegistryPort+                 , Holumbus.Distribution.DNode+                 , Holumbus.Distribution.DNode.Network+                 , Holumbus.Distribution.DNode.Base+                 , Holumbus.Distribution.DChan+                 , Holumbus.Distribution.DStreamPort+                 , Holumbus.Distribution.DFunction+                 , Holumbus.Distribution.DMVar+                 , Holumbus.Distribution.DValue                  , Holumbus.Console.Console+                 , Holumbus.Console.ServerConsole                                   , Holumbus.Common.Debug                  , Holumbus.Common.FileHandling                  , Holumbus.Common.Logging                  , Holumbus.Common.Threading                  , Holumbus.Common.Utils -  ghc-options: -Wall -threaded-  extensions: Arrows DeriveDataTypeable ExistentialQuantification+  ghc-options: -Wall+  extensions: Arrows DeriveDataTypeable ExistentialQuantification FlexibleContexts  executable PortRegistry   main-is: PortRegistry.hs   hs-source-dirs: Programs/PortRegistry source   ghc-options: -Wall -threaded-  extensions: Arrows DeriveDataTypeable ExistentialQuantification+  extensions: Arrows DeriveDataTypeable ExistentialQuantification FlexibleContexts++executable PortRegistryDaemon+  main-is: PortRegistryDaemon.hs+  hs-source-dirs: Programs/PortRegistryDaemon source+  ghc-options: -Wall -threaded+  extensions: Arrows DeriveDataTypeable ExistentialQuantification FlexibleContexts

@@ -1,8 +1,4 @@ -HOME		= ../..--SOURCE		= .- GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT) GHC		= ghc $(GHC_FLAGS) 

@@ -0,0 +1,26 @@++GHC_FLAGS	= -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT)+GHC		= ghc $(GHC_FLAGS)++RM_FLAGS	= -rf+RM		= rm $(RM_FLAGS)++PROG		= PortRegistryDaemon++OUTPUT		= output++all : $(PROG)++#prof : $(PROG).hs+#	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+#	$(RM) $(OUTPUT)/*+#	$(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+	[ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+	$(RM) $(OUTPUT)/*+	$(GHC) --make -o $@ $<++clean :+	$(RM) $(PROG) $(OUTPUT) registry.xml registry.port+	

@@ -0,0 +1,123 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.PortRegistryDaemon+  Copyright  : Copyright (C) 2009 Sebastian Reese+  License    : MIT++  Maintainer : Sebastian Reese (str@holumbus.org)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  This module provides the PortRegistryDeamonm, which is the Portregistry and a server console.++  Usage: PortRegistryDaemon PortRegistryPort ConsolePort Logfile++-}++-- ----------------------------------------------------------------------------++module Main(main) where++import           Holumbus.Common.Logging+import           Holumbus.Common.FileHandling+import           Holumbus.Common.Utils+import           Holumbus.Network.Port+import           Holumbus.Network.PortRegistry+import           Holumbus.Network.PortRegistry.PortRegistryData+import qualified Holumbus.Console.ServerConsole as Console+import           System.Environment (getArgs)+import           System.Log.Logger+import           System.Exit++version :: String+version = "Holumbus-PortRegistryDaemon 0.1"++prompt :: String+prompt = "# PortRegistryDaemon > "++localLogger :: String+localLogger = "Holumbus.PortRegistryDaemon"++pUsage :: IO ()+pUsage = do+  putStrLn "Usage: PortRegistryDaemon PortRegistryPort ConsolePort Logfile"++params :: IO [String]+params = do+  args <- getArgs+  if length args /= 3 then do+    errorM localLogger "Wrong argument count"+    pUsage+    exitFailure+    else+      return args++main :: IO ()+main+  = handleAll (\e -> errorM localLogger $ "EXCEPTION: " ++ show e) $ do+    (s_regport:s_cport:logfile:[]) <- params+    initializeFileLogging logfile [(localLogger, INFO),("Holumbus.Network.DoWithServer",INFO),("measure",ERROR)]+    let sn      = "portregistry"+    let regport = Just . fromInteger . read $ s_regport+    let cport   =                      read $ s_cport+    r <- newPortRegistryData sn regport+    p <- getPortRegistryRequestPort r+    writePortToFile p "/tmp/registry.port"+    saveToXmlFile "/tmp/registry.xml" p+    Console.startServerConsole createConsole r cport prompt++    +createConsole :: Console.ConsoleData PortRegistryData+createConsole =+  Console.addConsoleCommand "register"   hdlRegister   "registers a port manually" $+  Console.addConsoleCommand "unregister" hdlUnregister "unregisters a port manually" $+  Console.addConsoleCommand "lookup"   hdlLookup     "gets the socket id for a port" $+  Console.addConsoleCommand "ports"    hdlGetPorts   "lists all ports" $ +  Console.addConsoleCommand "version"    hdlVersion    "prints the version" $+  Console.initializeConsole+      ++hdlRegister :: PortRegistryData -> [String] -> IO String+hdlRegister prd opts+  = do+    handleAll (\_ -> return "usage: register <streamname> <hostname> <socketId>") $+      do+      let sn = head opts+      let hn = head $ tail opts+      let po = fromInteger . read . head . tail . tail $ opts+      registerPort sn (SocketId hn po) prd+      return "Register succesfull"+++hdlUnregister :: PortRegistryData -> [String] -> IO String+hdlUnregister prd opts+  = handleAll (\_ -> return "usage: unregister <streamname>") $+      do+      let sn = head opts+      unregisterPort sn prd+      return "Unregister succesfull"+++hdlLookup :: PortRegistryData -> [String] -> IO String+hdlLookup prd opts+  = handleAll (\_ -> return "usage: lookup <streamname>") $+      do+      let sn = head opts+      soid <- lookupPort sn prd+      return . show $ soid+++hdlGetPorts :: PortRegistryData -> [String] -> IO String+hdlGetPorts prd _+  = do+    ls <- getPorts prd+    return . show $ ls+++hdlVersion :: PortRegistryData -> [String] -> IO String+hdlVersion _ _+  = return version++-- ------------------------------------------------------------

@@ -1,6 +1,6 @@ This is the Holumbus-Distribution Library -Version 0.0.1+Version 0.1.0  Stefan Schmidt sts@holumbus.org @@ -12,57 +12,67 @@  Holumbus is a set of Haskell libraries. This package contains the  Holumbus-Distribution library for building and running a distributed systems.-One of the core elements of this library is an Erlang-like mailbox -system for interchanging messages between different threads or applications. +One of the core elements of this library is an Erlang-style communication +system for interchanging data between different threads or applications.  -This library itself is independent from other Holumbus libraries.+This library was completely rewritten between version 0.0.1 and 0.1.0. The old+packages can be found under Holumbus.Network.* while the new ones are+located at Holumbus.Distribution.*. The old packages will be removed in+future versions. +This library itself is independent from other Holumbus libraries although it+contains some modules which are only to be used by the other Holumbus+libraries but this may change in the near future. + Documentation: ------------------------------------------ -The "Chan" data type form Control.Concurrent.Chan is pretty useful.-You can easily use it as a queue for the consumer-producer-problem. One-thread writes into the channel, the other reads the data from it. But this-only works, if the two threads are running in the same address space.-That means the Chan data type cannot be used for the communication of-distinct applications, but this scenario is a normal use case.+Haskell offers great libraries and datatypes for the implementation+of intra-process communication, e.g. MVars, Chans, STM. But compared+to other functional languages like Erlang or Mozart/Oz (ok, the later+is a multi paradigm language) offer better support for building distributed+systems. Holumbus-Distribution wants to close this gap by offering multiple+distributed data structures. -Therefore it would be very useful to have a single framework for inter- -and intra-process communication. One thread can create a mailbox and other -threads can send messages to it, though it should make no difference, -if the threads are in the same address space or not.+The following datastructures are implemented so far: -This libary offers a "Stream-Port" implementation, borrowed from the-multi-paradigm programming language Mozart/Oz.+* distributed Chan (DChan)+  like the Chan datatype, but it allows the writing+  and reading (!) from other programs -A stream in this context is like a mailbox or a receiver. Other threads can send-messages to it via a port. A port in this context has nothing to do with the-port number of a unix-socket. Think of it as a sender.+* distributed MVar (DMVar)+  like the MVar datatype, but the content of the+  MVar can be shared among multiple programs. -To address a stream over the network easily, you can give him a unique name.-Then you can create a port, give him that stream name and the messages will-be send directly to the stream. This requires you to start the PortRegistry.+* distributed Functions (DFunction)+  an easy way to do remote procedure calls, just like+  haxr   -The PortRegistry keeps a log for all global communication streams in the-network. It MUST be started BEFORE all other programs. In future versions this-explicit execution sequence might not be necessary any more, but for now the-FIRST thing you have to start is the PortRegistry. There is ONLY ONE one-instance allowed in the whole communication network (this might also change).+* distributed Values (DValue)+  a variable which could only be written once and+  which could easily read by other programs -You can find the main parts for the communication in the module-"Holumbus.Network.Port"+* distributed Streams and Ports (DStream, DPort) +  just like the DChan, but this time, you are only+  allowed to read from the channel from one program -This library also contains some parts which might be useful, e.g. some-specialized maps (Holumbus.Data) and a commandline user interface-(Holumbus.Console). They are used in other Holumbus libraries. +To be able to use these data structures, your program needs to become a+distributed node (DNode, comparable to an Erlang-Node). After initializing+the node, you can create instances of the data structures described above.+You only need to register these resources at your node and then other nodes+are able to access them.  ++ Contents --------  Examples  Some example applications Programs  The applications you need to run a distributed system.+          (these are not needed any more an belongs to the old+           Holumbus.Network.* packages, they will be deleted in 0.2.0) source    Source code of the Holumbus-Distribution library.  @@ -94,63 +104,63 @@ $ runhaskell Setup.hs build $ runhaskell Setup.hs install --global # with root privileges -This will generate the library and the PortRegestry program.+This will generate the library and the PortRegestry programs.  For those who prefer to build it the old way with make:  $ make build $ make install # with root privileges -If you want to run your own distributed system, you'll need to compile the-PortRegistry, too. This can be done with -$ make programs-- Steps to make a distributed system running -------------------------------- -Before you can use the mailboxes in your own programs, you need to start the-PortRegistry BEFORE running any of you programs. If you have compiled the-PortRegistry, you can start it with: +How to use Streams and Ports: -$ cd Programs/PortRegistry-$ ./PortRegistry+On the receive side:+    -- Step 1:+    -- make a Haskell DNode, named "myReceiver" on the Port 7999+    -- this only needs to be called once during the runtime of the program+    _ <- initDNode $ (defaultDNodeConfig "myReceiver")+      { dnc_MinPort = (fromInteger 7999), dnc_MaxPort = (fromInteger 7999) }+    +    -- Step 2:+    -- make a new DStream, named "myStream"+    -- this stream can be used until you close it with "closeDStream"+    stream <- newDStream "myStream" -This will create a file "registry.xml" in your "/tmp" directory. This file-contains all information for your programs to access the PortRegistry.-It is wise to copy this file to every computer, on which you want to run a -program of your system.+    -- Step 3:+    -- wait for the next message, read it print it out to stdout+    msg <- (receive stream)::(IO String)+    putStrLn msg -Before you can send/receive messages in your programs, you'll need to-load the "registry.xml" file. This can be done by two simple lines in the-IO-monad:+    -- Step 4:+    -- we are behaving nicely and clean everything up before we leave+    closeDStream stream+    deinitDNode -p <- newPortRegistryFromXmlFile "/tmp/registry.xml"-setPortRegistry p+On the sender side: -To create a simple Mailbox for receiving messages, you can look a the following-demo-program:-      -main :: IO ()-main-  = do-    reg <- newPortRegistryFromXmlFile "/tmp/registry.xml"-    setPortRegistry reg-    gS <- (newGlobalStream "global"):: IO (Stream String)-    msg <- readStream gS-    putStrLn msg+    -- Step 1:+    -- make a Haskell DNode, we don't care about its name, so we leave it+    -- blank. The system will generate a unique random name on its own.+    -- this only needs to be called once during the runtime of the program+    _ <- initDNode $ defaultDNodeConfig "" -This program will set up a mailbox and wait till the first message is received-and print it out. Then is will terminate.-And the following program will send a message to your mailbox:      -      -main :: IO ()-main-  = do-    reg <- newPortRegistryFromXmlFile "/tmp/registry.xml"-    setPortRegistry reg-    gP <- (newGlobalPort "global")::IO (Port String)-    send gP "Hello World"-      -This program just sends one message to your mailbox.      +    -- Step 2:+    -- we need to know how to address the receiver node, so we have to provide+    -- its address, this only needs to be done once, or when the address of+    -- the receiver changes (which will not happen in most applications)+    addForeignDNode $ mkDNodeAddress "myReceiver" "localhost" (fromInteger 7999)++    -- Step 3:+    -- we make a new port, connected to "myStream" at the node "myReceiver"+    port <- newDPort "myStream" "myReceiver"++    -- Step 4:+    -- send the messages+    send port "Hello World"++    -- Step 5:+    -- we are behaving nicely and clean everything up before we leave+    deinitDNode

@@ -22,3 +22,4 @@    -- | Just print out some debug output.     printDebug :: m -> IO ()+  getDebug   :: m -> IO String

@@ -1,7 +1,7 @@ -- ---------------------------------------------------------------------------- {- |   Module     : Holumbus.Common.FileHandling-  Copyright  : Copyright (C) 2008 Stefan Schmidt+  Copyright  : Copyright (C) 2010 Stefan Schmidt   License    : MIT    Maintainer : Stefan Schmidt (stefanschmidt@web.de)@@ -39,7 +39,6 @@  -} -- ----------------------------------------------------------------------------- module Holumbus.Common.FileHandling     (       -- * xml files@@ -51,6 +50,7 @@     , appendToListFile     , readFromListFile     , parseByteStringToList+    , listToByteString        -- * bytestring file handling     , writeToBinFile@@ -70,10 +70,11 @@ import qualified Data.ByteString.Lazy as B import           Data.Char import           Foreign-import           System.IO+import           System.IO hiding (utf8) import           System.IO.Unsafe  import           Text.XML.HXT.Arrow+import           Control.Parallel.Strategies  -- ---------------------------------------------------------------------------- -- xml files@@ -94,7 +95,7 @@ saveToXmlFile :: (XmlPickler a) => FilePath -> a -> IO () saveToXmlFile f i    = do-    runX (constA i >>> xpickleDocument xpickle options f)+    _ <- runX (constA i >>> xpickleDocument xpickle options f)     return ()     where     options = [ (a_indent, v_1), (a_output_encoding, utf8), (a_validate, v_0) ]@@ -106,36 +107,21 @@  -- | Writes data to a list file. writeToListFile :: (Binary a) => FilePath -> [a] -> IO ()-writeToListFile fp bs = writeToBinFile fp $ B.concat $ map encode bs-+writeToListFile = encodeFile  -- | Appends data to a list file. appendToListFile :: (Binary a) => FilePath -> [a] -> IO ()-appendToListFile fp bs = appendToBinFile fp $ B.concat $ map encode bs+appendToListFile fp = appendToBinFile fp . listToByteString  -- | reads from a list file.-readFromListFile :: (Binary a) => FilePath -> IO [a]-readFromListFile f-   = do-     b <- readFromBinFile f -     return $ parseByteStringToList b+readFromListFile :: (NFData a, Binary a) => FilePath -> IO [a]+readFromListFile = decodeFile +listToByteString :: (Binary a) => [a] -> B.ByteString+listToByteString = encode --- | You'll need this function, if you read the files a a normal binary file,---   but the content itself is a list. This function encodes the bytestring---   into a list of the specified datatype. parseByteStringToList :: (Binary a) => B.ByteString -> [a]-parseByteStringToList b = reverse $ parse b []-  where-  parse :: (Binary a) => B.ByteString -> [a] -> [a]-  parse bs accu-    | (B.null bs) = accu-    | otherwise   = parse (B.drop count bs) ([nextElem] ++ accu) -    where-    nextElem = decode bs-    count    = B.length (encode nextElem)-     -+parseByteStringToList = decode  -- ----------------------------------------------------------------------------      -- strict functions, bytestrings only     @@ -154,12 +140,12 @@  -- | Reads the data from a binary file as a bytestring. readFromBinFile :: FilePath -> IO B.ByteString-readFromBinFile f  -   = bracket (openBinaryFile f ReadMode) hClose $ +readFromBinFile = B.readFile+{-   = bracket (openBinaryFile f ReadMode) hClose $         \h -> do        s <- hFileSize h        c <- B.hGetNonBlocking h (fromInteger s)-       return $! c    +       return $! c    -}   -- ----------------------------------------------------------------------------     @@ -211,3 +197,6 @@     | otherwise = do        w <- peekElemOff p len'        loop' (len'-1) p (chr (fromIntegral w):acc)+       +       + 

@@ -21,7 +21,8 @@ module Holumbus.Common.Logging ( -- * Configuration-initializeLogging+  initializeLogging+, initializeFileLogging ) where @@ -37,16 +38,43 @@ -- ---------------------------------------------------------------------------- -- Configuration -- ----------------------------------------------------------------------------+-- | configures the logging-parameters for the Holumbus-Framework+initializeFileLogging :: FilePath -> [(String, Priority)] -> IO ()+initializeFileLogging file ls = do +    +    handle <- openFile file AppendMode+    -- log all verbose+    v <- verboseStreamHandler handle DEBUG+    updateGlobalLogger "Holumbus" (setLevel WARNING)+    +    updateGlobalLogger rootLoggerName (setLevel DEBUG . setHandlers [v]) +    -- set all logLevels for all loggers+    mapM_ (\(s,p) -> updateGlobalLogger s (setLevel p)) ls++    -- log all to syslog+    -- s <- openlog "SyslogStuff" [PID] USER DEBUG+    -- updateGlobalLogger rootLoggerName (addHandler s)+    +    -- this will be removed in next versions    +    -- updateGlobalLogger "Holumbus.Network" (setLevel WARNING)+    -- updateGlobalLogger "Holumbus.MapReduce.TaskProcessor.task" (setLevel WARNING)+    -- updateGlobalLogger "Holumbus.FileSystem.Storage.FileStorage" (setLevel WARNING)+    -- updateGlobalLogger "Holumbus.FileSystem.Node.NodeData" (setLevel INFO)+    -- updateGlobalLogger "Holumbus.MapReduce.Types" (setLevel INFO)+    -- updateGlobalLogger "Holumbus.Network.Communication" (setLevel DEBUG)+    -- updateGlobalLogger "Holumbus.MapReduce.JobController.cycle" (setLevel WARNING)+ -- | configures the logging-parameters for the Holumbus-Framework initializeLogging :: [(String, Priority)] -> IO () initializeLogging ls = do -    -- log all verbose	++    -- log all verbose     v <- verboseStreamHandler stderr DEBUG     updateGlobalLogger rootLoggerName (setLevel DEBUG . setHandlers [v])-+    updateGlobalLogger "Holumbus" (setLevel WARNING)     -- set all logLevels for all loggers     mapM_ (\(s,p) -> updateGlobalLogger s (setLevel p)) ls @@ -55,7 +83,7 @@     -- updateGlobalLogger rootLoggerName (addHandler s)          -- this will be removed in next versions-    updateGlobalLogger "Holumbus" (setLevel WARNING)+    --updateGlobalLogger "Holumbus" (setLevel WARNING)     -- updateGlobalLogger "Holumbus.Network" (setLevel WARNING)     -- updateGlobalLogger "Holumbus.MapReduce.TaskProcessor.task" (setLevel WARNING)     -- updateGlobalLogger "Holumbus.FileSystem.Storage.FileStorage" (setLevel WARNING)

@@ -40,12 +40,10 @@ 					, handle 					) import           Data.Binary+--import           Holumbus.Common.MRBinary import qualified Data.ByteString.Lazy as B  -- import qualified Data.Map as Map--import           Data.Maybe-import           Data.Char  -- import           System.Exit 

@@ -51,13 +51,10 @@ where  import           Control.Concurrent-import           Control.Monad -import           Data.Char import qualified Data.Map as Map -import           System.IO-import           System.Console.Editline.Readline+import           System.Console.Readline  import           Holumbus.Common.Utils	( handleAll ) 

@@ -0,0 +1,216 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Console.ServerConsole+  Copyright  : Copyright (C) 2009 Sebastian Reese+  License    : MIT++  Maintainer : Sebastian Reese (str@holumbus.org)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  This module provides a tiny and nice implementation of a little command +  shell with communcation over a socket.+  +  It is basically a copy of Holumbus.Console.Console with some changes to fit network communication.++-}++-- ----------------------------------------------------------------------------+module Holumbus.Console.ServerConsole+    (+     -- * Console datatype+     ConsoleData+    +     -- * Operations+    , nextOption+    , parseOption+    , initializeConsole+    , addConsoleCommand+    , startServerConsole+    , defaultaction+    , defaultconverter+    )+where++import           Holumbus.Network.DoWithServer+import           System.IO+import           Control.Monad (forM)+import qualified Data.Map as Map+import           Holumbus.Common.Utils  ( handleAll )+import           Data.List++-- | Starts the server listening+startServerConsole ::+     ConsoleData a -- ^ the console data+  -> a             -- ^ the console config+  -> Int           -- ^ console port+  -> String        -- ^ a consoles prompt+  -> IO ()+startServerConsole cdata conf port prompt = doWithServer port (defaultaction cdata conf prompt) defaultconverter prompt++-- | This defaultimplementaion can be used if a simple INput -> Process command -> output patern is used+defaultaction :: ConsoleData a -> a -> String -> ServerAction String+defaultaction cdata conf prompt line sender clients = do+  clients' <-  handleAll (\_ -> return  $ [delete sender $ clients]) $ do+    forM (filter (==sender) clients) $+      \(Client _ handle _ _) -> do +        result <- handleInput line cdata conf+        if result == exitString then do +            hClose handle+            return . delete sender $ clients+          else do   +            hPutStrLn handle result+            hPutStr handle prompt+            hFlush handle+            return clients+  return . concat $ clients'++-- | default string to a converter. Converts the input lines into desired format. Here String+defaultconverter :: LineConverter String+defaultconverter = id+++-- ----------------------------------------------------------------------------+-- datatypes+-- ----------------------------------------------------------------------------+++-- | Map which contains all commands that the user can execute+type ConsoleData a = Map.Map String (ConsoleCommand a)+++-- | Console command, only a pair of a function which will be executed +--   and a description +type ConsoleCommand a = ( Maybe (ConsoleFunction a), String )+++-- | A console function. The string list represents the arguments+type ConsoleFunction a = (a -> [String] -> IO String )++-- ----------------------------------------------------------------------------+-- operations +-- ----------------------------------------------------------------------------+++-- | Creates a new console datatype+initializeConsole :: ConsoleData a+initializeConsole = Map.fromList [(exitString, exitCommand), (helpString, helpCommand)]+++-- | Adds a new console command to the function, an existing command with the+--   same name will be overwritten+addConsoleCommand +  :: String             -- ^ command string (the word the user has to enter when he wants to execute the command)+  -> ConsoleFunction a  -- ^ the function which should be executed+  -> String             -- ^ the function description+  -> ConsoleData a      -- ^ the old console data+  -> ConsoleData a+addConsoleCommand c f d m = Map.insert c (Just f, d) m+++-- | The exit function string.+exitString  :: String+exitString = "exit"+++-- | A dummy exit function (Just to print the help description, the command +--   is handled in the main loop.+exitCommand :: ConsoleCommand a+exitCommand = ( Nothing, "exit the console")+++-- | The help function string.+helpString :: String+helpString = "help"+++-- | A dummy help function (Just to print the help description, the command +--   is handled in the main loop.+helpCommand :: ConsoleCommand a+helpCommand = (Nothing, "print this help")+++-- | gets the next option from the command line as string+nextOption :: [String] -> IO (Maybe String, [String])+nextOption o+  = handleAll (\_ -> return (Nothing, o)) $+      do+      if ( null o ) then+          return (Nothing, o)+        else +          return (Just $ head o, tail o) ++-- | Simple "parser" for the commandline...+parseOption :: Read a => [String] -> IO (Maybe a, [String])+parseOption o+  = handleAll (\_ -> return (Nothing, o)) $+      do+      if ( null o ) then+          return (Nothing, o)+        else +          return (Just $ read $ head o, tail o) ++-- | The main loop. You know... read stdin, parse the input, execute command.+--   You can quit it by the exit-command.+handleInput :: String -> ConsoleData a -> a -> IO String+handleInput line cdata conf+  = do+    input <- return (words line)+    cmd   <- return (command input)+    args  <- return (arguments input)+    if (cmd == exitString) +      then do+        return exitString+      else do+        if (not $ null cmd) then handleCommand cdata conf cmd args else return "undefined"+    where+      command s = if (not $ null s) then head s else ""+      arguments s = tail s+++-- | Picks the command an execute the command function.+handleCommand :: ConsoleData a -> a -> String -> [String] -> IO String+handleCommand cdata conf cmd args+  = do+    if (cmd == helpString)+      then do+        printHelp cdata+      else do+        handleCommand' (Map.lookup cmd cdata)+    where+        handleCommand' Nothing              = do printError+        handleCommand' (Just (Nothing, _ )) = do printNoHandler+        handleCommand' (Just (Just f, _ ))  = do f conf args+++-- | Is executed when the function has no handler+printNoHandler :: IO String+printNoHandler = return "no function handler found"+++-- | Prints the "command-not-found" message.+printError :: IO String+printError = return "unknown command, try help for a list of available commands"+++-- | Prints the help text.+printHelp :: ConsoleData a -> IO String+printHelp cdata = return $ "available Commands:\n"++printCommands "" (Map.toAscList cdata)+    where+      printCommands acc [] = acc +      printCommands acc (x:xs) = printCommands (acc ++ "\n" ++ printCommand x) xs+      printCommand (c, (_, t)) = (prettyCommandName 15 c) ++ " -  " ++ t+++-- | Does some pretty printing for the function descriptions+prettyCommandName :: Int -> String -> String +prettyCommandName n s+  | n <= 0 = s+  | (n > 0) && (null s) = ' ' : prettyCommandName (n-1) s+  | otherwise           = x : prettyCommandName (n-1) xs+    where+      (x:xs) = s++

@@ -40,7 +40,7 @@ where  import           Prelude hiding (null, lookup)-+import qualified Data.List as L import qualified Data.Map as Map  @@ -109,7 +109,7 @@  -- | Creates an AccuMap from a list. fromList :: (Ord k) => [(k,[a])] -> AccuMap k a-fromList ks = foldl (\m (k,as) -> insertList k as m) empty ks+fromList ks = L.foldl' (\m (k,as) -> insertList k as m) empty ks   -- | Creates an AccuMap from a tuple list.

@@ -46,7 +46,6 @@ where  import           Prelude hiding (null, lookup)-import           Data.Maybe import qualified Data.Map as Map  -- | Every element of this map has to implement a key-function. which

@@ -0,0 +1,272 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Distribution.DChan+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +  This module offers a distributed channel datatype (DChan).++  It is similar to Control.Concurrent.Chan, except that you can use it+  between multiple processes on different computers. You can access a+  DChan (reading and writing) from your local process as well as from+  another one.+-}++-- ----------------------------------------------------------------------------++module Holumbus.Distribution.DChan +(+  -- * datatypes+    DChan+  +  -- * creating and closing channels+  , newDChan+  , newRemoteDChan+  , closeDChan+  +  -- * operations on a channel+  , writeDChan+  , readDChan+  , tryReadDChan+  , tryWaitReadDChan+  , isEmptyDChan+)+where++import           Control.Concurrent.Chan+import           Data.Binary+import qualified Data.ByteString.Lazy as B+import           System.IO+import           System.Log.Logger+import           System.Timeout++import           Holumbus.Distribution.DNode.Base++localLogger :: String+localLogger = "Holumbus.Distribution.DChan"+++dChanType :: DResourceType+dChanType = mkDResourceType "DCHAN"++mkDChanEntry :: (Binary a) => DChanReference a -> DResourceEntry+mkDChanEntry d = DResourceEntry {+    dre_Dispatcher   = dispatchDChanRequest d +  }+++data DChanRequestMessage+  = DCMReqRead+  | DCMReqWrite B.ByteString+  | DCMReqIsEmpty+  deriving (Show)++instance Binary DChanRequestMessage where+  put(DCMReqRead)    = putWord8 1+  put(DCMReqWrite a) = putWord8 2 >> put a+  put(DCMReqIsEmpty) = putWord8 3+  get+    = do+      t <- getWord8+      case t of+        1 -> return (DCMReqRead)+        2 -> get >>= \a -> return (DCMReqWrite a)+        3 -> return (DCMReqIsEmpty)+        _ -> error "DChanRequestMessage: wrong encoding"++  +data DChanResponseMessage+  = DCMRspRead B.ByteString+  | DCMRspWrite+  | DCMRspIsEmpty Bool+  deriving (Show)+++instance Binary DChanResponseMessage where+  put(DCMRspRead a)    = putWord8 1 >> put a+  put(DCMRspWrite)     = putWord8 2+  put(DCMRspIsEmpty b) = putWord8 3 >> put b+  get+    = do+      t <- getWord8+      case t of+        1 -> get >>= \a -> return (DCMRspRead a)+        2 -> return (DCMRspWrite)+        3 -> get >>= \b -> return (DCMRspIsEmpty b)+        _ -> error "DChanResponseMessage: wrong encoding"+++dispatchDChanRequest :: (Binary a) => DChanReference a -> DNodeId -> Handle -> IO () +dispatchDChanRequest dch _ hdl+  = do+    debugM localLogger "dispatcher: getting message from handle"+    raw <- getByteStringMessage hdl+    let msg = (decode raw)+    -- debugM localLogger $ "dispatcher: Message: " ++ show msg+    case msg of+      (DCMReqRead)    -> handleRead dch hdl+      (DCMReqWrite d) -> handleWrite dch (decode d) hdl+      (DCMReqIsEmpty) -> handleIsEmpty dch hdl++    +-- | The DChan datatype.+--   Notice that this datatype implements the Data.Binary typeclass.+--   That means that you can pass a DChan, so that another computer+--   can access the channel.+data DChan a+  = DChanLocal DResourceAddress (Chan a)+  | DChanRemote DResourceAddress++instance Binary (DChan a) where+  put(DChanLocal a _) = put a+  put(DChanRemote a)  = put a+  get = get >>= \a -> return (DChanRemote a)+++data DChanReference a = DChanReference DResourceAddress (Chan a) +++-- | Creates a new DChan on the local computer. The first parameter+--   is the name of the Channel which could be used in other processes to+--   access this stream. If you leave it empty, a random Id will be created.+newDChan :: (Binary a) => String -> IO (DChan a)+newDChan s+  = do+    dra <- genLocalResourceAddress dChanType s+    c <- newChan+    let dch = (DChanLocal dra c)+        dcr = (DChanReference dra c)+        dce = (mkDChanEntry dcr)+    addLocalResource dra dce+    return dch+    +-- TODO merge this with newDChan?+-- | Creates a reference to a DChan which was created in a different+--   process.+--   The first parameter is the name of the resource and the second one+--   the name of the node.+newRemoteDChan :: String -> String -> IO (DChan a)+newRemoteDChan r n+  = do+    return $ DChanRemote dra+    where+    dra = mkDResourceAddress dChanType r n+++-- | Closes a DChan object, could not be used anymore after this call.+closeDChan :: (DChan a) -> IO ()+closeDChan (DChanLocal dra _)+  = do+    delLocalResource dra+closeDChan (DChanRemote dra)+  = do+    delForeignResource dra++++requestRead :: (Binary a) => Handle -> IO a+requestRead hdl+  = do+    putByteStringMessage (encode $ DCMReqRead) hdl+    raw <- getByteStringMessage hdl+    let rsp = (decode raw)+    case rsp of+      (DCMRspRead d)  -> return $ decode d+      _ -> error "DChan - requestRead: invalid response"+    ++handleRead :: (Binary a) => DChanReference a -> Handle -> IO ()+handleRead (DChanReference _ ch) hdl+  = do+    a <- readChan ch+    putByteStringMessage (encode $ DCMRspRead $ encode a) hdl+++requestWrite :: (Binary a) => a -> Handle -> IO ()+requestWrite a hdl+  = do +    putByteStringMessage (encode $ DCMReqWrite $ encode a) hdl+    raw <- getByteStringMessage hdl+    let rsp = (decode raw)+    case rsp of+      (DCMRspWrite) -> return ()+      _ -> error "DChan - requestWrite: invalid response"+    ++handleWrite :: DChanReference a -> a -> Handle -> IO ()+handleWrite (DChanReference _ ch) a hdl+  = do+    writeChan ch a+    putByteStringMessage (encode $ DCMRspWrite) hdl+++requestIsEmpty :: Handle -> IO Bool+requestIsEmpty hdl+  = do+    putByteStringMessage (encode $ DCMReqIsEmpty) hdl+    raw <- getByteStringMessage hdl+    let rsp = (decode raw)+    case rsp of+      (DCMRspIsEmpty b) -> return b+      _ -> error "DChan - requestIsEmpty: invalid response"+    ++handleIsEmpty :: DChanReference a -> Handle -> IO ()+handleIsEmpty (DChanReference _ ch) hdl+  = do+    b <- isEmptyChan ch+    putByteStringMessage (encode $ DCMRspIsEmpty b) hdl +++-- | Writes data to a DChan.+writeDChan :: (Binary a) => DChan a -> a -> IO ()+writeDChan (DChanLocal _ c) v+  = do+    writeChan c v+writeDChan (DChanRemote a) v+  = do+    unsafeAccessForeignResource a (requestWrite v)+++-- | Reads data from a DChan, blocks if DChan is empty.+readDChan :: (Binary a) => DChan a -> IO a+readDChan (DChanLocal _ c)+  = do+    readChan c+readDChan (DChanRemote a)+  = do+    unsafeAccessForeignResource a requestRead+++-- | Tries to read data from a DChan, if the DChan is empty,+--   the function return with Nothing.+tryReadDChan :: (Binary a) => DChan a -> IO (Maybe a)+tryReadDChan dc+  = do+    empty <- isEmptyDChan dc+    if (not empty) +      then do timeout 1000 (readDChan dc)+      else do return Nothing+++-- | Reads data from a DChan. If the channel is empty, it waits+--   for a given time (in microseconds) an returns immediately+--   when new data arrives, otherwise it return Nothing.+tryWaitReadDChan :: (Binary a) => DChan a -> Int -> IO (Maybe a) +tryWaitReadDChan dc t = timeout t (readDChan dc)+    ++-- | Tests, if a DChan is empty.+isEmptyDChan :: DChan a -> IO Bool+isEmptyDChan (DChanLocal _ c)+  = do+    isEmptyChan c+isEmptyDChan (DChanRemote a)+  = do+    unsafeAccessForeignResource a requestIsEmpty

@@ -0,0 +1,224 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Distribution.DFunction+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +  This module offers distributed functions.++  This idea behind this is to implement RPC based on DNodes. You specify+  a function which could be called from other programs and register this+  as a resource in your local DNode. Then the foreign DNodes can create+  a link to this function an execute it. The function parameters will be+  serialized and send to the local DNode. There the parameters are deserialized+  and the function will be called. After this the return-value will be send+  back to the calling node. +-}++-- ----------------------------------------------------------------------------++{-# OPTIONS_GHC -XMultiParamTypeClasses -XFunctionalDependencies -XFlexibleInstances -XFlexibleContexts #-}+module Holumbus.Distribution.DFunction +(+  -- * datatypes+    DFunction+  , BinaryFunction+   +  -- * creating and closing function references+  , newDFunction+  , newRemoteDFunction+  , closeDFunction+  +  -- * invoking functions+  , accessDFunction+)+where++import           Prelude hiding (catch)++import           Control.Exception+import           Data.Binary+import qualified Data.ByteString.Lazy as B+import           System.IO+import           System.Log.Logger++import           Holumbus.Distribution.DNode.Base+++localLogger :: String+localLogger = "Holumbus.Distribution.DFunction"++-- ----------------------------------------------------------------------------++-- | Binary function typeclass. You can only use functions whose parameters+--   and return value are serializable. The idea of this typeclass comes from+--   the haxr library by Bjorn Bringert (http://www.haskell.org/haskellwiki/HaXR)+class BinaryFunction a where+    toFun :: a -> [B.ByteString] -> IO B.ByteString+    remoteCall :: ([B.ByteString] -> IO B.ByteString) -> a++instance (Binary a) => BinaryFunction (IO a) where+    toFun x [] = x >>= return . encode+    toFun _ _ = fail "Too many arguments"+    remoteCall f = f [] >>= return . decode++instance (Binary a, BinaryFunction b) => BinaryFunction (a -> b) where+    toFun f (x:xs) = toFun (f (decode x)) xs+    toFun _ _ = error "Too few arguments"+    remoteCall f x = remoteCall (\xs -> f (encode x:xs))+++++-- ----------------------------------------------------------------------------++dFunctionType :: DResourceType+dFunctionType = mkDResourceType "DFUNCTION"++mkDFunctionEntry :: (BinaryFunction a) => DFunctionReference a -> DResourceEntry+mkDFunctionEntry d = DResourceEntry {+    dre_Dispatcher   = dispatchDFunctionRequest d +  }+++data DFunctionRequestMessage+  = DFMReqCall [B.ByteString]+  deriving (Show)+  +instance Binary DFunctionRequestMessage where+  put(DFMReqCall bs) = put bs+  get = get >>= \bs -> return (DFMReqCall bs)+++data DFunctionResponseMessage+  = DFMRspCallResult B.ByteString+  | DFMRspCallException String++instance Binary DFunctionResponseMessage where+  put(DFMRspCallResult b) = putWord8 1 >> put b+  put(DFMRspCallException e) = putWord8 2 >> put e+  get+    = do+      t <- getWord8+      case t of+        1 -> get >>= \b -> return (DFMRspCallResult b)+        2 -> get >>= \e -> return (DFMRspCallException e)+        _ -> error "DFunctionResponseMessage: wrong encoding"+++dispatchDFunctionRequest :: (BinaryFunction a) => DFunctionReference a -> DNodeId -> Handle -> IO () +dispatchDFunctionRequest dfun _ hdl+  = do+    debugM localLogger "dispatcher: getting message from handle"+    raw <- getByteStringMessage hdl+    let msg = (decode raw)::(DFunctionRequestMessage)+    -- debugM localLogger $ "dispatcher: Message: " ++ show msg+    case msg of+      (DFMReqCall l)  -> handleCall dfun l hdl+++-- | The DFunction datatype. This is more like a reference to+--   a function located on a different node. You can call this+--   function via the accessDFunction function.+data DFunction a+  = DFunctionLocal DResourceAddress a+  | DFunctionRemote DResourceAddress++instance Binary (DFunction a) where+  put(DFunctionLocal dra _) = put dra+  put(DFunctionRemote dra)  = put dra+  get = get >>= \dra -> return (DFunctionRemote dra)+  +  +data DFunctionReference a = DFunctionReference DResourceAddress a+++-- | Creates a new distributed function. Only functions which are registered+--   at the local node can be called from the outside. The string parameter+--   specifies the name of the function which could the used by other nodes+--   to call it. If you leave it empty, a random name will be generated.+newDFunction :: (BinaryFunction a) => String -> a -> IO (DFunction a)+newDFunction s f+  = do+    a <- genLocalResourceAddress dFunctionType s+    let df  = (DFunctionLocal a f)+        dfr = (DFunctionReference a f)+        dfd = (mkDFunctionEntry dfr)+    addLocalResource a dfd+    return df+++-- | Created a reference to a function on a remote node. The first parameter+--   is the name of the function, the second parameter is the name of the node.+newRemoteDFunction :: (BinaryFunction a) => String -> String -> IO (DFunction a)+newRemoteDFunction r n+  = do+    return $ DFunctionRemote dra+    where+    dra = mkDResourceAddress dFunctionType r n+++-- | Closes a DFunction reference.+closeDFunction :: DFunction a -> IO ()+closeDFunction (DFunctionLocal dra _)+  = do+    delLocalResource dra+closeDFunction (DFunctionRemote dra)+  = do+    delForeignResource dra+++requestCall :: [B.ByteString] -> Handle -> IO B.ByteString+requestCall bs hdl+  = do+    putByteStringMessage (encode $ DFMReqCall bs) hdl+    raw <- getByteStringMessage hdl+    let rsp = (decode raw)+    case rsp of+      (DFMRspCallResult b) -> return b+      (DFMRspCallException e) -> throwIO $ DistributedException e "requestCall" "DFunction"+++handleCall :: BinaryFunction a => DFunctionReference a -> [B.ByteString] -> Handle -> IO ()+handleCall (DFunctionReference _ f) bs hdl+  = do+    catch+      (do+       b <- toFun f bs+       putByteStringMessage (encode $ DFMRspCallResult b) hdl)+      (\(SomeException e) -> do+       putByteStringMessage (encode $ DFMRspCallException (show e)) hdl)+++-- | Transforms a DFunction object to a normal function which could be called and passed around.+--   Because you have network tranfer everytime you call the function, this might throw a+--   DistributedException when the foreign node becomes unreachable.+accessDFunction :: (BinaryFunction a) => DFunction a -> a+accessDFunction (DFunctionLocal _ f) = f+accessDFunction (DFunctionRemote a)+  = remoteCall $ \bs ->+      unsafeAccessForeignResource a (requestCall bs)+++{-+Example Code:++addInt :: Int -> Int -> IO Int+addInt i1 i2 = return $ i1 + i2 ++test :: IO ()+test+  = do+    -- dfun <- newDFunction "add" (addInt)+    dfun <- (newRemoteDFunction address)::(IO (DFunction (Int -> Int -> IO Int)))+    let f = accessDFunction dfun+    res <- f 1 2+    putStrLn $ show res+    return ()+-}

@@ -0,0 +1,307 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Distribution.DMVar +  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +  This module offers the distributed MVar datatype.++  The datatype behaves just like a normal MVar, but the content of the+  variable may be stored on a different DNode. When accessing the DMVar,+  the content will be fetched from the external node and written back.++  It is guaranteed, that only one node at a time can take the content+  of the DMVar. Just like normal DMVars, you can produce deadlocks.++  When a node dies which holds the content of a DMVar, the node which+  created the variable will reset its value to the last known value.+  +  If the owner dies, the other nodes cannot access the content of the+  DMVar any more.+-}++-- ----------------------------------------------------------------------------++module Holumbus.Distribution.DMVar+(+  -- * datatype+    DMVar++  -- * creating and closing DMVars+  , newDMVar+  , newEmptyDMVar+  , newRemoteDMVar+  , closeDMVar++  -- * acccessing DMVars+  , readDMVar+  , takeDMVar+  , putDMVar+)+where++import           Prelude hiding (catch)++import           Control.Concurrent.MVar+import           Data.Binary+import qualified Data.ByteString.Lazy as B+import           System.IO+import           System.Log.Logger++import           Holumbus.Distribution.DNode.Base+++localLogger :: String+localLogger = "Holumbus.Distribution.DMVar"++dMVarType :: DResourceType+dMVarType = mkDResourceType "DMVAR"++mkDMVarEntry :: (Binary a) => DMVarReference a -> DResourceEntry+mkDMVarEntry d = DResourceEntry {+    dre_Dispatcher   = dispatchDMVarRequest d +  }+++data DMVarRequestMessage+  = DVMReqRead+  | DVMReqTake+  | DVMReqPut B.ByteString+  deriving (Show)+    +instance Binary DMVarRequestMessage where+  put(DVMReqRead)  = putWord8 1+  put(DVMReqTake)  = putWord8 2+  put(DVMReqPut b) = putWord8 3 >> put b+  get+    = do+      t <- getWord8+      case t of+        1 -> return (DVMReqRead)+        2 -> return (DVMReqTake)+        3 -> get >>= \b -> return (DVMReqPut b)+        _ -> error "DMVarRequestMessage: wrong encoding"+++data DMVarResponseMessage+  = DVMRspRead B.ByteString+  | DVMRspTake B.ByteString+  | DVMRspPut+  deriving (Show)++instance Binary DMVarResponseMessage where+  put(DVMRspRead b) = putWord8 1 >> put b+  put(DVMRspTake b) = putWord8 2 >> put b+  put(DVMRspPut)    = putWord8 3+  get+    = do+      t <- getWord8+      case t of+        1 -> get >>= \b -> return (DVMRspRead b)+        2 -> get >>= \b -> return (DVMRspTake b)+        3 -> return (DVMRspPut)+        _ -> error "DMVarResponseMessage: wrong encoding"+++dispatchDMVarRequest :: (Binary a) => DMVarReference a -> DNodeId -> Handle -> IO () +dispatchDMVarRequest dch dna hdl+  = do+    debugM localLogger "dispatcher: getting message from handle"+    raw <- getByteStringMessage hdl+    let msg = (decode raw)+    debugM localLogger $ "dispatcher: Message: " ++ show msg+    case msg of+      (DVMReqRead)  -> handleRead dch hdl+      (DVMReqTake)  -> handleTake dch dna hdl+      (DVMReqPut b) -> handlePut dch (decode b) hdl+++-- | The DMVar datatype.+data DMVar a+  = DMVarLocal DResourceAddress (MVar a) (MVar (a, Maybe DHandlerId))+  | DMVarRemote DResourceAddress++instance Binary (DMVar a) where+  put(DMVarLocal dra _ _) = put dra+  put(DMVarRemote dra)    = put dra+  get = get >>= \dra -> return (DMVarRemote dra)++data DMVarReference a = DMVarReference DResourceAddress (MVar a) (MVar (a, Maybe DHandlerId))+  ++-- | Creates a new local DMVar with a start value.+--   The string parameter specifies the name of the variable.+--   If you leave it empty, a random value will be generated.+newDMVar :: (Binary a) => String -> a -> IO (DMVar a)+newDMVar s d+  = do+    dra <- genLocalResourceAddress dMVarType s+    v <- newMVar d+    o <- newEmptyMVar+    let dmv = (DMVarLocal dra v o)+        dvr = (DMVarReference dra v o)+        dve = (mkDMVarEntry dvr)+    addLocalResource dra dve+    return dmv+++-- | Creates a new empty local DMVar. The string parameter specifies the name of+--   the variable. If you leave it empty, a random value will be generated.+newEmptyDMVar :: (Binary a) => String -> IO (DMVar a)+newEmptyDMVar s+  = do+    dra <- genLocalResourceAddress dMVarType s+    v <- newEmptyMVar+    o <- newMVar (undefined, Nothing)+    let dmv = (DMVarLocal dra v o)+        dvr = (DMVarReference dra v o)+        dve = (mkDMVarEntry dvr)+    addLocalResource dra dve+    return dmv+++-- | Creates a reference to an external DMVar.+--   The first parameter is the name of the resource and the second one+--   the name of the node.+newRemoteDMVar :: String -> String -> IO (DMVar a)+newRemoteDMVar r n+  = do+    return $ DMVarRemote dra+    where+    dra = mkDResourceAddress dMVarType r n+++-- | Closes a DMVar+closeDMVar :: (DMVar a) -> IO ()+closeDMVar (DMVarLocal dra _ _)+  = do+    delLocalResource dra+closeDMVar (DMVarRemote dra)+  = do+    delForeignResource dra++++requestRead :: (Binary a) => Handle -> IO a+requestRead hdl+  = do+    putByteStringMessage (encode $ DVMReqRead) hdl+    raw <- getByteStringMessage hdl+    let rsp = (decode raw)+    case rsp of+      (DVMRspRead d) -> return $ decode d+      _ -> error "DMVar - requestRead: invalid response"+++handleRead :: (Binary a) => DMVarReference a -> Handle -> IO ()+handleRead (DMVarReference _ v _) hdl+  = do+    a <- readMVar v+    putByteStringMessage (encode $ DVMRspRead $ encode a) hdl++++requestTake :: (Binary a) => Handle -> IO a+requestTake hdl+  = do+    putByteStringMessage (encode $ DVMReqTake) hdl+    raw <- getByteStringMessage hdl+    let rsp = (decode raw)+    case rsp of+      (DVMRspTake d) -> return $ decode d+      _ -> error "DMVar - requestTake: invalid response"+++handleTake :: (Binary a) => DMVarReference a -> DNodeId -> Handle -> IO ()+handleTake r@(DMVarReference _ v o) dni hdl+  = do+    debugM localLogger $ "handleTake: 1"+    a <- takeMVar v+    debugM localLogger $ "handleTake: 2"+    -- install handler and save backup+    mbDhi <- addForeignDNodeHandler False dni (handleErrorTake r)+    debugM localLogger $ "handleTake: 3"+    putMVar o (a, mbDhi)+    debugM localLogger $ "handleTake: 4"+    putByteStringMessage (encode $ DVMRspTake $ encode a) hdl+    debugM localLogger $ "handleTake: 5"+    ++handleErrorTake :: (Binary a) => DMVarReference a -> DHandlerId -> IO ()+handleErrorTake (DMVarReference _ v o) dhi+  = do+    debugM localLogger $ "handleErrorTake: 1"+    (a,_ ) <- takeMVar o+    delForeignHandler dhi+    debugM localLogger $ "handleErrorTake: 2"+    putMVar v a+++requestPut :: (Binary a) => a -> Handle -> IO ()+requestPut d hdl+  = do+    putByteStringMessage (encode $ DVMReqPut $ encode d) hdl+    raw <- getByteStringMessage hdl+    let rsp = (decode raw)+    case rsp of+      (DVMRspPut) -> return ()+      _ -> error "DMVar - requestWrite: invalid response"+++handlePut :: (Binary a) => DMVarReference a -> a -> Handle -> IO ()+handlePut (DMVarReference _ v o) a hdl+  = do+    -- delete backup and kill handler +    (_,mbDhi) <- takeMVar o+    case mbDhi of+      (Just dhi) -> delForeignHandler dhi+      (Nothing)  -> return ()+    putMVar v a+    putByteStringMessage (encode $ DVMRspPut) hdl++++-- | Reads the content of a DMVar. Blocks if the Variable is empty.+--   This may throw an exception if the owner of the variable is unreachable.+readDMVar :: (Binary a) => DMVar a -> IO a+readDMVar (DMVarLocal _ v _)+  = do+    readMVar v+readDMVar (DMVarRemote a)+  = do+    unsafeAccessForeignResource a requestRead+++-- | Takes the content of a DMVar. Blocks if the Variable is empty.+--   This may throw an exception if the owner of the variable is unreachable.+takeDMVar :: (Binary a) => DMVar a -> IO a+takeDMVar (DMVarLocal _ v o)+  = do+    a <- takeMVar v+    putMVar o (a, Nothing)+    return a+takeDMVar (DMVarRemote a)+  = do+    unsafeAccessForeignResource a requestTake+++-- | Writes a value to the DMvar. Blocks if the Variable is not empty.+--   This may throw an exception if the owner of the variable is unreachable.+putDMVar :: (Binary a) => DMVar a -> a -> IO ()  +putDMVar (DMVarLocal _ v o) d+  = do+    (_,mbDhi) <- takeMVar o+    case mbDhi of+      (Just dhi) -> delForeignHandler dhi+      (Nothing)  -> return ()+    putMVar v d+putDMVar (DMVarRemote a) d+  = do+    unsafeAccessForeignResource a (requestPut d)+

@@ -0,0 +1,46 @@++-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Distribution.DNode+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +  Public interface of the DNode datatype. See Holumbus.Distribution.DNode.Base+  for further documentation.+-}++-- ----------------------------------------------------------------------------+module Holumbus.Distribution.DNode+(+    DistributedException(..)+    +  , DNodeConfig(..)+  , defaultDNodeConfig+  +  , DNodeId+  , mkDNodeId+  , DNodeAddress+  , mkDNodeAddress+  +  , DHandlerId+      +  , initDNode+  , deinitDNode+  , addForeignDNode+  , delForeignDNode+  , checkForeignDNode+  , addForeignDNodeHandler+  , addForeignDResourceHandler+  , delForeignHandler+  +  , getDNodeData                  -- debug+)+where++import           Holumbus.Distribution.DNode.Base

@@ -0,0 +1,970 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Distribution.DNode.Base+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+  +  The main module for the implementation of the distributed data structures.+  It contains the DNode-datatype which is needed to register new local and+  remote resources. The main datatypes (Ids, Handlers, etc.) are also defined+  here.++  The module should only be used from within this library. User applications+  should refer to Holumbus.Distribution.DNode.+-}++-- ----------------------------------------------------------------------------++{-# OPTIONS_GHC -XDeriveDataTypeable -XScopedTypeVariables #-}+module Holumbus.Distribution.DNode.Base+(+    -- * datatypes+    DistributedException(..)+    +  , DNodeConfig(..)+  , defaultDNodeConfig+  +  , DNodeId+  , mkDNodeId+  , DNodeAddress+  , mkDNodeAddress+  +  , DHandlerId+  +  -- rf == only to be used by new resource implementations++  , DResourceType                -- rf+  , mkDResourceType              -- rf+  +  , DResourceId+  , DResourceAddress+  , mkDResourceAddress           -- rf+  +  , DResourceDispatcher          -- rf+  , DResourceEntry(..)           -- rf+    +  -- * Initializing and Deinitializing of the local node+  , initDNode+  , deinitDNode++  -- * adding, deleting other nodes+  , addForeignDNode+  , delForeignDNode++  -- * checking other nodes and resources+  , checkForeignDNode+  , addForeignDNodeHandler+  , addForeignDResourceHandler+  , delForeignHandler+  +  -- * needed by the resources+  , genLocalResourceAddress      -- rf+  , addLocalResource             -- rf+  , delLocalResource             -- rf+  , delForeignResource           -- rf+  , safeAccessForeignResource    -- rf+  , unsafeAccessForeignResource  -- rf+  , getByteStringMessage         -- rf -- reimported from Network-Module+  , putByteStringMessage         -- rf -- reimported from Network-Module+  , getDNodeData                 -- debug+)+where++import           Prelude hiding (catch)++import           Control.Exception+import           Control.Concurrent+import           Data.Typeable+import           Data.Binary+import           Data.Char+import           Data.Maybe+import           Data.Unique+import qualified Data.Map as Map+import qualified Data.Set as Set+import           Network.Socket (HostName, PortNumber)+import           System.IO+import           System.IO.Unsafe+import           System.Log.Logger+import           System.Random++import           Holumbus.Distribution.DNode.Network+++localLogger :: String+localLogger = "Holumbus.Distribution.DNode.Base"+++-- ----------------------------------------------------------------------------+-- Basic types which are commonly used+-- ----------------------------------------------------------------------------++-- | The exception type, used by distributed communication+data DistributedException = DistributedException {+    distEx_msg :: String -- ^ the message of the exception+  , distEx_fct :: String -- ^ the function in which the exception was thrown+  , distEx_mod :: String -- ^ the module in which the exception was thrown+  } deriving(Typeable, Show)+instance Exception DistributedException+++-- | A distributed Id type, could be named or randomly generated+data DId+  = DIdName String+  | DIdNumber Integer+  deriving (Show, Eq, Ord)+  +instance Binary DId where+  put (DIdName s) = putWord8 0 >> put s+  put (DIdNumber n) = putWord8 1 >> put n+  get+    = do+      t <- getWord8+      case t of+        0 -> get >>= \s -> return (DIdName s)+        _ -> get >>= \n -> return (DIdNumber n)+  +-- | Generates a distributed Id value. If the string is empty, a random+--   but unique id will be created+genDId :: String -> IO DId+genDId "" +  = do+    -- TODO make more unique -> uuids or other things+    i <- getStdRandom (randomR (1,1000000)):: IO Integer+    return $ DIdNumber i+genDId st+  = do+    return $ DIdName st+  +++-- ----------------------------------------------------------------------------+-- DNode related types+-- ----------------------------------------------------------------------------++-- | The configuration of a DNode. You need it to create a DNode and you can +--   use this data type it to alter its properties. This type is public to+--   allow users to create their own configuration.+data DNodeConfig = DNodeConfig {+    dnc_Name        :: String+  , dnc_MinPort     :: Int+  , dnc_MaxPort     :: Int+  , dnc_AccessDelay :: Int+  , dnc_PingDelay   :: Int+  } deriving (Show)++-- | A good default configuration. To create an unnamed node, just leave the+--   string empty. +defaultDNodeConfig :: String -> DNodeConfig+defaultDNodeConfig s+  = DNodeConfig s 8000 9000 1000000 1000000++++-- | The DNode identifier.+--   Every DNode has an Id, this could be named or randomly created. The id +--   could not be used to address a DNode directly over a Network connection +--   because the physical references are missing. The DNodeId is meant to +--   create a declarative reference which could be used to lookup purposes. +--   Think of the DNodeId as a domain name, without a DNS-Server to resolve +--   the physical address, it is worthless to establish a communication. +newtype DNodeId = DNodeId DId+  deriving (Show, Eq, Ord)++instance Binary DNodeId where+  put (DNodeId i) = put i+  get = get >>= \i -> return (DNodeId i)+  +-- | Generates a new DNodeId. If the string is empty, a random but unique+--   id will be created. +genDNodeId :: String -> IO DNodeId+genDNodeId s = genDId s >>= \i -> return (DNodeId i)++-- | Use this to make a new DNodeId from a String+mkDNodeId :: String -> DNodeId+mkDNodeId "" = error "the name of DNode is empty"+mkDNodeId s  = DNodeId $ DIdName s+++-- | The DNode address.+data DNodeAddress = DNodeAddress {+    dna_Id          :: DNodeId+  , dna_HostName    :: HostName+  , dna_ServiceName :: PortNumber+  } deriving (Show, Eq, Ord)++instance Binary DNodeAddress where+  put (DNodeAddress i hn po) = put i >> put hn >> put (toInteger po)+  get = get >>= \i -> get >>= \hn -> get >>= \po -> return (DNodeAddress i hn (fromInteger po))+++-- | use this to make a new DNodeAddress+mkDNodeAddress :: String -> HostName -> Int -> DNodeAddress+mkDNodeAddress s hn po = DNodeAddress i hn (fromIntegral po)+  where+  i = mkDNodeId s+  +-- | The type of the functions which could be called by a handler.+--   The DHandlerId is the Id of the handler which called the function,+--   you can use this to delete the handler.+type DHandlerFunction = DHandlerId -> IO ()++-- | Internal data of another DNode+data DNodeEntry = DNodeEntry {+    dne_Address              :: DNodeAddress+  , dne_PingThreadId         :: Maybe ThreadId+  , dne_HandlerFunctions     :: Map.Map DHandlerId DHandlerFunction+  , dne_TriggeredHandlers    :: Set.Set DHandlerId+  , dne_PositiveNodeHandlers :: Set.Set DHandlerId+  , dne_NegativeNodeHandlers :: Set.Set DHandlerId+  , dne_PositiveResourceHandlers :: Set.Set DHandlerId+  , dne_NegativeResourceHandlers :: Set.Set DHandlerId+  }++instance Show DNodeEntry where+  show _ = "{DNodeEntry}"++-- | The Id of a handler, is needed to stop the handler from further execution.+data DHandlerId = DHandlerId {+    dhi_Id      :: Unique+  , dhi_DNodeId :: DNodeId+  , dhi_DResourceId :: Maybe DResourceId+  } deriving (Eq, Ord)++  +-- ----------------------------------------------------------------------------+-- DResource related types+-- ----------------------------------------------------------------------------++-- | The ressouce type, it is used to separate between different kinds of+--   resources. The resource type is generated by the programmer of a+--   resource+newtype DResourceType = DResourceType String+  deriving (Show, Eq, Ord)++instance Binary DResourceType where+  put (DResourceType s) = put s+  get = get >>= \s -> return (DResourceType s)++mkDResourceType :: String -> DResourceType+mkDResourceType s = DResourceType (map toUpper s)++++-- | The DResource Id.+data DResourceId = DResourceId DId DResourceType+  deriving (Show, Eq, Ord)++instance Binary DResourceId where+  put (DResourceId i t) = put i >> put t +  get = get >>= \i -> get >>= \t -> return (DResourceId i t)++genDResourceId :: DResourceType -> String -> IO DResourceId+genDResourceId t s = genDId s >>= \i -> return (DResourceId i t)++mkDResourceId :: DResourceType -> String -> DResourceId+mkDResourceId _ "" = error "the name of the DResource is empty"+mkDResourceId t s  = DResourceId (DIdName s) t+++-- | The DResource address+data DResourceAddress = DResourceAddress {+    dra_Id     :: DResourceId+  , dra_NodeId :: DNodeId+  } deriving (Show, Eq, Ord)++instance Binary DResourceAddress where+  put (DResourceAddress i a) = put i >> put a+  get = get >>= \i -> get >>= \a -> return (DResourceAddress i a)++mkDResourceAddress :: DResourceType -> String -> String -> DResourceAddress+mkDResourceAddress t r n = DResourceAddress dri dni+  where+  dri = mkDResourceId t r+  dni = mkDNodeId n++-- | The DResource callback functions+type DResourceDispatcher = DNodeId -> Handle -> IO ()++-- | The container for the DResources+data DResourceEntry = DResourceEntry {+    dre_Dispatcher   :: DResourceDispatcher+  }++instance Show DResourceEntry where+  show _ = "{DResourceEntry}"++++-- | The local DNode, +--   manages the Resources,+--   keeps a record of foreign DNodes+data DNodeData = DNodeData {+    dnd_Address         :: DNodeAddress+  , dnd_SocketServer    :: SocketServer+  , dnd_AccessDelay     :: Int+  , dnd_PingDelay       :: Int+  , dnd_NodeMap         :: Map.Map DNodeId DNodeEntry+  , dnd_OwnResourceMap :: Map.Map DResourceId DResourceEntry+  } deriving (Show)++type DNode = MVar DNodeData+++{-# NOINLINE myDNode #-}+myDNode :: DNode+myDNode+  = do+    unsafePerformIO $ newEmptyMVar+++-- | Initializes the DNode of the program. You have to call this function+--   once BEFORE you can use other functions. +initDNode :: DNodeConfig -> IO DNodeId+initDNode c+  = do+    let minPort     = fromIntegral $ dnc_MinPort c+        maxPort     = fromIntegral $ dnc_MaxPort c+        accessDelay = dnc_AccessDelay c+        pingDelay   = dnc_PingDelay c+    n   <- genDNodeId (dnc_Name c)+    serverSocket <- startSocketServer (dispatcher) minPort maxPort+    case serverSocket of+      Nothing ->+        do+        errorM localLogger "no socket opened"+        error "socket could not be opened"+      (Just ss) ->+        do+        let hn = getSocketServerName ss+            po = getSocketServerPort ss+            dnd = DNodeData {+            dnd_Address         = DNodeAddress n hn po+          , dnd_SocketServer    = ss+          , dnd_AccessDelay     = accessDelay+          , dnd_PingDelay       = pingDelay+          , dnd_NodeMap         = Map.empty+          , dnd_OwnResourceMap = Map.empty+          }+        success <- tryPutMVar myDNode dnd+        if success+          then do return ()          +          else do+            stopSocketServer ss+            error "dnode already initialized"+    return n+++-- | deinitializes a DNode+deinitDNode :: IO ()+deinitDNode+  = do+    mbDnd <- tryTakeMVar myDNode+    case mbDnd of+      (Just dnd) ->+        do+        -- TODO close all threads and unregister from all other nodes...+        -- close the dispatcher thread, so no requests are handled...+        stopSocketServer (dnd_SocketServer dnd)+      (Nothing) ->+        do+        error "dnode already deinitialized"+    +    ++addNode :: DNodeAddress -> IO ()+addNode dna+  = do+    let dne = DNodeEntry dna Nothing Map.empty Set.empty Set.empty Set.empty Set.empty Set.empty+        dni = dna_Id dna+    modifyMVar_ myDNode $ \dnd -> return $ addNodeP dni dne dnd+   +      +addNodeP :: DNodeId -> DNodeEntry -> DNodeData -> DNodeData+addNodeP dni dne dnd = dnd { dnd_NodeMap = am }+  where+  am = Map.insert dni dne (dnd_NodeMap dnd)++deleteNode :: DNodeId -> IO ()+deleteNode i = modifyMVar_ myDNode $ \dnd -> return $ deleteNodeP i dnd ++deleteNodeP :: DNodeId -> DNodeData -> DNodeData+deleteNodeP i dnd = dnd { dnd_NodeMap = am }+  where+  am = Map.delete i (dnd_NodeMap dnd)++lookupNode :: DNodeId -> IO (Maybe DNodeEntry)+lookupNode i = withMVar myDNode $ \dnd -> return $ lookupNodeP i dnd++lookupNodeP :: DNodeId -> DNodeData -> Maybe DNodeEntry+lookupNodeP i dnd = Map.lookup i (dnd_NodeMap dnd)++-- lookupAllNodes :: IO ([DNodeEntry])+-- lookupAllNodes = withMVar myDNode $ \dnd -> return $ lookupAllNodesP dnd++-- lookupAllNodesP :: DNodeData -> [DNodeEntry]+-- lookupAllNodesP dnd = Map.elems (dnd_NodeMap dnd)++addResource :: DResourceId -> DResourceEntry -> IO ()+addResource i d = modifyMVar_ myDNode $ \dnd -> return $ addResourceP dnd i d++addResourceP :: DNodeData -> DResourceId -> DResourceEntry -> DNodeData+addResourceP dnd i d = dnd { dnd_OwnResourceMap = rm }+  where+  rm = Map.insert i d (dnd_OwnResourceMap dnd)++deleteResource :: DResourceId -> IO ()+deleteResource a = modifyMVar_ myDNode $ \dnd -> return $ deleteResourceP dnd a++deleteResourceP :: DNodeData -> DResourceId -> DNodeData+deleteResourceP dnd i = dnd { dnd_OwnResourceMap = rm }+  where+  rm = Map.delete i (dnd_OwnResourceMap dnd)++lookupResource :: DResourceId -> IO (Maybe DResourceEntry)+lookupResource i = withMVar myDNode $ \dnd -> return $ lookupResourceP dnd i++lookupResourceP :: DNodeData -> DResourceId -> Maybe DResourceEntry+lookupResourceP dnd i = Map.lookup i (dnd_OwnResourceMap dnd)+++getDNodeData :: IO (DNodeData)+getDNodeData = readMVar myDNode+++lookupDNodeReqestInfo :: DNodeId -> IO (DNodeAddress, Maybe DNodeAddress)+lookupDNodeReqestInfo dni+  = withMVar myDNode $ \dnd ->+      do+      let myDna      = dnd_Address dnd+          mbDne      = lookupNodeP dni dnd+          mbOtherDna = if (isJust mbDne) then (Just $ dne_Address $ fromJust mbDne) else Nothing +      return (myDna, mbOtherDna)+++checkDNodeRequest :: DNodeAddress -> DNodeAddress -> IO Bool+checkDNodeRequest s r+  = modifyMVar myDNode $ \dnd ->+      do+      let isValid = (dna_Id r) == (dna_Id $ dnd_Address dnd)+          dni     = dna_Id s+          dne     = DNodeEntry s Nothing Map.empty Set.empty Set.empty Set.empty Set.empty Set.empty+          dnd'    = if (Map.member dni (dnd_NodeMap dnd))+                    then dnd +                    else addNodeP dni dne dnd+      return (dnd', isValid)    ++-- ----------------------------------------------------------------------------++-- generic message container+data DNodeRequestMessage = DNodeRequestMessage {+    dnm_Req_Sender   :: DNodeAddress+  , dnm_Req_Receiver :: DNodeAddress+  , dnm_Req_Message  :: DNodeRequest+  } deriving (Show)+  +instance Binary DNodeRequestMessage where+  put (DNodeRequestMessage s r m) = put s >> put r >> put m+  get = get >>= \s -> get >>= \r -> get >>= \m -> return (DNodeRequestMessage s r m)++data DNodeRequest+  = DNMReqPing DNodeId [DResourceId]+  | DNMReqResourceMsg DResourceAddress+  deriving (Show)++instance Binary DNodeRequest where+  -- put (DNMReqRegister dna)   = putWord8 1 >> put dna+  -- put (DNMReqUnregister i)   = putWord8 2 >> put i+  put (DNMReqPing i rs)      = putWord8 3 >> put i >> put rs+  put (DNMReqResourceMsg a) = putWord8 4 >> put a+  get+    = do+      t <- getWord8+      case t of+        -- 1 -> get >>= \dna -> return (DNMReqRegister dna)+        -- 2 -> get >>= \i -> return (DNMReqUnregister i)+        3 -> get >>= \i -> get >>= \rs -> return (DNMReqPing i rs)+        4 -> get >>= \a -> return (DNMReqResourceMsg a)+        _ -> error "DNodeRequestMessage: wrong encoding"+++data DNodeResponseMessage+  = DNMRspOk+  | DNMRspPing [(DResourceId,Bool)]+  | DNMRspError String+  deriving (Show)+  +instance Binary DNodeResponseMessage where+  put(DNMRspOk)      = putWord8 1+  put(DNMRspPing rs) = putWord8 2 >> put rs+  put(DNMRspError e) = putWord8 3 >> put e+  get+    = do+      t <- getWord8+      case t of+        1 -> return (DNMRspOk)+        2 -> get >>= \rs -> return (DNMRspPing rs)+        3 -> get >>= \e -> return (DNMRspError e)+        _ -> error "DNodeResponseMessage: wrong encoding"++-- ----------------------------------------------------------------------------++    +-- | Delegates new incomming messages on a unix-socket to their streams.+dispatcher :: Handle -> IO ()+dispatcher hdl+  = do+    debugM localLogger "dispatcher: reading message from connection"+    raw <- getByteStringMessage hdl+    debugM localLogger "dispatcher: message received starting dispatching"+    let msg      = (decode raw)::DNodeRequestMessage+    let sender   = dnm_Req_Sender msg+    let receiver = dnm_Req_Receiver msg+    -- check id and add new addresses+    isValid <- checkDNodeRequest sender receiver+    if isValid+      then do+        debugM localLogger $ "dispatcher: Message: " ++ show msg+        case (dnm_Req_Message msg) of+          -- (DNMReqRegister dna) -> handleRegister dna hdl+          -- (DNMReqUnregister i) -> handleUnregister i hdl+          (DNMReqPing i rs)      -> handlePing i rs hdl+          (DNMReqResourceMsg a) -> handleResourceMessage (dna_Id sender) a hdl+      else do+        warningM localLogger $ "message for other node received... dropping request: " ++ show msg+        putByteStringMessage (encode $ DNMRspError "unknown receiver") hdl++{-+requestRegister :: Handle -> IO ()+requestRegister hdl+  = do+    dnd <- readMVar myDNode+    putMessage (encode $ DNMReqRegister $ dnd_Address dnd) hdl+    -- get the response+    raw <- getMessage hdl+    let rsp = (decode raw)::DNodeResponseMessage+    case rsp of+      (DNMRspOk)      -> return ()+      (DNMRspError e) -> error e+      _               -> error "false response"+++handleRegister :: DNodeAddress -> Handle -> IO ()+handleRegister dan hdl+  = do+    addNode dan+    -- put the response+    putMessage (encode $ DNMRspOk) hdl+    ++requestUnregister :: Handle -> IO ()+requestUnregister hdl+  = do+    dnd <- readMVar myDNode+    let i = dna_Id $ dnd_Address dnd+    putMessage (encode $ DNMReqUnregister i) hdl+    -- get the response+    raw <- getMessage hdl+    let rsp = (decode raw)::DNodeResponseMessage+    case rsp of+      (DNMRspOk)      -> return ()+      (DNMRspError e) -> error e+      _               -> error "false response"+ ++handleUnregister :: DNodeId -> Handle -> IO ()+handleUnregister i hdl+  = do+    deleteNode i+    -- put the response+    putMessage (encode $ DNMRspOk) hdl+-}++requestPing :: DNodeAddress -> DNodeAddress -> DNodeId -> [DResourceId] -> Handle -> IO (Maybe [(DResourceId, Bool)])+requestPing s r i rs hdl+  = do+    let request = DNodeRequestMessage s r (DNMReqPing i rs)+    putByteStringMessage (encode $ request) hdl+    raw <- getByteStringMessage hdl+    let rsp = (decode raw)::DNodeResponseMessage+    case rsp of+      (DNMRspOk)      -> return (Just [])+      (DNMRspPing ls) -> return (Just ls)+      (DNMRspError e) -> do+        errorM localLogger e+        return Nothing+++handlePing :: DNodeId -> [DResourceId] -> Handle -> IO ()+handlePing otherDni rs hdl+  = do+    dnd <- readMVar myDNode+    let myDni = dna_Id $ dnd_Address dnd+    if (myDni == otherDni)+      then do +        let orm = dnd_OwnResourceMap dnd+            ls  = map (\i -> (i, isJust $ Map.lookup i orm)) rs+        putByteStringMessage (encode $ DNMRspPing ls) hdl+      else do+        putByteStringMessage (encode $ DNMRspError "false ping - ids do not match") hdl++      +requestResourceMessage :: DNodeAddress -> DNodeAddress -> DResourceAddress -> (Handle -> IO a) -> Handle -> IO a+requestResourceMessage s r dra requester hdl+  = do+    let request = DNodeRequestMessage s r (DNMReqResourceMsg dra)+    -- ask node for resource+    debugM localLogger "requestResourceMessage: asking node for resource"+    putByteStringMessage (encode $ request) hdl+    -- get the response+    debugM localLogger "requestResourceMessage: getting the response"+    raw <- getByteStringMessage hdl+    debugM localLogger "requestResourceMessage: parsing the response"+    let rsp = (decode raw)::DNodeResponseMessage+    case rsp of+      (DNMRspOk)      -> do requester hdl+      (DNMRspError e) -> error e+      _               -> error "false response"+-- TODO throw real exception here... or something else+      +handleResourceMessage :: DNodeId -> DResourceAddress -> Handle -> IO ()+handleResourceMessage sender dra hdl+  = do+    let i = dra_Id dra+    mbDrd <- lookupResource i+    case mbDrd of+      (Just drd) ->+        do+        putByteStringMessage (encode $ DNMRspOk) hdl+        let handler = dre_Dispatcher drd+        handler sender hdl+      (Nothing) -> +        putByteStringMessage (encode $ DNMRspError "resource not found") hdl+      +-- ----------------------------------------------------------------------------+-- Functions for adding/deleting other nodes to the system+-- ----------------------------------------------------------------------------+++-- | Add a foreign DNode to the list of known DNodes.+--   Only DNodes in this list could be reached by the local node.+addForeignDNode :: DNodeAddress -> IO ()+addForeignDNode dna+  = do+    -- let hn = dna_HostName dna+    --     po = dna_PortNumber dna+    -- performSafeSendRequest requestRegister () hn po+    addNode dna +++-- | removes a foreign DNode entry. You should clean up the foreign DNode+--   entries.+delForeignDNode :: DNodeId -> IO ()+delForeignDNode i+  = do+    mDne <- lookupNode i+    case mDne of+      (Just _) ->+        do+        -- let dna = dne_Address dne+        --     hn  = dna_HostName dna+        --     po  = dna_PortNumber dna+        -- performSafeSendRequest requestUnregister () hn po+        deleteNode i+      (Nothing) -> return ()+++-- | Manually Checks, if another DNode is reachable. Returns true if this is the case,+--   otherwise false. Always returns, does not throw an exception caused by network failures.+checkForeignDNode :: DNodeId -> IO (Bool)+checkForeignDNode dni+  = do+    (myDna, mbOtherDna) <- lookupDNodeReqestInfo dni    +    case mbOtherDna of+      (Just otherDna) ->+        do+        let hn  = dna_HostName otherDna+            po  = dna_ServiceName otherDna+        res <- performSafeSendRequest (requestPing myDna otherDna dni []) Nothing hn po+        return $ isJust res+      (Nothing) -> return False+++-- | Adds a handler function which periodically checks the existences (or non-existence)+--   of other DNodes. The first parameter indicates the type of the handler.+--   If you want to install a handler which is fired when a Node becomes reachable (positive trigger),+--   it needs to be true. If you want to monitor the event when a specific node disappears,+--   pass false.+addForeignDNodeHandler :: Bool -> DNodeId -> DHandlerFunction -> IO (Maybe DHandlerId)+addForeignDNodeHandler positive dni f+  = modifyMVar myDNode $ \dnd ->+      do+      case (lookupNodeP dni dnd) of+        (Just dne) ->+          do+          let oldTid = dne_PingThreadId dne+          -- do we have an existing ping thread?+          tid <- if (isNothing oldTid)+            then do +              t <- startPingThread dni+              return $ Just t+            else return oldTid+          uid <- newUnique+          let dhi  = DHandlerId uid dni Nothing+              dne' = dne { dne_HandlerFunctions = Map.insert dhi f (dne_HandlerFunctions dne)+                         , dne_PositiveNodeHandlers = if positive +                                                        then Set.insert dhi (dne_PositiveNodeHandlers dne)+                                                        else (dne_PositiveNodeHandlers dne)+                         , dne_NegativeNodeHandlers = if positive+                                                        then (dne_NegativeNodeHandlers dne)+                                                        else Set.insert dhi (dne_NegativeNodeHandlers dne)+                         , dne_PingThreadId = tid }+          return ((addNodeP dni dne' dnd), Just dhi)+        (Nothing) -> return (dnd, Nothing)+++-- | Adds a handler function which periodically checks the existences (or non-existence)+--   of resources on other DNodes. The first parameter indicates the type of the handler.+--   If you want to install a handler which is fired when a Node becomes reachable (positive trigger),+--   it needs to be true. If you want to monitor the event when a specific node disappears,+--   pass false.+addForeignDResourceHandler :: Bool -> DResourceAddress -> DHandlerFunction -> IO (Maybe DHandlerId)+addForeignDResourceHandler positive dra f+  = modifyMVar myDNode $ \dnd ->+      do+      let dni   = dra_NodeId dra+          dri   = dra_Id dra+      case (lookupNodeP dni dnd) of+        (Just dne) ->+          do+          let oldTid = dne_PingThreadId dne+          -- do we have an existing ping thread?+          tid <- if (isNothing $ oldTid)+            then do +              t <- startPingThread dni+              return $ Just t+            else return oldTid+          uid <- newUnique+          let dhi  = DHandlerId uid dni (Just dri)+              dne' = dne { dne_HandlerFunctions = Map.insert dhi f (dne_HandlerFunctions dne)+                         , dne_PositiveResourceHandlers = if positive +                                                            then Set.insert dhi (dne_PositiveResourceHandlers dne)+                                                            else (dne_PositiveResourceHandlers dne)+                         , dne_NegativeResourceHandlers = if positive+                                                            then (dne_NegativeResourceHandlers dne)+                                                            else Set.insert dhi (dne_NegativeResourceHandlers dne)+                         , dne_PingThreadId = tid }+          return ((addNodeP dni dne' dnd), Just dhi)+        (Nothing) -> return (dnd, Nothing)+++-- | Deletes a Handler from the system, will not be called anymore.+delForeignHandler :: DHandlerId -> IO ()+delForeignHandler dhi+  = do+    modifyMVar_ myDNode $ \dnd ->+      do+      let dni   = dhi_DNodeId dhi+      case (lookupNodeP dni dnd) of+        (Just dne) ->+          do+          let dne' = dne { dne_HandlerFunctions = Map.delete dhi (dne_HandlerFunctions dne)+                     , dne_TriggeredHandlers = Set.delete dhi (dne_TriggeredHandlers dne)+                     , dne_PositiveResourceHandlers = Set.delete dhi (dne_PositiveResourceHandlers dne)+                     , dne_NegativeResourceHandlers = Set.delete dhi (dne_NegativeResourceHandlers dne)+                     , dne_PositiveNodeHandlers = Set.delete dhi (dne_PositiveNodeHandlers dne)+                     , dne_NegativeNodeHandlers = Set.delete dhi (dne_NegativeNodeHandlers dne)+                     }+          return $ addNodeP dni dne' dnd+        (Nothing) -> return dnd+++genLocalResourceAddress :: DResourceType -> String -> IO DResourceAddress+genLocalResourceAddress t s+  = do+    i <- genDResourceId t s+    dnd <- readMVar myDNode+    return $ DResourceAddress i (dna_Id $ dnd_Address dnd)+    ++addLocalResource :: DResourceAddress -> DResourceEntry -> IO ()+addLocalResource a d+  = do+    let i = dra_Id a+    addResource i d+++delLocalResource :: DResourceAddress -> IO ()+delLocalResource a+  = do+    let i = dra_Id a+    deleteResource i+++delForeignResource :: DResourceAddress -> IO ()+delForeignResource dra+  = modifyMVar_ myDNode $ \dnd ->+      do+      let dni = dra_NodeId dra+          dri = dra_Id dra+      case (lookupNodeP dni dnd) of+        (Just dne) ->+          do+          let handlerIds = Set.fromList $ filter (\dhi -> (dhi_DResourceId dhi) == (Just dri)) $ Map.keys (dne_HandlerFunctions dne)+              dne'   = dne { dne_HandlerFunctions = Map.filterWithKey (\k _ -> Set.notMember k handlerIds) (dne_HandlerFunctions dne)+                           , dne_TriggeredHandlers = Set.difference (dne_TriggeredHandlers dne) handlerIds+                           , dne_PositiveResourceHandlers = Set.difference (dne_PositiveResourceHandlers dne) handlerIds+                           , dne_NegativeResourceHandlers = Set.difference (dne_NegativeResourceHandlers dne) handlerIds+                           }+          return $ addNodeP dni dne' dnd+        (Nothing) -> return dnd+++safeAccessForeignResource :: DResourceAddress -> (Handle -> IO a) -> IO a+safeAccessForeignResource dra requester+  = do+    debugM localLogger $ "accessForeignResource: " ++ show dra +    let dni = dra_NodeId dra+    (myDna, mbOtherDna) <- lookupDNodeReqestInfo dni    +    case mbOtherDna of+      (Just otherDna) ->+        do+        debugM localLogger $ "accessForeignResource: node in list found"+        let hn  = dna_HostName otherDna+            po  = dna_ServiceName otherDna+        mbres <- performMaybeSendRequest (requestResourceMessage myDna otherDna dra requester) hn po+        case mbres of+          (Just res) -> return res+          (Nothing)  -> retryAccess+      (Nothing) -> retryAccess+    where+      retryAccess+        = do+          dnd <- readMVar myDNode+          threadDelay $ dnd_AccessDelay dnd+          yield+          safeAccessForeignResource dra requester+++unsafeAccessForeignResource :: DResourceAddress -> (Handle -> IO a) -> IO a+unsafeAccessForeignResource dra requester+  = do+    debugM localLogger $ "accessForeignResource: " ++ show dra +    let dni = dra_NodeId dra+    (myDna, mbOtherDna) <- lookupDNodeReqestInfo dni    +    case mbOtherDna of+      (Just otherDna) ->+        do+        debugM localLogger $ "accessForeignResource: node in list found"+        let hn  = dna_HostName otherDna+            po  = dna_ServiceName otherDna+        catch (performUnsafeSendRequest (requestResourceMessage myDna otherDna dra requester) hn po)+          (\(e ::IOException) -> +            do+            debugM localLogger $  show e+            throwIO $ DistributedException (show e) "unsafeAccessForeignResource" "DNode")+      (Nothing) -> +        throwIO $ DistributedException "node not registered" "unsafeAccessForeignResource" "DNode"+    ++startPingThread :: DNodeId -> IO ThreadId+startPingThread otherDNodeId+  = do+    forkIO $ pingLoop+    where+    -- the function for the main ping loop+    pingLoop = do+      myDnd <- readMVar myDNode+      let myDna = dnd_Address myDnd+      -- get NodeData from local DNode+      mbDne <- lookupNode otherDNodeId+      case mbDne of+        (Just dne) -> do+          myTid <- myThreadId+          if ((Just myTid) == (dne_PingThreadId dne))+            then do+              -- do the pinging+              pingRes <- doPinging myDna dne+              -- evaluate the results and collect the handlers to execute+              (hdls,pingDelay) <- evaluatePingResult pingRes+              -- execute handlers+              sequence_ $ map (\(dhi,f) -> forkIO $ f dhi) hdls+              -- wait and redo the pinging+              threadDelay pingDelay+              pingLoop+            else do+              debugM localLogger $ "the thread ids don't match - leaving thread" ++ show otherDNodeId+              return ()+        (Nothing) -> do+          debugM localLogger $ "resource not found - leaving thread: " ++ show otherDNodeId+          return ()+    -- the function which does the pinging of the external node+    doPinging myDna dne = do+      let resources = mapMaybe (dhi_DResourceId) $ Map.keys $ dne_HandlerFunctions dne+          otherDna  = dne_Address dne+          hn        = dna_HostName otherDna+          po        = dna_ServiceName otherDna+      performSafeSendRequest (requestPing myDna otherDna otherDNodeId resources) Nothing hn po+    -- the function which evaluates the result of the ping (changes state of local DNode)+    evaluatePingResult res =+      modifyMVar myDNode $ \dnd -> do+        let pingDelay = dnd_PingDelay dnd+        (dnd', hdls) <- case (Map.lookup otherDNodeId (dnd_NodeMap dnd)) of+          -- the node entry still exists+          (Just dne) -> do    +            let tid            = if (hasAnyHandlers dne)then (dne_PingThreadId dne) else Nothing+                allPosNodeHdls = dne_PositiveNodeHandlers dne+                allPosResHdls  = dne_PositiveResourceHandlers dne+                allNegNodeHdls = dne_NegativeNodeHandlers dne+                allNegResHdls  = dne_NegativeResourceHandlers dne+                trigHdls       = dne_TriggeredHandlers dne+                hdlFuncs       = dne_HandlerFunctions dne+            (allTrigHdls, hdls) <- case (res) of+              -- we've got a response with the list of resources+              (Just ls) -> do+                let untrigPosNodeHdls = Set.difference allPosNodeHdls trigHdls+                    existingRes = Set.fromList $ map fst $ filter (snd) ls+                    missingRes = Set.fromList $ map fst $ filter (not . snd) ls+                    existingPosResHdls = Set.filter (isMatchingResHdl existingRes) allPosResHdls+                    missingPosResHdls = Set.filter (isMatchingResHdl missingRes) allPosResHdls+                    existingNegResHdls = Set.filter (isMatchingResHdl existingRes) allNegResHdls+                    missingNegResHdls = Set.filter (isMatchingResHdl missingRes) allNegResHdls+                    untrigPosResHdls = Set.difference existingPosResHdls trigHdls+                    untrigNegResHdls = Set.difference missingNegResHdls trigHdls+                    untrigHdls = Set.unions [untrigPosNodeHdls, untrigPosResHdls, untrigNegResHdls]+                    reUntrigHdls = Set.unions [allNegNodeHdls, missingPosResHdls, existingNegResHdls]+                    -- new triggered handler set+                    allTrigHdls = Set.union untrigHdls $ Set.difference trigHdls reUntrigHdls+                    -- list of Handlers to execute+                    hdls = filter (\(dhi,_) -> Set.member dhi untrigHdls) $ Map.toList hdlFuncs+                return (allTrigHdls, hdls)+              -- external node was not reachable+              (Nothing) -> do+                let untrigNegNodeHdls = Set.difference allNegNodeHdls trigHdls+                    untrigNegResHdls = Set.difference allNegResHdls trigHdls+                    untrigHdls = Set.union untrigNegNodeHdls untrigNegResHdls+                    -- new triggered handler set+                    allTrigHdls = Set.union allNegNodeHdls allNegResHdls+                    -- list of Handlers to execute+                    hdls = filter (\(dhi,_) -> Set.member dhi untrigHdls) $ Map.toList hdlFuncs+                return (allTrigHdls, hdls)+            let dne' = dne { dne_PingThreadId = tid, dne_TriggeredHandlers = allTrigHdls }+                dnd' = addNodeP otherDNodeId dne' dnd+            return (dnd', hdls)+          -- in the meantime, the node entry was deleted+          (Nothing) -> return (dnd, [])+        return (dnd', (hdls, pingDelay))+    isMatchingResHdl :: Set.Set DResourceId -> DHandlerId -> Bool+    isMatchingResHdl resIds dhi = if (isJust mbResId) then isMember else False+      where+        mbResId = dhi_DResourceId dhi+        resId = fromJust mbResId+        isMember = Set.member resId resIds+    hasAnyHandlers :: DNodeEntry -> Bool+    hasAnyHandlers dne = not $ Map.null (dne_HandlerFunctions dne)+

@@ -0,0 +1,305 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Distribution.DNode.Network+  Copyright  : Copyright (C) 2008 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  The Server-Module for the Holumbus framework.+  +  It contains the lowlevel functions, like the socket handling (opening, +  reading, writing, ...).+  +-}++-- ----------------------------------------------------------------------------++{-# OPTIONS -fglasgow-exts #-}+module Holumbus.Distribution.DNode.Network+    (+      -- * Server-Datatypes+      SocketServer+    , getSocketServerName+    , getSocketServerPort+    +    , HandlerFunction+    +    -- * Server-Operations+    , startSocketServer+    , stopSocketServer++    -- * Client-Operations+    , performUnsafeSendRequest+    , performSafeSendRequest    +    , performMaybeSendRequest++    -- * Handle-Operations+    , putByteStringMessage+    , getByteStringMessage+    )+where++import           Prelude hiding         ( catch )++import           Control.Concurrent+import           Control.Exception      (+            Exception+          , IOException+          , bracket+          , catch+          )++import qualified Data.ByteString.Lazy   as B+import           Data.Typeable++import           Network+import qualified Network.Socket         as Socket++import           System.Log.Logger+import           System.CPUTime+import           System.IO+import           System.Posix++import           Text.Printf++import           Holumbus.Common.Utils (handleAll)+++-- | Logger+localLogger :: String+localLogger = "Holumbus.Distribution.DNode.Network"+++-- ----------------------------------------------------------------------------+-- exception stuff+-- ----------------------------------------------------------------------------++-- | This exception is needed to stop the socket server threads in a controlled+--   way and to distinguish this kind of exception from IOExceptions.+data SocketServerException = SocketServerException ThreadId+  deriving (Typeable, Show)++instance Exception SocketServerException where+++-- ----------------------------------------------------------------------------+-- Server-Datatypes+-- ----------------------------------------------------------------------------++data SocketServer = SocketServer { +    ss_ThreadId    :: ! ThreadId+  , ss_HostName    :: ! HostName+  , ss_PortNumber  :: ! PortNumber+  } deriving (Show)++getSocketServerName :: SocketServer -> HostName+getSocketServerName = ss_HostName++getSocketServerPort :: SocketServer -> PortNumber+getSocketServerPort = ss_PortNumber+++type HandlerFunction a = Handle -> IO a+++-- ----------------------------------------------------------------------------+-- Server-Operations+-- ----------------------------------------------------------------------------++-- | Creates a new (unix-)socket and starts the listener in its own thread.+--   You'll get a reference to the listener Thread, so you can kill it with+--   stopSocketServer.+--   It is also possible to give a range of PortNumbers on which the socket+--   will be opened. The first portnumber available will be taken.+startSocketServer+  :: HandlerFunction () -- ^ dispatcher function+  -> Int                -- ^ start port number+  -> Int                -- ^ end port number+  -> IO (Maybe SocketServer)+startSocketServer f actPo maxPo+  = do+    s <- (getFirstSocket actPo maxPo)+    case s of+      Nothing -> +        return Nothing+      (Just (so, po)) ->+        do+        hn <- getHostName+        tid <- forkIO $ +          do+          handleAll+            (\e ->+              do+              putStrLn $ "ERROR - socket closed with exception: " ++ show e +              sClose so+            ) $+            do+            {- 6.8+            catchDyn (waitForRequests f so (SocketId hn po)) (handler so)+            -}+            catch+              (waitForRequests f so)+              (handler so)+        return (Just $ SocketServer tid hn po)+    where+    handler :: Socket -> SocketServerException -> IO ()+    handler so (SocketServerException i)+        = do+          sClose so+          putStrLn $ "socket normally closed by thread " ++ show i +++-- | Stops a socker server by its internal thread id.+stopSocketServer :: SocketServer -> IO ()+stopSocketServer ss+  = do+    let i = ss_ThreadId ss+    me <- myThreadId+    debugM localLogger $ "stopping socket server... with threadId: " ++ show i ++ " - form threadId: " ++ show me+    throwTo i (SocketServerException me)+    yield+++-- | Gets the hostname of the computer of just "localhost".+getHostName :: IO (HostName)+getHostName+  = do+    (hn, _) <- Socket.getNameInfo [] True False (Socket.SockAddrUnix "localhost")+    return (maybe "localhost" id hn)+++-- | Gets the first free port number and creates of new socket on it.+getFirstSocket :: Int -> Int -> IO (Maybe (Socket, PortNumber))+getFirstSocket actPo maxPo+  = do+    if (actPo > maxPo)+      then do+        return Nothing +      else do+        handleAll (return (getFirstSocket (actPo+1) maxPo)) $+          do +          debugM localLogger $ "getFirstSocket: getting Socket for: " ++ show actPo+          socket <- getSocket $ fromIntegral actPo+          return (Just (socket, fromIntegral actPo))     +++-- | Opens a socket on a port number.+getSocket :: PortNumber -> IO (Socket)+getSocket po =+  -- for MS-Windows Systems+  withSocketsDo $ do+  -- Don't let the server be terminated by sockets closed unexpectedly by the client.+  _ <- installHandler sigPIPE Ignore Nothing+  socket <- listenOn (PortNumber po)+  return socket+++-- | Listens to a socket and opens a new dispatcher thread for every incomming+--   data.+waitForRequests :: HandlerFunction () -> Socket -> IO ()+waitForRequests f socket = +  do+  (hdl,_,_) <- accept socket+  _ <- forkIO $ processRequest f hdl  -- Spawn new thread to answer the current request.+  waitForRequests f socket       -- Wait for more requests.+++-- | A wrapper around the user defined dispatcher function.+--   Mesures the time and catches unhandled exceptions.+processRequest :: HandlerFunction () -> Handle -> IO ()+processRequest f conn = +  bracket (return conn) (hClose) (processRequest')+    where+    processRequest' hdl = +      do+      hSetBuffering hdl $ BlockBuffering Nothing+      -- Dispatch the request and measure the processing time.+      t1 <- getCPUTime+      debugM localLogger "starting to dispatch request"+      handleAll (\e -> errorM localLogger $ "UnknownError: " ++ show e) $ f hdl+      t2 <- getCPUTime+      d <- return ((fromIntegral (t2 - t1) / 1000000000000) :: Float)+      ds <- return (printf "%.4f" d)+      infoM localLogger ("request processed in " ++ ds ++ " sec")+++-- ----------------------------------------------------------------------------+-- Client-Operations+-- ----------------------------------------------------------------------------++    +-- | Send the query to a server and merge the result with the global result.+sendRequest :: HandlerFunction a -> HostName -> PortNumber -> IO a+sendRequest f n p = +  withSocketsDo $ do +    _ <- installHandler sigPIPE Ignore Nothing+    +    --TODO exception handling+    --handle (\e -> do putStrLn $ show e return False) $+    bracket (connectTo n (PortNumber p)) (hClose) (send)+    where    +    send hdl +      = do+        hSetBuffering hdl $ BlockBuffering Nothing+        f hdl +++-- no exception handling+performUnsafeSendRequest :: HandlerFunction a -> HostName -> PortNumber -> IO a+performUnsafeSendRequest = sendRequest+++-- all IOExceptions handled return of default value+performSafeSendRequest :: HandlerFunction a -> a -> HostName -> PortNumber -> IO a+performSafeSendRequest f d n p+  = catch (sendRequest f n p)+          (\(e ::IOException) -> +            do+            debugM localLogger $  show e +            return d)+++-- all IOExceptions handled return of Nothing+performMaybeSendRequest :: HandlerFunction a -> HostName -> PortNumber -> IO (Maybe a)+performMaybeSendRequest f n p+  = catch (do+           res <- sendRequest f n p+           return (Just res))+          (\(e ::IOException) -> +            do+            debugM localLogger $  show e +            return Nothing)+++-- ----------------------------------------------------------------------------+-- Handle-Operations+-- ----------------------------------------------------------------------------++-- | Puts a bytestring to a handle. But to make the reading easier, we write+--   the length of the data as a message-header to the handle, too. +putByteStringMessage :: B.ByteString -> Handle -> IO ()+putByteStringMessage msg hdl+  = do+    handleAll (\e -> do+      errorM localLogger $ "putMessage: " ++ show e+      errorM localLogger $ "message: " ++ show msg +     ) $ do+      hPutStrLn hdl ((show $ B.length msg) ++ " ")+      B.hPut hdl msg+      hFlush hdl+++-- | Reads data from a stream. We define, that the first line of the message+--   is the message header which tells us how much bytes we have to read.+getByteStringMessage :: Handle -> IO (B.ByteString)+getByteStringMessage hdl+  = do+    line <- hGetLine hdl+    let pkg = words line+    raw <- B.hGet hdl (read $ head pkg)+    return raw

@@ -0,0 +1,194 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Distribution.DStreamPort+  Copyright  : Copyright (C) 2009 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  This module offers distributed streams and ports.++  Because a DChan allows external read access, the idea came up to+  split a DChan into two parts: a stream and a port. A stream only allows+  you to read data from it. The read-access is limited to the local process which+  created the stream. To send data to a stream, you need a port. This can be used+  on forgein nodes to send data to your local stream.+  +-}++-- ----------------------------------------------------------------------------++module Holumbus.Distribution.DStreamPort+(+  -- * datatypes+    DStream+  , DPort+  , StreamPortMessage(..)+  +  -- * creating and closing a stream+  , newDStream+  , closeDStream++  -- * operations on a stream+  , isEmptyDStream  +  , receive+  , receiveMsg+  , tryReceive+  , tryReceiveMsg+  , tryWaitReceive+  , tryWaitReceiveMsg+  , withStream+    +  -- * creating a port+  , newDPortFromStream+  , newDPort++  -- * operations on a port+  , send+  , sendWithGeneric+  , sendWithMaybeGeneric+)+where++import           Data.Binary+--import           Holumbus.Common.MRBinary+import qualified Data.ByteString.Lazy as B+import           System.Log.Logger++import           Holumbus.Distribution.DChan+++localLogger :: String+localLogger = "Holumbus.Distribution.DStreamPort"++-- ----------------------------------------------------------------------------+-- | Message Datatype.+--   We are sending additional information, to do debugging+data (Binary a) => StreamPortMessage a = StreamPortMessage {+    spm_Data           :: ! a                    -- ^ the data  +  , spm_Generic        :: ! (Maybe B.ByteString) -- ^ some generic data -- could be another port+  }++instance (Binary a) => Binary (StreamPortMessage a) where+  put(StreamPortMessage d g) = put d >> put g+  get = get >>= \d -> get >>= \g -> return $ (StreamPortMessage d g)+++-- ----------------------------------------------------------------------------+newtype DStream a = DStream (DChan (StreamPortMessage a))+++-- | Creates a new local stream.+newDStream :: (Binary a) => String -> IO (DStream a) +newDStream s+  = do+    dc <- newDChan s+    return (DStream dc)+++-- | Closes a stream.+closeDStream :: DStream a -> IO ()+closeDStream (DStream dc) = closeDChan dc+++-- | Tests, if a stream has no more data to read.+isEmptyDStream :: DStream a -> IO Bool+isEmptyDStream (DStream dc) = isEmptyDChan dc+++getMaybeMsgData :: (Binary a) => Maybe (StreamPortMessage a) -> Maybe a+getMaybeMsgData (Nothing)  = Nothing+getMaybeMsgData (Just msg) = Just (spm_Data msg)++-- | Reads the data packet of the next message from a stream.+--   If stream is empty, this function will block until a new message arrives.+receive :: (Binary a) => DStream a -> IO a+receive s = (receiveMsg s) >>= \msg -> return (spm_Data msg)+++-- | Reads the next message from a stream (data packet + message header).+--   If stream is empty, this function will block until a new message arrives.+receiveMsg :: (Binary a) => DStream a -> IO (StreamPortMessage a)+receiveMsg (DStream dc) = readDChan dc+++-- | Reads the data packet of the next message from a stream.+--   If stream is empty, this function will immediately return with Nothing.+tryReceive :: (Binary a) => DStream a -> IO (Maybe a)+tryReceive s = tryReceiveMsg s >>= \msg -> return $ getMaybeMsgData msg+++-- | Reads the next message from a stream (data packet + message header).+--   If stream is empty, this function will immediately return with Nothing.+tryReceiveMsg :: (Binary a) => DStream a -> IO (Maybe (StreamPortMessage a))+tryReceiveMsg (DStream dc) = tryReadDChan dc+++-- | Reads the data packet of the next message from a stream.+--   If stream is empty, this function will wait for new messages until the +--   time is up and if no message has arrived, return with Nothing.+tryWaitReceive :: (Binary a) => DStream a -> Int -> IO (Maybe a)+tryWaitReceive s t = (tryWaitReceiveMsg s t) >>= \msg -> return $ getMaybeMsgData msg+++-- | Reads the next message from a stream (data packet + message header).+--   If stream is empty, this function will wait for new messages until the +--   time is up and if no message has arrived, return with Nothing.+tryWaitReceiveMsg :: (Binary a) => DStream a -> Int -> IO (Maybe (StreamPortMessage a))+tryWaitReceiveMsg (DStream dc) t = tryWaitReadDChan dc t++    +-- | Encapsulates a stream.+--   A new stream is created, then some user-action is done an after that the+--   stream is closed. +withStream :: (Binary a) => (DStream a -> IO b) -> IO b+withStream f+  = do+    debugM localLogger "withStream: creating new stream"+    s <- newDStream ""+    debugM localLogger "withStream: new stream created"+    res <- f s+    closeDStream s+    return res++   +-- ----------------------------------------------------------------------------+newtype DPort a = DPort (DChan (StreamPortMessage a))++instance (Binary a) => Binary (DPort a) where+  put(DPort dc) = put dc+  get = get >>= \dc -> return $ (DPort dc)+    +    +-- | Creates a new Port, which is bound to a stream.+newDPortFromStream :: DStream a -> IO (DPort a)+newDPortFromStream (DStream dc) = return (DPort dc)+++-- | Creates a new port from a streamname and its socketId.+--   The first parameter is the name of the resource and the second one+--   the name of the node.+newDPort :: String -> String -> IO (DPort a)+newDPort r n = newRemoteDChan r n >>= \dc -> return (DPort dc)+    +   +-- | Send data to the stream of the port.+--   The data is send via network, if the stream is located on an external+--   processor+send :: (Binary a) => DPort a -> a -> IO ()+send p d = sendWithMaybeGeneric p d Nothing+++-- | Like "send", but here we can give some generic data (e.g. a port for reply +--   messages).+sendWithGeneric :: (Binary a) => DPort a -> a -> B.ByteString -> IO ()+sendWithGeneric p d rp = sendWithMaybeGeneric p d (Just rp) +++-- | Like "sendWithGeneric", but the generic data is optional+sendWithMaybeGeneric :: (Binary a) => DPort a -> a -> Maybe B.ByteString -> IO ()+sendWithMaybeGeneric (DPort dc) d rp = writeDChan dc (StreamPortMessage d rp)

@@ -0,0 +1,122 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Distribution.DValue+  Copyright  : Copyright (C) 2010 Stefan Schmidt+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  This module offers the distrubted value datatype.+  +  A DValue is like a distributed MVar, but is can only be+  set once on the local node. If the value is set, it cannot be+  changed any more.++  I don't know if this is useful at all, so the implementation is not+  quite finished an some things could be improved.+-}++-- ----------------------------------------------------------------------------++module Holumbus.Distribution.DValue+(+  -- * datatypes+    DValue+    +  -- * creating and closing a DValue+  , newDValue+  , newRemoteDValue+  , closeDValue++  -- * access a DValue+  , getDValue+  -- , flushDValue+)+where++import           Control.Concurrent.MVar++import           Data.Binary++import           Holumbus.Distribution.DMVar+++-- localLogger :: String+--localLogger = "Holumbus.Distribution.DValue"+++data DValue a+  = DValueLocal (DMVar a)+  | DValueRemote (DMVar a) (MVar a)+++-- | Creates new DValue on the local DNode. The first parameter+--   is the name of the value which could be used in other processes to+--   access this stream. If you leave it empty, a random Id will be created.+newDValue :: (Binary a) => String -> a -> IO (DValue a)+newDValue s a+  = do+    dv <- newDMVar s a+    return (DValueLocal dv)+    ++-- | Creates a new DValue.+--   The first parameter is the name of the resource and the second one+--   the name of the node.+newRemoteDValue :: String -> String -> IO (DValue a)+newRemoteDValue r n+  = do+    dv <- newRemoteDMVar r n+    lv <- newEmptyMVar+    return (DValueRemote dv lv)+++-- | Closes a DValue.+closeDValue :: (DValue a) -> IO ()+closeDValue (DValueLocal dv)+  = do+    closeDMVar dv+closeDValue (DValueRemote dv _)+  = do+    closeDMVar dv+++-- | Gets value. It will only access the network on the first+--   time and my throw an exception. It returns always (Just a)+--    but this may change in the future, so the type sticks to (Maybe a)+getDValue :: (Binary a) => DValue a -> IO (Maybe a)+getDValue (DValueLocal dv)+  = do+    a <- readDMVar dv+    return (Just a)+getDValue (DValueRemote dv lv)+  = do+    e <- isEmptyMVar lv+    if e+      then do+        --TODO exception handling here...+        a <- readDMVar dv+        putMVar lv a+        return (Just a)+      else do+        a <- readMVar lv+        return (Just a)+++{-+flushDValue :: DValue a -> IO ()+flushDValue (DValueLocal _) = return ()+flushDValue (DValueRemote _ lv)+  = do+    e <- isEmptyMVar lv+    if e+      then return ()+      else do+        _ <- takeMVar lv+        return ()+-}+

@@ -28,6 +28,7 @@  {-# OPTIONS -fglasgow-exts #-} module Holumbus.Network.Communication+{-# DEPRECATED "this module will be remove in the next release, please use the packages from Holumbus.Distribution.*" #-} (   StreamName       -- (reexport) , SocketId         -- (reexport)@@ -35,6 +36,7 @@  -- time constants , time30           -- (reexport)+, time60           -- (reexport) , timeIndefinitely -- (reexport)  , IdType@@ -54,16 +56,18 @@  -- * client operations , ClientClass(..)-, Client+, ClientData(..)+, Client(..) , newClient , closeClient-, ClientPort+, ClientPort(..) , sendRequestToClient  ) where  import           Control.Concurrent import           Data.Binary+--import           Holumbus.Common.MRBinary import qualified Data.ByteString.Lazy as B import qualified Data.Map as Map import           Data.Maybe@@ -275,7 +279,7 @@     debugM localLogger "closeServer: closeStream"     closeStream stream     debugM localLogger "closeServer: unregister clients"-    mapM (\i -> do unregisterClient i s) allIds+    _ <- mapM (\i -> do unregisterClient i s) allIds     debugM localLogger "closeServer: end"     return () @@ -479,6 +483,17 @@         putStrLn $ "SiteToClientMap: " ++ show (sd_SiteToClientMap sd)         putStrLn $ "SiteMap:         " ++ show (sd_SiteMap sd)         putStrLn $ "NextId:          " ++ show (sd_NextId sd)+  getDebug (Server server)+    = withMVar server $+        \sd ->+        do+        return   $ ("printServer"+          ++"\n"++ "OwnStream:       " ++ show (sd_OwnStream sd)+          ++"\n"++ "OwnPort:         " ++ show (sd_OwnPort sd)+          ++"\n"++ "ClientMap:       " ++ show (sd_ClientMap sd)+          ++"\n"++ "SiteToClientMap: " ++ show (sd_SiteToClientMap sd)+          ++"\n"++ "SiteMap:         " ++ show (sd_SiteMap sd)+          ++"\n"++ "NextId:          " ++ show (sd_NextId sd))++"\n"     @@ -658,7 +673,7 @@ newClient   :: (Binary a, Binary b)   => StreamName -> Maybe SocketId-  -> (a -> IO (Maybe b))  -- ^ the individual request dispatcher for the client+  -> (a -> IO (Maybe b))  -- ^ the individual request dispatcher for the client -- (ReqM -> IO (RespM)   -> IO Client newClient sn soid action   = do  @@ -675,7 +690,7 @@     -- first, we start the server, because we can't handle requests without it     startRequestDispatcher stid st (dispatchClientRequest client action)     -- then we try to register a the server-    setThreadDelay 5000000 ptid+    setThreadDelay 2000000 ptid     setThreadAction (checkServer sp sid po client) ptid     setThreadErrorHandler (handleCheckServerError client) ptid     startThread ptid@@ -777,7 +792,17 @@         putStrLn $ "OwnStream:  " ++ show (cd_OwnStream cd)         putStrLn $ "OwnPort:    " ++ show (cd_OwnPort cd)         putStrLn $ "ServerPort: " ++ show (cd_ServerPort cd)-+  getDebug (Client client)+    = withMVar client $+        \cd ->+        do+        return ( "printClient"+          ++"\n"++ "Id:         " ++ show (cd_Id cd)+          ++"\n"++ "LifeValue   " ++ show (cd_LifeValue cd)+          ++"\n"++ "Site:       " ++ show (cd_SiteId cd)+          ++"\n"++ "OwnStream:  " ++ show (cd_OwnStream cd)+          ++"\n"++ "OwnPort:    " ++ show (cd_OwnPort cd)+          ++"\n"++ "ServerPort: " ++ show (cd_ServerPort cd)++"\n")   @@ -787,7 +812,22 @@  -- | Just a wrapper around a port. data ClientPort = ClientPort (Port ClientRequestMessage)-  deriving (Show)+  deriving (Show,Eq)++instance Ord ClientPort where  --p_StreamName+  compare (ClientPort p1) (ClientPort p2) = compare (p_StreamName p1) (p_StreamName p2)+  (<) (ClientPort p1) (ClientPort p2)     = (<)     (p_StreamName p1) (p_StreamName p2)+  (>) (ClientPort p1) (ClientPort p2)     = (>)     (p_StreamName p1) (p_StreamName p2)+  (>=) (ClientPort p1) (ClientPort p2)    = (>=)    (p_StreamName p1) (p_StreamName p2)+  (<=)  (ClientPort p1) (ClientPort p2)   = (<=)    (p_StreamName p1) (p_StreamName p2)+  max c1@(ClientPort p1) c2@(ClientPort p2) = if (max s1 s2) == s1 then c1 else c2+    where+    s1 = p_StreamName p1+    s2 = p_StreamName p2+  min c1@(ClientPort p1) c2@(ClientPort p2) = if (min s1 s2) == s1 then c1 else c2+    where+    s1 = p_StreamName p1+    s2 = p_StreamName p2  instance Binary ClientPort where   put (ClientPort p) = put p

@@ -21,6 +21,7 @@  {-# OPTIONS -fglasgow-exts #-} module Holumbus.Network.Core+{-# DEPRECATED "this module will be remove in the next release, please use the packages from Holumbus.Distribution.*" #-}     (       -- * Socket-Descriptor       SocketId(..)@@ -29,7 +30,13 @@     , startSocket        -- * Client-Operations+      -- deprecated     , sendRequest+    +      -- use this+    , performUnsafeSendRequest+    , performSafeSendRequest    +    , performMaybeSendRequest        -- * Handle-Operations     , putMessage@@ -43,12 +50,13 @@  import           Control.Concurrent import           Control.Exception      ( Exception-					, bracket-					, catch-					)-import           Control.Monad+          , IOException+          , bracket+          , catch+          )  import           Data.Binary+--import           Holumbus.Common.MRBinary import qualified Data.ByteString.Lazy   as B import           Data.Typeable @@ -70,7 +78,7 @@ localLogger = "Holumbus.Network.Core"  -type ServerDispatcher = SocketId -> Handle -> HostName -> PortNumber -> IO ()+type ServerDispatcher = SocketId -> Handle -> SocketId -> IO ()  -- ------------------------------------------------------------ --@@ -187,7 +195,7 @@   -- for MS-Windows Systems   withSocketsDo $ do   -- Don't let the server be terminated by sockets closed unexpectedly by the client.-  installHandler sigPIPE Ignore Nothing+  _ <- installHandler sigPIPE Ignore Nothing   socket <- listenOn po   return socket @@ -198,7 +206,7 @@ waitForRequests f socket soid =    do   client <- accept socket-  forkIO $ processRequest f soid client   -- Spawn new thread to answer the current request.+  _ <- forkIO $ processRequest f soid client   -- Spawn new thread to answer the current request.   waitForRequests f socket soid       -- Wait for more requests.  @@ -215,7 +223,7 @@       t1 <- getCPUTime       debugM localLogger "starting to dispatch request"       handleAll (\e -> errorM localLogger $ "UnknownError: " ++ show e) $ do-        f soid hdl hst prt+        f soid hdl (SocketId hst prt)       t2 <- getCPUTime       d <- return ((fromIntegral (t2 - t1) / 1000000000000) :: Float)       ds <- return (printf "%.4f" d)@@ -228,21 +236,45 @@       -- | Send the query to a server and merge the result with the global result.-sendRequest :: (Handle -> IO a) -> HostName -> PortID -> IO a+sendRequest :: (Handle -> IO a) -> HostName -> PortNumber -> IO a sendRequest f n p =    withSocketsDo $ do -    installHandler sigPIPE Ignore Nothing+    _ <- installHandler sigPIPE Ignore Nothing          --TODO exception handling     --handle (\e -> do putStrLn $ show e return False) $-    bracket (connectTo n p) (hClose) (send)+    bracket (connectTo n (PortNumber p)) (hClose) (send)     where         send hdl        = do         hSetBuffering hdl NoBuffering-        f hdl+        f hdl  +-- no exception handling+performUnsafeSendRequest :: (Handle -> IO a) -> HostName -> PortNumber -> IO a+performUnsafeSendRequest = sendRequest +-- all IOExceptions handled return of default value+performSafeSendRequest :: (Handle -> IO a) -> a -> HostName -> PortNumber -> IO a+performSafeSendRequest f d n p+  = catch (sendRequest f n p)+          (\(e ::IOException) -> +            do+            debugM localLogger $  show e +            return d)++-- all IOExceptions handled return of Nothing+performMaybeSendRequest :: (Handle -> IO a) -> HostName -> PortNumber -> IO (Maybe a)+performMaybeSendRequest f n p+  = catch (do+           res <- sendRequest f n p+           return (Just res))+          (\(e ::IOException) -> +            do+            debugM localLogger $  show e +            return Nothing)++ -- ---------------------------------------------------------------------------- -- Handle-Operations -- ----------------------------------------------------------------------------@@ -256,6 +288,7 @@       errorM localLogger $ "putMessage: " ++ show e       errorM localLogger $ "message: " ++ show msg       ) $ do+      debugM "measure.putMessage" "1"       hPutStrLn hdl ((show $ B.length msg) ++ " ")       B.hPut hdl msg @@ -265,6 +298,7 @@ getMessage :: Handle -> IO (B.ByteString) getMessage hdl   = do+    debugM "measure.getMessage" "1"     line <- hGetLine hdl     let pkg = words line     raw <- B.hGet hdl (read $ head pkg)

@@ -0,0 +1,136 @@+-- ----------------------------------------------------------------------------++{- |+  Module     : Holumbus.Network.DoWithServer+  Copyright  : Copyright (C) 2009 Sebastian Reese+  License    : MIT++  Maintainer : Sebastian Reese (str@holumbus.org)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  This module provides an interface to start a server listeing on a tcp socket.+  ( Module is based on simple tcp server found here http://sequence.complete.org/node/258 )+  +  You have to implement 2 functions+       type ServerAction  a = a -> Client a -> [Client a] -> IO [Client a]+    +    and+    +      type LineConverter a = String -> a+      +   to start a server use the+     doWithServer :: (Show a) => Int -> ServerAction a -> LineConverter a -> String -> IO ()+   function.++-}++-- ----------------------------------------------------------------------------+module Holumbus.Network.DoWithServer+{-# DEPRECATED "this module will be remove in the next release, please use the packages from Holumbus.Distribution.*" #-}+(+    Client (..)+  , ServerAction+  , LineConverter+  , doWithServer+)+ where ++--import           Holumbus.Common.Logging+import System.Log.Logger+import System.IO+import Control.Concurrent+import Control.Concurrent.STM+import Control.Monad (liftM)+import Network+import Holumbus.Common.Utils  ( handleAll )++localLogger :: String+localLogger = "Holumbus.Network.DoWithServer"++------------------------------------------------------------------------------------------------------+-- Datatypes++-- | the connecting client type+data Client        a = Client (TChan a) Handle HostName PortNumber++instance Show (Client a) where+  show (Client _ _ host port) = "Client " ++ host ++ ":"++show port+  +instance Eq (Client a) where+  (==) (Client _ _ host1 port1) (Client _ _ host2 port2) = host1 == host2 && port1 == port2+  (/=) c1 c2 = not (c1 == c2)+  +instance Ord (Client a) where+  compare (Client _ _ host1 port1) (Client _ _ host2 port2) = compare (host1,port1) (host2,port2)++-- | type for call back function that is executed within server cycle+type ServerAction  a = a -> Client a -> [Client a] -> IO [Client a]++-- | converts the string read by hGetLine to your datatype+type LineConverter a = String -> a+------------------------------------------------------------------------------------------------------++-- | execute a ServerAction wrapped by the tcp server+doWithServer :: (Show a) => Int -> ServerAction a -> LineConverter a -> String -> IO ()+doWithServer port action converter prompt = +  withSocketsDo $ do+    servSock <- listenOn . PortNumber . fromIntegral $ port+    handleAll (\_ -> sClose servSock) $ do+      acceptChan <- atomically newTChan+      _ <- forkIO $ acceptLoop servSock acceptChan converter prompt+      mainLoop servSock acceptChan [] action prompt --`finally` sClose servSock+++-- | the accept loop for new connections+acceptLoop :: Socket -> TChan (Client a) -> LineConverter a -> String -> IO ()+acceptLoop servSock chan convert prompt = do+  (handle, host, port) <- accept servSock+  hPrompt handle prompt+  ch <- atomically newTChan+  _ <- forkIO $ clientLoop (Client ch handle host port) convert+  atomically $ writeTChan chan (Client ch handle host port)+  acceptLoop servSock chan convert prompt++-- | the listenerloops for getting the clients commands+clientLoop :: Client a -> LineConverter a -> IO ()+clientLoop client@(Client chan handle _ _) convert =+  handleAll (\_ -> do { infoM localLogger $ "Client disconnected: "++show client; hClose handle; return () }) $ do+    listenLoop (liftM convert $ hGetLine handle) chan++-- | the listenerloops for getting the clients commands+listenLoop :: IO a -> TChan a -> IO ()+listenLoop act chan =+  sequence_ (repeat (act >>= atomically . writeTChan chan)) ++-- | the main loop combines new client actions (Left) and new command actions (Right)+mainLoop :: (Show a) => Socket -> TChan (Client a) -> [(Client a)] -> ServerAction a -> String-> IO ()+mainLoop servSock acceptChan clients f prompt= do+  r <- atomically $ (Left `fmap` readTChan acceptChan)+                    `orElse`+                    (Right `fmap` tselect clients)+  case r of+    -- a new client+    Left sender -> do+      infoM localLogger $ "new client" ++ show sender+      mainLoop servSock acceptChan (sender:clients) f prompt+   +    -- a new command+    Right (line, sender) -> do+      debugM localLogger $ (show sender) ++ " [ " ++ show line ++ " ]"+      clients' <- f line sender clients+      debugM localLogger $ "Num of Clients: " ++ show (length clients')+      mainLoop servSock acceptChan (clients') f prompt+++hPrompt :: Handle -> String -> IO ()+hPrompt h p = do+  openhandle <- hIsOpen h+  if openhandle then do+    hPutStr h p+    hFlush h+    else return ()++tselect :: [(Client a)] -> STM (a, Client a)+tselect = foldl orElse retry . map (\client@(Client ch _ _ _) -> (\line -> (line,client)) `fmap` readTChan ch) 

@@ -26,6 +26,7 @@  {-# OPTIONS -fglasgow-exts #-} module Holumbus.Network.Messages+{-# DEPRECATED "this module will be remove in the next release, please use the packages from Holumbus.Distribution.*" #-} ( -- * Message-Class   RspMsg(..)@@ -49,6 +50,7 @@ -}  import           Data.Binary+--import           Holumbus.Common.MRBinary import           Data.Maybe import           Data.Typeable @@ -233,6 +235,7 @@         -- yield       else do         -- do the dispatching in a new process...+        infoM localLogger $ "forking dispatch with response port: "  ++ (show responsePort)         _ <- forkIO $ dispatcher dat $ (fromJust responsePort)         return () 

@@ -20,6 +20,7 @@ {-# OPTIONS_GHC -fno-warn-unused-binds #-}	-- for unused record field selectors  module Holumbus.Network.Port+{-# DEPRECATED "this module will be remove in the next release, please use the packages from Holumbus.Distribution.*" #-} ( -- * Constants   time1@@ -36,7 +37,7 @@ , StreamName , StreamType(..) , Stream-, Port+, Port(..)  -- * Message-Operations , getMessageType@@ -85,6 +86,7 @@ import           Control.Monad  import           Data.Binary+--import           Holumbus.Common.MRBinary import qualified Data.ByteString.Lazy as B import           Data.Char import qualified Data.Map as Map@@ -222,7 +224,7 @@ data Port a = Port { p_StreamName :: StreamName      -- ^ the name of the destination stream 		   , p_SocketId   :: Maybe SocketId 		   }-	      deriving (Show)+	      deriving (Show,Eq)  instance (Show a, Binary a) => Binary (Port a) where   put (Port sn soid)	= put sn >> put soid@@ -795,13 +797,13 @@   -- | Delegates new incomming messages on a unix-socket to their streams.-streamDispatcher :: SocketId -> Handle -> HostName -> PortNumber -> IO ()-streamDispatcher (SocketId _ ownPo) hdl hn po+streamDispatcher :: SocketId -> Handle -> SocketId -> IO ()+streamDispatcher (SocketId _ ownPo) hdl (SocketId hn po)   = do     debugM localLogger "streamDispatcher: getting message from handle"     raw <- getMessage hdl     let msg = (decode raw)::Message B.ByteString-    debugM localLogger $ "streamDispatcher: Message: " ++ show msg+    -- debugM localLogger $ "streamDispatcher: Message: " ++ show msg     let sn = getMessageReceiver msg     sns <- getStreamNamesForPort ownPo     sd <- getStreamData sn@@ -910,7 +912,7 @@             do             msg <- newMessage MTExternal sn d rp             let raw = encode $ encodeMessage $ updateReceiverSocket msg so-            sendRequest (putMessage raw) hn (PortNumber po)+            sendRequest (putMessage raw) hn po             return ()           -- we don't know it's port           (Nothing) ->@@ -921,7 +923,7 @@                 do                 msg <- newMessage MTExternal sn d rp                 let raw = encode $ encodeMessage $ updateReceiverSocket msg so-                sendRequest (putMessage raw) hn (PortNumber po)+                sendRequest (putMessage raw) hn po                 return ()               (Nothing) ->                 do

@@ -18,6 +18,7 @@  {-# OPTIONS -fglasgow-exts #-} module Holumbus.Network.PortRegistry+{-# DEPRECATED "this module will be remove in the next release, please use the packages from Holumbus.Distribution.*" #-}     (       -- * Type-Classes       PortRegistry(..)

@@ -16,6 +16,7 @@ -- ----------------------------------------------------------------------------  module Holumbus.Network.PortRegistry.Messages+{-# DEPRECATED "this module will be remove in the next release, please use the packages from Holumbus.Distribution.*" #-} (   PortRegistryRequestStream , PortRegistryRequestPort@@ -29,9 +30,9 @@ where  import           Data.Binary+--import           Holumbus.Common.MRBinary  import           Holumbus.Network.Port-import           Holumbus.Network.Core import           Holumbus.Network.Messages  

@@ -16,6 +16,7 @@ -- ----------------------------------------------------------------------------  module Holumbus.Network.PortRegistry.PortRegistryData+{-# DEPRECATED "this module will be remove in the next release, please use the packages from Holumbus.Distribution.*" #-} ( -- * Datatypes   PortRegistryData

@@ -16,6 +16,7 @@ -- ----------------------------------------------------------------------------  module Holumbus.Network.PortRegistry.PortRegistryPort+{-# DEPRECATED "this module will be remove in the next release, please use the packages from Holumbus.Distribution.*" #-} ( -- * Datatypes   PortRegistryPort

@@ -18,6 +18,7 @@ -- ----------------------------------------------------------------------------  module Holumbus.Network.Site+{-# DEPRECATED "this module will be remove in the next release, please use the packages from Holumbus.Distribution.*" #-} ( -- * Datatypes   SiteId@@ -42,6 +43,7 @@ where  import           Data.Binary+--import           Holumbus.Common.MRBinary import qualified Data.List as List import           Network.Socket import           System.Posix