legion 0.9.0.2 → 0.10.0.0
raw patch · 12 files changed
+1883/−1841 lines, 12 filesdep +stm
Dependencies added: stm
Files
- legion.cabal +5/−3
- src/Network/Legion.hs +4/−2
- src/Network/Legion/Admin.hs +62/−88
- src/Network/Legion/Fork.hs +46/−17
- src/Network/Legion/Launch.hs +41/−0
- src/Network/Legion/Runtime.hs +529/−892
- src/Network/Legion/Runtime/ConnectionManager.hs +139/−73
- src/Network/Legion/Runtime/PeerMessage.hs +3/−4
- src/Network/Legion/Runtime/State.hs +1035/−0
- src/Network/Legion/SocketUtil.hs +19/−0
- src/Network/Legion/StateMachine.hs +0/−597
- src/Network/Legion/StateMachine/Monad.hs +0/−165
legion.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: legion-version: 0.9.0.2+version: 0.10.0.0 synopsis: Distributed, stateful, homogeneous microservice framework. description: Legion is a framework for writing distributed, homogeneous, stateful microservices in Haskell.@@ -37,6 +37,7 @@ Network.Legion.Index Network.Legion.KeySet Network.Legion.LIO+ Network.Legion.Launch Network.Legion.Lift Network.Legion.PartitionKey Network.Legion.PartitionState@@ -45,9 +46,9 @@ Network.Legion.Runtime Network.Legion.Runtime.ConnectionManager Network.Legion.Runtime.PeerMessage+ Network.Legion.Runtime.State Network.Legion.Settings- Network.Legion.StateMachine- Network.Legion.StateMachine.Monad+ Network.Legion.SocketUtil Network.Legion.UUID Paths_legion -- other-extensions:@@ -70,6 +71,7 @@ network >= 2.6.2.1 && < 2.7, scotty >= 0.11.0 && < 0.12, scotty-resource >= 0.1 && < 0.3,+ stm >= 2.4.4.1 && < 2.5, text >= 1.2.2.0 && < 1.3, time >= 1.6.0.1 && < 1.7, transformers >= 0.3.0.0 && < 0.6,
src/Network/Legion.hs view
@@ -84,11 +84,13 @@ import Network.Legion.Index (Tag(Tag, unTag), IndexRecord(IndexRecord, irTag, irKey), SearchTag(SearchTag, stTag, stKey), Indexable(indexEntries))+import Network.Legion.Launch (forkLegionary) import Network.Legion.PartitionKey (PartitionKey(K, unKey)) import Network.Legion.PartitionState (PartitionPowerState) import Network.Legion.PowerState (Event(apply))-import Network.Legion.Runtime (StartupMode(NewCluster, JoinCluster, Recover),- forkLegionary, Runtime, makeRequest, search)+import Network.Legion.Runtime (Runtime, makeRequest, search)+import Network.Legion.Runtime.State (StartupMode(NewCluster, JoinCluster,+ Recover)) import Network.Legion.Settings (RuntimeSettings(RuntimeSettings, adminHost, adminPort, peerBindAddr, joinBindAddr))
src/Network/Legion/Admin.hs view
@@ -2,38 +2,31 @@ {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-}-{- |- This module contains the admin interface code.--}++{- | This module contains the admin interface code. -} module Network.Legion.Admin ( runAdmin,- AdminMessage(..),+ forkAdmin, ) where + import Canteven.HTTP (requestLogging, logExceptionsAndContinue)-import Control.Concurrent (forkIO, newChan, newEmptyMVar, writeChan,- putMVar, takeMVar, Chan)-import Control.Monad (void)-import Control.Monad.Logger (askLoggerIO, runLoggingT, logDebug)+import Control.Monad.IO.Class (liftIO)+import Control.Monad.Logger (askLoggerIO, runLoggingT, logDebug,+ MonadLoggerIO) import Control.Monad.Trans.Class (lift)-import Data.Conduit (Source) import Data.Default.Class (def)-import Data.Map (Map)-import Data.Set (Set) import Data.Text.Encoding (encodeUtf8) import Data.Text.Lazy (Text)-import Data.Time (UTCTime) import Data.Version (showVersion) import Network.HTTP.Types (notFound404) import Network.Legion.Application (LegionConstraints)-import Network.Legion.Conduit (chanToSource)-import Network.Legion.Distribution (Peer)-import Network.Legion.Index (IndexRecord)+import Network.Legion.Fork (forkC) import Network.Legion.LIO (LIO)-import Network.Legion.Lift (lift2) import Network.Legion.PartitionKey (PartitionKey(K), unKey)-import Network.Legion.PartitionState (PartitionPowerState)-import Network.Legion.StateMachine.Monad (NodeState)+import Network.Legion.Runtime (Runtime, debugRuntimeState, getDivergent,+ debugLocalPartitions, debugPartition, eject, debugIndex)+import Network.Legion.Settings (RuntimeSettings, adminPort, adminHost) import Network.Wai (Middleware, modifyResponse) import Network.Wai.Handler.Warp (HostPreference, defaultSettings, Port, setHost, setPort)@@ -42,63 +35,64 @@ import Paths_legion (version) import Text.Read (readMaybe) import Web.Scotty.Resource.Trans (resource, get, delete)-import Web.Scotty.Trans (Options, scottyOptsT, settings, ScottyT, ActionT,- param, middleware, status, json)+import Web.Scotty.Trans (Options, scottyOptsT, settings, ScottyT, param,+ middleware, status, json, text) import qualified Data.Map as Map import qualified Data.Text as T+import qualified Data.Text.Lazy as TL -{- |- Start the admin service in a background thread.--}-runAdmin :: (LegionConstraints e o s)+{- | Fork the admin website in a background thread. -}+forkAdmin :: (LegionConstraints e o s, MonadLoggerIO m)+ => RuntimeSettings+ {- ^ Settings and configuration of the legion framework. -}+ -> Runtime e o s+ -> m ()+forkAdmin legionSettings runtime = do+ logging <- askLoggerIO+ liftIO . (`runLoggingT` logging) . forkC "admin thread" $+ runAdmin (adminPort legionSettings) (adminHost legionSettings) runtime+++{- | Run the admin service. Does not return. -}+runAdmin :: (LegionConstraints e o s, MonadLoggerIO io) => Port -> HostPreference- -> LIO (Source LIO (AdminMessage e o s))-runAdmin addr host = do- logging <- askLoggerIO- chan <- lift newChan- void . lift . forkIO . (`runLoggingT` logging) $- let- website :: ScottyT Text LIO ()- website = do- middleware- $ requestLogging logging- . setServer- . logExceptionsAndContinue logging+ -> Runtime e o s+ -> io ()+runAdmin addr host runtime = do+ logging <- askLoggerIO+ let+ website :: ScottyT Text LIO ()+ website = do+ middleware+ $ requestLogging logging+ . setServer+ . logExceptionsAndContinue logging - resource "/clusterstate" $- get $ json =<< send chan GetState- resource "/index" $- get $ json =<< send chan GetIndex- resource "/divergent" $- get $- json . Map.mapKeys show =<< send chan GetDivergent- resource "/partitions" $- get $- json . Map.mapKeys (show . toInteger . unKey) =<< send chan GetStates- resource "/partitions/:key" $- get $ do- key <- K . read <$> param "key"- json =<< send chan (GetPart key)- resource "/peers/:peer" $- delete $- readMaybe <$> param "peer" >>= \case- Nothing -> status notFound404- Just peer -> do- lift . $(logDebug) . T.pack $ "Ejecting peer: " ++ show peer- send chan (Eject peer)+ resource "/rts" $+ get $ text . TL.pack . show =<< debugRuntimeState runtime+ resource "/index" $+ get $ json =<< debugIndex runtime+ resource "/divergent" $+ get $+ json . Map.mapKeys show =<< getDivergent runtime+ resource "/partitions" $+ get $+ json . Map.mapKeys (show . toInteger . unKey)+ =<< debugLocalPartitions runtime+ resource "/partitions/:key" $+ get $ do+ key <- K . read <$> param "key"+ json =<< debugPartition runtime key+ resource "/peers/:peer" $+ delete $+ readMaybe <$> param "peer" >>= \case+ Nothing -> status notFound404+ Just peer -> do+ lift . $(logDebug) . T.pack $ "Ejecting peer: " ++ show peer+ eject runtime peer - in scottyOptsT (options addr host) (`runLoggingT` logging) website- return (chanToSource chan)- where- send- :: Chan (AdminMessage e o s)- -> ((a -> LIO ()) -> AdminMessage e o s)- -> ActionT Text LIO a- send chan msg = lift2 $ do- mvar <- newEmptyMVar- writeChan chan (msg (lift . putMVar mvar))- takeMVar mvar+ scottyOptsT (options addr host) (`runLoggingT` logging) website {- | Build some warp settings based on the configured socket address. -}@@ -125,25 +119,5 @@ {- | The value of the @Server:@ header. -} serverValue = encodeUtf8 (T.pack ("legion-admin/" ++ showVersion version))---{- |- The type of messages sent by the admin service.--}-data AdminMessage e o s- = GetState (NodeState e o s -> LIO ())- | GetPart PartitionKey (PartitionPowerState e o s -> LIO ())- | Eject Peer (() -> LIO ())- | GetIndex (Set IndexRecord -> LIO ())- | GetDivergent (Map Peer (Maybe UTCTime) -> LIO ())- | GetStates (Map PartitionKey (PartitionPowerState e o s) -> LIO ())--instance Show (AdminMessage e o s) where- show (GetState _) = "(GetState _)"- show (GetPart k _) = "(GetPart " ++ show k ++ " _)"- show (Eject p _) = "(Eject " ++ show p ++ " _)"- show (GetIndex _) = "(GetIndex _)"- show (GetDivergent _) = "(GetDivergent _)"- show (GetStates _) = "(GetStates _)"
src/Network/Legion/Fork.hs view
@@ -1,16 +1,22 @@+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE TemplateHaskell #-}+ {- | This module holds `forkC`, because we use it in at least two other modules. -} module Network.Legion.Fork (- forkC+ forkC,+ forkL,+ ForkM(..), ) where import Control.Concurrent (forkIO)-import Control.Exception (SomeException, try)+import Control.Exception (SomeException) import Control.Monad (void)-import Control.Monad.Logger (logError, askLoggerIO, runLoggingT)-import Control.Monad.Trans.Class (lift)+import Control.Monad.Catch (try, MonadCatch)+import Control.Monad.IO.Class (liftIO)+import Control.Monad.Logger (logError, askLoggerIO, runLoggingT,+ MonadLoggerIO, LoggingT, MonadLoggerIO) import Data.Text (pack) import Network.Legion.LIO (LIO) import System.Exit (ExitCode(ExitFailure))@@ -23,16 +29,15 @@ we should crash the program instead of running in some kind of zombie broken state. -}-forkC- :: String+forkC :: (ForkM m, MonadCatch m, MonadLoggerIO m)+ => String -- ^ The name of the critical thread, used for logging.- -> LIO ()+ -> m () -- ^ The IO to execute.- -> LIO ()-forkC name io = do- logging <- askLoggerIO- lift . void . forkIO $ do- result <- try (runLoggingT io logging)+ -> m ()+forkC name action =+ forkM $ do+ result <- try action case result of Left err -> do let msg =@@ -40,10 +45,34 @@ ++ ". We are crashing the entire program because we can't " ++ "continue without this thread. The error was: " ++ show (err :: SomeException)- -- write the message to every place we can think of.- (`runLoggingT` logging) . $(logError) . pack $ msg- putStrLn msg- hPutStrLn stderr msg- exitImmediately (ExitFailure 1)+ {- write the message to every place we can think of. -}+ $(logError) . pack $ msg+ liftIO (putStrLn msg)+ liftIO (hPutStrLn stderr msg)+ liftIO (exitImmediately (ExitFailure 1)) Right v -> return v+++{- | Fork a normal, noncritical thread in 'MonadLoggerIO'. -}+forkL :: (MonadLoggerIO io)+ => LIO ()+ -> io ()+forkL io = liftIO . void . forkIO . runLoggingT io =<< askLoggerIO+++{- |+ Class of monads that can be forked. I'm sure there is a better solution for+ this, maybe using MonadBaseControl or something. This needs looking into.+-}+class ForkM m where+ forkM :: m () -> m ()++instance ForkM IO where+ forkM = void . forkIO++instance ForkM (LoggingT IO) where+ forkM action = do+ logging <- askLoggerIO+ liftIO . forkM $ runLoggingT action logging+
+ src/Network/Legion/Launch.hs view
@@ -0,0 +1,41 @@+{- |+ This module takes care of launching the Legion runtime system and associated+ admin tools. It mainly exists to avoid a module dependency cycle between the+ runtime stuff and the admin stuff.+-}+module Network.Legion.Launch (+ forkLegionary+) where++import Control.Monad.Logger (MonadLoggerIO)+import Network.Legion.Admin (forkAdmin)+import Network.Legion.Application (LegionConstraints, Persistence)+import Network.Legion.Runtime (forkRuntime, Runtime)+import Network.Legion.Runtime.State (StartupMode)+import Network.Legion.Settings (RuntimeSettings)++{- |+ Forks the legion framework in a background thread, and returns a handle+ on the runtime environment, which can be used to make user requests.++ - @__e__@ is the type of request your application will handle. @__e__@ stands+ for __"event"__.+ - @__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"__.+-}+forkLegionary :: (LegionConstraints e o s, MonadLoggerIO m)+ => Persistence e o s+ {- ^ The persistence layer used to back the legion framework. -}+ -> RuntimeSettings+ {- ^ Settings and configuration of the legion framework. -}+ -> StartupMode+ -> m (Runtime e o s)+forkLegionary persistence settings startupMode = do+ runtime <- forkRuntime persistence settings startupMode+ forkAdmin settings runtime+ return runtime++
src/Network/Legion/Runtime.hs view
@@ -1,895 +1,532 @@ {-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE TemplateHaskell #-}-{- |- This module is responsible for the runtime operation of the legion- framework. This mostly means opening sockets and piping data around to the- various connected pieces.--}-module Network.Legion.Runtime (- forkLegionary,- StartupMode(..),- Runtime,- makeRequest,- search,-) where--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.Catch (catchAll, try, SomeException, throwM)-import Control.Monad.IO.Class (MonadIO, liftIO)-import Control.Monad.Logger (logWarn, logError, logInfo, LoggingT,- MonadLoggerIO, runLoggingT, askLoggerIO, logDebug)-import Control.Monad.Trans.Class (lift)-import Control.Monad.Trans.State (StateT, runStateT, get, put, modify)-import Data.Binary (encode, Binary)-import Data.Conduit (Source, ($$), (=$=), yield, await, awaitForever,- transPipe, ConduitM, runConduit, Sink)-import Data.Conduit.Network (sourceSocket)-import Data.Conduit.Serialization.Binary (conduitDecode)-import Data.Map (Map)-import Data.Set (Set)-import Data.String (IsString, fromString)-import Data.Text (pack)-import Data.Time (UTCTime, getCurrentTime)-import GHC.Generics (Generic)-import Network.Legion.Admin (runAdmin, AdminMessage(GetState, GetPart,- Eject, GetIndex, GetDivergent, GetStates))-import Network.Legion.Application (LegionConstraints, Persistence,- list, saveCluster)-import Network.Legion.BSockAddr (BSockAddr(BSockAddr))-import Network.Legion.ClusterState (ClusterPowerState)-import Network.Legion.Conduit (merge, chanToSink, chanToSource)-import Network.Legion.Distribution (Peer, newPeer)-import Network.Legion.Fork (forkC)-import Network.Legion.Index (IndexRecord(IndexRecord), irTag, irKey,- SearchTag(SearchTag), indexEntries, Indexable)-import Network.Legion.LIO (LIO)-import Network.Legion.Lift (lift2, lift3)-import Network.Legion.PartitionKey (PartitionKey)-import Network.Legion.PartitionState (PartitionPowerState)-import Network.Legion.PowerState (Event)-import Network.Legion.Runtime.ConnectionManager (newConnectionManager,- ConnectionManager, newPeers)-import Network.Legion.Runtime.PeerMessage (PeerMessage(PeerMessage),- PeerMessagePayload(ForwardRequest, ForwardResponse, ClusterMerge,- PartitionMerge, Search, SearchResponse, JoinNext, JoinNextResponse),- MessageId, newSequence, nextMessageId, JoinNextResponse(Joined,- JoinFinished))-import Network.Legion.Settings (RuntimeSettings(RuntimeSettings,- adminHost, adminPort, peerBindAddr, joinBindAddr))-import Network.Legion.StateMachine (partitionMerge, clusterMerge,- newNodeState, UserResponse(Forward, Respond), userRequest, eject,- minimumCompleteServiceSet, joinNext, joinNextResponse)-import Network.Legion.StateMachine.Monad (NodeState, runSM, ClusterAction,- SM, popActions, nsIndex)-import Network.Legion.UUID (getUUID)-import Network.Socket (Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN),- 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 System.IO (stderr, hPutStrLn)-import qualified Data.Conduit.List as CL-import qualified Data.Map as Map-import qualified Data.Set as Set-import qualified Network.Legion.ClusterState as C-import qualified Network.Legion.PowerState as PS-import qualified Network.Legion.Runtime.ConnectionManager as CM-import qualified Network.Legion.StateMachine as SM-import qualified Network.Legion.StateMachine.Monad as SMM---{- |- Run the legion node framework program, with the given user definitions,- framework settings, and request source. This function never returns- (except maybe with an exception if something goes horribly wrong).--}-runLegionary :: (LegionConstraints e o s)- => Persistence e o s- {- ^ The persistence layer used to back the legion framework. -}- -> RuntimeSettings- {- ^ Settings and configuration of the legionframework. -}- -> StartupMode- -> Source IO (RequestMsg e o)- {- ^ A source of requests, together with a way to respond to the requets. -}- -> LoggingT IO ()- {-- Don't expose 'LIO' here because 'LIO' is a strictly internal- symbol. 'LoggingT IO' is what we expose to the world.- -}--runLegionary- persistence- settings@RuntimeSettings {adminHost, adminPort}- startupMode- requestSource- = do- {- Start the various messages sources. -}- peerS <- loggingC =<< startPeerListener settings- adminS <- loggingC =<< runAdmin adminPort adminHost- joinS <- loggingC (joinMsgSource settings)-- (self, nodeState, peers) <- makeNodeState persistence settings startupMode- rts <- newRuntimeState self peers- let- messageSource = transPipe lift (- (joinS =$= CL.map J) `merge`- (peerS =$= CL.map P) `merge`- (requestSource =$= CL.map R) `merge`- (adminS =$= CL.map A)- )- void . runRTS persistence nodeState rts . runConduit $- messageSource- =$= messageSink- where- newRuntimeState :: (Binary e, Binary o, Binary s)- => Peer- -> Map Peer BSockAddr- -> LoggingT IO (RuntimeState e o s)- newRuntimeState self peers = do- cm <- newConnectionManager peers- firstMessageId <- newSequence- return RuntimeState {- forwarded = Map.empty,- nextId = firstMessageId,- cm,- self,- commClock = Map.empty,- searches = Map.empty- }-- {- |- Turn an LIO-based conduit into an IO-based conduit, so that it- will work with `merge`.- -}- loggingC :: ConduitM e o LIO r -> LIO (ConduitM e o IO r)- loggingC c = do- logging <- askLoggerIO- return (transPipe (`runLoggingT` logging) c)---{- |- This is how requests are packaged when they are sent to the legion framework- for handling. It includes the request information itself, a partition key to- which the request is directed, and a way for the framework to deliver the- response to some interested party.--}-data RequestMsg e o- = Request PartitionKey e (o -> IO ())- | SearchDispatch SearchTag (Maybe IndexRecord -> IO ())-instance (Show e) => Show (RequestMsg e o) where- show (Request k e _) = "(Request " ++ show k ++ " " ++ show e ++ " _)"- show (SearchDispatch s _) = "(SearchDispatch " ++ show s ++ " _)"---messageSink :: (LegionConstraints e o s)- => Sink (RuntimeMessage e o s) (RTS e o s) ()-messageSink = awaitForever (\msg -> do- $(logDebug) . pack $ "Receieved: " ++ show msg- lift $ do- case msg of- P (PeerMessage source _ _) ->- updateRecvClock source- _ -> return ()- handleMessage msg- updatePeers- clusterActions- )---{- | Make progress on outstanding cluster actions. -}-clusterActions :: RTS e o s ()-clusterActions =- mapM_ clusterAction =<< popActions- where- {- |- Actually perform a cluster action as directed by the state- machine.- -}- clusterAction- :: ClusterAction e o s- -> RTS e o s ()-- clusterAction (SMM.ClusterMerge peer ps) =- void $ send peer (ClusterMerge ps)-- clusterAction (SMM.PartitionMerge peer key ps) =- void $ send peer (PartitionMerge key ps)-- clusterAction (SMM.PartitionJoin peer keys) =- void $ send peer (JoinNext keys)---{- |- Make sure the connection manager knows about any new peers that have- joined the cluster.--}-updatePeers :: RTS e o s ()-updatePeers = do- peers <- SM.getPeers- RuntimeState {cm} <- lift get- lift2 $ newPeers cm peers---{- |- 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 e o s)- => RuntimeMessage e o s- -> RTS e o s ()--handleMessage {- Join Next Response -}- (P (PeerMessage source _ (JoinNextResponse _messageId response)))- =- joinNextResponse source (toMaybe response)- where- toMaybe- :: JoinNextResponse e o s- -> Maybe (PartitionKey, PartitionPowerState e o s)- toMaybe (Joined key partition) = Just (key, partition)- toMaybe JoinFinished = Nothing--handleMessage {- Join Next -}- (P (PeerMessage source messageId (JoinNext askKeys)))- =- joinNext source askKeys >>= \case- Nothing -> void $- send source (JoinNextResponse messageId JoinFinished)- Just (gotKey, partition) -> void $- send source (JoinNextResponse messageId (Joined gotKey partition))--handleMessage {- Partition Merge -}- (P (PeerMessage _ _ (PartitionMerge key ps)))- =- partitionMerge key ps--handleMessage {- Cluster Merge -}- (P (PeerMessage _ _ (ClusterMerge cs)))- =- clusterMerge cs--handleMessage {- Forward Request -}- (P (msg@(PeerMessage source mid (ForwardRequest key request))))- = do- output <- userRequest key request- case output of- Respond response -> void $ send source (ForwardResponse mid response)- Forward peer -> forward peer msg--handleMessage {- Forward Response -}- (msg@(P (PeerMessage _ _ (ForwardResponse mid response))))- = do- rts <- lift get- case lookupDelete mid (forwarded rts) of- (Nothing, fwd) -> do- $(logWarn) . pack $ "Unsolicited ForwardResponse: " ++ show msg- (lift . put) rts {forwarded = fwd}- (Just respond, fwd) -> do- lift2 $ respond response- (lift . put) rts {forwarded = fwd}--handleMessage {- User Request -}- (R (Request key request respond))- = do- output <- userRequest key request- case output of- Respond response -> lift3 (respond response)- Forward peer -> do- messageId <- send peer (ForwardRequest key request)- (lift . modify) $ \rts@RuntimeState {forwarded} -> rts {- forwarded = Map.insert messageId (lift . respond) forwarded- }--handleMessage {- Search Dispatch -}- {-- This is where we send out search request to all the appropriate- nodes in the cluster.- -}- (R (SearchDispatch searchTag respond))- =- Map.lookup searchTag . searches <$> lift get >>= \case- Nothing -> do- {-- No identical search is currently being executed, kick off a- new one.- -}- mcss <- minimumCompleteServiceSet- mapM_ sendOne (Set.toList mcss)- rts@RuntimeState {searches} <- lift get- (lift . put) rts {- searches = Map.insert- searchTag- (mcss, Nothing, [lift . respond])- searches- }- Just (peers, best, responders) -> do- {-- A search for this tag is already in progress, just add the- responder to the responder list.- -}- rts@RuntimeState {searches} <- lift get- (lift . put) rts {- searches = Map.insert- searchTag- (peers, best, (lift . respond):responders)- searches- }- where- sendOne :: Peer -> RTS e o s ()- sendOne peer =- void $ send peer (Search searchTag)--handleMessage {- Search Execution -}- {- This is where we handle local search execution. -}- (P (PeerMessage source _ (Search searchTag)))- = do- output <- SM.search searchTag - void $ send source (SearchResponse searchTag output)--handleMessage {- Search Response -}- {-- This is where we gather all the responses from the various peers- to which we dispatched search requests.- -}- (msg@(P (PeerMessage source _ (SearchResponse searchTag response))))- =- {- TODO: see if this function can't be made more elegant. -}- Map.lookup searchTag . searches <$> lift get >>= \case- Nothing ->- {- There is no search happening. -}- $(logWarn) . pack $ "Unsolicited SearchResponse: " ++ show msg- Just (peers, best, responders) ->- if source `Set.member` peers- then- let peers2 = Set.delete source peers- in if null peers2- then do- {-- All peers have responded, go ahead and respond to- the client.- -}- lift2 $ mapM_ ($ bestOf best response) responders- rts@RuntimeState {searches} <- lift get- (lift . put) rts {searches = Map.delete searchTag searches}- else do- {- We are still waiting on some outstanding requests. -}- rts@RuntimeState {searches} <- lift get- (lift . put) rts {- searches = Map.insert- searchTag- (peers2, bestOf best response, responders)- searches- }- else- {-- There is a search happening, but the peer that responded- is not part of it.- -}- $(logWarn) . pack $ "Unsolicited SearchResponse: " ++ show msg- where- {- |- Figure out which index record returned to us by the various peers- is the most appropriate to return to the user. This is mostly like- 'min' but we can't use 'min' (or fancy applicative formulations)- because we want to favor 'Just' instead of 'Nothing'.- -}- bestOf :: Maybe IndexRecord -> Maybe IndexRecord -> Maybe IndexRecord- bestOf (Just a) (Just b) = Just (min a b)- bestOf Nothing b = b- bestOf a Nothing = a--handleMessage {- Join Request -}- (J (JoinRequest addy, respond))- = do- (peer, cluster) <- SM.join addy- lift2 $ respond (JoinOk peer cluster)--handleMessage {- Admin Get State -}- (A (GetState respond))- = - lift2 . respond =<< SMM.getNodeState--handleMessage {- Admin Get Partition -}- (A (GetPart key respond))- =- lift2 . respond =<< SM.getPartition key--handleMessage {- Admin Eject Peer -}- (A (Eject peer respond))- = do- eject peer- lift2 $ respond ()--handleMessage {- Admin Get Index -}- (A (GetIndex respond))- =- lift2 . respond =<< SMM.nsIndex <$> SMM.getNodeState--handleMessage {- Admin Get Divergent -}- (A (GetDivergent respond))- = do- RuntimeState {commClock} <- lift get- diverging <- divergentPeers . SMM.partitions <$> SMM.getNodeState- lift2 . respond $ Map.fromAscList [- (peer, r)- | (peer, (_, r)) <- Map.toAscList commClock- , peer `Set.member` diverging- ]- where- divergentPeers :: Map PartitionKey (PartitionPowerState e o s) -> Set Peer- divergentPeers =- foldr Set.union Set.empty . fmap (PS.divergent . snd) . Map.toList--handleMessage {- Admin Get States -}- (A (GetStates respond))- = do- persistence <- SMM.getPersistence- lift2 . respond . Map.fromList =<< runConduit (- transPipe liftIO (list persistence)- =$= CL.consume- )---{- | This defines the various ways a node can be spun up. -}-data StartupMode- = NewCluster- {- ^ Indicates that we should bootstrap a new cluster at startup. -}- | JoinCluster SockAddr- {- ^ Indicates that the node should try to join an existing cluster. -}- | Recover Peer ClusterPowerState- {- ^- Recover from a crash as the given peer, using the given cluster- state.- -}- deriving (Show, Eq)---{- |- Construct a source of incoming peer messages. We have to start the- peer listener first before we spin up the cluster management, which- is why this is an @LIO (Source LIO PeerMessage)@ instead of a- @Source LIO PeerMessage@.--}-startPeerListener :: (LegionConstraints e o s)- => RuntimeSettings- -> LIO (Source LIO (PeerMessage e o s))--startPeerListener RuntimeSettings {peerBindAddr} =- catchAll (do- (inputChan, so) <- lift $ do- inputChan <- newChan- so <- socket (fam peerBindAddr) Stream defaultProtocol- setSocketOption so ReuseAddr 1- bind so peerBindAddr- listen so 5- return (inputChan, so)- forkC "peer socket acceptor" $ acceptLoop so inputChan- return (chanToSource inputChan)- ) (\err -> do- $(logError) . pack- $ "Couldn't start incomming peer message service, because of: "- ++ show (err :: SomeException)- throwM err- )- where- acceptLoop :: (LegionConstraints e o s)- => Socket- -> Chan (PeerMessage e o s)- -> LIO ()- acceptLoop so inputChan =- catchAll (- forever $ do- (conn, _) <- lift $ accept so- remoteAddr <- lift $ getPeerName conn- logging <- askLoggerIO- let runSocket =- sourceSocket conn- =$= conduitDecode- $$ msgSink- void- . lift- . forkIO- . (`runLoggingT` logging)- . logErrors remoteAddr- $ runSocket- ) (\err -> do- $(logError) . pack- $ "error in peer message accept loop: "- ++ show (err :: SomeException)- throwM err- )- where- msgSink = chanToSink inputChan-- logErrors :: SockAddr -> LIO () -> LIO ()- logErrors remoteAddr io = do- result <- try io- case result of- Left err ->- $(logWarn) . pack- $ "Incomming peer connection (" ++ show remoteAddr- ++ ") crashed because of: " ++ show (err :: SomeException)- Right v -> return v---{- | Figure out how to construct the initial node state. -}-makeNodeState :: (Event e o s, Indexable s)- => Persistence e o s- -> RuntimeSettings- -> StartupMode- -> LIO (Peer, NodeState e o s, Map Peer BSockAddr)--makeNodeState- persistence- settings@RuntimeSettings {peerBindAddr}- NewCluster- = do- {- Build a brand new node state, for the first node in a cluster. -}- verifyClearPersistence persistence- self <- newPeer- clusterId <- getUUID- let- cluster = C.new clusterId self peerBindAddr- makeNodeState persistence settings (Recover self cluster)--makeNodeState- persistence- settings@RuntimeSettings {peerBindAddr}- (JoinCluster addr)- = do- {-- Join a cluster by either starting fresh, or recovering from a- shutdown or crash.- -}- verifyClearPersistence persistence- $(logInfo) "Trying to join an existing cluster."- (self, cluster) <- joinCluster (JoinRequest (BSockAddr peerBindAddr))- makeNodeState persistence settings (Recover self cluster)- where- joinCluster :: JoinRequest -> LIO (Peer, ClusterPowerState)- joinCluster joinMsg = liftIO $ do- so <- socket (fam addr) Stream defaultProtocol- connect so addr- sendAll so (encode joinMsg)- {-- using sourceSocket and conduitDecode is easier than building- a recive/decode state loop, even though we only read a single- response.- -}- sourceSocket so =$= conduitDecode $$ do- response <- await- case response of- Nothing -> fail- $ "Couldn't join a cluster because there was no response "- ++ "to our join request!"- Just (JoinOk self cps) ->- return (self, cps)--makeNodeState persistence _ (Recover self cluster) = do- {- Make sure to rebuild the index in the case of recovery. -}- index <- runConduit . transPipe liftIO $- list persistence- =$= CL.fold addIndexRecords Set.empty- let- nodeState = (newNodeState self cluster) {nsIndex = index}- liftIO $ saveCluster persistence self cluster- return (self, nodeState, C.getPeers cluster)- where- addIndexRecords :: (Indexable s, Event e o s)- => Set IndexRecord- -> (PartitionKey, PartitionPowerState e o s)- -> Set IndexRecord- addIndexRecords index (key, partition) =- let- newRecords =- Set.map- (`IndexRecord` key)- (indexEntries (PS.projectedValue partition))- in Set.union index newRecords---{- |- Helper for 'makeNodeState'. Verify that there is nothing in the- persistence layer.--}-verifyClearPersistence :: (MonadLoggerIO io) => Persistence e o s -> io ()-verifyClearPersistence persistence = - liftIO (runConduit (list persistence =$= CL.head)) >>= \case- Just _ -> do- let- msg :: (IsString a) => a- msg = fromString- $ "We are trying to start up a new peer, but the persistence "- ++ "layer already has data in it. This is an invalid state. "- ++ "New nodes must be started from a totally clean, empty state."- $(logError) msg- liftIO $ do- hPutStrLn stderr msg- putStrLn msg- error msg- Nothing ->- return ()---{- | A source of cluster join request messages. -}-joinMsgSource- :: RuntimeSettings- -> Source LIO (JoinRequest, JoinResponse -> LIO ())--joinMsgSource RuntimeSettings {joinBindAddr} = join . lift $- catchAll (do- (chan, so) <- lift $ do- chan <- newChan- so <- socket (fam joinBindAddr) Stream defaultProtocol- setSocketOption so ReuseAddr 1- bind so joinBindAddr- listen so 5- return (chan, so)- forkC "join socket acceptor" $ acceptLoop so chan- return (chanToSource chan)- ) (\err -> do- $(logError) . pack- $ "Couldn't start join request service, because of: "- ++ show (err :: SomeException)- throwM err- )- where- acceptLoop :: Socket -> Chan (JoinRequest, JoinResponse -> LIO ()) -> LIO ()- acceptLoop so chan =- catchAll (- forever $ do- (conn, _) <- lift $ accept so- logging <- askLoggerIO- (void . lift . forkIO . (`runLoggingT` logging) . logErrors) (- sourceSocket conn- =$= conduitDecode- =$= attachResponder conn- $$ chanToSink chan- )- ) (\err -> do- $(logError) . pack- $ "error in join request accept loop: "- ++ show (err :: SomeException)- throwM err- )- where- logErrors :: LIO () -> LIO ()- logErrors io = do- result <- try io- case result of- Left err ->- $(logWarn) . pack- $ "Incomming join connection crashed because of: "- ++ show (err :: SomeException)- Right v -> return v-- attachResponder- :: Socket- -> ConduitM JoinRequest (JoinRequest, JoinResponse -> LIO ()) LIO ()- attachResponder conn = awaitForever (\msg -> do- mvar <- liftIO newEmptyMVar- yield (msg, lift . putMVar mvar)- response <- liftIO $ takeMVar mvar- liftIO $ sendAll conn (encode response)- )---{- | 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---{- |- 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.-- - @__e__@ is the type of request your application will handle. @__e__@ stands- for __"event"__.- - @__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"__.--}-forkLegionary :: (LegionConstraints e o s, MonadLoggerIO io)- => Persistence e o s- {- ^ The persistence layer used to back the legion framework. -}- -> RuntimeSettings- {- ^ Settings and configuration of the legion framework. -}- -> StartupMode- -> io (Runtime e o)--forkLegionary persistence settings startupMode = do- logging <- askLoggerIO- liftIO . (`runLoggingT` logging) $ do- chan <- liftIO newChan- forkC "main legion thread" $- runLegionary persistence settings startupMode (chanToSource chan)- return Runtime {- rtMakeRequest = \key request -> liftIO $ do- responseVar <- newEmptyMVar- writeChan chan (Request key request (putMVar responseVar))- takeMVar responseVar,- rtSearch =- let- findNext :: SearchTag -> IO (Maybe IndexRecord)- findNext searchTag = do- responseVar <- newEmptyMVar- writeChan chan (SearchDispatch searchTag (putMVar responseVar))- takeMVar responseVar- in findNext-- }---{- |- This type represents a handle to the runtime environment of your- Legion application. This allows you to make requests and access the- partition index.-- 'Runtime' is an opaque structure. Use 'makeRequest' and 'search' to- access it.--}-data Runtime e o = Runtime {- {- |- Send an application request to the legion runtime, and get back- a response.- -}- rtMakeRequest :: PartitionKey -> e -> IO o,-- {- | Query the index to find a set of partition keys. -}- rtSearch :: SearchTag -> IO (Maybe IndexRecord)- }---{- | Send a user request to the legion runtime. -}-makeRequest :: (MonadIO io) => Runtime e o -> PartitionKey -> e -> io o-makeRequest rt key = liftIO . rtMakeRequest rt key---{- |- Send a search request to the legion runtime. Returns results that are- __strictly greater than__ the provided 'SearchTag'.--}-search :: (MonadIO io) => Runtime e o -> SearchTag -> Source io IndexRecord-search rt tag =- liftIO (rtSearch rt tag) >>= \case- Nothing -> return ()- Just record@IndexRecord {irTag, irKey} -> do- yield record- search rt (SearchTag irTag (Just irKey))---{- | This is the type of message passed around in the runtime. -}-data RuntimeMessage e o s- = P (PeerMessage e o s)- | R (RequestMsg e o)- | J (JoinRequest, JoinResponse -> LIO ())- | A (AdminMessage e o s)-instance (Show e, Show o, Show s) => Show (RuntimeMessage e o s) where- show (P m) = "(P " ++ show m ++ ")"- show (R m) = "(R " ++ show m ++ ")"- show (J (jr, _)) = "(J (" ++ show jr ++ ", _))"- show (A a) = "(A (" ++ show a ++ "))"---{- |- The runtime state.-- The 'searches' field is a little weird.-- It turns out that searches are deterministic over the parameters of- 'SearchTag' and cluster state. This should make sense, because everything in- Haskell is deterministic given __all__ the parameters. Since the cluster- state only changes over time, searches that happen "at the same time" and- for the same 'SearchTag' can be considered identical. I don't think it is too- much of a stretch to say that searches that have overlapping execution times- can be considered to be happening "at the same time", therefore the- search tag becomes determining factor in the result of the search.-- This is a long-winded way of justifying the fact that, if we are currently- executing a search and an identical search requests arrives, then the second- identical search is just piggy-backed on the results of the currently- executing search. Whether this counts as a premature optimization hack or a- beautifully elegant expression of platonic reality is left as an exercise for- the reader. It does help simplify the code a little bit because we don't have- to specify some kind of UUID to differentiate otherwise identical searches.--}-data RuntimeState e o s = RuntimeState {- self :: Peer,- forwarded :: Map MessageId (o -> LIO ()),- nextId :: MessageId,- cm :: ConnectionManager e o s,- commClock :: Map Peer (Maybe UTCTime, Maybe UTCTime),- {- ^ When did we last communicate with a peer. (sent, recv). -}- searches :: Map- SearchTag- (Set Peer, Maybe IndexRecord, [Maybe IndexRecord -> LIO ()])- }---{- | This is the type of a join request message. -}-newtype JoinRequest = JoinRequest BSockAddr- deriving (Generic, Show)-instance Binary JoinRequest---{- | The response to a JoinRequst message -}-data JoinResponse- = JoinOk Peer ClusterPowerState- 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))---{- | The runtime monad. -}-type RTS e o s =- SM e o s (- StateT (RuntimeState e o s)- LIO)---{- | Shorthand for running the RTS monad. -}-runRTS- :: Persistence e o s- -> NodeState e o s- -> RuntimeState e o s- -> RTS e o s a- -> LIO (a, NodeState e o s, [ClusterAction e o s], RuntimeState e o s)-runRTS persistence ns rts =- fmap flatten- . (`runStateT` rts)- . runSM persistence ns- where- flatten ((a, b, c), d) = (a, b, c, d)---{- |- Send a peer message in the RTS monad, automatically taking care of- necessary state updates.--}-send :: Peer -> PeerMessagePayload e o s -> RTS e o s MessageId-send target payload = do- rts@RuntimeState {cm, self, nextId} <- lift get- (lift . put) rts {nextId = nextMessageId nextId}- lift2 $ CM.send cm target (PeerMessage self nextId payload)- return nextId---{- | Forward an existing message to another peer. -}-forward :: Peer -> PeerMessage e o s -> RTS e o s ()-forward target message = do- RuntimeState {cm} <- lift get- lift2 $ CM.send cm target message---{- | Update the time when we last received a message from a peer. -}-updateRecvClock :: Peer -> RTS e o s ()-updateRecvClock peer = do- now <- liftIO getCurrentTime- (lift . modify) (\rts@RuntimeState {commClock} ->- let- newCommClock = case Map.lookup peer commClock of- Nothing -> Map.insert peer (Nothing, Just now) commClock- Just (s, _) -> Map.insert peer (s, Just now) commClock- in newCommClock `seq` rts {- commClock = newCommClock- }- )+{-# LANGUAGE ExplicitForAll #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}++module Network.Legion.Runtime (+ -- * User-facing interface.+ forkRuntime,+ makeRequest,+ search,+ Runtime,++ -- * Internal interface.+ eject,+ getDivergent,++ -- * Debugging interface.+ debugLocalPartitions,+ debugRuntimeState,+ debugPartition,+ debugIndex,++) where+++import Control.Concurrent (writeChan, newChan, Chan)+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)+import Control.Monad (void, forever)+import Control.Monad.Catch (catchAll, try, SomeException, throwM,+ MonadCatch)+import Control.Monad.IO.Class (liftIO, MonadIO)+import Control.Monad.Logger (MonadLoggerIO, logError, logWarn, logDebug,+ askLoggerIO, runLoggingT)+import Control.Monad.Trans.Class (lift)+import Data.Aeson (Value)+import Data.Binary (encode)+import Data.Conduit ((.|), runConduit, awaitForever, Source, yield)+import Data.Conduit.Network (sourceSocket)+import Data.Conduit.Serialization.Binary (conduitDecode)+import Data.Map (Map)+import Data.Set (Set)+import Data.Time (UTCTime)+import Network.Legion.Application (LegionConstraints, Persistence)+import Network.Legion.Conduit (chanToSource)+import Network.Legion.Distribution (Peer)+import Network.Legion.Fork (forkC, forkL, ForkM)+import Network.Legion.Index (IndexRecord(IndexRecord),+ SearchTag(SearchTag), irTag, irKey)+import Network.Legion.LIO (LIO)+import Network.Legion.PartitionKey (PartitionKey)+import Network.Legion.PartitionState (PartitionPowerState)+import Network.Legion.Runtime.ConnectionManager (send)+import Network.Legion.Runtime.PeerMessage (PeerMessage(PeerMessage),+ PeerMessagePayload(PartitionMerge, ForwardRequest, ForwardResponse,+ ClusterMerge, Search, SearchResponse, JoinNext, JoinNextResponse),+ payload, source, messageId, JoinNextResponse(JoinFinished, Joined))+import Network.Legion.Runtime.State (makeRuntimeState, StartupMode,+ RuntimeT, JoinRequest(JoinRequest), JoinResponse(JoinOk), runRuntimeT,+ updateRecvClock, userRequest, forwardResponse, clusterMerge,+ getCM, searchResponse, joinNext, partitionMerge, joinNextResponse,+ forwardedRequest, searchDispatch)+import Network.Legion.Settings (RuntimeSettings(RuntimeSettings),+ peerBindAddr, joinBindAddr)+import Network.Legion.SocketUtil (fam)+import Network.Socket (SocketOption(ReuseAddr), SocketType(Stream),+ accept, bind, defaultProtocol, listen, setSocketOption, socket,+ SockAddr, getPeerName, Socket)+import Network.Socket.ByteString.Lazy (sendAll)+import qualified Data.Text as T+import qualified Network.Legion.Runtime.State as S+++{- |+ This type represents a handle to the runtime environment of your+ Legion application. This allows you to make requests and access the+ partition index.++ 'Runtime' is an opaque structure. Use 'makeRequest' and 'search' to+ access it.+-}+newtype Runtime e o s = Runtime {+ unRuntime :: Chan (RuntimeMessage e o s)+ }+++{- | Fork the runtime in a background thread. -}+forkRuntime :: (LegionConstraints e o s, MonadLoggerIO m)+ => Persistence e o s+ {- ^ The persistence layer used to back the legion framework. -}+ -> RuntimeSettings+ {- ^ Settings and configuration of the legion framework. -}+ -> StartupMode+ -> m (Runtime e o s)+forkRuntime persistence settings startupMode = do+ runtime <- Runtime <$> liftIO newChan+ logging <- askLoggerIO+ liftIO . (`runLoggingT` logging) . forkC "main legion thread" $+ executeRuntime persistence settings startupMode runtime+ return runtime+++{- | Send a user request to the legion runtime. -}+makeRequest :: (MonadIO m) => Runtime e o s -> PartitionKey -> e -> m o+makeRequest runtime key e = call runtime (RMUserRequest key e)+++{- |+ Send a search request to the legion runtime. Returns results that are+ __strictly greater than__ the provided 'SearchTag'.+-}+search :: (MonadIO m) => Runtime e o s -> SearchTag -> Source m IndexRecord+search runtime tag =+ call runtime (RMUserSearch tag) >>= \case+ Nothing -> return ()+ Just record@IndexRecord {irTag, irKey} -> do+ yield record+ search runtime (SearchTag irTag (Just irKey))+++{- | Get the runtime state for debugging. -}+debugRuntimeState :: (MonadIO m)+ => Runtime e o s+ -> m Value+debugRuntimeState runtime = call runtime RMDebugRuntimeState+++{- | Get a partition for debugging. -}+debugPartition :: (MonadIO m)+ => Runtime e o s+ -> PartitionKey+ -> m (Maybe (PartitionPowerState e o s))+debugPartition runtime = call runtime . RMDebugPartition+++{- | Eject a peer. -}+eject :: (MonadIO m)+ => Runtime e o s+ -> Peer+ -> m ()+eject runtime = call runtime . RMEject+++{- | Get the index for debugging. -}+debugIndex :: (MonadIO m)+ => Runtime e o s+ -> m (Set IndexRecord)+debugIndex runtime = call runtime RMDebugIndex+++{- | Get the divergent peers. -}+getDivergent :: (MonadIO m)+ => Runtime e o s+ -> m (Map Peer (Maybe UTCTime))+getDivergent runtime = call runtime RMGetDivergent+++{- | Dump all of the locally managed partitions, for debugging. -}+debugLocalPartitions :: (MonadIO m)+ => Runtime e o s+ -> m (Map PartitionKey (PartitionPowerState e o s))+debugLocalPartitions runtime = call runtime RMDebugLocalPartitions+++{- |+ Execute the Legion runtime, with the given user definitions, and+ framework settings. This function never returns (except maybe with an+ exception if something goes horribly wrong).+-}+executeRuntime :: (+ ForkM m,+ LegionConstraints e o s,+ MonadCatch m,+ MonadLoggerIO m+ )+ => Persistence e o s+ {- ^ The persistence layer used to back the legion framework. -}+ -> RuntimeSettings+ {- ^ Settings and configuration of the legionframework. -}+ -> StartupMode+ -> Runtime e o s+ {- ^ A source of requests, together with a way to respond to the requets. -}+ -> m ()+executeRuntime+ persistence+ settings+ startupMode+ runtime+ = do+ {- Start the various messages sources. -}+ startPeerListener settings runtime+ startJoinListener settings runtime++ rts <- makeRuntimeState persistence settings startupMode+ void . runRuntimeT persistence rts . runConduit $+ chanToSource (unRuntime runtime)+ .| awaitForever (\msg -> do+ $(logDebug) . T.pack $ "Receieved: " ++ show msg+ lift (handleRuntimeMessage runtime msg)+ )+++{- | Handle runtime message. -}+handleRuntimeMessage :: (+ ForkM m,+ LegionConstraints e o s,+ MonadCatch m,+ MonadLoggerIO m+ )+ => Runtime e o s+ {- ^+ A handle on our own runtime, used to send messages back to+ ourselves.+ -}+ -> RuntimeMessage e o s+ -> RuntimeT e o s m ()++handleRuntimeMessage+ runtime+ (RMPeerMessage msg@(PeerMessage source _ _))+ = do+ updateRecvClock source+ handlePeerMessage runtime msg++handleRuntimeMessage _ (RMJoinRequest (JoinRequest addr) responder) = do+ (peer, cluster) <- S.joinCluster addr+ respond responder (JoinOk peer cluster)++handleRuntimeMessage _ (RMDebugRuntimeState responder) =+ respond responder =<< S.debugRuntimeState++handleRuntimeMessage _ (RMDebugPartition key responder) =+ respond responder =<< S.debugPartition key++handleRuntimeMessage _ (RMEject peer responder) =+ respond responder =<< S.eject peer++handleRuntimeMessage _ (RMDebugIndex responder) =+ respond responder =<< S.debugIndex++handleRuntimeMessage _ (RMGetDivergent responder) =+ respond responder =<< S.getDivergent++handleRuntimeMessage _ (RMDebugLocalPartitions responder) =+ respond responder =<< S.debugLocalPartitions++handleRuntimeMessage _ (RMUserRequest key request responder) =+ userRequest key request (respond responder)++handleRuntimeMessage _ (RMUserSearch tag responder) =+ searchDispatch tag (respond responder)+++{- | Handle a peer message. -}+handlePeerMessage :: (+ ForkM m,+ LegionConstraints e o s,+ MonadCatch m,+ MonadLoggerIO m+ )+ => Runtime e o s+ {- ^+ A handle on our own runtime, used to send messages back to+ ourselves.+ -}+ -> PeerMessage e o s+ -> RuntimeT e o s m ()++handlePeerMessage {- PartitionMerge -}+ _runtime+ PeerMessage {+ payload = (PartitionMerge key partition)+ }+ =+ partitionMerge key partition++handlePeerMessage {- ForwardRequest -}+ _runtime+ PeerMessage {+ source,+ messageId,+ payload = ForwardRequest key event+ }+ =+ forwardedRequest source messageId key event++handlePeerMessage {- ForwardResponse -}+ _runtime+ PeerMessage {+ payload = ForwardResponse forMessageId output+ }+ =+ forwardResponse forMessageId output++handlePeerMessage {- ClusterMerge -}+ _runtime+ PeerMessage {+ payload = ClusterMerge cluster+ }+ =+ clusterMerge cluster++handlePeerMessage {- Search -}+ {- This is where we handle local search execution. -}+ _runtime+ PeerMessage {+ source,+ payload = Search searchTag+ }+ = do+ searchResult <- S.search searchTag+ cm <- getCM+ void $ send cm source (SearchResponse searchTag searchResult)++handlePeerMessage {- SearchResponse -}+ {-+ This is where we gather all the responses from the various peers+ to which we dispatched search requests.+ -}+ _runtime+ PeerMessage {+ source,+ payload = SearchResponse searchTag record+ }+ =+ searchResponse source searchTag record++handlePeerMessage {- JoinNext -}+ _runtime+ PeerMessage {+ source,+ messageId,+ payload = JoinNext askKeys+ }+ = do+ cm <- getCM+ joinNext source askKeys (\case+ Nothing -> void $+ send cm source (JoinNextResponse messageId JoinFinished)+ Just (gotKey, partition) -> void $+ send cm source (JoinNextResponse messageId (Joined gotKey partition))+ )++handlePeerMessage {- JoinNextResponse -}+ _runtime+ PeerMessage {+ source,+ payload = JoinNextResponse _toMessageId response+ }+ =+ joinNextResponse source (toMaybe response)+ where+ toMaybe+ :: JoinNextResponse e o s+ -> Maybe (PartitionKey, PartitionPowerState e o s)+ toMaybe (Joined key partition) = Just (key, partition)+ toMaybe JoinFinished = Nothing+++{- | A way for the runtime to respond to a message. -}+newtype Responder a = Responder {+ unResponder :: a -> IO ()+ }+instance Show (Responder a) where+ show _ = "Responder"+++{- | Respond to a messag, using the given responder, in 'MonadIO'. -}+respond :: (MonadIO m) => Responder a -> a -> m ()+respond responder = liftIO . unResponder responder+++{- | Send a message to the runtime that blocks on a response. -}+call :: (MonadIO m)+ => Runtime e o s+ -> (Responder a -> RuntimeMessage e o s)+ -> m a+call runtime withResonder = liftIO $ do+ mvar <- newEmptyMVar+ cast runtime (withResonder (Responder (putMVar mvar)))+ takeMVar mvar+++{- | Send a message to the runtime. Do not wait for a result. -}+cast :: Runtime e o s -> RuntimeMessage e o s -> IO ()+cast runtime = writeChan (unRuntime runtime)+++data RuntimeMessage e o s+ = RMPeerMessage (PeerMessage e o s)+ | RMJoinRequest JoinRequest (Responder JoinResponse)+ | RMDebugRuntimeState (Responder Value)+ | RMDebugPartition+ PartitionKey+ (Responder (Maybe (PartitionPowerState e o s)))+ | RMEject Peer (Responder ())+ | RMDebugIndex (Responder (Set IndexRecord))+ | RMGetDivergent (Responder (Map Peer (Maybe UTCTime)))+ | RMDebugLocalPartitions+ (Responder (Map PartitionKey (PartitionPowerState e o s)))+ | RMUserRequest PartitionKey e (Responder o)+ | RMUserSearch SearchTag (Responder (Maybe IndexRecord))+ deriving (Show)+++{- |+ Start the peer listener, which accepts peer messages from the network+ and sends them to the runtime.+-}+startPeerListener :: (+ ForkM m,+ LegionConstraints e o s,+ MonadCatch m,+ MonadLoggerIO m+ )+ => RuntimeSettings+ -> Runtime e o s+ -> m ()++startPeerListener RuntimeSettings {peerBindAddr} runtime =+ catchAll (do+ so <- liftIO $ do+ so <- socket (fam peerBindAddr) Stream defaultProtocol+ setSocketOption so ReuseAddr 1+ bind so peerBindAddr+ listen so 5+ return so+ forkC "peer socket acceptor" $ acceptLoop so runtime+ ) (\err -> do+ $(logError) . T.pack+ $ "Couldn't start incomming peer message service, because of: "+ ++ show (err :: SomeException)+ throwM err+ )+ where+ acceptLoop :: (MonadLoggerIO m, LegionConstraints e o s, MonadCatch m)+ => Socket+ -> Runtime e o s+ -> m ()+ acceptLoop so runtime_ =+ catchAll (+ forever $ do+ (conn, _) <- liftIO $ accept so+ remoteAddr <- liftIO $ getPeerName conn+ void+ . forkL+ . logErrors remoteAddr+ $ runConduit (+ sourceSocket conn+ .| conduitDecode+ .| awaitForever (liftIO . cast runtime_ . RMPeerMessage)+ )+ ) (\err -> do+ $(logError) . T.pack+ $ "error in peer message accept loop: "+ ++ show (err :: SomeException)+ throwM err+ )+ where+ logErrors :: SockAddr -> LIO () -> LIO ()+ logErrors remoteAddr io = do+ result <- try io+ case result of+ Left err ->+ $(logWarn) . T.pack+ $ "Incomming peer connection (" ++ show remoteAddr+ ++ ") crashed because of: " ++ show (err :: SomeException)+ Right v -> return v+++{- |+ Starts the join listener, which accepts cluster join requests from+ the network and sends them to the runtime.+-}+startJoinListener :: (MonadCatch m, MonadLoggerIO m, ForkM m)+ => RuntimeSettings+ -> Runtime e o s+ -> m ()++startJoinListener RuntimeSettings {joinBindAddr} runtime =+ catchAll (do+ so <- liftIO $ do+ so <- socket (fam joinBindAddr) Stream defaultProtocol+ setSocketOption so ReuseAddr 1+ bind so joinBindAddr+ listen so 5+ return so+ forkC "join socket acceptor" $ acceptLoop so+ ) (\err -> do+ $(logError) . T.pack+ $ "Couldn't start join request service, because of: "+ ++ show (err :: SomeException)+ throwM err+ )+ where+ acceptLoop :: (MonadCatch m, MonadLoggerIO m) => Socket -> m ()+ acceptLoop so =+ catchAll (+ forever $ do+ (conn, _) <- liftIO (accept so)+ void+ . forkL+ . logErrors+ . liftIO+ $ runConduit (+ sourceSocket conn+ .| conduitDecode+ .| awaitForever (\req -> liftIO $+ sendAll conn . encode+ =<< call runtime (RMJoinRequest req)+ )+ )+ ) (\err -> do+ $(logError) . T.pack+ $ "error in join request accept loop: "+ ++ show (err :: SomeException)+ throwM err+ )+ where+ logErrors :: (MonadCatch m, MonadLoggerIO m) => m () -> m ()+ logErrors m = do+ result <- try m+ case result of+ Left err ->+ $(logWarn) . T.pack+ $ "Incomming join connection crashed because of: "+ ++ show (err :: SomeException)+ Right v -> return v
src/Network/Legion/Runtime/ConnectionManager.hs view
@@ -9,28 +9,35 @@ ConnectionManager, newConnectionManager, send,+ forward, newPeers, ) where import Prelude hiding (lookup) -import Control.Concurrent (Chan, writeChan, newChan, readChan)-import Control.Exception (try, SomeException, bracketOnError)+import Control.Concurrent (Chan, writeChan, newChan, readChan,+ newEmptyMVar, putMVar, takeMVar)+import Control.Exception (SomeException, bracketOnError) import Control.Monad (void)-import Control.Monad.Logger (logInfo, logWarn)+import Control.Monad.Catch (MonadCatch, try)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Logger (logInfo, logWarn, MonadLoggerIO) import Control.Monad.Trans.Class (lift) import Data.Binary (Binary, encode) import Data.ByteString.Lazy (ByteString)+import Data.Conduit (Sink, runConduit, (.|), await) import Data.Map (toList, insert, empty, Map, lookup) import Data.Text (pack) import Network.Legion.BSockAddr (BSockAddr(BSockAddr))+import Network.Legion.Conduit (chanToSource) import Network.Legion.Distribution (Peer)-import Network.Legion.Fork (forkC)-import Network.Legion.LIO (LIO)-import Network.Legion.Runtime.PeerMessage (PeerMessage)+import Network.Legion.Fork (forkC, ForkM)+import Network.Legion.Runtime.PeerMessage (PeerMessage(PeerMessage),+ MessageId, PeerMessagePayload, source, messageId, payload,+ nextMessageId, newSequence)+import Network.Legion.SocketUtil (fam) 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))+ defaultProtocol, connect, close, SockAddr) import Network.Socket.ByteString.Lazy (sendAll) {- |@@ -44,63 +51,113 @@ {- | Create a new connection manager. -}-newConnectionManager :: (Binary e, Binary o, Binary s)- => Map Peer BSockAddr- -> LIO (ConnectionManager e o s)-newConnectionManager initPeers = do- chan <- lift newChan+newConnectionManager :: (+ Binary e,+ Binary o,+ Binary s,+ ForkM m,+ MonadCatch m,+ MonadLoggerIO m+ )+ => Peer+ -> Map Peer BSockAddr+ -> m (ConnectionManager e o s)+newConnectionManager self initPeers = do+ chan <- liftIO newChan+ nextId <- newSequence forkC "connection manager thread" $- manager chan S {connections = empty}+ manager chan S {+ nextId,+ connections = empty+ } let cm = C chan newPeers cm initPeers return cm where- manager :: (Binary s, Binary o, Binary e)+ manager :: (+ Binary e,+ Binary o,+ Binary s,+ ForkM m,+ MonadCatch m,+ MonadLoggerIO m+ ) => Chan (Message e o s) -> State e o s- -> LIO ()- manager chan state = lift (readChan chan) >>= handle state >>= manager chan+ -> m ()+ manager chan state =+ runConduit (chanToSource chan .| handle state) - handle :: (Binary e, Binary o, Binary s)+ handle :: (+ Binary e,+ Binary o,+ Binary s,+ ForkM m,+ MonadCatch m,+ MonadLoggerIO m+ ) => State e o s- -> Message e o s- -> LIO (State e 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+ -> Sink (Message e o s) m ()+ handle s@S {connections, nextId} =+ await >>= \case+ Nothing -> return ()+ Just (NewPeer peer addr) ->+ handle =<< case lookup peer connections of+ Nothing -> do+ conn <- lift (connection addr)+ return s {+ connections = insert peer conn connections+ }+ Just _ ->+ return s+ Just (Send peer payload respond) -> do+ case lookup peer connections of+ Nothing -> $(logWarn) . pack $ "unknown peer: " ++ show peer+ Just conn -> liftIO $+ writeChan conn PeerMessage {+ source = self,+ messageId = nextId,+ payload+ }+ liftIO (respond nextId)+ handle s {nextId = nextMessageId nextId}+ Just (Forward peer msg) ->+ case lookup peer connections of+ Nothing -> $(logWarn) . pack $ "unknown peer: " ++ show peer+ Just conn -> liftIO $ writeChan conn msg -{- |- Build a new connection.--}-connection :: (Binary e, Binary o, Binary s)+{- | Build a new connection. -}+connection :: (+ Binary e,+ Binary o,+ Binary s,+ ForkM m,+ MonadCatch m,+ MonadIO m,+ MonadLoggerIO m+ ) => SockAddr- -> LIO (Chan (PeerMessage e o s))+ -> m (Chan (PeerMessage e o s)) connection addr = do- chan <- lift newChan+ chan <- liftIO newChan forkC ("connection to: " ++ show addr) $ handle chan Nothing return chan where- handle :: (Binary e, Binary o, Binary s)+ handle :: (+ Binary e,+ Binary o,+ Binary s,+ MonadCatch m,+ MonadLoggerIO m+ ) => Chan (PeerMessage e o s) -> Maybe Socket- -> LIO ()+ -> m () handle chan so =- lift (readChan chan) >>= sendWithRetry so . encode >>= handle chan+ liftIO (readChan chan) >>= sendWithRetry so . encode >>= handle chan {- | Open a socket. -} openSocket :: IO Socket@@ -122,9 +179,12 @@ create a new socket and retry sending the payload. Return whatever the "working" socket is. -}- sendWithRetry :: Maybe Socket -> ByteString -> LIO (Maybe Socket)+ sendWithRetry :: (MonadCatch m, MonadLoggerIO m)+ => Maybe Socket+ -> ByteString+ -> m (Maybe Socket) sendWithRetry Nothing payload =- (lift . try) openSocket >>= \case+ try (liftIO openSocket) >>= \case Left err -> do $(logWarn) . pack $ "Can't connect to: " ++ show addr ++ ". Dropping message on "@@ -132,7 +192,7 @@ ++ show (err :: SomeException) return Nothing Right so -> do- result2 <- (lift . try) (sendAll so payload)+ result2 <- try (liftIO (sendAll so payload)) case result2 of Left err -> $(logWarn) . pack $ "An error happend when trying to send a payload over a socket "@@ -142,43 +202,57 @@ Right _ -> return () return (Just so) sendWithRetry (Just so) payload =- (lift . try) (sendAll so payload) >>= \case+ try (liftIO (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 ()))+ (liftIO . void) (try (close so) :: IO (Either SomeException ())) sendWithRetry Nothing payload Right _ -> return (Just so) -{- |- Send a message to a peer.--}-send- :: ConnectionManager e o s+{- | Send a message to a peer. -}+send :: (MonadIO m)+ => ConnectionManager e o s -> Peer+ -> PeerMessagePayload e o s+ -> m MessageId+send (C chan) peer payload = do+ mvar <- liftIO newEmptyMVar+ liftIO . writeChan chan $ Send peer payload (putMVar mvar)+ liftIO (takeMVar mvar)+++{- | Forward a message. -}+forward :: (MonadIO m)+ => ConnectionManager e o s+ -> Peer -> PeerMessage e o s- -> LIO ()-send (C chan) peer = lift . writeChan chan . Send peer+ -> m ()+forward (C chan) peer =+ liftIO . writeChan chan . Forward peer {- | Tell the connection manager about a new peer. -}-newPeer- :: ConnectionManager e o s+newPeer :: (MonadIO io)+ => ConnectionManager e o s -> Peer -> SockAddr- -> LIO ()-newPeer (C chan) peer addr = lift $ writeChan chan (NewPeer peer addr)+ -> io ()+newPeer (C chan) peer addr = liftIO $ writeChan chan (NewPeer peer addr) {- | Tell the connection manager about all the peers known to the cluster state. -}-newPeers :: ConnectionManager e o s -> Map Peer BSockAddr -> LIO ()+newPeers :: (MonadIO io)+ => ConnectionManager e o s+ -> Map Peer BSockAddr+ -> io () newPeers cm peers = mapM_ oneNewPeer (toList peers) where@@ -188,7 +262,8 @@ {- | The internal state of the connection manager. -}-newtype State e o s = S {+data State e o s = S {+ nextId :: MessageId, connections :: Map Peer (Chan (PeerMessage e o s)) } @@ -198,16 +273,7 @@ -} data Message e o s = NewPeer Peer SockAddr- | Send Peer (PeerMessage e 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+ | Forward Peer (PeerMessage e o s)+ | Send Peer (PeerMessagePayload e o s) (MessageId -> IO ())
src/Network/Legion/Runtime/PeerMessage.hs view
@@ -12,7 +12,7 @@ nextMessageId, ) where -import Control.Monad.Trans.Class (lift)+import Control.Monad.IO.Class (MonadIO) import Data.Binary (Binary) import Data.UUID (UUID) import Data.Word (Word64)@@ -21,7 +21,6 @@ import Network.Legion.Distribution (Peer) import Network.Legion.Index (SearchTag, IndexRecord) import Network.Legion.KeySet (KeySet)-import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.PartitionState (PartitionPowerState) import Network.Legion.UUID (getUUID)@@ -77,8 +76,8 @@ 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+newSequence :: (MonadIO io) => io MessageId+newSequence = do sid <- getUUID return (M sid 0)
+ src/Network/Legion/Runtime/State.hs view
@@ -0,0 +1,1035 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}++{- | This module contains low level state types. -}+module Network.Legion.Runtime.State (+ -- * Runtime State+ RuntimeState,+ makeRuntimeState,++ -- * Runtime Monads+ RuntimeT,+ runRuntimeT,+ runConcurrentT,++ -- * Runtime Monad Operations+ updateRecvClock,+ joinCluster,+ eject,+ getDivergent,+ userRequest,+ forwardedRequest,+ forwardResponse,+ clusterMerge,+ getCM,+ searchDispatch,+ search,+ searchResponse,+ joinNext,+ joinNextResponse,+ partitionMerge,+ ++ -- * Debug Monad Operations+ debugIndex,+ debugRuntimeState,+ debugLocalPartitions,+ debugPartition,++ -- * Other Types+ StartupMode(..),+ JoinRequest(..),+ JoinResponse(..),+ UserResponse(..),+) where+++import Control.Concurrent.STM (TVar, atomically, readTVar, newTVar,+ STM, writeTVar, modifyTVar)+import Control.Monad (unless, void)+import Control.Monad.Catch (throwM, MonadThrow, MonadCatch)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Logger (MonadLoggerIO, MonadLogger, logInfo,+ logError, logWarn, logDebug)+import Control.Monad.Trans.Class (lift, MonadTrans)+import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask)+import Control.Monad.Trans.State (StateT, runStateT, modify, get, put)+import Data.Aeson (Value, toJSON, ToJSON)+import Data.Binary (encode, Binary)+import Data.Bool (bool)+import Data.Conduit ((.|), await, transPipe, runConduit)+import Data.Conduit.Network (sourceSocket)+import Data.Conduit.Serialization.Binary (conduitDecode)+import Data.Default.Class (Default)+import Data.Map (Map)+import Data.Maybe (catMaybes, fromMaybe)+import Data.Set (Set, (\\))+import Data.String (IsString, fromString)+import Data.Text (Text)+import Data.Time (UTCTime, getCurrentTime)+import GHC.Generics (Generic)+import Network.Legion.Application (Persistence, list, saveCluster,+ getState, saveState)+import Network.Legion.BSockAddr (BSockAddr(BSockAddr))+import Network.Legion.ClusterState (ClusterPowerState, RebalanceOrd,+ ClusterPowerStateT)+import Network.Legion.Distribution (Peer, newPeer)+import Network.Legion.Fork (ForkM, forkM)+import Network.Legion.Index (IndexRecord(IndexRecord),+ SearchTag(SearchTag), Indexable, indexEntries, stTag, stKey, irTag,+ irKey)+import Network.Legion.KeySet (KeySet)+import Network.Legion.PartitionKey (PartitionKey)+import Network.Legion.PartitionState (PartitionPowerState,+ PartitionPowerStateT)+import Network.Legion.PowerState (Event)+import Network.Legion.PowerState.Monad (PropAction(Send, DoNothing))+import Network.Legion.Runtime.ConnectionManager (ConnectionManager,+ newConnectionManager)+import Network.Legion.Runtime.PeerMessage (MessageId, newSequence,+ PeerMessagePayload(ClusterMerge, PartitionMerge, JoinNext,+ ForwardRequest, ForwardResponse, Search), PeerMessage(PeerMessage),+ source, messageId, payload)+import Network.Legion.Settings (RuntimeSettings(RuntimeSettings,+ peerBindAddr))+import Network.Legion.SocketUtil (fam)+import Network.Legion.UUID (getUUID)+import Network.Socket (SocketType(Stream), defaultProtocol, socket,+ SockAddr, connect)+import Network.Socket.ByteString.Lazy (sendAll)+import System.IO (stderr, hPutStrLn)+import qualified Data.Aeson as A+import qualified Data.ByteString.Lazy.Char8 as BSL8+import qualified Data.Conduit.List as CL+import qualified Data.Map as Map+import qualified Data.Set as Set+import qualified Data.Text as T+import qualified Network.Legion.ClusterState as C+import qualified Network.Legion.Distribution as D+import qualified Network.Legion.KeySet as KS+import qualified Network.Legion.PowerState as PS+import qualified Network.Legion.PowerState.Monad as PM+import qualified Network.Legion.Runtime.ConnectionManager as CM+++{- | The state of the runtime system. -}+data RuntimeState e o s m = RuntimeState {+ self :: Peer,+ cluster :: ClusterPowerState,+ partitions :: Map PartitionKey (TVar (PartitionWorkerState e o s m)),+ rtsIndex :: Set IndexRecord,+ joins :: Map Peer KeySet,+ {- ^ Outstanding joins. -}+ lastRebalance :: RebalanceOrd,+ forwarded :: Map MessageId (o -> m ()),+ nextId :: MessageId,+ cm :: ConnectionManager e o s,+ recvClock :: Map Peer (Maybe UTCTime),+ {- ^ When did we last receive a message from a peer. -}+ searches :: Map+ SearchTag+ (+ Set Peer,+ Maybe IndexRecord,+ [Maybe IndexRecord -> m ()]+ )+ {- ^+ A map of currently dispatched searches. The values+ are the peers from which we are still expecting+ a result, best result so far, and the responders+ to which to send the eventual best result.++ The 'searches' field is a little weird.++ It turns out that searches are deterministic+ over the parameters of 'SearchTag' and cluster+ state. This should make sense, because everything+ in Haskell is deterministic given __all__+ the parameters. Since the cluster state only+ changes over time, searches that happen "at the+ same time" and for the same 'SearchTag' can be+ considered identical. I don't think it is too+ much of a stretch to say that searches that have+ overlapping execution times can be considered+ to be happening "at the same time", therefore+ the search tag becomes determining factor in the+ result of the search.++ This is a long-winded way of justifying the fact+ that, if we are currently executing a search and+ an identical search requests arrives, then the+ second identical search is just piggy-backed on the+ results of the currently executing search. Whether+ this counts as a premature optimization hack+ or a beautifully elegant expression of platonic+ reality is left as an exercise for the reader. It+ does help simplify the code a little bit because+ we don't have to specify some kind of UUID to+ differentiate otherwise identical searches.+ -}+ }+instance Show (RuntimeState e o s m) where+ show = BSL8.unpack . A.encode+instance ToJSON (RuntimeState e o s m) where+ toJSON _ = toJSON ("RuntimeState" :: Text)+++{- | The state of an individual asynchronous partition worker. -}+data PartitionWorkerState e o s m = PWS {+ pwsCm :: ConnectionManager e o s,+ pwsKey :: PartitionKey,+ pwsSelf :: Peer,+ pwsPersistence :: Persistence e o s,+ pwsCacheVal :: Maybe (PartitionPowerState e o s),+ pwsJobQueue :: [(ClusterPowerState, PartitionPowerStateT e o s m ())]+ }+++{- | The Runtime Monad Transformer. -}+newtype RuntimeT e o s m a = RuntimeT {+ unRuntimeT :: StateT (RuntimeState e o s m) (ReaderT (Persistence e o s) m) a+ }+ deriving (Functor, Applicative, Monad, MonadIO, MonadLogger, MonadThrow)+instance MonadTrans (RuntimeT e o s) where+ lift = RuntimeT . lift . lift+++{- | Execute a 'RuntimeT'. -}+runRuntimeT+ :: Persistence e o s+ -> RuntimeState e o s m+ -> RuntimeT e o s m a+ -> m (a, RuntimeState e o s m)+runRuntimeT persistence rts =+ (`runReaderT` persistence) . (`runStateT` rts) . unRuntimeT+++{- | Initialize the runtime state. -}+makeRuntimeState :: (+ Binary e,+ Binary o,+ Binary s,+ Event e o s,+ ForkM m,+ Indexable s,+ MonadCatch m,+ MonadLoggerIO m+ )+ => Persistence e o s+ -> RuntimeSettings+ -> StartupMode+ -> m (RuntimeState e o s m)++makeRuntimeState+ persistence+ settings@RuntimeSettings {peerBindAddr}+ NewCluster+ = do+ {- Build a brand new node state, for the first node in a cluster. -}+ verifyClearPersistence persistence+ self <- newPeer+ clusterId <- getUUID+ let+ cluster = C.new clusterId self peerBindAddr+ makeRuntimeState persistence settings (Recover self cluster)++makeRuntimeState+ persistence+ settings@RuntimeSettings {peerBindAddr}+ (JoinCluster addr)+ = do+ {-+ Join a cluster by either starting fresh, or recovering from a+ shutdown or crash.+ -}+ verifyClearPersistence persistence+ $(logInfo) "Trying to join an existing cluster."+ (self, cluster) <- requestJoin (JoinRequest (BSockAddr peerBindAddr))+ makeRuntimeState persistence settings (Recover self cluster)+ where+ requestJoin :: (MonadLoggerIO io)+ => JoinRequest+ -> io (Peer, ClusterPowerState)+ requestJoin joinMsg = liftIO $ do+ so <- socket (fam addr) Stream defaultProtocol+ connect so addr+ sendAll so (encode joinMsg)+ {-+ using sourceSocket and conduitDecode is easier than building+ a recive/decode state loop, even though we only read a single+ response.+ -}+ runConduit $ sourceSocket so .| conduitDecode .| do+ response <- await+ case response of+ Nothing -> fail+ $ "Couldn't join a cluster because there was no response "+ ++ "to our join request!"+ Just (JoinOk self cps) ->+ return (self, cps)++makeRuntimeState persistence _ (Recover self cluster) = do+ {- Make sure to rebuild the index in the case of recovery. -}+ rtsIndex <- runConduit . transPipe liftIO $+ list persistence+ .| CL.fold addIndexRecords Set.empty+ firstMessageId <- newSequence+ cm <- newConnectionManager self (C.getPeers cluster)+ liftIO $ saveCluster persistence self cluster+ return RuntimeState {+ self,+ cluster,+ partitions = Map.empty,+ rtsIndex,+ joins = Map.empty,+ lastRebalance = minBound,+ forwarded = Map.empty,+ nextId = firstMessageId,+ cm,+ recvClock = Map.empty,+ searches = Map.empty+ }+ where+ addIndexRecords :: (Indexable s, Event e o s)+ => Set IndexRecord+ -> (PartitionKey, PartitionPowerState e o s)+ -> Set IndexRecord+ addIndexRecords index (key, partition) =+ let+ newRecords =+ Set.map+ (`IndexRecord` key)+ (indexEntries (PS.projectedValue partition))+ in Set.union index newRecords+++{- | This defines the various ways a node can be spun up. -}+data StartupMode+ = NewCluster+ {- ^ Indicates that we should bootstrap a new cluster at startup. -}+ | JoinCluster SockAddr+ {- ^ Indicates that the node should try to join an existing cluster. -}+ | Recover Peer ClusterPowerState+ {- ^+ Recover from a crash as the given peer, using the given cluster+ state.+ -}+ deriving (Show, Eq)+++{- | This is the type of a join request message. -}+newtype JoinRequest = JoinRequest BSockAddr+ deriving (Generic, Show)+instance Binary JoinRequest+++{- | The response to a JoinRequest message -}+data JoinResponse+ = JoinOk Peer ClusterPowerState+ deriving (Generic)+instance Binary JoinResponse+++{- |+ Helper for 'makeRuntimeState'. Verify that there is nothing in the+ persistence layer.+-}+verifyClearPersistence :: (MonadLoggerIO io) => Persistence e o s -> io ()+verifyClearPersistence persistence = + liftIO (runConduit (list persistence .| CL.head)) >>= \case+ Just _ -> do+ let+ msg :: (IsString a) => a+ msg = fromString+ $ "We are trying to start up a new peer, but the persistence "+ ++ "layer already has data in it. This is an invalid state. "+ ++ "New nodes must be started from a totally clean, empty state."+ $(logError) msg+ liftIO $ do+ hPutStrLn stderr msg+ putStrLn msg+ error msg+ Nothing ->+ return ()+++{- | Update the time when we last received a message from a peer. -}+updateRecvClock :: (MonadIO m) => Peer -> RuntimeT e o s m ()+updateRecvClock peer = RuntimeT $ do+ now <- liftIO getCurrentTime+ modify (\rts@RuntimeState {recvClock} ->+ let+ newRecvClock = Map.insert peer (Just now) recvClock+ in newRecvClock `seq` rts {+ recvClock = newRecvClock+ }+ )+++{- | Return the current index for debugging. -}+debugIndex :: (Monad m) => RuntimeT e o s m (Set IndexRecord)+debugIndex = RuntimeT $ rtsIndex <$> get+++{- | Return the runtime state for debugging. -}+debugRuntimeState :: (Monad m) => RuntimeT e o s m Value+debugRuntimeState = toJSON <$> RuntimeT get+++{- | Return all of the local partitions for debugging. -}+debugLocalPartitions :: (MonadIO m)+ => RuntimeT e o s m (Map PartitionKey (PartitionPowerState e o s))+debugLocalPartitions = do+ persistence <- RuntimeT (lift ask)+ Map.fromList <$> runConduit (+ transPipe liftIO (list persistence)+ .| CL.consume+ )+ +++{- | Return a specific partition, for debugging. -}+debugPartition :: (MonadIO m)+ => PartitionKey+ -> RuntimeT e o s m (Maybe (PartitionPowerState e o s))+debugPartition key = RuntimeT $ do+ persistence <- lift ask+ liftIO (getState persistence key)+++{- | Let a new peer join the cluster. -}+joinCluster :: (MonadIO m, MonadThrow m)+ => BSockAddr+ -> RuntimeT e o s m (Peer, ClusterPowerState)+joinCluster addr = do+ peer <- newPeer+ runClusterPowerStateT (C.joinCluster peer addr)+ cluster <- getCluster+ return (peer, cluster)+++{- | Eject a peer from the cluster. -}+eject :: (MonadIO m, MonadThrow m) => Peer -> RuntimeT e o s m ()+eject peer = do+ {-+ 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?+ -}+ runClusterPowerStateT (C.eject peer)+ {-+ 'runClusterPowerStateT (C.eject peer)' will cause us to attempt to+ notify the peer that they have been ejected, but that notification+ is almost certainly going to go unacknowledged because the peer+ is probably down.+ + This call to 'eject' was presumably invoked as a result of user+ action, and we must therefore trust the user to know that the peer+ is really down and not coming back. This "guarantee" allows us to+ acknowledge the ejection on the peer's behalf.++ This call will acknowledge the drop on behalf of the peer, and also+ remove that peer from the keyspace distribution map.+ -}+ runClusterPowerStateTAs peer (return ())+++{- |+ Gets the peers that the local node thinks are diverging, and the time+ we last received a message from those peers.+-}+getDivergent :: (MonadIO m) => RuntimeT e o s m (Map Peer (Maybe UTCTime))+getDivergent = RuntimeT $ do+ RuntimeState {recvClock, partitions} <- get+ diverging <- lift . lift $ divergentPeers partitions+ return $ Map.fromAscList [+ (peer, r)+ | (peer, r) <- Map.toAscList recvClock+ , peer `Set.member` diverging+ ]+ where+ divergentPeers :: (MonadIO m)+ => Map PartitionKey (TVar (PartitionWorkerState e o s m))+ -> m (Set Peer)+ divergentPeers partitions = liftIO $+ foldr Set.union Set.empty . catMaybes <$> sequence [+ fmap PS.divergent . pwsCacheVal <$> atomically (readTVar tvar)+ | (_key, tvar) <- Map.toList partitions+ ]+++{- | Handle a user request, and pass the response to the continuation. -}+userRequest :: (+ Default s,+ Eq e,+ Event e o s,+ ForkM m,+ Indexable s,+ MonadCatch m,+ MonadLoggerIO m,+ Show e,+ Show s+ )+ => PartitionKey+ -> e+ -> (o -> m ())+ -> RuntimeT e o s m ()+userRequest key request k = do+ RuntimeState {self, cm} <- RuntimeT get+ route key >>= \case+ p | p == self -> + runConcurrentT key (+ lift . k =<< PM.event request+ )+ p -> do+ messageId <- CM.send cm p (ForwardRequest key request)+ (RuntimeT . modify) (\rts@RuntimeState {forwarded} -> rts {+ forwarded = Map.insert messageId k forwarded+ })+++{- | Handle a forwarded request. -}+forwardedRequest :: (+ Default s,+ Eq e,+ Event e o s,+ ForkM m,+ MonadCatch m,+ MonadLoggerIO m+ )+ => Peer+ -> MessageId+ -> PartitionKey+ -> e+ -> RuntimeT e o s m ()+forwardedRequest source messageId key event = do+ RuntimeState {self, cm} <- RuntimeT get+ route key >>= \case+ p | p == self ->+ runConcurrentT key (do+ o <- PM.event event+ (void . lift) (CM.send cm source (ForwardResponse messageId o))+ )+ p ->+ {-+ No need to keep track of the forwarded message, we just need to+ reconstruct the original message and send it on its way.++ TODO think about implementing cycle detection. Cycles should+ not exists, so if we detect one then that classifies as+ a bug. So really this is more of an opportunity for bug+ detection.+ -}+ CM.forward cm p PeerMessage {+ source,+ messageId,+ payload = ForwardRequest key event+ }+++{- | Find the route for a user request. -}+route :: (MonadLogger m)+ => PartitionKey+ -> RuntimeT e o s m Peer+route key = RuntimeT $ do+ RuntimeState {self, cluster} <- get+ let routes = C.findRoute key cluster+ if self `Set.member` routes+ then return self+ else case Set.toList routes of+ [] -> do+ let msg = "No routes for key: " ++ show key+ $(logError) . T.pack $ msg+ error msg+ peer:_ -> return peer+ ++++{- | Receive a response to a forwarded user request. -}+forwardResponse :: (MonadLoggerIO m, Show o)+ => MessageId+ -> o+ -> RuntimeT e o s m ()+forwardResponse forMessageId output = do+ rts@RuntimeState{forwarded} <- RuntimeT get+ let (r, fwd) = lookupAndDelete forMessageId forwarded+ RuntimeT $ put rts {forwarded = fwd}+ case r of+ Nothing ->+ $(logWarn) . T.pack+ $ "Received unexpected forward response: "+ ++ show (forMessageId, output)+ Just respond ->+ lift (respond output)+++{- | Merge a forenig cluster power state. -}+clusterMerge :: (MonadIO m, MonadThrow m)+ => ClusterPowerState+ -> RuntimeT e o s m ()+clusterMerge cluster =+ runClusterPowerStateT (PM.merge cluster)+++{- | Return the handle to the connection manager. -}+getCM :: (Monad m) => RuntimeT e o s m (ConnectionManager e o s)+getCM = RuntimeT $ cm <$> get+++{- | Dispatch a distributed search request. -}+searchDispatch :: (MonadIO m)+ => SearchTag+ -> (Maybe IndexRecord -> m ())+ -> RuntimeT e o s m ()+searchDispatch searchTag k =+ Map.lookup searchTag . searches <$> RuntimeT get >>= \case+ Nothing -> do+ {-+ No identical search is currently being executed, kick off a+ new one.+ -}+ mcss <- minimumCompleteServiceSet+ mapM_ sendSearch (Set.toList mcss)+ (RuntimeT . modify) (\rts@RuntimeState {searches} -> rts {+ searches = Map.insert+ searchTag+ (mcss, Nothing, [k])+ searches+ })+ Just (peers, best, responders) ->+ {-+ A search for this tag is already in progress, just add the+ responder to the responder list.+ -}+ (RuntimeT . modify) (\rts@RuntimeState {searches} -> rts {+ searches = Map.insert+ searchTag+ (peers, best, k:responders)+ searches+ })+ where+ sendSearch :: (MonadIO m)+ => Peer+ -> RuntimeT e o s m ()+ sendSearch peer = do+ cm <- getCM+ void $ CM.send cm peer (Search searchTag)+++{- |+ Search the index, and return the first record that is __strictly+ greater than__ the provided search tag, if such a record exists.+-}+search :: (Monad m)+ => SearchTag+ -> RuntimeT e o s m (Maybe IndexRecord)++search SearchTag {stTag, stKey = Nothing} = RuntimeT $ do+ index <- rtsIndex <$> get+ return (Set.lookupGE IndexRecord {irTag = stTag, irKey = minBound} index)++search SearchTag {stTag, stKey = Just key} = RuntimeT $ do+ index <- rtsIndex <$> get+ return (Set.lookupGT IndexRecord {irTag = stTag, irKey = key} index)+++{- | Handle an incomming search response. -}+searchResponse :: (MonadLogger m)+ => Peer+ -> SearchTag+ -> Maybe IndexRecord+ -> RuntimeT e o s m ()++searchResponse source searchTag response =+ {- TODO: see if this function can't be made more elegant. -}+ Map.lookup searchTag . searches <$> RuntimeT get >>= \case+ Nothing ->+ {- There is no search happening. -}+ $(logWarn) . T.pack+ $ "Unsolicited SearchResponse: "+ ++ show (source, searchTag, response)+ Just (peers, best, responders) ->+ if source `Set.member` peers+ then+ let peers2 = Set.delete source peers+ in if null peers2+ then do+ {-+ All peers have responded, go ahead and respond to+ the client.+ -}+ lift $ mapM_ ($ bestOf best response) responders+ rts@RuntimeState {searches} <- RuntimeT get+ (RuntimeT . put) rts {searches = Map.delete searchTag searches}+ else do+ {- We are still waiting on some outstanding requests. -}+ rts@RuntimeState {searches} <- RuntimeT get+ (RuntimeT . put) rts {+ searches = Map.insert+ searchTag+ (peers2, bestOf best response, responders)+ searches+ }+ else+ {-+ There is a search happening, but the peer that responded+ is not part of it.+ -}+ $(logWarn) . T.pack+ $ "Unsolicited SearchResponse: "+ ++ show (source, searchTag, response)+ where+ {- |+ Figure out which index record returned to us by the various peers+ is the most appropriate to return to the user. This is mostly like+ 'min' but we can't use 'min' (or fancy applicative formulations)+ because we want to favor 'Just' instead of 'Nothing'.+ -}+ bestOf :: Maybe IndexRecord -> Maybe IndexRecord -> Maybe IndexRecord+ bestOf (Just a) (Just b) = Just (min a b)+ bestOf Nothing b = b+ bestOf a Nothing = a+++{- |+ Allow a peer to participate in the replication of the partition with the+ __minimum__ key that is within the indicated partition key set. Calls+ the continuation with @Nothing@ if there is no such partition, or @Just+ (key, partition)@ where @key@ is the partition key that was joined+ and @partition@ is the resulting partition power state.+-}+joinNext :: (+ Default s,+ Eq e,+ Event e o s,+ ForkM m,+ MonadCatch m,+ MonadLoggerIO m+ )+ => Peer+ -> KeySet+ -> (Maybe (PartitionKey, PartitionPowerState e o s) -> m ())+ -> RuntimeT e o s m ()+joinNext peer askKeys k = do+ persistence <- RuntimeT (lift ask)+ (lift . runConduit) (+ transPipe liftIO (list persistence)+ .| CL.filter ((`KS.member` askKeys) . fst)+ .| CL.head+ ) >>= \case+ Nothing -> lift (k Nothing)+ Just (gotKey, _) ->+ runConcurrentT gotKey (do+ PM.participate peer+ PM.acknowledge+ partition <- PM.getPowerState+ lift (k (Just (gotKey, partition)))+ )+++{- | Receive the result of a JoinNext request. -}+joinNextResponse :: (+ Default s,+ Eq e,+ Event e o s,+ ForkM m,+ MonadLoggerIO m,+ MonadCatch m,+ Show e,+ Show s+ )+ => Peer+ -> Maybe (PartitionKey, PartitionPowerState e o s)+ -> RuntimeT e o s m ()+joinNextResponse peer response = do+ RuntimeState {cluster, lastRebalance} <- RuntimeT get+ if lastRebalance > fst (C.nextAction cluster)+ then+ {- We are receiving messages from an old rebalance. Log and ignore. -}+ $(logWarn) . T.pack+ $ "Received an old join response: "+ ++ show (peer, response, cluster, lastRebalance)+ else do+ case response of+ Just (key, partition) -> do+ partitionMerge key partition+ RuntimeState {joins, cm} <- RuntimeT get+ case (KS.\\ KS.fromRange minBound key) <$> Map.lookup peer joins of+ Nothing ->+ {- An unexpected peer sent us this message, Ignore. -}+ $(logWarn) . T.pack+ $ "Unexpected join next: " ++ show (peer, response)+ Just needsJoinSet -> do+ unless (KS.null needsJoinSet) (+ void $ CM.send cm peer (JoinNext needsJoinSet)+ )+ (RuntimeT . modify) (\rts -> rts {+ joins = Map.filter+ (not . KS.null)+ (Map.insert peer needsJoinSet joins)+ })+ Nothing ->+ (RuntimeT . modify) (\rts@RuntimeState {joins} -> rts {+ joins = Map.delete peer joins+ })+ Map.null . joins <$> RuntimeT get >>= bool+ (return ())+ (runClusterPowerStateT C.finishRebalance)+++{- | Merge a foreign partition replica with the local partion replica. -}+partitionMerge :: (+ Default s,+ Eq e,+ Event e o s,+ ForkM m,+ MonadCatch m,+ MonadLoggerIO m+ )+ => PartitionKey+ -> PartitionPowerState e o s+ -> RuntimeT e o s m ()+partitionMerge key foreignPartition =+ runConcurrentT key (PM.merge foreignPartition)+++{- | Get the current cluster state. -}+getCluster :: (Monad m) => RuntimeT e o s m ClusterPowerState+getCluster = RuntimeT $ cluster <$> get+++{- |+ Run a 'ClusterPowerStateT', and perform any resulting propagation+ actions.+-}+runClusterPowerStateT :: (MonadIO m, MonadThrow m)+ => ClusterPowerStateT m a+ -> RuntimeT e o s m a+runClusterPowerStateT m = do+ as <- RuntimeT $ self <$> get+ runClusterPowerStateTAs as m+++{- |+ Run a clusterstate-flavored 'PowerStateT' in the 'RuntimeT' monad,+ automatically acknowledging the resulting power state.++ Generalized to run as any peer, in order to support exceptional cases+ like 'eject'.+-}+runClusterPowerStateTAs :: (MonadIO m, MonadThrow m)+ => Peer {- ^ The peer to run as. -}+ -> ClusterPowerStateT m a+ -> RuntimeT e o s m a+runClusterPowerStateTAs as m = do+ RuntimeState {cluster, self} <- RuntimeT get+ persistence <- RuntimeT (lift ask)+ lift (PM.runPowerStateT as cluster (m <* PM.acknowledge)) >>= \case+ Left err -> throwM err+ Right (a, action, cluster2, _outputs) -> do+ RuntimeT (modify (\rts -> rts {cluster = cluster2}))+ liftIO (saveCluster persistence self cluster2)+ case action of+ Send -> sequence_ [+ getCM >>= (\cm -> CM.send cm p (ClusterMerge cluster2))+ | p <- Set.toList (PS.allParticipants cluster2)+ , p /= self+ ]+ DoNothing -> return ()+ return a+++{- |+ The type of response to a user request, either forward to another node,+ or respond directly.+-}+data UserResponse o+ = Forward Peer+ | Respond o+++{- | The action is executed in a background thread. -}+runConcurrentT :: (+ Default s,+ Eq e,+ Event e o s,+ ForkM m,+ MonadCatch m,+ MonadLoggerIO m+ )+ => PartitionKey+ -> PartitionPowerStateT e o s m ()+ -> RuntimeT e o s m ()+runConcurrentT key action_ = do+ rts@RuntimeState {partitions} <- RuntimeT get+ persistence <- RuntimeT (lift ask)+ let job = (cluster rts, action_)+ case Map.lookup key partitions of+ Nothing -> do+ tvar <- liftIO (atomically (newTVar PWS {+ pwsCm = cm rts,+ pwsKey = key,+ pwsSelf = self rts,+ pwsPersistence = persistence,+ pwsCacheVal = Nothing,+ pwsJobQueue = []+ }))+ RuntimeT $ put rts {partitions = Map.insert key tvar partitions}+ lift =<< liftIO (atomically (queueAction tvar job))+ Just tvar ->+ lift =<< liftIO (atomically (queueAction tvar job))+ where+ {- |+ Put the partition action on the execution queue, and maybe also+ start the execution thread if there isn't already one running.+ -}+ queueAction :: (+ Default s,+ Eq e,+ Event e o s,+ ForkM m,+ MonadLoggerIO m+ )+ => TVar (PartitionWorkerState e o s m)+ -> (ClusterPowerState, PartitionPowerStateT e o s m ())+ -> STM (m ())+ queueAction tvar job = do+ pws@PWS {pwsJobQueue} <- readTVar tvar+ let+ forkJobWorker = return (forkM (jobWorker tvar job))+ writeTVar tvar pws {pwsJobQueue = pwsJobQueue ++ [job]}+ if null pwsJobQueue+ then forkJobWorker+ else return (return ())++ jobWorker :: (MonadLoggerIO m, Default s, Event e o s, Eq e)+ => TVar (PartitionWorkerState e o s m)+ -> (ClusterPowerState, PartitionPowerStateT e o s m ())+ -> m ()+ jobWorker tvar job =+ doJob tvar job >> nextJob >>= maybe shutdown (jobWorker tvar)+ where+ nextJob = liftIO . atomically $ do+ pws@PWS {pwsCacheVal, pwsJobQueue} <- readTVar tvar+ case pwsJobQueue of+ _:next:more -> do+ {- Pop the last job off the queue, and promote the next job. -}+ writeTVar tvar pws {pwsJobQueue = next:more}+ return (Just next)+ [_] -> do+ {-+ All jobs complete. Pop the last job off the stack and+ clean up the cache if necessary.+ -}+ writeTVar tvar $+ case Set.null . PS.divergent <$> pwsCacheVal of+ Just False -> pws {pwsJobQueue = []}+ _ -> pws {pwsCacheVal = Nothing, pwsJobQueue = []}+ return Nothing+ [] ->+ {- This shouldn't happen. See about using non-empty lists. -}+ return Nothing+ shutdown = return ()++ doJob :: (MonadLoggerIO m, Default s, Event e o s, Eq e)+ => TVar (PartitionWorkerState e o s m)+ -> (ClusterPowerState, PartitionPowerStateT e o s m ())+ -> m ()+ doJob tvar (cluster, action) = do+ $(logDebug) . T.pack $ "Starting job on " ++ show key+ PWS {+ pwsCm,+ pwsSelf,+ pwsPersistence,+ pwsCacheVal+ } <- liftIO (atomically (readTVar tvar))+ partition <- case pwsCacheVal of+ Nothing ->+ fromMaybe (PS.new key (C.findOwners key cluster))+ <$> liftIO (getState pwsPersistence key)+ Just partition -> return partition+ PM.runPowerStateT pwsSelf partition (+ action <* (removeObsolete >> PM.acknowledge)+ ) >>= \case+ Left err ->+ $(logError) . T.pack+ $ "Partition error: " ++ show (err, key)+ Right ((), propAction, newPartition, _outputs) -> do+ liftIO . atomically . modifyTVar tvar $ (\pws ->+ pws {pwsCacheVal = Just newPartition}+ )+ liftIO (saveState pwsPersistence key (Just newPartition))+ case propAction of+ Send -> sequence_ [+ CM.send pwsCm p (PartitionMerge key newPartition)+ | p <- Set.toList (PS.allParticipants newPartition)+ , p /= pwsSelf+ ]+ DoNothing -> return ()+ $(logDebug) . T.pack $ "Finished job on " ++ show key+ where+ {- |+ Remove obsolete peers. Obsolete peers are peers that are no longer+ participating in the replication of this partition, due to a+ rebalance. Such peers are removed lazily here at read time.+ -}+ removeObsolete :: (Monad m, Event e o s, Eq e)+ => PartitionPowerStateT e o s m ()+ removeObsolete = do+ let owners = C.findOwners key cluster+ peers <- PS.projParticipants <$> PM.getPowerState+ let obsolete = peers \\ owners+ mapM_+ (\peer -> PM.disassociate peer >> PM.acknowledgeAs peer)+ (Set.toList obsolete)+++{- | Lookup a key from a map, and also delete the key if it exists. -}+lookupAndDelete :: (Ord k) => k -> Map k v -> (Maybe v, Map k v)+lookupAndDelete = Map.updateLookupWithKey (const (const Nothing))+++{- |+ Figure out the set of nodes to which search requests should be+ dispatched. "Minimum complete service set" means the minimum set+ of peers that, together, service the whole partition key space;+ thereby guaranteeing that if any particular partition is indexed,+ the corresponding index record will exist on one of these peers.++ Implementation considerations:++ There will usually be more than one solution for the MCSS. For now,+ we just compute a deterministic solution, but we should implement+ a random (or pseudo-random) solution in order to maximally balance+ cluster resources.++ Also, it is not clear that the minimum complete service set is even+ what we really want. MCSS will reduce overall network utilization,+ but it may actually increase latency. If we were to dispatch redundant+ requests to multiple nodes, we could continue with whichever request+ returns first, and ignore the slow responses. This is probably the+ best solution. We will call this "fastest competitive search".++ TODO: implement fastest competitive search.+-}+minimumCompleteServiceSet :: (Monad m) => RuntimeT e o s m (Set Peer)+minimumCompleteServiceSet = do+ RuntimeState {cluster} <- RuntimeT get+ return (D.minimumCompleteServiceSet (C.getDistribution cluster))++
+ src/Network/Legion/SocketUtil.hs view
@@ -0,0 +1,19 @@++{- | Socket utilities. -}+module Network.Legion.SocketUtil (+ fam,+) where+++import Network.Socket (SockAddr, SockAddr(SockAddrInet, SockAddrInet6,+ SockAddrUnix, SockAddrCan), Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN))+++{- | 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/StateMachine.hs
@@ -1,597 +0,0 @@-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE TemplateHaskell #-}-{- |- 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.-- 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.)-- 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(- -- * Running the state machine.- newNodeState,-- -- * State machine inputs.- userRequest,- partitionMerge,- clusterMerge,- eject,- join,- minimumCompleteServiceSet,- search,-- joinNext,- joinNextResponse,-- -- * State machine outputs.- UserResponse(..),-- -- * State inspection- getPeers,- getPartition,-) where--import Control.Monad (void, unless)-import Control.Monad.Catch (throwM, MonadThrow)-import Control.Monad.IO.Class (MonadIO, liftIO)-import Control.Monad.Logger (logDebug, logError, MonadLoggerIO, logWarn)-import Control.Monad.Trans.Class (lift)-import Data.Bool (bool)-import Data.Conduit ((=$=), runConduit, transPipe, awaitForever)-import Data.Default.Class (Default)-import Data.Map (Map)-import Data.Maybe (fromMaybe)-import Data.Set (Set, (\\), member)-import Data.Text (pack)-import Network.Legion.Application (getState, saveState, list, saveCluster)-import Network.Legion.BSockAddr (BSockAddr)-import Network.Legion.ClusterState (ClusterPowerState, ClusterPowerStateT)-import Network.Legion.Distribution (Peer, newPeer, RebalanceAction(Invite,- Drop))-import Network.Legion.Index (IndexRecord(IndexRecord), stTag, stKey,- irTag, irKey, SearchTag(SearchTag), indexEntries, Indexable)-import Network.Legion.KeySet (KeySet)-import Network.Legion.PartitionKey (PartitionKey)-import Network.Legion.PartitionState (PartitionPowerState, PartitionPowerStateT)-import Network.Legion.PowerState (Event)-import Network.Legion.PowerState.Monad (PropAction(Send, DoNothing))-import Network.Legion.StateMachine.Monad (SM, NodeState(NodeState),- ClusterAction(PartitionMerge, ClusterMerge, PartitionJoin),- self, cluster, partitions, nsIndex, getPersistence, getNodeState,- modifyNodeState, pushActions, joins, lastRebalance)-import qualified Data.Conduit.List as CL-import qualified Data.Map as Map-import qualified Data.Set as Set-import qualified Network.Legion.ClusterState as C-import qualified Network.Legion.Distribution as D-import qualified Network.Legion.KeySet as KS-import qualified Network.Legion.PowerState as PS-import qualified Network.Legion.PowerState.Monad as PM---{- | Make a new node state. -}-newNodeState :: Peer -> ClusterPowerState -> NodeState e o s-newNodeState self cluster =- NodeState {- self,- cluster,- partitions = Map.empty,- nsIndex = Set.empty,- joins = Map.empty,- lastRebalance = minBound- }---{- | Handle a user request. -}-userRequest :: (- Default s,- Eq e,- Event e o s,- Indexable s,- MonadLoggerIO m,- MonadThrow m,- Show e,- Show s- )- => PartitionKey- -> e- -> SM e o s m (UserResponse o)-userRequest key request = do- NodeState {self, cluster} <- getNodeState- let routes = C.findRoute key cluster- if self `Set.member` routes- then do- (response, _) <- runPartitionPowerStateT key (- PM.event request- )- return (Respond response)-- else case Set.toList routes of- [] -> do- let msg = "No routes for key: " ++ show key- $(logError) . pack $ msg- error msg- peer:_ -> return (Forward peer)---{- |- Handle the state transition for a partition merge event. Returns 'Left'- if there is an error, and 'Right' if everything went fine.--}-partitionMerge :: (- Default s,- Eq e,- Event e o s,- Indexable s,- MonadLoggerIO m,- MonadThrow m,- Show e,- Show s- )- => PartitionKey- -> PartitionPowerState e o s- -> SM e o s m ()-partitionMerge key foreignPartition =- void $ runPartitionPowerStateT key (PM.merge foreignPartition)---{- | Handle the state transition for a cluster merge event. -}-clusterMerge :: (- Default s,- Eq e,- Event e o s,- Indexable s,- MonadLoggerIO m,- MonadThrow m,- Show e,- Show s- )- => ClusterPowerState- -> SM e o s m ()-clusterMerge foreignCluster = do- runClusterPowerStateT (PM.merge foreignCluster)- nodeState@NodeState {lastRebalance, cluster, self} <- getNodeState- $(logDebug) . pack- $ "Next Rebalance: "- ++ show (lastRebalance, C.nextAction cluster, nodeState)- case C.nextAction cluster of- (ord, Invite peer keys) | ord > lastRebalance && peer == self -> do- {-- The current action is an Invite, and this peer is the target.-- Send the join request message to every peer, update lastRebalance- so we don't repeat this on every trivial cluster merge, update- the expected joins so we can keep track of progress, then sit- back and wait.- -}- let- askPeers =- Set.toList . Set.delete self . Map.keysSet . C.getPeers $ cluster- pushActions [- PartitionJoin p keys- | p <- askPeers- ]- modifyNodeState (\ns -> ns {- joins = Map.fromList [- (p, keys)- | p <- askPeers- ],- lastRebalance = ord- })- (ord, Drop peer keys) | ord > lastRebalance && peer == self -> do- persistence <- getPersistence- runConduit (- transPipe liftIO (list persistence)- =$= CL.map fst- =$= CL.filter (`KS.member` keys)- =$= awaitForever (\key ->- lift $ runPartitionPowerStateT key (- PM.disassociate self- )- )- )- modifyNodeState (\ns -> ns {- lastRebalance = ord- })- runClusterPowerStateT C.finishRebalance- _ -> return ()---{- | Eject a peer from the cluster. -}-eject :: (MonadLoggerIO m, MonadThrow m) => Peer -> SM e o s m ()-eject peer = do- {-- 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?- -}- runClusterPowerStateT (C.eject peer)- {-- 'runClusterPowerStateT (C.eject peer)' will cause us to attempt to- notify the peer that they have been ejected, but that notification- is almost certainly going to go unacknowledged because the peer- is probably down.- - This call to 'eject' was presumably invoked as a result of user- action, and we must therefore trust the user to know that the peer- is really down and not coming back. This "guarantee" allows us to- acknowledge the ejection on the peer's behalf.-- This call will acknowledge the drop on behalf of the peer, and also- remove that peer from the keyspace distribution map.- -}- runClusterPowerStateTAs peer (return ())---{- | Handle a peer join request. -}-join :: (MonadIO m, MonadThrow m)- => BSockAddr- -> SM e o s m (Peer, ClusterPowerState)-join peerAddr = do- peer <- newPeer- void $ runClusterPowerStateT (C.joinCluster peer peerAddr)- NodeState {cluster} <- getNodeState- return (peer, cluster)---{- |- Figure out the set of nodes to which search requests should be- dispatched. "Minimum complete service set" means the minimum set- of peers that, together, service the whole partition key space;- thereby guaranteeing that if any particular partition is indexed,- the corresponding index record will exist on one of these peers.-- Implementation considerations:-- There will usually be more than one solution for the MCSS. For now,- we just compute a deterministic solution, but we should implement- a random (or pseudo-random) solution in order to maximally balance- cluster resources.-- Also, it is not clear that the minimum complete service set is even- what we really want. MCSS will reduce overall network utilization,- but it may actually increase latency. If we were to dispatch redundant- requests to multiple nodes, we could continue with whichever request- returns first, and ignore the slow responses. This is probably the- best solution. We will call this "fastest competitive search".-- TODO: implement fastest competitive search.--}-minimumCompleteServiceSet :: (Monad m) => SM e o s m (Set Peer)-minimumCompleteServiceSet = do- NodeState {cluster} <- getNodeState- return (D.minimumCompleteServiceSet (C.getDistribution cluster))---{- |- Search the index, and return the first record that is __strictly- greater than__ the provided search tag, if such a record exists.--}-search :: (Monad m) => SearchTag -> SM e o s m (Maybe IndexRecord)-search SearchTag {stTag, stKey = Nothing} = do- NodeState {nsIndex} <- getNodeState- return (Set.lookupGE IndexRecord {irTag = stTag, irKey = minBound} nsIndex)-search SearchTag {stTag, stKey = Just key} = do- NodeState {nsIndex} <- getNodeState- return (Set.lookupGT IndexRecord {irTag = stTag, irKey = key} nsIndex)---{- |- Allow a peer to participate in the replication of the partition that is- __greater than or equal to__ the indicated partition key. Returns @Nothing@- if there is no such partition, or @Just (key, partition)@ where @key@ is the- partition key that was joined and @partition@ is the resulting partition- power state.--}-joinNext :: (- Default s,- Eq e,- Event e o s,- Indexable s,- MonadLoggerIO m,- MonadThrow m- )- => Peer- -> KeySet- -> SM e o s m (Maybe (PartitionKey, PartitionPowerState e o s))-joinNext peer askKeys = do- persistence <- getPersistence- runConduit (- transPipe liftIO (list persistence)- =$= CL.filter ((`KS.member` askKeys) . fst)- =$= CL.head- ) >>= \case- Nothing -> return Nothing- Just (gotKey, partition) -> do- {-- This is very similar to the 'runPartitionPowerStateT' code,- but there are some important differences. First, 'list' has- already done to the trouble of fetching the partition value,- so we don't want to have 'runPartitionPowerStateT' do it- again. Second, and more importantly, 'runPartitionPowerStateT'- will cause a 'PartitionMerge' message to be sent to @peer@, but- that message would be redundant, because it contains a subset- of the information contained within the 'JoinNextResponse'- message that this function produces.- -}- NodeState {self} <- getNodeState- PM.runPowerStateT self partition (do- PM.participate peer- PM.acknowledge- ) >>= \case- Left err -> throwM err- Right ((), action, partition2, _infOutputs) -> do- case action of- Send -> pushActions [- PartitionMerge p gotKey partition2- | p <- Set.toList (PS.allParticipants partition2)- {-- Don't send a 'PartitionMerge' to @peer@. We- are already going to send it a more informative- 'JoinNextResponse'- -}- , p /= peer- , p /= self- ]- DoNothing -> return ()- savePartition gotKey partition2- return (Just (gotKey, partition2))---{- | Receive the result of a JoinNext request. -}-joinNextResponse :: (- Default s,- Eq e,- Event e o s,- Indexable s,- MonadLoggerIO m,- MonadThrow m,- Show e,- Show s- )- => Peer- -> Maybe (PartitionKey, PartitionPowerState e o s)- -> SM e o s m ()-joinNextResponse peer response = do- NodeState {cluster, lastRebalance} <- getNodeState- if lastRebalance > fst (C.nextAction cluster)- then- {- We are receiving messages from an old rebalance. Log and ignore. -}- $(logWarn) . pack- $ "Received an old join response: "- ++ show (peer, response, cluster, lastRebalance)- else do- case response of- Just (key, partition) -> do- partitionMerge key partition- NodeState {joins} <- getNodeState- case (KS.\\ KS.fromRange minBound key) <$> Map.lookup peer joins of- Nothing ->- {- An unexpected peer sent us this message, Ignore. TODO log. -}- return ()- Just needsJoinSet -> do- unless (KS.null needsJoinSet)- (pushActions [PartitionJoin peer needsJoinSet])- modifyNodeState (\ns -> ns {- joins = Map.filter- (not . KS.null)- (Map.insert peer needsJoinSet joins)- })- Nothing ->- modifyNodeState (\ns@NodeState {joins} -> ns {- joins = Map.delete peer joins- })- Map.null . joins <$> getNodeState >>= bool- (return ())- (runClusterPowerStateT C.finishRebalance)---{- |- The type of response to a user request, either forward to another node,- or respond directly.--}-data UserResponse o- = Forward Peer- | Respond o---{- | Get the known peer data from the cluster. -}-getPeers :: (Monad m) => SM e o s m (Map Peer BSockAddr)-getPeers = C.getPeers . cluster <$> getNodeState---{- | Gets a partition state. -}-getPartition :: (Default s, MonadIO m)- => PartitionKey- -> SM e o s m (PartitionPowerState e o s)-getPartition key = do- persistence <- getPersistence- NodeState {partitions, cluster} <- getNodeState- case Map.lookup key partitions of- Nothing ->- fromMaybe (PS.new key (C.findOwners key cluster)) <$>- liftIO (getState persistence key)- Just partition -> return partition---{- |- Saves a partition state. This function automatically handles the cache- for active propagations, as well as reindexing of partitions.--}-savePartition :: (Default s, Event e o s, Indexable s, MonadLoggerIO m)- => PartitionKey- -> PartitionPowerState e o s- -> SM e o s m ()-savePartition key partition = do- persistence <- getPersistence- oldTags <- indexEntries . PS.projectedValue <$> getPartition key- let- currentTags = indexEntries (PS.projectedValue partition)- {- TODO: maybe use Set.mapMonotonic for performance? -}- obsoleteRecords = Set.map (flip IndexRecord key) (oldTags \\ currentTags)- newRecords = Set.map (flip IndexRecord key) currentTags-- $(logDebug) . pack- $ "Tagging " ++ show key ++ " with: "- ++ show (currentTags, obsoleteRecords, newRecords)-- NodeState {self} <- getNodeState- liftIO (saveState persistence key (- if self `member` PS.allParticipants partition- then Just partition- else Nothing- ))- modifyNodeState (\ns@NodeState {partitions, nsIndex} ->- nsIndex `seq`- ns {- partitions = if Set.null (PS.divergent 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,- nsIndex = (nsIndex \\ obsoleteRecords) `Set.union` newRecords- }- )----- {- |--- Create the log message for origin conflict errors. The reason this--- function only creates the log message, instead of doing the logging--- as well, is because doing the logging here would screw up the source--- location that the template-haskell logging functions generate for us.--- -}--- originError :: (Show o) => DifferentOrigins o -> Text--- originError (DifferentOrigins a b) = pack--- $ "Tried to merge powerstates with different origins: "--- ++ show (a, b)---{- | Run a partition-flavored 'PowerStateT' in the 'SM' monad. -}-runPartitionPowerStateT :: (- Default s,- Eq e,- Event e o s,- Indexable s,- MonadLoggerIO m,- MonadThrow m,- Show e,- Show s- )- => PartitionKey- -> PartitionPowerStateT e o s (SM e o s m) a- -> SM e o s m (a, PartitionPowerState e o s)-runPartitionPowerStateT key m = do- NodeState {self} <- getNodeState- partition <- getPartition key- PM.runPowerStateT self partition (- m <* (removeObsolete >> PM.acknowledge)- ) >>= \case- Left err -> throwM err- Right (a, action, partition2, _infOutputs) -> do- case action of- Send -> pushActions [- PartitionMerge p key partition2- | p <- Set.toList (PS.allParticipants partition2)- , p /= self- ]- DoNothing -> return ()- $(logDebug) . pack- $ "Partition update: " ++ show partition- ++ " --> " ++ show partition2 ++ " :: " ++ show action- savePartition key partition2- return (a, partition2)- where- {- |- Remove obsolete peers. Obsolete peers are peers that are no longer- participating in the replication of this partition, due to a- rebalance. Such peers are removed lazily here at read time.- -}- removeObsolete :: (Eq e, Event e o s, Monad m)- => PartitionPowerStateT e o s (SM e o s m) ()- removeObsolete = do- owners <- C.findOwners key . cluster <$> lift getNodeState- peers <- PS.projParticipants <$> PM.getPowerState- let obsolete = peers \\ owners- mapM_- (\peer -> PM.disassociate peer >> PM.acknowledgeAs peer)- (Set.toList obsolete)---{- | Like 'runClusterPowerStateTAs', but run as the local peer. -}-runClusterPowerStateT :: (MonadThrow m, MonadIO m)- => ClusterPowerStateT (SM e o s m) a- -> SM e o s m a-runClusterPowerStateT m = do- NodeState {self} <- getNodeState- runClusterPowerStateTAs self m---{- |- Run a clusterstate-flavored 'PowerStateT' in the 'SM' monad,- automatically acknowledging the resulting power state.-- Generalized to run as any peer, in order to support exceptional cases- like 'eject'.--}-runClusterPowerStateTAs :: (MonadThrow m, MonadIO m)- => Peer {- ^ The peer to run as. -}- -> ClusterPowerStateT (SM e o s m) a- -> SM e o s m a-runClusterPowerStateTAs as m = do- NodeState {cluster, self} <- getNodeState- PM.runPowerStateT as cluster (m <* PM.acknowledge) >>= \case- Left err -> throwM err- Right (a, action, cluster2, _outputs) -> do- getPersistence >>= \p -> liftIO (saveCluster p self cluster2)- case action of- Send -> pushActions [- ClusterMerge p cluster2- | p <- Set.toList (PS.allParticipants cluster2)- , p /= self- ]- DoNothing -> return ()- modifyNodeState (\ns -> ns {cluster = cluster2})- return a--
− src/Network/Legion/StateMachine/Monad.hs
@@ -1,165 +0,0 @@-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE OverloadedStrings #-}-{- |- This module contains the legion state machine monad and some- primitives for manipulating the state. It is the foundation upon wish- the 'Network.Legion.StateMachine' module is built. It is separate from- that module because some of the primitives we export here go some small- way to avoiding bugs that might arise if that module had direct access- to the internals of this monad.--}-module Network.Legion.StateMachine.Monad (- -- * Run the monad- runSM,-- -- * State Inspection- getPersistence,- getNodeState,-- -- * State Modification- modifyNodeState,- pushActions,- popActions,-- -- * Other symbols- SM,- NodeState(..),- ClusterAction(..),-) where--import Control.Monad.Catch (MonadThrow)-import Control.Monad.IO.Class (MonadIO)-import Control.Monad.Logger (MonadLogger)-import Control.Monad.Trans.Class (lift, MonadTrans)-import Control.Monad.Trans.Reader (ReaderT, runReaderT, ask)-import Control.Monad.Trans.State (StateT, runStateT, get, modify, put)-import Data.Aeson (ToJSON, toJSON, object, (.=), encode)-import Data.ByteString.Lazy (toStrict)-import Data.Map (Map)-import Data.Set (Set)-import Data.Text (unpack)-import Data.Text.Encoding (decodeUtf8)-import Network.Legion.Application (Persistence)-import Network.Legion.ClusterState (ClusterPowerState, RebalanceOrd)-import Network.Legion.Distribution (Peer)-import Network.Legion.Index (IndexRecord)-import Network.Legion.KeySet (KeySet)-import Network.Legion.Lift (lift2, lift3)-import Network.Legion.PartitionKey (PartitionKey)-import Network.Legion.PartitionState (PartitionPowerState)-import qualified Data.Map as Map---{- |- Run an SM action.--}-runSM :: (Functor m)- => Persistence e o s- -> NodeState e o s- -> SM e o s m a- -> m (a, NodeState e o s, [ClusterAction e o s])-runSM p ns =- fmap flatten- . (`runStateT` [])- . (`runStateT` ns)- . (`runReaderT` p)- . unSM- where- flatten :: ((a, b), c) -> (a, b, c)- flatten ((a, b), c) = (a, b, c)---{- | Get the handle to the persistence layer. -}-getPersistence :: (Monad m) => SM e o s m (Persistence e o s)-getPersistence = SM ask---{- | Get the current node state. -}-getNodeState :: (Monad m) => SM e o s m (NodeState e o s)-getNodeState = (SM . lift) get---{- | Update current node state. -}-modifyNodeState :: (Monad m)- => (NodeState e o s -> NodeState e o s)- -> SM e o s m ()-modifyNodeState = SM . lift . modify---{- | Accumulate some cluster propagation actions. -}-pushActions :: (Monad m) => [ClusterAction e o s] -> SM e o s m ()-pushActions = SM . lift2 . modify . flip (++)---{- | Return and reset the accumulated cluster actions. -}-popActions :: (Monad m) => SM e o s m [ClusterAction e o s]-popActions = SM . lift2 $ do- actions <- get- put []- return actions---{- |- This monad encapsulates the global state of the legion node (not- counting the runtime stuff, like open connections and what have- you).-- 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 e o s m a = SM {- unSM ::- ReaderT (Persistence e o s) (- StateT (NodeState e o s) (- StateT [ClusterAction e o s]- m)) a- }- deriving (Functor, Applicative, Monad, MonadLogger, MonadIO, MonadThrow)-instance MonadTrans (SM e o s) where- lift = SM . lift3---{- |- This is the portion of the local node state that is not persistence- related.--}-data NodeState e o s = NodeState {- self :: Peer,- cluster :: ClusterPowerState,- partitions :: Map PartitionKey (PartitionPowerState e o s),- nsIndex :: Set IndexRecord,- joins :: Map Peer KeySet,- lastRebalance :: RebalanceOrd- }-instance (Show e, Show s) => Show (NodeState e o 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 e, Show s) => ToJSON (NodeState e o s) where- toJSON (NodeState self_ cluster_ partitions_ nsIndex_ joins_ lastUpdate_) =- object [- "self" .= show self_,- "cluster" .= cluster_,- "partitions" .= Map.map show (Map.mapKeys show partitions_),- "nsIndex" .= show nsIndex_,- "joins" .= Map.map show (Map.mapKeys show joins_),- "lastRebalance" .= show lastUpdate_- ]---{- |- 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.--}-data ClusterAction e o s- = PartitionMerge Peer PartitionKey (PartitionPowerState e o s)- | ClusterMerge Peer ClusterPowerState- | PartitionJoin Peer KeySet- deriving (Show)--