raketka 1.1 → 1.1.1

raw patch · 9 files changed

+39/−29 lines, 9 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• changelog.md +5/−0
• raketka.cabal +1/−1
• src/Control/Distributed/Raketka/HandleMsg.hs +4/−5
• src/Control/Distributed/Raketka/Impl/Inst.hs +2/−2
• src/Control/Distributed/Raketka/Master.hs +5/−3
• src/Control/Distributed/Raketka/NewServerInfo.hs +10/−9
• src/Control/Distributed/Raketka/Process/Server.hs +4/−5
• src/Control/Distributed/Raketka/Type/Server.hs +6/−3
• src/Main.hs +2/−1

@@ -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     

@@ -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. 

@@ -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 

@@ -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 =

@@ -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

@@ -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

@@ -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         }

@@ -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                

@@ -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)+                ())