raketka 1.1.3 → 1.2.0

raw patch · 11 files changed

+150/−102 lines, 11 filesdep +splitmixdep −taggeddep ~basesetup-changedPVP ok

version bump matches the API change (PVP)

Dependencies added: splitmix

Dependencies removed: tagged

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• Setup.hs +0/−2
• changelog.md +10/−1
• raketka.cabal +25/−5
• src/Control/Distributed/Raketka/HandleMsg.hs +25/−16
• src/Control/Distributed/Raketka/Impl/Inst.hs +15/−13
• src/Control/Distributed/Raketka/Master.hs +8/−7
• src/Control/Distributed/Raketka/NewServerInfo.hs +14/−8
• src/Control/Distributed/Raketka/Process/Send.hs +12/−10
• src/Control/Distributed/Raketka/Process/Server.hs +7/−8
• src/Control/Distributed/Raketka/Type/Server.hs +17/−21
• src/Main.hs +17/−11

@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain

@@ -1,5 +1,14 @@+#####  1.2.0+    remove dependency on tagged package++    error message in case of incorrect args++    add legacy_libc flag++    + .gitignore+ #####  1.1.3-    fix broken build. Reason: missing -main-is ghc-option  +    fix broken build. Reason: missing -main-is ghc-option  #####  1.1.2     fix broken build. Reason: API change in dependency network-transport-tcp

@@ -1,5 +1,5 @@ name:                raketka-version:             1.1.3+version:             1.2.0 build-type:          Simple synopsis:            distributed-process node description:         peer node with simplelocalnet backend @@ -17,6 +17,21 @@    location: https://github.com/ciez/raketka.git  +flag legacy_libc+   description: set this flag to True if you see this error message :+         cbits-unix/init.c:3:10: error:+         fatal error: sys/random.h: No such file or directory+              3 | #include <sys/random.h>+                |          ^~~~~~~~~++         how to set this flag to true:+            in cabal.project or cabal.project.local+               add this line (0 indent):+                   constraints: raketka +legacy_libc+   default: False+   manual: True++ library   exposed-modules:           Control.Distributed.Raketka.NodeId@@ -38,7 +53,6 @@                        stm,                        distributed-process,                        distributed-process-simplelocalnet,-                       tagged,                        random,                        async,                        network,@@ -89,7 +103,6 @@                        bytestring,                        containers,                         stm,-                       tagged,                        distributed-process,                        distributed-process-simplelocalnet,                        async,@@ -101,6 +114,9 @@                        conf-json,                        template-haskell +  if flag(legacy_libc)+      build-depends: splitmix <= 0.1.1.0+                            hs-source-dirs:      src   default-language:    Haskell2010@@ -142,6 +158,10 @@    main-is: TestMain.hs -  build-depends:  base >= 4.8,+  build-depends:  base >=4.8 && <5.0,                   hspec >= 2.1.7,-                  QuickCheck >= 2.8.1                      +                  QuickCheck >= 2.8.1++  if flag(legacy_libc)+      build-depends: splitmix <= 0.1.1.0+

