packages feed

raketka 1.0 → 1.1

raw patch · 14 files changed

+176/−104 lines, 14 files

Files

README.md view
@@ -1,36 +1,43 @@ ### raketka -    basic distributed-process node+    note : package build may fail ++    due to version clash between aeson, vector, primitive packages     -    configurable peers+    https://github.com/ciez/vector is a patched version of vector-0.11.0.0 +++    * basic distributed-process node+    * configurable peers+     Begin reading at Control.Distributed.Raketka.Type.Arg -    see also +    see also:      -    [parconc-examples](https://hackage.haskell.org/package/parconc-examples)+        *   [parconc-examples](https://hackage.haskell.org/package/parconc-examples)+        *   [haskell-distributed tutorials](http://haskell-distributed.github.io/tutorials/1ch.html)       -    [haskell-distributed tutorials](http://haskell-distributed.github.io/tutorials/1ch.html)  +    This library:      -    This library has basic working code -    to enable to start nodes and to connect them with each other. +        functionality: +            * start nodes +            * connect them with each other+            * exchanges pings with other nodes which are expected to pong back+            * received pings & pongs are output to stdout+            * when 1 node is disconnected or stops, the other nodes stdout notifications about this     -    The library may be extended -    however you are most likely to write your own code that does much more. +        may be extended +        however you are most likely to write your own code that does much more.      -    This library is simple on purpose. -    It took me a few days to write this code and make it work.    +        is simple on purpose, does only a few things which seem to be common in distributed arch           To start the program that comes with this library:     -    start the same program in multiple consoles: 1 per node+        start the same program in multiple consoles: 1 per node      -    pass 2 args:-    -        1. path to config.json (see enclosed test-conf.json)-        2. idx of this node in the cluster: 0 - (length Cluster -1)-    -    The program exchanges pings with other nodes which are expected to pong back. -    Both pings & pongs are output to stdout+        2 args are expected:     -    When you shut one node, the other nodes stdout notifications about that node     +            1. path to config.json (see test-conf.json)+            2. idx of this node in the cluster: 0 .. (length Cluster -1)    +
changelog.md view
@@ -1,3 +1,19 @@+#####  1.1+    add custom state to Server+    +    peer state type is a configurable param+    +    note : package build may fail ++    due to version clash between aeson, vector, primitive packages+    +    https://github.com/ciez/vector is a patched version of vector-0.11.0.0+    +    I do not claim to have any knowledge about vector package workings. ++    Did basic refactoring: moved some code to a different module and followed some GHC suggestions. It builds and it works with raketka.  +    + #####  1.0     start multiple nodes
raketka.cabal view
@@ -1,5 +1,5 @@ name:                raketka-version:             1.0+version:             1.1 build-type:          Simple synopsis:            basic distributed-process node with configurable peers description:         start multiple nodes, let them communicate. @@ -72,8 +72,7 @@ executable raketka   main-is:             Main.hs   ghc-options:  -fwarn-unused-imports         -  other-modules:      Control.Distributed.Raketka.Impl.LByteString-                      Control.Distributed.Raketka.Impl.Inst+  other-modules:      Control.Distributed.Raketka.Impl.Inst    build-depends:       base >=4.8 && <5.0,                        binary,
src/Control/Distributed/Raketka/HandleMsg.hs view
@@ -18,18 +18,18 @@          depending on message type -} -handleRemoteMessage::Content c =>-    Tagged c Server -> Message c -> Process ()+handleRemoteMessage::Content tag ps s c =>+    Tagged tag (Server ps s) -> Message c -> Process () handleRemoteMessage s1@(Tagged Server{..}) msg0 =     case msg0 of     Info ping1 pid1 -> newServerInfo s1 ping1 pid1      Message msg1 -> handleMessage s1 msg1  -{- | replies to 'whereisRemoteAsync' run on init +{- | handles replies to 'whereisRemoteAsync', run on init  in "Control.Distributed.Raketka.Master" -}-handleWhereIsReply::Content c =>-    Tagged c Server -> WhereIsReply -> Process () +handleWhereIsReply::Content tag ps s c =>+    Tagged tag (Server ps s) -> WhereIsReply -> Process ()  handleWhereIsReply _ (P.WhereIsReply _ Nothing) = pure () handleWhereIsReply s1@(Tagged Server{..}) (WhereIsReply _ (Just pid0)) =   liftIO $ atomically $ @@ -37,11 +37,12 @@   {- | 'ProcessMonitorNotification' e.g. connection lost  -} -handleMonitorNotification::Content c =>-    Tagged c Server -> ProcessMonitorNotification -> Process ()+handleMonitorNotification::Content tag ps s c =>+    Tagged tag (Server ps s) -> ProcessMonitorNotification -> Process () handleMonitorNotification-       (Tagged Server{..}) (ProcessMonitorNotification _ pid0 _) = do-  say (printf "server on %s dropped connection" (show pid0))+       s1@(Tagged Server{..}) (ProcessMonitorNotification _ pid0 _) = do+  say (printf "server on %s dropped connection" pid0)   liftIO $ atomically $ do     old_pids1 <- readTVar servers-    writeTVar servers (filter (/= pid0) old_pids1)+    writeTVar servers $ onPeerDisConnected' old_pids1 pid0+  onPeerDisConnected s1 pid0
src/Control/Distributed/Raketka/Impl/Inst.hs view
@@ -1,32 +1,42 @@ module Control.Distributed.Raketka.Impl.Inst where -import Data.Data-import Data.Binary 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 import qualified Control.Distributed.Raketka.Process.Server as S import Debug.Trace+import Text.Printf  -newtype TByteString = TByteString ByteString deriving (Data, Typeable, Binary, Show)+data Slb = Slb  -- ^ Stateless ByteString    +type Server_slb = Server (Set ProcessId) () -instance Specific TByteString where-    handleMessage::Tagged TByteString Server -> TByteString -> Process ()++instance Specific Slb (Set ProcessId) () ByteString where+    handleMessage::Tagged Slb Server_slb -> ByteString -> Process ()     handleMessage serv0 msg0 =         trace "handleMessage" $          pure ()  --  todo  -    startServer::Tagged TByteString ServerId -> Process ()+    startServer::Tagged Slb ServerId -> Process ()     startServer (Tagged id0) = server id0 +    onPeerConnected::Tagged Slb Server_slb -> ProcessId -> Process ()+    onPeerConnected (Tagged s0) pid0 =+        say $ printf "connected to %s" pid0+    +    onPeerDisConnected::Tagged Slb Server_slb -> ProcessId -> Process ()+    onPeerDisConnected (Tagged s0) pid0 =+        say $ printf "disconnected %s" pid0 + server::ServerId -> Process () server id0 = newServer' >>= flip S.server id0  -newServer'::Process (Tagged TByteString Server)-newServer' = S.newServer []+newServer'::Process (Tagged Slb Server_slb)+newServer' = S.newServer () S.empty
− src/Control/Distributed/Raketka/Impl/LByteString.hs 
@@ -1,12 +0,0 @@-module Control.Distributed.Raketka.Impl.LByteString where--import qualified Data.ByteString as B (ByteString,concat)-import qualified Data.ByteString.Lazy as L (ByteString,toChunks,fromChunks)---toStrict::L.ByteString -> B.ByteString-toStrict = B.concat . L.toChunks---toLazy::B.ByteString -> L.ByteString-toLazy bs = L.fromChunks [bs]
src/Control/Distributed/Raketka/Master.hs view
@@ -9,11 +9,14 @@ import Control.Distributed.Raketka.NodeId as N  {-| * registers own pid-    * calls 'whereisRemoteAsync'+    * calls 'whereisRemoteAsync' for each suggested peer     * calls 'startServer'   -}-master::Specific c =>-    Backend -> Cluster -> Tagged c Int -> Process ()+master::Specific tag ps s c =>+    Backend +    -> Cluster          -- ^ server ids from config   +    -> Tagged tag Int     -- ^ this server's idx in cluster  +    -> Process () master backend0 (Cluster ids0) idx1@(Tagged idx0) = do   mynode1 <- getSelfNode
src/Control/Distributed/Raketka/NewServerInfo.hs view
@@ -11,19 +11,22 @@ import Control.Distributed.Raketka.Process.Send  {- | 'Ping' and 'Pong' handler -}-newServerInfo::Content c =>-    Tagged c Server +newServerInfo::Content tag ps s c =>+    Tagged tag (Server ps s)      -> Ping      -> ProcessId      -> Process ()  newServerInfo s1@(Tagged server0@Server{..}) ping0 pid0 = do-  liftIO $ printf "%s received %s from %s\n" (show spid) (show ping0) (show pid0)-  join $ liftIO $ atomically $ do-    old_pids1 <- readTVar servers-    writeTVar servers (pid0 : filter (/= pid0) old_pids1)+    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 -    when (ping0 == Ping) $ sendRemote s1 pid0 $ Info Pong spid +        when (ping0 == Ping) $ sendRemote s1 pid0 $ Info Pong spid +    +        -- monitor the new server+        pure (when (pid0 `notElem` old_pids2) $ void $ monitor pid0) -    -- monitor the new server-    return (when (pid0 `notElem` old_pids1) $ void $ monitor pid0)+    onPeerConnected s1 pid0
src/Control/Distributed/Raketka/Process/Send.hs view
@@ -14,14 +14,14 @@  Other ways of sending messages are not implemented to keep the code basic.   -}-sendRemote::Content c =>-    Tagged c Server -> ProcessId -> Message c -> STM ()+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)  {- | broadcast message to all known peers -}-sendRemoteAll::Content c =>-    Tagged c Server -> Message c -> STM ()+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) pids1+    mapM_ (\pid1 -> sendRemote s1 pid1 pmsg0) $ peer_pids pids1
src/Control/Distributed/Raketka/Process/Server.hs view
@@ -15,8 +15,8 @@ {- | * init 'proxy'     * send out pings     * 'receiveWait's for messages       -}-server::Content c =>-    Tagged c Server -> ServerId -> Process ()+server::Content tag ps s c =>+    Tagged tag (Server ps s) -> ServerId -> Process () server s1@(Tagged server0) id0  = do   P.spawnLocal (proxy s1) @@ -34,24 +34,26 @@       ]  --- | pipeline for sending msgs-proxy::Content c =>-    Tagged c Server -> Process ()+-- | read & run /send processes/ from the pipeline +proxy::Content tag ps s c =>+    Tagged tag (Server ps s) -> Process () proxy (Tagged Server{..}) = forever $ join $     liftIO $ atomically $ readTChan proxychan   -- | init 'Server' store-newServer::Content c =>-     [ProcessId] -> Process (Tagged c Server)-newServer pids0 = do+newServer::Content tag ps s c =>+     s -> ps -> Process (Tagged tag (Server ps s))+newServer state0 pids0 = do   pid1 <- getSelfPid   liftIO $ do-    s1 <- newTVarIO pids0+    ps1 <- newTVarIO pids0+    s1 <- newTVarIO state0     c1 <- newTVarIO Map.empty     o1 <- newTChanIO     pure $ Tagged Server { -            servers = s1, +            servers = ps1,              proxychan = o1, -            spid = pid1+            spid = pid1,+            state = s1         }
src/Control/Distributed/Raketka/Type/Arg.hs view
@@ -4,21 +4,28 @@  <http://haskell-distributed.github.io/tutorials/1ch.html>  -This library has basic working code to enable to start nodes and to connect them with each other. The library may be extended however you are most likely to write your own code that does much more. This library is simple on purpose. It took me a few days to write this code and make it work.    +This library: -To start the program that comes with this library:+    * has basic working code to enable to start nodes and to connect them with each other+    * may be extended however you are most likely to write your own code that does much more +    * is simple on purpose+    * exchanges pings with other nodes which are expected to pong back +    * both pings & pongs are output to stdout+    * when a node disconnects or stops, the other nodes stdout notifications about this -start the same program in multiple consoles: 1 per node+==== How to use the program that comes with this library:++    start the same program in multiple consoles: 1 per node  -pass 2 args:+    pass 2 args: -    1. path to config.json (see enclosed test-conf.json)-    2. idx of this node in the cluster: 0 - (length Cluster -1)+        1. path to config.json (see enclosed test-conf.json)+        2. idx of this node in the cluster: 0 .. (length Cluster -1)         -The program exchanges pings with other nodes which are expected to pong back. Both pings & pongs are output to stdout+ >>>  ./raketka ./test-conf.json 0     -When you shut one node, the other nodes stdout notifications about that node    --}+    stop (ctrl-C) any nodes, see notifications in live consoles     -}+     module Control.Distributed.Raketka.Type.Arg where  import GHC.Generics
src/Control/Distributed/Raketka/Type/Message.hs view
@@ -4,8 +4,9 @@ import GHC.Generics (Generic) import Data.Binary import Data.Typeable-+import Text.Printf +-- | content type is implementation-specific data Message content = Info Ping ProcessId  -- ^ message sent when nodes join cluster                         | Message content    -- ^ other messages                         deriving (Typeable, Generic, Show)@@ -14,3 +15,9 @@  instance Binary content => Binary (Message content) instance Binary Ping+++instance PrintfArg ProcessId  where+    formatArg pid0 = \(f1::FieldFormat) ->+                     (show pid0 ++)         +
src/Control/Distributed/Raketka/Type/Server.hs view
@@ -6,28 +6,57 @@ import Control.Distributed.Process hiding (Message) import Control.Distributed.Raketka.Type.Arg import Data.Tagged+import Data.Set --- | constraint type-type Content c = (Specific c, Serializable c, Show c) -{- | see example implementation in "Control.Distributed.Raketka.Impl.Inst"+{- | constraint type -    this file is part of the package, is not displayed in docs because it is part of a program, not the library.+    __c__ is Message content type, implementation-specific      -}+type Content tag ps s c = (PeerInfo ps, Specific tag ps s c, Serializable c, Show c)++{- | methods in this instance are called in library, defined in the program (this or another program that consumes this library)     -    see also Main.hs there is important code there +    see example implementation in "Control.Distributed.Raketka.Impl.Inst"++    "Control.Distributed.Raketka.Impl.Inst" is part of the package, is not displayed in docs because it is part of a program, not the library.+    +    see also Main.hs there is important code there+    +    __c__ is Message content type, implementation-specific   -}-class Specific c where-    handleMessage::Tagged c Server -> c -> Process ()-    startServer::Tagged c ServerId -> Process ()+class Specific tag ps s c | tag -> ps, tag -> s, tag -> c where+    startServer::Tagged tag ServerId -> 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 ()  +class PeerInfo ps where+    onPeerConnected'::ps -> ProcessId -> ps+    onPeerDisConnected'::ps -> ProcessId -> ps+    peer_pids::ps -> [ProcessId]+++instance PeerInfo (Set ProcessId) where+    onPeerConnected'::Set ProcessId -> ProcessId -> Set ProcessId+    onPeerConnected' s0 pid0 = insert pid0 s0 +    +    onPeerDisConnected'::Set ProcessId -> ProcessId -> Set ProcessId+    onPeerDisConnected' s0 pid0 = delete pid0 s0+    +    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 = Server+data Server ps s = Server       { proxychan::TChan (Process ())  -- ^ pipeline for sending messages -        , servers::TVar [ProcessId]    -- ^ to broadcast to entire cluster -        , spid::ProcessId              -- ^ this node's pid   +        , servers::TVar ps      -- ^ peer specific store +        , spid::ProcessId       -- ^ this node's pid+        , state::TVar s         -- ^ this node's common store          }
src/Main.hs view
@@ -11,7 +11,7 @@ import Control.Distributed.Raketka.Impl.Inst as I  -remotable ['server]+remotable ['I.server]   main::IO()@@ -25,4 +25,4 @@     node1 <- newLocalNode backend1     Node.runProcess node1              (master backend1 c1 -                (Tagged $ read idx0::Tagged TByteString Int))+                (Tagged $ read idx0::Tagged Slb Int))