legion 0.1.0.1 → 0.2.0.0
raw patch · 16 files changed
+1063/−905 lines, 16 filesdep +aesondep +canteven-log
Dependencies added: aeson, canteven-log
Files
- legion.cabal +5/−2
- src/Network/Legion.hs +4/−1
- src/Network/Legion/Admin.hs +19/−2
- src/Network/Legion/Application.hs +11/−3
- src/Network/Legion/BSockAddr.hs +1/−4
- src/Network/Legion/ClusterState.hs +11/−1
- src/Network/Legion/ConnectionManager.hs +0/−208
- src/Network/Legion/Distribution.hs +7/−32
- src/Network/Legion/PartitionKey.hs +3/−9
- src/Network/Legion/PartitionState.hs +13/−1
- src/Network/Legion/PowerState.hs +17/−0
- src/Network/Legion/Propagation.hs +24/−0
- src/Network/Legion/Runtime.hs +324/−56
- src/Network/Legion/Runtime/ConnectionManager.hs +208/−0
- src/Network/Legion/Runtime/PeerMessage.hs +79/−0
- src/Network/Legion/StateMachine.hs +337/−586
legion.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: legion-version: 0.1.0.1+version: 0.2.0.0 synopsis: Distributed, stateful, homogeneous microservice framework. description: Legion is a framework for writing distributed, homogeneous, stateful microservices in Haskell.@@ -31,7 +31,6 @@ Network.Legion.Basics Network.Legion.ClusterState Network.Legion.Conduit- Network.Legion.ConnectionManager Network.Legion.Distribution Network.Legion.Fork Network.Legion.KeySet@@ -41,6 +40,8 @@ Network.Legion.PowerState Network.Legion.Propagation Network.Legion.Runtime+ Network.Legion.Runtime.ConnectionManager+ Network.Legion.Runtime.PeerMessage Network.Legion.Settings Network.Legion.StateMachine Network.Legion.UUID@@ -48,12 +49,14 @@ -- other-extensions: build-depends: Ranged-sets >= 0.3.0 && < 0.4,+ aeson >= 0.11.2.0 && < 0.12, attoparsec >= 0.13.0.1 && < 0.14, base >= 4.8 && < 4.9, binary >= 0.7.5 && < 0.9, binary-conduit >= 1.2.3 && < 1.3, bytestring >= 0.10.4.0 && < 0.11, canteven-http >= 0.1.1.1 && < 0.2,+ canteven-log >= 1.0.0.0 && < 1.1, conduit >= 1.2.4 && < 1.3, conduit-extra >= 1.1.9 && < 1.2, containers >= 0.5.5.1 && < 0.6,
src/Network/Legion.hs view
@@ -161,7 +161,10 @@ -------------------------------------------------------------------------------- -- $invocation--- Notes on invocation.+-- While this section is being worked on, you can check out the+-- [legion-cache](https://github.com/taphu/legion-cache) project for a+-- working example of how to build a basic distributed key-value store+-- using Legion. --------------------------------------------------------------------------------
src/Network/Legion/Admin.hs view
@@ -7,6 +7,7 @@ -} module Network.Legion.Admin ( runAdmin,+ AdminMessage(..), ) where import Canteven.HTTP (requestLogging, logExceptionsAndContinue)@@ -23,10 +24,11 @@ import Network.HTTP.Types (notFound404) import Network.Legion.Application (LegionConstraints) import Network.Legion.Conduit (chanToSource)+import Network.Legion.Distribution (Peer) import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey(K))-import Network.Legion.StateMachine (AdminMessage(GetState, GetPart,- Eject))+import Network.Legion.PartitionState (PartitionPowerState)+import Network.Legion.StateMachine (NodeState) import Network.Wai (Middleware, modifyResponse) import Network.Wai.Handler.Warp (HostPreference, defaultSettings, Port, setHost, setPort)@@ -120,4 +122,19 @@ -} serverValue = encodeUtf8 (T.pack ("legion-admin/" ++ showVersion version))+++{- |+ The type of messages sent by the admin service.+-}+data AdminMessage i o s+ = GetState (NodeState i s -> LIO ())+ | GetPart PartitionKey (Maybe (PartitionPowerState i s) -> LIO ())+ | Eject Peer (() -> LIO ())++instance Show (AdminMessage i o s) where+ show (GetState _) = "(GetState _)"+ show (GetPart k _) = "(GetPart " ++ show k ++ " _)"+ show (Eject p _) = "(Eject " ++ show p ++ " _)"+
src/Network/Legion/Application.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ConstraintKinds #-} {- | This module contains the data types necessary for implementing the user application.@@ -26,15 +26,23 @@ > Show o, Show s, Eq i > ) -}-class (+type LegionConstraints i o s = ( ApplyDelta i s, Default s, Binary i, Binary o, Binary s, Show i, Show o, Show s, Eq i- ) => LegionConstraints i o s where+ ) {- | This is the type of a user-defined Legion application. Implement this and allow the Legion framework to manage your cluster.++ - @__i__@ is the type of request your application will handle. @__i__@ stands+ for __"input"__.+ - @__o__@ is the type of response produced by your application. @__o__@ stands+ for __"output"__+ - @__s__@ is the type of state maintained by your application. More+ precisely, it is the type of the individual partitions that make up+ your global application state. @__s__@ stands for __"state"__. -} data Legionary i o s = Legionary { {- |
src/Network/Legion/BSockAddr.hs view
@@ -11,9 +11,7 @@ SockAddrCan)) -{- |- A type useful only for creating a `Binary` instance of `SockAddr`.--}+{- | A type useful only for creating a `Binary` instance of `SockAddr`. -} newtype BSockAddr = BSockAddr {getAddr :: SockAddr} deriving (Show, Eq) instance Binary BSockAddr where@@ -48,6 +46,5 @@ $ "Can't decode BSockAddr because the constructor tag " ++ "was not understood. Probably this data is representing " ++ "something else."-
src/Network/Legion/ClusterState.hs view
@@ -2,6 +2,7 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-} {- | This module contains the data types related to the distributed cluster state. -}@@ -24,6 +25,7 @@ heartbeat, ) where +import Data.Aeson (ToJSON, toJSON, object, (.=)) import Data.Binary (Binary) import Data.Default.Class (Default(def)) import Data.Map (Map)@@ -59,6 +61,14 @@ distribution = D.empty, peers = Map.empty }+instance ToJSON ClusterState where+ toJSON ClusterState {distribution, peers} = object [+ "distribution" .= distribution,+ "peers" .= Map.fromList [+ (show p, show a)+ | (p, a) <- Map.toList peers+ ]+ ] {- |@@ -75,7 +85,7 @@ -} newtype ClusterPropState = ClusterPropState { unPropState :: PropState UUID ClusterState Peer Update- } deriving (Show)+ } deriving (Show, ToJSON) {- |
− src/Network/Legion/ConnectionManager.hs
@@ -1,208 +0,0 @@-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE TemplateHaskell #-}-{- |- This module manages connections to other nodes in the cluster.--}-module Network.Legion.ConnectionManager (- ConnectionManager,- newConnectionManager,- send,- newPeers,-) where--import Prelude hiding (lookup)--import Control.Concurrent (Chan, writeChan, newChan, readChan)-import Control.Exception (try, SomeException)-import Control.Monad (void)-import Control.Monad.Logger (logInfo, logWarn)-import Control.Monad.Trans.Class (lift)-import Data.Binary (Binary, encode)-import Data.ByteString.Lazy (ByteString)-import Data.Map (toList, insert, empty, Map, lookup)-import Data.Text (pack)-import Network.Legion.BSockAddr (BSockAddr(BSockAddr))-import Network.Legion.Distribution (Peer)-import Network.Legion.Fork (forkC)-import Network.Legion.LIO (LIO)-import Network.Legion.StateMachine (PeerMessage)-import Network.Socket (SockAddr, Socket, socket, SocketType(Stream),- defaultProtocol, connect, close, SockAddr(SockAddrInet, SockAddrInet6,- SockAddrUnix, SockAddrCan), Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN))-import Network.Socket.ByteString.Lazy (sendAll)--{- |- A handle on the connection manager--}-data ConnectionManager i o s = C (Chan (Message i o s))-instance Show (ConnectionManager i o s) where- show _ = "ConnectionManager"---{- |- Create a new connection manager.--}-newConnectionManager :: (Binary i, Binary o, Binary s)- => Map Peer BSockAddr- -> LIO (ConnectionManager i o s)-newConnectionManager initPeers = do- chan <- lift newChan- forkC "connection manager thread" $- manager chan S {connections = empty}- let cm = C chan- newPeers cm initPeers- return cm- where- manager :: (Binary s, Binary o, Binary i)- => Chan (Message i o s)- -> State i o s- -> LIO ()- manager chan state = lift (readChan chan) >>= handle state >>= manager chan-- handle :: (Binary i, Binary o, Binary s)- => State i o s- -> Message i o s- -> LIO (State i o s)- handle s@S {connections} (NewPeer peer addr) =- case lookup peer connections of- Nothing -> do- conn <- connection addr- return s {- connections = insert peer conn connections- }- Just _ ->- return s-- handle s@S {connections} (Send peer msg) = do- case lookup peer connections of- Nothing -> $(logWarn) . pack $ "unknown peer: " ++ show peer- Just conn -> lift $ writeChan conn msg- return s---{- |- Build a new connection.--}-connection :: (Binary i, Binary o, Binary s)- => SockAddr- -> LIO (Chan (PeerMessage i o s))--connection addr = do- chan <- lift newChan- forkC ("connection to: " ++ show addr) $- handle chan Nothing- return chan- where- handle :: (Binary i, Binary o, Binary s)- => Chan (PeerMessage i o s)- -> Maybe Socket- -> LIO ()- handle chan so =- lift (readChan chan) >>= sendWithRetry so . encode >>= handle chan-- {- |- Open a socket.- -}- openSocket :: IO Socket- openSocket = do- so <- socket (fam addr) Stream defaultProtocol- connect so addr- return so-- {- |- Try to send the payload over the socket, and if that fails, then try to- create a new socket and retry sending the payload. Return whatever the- "working" socket is.- -}- sendWithRetry :: Maybe Socket -> ByteString -> LIO (Maybe Socket)- sendWithRetry Nothing payload =- (lift . try) openSocket >>= \case- Left err -> do- $(logWarn) . pack- $ "Can't connect to: " ++ show addr ++ ". Dropping message on "- ++ "the floor: " ++ show payload ++ ". The error was: "- ++ show (err :: SomeException)- return Nothing- Right so -> do- result2 <- (lift . try) (sendAll so payload)- case result2 of- Left err -> $(logWarn) . pack- $ "An error happend when trying to send a payload over a socket "- ++ "to the address: " ++ show addr ++ ". The error was: "- ++ show (err :: SomeException) ++ ". This is the last straw, we "- ++ "are not retrying. The message is being dropped on the floor. "- ++ "The message was: " ++ show payload- Right _ -> return ()- return (Just so)- sendWithRetry (Just so) payload =- (lift . try) (sendAll so payload) >>= \case- Left err -> do- $(logInfo) . pack- $ "Socket to " ++ show addr ++ " died. Retrying on a new "- ++ "socket. The error was: " ++ show (err :: SomeException)- (lift . void) (try (close so) :: IO (Either SomeException ()))- sendWithRetry Nothing payload- Right _ ->- return (Just so)---{- |- Send a message to a peer.--}-send- :: ConnectionManager i o s- -> Peer- -> PeerMessage i o s- -> LIO ()-send (C chan) peer = lift . writeChan chan . Send peer---{- |- Tell the connection manager about a new peer.--}-newPeer- :: ConnectionManager i o s- -> Peer- -> SockAddr- -> LIO ()-newPeer (C chan) peer addr = lift $ writeChan chan (NewPeer peer addr)---{- |- Tell the connection manager about all the peers known to the cluster state.--}-newPeers :: ConnectionManager i o s -> Map Peer BSockAddr -> LIO ()-newPeers cm peers =- mapM_ oneNewPeer (toList peers)- where- oneNewPeer (peer, BSockAddr addy) = newPeer cm peer addy---{- |- The internal state of the connection manager.--}-data State i o s = S {- connections :: Map Peer (Chan (PeerMessage i o s))- }---{- |- The types of messages that the ConnectionManager understands.--}-data Message i o s- = NewPeer Peer SockAddr- | Send Peer (PeerMessage i o s)---{- |- Guess the family of a `SockAddr`.--}-fam :: SockAddr -> Family-fam SockAddrInet {} = AF_INET-fam SockAddrInet6 {} = AF_INET6-fam SockAddrUnix {} = AF_UNIX-fam SockAddrCan {} = AF_CAN--
src/Network/Legion/Distribution.hs view
@@ -17,10 +17,12 @@ import Prelude hiding (null) +import Data.Aeson (ToJSON, toJSON, object, (.=)) import Data.Binary (Binary) import Data.Function (on) import Data.List (sort, sortBy) import Data.Set (Set, toList)+import Data.Text (pack) import Data.UUID (UUID) import GHC.Generics (Generic) import Network.Legion.KeySet (KeySet, member, (\\), null)@@ -47,6 +49,11 @@ newtype ParticipationDefaults = D { unD :: [(KeySet, Set Peer)] } deriving (Show, Binary)+instance ToJSON ParticipationDefaults where+ toJSON (D dist) = object [+ pack (show ks) .= Set.map show peers+ | (ks, peers) <- dist+ ] {- |@@ -134,38 +141,6 @@ weightOf p = sum [KS.size ks | (ks, ps) <- dist, p `Set.member` ps] --- -- TODO: first figure out if any replicas need re-building.--- case sortBy (weight `on` snd) (toList dist) of--- (p, keyspace):remaining | p == peer ->--- case reverse remaining of--- [] -> Nothing--- (target, targetSpace):_ ->--- let--- {- |--- Keys that already exist at the target are not eligible--- for being moved.--- -}--- eligibleSpace = keyspace \\ targetSpace--- migrationSize = (size keyspace - size targetSpace) `div` 2--- migrants = pickMigrants migrationSize eligibleSpace--- in--- case migrants of--- Just keys -> Just (Move target keys)--- Nothing -> Nothing--- _ -> Nothing--- where--- weight--- :: KeySet--- -> KeySet--- -> Ordering--- weight = flip compare `on` size--- --- pickMigrants :: Integer -> KeySet -> Maybe KeySet--- pickMigrants n keyspace =--- let migrants = take n keyspace in--- if size migrants > 0--- then Just migrants--- else Nothing {- |
src/Network/Legion/PartitionKey.hs view
@@ -19,9 +19,7 @@ import Data.Word (Word64) -{- |- This is how partitions are identified and referenced.--}+{- | This is how partitions are identified and referenced. -} newtype PartitionKey = K {unkey :: Word256} deriving (Eq, Ord, Show, Bounded) instance Binary PartitionKey where@@ -35,9 +33,7 @@ adjacentBelow (K k) = if k == minBound then Nothing else Just (K (pred k)) -{- |- Convert a `PartitionKey` into a hex string--}+{- | Convert a `PartitionKey` into a hex string. -} toHex :: PartitionKey -> String toHex (K (Word256 (Word128 a b) (Word128 c d))) = concatMap toHex64 [a, b, c, d]@@ -76,9 +72,7 @@ (True, True, True, True) -> 'f' -{- |- Maybe convert a hex string into a partition key--}+{- | Maybe convert a hex string into a partition key -} fromHex :: String -> Either String PartitionKey fromHex str | length str > 64 =
src/Network/Legion/PartitionState.hs view
@@ -19,8 +19,10 @@ projParticipants, projected, infimum,+ complete, ) where +import Data.Aeson (ToJSON) import Data.Binary (Binary) import Data.Default.Class (Default) import Data.Set (Set)@@ -52,7 +54,7 @@ -} newtype PartitionPropState i s = PartitionPropState { unPropState :: PropState PartitionKey s Peer i- } deriving (Eq, Show)+ } deriving (Eq, Show, ToJSON) -- {- |@@ -178,5 +180,15 @@ -} infimum :: PartitionPowerState i s -> s infimum = P.infimum . unPowerState+++{- |+ Figure out if this propagation state has any work to do. Return 'True' if all+ known propagation work has been completed. The implication here is that the+ only way more work can happen is if new deltas are applied, either directly+ or via a merge.+-}+complete :: PartitionPropState i s -> Bool+complete = P.complete . unPropState
src/Network/Legion/PowerState.hs view
@@ -1,6 +1,7 @@ {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-} {- | This module contains the fundamental distributed data object. -}@@ -28,6 +29,7 @@ import Prelude hiding (null) +import Data.Aeson (ToJSON, toJSON, object, (.=)) import Data.Binary (Binary(put, get)) import Data.Default.Class (Default(def)) import Data.DoubleWord (Word256(Word256), Word128(Word128))@@ -50,6 +52,15 @@ deltas :: Map (StateId p) (Delta p d, Set p) } deriving (Generic, Show, Eq) instance (Binary o, Binary s, Binary p, Binary d) => Binary (PowerState o s p d)+instance (Show o, Show s, Show p, Show d) => ToJSON (PowerState o s p d) where+ toJSON PowerState {origin, infimum, deltas} = object [+ "origin" .= show origin,+ "infimum" .= infimum,+ "deltas" .= Map.fromList [+ (show sid, (show d, Set.map show ps))+ | (sid, (d, ps)) <- Map.toList deltas+ ]+ ] {- |@@ -66,6 +77,12 @@ Infimum s1 _ _ == Infimum s2 _ _ = s1 == s2 instance (Ord p) => Ord (Infimum s p) where compare (Infimum s1 _ _) (Infimum s2 _ _) = compare s1 s2+instance (Show s, Show p) => ToJSON (Infimum s p) where+ toJSON Infimum {stateId, participants, stateValue} = object [+ "stateId" .= show stateId,+ "participants" .= Set.map show participants,+ "stateValue" .= show stateValue+ ] {- |
src/Network/Legion/Propagation.hs view
@@ -1,6 +1,7 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-} {- | This module defines how to propagate a PowerState amoung its participants. -}@@ -26,10 +27,12 @@ projParticipants, projected, infimum,+ complete, ) where import Prelude hiding (lookup) +import Data.Aeson (ToJSON, object, (.=), toJSON) import Data.Binary (Binary) import Data.Default.Class (Default) import Data.Map (Map, lookup)@@ -68,6 +71,16 @@ self :: p, now :: Time } deriving (Eq, Show)+instance (Show o, Show s, Show p, Show d) => ToJSON (PropState o s p d) where+ toJSON PropState {powerState, peerStates, self, now} = object [+ "powerState" .= powerState,+ "peerStates" .= Map.fromList [+ (show p, show s)+ | (p, s) <- Map.toList peerStates+ ],+ "self" .= show self,+ "now" .= show now+ ] {- |@@ -364,5 +377,16 @@ -} infimum :: PropPowerState o s p d -> s infimum = PS.infimumValue . unPowerState+++{- |+ Figure out if this propagation state has any work to do. Return 'True' if all+ known propagation work has been completed. The implication here is that the+ only way more work can happen is if new deltas are applied, either directly+ or via a merge.+-}+complete :: (Ord p) => PropState o s p d -> Bool+complete PropState {powerState, peerStates} =+ Map.null peerStates && Set.null (divergent powerState)
src/Network/Legion/Runtime.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-}@@ -17,45 +18,52 @@ import Control.Concurrent (forkIO) import Control.Concurrent.Chan (writeChan, newChan, Chan) import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)-import Control.Monad (void, forever, join)+import Control.Monad (void, forever, join, (>=>)) import Control.Monad.Catch (catchAll, try, SomeException, throwM)-import Control.Monad.IO.Class (liftIO)+import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Logger (logWarn, logError, logInfo, LoggingT,- MonadLoggerIO, runLoggingT, askLoggerIO)+ MonadLoggerIO, runLoggingT, askLoggerIO, logDebug) import Control.Monad.Trans.Class (lift)-import Data.Binary (encode)+import Data.Binary (encode, Binary) import Data.Conduit (Source, ($$), (=$=), yield, await, awaitForever,- transPipe, ConduitM, runConduit)+ transPipe, ConduitM, runConduit, Sink) import Data.Conduit.Network (sourceSocket) import Data.Conduit.Serialization.Binary (conduitDecode) import Data.Map (Map) import Data.Text (pack)-import Network.Legion.Admin (runAdmin)-import Network.Legion.Application (LegionConstraints, Legionary,- RequestMsg)+import GHC.Generics (Generic)+import Network.Legion.Admin (runAdmin, AdminMessage(GetState, GetPart,+ Eject))+import Network.Legion.Application (LegionConstraints,+ Legionary(Legionary), RequestMsg, persistence, getState) import Network.Legion.BSockAddr (BSockAddr(BSockAddr)) import Network.Legion.ClusterState (ClusterPowerState) import Network.Legion.Conduit (merge, chanToSink, chanToSource)-import Network.Legion.ConnectionManager (newConnectionManager, send,- newPeers) import Network.Legion.Distribution (Peer, newPeer) import Network.Legion.Fork (forkC) import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey)+import Network.Legion.Runtime.ConnectionManager (newConnectionManager,+ send, ConnectionManager, newPeers)+import Network.Legion.Runtime.PeerMessage (PeerMessage(PeerMessage),+ PeerMessagePayload(ForwardRequest, ForwardResponse, ClusterMerge,+ PartitionMerge), MessageId, newSequence, next) import Network.Legion.Settings (LegionarySettings(LegionarySettings, adminHost, adminPort, peerBindAddr, joinBindAddr))-import Network.Legion.StateMachine (stateMachine, LInput(J, P, R,- A), JoinRequest(JoinRequest), JoinResponse(JoinOk, JoinRejected),- LOutput(Send, NewPeers), AdminMessage, NodeState, PeerMessage,- newNodeState)+import Network.Legion.StateMachine (partitionMerge, clusterMerge,+ NodeState, newNodeState, runSM, UserResponse(Forward, Respond),+ userRequest, heartbeat, rebalance, migrate, propagate, ClusterAction,+ eject) import Network.Legion.UUID (getUUID) import Network.Socket (Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN),- SocketOption(ReuseAddr), SocketType(Stream), accept, bindSocket,+ SocketOption(ReuseAddr), SocketType(Stream), accept, bind, defaultProtocol, listen, setSocketOption, socket, SockAddr(SockAddrInet, SockAddrInet6, SockAddrUnix, SockAddrCan), connect, getPeerName, Socket) import Network.Socket.ByteString.Lazy (sendAll) import qualified Data.Conduit.List as CL+import qualified Data.Map as Map import qualified Network.Legion.ClusterState as C+import qualified Network.Legion.StateMachine as SM {- |@@ -70,17 +78,17 @@ -} runLegionary :: (LegionConstraints i o s) => Legionary i o s- -- ^ The user-defined legion application to run.+ {- ^ The user-defined legion application to run. -} -> LegionarySettings- -- ^ Settings and configuration of the legionary framework.+ {- ^ Settings and configuration of the legionary framework. -} -> StartupMode -> Source IO (RequestMsg i o)- -- ^ A source of requests, together with a way to respond to the requets.+ {- ^ A source of requests, together with a way to respond to the requets. -}+ -> LoggingT IO () {-- We don't use `LIO` in the type signature here because we don't- export the `LIO` symbol.+ Don't expose 'LIO' here because 'LIO' is a strictly internal+ symbol. 'LoggingT IO' is what we expose to the world. -}- -> LoggingT IO () runLegionary legionary@@ -88,31 +96,36 @@ startupMode requestSource = do+ {- Start the various messages sources. -} peerS <- loggingC =<< startPeerListener settings- (nodeState, peers) <- makeNodeState settings startupMode- cm <- newConnectionManager peers- $(logInfo) . pack- $ "The initial node state is: " ++ show nodeState adminS <- loggingC =<< runAdmin adminPort adminHost joinS <- loggingC (joinMsgSource settings)++ (self, nodeState, peers) <- makeNodeState settings startupMode+ cm <- newConnectionManager peers++ firstMessageId <- newSequence+ let+ rts = RuntimeState {+ forwarded = Map.empty,+ nextId = firstMessageId,+ cm,+ self+ } runConduit $ (joinS `merge` (peerS `merge` (requestSource `merge` adminS))) =$= CL.map toMessage- =$= stateMachine legionary nodeState- =$= handleOutput cm+ =$= messageSink legionary (rts, nodeState) where- handleOutput cm = awaitForever (lift . \case- Send peer message -> send cm peer message- NewPeers peers -> newPeers cm peers- )- toMessage :: Either (JoinRequest, JoinResponse -> LIO ()) (Either (PeerMessage i o s)- (Either (RequestMsg i o) (AdminMessage i o s)))- -> LInput i o s+ (Either+ (RequestMsg i o)+ (AdminMessage i o s)))+ -> RuntimeMessage i o s toMessage (Left m) = J m toMessage (Right (Left m)) = P m toMessage (Right (Right (Left m))) = R m@@ -128,16 +141,228 @@ return (transPipe (`runLoggingT` logging) c) -{- | This defines the various ways a node can be spun up. -}+messageSink :: (LegionConstraints i o s)+ => Legionary i o s+ -> (RuntimeState i o s, NodeState i s)+ -> Sink (RuntimeMessage i o s) LIO ()+messageSink legionary states =+ await >>= \case+ Nothing -> return ()+ Just msg -> do+ $(logDebug) . pack+ $ "Receieved: " ++ show msg+ lift . handleMessage legionary msg+ >=> lift . updatePeers legionary+ >=> lift . clusterHousekeeping legionary+ >=> messageSink legionary+ $ states+++{- |+ Make sure the connection manager knows about any new peers that have+ joined the cluster.+-}+updatePeers+ :: Legionary i o s+ -> (RuntimeState i o s, NodeState i s)+ -> LIO (RuntimeState i o s, NodeState i s)+updatePeers legionary (rts, ns) = do+ (peers, ns2) <- runSM legionary ns SM.getPeers+ newPeers (cm rts) peers+ return (rts, ns2)+++{- |+ Perform any cluster management actions, and update the state+ appropriately.+-}+clusterHousekeeping :: (LegionConstraints i o s)+ => Legionary i o s+ -> (RuntimeState i o s, NodeState i s)+ -> LIO (RuntimeState i o s, NodeState i s)+clusterHousekeeping legionary (rts, ns) = do+ (actions, ns2) <- runSM legionary ns (+ heartbeat+ >> rebalance+ >> migrate+ >> propagate+ )+ rts2 <- foldr (>=>) return (clusterAction <$> actions) rts+ return (rts2, ns2)+++{- |+ Actually perform a cluster action as directed by the state+ machine.+-}+clusterAction+ :: ClusterAction i s+ -> RuntimeState i o s+ -> LIO (RuntimeState i o s)++clusterAction+ (SM.ClusterMerge peer ps)+ rts@RuntimeState {self, nextId, cm}+ = do+ send cm peer (PeerMessage self nextId (ClusterMerge ps))+ return rts {nextId = next nextId}++clusterAction+ (SM.PartitionMerge peer key ps)+ rts@RuntimeState {self, nextId, cm}+ = do+ send cm peer (PeerMessage self nextId (PartitionMerge key ps))+ return rts {nextId = next nextId}+++{- |+ Handle an individual runtime message, accepting an initial runtime+ state and an initial node state, and producing an updated runtime+ state and node state.+-}+handleMessage :: (LegionConstraints i o s)+ => Legionary i o s+ -> RuntimeMessage i o s+ -> (RuntimeState i o s, NodeState i s)+ -> LIO (RuntimeState i o s, NodeState i s)++handleMessage {- Partition Merge -}+ legionary+ (P (PeerMessage source _ (PartitionMerge key ps)))+ (rts, ns)+ = do+ ((), ns2) <- runSM legionary ns (partitionMerge source key ps)+ return (rts, ns2)+ +handleMessage {- Cluster Merge -}+ legionary+ (P (PeerMessage source _ (ClusterMerge cs)))+ (rts, ns)+ = do+ ((), ns2) <- runSM legionary ns (clusterMerge source cs)+ return (rts, ns2)++handleMessage {- Forward Request -}+ legionary+ (P (msg@(PeerMessage source mid (ForwardRequest key request))))+ (rts@RuntimeState {nextId, cm, self}, ns)+ = do+ (output, ns2) <- runSM legionary ns (userRequest key request)+ case output of+ Respond response -> do+ send cm source (+ PeerMessage self nextId (ForwardResponse mid response)+ )+ return (rts {nextId = next nextId}, ns2)+ Forward peer -> do+ send cm peer msg+ return (rts {nextId = next nextId}, ns2)+ +handleMessage {- Forward Response -}+ _legionary+ (msg@(P (PeerMessage _ _ (ForwardResponse mid response))))+ (rts, ns)+ =+ case lookupDelete mid (forwarded rts) of+ (Nothing, fwd) -> do+ $(logWarn) . pack $ "Unsolicited ForwardResponse: " ++ show msg+ return (rts {forwarded = fwd}, ns)+ (Just respond, fwd) -> do+ respond response+ return (rts {forwarded = fwd}, ns)++handleMessage {- User Request -}+ legionary+ (R ((key, request), respond))+ (rts@RuntimeState {self, cm, nextId, forwarded}, ns)+ = do+ (output, ns2) <- runSM legionary ns (userRequest key request)+ case output of+ Respond response -> do+ lift (respond response)+ return (rts, ns2)+ Forward peer -> do+ send cm peer (+ PeerMessage self nextId (ForwardRequest key request)+ )+ return (+ rts {+ forwarded = Map.insert nextId (lift . respond) forwarded,+ nextId = next nextId+ },+ ns2+ )++handleMessage {- Join Request -}+ legionary+ (J (JoinRequest addy, respond))+ (rts, ns)+ = do+ ((peer, cluster), ns2) <- runSM legionary ns (SM.join addy)+ respond (JoinOk peer cluster)+ return (rts, ns2)++handleMessage {- Admin Get State -}+ _legionary+ (A (GetState respond))+ (rts, ns)+ =+ respond ns >> return (rts, ns)++handleMessage {- Admin Get Partition -}+ Legionary {persistence}+ (A (GetPart key respond))+ (rts, ns)+ = do+ respond =<< lift (getState persistence key)+ return (rts, ns)++handleMessage {- Admin Eject Peer -}+ legionary+ (A (Eject peer respond))+ (rts, ns)+ = do+ {-+ TODO: we should attempt to notify the ejected peer that it has+ been ejected instead of just cutting it off and washing our hands+ of it. I have a vague notion that maybe ejected peers should be+ permanently recorded in the cluster state so that if they ever+ reconnect then we can notify them that they are no longer welcome+ to participate.++ On a related note, we need to think very hard about the split brain+ problem. A random thought about that is that we should consider the+ extreme case where the network just fails completely and every node+ believes that every other node should be or has been ejected. This+ would obviously be catastrophic in terms of data durability unless+ we have some way to reintegrate an ejected node. So, either we+ have to guarantee that such a situation can never happen, or else+ implement a reintegration strategy. It might be acceptable for+ the reintegration strategy to be very costly if it is characterized+ as an extreme recovery scenario.++ Question: would a reintegration strategy become less costly if the+ "next state id" for a peer were global across all power states+ instead of local to each power state?+ -}+ ((), ns2) <- runSM legionary ns (eject peer)+ respond ()+ return (rts, ns2)+++{- | This defines the various ways a node can be spun up. -} data StartupMode = NewCluster- -- ^ Indicates that we should bootstrap a new cluster at startup. The- -- persistence layer may be safely pre-populated because the new- -- node will claim the entire keyspace. + {- ^+ Indicates that we should bootstrap a new cluster at startup. The+ persistence layer may be safely pre-populated because the new node+ will claim the entire keyspace.+ -} | JoinCluster SockAddr- -- ^ Indicates that the node should try to join an existing cluster,- -- either by starting fresh, or by recovering from a shutdown- -- or crash.+ {- ^+ Indicates that the node should try to join an existing cluster,+ either by starting fresh, or by recovering from a shutdown or crash.+ -} deriving (Show, Eq) @@ -157,7 +382,7 @@ inputChan <- newChan so <- socket (fam peerBindAddr) Stream defaultProtocol setSocketOption so ReuseAddr 1- bindSocket so peerBindAddr+ bind so peerBindAddr listen so 5 return (inputChan, so) forkC "peer socket acceptor" $ acceptLoop so inputChan@@ -182,7 +407,7 @@ let runSocket = sourceSocket conn =$= conduitDecode- $$ msgSink+ $$ msgSink void . lift . forkIO@@ -210,18 +435,19 @@ {- | Figure out how to construct the initial node state. -}-makeNodeState :: (LegionConstraints i o s)+makeNodeState :: (Show i) => LegionarySettings -> StartupMode- -> LIO (NodeState i o s, Map Peer BSockAddr)+ -> LIO (Peer, NodeState i s, Map Peer BSockAddr) makeNodeState LegionarySettings {peerBindAddr} NewCluster = do {- Build a brand new node state, for the first node in a cluster. -} self <- newPeer clusterId <- getUUID- let cluster = C.new clusterId self peerBindAddr- nodeState <- newNodeState self cluster- return (nodeState, C.getPeers cluster)+ let+ cluster = C.new clusterId self peerBindAddr+ nodeState = newNodeState self cluster+ return (self, nodeState, C.getPeers cluster) makeNodeState LegionarySettings {peerBindAddr} (JoinCluster addr) = do {-@@ -230,9 +456,10 @@ -} $(logInfo) "Trying to join an existing cluster." (self, clusterPS) <- joinCluster (JoinRequest (BSockAddr peerBindAddr))- let cluster = C.initProp self clusterPS- nodeState <- newNodeState self cluster- return (nodeState, C.getPeers cluster)+ let+ cluster = C.initProp self clusterPS+ nodeState = newNodeState self cluster+ return (self, nodeState, C.getPeers cluster) where joinCluster :: JoinRequest -> LIO (Peer, ClusterPowerState) joinCluster joinMsg = liftIO $ do@@ -268,7 +495,7 @@ chan <- newChan so <- socket (fam joinBindAddr) Stream defaultProtocol setSocketOption so ReuseAddr 1- bindSocket so joinBindAddr+ bind so joinBindAddr listen so 5 return (chan, so) forkC "join socket acceptor" $ acceptLoop so chan@@ -320,7 +547,7 @@ ) -{- | Guess the family of a `SockAddr`. -}+{- | Guess the family of a `SockAddr`. -} fam :: SockAddr -> Family fam SockAddrInet {} = AF_INET fam SockAddrInet6 {} = AF_INET6@@ -332,13 +559,13 @@ Forks the legion framework in a background thread, and returns a way to send user requests to it and retrieve the responses to those requests. -}-forkLegionary :: (LegionConstraints i o s, MonadLoggerIO io)+forkLegionary :: (LegionConstraints i o s, MonadLoggerIO io, MonadIO io2) => Legionary i o s {- ^ The user-defined legion application to run. -} -> LegionarySettings {- ^ Settings and configuration of the legionary framework. -} -> StartupMode- -> io (PartitionKey -> i -> IO o)+ -> io (PartitionKey -> i -> io2 o) forkLegionary legionary settings startupMode = do logging <- askLoggerIO@@ -346,10 +573,51 @@ chan <- liftIO newChan forkC "main legion thread" $ runLegionary legionary settings startupMode (chanToSource chan)- return (\ key request -> do+ return (\ key request -> liftIO $ do responseVar <- newEmptyMVar writeChan chan ((key, request), putMVar responseVar) takeMVar responseVar )+++{- | This is the type of message passed around in the runtime. -}+data RuntimeMessage i o s+ = P (PeerMessage i o s)+ | R (RequestMsg i o)+ | J (JoinRequest, JoinResponse -> LIO ())+ | A (AdminMessage i o s)+instance (Show i, Show o, Show s) => Show (RuntimeMessage i o s) where+ show (P m) = "(P " ++ show m ++ ")"+ show (R ((p, i), _)) = "(R ((" ++ show p ++ ", " ++ show i ++ "), _))"+ show (J (jr, _)) = "(J (" ++ show jr ++ ", _))"+ show (A a) = "(A (" ++ show a ++ "))"+++{- | The runtime state. -}+data RuntimeState i o s = RuntimeState {+ self :: Peer,+ forwarded :: Map MessageId (o -> LIO ()),+ nextId :: MessageId,+ cm :: ConnectionManager i o s+ }+++{- | This is the type of a join request message. -}+data JoinRequest = JoinRequest BSockAddr+ deriving (Generic, Show)+instance Binary JoinRequest+++{- | The response to a JoinRequst message -}+data JoinResponse+ = JoinOk Peer ClusterPowerState+ | JoinRejected String+ deriving (Generic)+instance Binary JoinResponse+++{- | Lookup a key from a map, and also delete the key if it exists. -}+lookupDelete :: (Ord k) => k -> Map k v -> (Maybe v, Map k v)+lookupDelete = Map.updateLookupWithKey (const (const Nothing))
+ src/Network/Legion/Runtime/ConnectionManager.hs view
@@ -0,0 +1,208 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE TemplateHaskell #-}+{- |+ This module manages connections to other nodes in the cluster.+-}+module Network.Legion.Runtime.ConnectionManager (+ ConnectionManager,+ newConnectionManager,+ send,+ newPeers,+) where++import Prelude hiding (lookup)++import Control.Concurrent (Chan, writeChan, newChan, readChan)+import Control.Exception (try, SomeException)+import Control.Monad (void)+import Control.Monad.Logger (logInfo, logWarn)+import Control.Monad.Trans.Class (lift)+import Data.Binary (Binary, encode)+import Data.ByteString.Lazy (ByteString)+import Data.Map (toList, insert, empty, Map, lookup)+import Data.Text (pack)+import Network.Legion.BSockAddr (BSockAddr(BSockAddr))+import Network.Legion.Distribution (Peer)+import Network.Legion.Fork (forkC)+import Network.Legion.LIO (LIO)+import Network.Legion.Runtime.PeerMessage (PeerMessage)+import Network.Socket (SockAddr, Socket, socket, SocketType(Stream),+ defaultProtocol, connect, close, SockAddr(SockAddrInet, SockAddrInet6,+ SockAddrUnix, SockAddrCan), Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN))+import Network.Socket.ByteString.Lazy (sendAll)++{- |+ A handle on the connection manager+-}+data ConnectionManager i o s = C (Chan (Message i o s))+instance Show (ConnectionManager i o s) where+ show _ = "ConnectionManager"+++{- |+ Create a new connection manager.+-}+newConnectionManager :: (Binary i, Binary o, Binary s)+ => Map Peer BSockAddr+ -> LIO (ConnectionManager i o s)+newConnectionManager initPeers = do+ chan <- lift newChan+ forkC "connection manager thread" $+ manager chan S {connections = empty}+ let cm = C chan+ newPeers cm initPeers+ return cm+ where+ manager :: (Binary s, Binary o, Binary i)+ => Chan (Message i o s)+ -> State i o s+ -> LIO ()+ manager chan state = lift (readChan chan) >>= handle state >>= manager chan++ handle :: (Binary i, Binary o, Binary s)+ => State i o s+ -> Message i o s+ -> LIO (State i o s)+ handle s@S {connections} (NewPeer peer addr) =+ case lookup peer connections of+ Nothing -> do+ conn <- connection addr+ return s {+ connections = insert peer conn connections+ }+ Just _ ->+ return s++ handle s@S {connections} (Send peer msg) = do+ case lookup peer connections of+ Nothing -> $(logWarn) . pack $ "unknown peer: " ++ show peer+ Just conn -> lift $ writeChan conn msg+ return s+++{- |+ Build a new connection.+-}+connection :: (Binary i, Binary o, Binary s)+ => SockAddr+ -> LIO (Chan (PeerMessage i o s))++connection addr = do+ chan <- lift newChan+ forkC ("connection to: " ++ show addr) $+ handle chan Nothing+ return chan+ where+ handle :: (Binary i, Binary o, Binary s)+ => Chan (PeerMessage i o s)+ -> Maybe Socket+ -> LIO ()+ handle chan so =+ lift (readChan chan) >>= sendWithRetry so . encode >>= handle chan++ {- |+ Open a socket.+ -}+ openSocket :: IO Socket+ openSocket = do+ so <- socket (fam addr) Stream defaultProtocol+ connect so addr+ return so++ {- |+ Try to send the payload over the socket, and if that fails, then try to+ create a new socket and retry sending the payload. Return whatever the+ "working" socket is.+ -}+ sendWithRetry :: Maybe Socket -> ByteString -> LIO (Maybe Socket)+ sendWithRetry Nothing payload =+ (lift . try) openSocket >>= \case+ Left err -> do+ $(logWarn) . pack+ $ "Can't connect to: " ++ show addr ++ ". Dropping message on "+ ++ "the floor: " ++ show payload ++ ". The error was: "+ ++ show (err :: SomeException)+ return Nothing+ Right so -> do+ result2 <- (lift . try) (sendAll so payload)+ case result2 of+ Left err -> $(logWarn) . pack+ $ "An error happend when trying to send a payload over a socket "+ ++ "to the address: " ++ show addr ++ ". The error was: "+ ++ show (err :: SomeException) ++ ". This is the last straw, we "+ ++ "are not retrying. The message is being dropped on the floor. "+ ++ "The message was: " ++ show payload+ Right _ -> return ()+ return (Just so)+ sendWithRetry (Just so) payload =+ (lift . try) (sendAll so payload) >>= \case+ Left err -> do+ $(logInfo) . pack+ $ "Socket to " ++ show addr ++ " died. Retrying on a new "+ ++ "socket. The error was: " ++ show (err :: SomeException)+ (lift . void) (try (close so) :: IO (Either SomeException ()))+ sendWithRetry Nothing payload+ Right _ ->+ return (Just so)+++{- |+ Send a message to a peer.+-}+send+ :: ConnectionManager i o s+ -> Peer+ -> PeerMessage i o s+ -> LIO ()+send (C chan) peer = lift . writeChan chan . Send peer+++{- |+ Tell the connection manager about a new peer.+-}+newPeer+ :: ConnectionManager i o s+ -> Peer+ -> SockAddr+ -> LIO ()+newPeer (C chan) peer addr = lift $ writeChan chan (NewPeer peer addr)+++{- |+ Tell the connection manager about all the peers known to the cluster state.+-}+newPeers :: ConnectionManager i o s -> Map Peer BSockAddr -> LIO ()+newPeers cm peers =+ mapM_ oneNewPeer (toList peers)+ where+ oneNewPeer (peer, BSockAddr addy) = newPeer cm peer addy+++{- |+ The internal state of the connection manager.+-}+data State i o s = S {+ connections :: Map Peer (Chan (PeerMessage i o s))+ }+++{- |+ The types of messages that the ConnectionManager understands.+-}+data Message i o s+ = NewPeer Peer SockAddr+ | Send Peer (PeerMessage i o s)+++{- |+ Guess the family of a `SockAddr`.+-}+fam :: SockAddr -> Family+fam SockAddrInet {} = AF_INET+fam SockAddrInet6 {} = AF_INET6+fam SockAddrUnix {} = AF_UNIX+fam SockAddrCan {} = AF_CAN++
+ src/Network/Legion/Runtime/PeerMessage.hs view
@@ -0,0 +1,79 @@+{-# LANGUAGE DeriveGeneric #-}+{- |+ This module contains the type of runtime messages that can be exchanged+ between peers.+-}+module Network.Legion.Runtime.PeerMessage (+ PeerMessage(..),+ PeerMessagePayload(..),+ MessageId,+ newSequence,+ next,+) where++import Control.Monad.Trans.Class (lift)+import Data.Binary (Binary)+import Data.UUID (UUID)+import Data.Word (Word64)+import GHC.Generics (Generic)+import Network.Legion.ClusterState (ClusterPowerState)+import Network.Legion.Distribution (Peer)+import Network.Legion.LIO (LIO)+import Network.Legion.PartitionKey (PartitionKey)+import Network.Legion.PartitionState (PartitionPowerState)+import Network.Legion.UUID (getUUID)+++{- |+ The type of messages sent between peers.+-}+data PeerMessage i o s = PeerMessage {+ source :: Peer,+ messageId :: MessageId,+ payload :: PeerMessagePayload i o s+ }+ deriving (Generic, Show)+instance (Binary i, Binary o, Binary s) => Binary (PeerMessage i o s)+++{- |+ The data contained within a peer message.++ When we get around to implementing durability and data replication,+ the sustained inability to confirm that a node has received one of+ these messages should result in the ejection of that node from the+ cluster and the blacklisting of that node so that it can never re-join.+-}+data PeerMessagePayload i o s+ = PartitionMerge PartitionKey (PartitionPowerState i s)+ | ForwardRequest PartitionKey i+ | ForwardResponse MessageId o+ | ClusterMerge ClusterPowerState+ deriving (Generic, Show)+instance (Binary i, Binary o, Binary s) => Binary (PeerMessagePayload i o s)+++data MessageId = M UUID Word64 deriving (Generic, Show, Eq, Ord)+instance Binary MessageId+++{- |+ Initialize a new sequence of messageIds. It would be perfectly fine to ensure+ unique message ids by generating a unique UUID for each one, but generating+ UUIDs is not free, and we are probably going to be generating a lot of these.+-}+newSequence :: LIO MessageId+newSequence = lift $ do+ sid <- getUUID+ return (M sid 0)+++{- |+ Generate the next message id in the sequence. We would normally use+ `succ` for this kind of thing, but making `MessageId` an instance of+ `Enum` really isn't appropriate.+-}+next :: MessageId -> MessageId+next (M sequenceId ord) = M sequenceId (ord + 1)++
src/Network/Legion/StateMachine.hs view
@@ -1,75 +1,91 @@-{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TemplateHaskell #-} {- |- This module contains the state machine implementation of a legion node.+ This module contains the "pure-ish" state machine that defines what+ it means to be a legion node. As described on 'SM', the state machine+ is modeled in monadic fashion, where the state machine sate is modeled+ as monadic context, state machine input is modeled as various monadic+ functions, and state machine output is modeled as the result of those+ monadic functions. - Discussion:+ The reason the state lives behind a monad is because part of the+ node state (i.e. the persistence layer) really does live behind IO,+ and cannot be accessed purely. Therefore, the state is divided into a+ pure part, modeled by 'NodeState'; and an impure part, modeled by the+ persistence layer interface. We wrap these two components inside+ of a new, opaque, monad called 'SM' by using a monad transformation+ stack, where 'StateT' wraps the pure part of the state, and IO wraps+ the impure part of the state. (This is a simplified description. The+ actual monad transformation stack is more complicated, because it+ incorporates logging and access to the user-defined request handler.) - This is a first attempt to discover a pure legion state machine and isolated- it from the runtime IO considerations. It is obviously not perfect, because- everything still lives in 'LIO', which is 'IO'-backed; but mostly this is- because access to the persistence layer still happens here. Once we pull that- out into the 'Network.Legion.Runtime' module we should be clear to remove IO- and make this thing look more like a pure state machine. - Rick+ The overall purpose of all of this is to separate as much as+ possible the abstract idea of what a legion node is with its runtime+ considerations. The state machine contained in this module defines how a+ legion node should behave when faced with various inputs, and it would+ be completely pure but for the persistence layer interface. The runtime+ system 'Network.Legion.Runtime' implements the mechanisms by which+ such input is collected and any behavior associated with the output+ (e.g. managing network connections, sending data across the wire,+ reading data from the wire, transforming those data into inputs to+ the state machine, etc.). -}-module Network.Legion.StateMachine (- stateMachine,- LInput(..),- LOutput(..),- JoinRequest(..),- JoinResponse(..),- AdminMessage(..),+module Network.Legion.StateMachine(+ -- * Running the state machine. NodeState,- Forwarded(..),- PeerMessage(..),- PeerMessagePayload(..),- MessageId,- next, newNodeState,-) where+ SM,+ runSM, -import Prelude hiding (lookup)+ -- * State machine inputs.+ userRequest,+ partitionMerge,+ clusterMerge,+ migrate,+ propagate,+ rebalance,+ heartbeat,+ eject,+ join, -import Control.Exception (throw)+ -- * State machine outputs.+ ClusterAction(..),+ UserResponse(..),++ -- * State inspection+ getPeers,+) where+ import Control.Monad (unless)-import Control.Monad.Catch (try, SomeException, MonadCatch, MonadThrow)-import Control.Monad.IO.Class (MonadIO, liftIO)-import Control.Monad.Logger (logDebug, logWarn, logError, logInfo,- MonadLogger)-import Control.Monad.Trans.Class (MonadTrans, lift)-import Control.Monad.Trans.State (StateT, runStateT, get, put)-import Data.Binary (Binary)-import Data.Conduit (Source, Conduit, ($$), await, awaitForever,- transPipe, ConduitM, yield, ($=))+import Control.Monad.IO.Class (MonadIO)+import Control.Monad.Logger (MonadLogger, logWarn, logDebug, logError)+import Control.Monad.Trans.Class (lift, MonadTrans)+import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask)+import Control.Monad.Trans.State (StateT, runStateT, get, put, modify)+import Data.Aeson (ToJSON, toJSON, object, (.=), encode)+import Data.ByteString.Lazy (toStrict)+import Data.Conduit (($=), ($$), Sink, transPipe, awaitForever) import Data.Default.Class (Default)-import Data.Map (Map, insert, lookup)+import Data.Map (Map) import Data.Maybe (fromMaybe)-import Data.Set (member, minView, (\\))-import Data.Text (pack)+import Data.Set ((\\))+import Data.Text (pack, unpack)+import Data.Text.Encoding (decodeUtf8) import Data.Time.Clock (getCurrentTime)-import Data.UUID (UUID)-import Data.Word (Word64)-import GHC.Generics (Generic)-import Network.Legion.Application (Legionary, LegionConstraints,- Persistence(getState, saveState, list), Legionary(Legionary,- persistence, handleRequest), RequestMsg)+import Network.Legion.Application (Legionary(Legionary), getState,+ saveState, list, persistence, handleRequest) import Network.Legion.BSockAddr (BSockAddr)-import Network.Legion.ClusterState (claimParticipation, ClusterPropState,- getPeers, getDistribution, ClusterPowerState)-import Network.Legion.Distribution (rebalanceAction, RebalanceAction(- Invite), Peer, newPeer)-import Network.Legion.KeySet (union, KeySet)+import Network.Legion.ClusterState (ClusterPropState, ClusterPowerState)+import Network.Legion.Distribution (Peer, rebalanceAction, newPeer,+ RebalanceAction(Invite))+import Network.Legion.KeySet (KeySet, union) import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.PartitionState (PartitionPowerState, PartitionPropState) import Network.Legion.PowerState (ApplyDelta)-import Network.Legion.UUID (getUUID) import qualified Data.Conduit.List as CL import qualified Data.Map as Map import qualified Data.Set as Set@@ -79,262 +95,133 @@ {- |- This conduit houses the main legionary state machine. The conduit's- input, internal state, and output are analogous to a "real" state- machine's input, state, and output. If this seems like an odd use of- conduit, that's ok. Hopefully we can make this look more like a pure- state machine once we remove 'IO' from this module.+ This is the portion of the local node state that is not persistence+ related. -}-stateMachine :: (LegionConstraints i o s)- => Legionary i o s- -> NodeState i o s- -> Conduit (LInput i o s) LIO (LOutput i o s)-stateMachine l n = awaitForever (\msg -> do- newState <- runStateMT n $ do- handleMessage l msg- heartbeat- migrate l- propagate- rebalance l- logState- stateMachine l newState- )- where- logState = lift . logNodeState =<< getS+data NodeState i s = NodeState {+ self :: Peer,+ cluster :: ClusterPropState,+ partitions :: Map PartitionKey (PartitionPropState i s),+ migration :: KeySet+ }+instance (Show i, Show s) => Show (NodeState i s) where+ show = unpack . decodeUtf8 . toStrict . encode+{-+ The ToJSON instance is mainly for debugging. The Haskell-generated 'Show'+ instance is very hard to read.+-}+instance (Show i, Show s) => ToJSON (NodeState i s) where+ toJSON (NodeState self cluster partitions migration) =+ object [+ "self" .= show self,+ "cluster" .= cluster,+ "partitions" .= Map.mapKeys show partitions,+ "migration" .= show migration+ ] -{- | Handle one incomming message. -}-handleMessage :: (LegionConstraints i o s)- => Legionary i o s- -> LInput i o s- -> StateM i o s ()+{- |+ Make a new node state.+-}+newNodeState :: Peer -> ClusterPropState -> NodeState i s+newNodeState self cluster =+ NodeState {+ self,+ cluster,+ partitions = Map.empty,+ migration = KS.empty+ } -handleMessage l msg = do- NodeState {cluster} <- getS- let- {- | Return `True` if the peer is a known peer, false otherwise. -}- known peer = peer `member` C.allParticipants cluster- $(logDebug) . pack $ "Receiving: " ++ show msg- case msg of- P peerMsg@PeerMessage {source} ->- if known source- then handlePeerMessage l peerMsg- else- $(logWarn) . pack- $ "Dropping message from unknown peer: " ++ show source- R ((key, request), respond) ->- case minView (C.findPartition key cluster) of- Nothing ->- $(logError) . pack- $ "Keyspace does not contain key: " ++ show key ++ ". This "- ++ "is a very bad thing and probably means there is a bug, "- ++ "or else this node has not joined a cluster yet."- Just (peer, _) ->- forward peer key request respond- J m -> handleJoinRequest m- A m -> handleAdminMessage l m +{- |+ This monad encapsulates the global state of the legion node (not+ counting the runtime stuff, like open connections and what have+ you). -{- | Handles one incomming message from a peer. -}-handlePeerMessage :: (LegionConstraints i o s)- => Legionary i o s- -> PeerMessage i o s- -> StateM i o s ()+ The main reason that the state is hidden behind a monad is because part+ of the sate (i.e. the partition data) lives behind 'IO'. Therefore,+ if we want to model the global state of the node as a single unit,+ we have to do so using a monad.+-}+newtype SM i o s a = SM {+ unSM :: ReaderT (Legionary i o s) (StateT (NodeState i s) LIO) a+ }+ deriving (Functor, Applicative, Monad, MonadLogger, MonadIO) -handlePeerMessage -- PartitionMerge- Legionary {- persistence- }- msg@PeerMessage {- source,- payload = PartitionMerge key ps- }- = do- nodeState@NodeState {self, propStates, cluster} <- getS- propState <- lift $ maybe- (getStateL persistence self cluster key)- return- (lookup key propStates)- let- owners = C.findPartition key cluster- case P.mergeEither source ps propState of- Left err ->- $(logWarn) . pack- $ "Can't apply incomming partition action message "- ++ show msg ++ "because of: " ++ show err- Right newPropState -> do- $(logDebug) "Saving because of PartitionMerge"- lift $ saveStateL persistence key (- if P.participating newPropState- then Just (P.getPowerState newPropState)- else Nothing- )- putS nodeState {- propStates = if newPropState == P.new key self owners- then Map.delete key propStates- else insert key newPropState propStates- } -handlePeerMessage -- ForwardRequest- Legionary {handleRequest, persistence}- msg@PeerMessage {- payload = ForwardRequest key request,- source,- messageId- }- = do- ns@NodeState {self, cluster, propStates} <- getS- let owners = C.findPartition key cluster- if self `member` owners- then do- let- respond = send source . ForwardResponse messageId-- -- TODO - -- - figure out some slick concurrency here, by maintaining- -- a map of keys that are currently being accessed or- -- something- -- - either (respond . rethrow) respond =<< try (do - prop <- lift $ getStateL persistence self cluster key- let response = handleRequest key request (P.ask prop)- newProp = P.delta request prop- $(logDebug) "Saving because of ForwardRequest"- lift $ saveStateL persistence key (Just (P.getPowerState newProp))- $(logInfo) . pack- $ "Handling user request: " ++ show request- $(logDebug) . pack- $ "Request details request: " ++ show prop ++ " ++ "- ++ show request ++ " --> " ++ show (response, newProp)- putS ns {propStates = insert key newProp propStates}- return response- )- else- {-- we don't own the key after all, someone was wrong to forward- us this request.- -}- case minView owners of- Nothing -> $(logError) . pack- $ "Can't find any owners for the key: " ++ show key- Just (peer, _) ->- emit (Send peer msg)- where- {- |- rethrow is just a reification of `throw`.- -}- rethrow :: SomeException -> a- rethrow = throw--handlePeerMessage -- ForwardResponse- Legionary {}- msg@PeerMessage {- payload = ForwardResponse messageId response- }- = do- nodeState@NodeState {forwarded} <- getS- case lookup messageId (unF forwarded) of- Nothing -> $(logWarn) . pack- $ "This peer received a response for a forwarded request that it "- ++ "didn't send. The only time you might expect to see this is if "- ++ "this peer recently crashed and was immediately restarted. If "- ++ "you are seeing this in other circumstances then probably "- ++ "something is very wrong at the network level. The message was: "- ++ show msg- Just respond ->- lift $ respond response- putS nodeState {- forwarded = F . Map.delete messageId . unF $ forwarded- }--handlePeerMessage -- ClusterMerge- Legionary {}- msg@PeerMessage {- source,- payload = ClusterMerge ps- }- = do- nodeState@NodeState {migration, cluster} <- getS- case C.mergeEither source ps cluster of- Left err ->- $(logWarn) . pack- $ "Can't apply incomming cluster action message "- ++ show msg ++ "because of: " ++ show err- Right (newCluster, newMigration) ->- putS nodeState {- migration = migration `union` newMigration,- cluster = newCluster- }+{- |+ Run an SM action.+-}+runSM+ :: Legionary i o s+ -> NodeState i s+ -> SM i o s a+ -> LIO (a, NodeState i s)+runSM l ns action = runStateT (runReaderT (unSM action) l) ns -{- | Handle a join request message -}-handleJoinRequest- :: (JoinRequest, JoinResponse -> LIO ())- -> StateM i o s ()+{- | Handle a user request. -}+userRequest :: (ApplyDelta i s, Default s)+ => PartitionKey+ -> i+ -> SM i o s (UserResponse o)+userRequest key request = SM $ do+ NodeState {self, cluster} <- lift get+ Legionary {handleRequest} <- ask+ let owners = C.findPartition key cluster+ if self `Set.member` owners+ then do+ partition <- unSM $ getPartition key+ let+ response = handleRequest key request (P.ask partition)+ partition2 = P.delta request partition+ unSM $ savePartition key partition2+ return (Respond response) -handleJoinRequest (JoinRequest peerAddr, respond) = do- ns@NodeState {cluster} <- getS- peer <- lift newPeer- let newCluster = C.joinCluster peer peerAddr cluster- emit . NewPeers . getPeers $ newCluster- lift $ respond (JoinOk peer (C.getPowerState newCluster))- putS ns {cluster = newCluster}+ else case Set.toList owners of+ [] -> do+ let msg = "No owners for key: " ++ show key+ $(logError) . pack $ msg+ error msg+ peer:_ -> return (Forward peer) {- |- Handle a message from the admin service.+ Handle the state transition for a partition merge event. Returns 'Left'+ if there is an error, and 'Right' if everything went fine. -}-handleAdminMessage- :: Legionary i o s- -> AdminMessage i o s- -> StateM i o s ()-handleAdminMessage _ (GetState respond) =- lift . respond =<< getS-handleAdminMessage Legionary {persistence} (GetPart key respond) = lift $ do- partitionVal <- lift (getState persistence key)- respond partitionVal-handleAdminMessage _ (Eject peer respond) = do- {-- TODO: we should attempt to notify the ejected peer that it has- been ejected instead of just cutting it off and washing our hands- of it. I have a vague notion that maybe ejected peers should be- permanently recorded in the cluster state so that if they ever- reconnect then we can notify them that they are no longer welcome- to participate.-- On a related note, we need to think very hard about the split brain- problem. A random thought about that is that we should consider the- extreme case where the network just fails completely and every node- believes that every other node should be or has been ejected. This- would obviously be catastrophic in terms of data durability unless- we have some way to reintegrate an ejected node. So, either we- have to guarantee that such a situation can never happen, or else- implement a reintegration strategy. It might be acceptable for- the reintegration strategy to be very costly if it is characterized- as an extreme recovery scenario.-- Question: would a reintegration strategy become less costly if the- "next state id" for a peer were global across all power states- instead of local to each power state?- -}- modifyS eject- lift $ respond ()- where- eject ns@NodeState {cluster} = ns {cluster = C.eject peer cluster}+partitionMerge :: (Show i, Show s, ApplyDelta i s, Default s)+ => Peer+ -> PartitionKey+ -> PartitionPowerState i s+ -> SM i o s ()+partitionMerge source key foreignPartition = do+ partition <- getPartition key+ case P.mergeEither source foreignPartition partition of+ Left err -> $(logWarn) . pack+ $ "Can't apply incomming partition merge from "+ ++ show source ++ ": " ++ show foreignPartition+ ++ ". because of: " ++ show err+ Right newPartition -> savePartition key newPartition -{- | Update all of the propagation states with the current time. -}-heartbeat :: StateM i o s ()-heartbeat = do- now <- liftIO getCurrentTime- ns@NodeState {cluster, propStates} <- getS- putS ns {- cluster = C.heartbeat now cluster,- propStates = Map.fromAscList [- (k, P.heartbeat now p)- | (k, p) <- Map.toAscList propStates- ]- }+{- | Handle the state transition for a cluster merge event. -}+clusterMerge+ :: Peer+ -> ClusterPowerState+ -> SM i o s ()+clusterMerge source foreignCluster = SM . lift $ do+ nodeState@NodeState {migration, cluster} <- get+ case C.mergeEither source foreignCluster cluster of+ Left err -> $(logWarn) . pack+ $ "Can't apply incomming cluster merge from "+ ++ show source ++ ": " ++ show foreignCluster+ ++ ". because of: " ++ show err+ Right (newCluster, newMigration) ->+ put nodeState {+ migration = migration `union` newMigration,+ cluster = newCluster+ } {- |@@ -349,351 +236,215 @@ peer to a partition. This will cause the data to be transfered in the normal course of propagation. -}-migrate :: (LegionConstraints i o s) => Legionary i o s -> StateM i o s ()-migrate Legionary{persistence} = do- ns@NodeState {migration} <- getS+migrate :: (ApplyDelta i s) => SM i o s ()+migrate = do+ NodeState {migration} <- (SM . lift) get+ Legionary {persistence} <- SM ask unless (KS.null migration) $- putS =<< lift (- listL persistence- $= CL.filter ((`KS.member` migration) . fst)- $$ accum ns {migration = KS.empty}- )+ transPipe (SM . lift3) (list persistence)+ $= CL.filter ((`KS.member` migration) . fst)+ $$ accum+ (SM . lift) $ modify (\ns -> ns {migration = KS.empty}) where- accum ns@NodeState {self, cluster, propStates} = await >>= \case- Nothing -> return ns- Just (key, ps) -> - let- origProp = fromMaybe (P.initProp self ps) (lookup key propStates)- newPeers_ = C.findPartition key cluster \\ P.projParticipants origProp- {- This 'P.participate' is where the magic happens. -}- newProp = foldr P.participate origProp (Set.toList newPeers_)- in do- $(logDebug) . pack $ "Migrating: " ++ show key- lift (saveStateL persistence key (Just (P.getPowerState newProp)))- accum ns {- propStates = Map.insert key newProp propStates- }+ accum :: (ApplyDelta i s)+ => Sink (PartitionKey, PartitionPowerState i s) (SM i o s) ()+ accum = awaitForever $ \ (key, ps) -> do+ NodeState {self, cluster, partitions} <- (lift . SM . lift) get+ let+ partition = fromMaybe (P.initProp self ps) (Map.lookup key partitions)+ newPeers = C.findPartition key cluster \\ P.projParticipants partition+ newPartition = foldr P.participate partition (Set.toList newPeers)+ $(logDebug) . pack $ "Migrating: " ++ show key+ lift (savePartition key newPartition) {- | Handle all cluster and partition state propagation actions, and return an updated node state. -}-propagate :: (LegionConstraints i o s) => StateM i o s ()-propagate = do- ns@NodeState {cluster, propStates, self} <- getS- let (peers, ps, cluster2) = C.actions cluster- $(logDebug) . pack $ "Cluster Actions: " ++ show (peers, ps)- mapM_ (doClusterAction ps) (Set.toList peers)- propStates2 <- mapM doPartitionActions (Map.toList propStates)- putS ns {- cluster = cluster2,- propStates = Map.fromAscList [- (k, p)- | (k, p) <- propStates2- , p /= P.initProp self (P.getPowerState p)- ]- }+propagate :: SM i o s [ClusterAction i s]+propagate = SM $ do+ partitionActions <- getPartitionActions+ clusterActions <- unSM getClusterActions+ return (clusterActions ++ partitionActions) where- doClusterAction ps peer =- send peer (ClusterMerge ps)+ getPartitionActions = do+ ns@NodeState {partitions} <- lift get+ let+ updates = [+ (key, newPartition, [+ PartitionMerge peer key ps+ | peer <- Set.toList peers_+ ])+ | (key, partition) <- Map.toAscList partitions+ , let (peers_, ps, newPartition) = P.actions partition+ ]+ actions = [a | (_, _, as) <- updates, a <- as]+ newPartitions = Map.fromAscList [+ (key, newPartition)+ | (key, newPartition, _) <- updates+ , not (P.complete newPartition)+ ]+ (lift . put) ns {+ partitions = newPartitions+ }+ return actions - doPartitionActions (key, propState) = do- let (peers, ps, propState2) = P.actions propState- mapM_ (perform ps) (Set.toList peers)- return (key, propState2)- where- perform ps peer =- send peer (PartitionMerge key ps)+ getClusterActions :: SM i o s [ClusterAction i s]+ getClusterActions = SM $ do+ ns@NodeState {cluster} <- lift get+ let+ (peers, cs, newCluster) = C.actions cluster+ actions = [ClusterMerge peer cs | peer <- Set.toList peers]+ (lift . put) ns {+ cluster = newCluster+ }+ return actions {- | Figure out if any rebalancing actions must be taken by this node, and kick them off if so. -}-rebalance :: (LegionConstraints i o s) => Legionary i o s -> StateM i o s ()-rebalance _ = do- ns@NodeState {self, cluster} <- getS+rebalance :: SM i o s ()+rebalance = SM $ do+ ns@NodeState {self, cluster} <- lift get let- allPeers = (Set.fromList . Map.keys . getPeers) cluster- dist = getDistribution cluster+ allPeers = (Set.fromList . Map.keys . C.getPeers) cluster+ dist = C.getDistribution cluster action = rebalanceAction self allPeers dist $(logDebug) . pack $ "The rebalance action is: " ++ show action- putS ns {+ (lift . put) ns { cluster = case action of Nothing -> cluster- Just (Invite ks) -> claimParticipation self ks cluster+ Just (Invite ks) -> C.claimParticipation self ks cluster } -{- | This is the type of input accepted by the legionary state machine. -}-data LInput i o s- = P (PeerMessage i o s)- | R (RequestMsg i o)- | J (JoinRequest, JoinResponse -> LIO ())- | A (AdminMessage i o s)--instance (Show i, Show o, Show s) => Show (LInput i o s) where- show (P m) = "(P " ++ show m ++ ")"- show (R ((p, i), _)) = "(R ((" ++ show p ++ ", " ++ show i ++ "), _))"- show (J (jr, _)) = "(J (" ++ show jr ++ ", _))"- show (A a) = "(A (" ++ show a ++ "))"---{- | This is the type of output produced by the legionary state machine. -}-data LOutput i o s- = Send Peer (PeerMessage i o s)- | NewPeers (Map Peer BSockAddr)---{- | A helper function to log the state of the node: -}-logNodeState :: (LegionConstraints i o s) => NodeState i o s -> LIO ()-logNodeState ns = $(logDebug) . pack- $ "The current node state is: " ++ show ns---{- | Like `getState`, but in LIO, and provides the correct bottom value. -}-getStateL :: (ApplyDelta i s, Default s)- => Persistence i s- -> Peer- -> ClusterPropState- -> PartitionKey- -> LIO (PartitionPropState i s)--getStateL p self cluster key =- {- dp == default participants -}- let dp = C.findPartition key cluster- in maybe- (P.new key self dp)- (P.initProp self)- <$> lift (getState p key)---{- | Like `saveState`, but in LIO. -}-saveStateL- :: Persistence i s- -> PartitionKey- -> Maybe (PartitionPowerState i s)- -> LIO ()-saveStateL p k = lift . saveState p k---{- | Like `list`, but in LIO. -}-listL :: Persistence i s -> Source LIO (PartitionKey, PartitionPowerState i s)-listL p = transPipe lift (list p)---{- | This is the type of a join request message. -}-data JoinRequest = JoinRequest BSockAddr- deriving (Generic, Show)-instance Binary JoinRequest---{- | The response to a JoinRequst message -}-data JoinResponse- = JoinOk Peer ClusterPowerState- | JoinRejected String- deriving (Generic)-instance Binary JoinResponse---{- |- The type of messages sent by the admin service.--}-data AdminMessage i o s- = GetState (NodeState i o s -> LIO ())- | GetPart PartitionKey (Maybe (PartitionPowerState i s) -> LIO ())- | Eject Peer (() -> LIO ())--instance Show (AdminMessage i o s) where- show (GetState _) = "(GetState _)"- show (GetPart k _) = "(GetPart " ++ show k ++ " _)"- show (Eject p _) = "(Eject " ++ show p ++ " _)"---{- | Defines the local state of a node in the cluster. -}-data NodeState i o s = NodeState {- self :: Peer,- cluster :: ClusterPropState,- forwarded :: Forwarded o,- propStates :: Map PartitionKey (PartitionPropState i s),- migration :: KeySet,- nextId :: MessageId- }- deriving (Show)---{- | A set of forwarded messages. -}-newtype Forwarded o = F {unF :: Map MessageId (o -> LIO ())}-instance Show (Forwarded o) where- show = show . Map.keys . unF---{- |- The type of messages sent to us from other peers.--}-data PeerMessage i o s = PeerMessage {- source :: Peer,- messageId :: MessageId,- payload :: PeerMessagePayload i o s- }- deriving (Generic, Show)-instance (Binary i, Binary o, Binary s) => Binary (PeerMessage i o s)+{- | Update all of the propagation states with the current time. -}+heartbeat :: SM i o s ()+heartbeat = SM $ do+ now <- lift3 getCurrentTime+ ns@NodeState {cluster, partitions} <- lift get+ (lift . put) ns {+ cluster = C.heartbeat now cluster,+ partitions = Map.fromAscList [+ (k, P.heartbeat now p)+ | (k, p) <- Map.toAscList partitions+ ]+ } -{- |- The data contained within a peer message.-- When we get around to implementing durability and data replication,- the sustained inability to confirm that a node has received one of- these messages should result in the ejection of that node from the- cluster and the blacklisting of that node so that it can never re-join.--}-data PeerMessagePayload i o s- = PartitionMerge PartitionKey (PartitionPowerState i s)- | ForwardRequest PartitionKey i- | ForwardResponse MessageId o- | ClusterMerge ClusterPowerState- deriving (Generic, Show)-instance (Binary i, Binary o, Binary s) => Binary (PeerMessagePayload i o s)+{- | Eject a peer from the cluster. -}+eject :: Peer -> SM i o s ()+eject peer = SM . lift $ do+ ns@NodeState {cluster} <- get+ put ns {cluster = C.eject peer cluster} -data MessageId = M UUID Word64 deriving (Generic, Show, Eq, Ord)-instance Binary MessageId+{- | Handle a peer join request. -}+join :: BSockAddr -> SM i o s (Peer, ClusterPowerState)+join peerAddr = SM $ do+ peer <- lift2 newPeer+ ns@NodeState {cluster} <- lift get+ let newCluster = C.joinCluster peer peerAddr cluster+ (lift . put) ns {cluster = newCluster}+ return (peer, C.getPowerState newCluster) {- |- Generate the next message id in the sequence. We would normally use- `succ` for this kind of thing, but making `MessageId` an instance of- `Enum` really isn't appropriate.+ These are the actions that a node can take which allow it to coordinate+ with other nodes. It is up to the runtime system to implement the+ actions. -}-next :: MessageId -> MessageId-next (M sequenceId ord) = M sequenceId (ord + 1)+data ClusterAction i s+ = ClusterMerge Peer ClusterPowerState+ | PartitionMerge Peer PartitionKey (PartitionPowerState i s) {- |- Initialize a new sequence of messageIds+ The type of response to a user request, either forward to another node,+ or respond directly. -}-newSequence :: LIO MessageId-newSequence = lift $ do- sid <- getUUID- return (M sid 0)+data UserResponse o+ = Forward Peer+ | Respond o -{- |- Make a new node state.--}-newNodeState :: Peer -> ClusterPropState -> LIO (NodeState i o s)-newNodeState self cluster = do- nextId <- newSequence- return NodeState {- self,- nextId,- cluster,- forwarded = F Map.empty,- propStates = Map.empty,- migration = KS.empty- }+{- | Get the known peer data from the cluster. -}+getPeers :: SM i o s (Map Peer BSockAddr)+getPeers = SM $ C.getPeers . cluster <$> lift get -send :: Peer -> PeerMessagePayload i o s -> StateM i o s ()-send peer payload = do- ns@NodeState {self, nextId} <- getS- emit (Send peer PeerMessage {- source = self,- messageId = nextId,- payload- })- putS ns {nextId = next nextId}+{- | Gets a partition state. -}+getPartition :: (Default s, ApplyDelta i s)+ => PartitionKey+ -> SM i o s (PartitionPropState i s)+getPartition key = SM $ do+ Legionary {persistence} <- ask+ NodeState {self, partitions, cluster} <- lift get+ case Map.lookup key partitions of+ Nothing ->+ lift3 (getState persistence key) <&> \case+ Nothing -> P.new key self (C.findPartition key cluster)+ Just partition -> P.initProp self partition+ Just partition -> return partition -{- |- Forward a user request to a peer for handling, being sure to do all- the node state accounting.--}-forward- :: Peer- -> PartitionKey- -> i- -> (o -> IO ())- -> StateM i o s ()-forward peer key request respond = do- ns@NodeState {nextId, self, forwarded} <- getS- emit (Send peer PeerMessage {- source = self,- messageId = nextId,- payload = ForwardRequest key request- })- putS ns {- nextId = next nextId,- forwarded = F . insert nextId (lift . respond) . unF $ forwarded+{- | Saves a partition state. -}+savePartition :: PartitionKey -> PartitionPropState i s -> SM i o s ()+savePartition key partition = SM $ do+ Legionary {persistence} <- ask+ ns@NodeState {partitions} <- lift get+ lift3 (saveState persistence key (+ if P.participating partition+ then Just (P.getPowerState partition)+ else Nothing+ ))+ lift $ put ns {+ partitions = if P.complete partition+ then+ {-+ Remove the partition from the working cache because there+ is no remaining work that needs to be done to propagage+ its changes.+ -}+ Map.delete key partitions+ else+ Map.insert key partition partitions } -{- |- The monad in which the internals of the state machine run. This is really- just a conduit, but we wrap it because we only want to allow `yield`, which- we have re-named `emit`.--}-newtype StateMT i o s m r = StateMT {- unStateMT ::- StateT- (NodeState i o s)- (ConduitM (LInput i o s) (LOutput i o s) m)- r- } deriving (- Functor, Applicative, Monad, MonadLogger, MonadCatch,- MonadThrow, MonadIO- )-{-- We can lift things from the underlying monad straight to 'StateT',- bypassing the `CondutM` layer.--}-instance MonadTrans (StateMT i o s) where- lift = StateMT . lift . lift---{- |- The state machine monad, in LIO.--}-type StateM i o s r = StateMT i o s LIO r---{- |- Run the state machine monad, starting with the initial node state.--}-runStateMT- :: NodeState i o s- -> StateMT i o s m ()- -> ConduitM (LInput i o s) (LOutput i o s) m (NodeState i o s)-runStateMT ns = fmap snd . (`runStateT` ns) . unStateMT---{- |- Emit some output from the state machine.--}-emit :: LOutput i o s -> StateM i o s ()-emit = StateMT . lift . yield---{- |- Get the node State.--}-getS :: StateMT i o s m (NodeState i o s)-getS = StateMT get+{- | Borrowed from 'lens', like @flip fmap@. -}+(<&>) :: (Functor f) => f a -> (a -> b) -> f b+(<&>) = flip fmap -{- |- Put the node state.--}-putS :: NodeState i o s -> StateMT i o s m ()-putS = StateMT . put+{- | Lift from two levels down in a monad transformation stack. -}+lift2+ :: (+ MonadTrans a,+ MonadTrans b,+ Monad m,+ Monad (b m)+ )+ => m r+ -> a (b m) r+lift2 = lift . lift -{- |- Modify the node state.--}-modifyS :: (NodeState i o s -> NodeState i o s) -> StateMT i o s m ()-modifyS f = putS . f =<< getS+{- | Lift from three levels down in a monad transformation stack. -}+lift3+ :: (+ MonadTrans a,+ MonadTrans b,+ MonadTrans c,+ Monad m,+ Monad (c m),+ Monad (b (c m))+ )+ => m r+ -> a (b (c m)) r+lift3 = lift . lift . lift