@@ -1,6 +1,5 @@ module Control.Distributed.Raketka.HandleMsg where -import Data.Tagged import Text.Printf import Control.Distributed.Process as P hiding (Message, handleMessage) import Control.Concurrent.STM@@ -18,30 +17,40 @@          depending on message type -} -handleRemoteMessage::Content tag ps s c =>-    Tagged tag (Server ps s) -> Message c -> Process ()-handleRemoteMessage s1@(Tagged Server{..}) msg0 =  +handleRemoteMessage :: Content tag ps s c =>+    tag (Server ps s) -> Message c -> Process ()+handleRemoteMessage server0 msg0 =   case msg0 of-    Info ping1 pid1 -> newServerInfo s1 ping1 pid1 -    Message msg1 -> handleMessage s1 msg1+    Info ping1 pid1 -> newServerInfo server0 ping1 pid1+    Message msg1 -> handleMessage server0 msg1   {- | handles replies to 'whereisRemoteAsync', run on init  in "Control.Distributed.Raketka.Master" -}-handleWhereIsReply::Content tag ps s c =>-    Tagged tag (Server ps s) -> WhereIsReply -> Process () +handleWhereIsReply :: Content tag ps s c =>+    tag (Server ps s)+    -> WhereIsReply+    -> Process () handleWhereIsReply _ (P.WhereIsReply _ Nothing) = pure ()-handleWhereIsReply s1@(Tagged Server{..}) (WhereIsReply _ (Just pid0)) =-  la $ sendRemote s1 pid0 $ Info Ping spid+handleWhereIsReply+         server0+         (WhereIsReply _ (Just pid0)) =+  la $ sendRemote server0 pid0 $+         Info Ping $+         spid $ untag server0   {- | 'ProcessMonitorNotification' e.g. connection lost  -} -handleMonitorNotification::Content tag ps s c =>-    Tagged tag (Server ps s) -> ProcessMonitorNotification -> Process ()+handleMonitorNotification :: Content tag ps s c =>+    tag (Server ps s)+    -> ProcessMonitorNotification+    -> Process () handleMonitorNotification-       s1@(Tagged Server{..}) (ProcessMonitorNotification _ pid0 reason0) = do+       server0+       (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 +    old_pids1 <- readTVar servers1+    writeTVar servers1 $ onPeerDisConnected' old_pids1 pid0+  onPeerDisConnected server0 pid0+  where servers1 = servers $ untag server0

@@ -2,7 +2,6 @@  import Data.ByteString import Data.Set as S-import Data.Tagged import Control.Distributed.Process import Control.Distributed.Raketka.Type.Arg import Control.Distributed.Raketka.Type.Server@@ -11,28 +10,31 @@ import Text.Printf  -data Slb = Slb  -- ^ Stateless ByteString   +data Slb a = Slb a  -- ^ Stateless ByteString  type Server_slb = Server (Set ProcessId) ()   instance Specific Slb (Set ProcessId) () ByteString where-    handleMessage::Tagged Slb Server_slb -> ByteString -> Process ()-    handleMessage serv0 msg0 =+    handleMessage :: Slb Server_slb -> ByteString -> Process ()+    handleMessage (Slb serv0) msg0 =         trace "handleMessage" $          pure ()  --  todo  -    startServer::Tagged Slb ServerId -> () -> Process ()-    startServer (Tagged id0) _ = server id0 +    startServer :: Slb ServerId -> () -> Process ()+    startServer (Slb id0) _ = server id0 -    onPeerConnected::Tagged Slb Server_slb -> ProcessId -> Process ()-    onPeerConnected (Tagged s0) = tracePid "connected to %s"+    onPeerConnected :: Slb Server_slb -> ProcessId -> Process ()+    onPeerConnected (Slb s0) = tracePid "connected to %s" -    onPeerDisConnected::Tagged Slb Server_slb -> ProcessId -> Process ()-    onPeerDisConnected (Tagged s0) = tracePid "disconnected %s"+    onPeerDisConnected :: Slb Server_slb -> ProcessId -> Process ()+    onPeerDisConnected (Slb s0) = tracePid "disconnected %s" +    tag = Slb+    untag (Slb a) = a -tracePid::String    -- ^ formattable string with a placeholder for pid++tracePid :: String    -- ^ formattable string with a placeholder for pid         -> ProcessId         -> Process () tracePid msg0 pid0 =@@ -41,9 +43,9 @@         where msg1 = printf msg0 pid0  -server::ServerId -> Process ()+server :: ServerId -> Process () server id0 = newServer' >>= flip S.server id0  -newServer'::Process (Tagged Slb Server_slb)+newServer' :: Process (Slb Server_slb) newServer' = S.newServer () S.empty

@@ -1,6 +1,5 @@ module Control.Distributed.Raketka.Master where -import Data.Tagged import Control.Distributed.Process as P import Control.Monad import Control.Distributed.Process.Backend.SimpleLocalnet@@ -12,18 +11,20 @@     * calls 'whereisRemoteAsync' for each suggested peer     * calls 'startServer'   -}-master::Specific tag ps s c =>+master :: forall tag ps s c.+   Specific tag ps s c =>     Backend      -> Cluster          -- ^ server ids from config   -    -> Tagged tag Int   -- ^ this server's idx in cluster  +    -> tag Int   -- ^ this server's idx in cluster     -> s                -- ^ init custom state     -> Process ()-master backend0 (Cluster ids0) idx1@(Tagged idx0) state0 = do+master backend0 (Cluster ids0) idx0 state0 = do   mynode1 <- getSelfNode    let peers1 = N.nodeId <$> ids0-      this1 = ids0 !! idx0-      service1 = T.service this1+      thisServerId :: ServerId = ids0 !! (untag idx0)+      thisServerIdTagged :: tag ServerId = tag $ thisServerId+      service1 = T.service thisServerId       peers2 = filter (/= mynode1) peers1    mypid1 <- getSelfPid@@ -32,4 +33,4 @@    forM_ peers2 $ \(peer1::NodeId) -> P.whereisRemoteAsync peer1 service1 -  startServer (passTag idx1 this1) state0+  startServer thisServerIdTagged state0

