diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,8 @@ +##### 1.1.1 + change common state from TVar s to s + + pass init common state + ##### 1.1 add custom state to Server diff --git a/raketka.cabal b/raketka.cabal --- a/raketka.cabal +++ b/raketka.cabal @@ -1,5 +1,5 @@ name: raketka -version: 1.1 +version: 1.1.1 build-type: Simple synopsis: basic distributed-process node with configurable peers description: start multiple nodes, let them communicate. diff --git a/src/Control/Distributed/Raketka/HandleMsg.hs b/src/Control/Distributed/Raketka/HandleMsg.hs --- a/src/Control/Distributed/Raketka/HandleMsg.hs +++ b/src/Control/Distributed/Raketka/HandleMsg.hs @@ -32,17 +32,16 @@ Tagged tag (Server ps s) -> WhereIsReply -> Process () handleWhereIsReply _ (P.WhereIsReply _ Nothing) = pure () handleWhereIsReply s1@(Tagged Server{..}) (WhereIsReply _ (Just pid0)) = - liftIO $ atomically $ - sendRemote s1 pid0 $ Info Ping spid + la $ sendRemote s1 pid0 $ Info Ping spid {- | 'ProcessMonitorNotification' e.g. connection lost -} handleMonitorNotification::Content tag ps s c => Tagged tag (Server ps s) -> ProcessMonitorNotification -> Process () handleMonitorNotification - s1@(Tagged Server{..}) (ProcessMonitorNotification _ pid0 _) = do - say (printf "server on %s dropped connection" pid0) - liftIO $ atomically $ do + s1@(Tagged Server{..}) (ProcessMonitorNotification _ pid0 reason0) = do + say (printf "server on %s dropped connection. reason: %s" pid0 (show reason0)) + la $ do old_pids1 <- readTVar servers writeTVar servers $ onPeerDisConnected' old_pids1 pid0 onPeerDisConnected s1 pid0 diff --git a/src/Control/Distributed/Raketka/Impl/Inst.hs b/src/Control/Distributed/Raketka/Impl/Inst.hs --- a/src/Control/Distributed/Raketka/Impl/Inst.hs +++ b/src/Control/Distributed/Raketka/Impl/Inst.hs @@ -22,8 +22,8 @@ trace "handleMessage" $ pure () -- todo - startServer::Tagged Slb ServerId -> Process () - startServer (Tagged id0) = server id0 + startServer::Tagged Slb ServerId -> () -> Process () + startServer (Tagged id0) _ = server id0 onPeerConnected::Tagged Slb Server_slb -> ProcessId -> Process () onPeerConnected (Tagged s0) pid0 = diff --git a/src/Control/Distributed/Raketka/Master.hs b/src/Control/Distributed/Raketka/Master.hs --- a/src/Control/Distributed/Raketka/Master.hs +++ b/src/Control/Distributed/Raketka/Master.hs @@ -15,9 +15,10 @@ master::Specific tag ps s c => Backend -> Cluster -- ^ server ids from config - -> Tagged tag Int -- ^ this server's idx in cluster + -> Tagged tag Int -- ^ this server's idx in cluster + -> s -- ^ init custom state -> Process () -master backend0 (Cluster ids0) idx1@(Tagged idx0) = do +master backend0 (Cluster ids0) idx1@(Tagged idx0) state0 = do mynode1 <- getSelfNode let peers1 = N.nodeId <$> ids0 @@ -26,8 +27,9 @@ peers2 = filter (/= mynode1) peers1 mypid1 <- getSelfPid + register service1 mypid1 forM_ peers2 $ \(peer1::NodeId) -> P.whereisRemoteAsync peer1 service1 - startServer $ passTag idx1 this1 + startServer (passTag idx1 this1) state0 diff --git a/src/Control/Distributed/Raketka/NewServerInfo.hs b/src/Control/Distributed/Raketka/NewServerInfo.hs --- a/src/Control/Distributed/Raketka/NewServerInfo.hs +++ b/src/Control/Distributed/Raketka/NewServerInfo.hs @@ -18,15 +18,16 @@ -> Process () newServerInfo s1@(Tagged server0@Server{..}) ping0 pid0 = do - liftIO $ printf "%s received %s from %s\n" spid (show ping0) pid0 - join $ liftIO $ atomically $ do - old_pids1 <- readTVar servers - let old_pids2 = peer_pids old_pids1 - writeTVar servers $ onPeerConnected' old_pids1 pid0 + say $ printf "%s received %s from %s\n" spid (show ping0) pid0 + old_pids1 <- la $ readTVar servers + let old_pids2 = peer_pids old_pids1 + la $ writeTVar servers $ onPeerConnected' old_pids1 pid0 - when (ping0 == Ping) $ sendRemote s1 pid0 $ Info Pong spid - - -- monitor the new server - pure (when (pid0 `notElem` old_pids2) $ void $ monitor pid0) + if (ping0 == Ping) then + la (sendRemote s1 pid0 $ Info Pong spid) + else pure () + + -- monitor the new server + when (pid0 `notElem` old_pids2) $ void $ monitor pid0 onPeerConnected s1 pid0 diff --git a/src/Control/Distributed/Raketka/Process/Server.hs b/src/Control/Distributed/Raketka/Process/Server.hs --- a/src/Control/Distributed/Raketka/Process/Server.hs +++ b/src/Control/Distributed/Raketka/Process/Server.hs @@ -21,9 +21,9 @@ P.spawnLocal (proxy s1) pid1 <- getSelfPid - liftIO $ atomically $ sendRemoteAll s1 $ Info Ping pid1 + la $ sendRemoteAll s1 $ Info Ping pid1 - forever $ + forever $ P.receiveWait [ P.match $ H.handleRemoteMessage s1 , P.match $ handleMonitorNotification s1 @@ -38,7 +38,7 @@ proxy::Content tag ps s c => Tagged tag (Server ps s) -> Process () proxy (Tagged Server{..}) = forever $ join $ - liftIO $ atomically $ readTChan proxychan + la $ readTChan proxychan -- | init 'Server' store @@ -48,12 +48,11 @@ pid1 <- getSelfPid liftIO $ do ps1 <- newTVarIO pids0 - s1 <- newTVarIO state0 c1 <- newTVarIO Map.empty o1 <- newTChanIO pure $ Tagged Server { servers = ps1, proxychan = o1, spid = pid1, - state = s1 + state = state0 } diff --git a/src/Control/Distributed/Raketka/Type/Server.hs b/src/Control/Distributed/Raketka/Type/Server.hs --- a/src/Control/Distributed/Raketka/Type/Server.hs +++ b/src/Control/Distributed/Raketka/Type/Server.hs @@ -25,7 +25,7 @@ __c__ is Message content type, implementation-specific -} class Specific tag ps s c | tag -> ps, tag -> s, tag -> c where - startServer::Tagged tag ServerId -> Process () + startServer::Tagged tag ServerId -> s -> Process () handleMessage::Tagged tag (Server ps s) -> c -> Process () onPeerConnected::Tagged tag (Server ps s) -> ProcessId -> Process () onPeerDisConnected::Tagged tag (Server ps s) -> ProcessId -> Process () @@ -57,6 +57,9 @@ { proxychan::TChan (Process ()) -- ^ pipeline for sending messages , servers::TVar ps -- ^ peer specific store , spid::ProcessId -- ^ this node's pid - , state::TVar s -- ^ this node's common store + , state::s -- ^ this node's common store } - + + +la::STM a -> Process a +la = liftIO . atomically diff --git a/src/Main.hs b/src/Main.hs --- a/src/Main.hs +++ b/src/Main.hs @@ -25,4 +25,5 @@ node1 <- newLocalNode backend1 Node.runProcess node1 (master backend1 c1 - (Tagged $ read idx0::Tagged Slb Int)) + (Tagged $ read idx0::Tagged Slb Int) + ())