@@ -1,6 +1,5 @@ module Control.Distributed.Raketka.NewServerInfo where -import Data.Tagged import Control.Distributed.Process   hiding (Message, mask, finally, handleMessage) import Control.Monad@@ -12,22 +11,29 @@  {- | 'Ping' and 'Pong' handler -} newServerInfo::Content tag ps s c =>-    Tagged tag (Server ps s) +    tag (Server ps s)     -> Ping      -> ProcessId      -> Process () -newServerInfo s1@(Tagged server0@Server{..}) ping0 pid0 = do-    say $ printf "%s received %s from %s\n" spid (show ping0) pid0-    old_pids1 <- la $ readTVar servers+newServerInfo server0 ping0 pid0 = do+    say $ printf "%s received %s from %s\n"+            (spid server1)+            (show ping0)+            pid0++    old_pids1 <- la $ readTVar $ servers1     let old_pids2 = peer_pids old_pids1-    la $ writeTVar servers $ onPeerConnected' old_pids1 pid0+    la $ writeTVar servers1 $ onPeerConnected' old_pids1 pid0      if (ping0 == Ping) then -        la (sendRemote s1 pid0 $ Info Pong spid)+        la $ sendRemote server0 pid0 $+               Info Pong $ spid server1         else pure ()      -- monitor the new server     when (pid0 `notElem` old_pids2) $ void $ monitor pid0 -    onPeerConnected s1 pid0+    onPeerConnected server0 pid0+    where server1 = untag server0+          servers1 = servers server1

@@ -2,7 +2,6 @@    (sendRemote,    sendRemoteAll)  where -import Data.Tagged import Control.Concurrent.STM import Control.Distributed.Raketka.Type.Server import Control.Distributed.Process@@ -14,14 +13,17 @@  Other ways of sending messages are not implemented to keep the code basic.   -}-sendRemote::Content tag ps s c =>-    Tagged tag (Server ps s) -> ProcessId -> Message c -> STM ()-sendRemote (Tagged Server{..}) pid pmsg =-    writeTChan proxychan (send pid pmsg)+sendRemote :: Content tag ps s c =>+    tag (Server ps s) -> ProcessId -> Message c -> STM ()+sendRemote taggedServer pid pmsg = writeTChan (proxychan server) (send pid pmsg)+   where server = untag taggedServer  {- | broadcast message to all known peers -}-sendRemoteAll::Content tag ps s c =>-    Tagged tag (Server ps s) -> Message c -> STM ()-sendRemoteAll s1@(Tagged server0@Server{..}) pmsg0 = do-    pids1 <- readTVar servers-    mapM_ (\pid1 -> sendRemote s1 pid1 pmsg0) $ peer_pids pids1+sendRemoteAll :: Content tag ps s c =>+    tag (Server ps s) -> Message c -> STM ()+sendRemoteAll taggedServer0 pmsg0 = do+    pids1 <- readTVar servers1+    mapM_+         (\pid1 -> sendRemote taggedServer0 pid1 pmsg0) $+         peer_pids pids1+    where servers1 = servers $ untag taggedServer0

@@ -1,6 +1,5 @@ module Control.Distributed.Raketka.Process.Server where -import Data.Tagged hiding (proxy) import Control.Concurrent.STM import Control.Distributed.Raketka.HandleMsg as H import Control.Distributed.Process as P hiding (proxy)@@ -16,8 +15,8 @@     * send out pings     * 'receiveWait's for messages       -} server::Content tag ps s c =>-    Tagged tag (Server ps s) -> ServerId -> Process ()-server s1@(Tagged server0) id0  = do+    tag (Server ps s) -> ServerId -> Process ()+server s1 id0  = do   P.spawnLocal (proxy s1)    pid1 <- getSelfPid@@ -36,21 +35,21 @@  -- | read & run /send processes/ from the pipeline  proxy::Content tag ps s c =>-    Tagged tag (Server ps s) -> Process ()-proxy (Tagged Server{..}) = forever $ join $-    la $ readTChan proxychan+    tag (Server ps s) -> Process ()+proxy taggedServer0 = forever $ join $+    la $ readTChan $ proxychan $ untag taggedServer0   -- | init 'Server' store newServer::Content tag ps s c =>-     s -> ps -> Process (Tagged tag (Server ps s))+     s -> ps -> Process (tag (Server ps s)) newServer state0 pids0 = do   pid1 <- getSelfPid   liftIO $ do     ps1 <- newTVarIO pids0     c1 <- newTVarIO Map.empty     o1 <- newTChanIO-    pure $ Tagged Server { +    pure $ tag Server {             servers = ps1,              proxychan = o1,              spid = pid1,

@@ -5,7 +5,6 @@ import Control.Distributed.Process.Serializable import Control.Distributed.Process hiding (Message) import Control.Distributed.Raketka.Type.Arg-import Data.Tagged import Data.Set  @@ -25,41 +24,38 @@     __c__ is Message content type, implementation-specific   -} class Specific tag ps s c | tag -> ps, tag -> s, tag -> c where-    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 ()+    startServer :: tag ServerId -> s -> Process ()+    handleMessage :: tag (Server ps s) -> c -> Process ()+    onPeerConnected :: tag (Server ps s) -> ProcessId -> Process ()+    onPeerDisConnected :: tag (Server ps s) -> ProcessId -> Process ()+    tag :: forall a. a -> tag a+    untag :: forall a. tag a -> a   class PeerInfo ps where-    onPeerConnected'::ps -> ProcessId -> ps-    onPeerDisConnected'::ps -> ProcessId -> ps-    peer_pids::ps -> [ProcessId]+    onPeerConnected' :: ps -> ProcessId -> ps+    onPeerDisConnected' :: ps -> ProcessId -> ps+    peer_pids :: ps -> [ProcessId]   instance PeerInfo (Set ProcessId) where-    onPeerConnected'::Set ProcessId -> ProcessId -> Set ProcessId+    onPeerConnected' :: Set ProcessId -> ProcessId -> Set ProcessId     onPeerConnected' s0 pid0 = insert pid0 s0      -    onPeerDisConnected'::Set ProcessId -> ProcessId -> Set ProcessId+    onPeerDisConnected' :: Set ProcessId -> ProcessId -> Set ProcessId     onPeerDisConnected' s0 pid0 = delete pid0 s0     -    peer_pids::Set ProcessId -> [ProcessId]+    peer_pids :: Set ProcessId -> [ProcessId]     peer_pids = toList         --- | pass tag between different types -passTag::Tagged a b -> c -> Tagged a c-passTag _ = Tagged-- data Server ps s = Server-      { proxychan::TChan (Process ())  -- ^ pipeline for sending messages -        , servers::TVar ps      -- ^ peer specific store -        , spid::ProcessId       -- ^ this node's pid-        , state::s              -- ^ this node's common store +      { proxychan :: TChan (Process ())  -- ^ pipeline for sending messages+        , servers :: TVar ps      -- ^ peer specific store+        , spid :: ProcessId       -- ^ this node's pid+        , state :: s              -- ^ this node's common store         }  -la::STM a -> Process a+la :: STM a -> Process a la = liftIO . atomically                

@@ -1,6 +1,6 @@ module Main where -import Data.Tagged+import Debug.Trace import Data.Conf.Json import System.Environment import Control.Distributed.Process.Closure@@ -9,28 +9,34 @@ import Control.Distributed.Raketka.Master import Control.Distributed.Raketka.Type.Arg import Control.Distributed.Raketka.Impl.Inst as I+import Control.Distributed.Raketka.Type.Server   remotable ['I.server]  -main::IO()+main :: IO() main = do-    [path0, idx0] <- getArgs -    Right c1 @ (Cluster conf1) <- readParse path0::IO (Either String Cluster)-    let idx1 = read idx0::Int-    idKnown c1 $ Tagged idx1+    args <- getArgs+    case args of+         [path0, idx0] -> do+             Right c1 @ (Cluster conf1) <- readParse path0::IO (Either String Cluster)+             let idx1 = read idx0 :: Int+             idKnown c1 $ Slb idx1+         _ -> do+                  traceIO "expect 2 args :"+                  traceIO "1) path to a config file. See ./test-conf.json for an example"+                  traceIO "2) 0 based node index"  -idKnown::Cluster -> Tagged Slb Int -> IO()-idKnown cl1@(Cluster cl0) idx1@(Tagged idx0) = do+idKnown :: Cluster -> Slb Int -> IO()+idKnown cl1@(Cluster cl0) idx0 = do     backend1 <- initializeBackend (host id1) thisport1                     (__remoteTable initRemoteTable)     node1 <- newLocalNode backend1     Node.runProcess node1              (master backend1 cl1 -                idx1+                idx0                 ())     where thisport1 = show $ port id1-          id1 = cl0 !! idx0  -                +          id1 = cl0 !! (untag idx0)