diff --git a/legion.cabal b/legion.cabal --- a/legion.cabal +++ b/legion.cabal @@ -1,8 +1,8 @@ --- Initial keyspace-partition.cabal generated by cabal init. For further +-- Initial keyspace-partition.cabal generated by cabal init. For further -- documentation, see http://haskell.org/cabal/users-guide/ name: legion -version: 0.2.0.0 +version: 0.3.0.0 synopsis: Distributed, stateful, homogeneous microservice framework. description: Legion is a framework for writing distributed, homogeneous, stateful microservices in Haskell. @@ -13,8 +13,9 @@ maintainer: rick@owenssoftware.com copyright: 2015-2016 Rick Owens category: Concurrency, Network +stability: experimental build-type: Simple --- extra-source-files: +-- extra-source-files: cabal-version: >=1.10 source-repository head @@ -22,9 +23,9 @@ location: git@github.com:taphu/legion.git library - exposed-modules: + exposed-modules: Network.Legion - other-modules: + other-modules: Network.Legion.Admin Network.Legion.Application Network.Legion.BSockAddr @@ -33,6 +34,7 @@ Network.Legion.Conduit Network.Legion.Distribution Network.Legion.Fork + Network.Legion.Index Network.Legion.KeySet Network.Legion.LIO Network.Legion.PartitionKey @@ -46,7 +48,7 @@ Network.Legion.StateMachine Network.Legion.UUID Paths_legion - -- other-extensions: + -- other-extensions: build-depends: Ranged-sets >= 0.3.0 && < 0.4, aeson >= 0.11.2.0 && < 0.12, diff --git a/src/Network/Legion.hs b/src/Network/Legion.hs --- a/src/Network/Legion.hs +++ b/src/Network/Legion.hs @@ -25,16 +25,24 @@ module Network.Legion ( -- * Service Implementation -- $service-implementaiton + + -- ** Indexing + -- $indexing + Legionary(..), LegionConstraints, Persistence(..), ApplyDelta(..), - RequestMsg, + Tag(..), -- * Invoking Legion -- $invocation forkLegionary, - runLegionary, StartupMode(..), + Runtime, + makeRequest, + search, + SearchTag(..), + IndexRecord(..), -- * Fundamental Types PartitionKey(..), PartitionPowerState, @@ -52,13 +60,15 @@ import Network.Legion.Application (LegionConstraints, Persistence(Persistence, getState, saveState, list), - Legionary(Legionary, persistence, handleRequest), RequestMsg) + Legionary(Legionary, persistence, handleRequest, index)) import Network.Legion.Basics (newMemoryPersistence, diskPersistence) -import Network.Legion.PartitionKey (PartitionKey(K, unkey)) +import Network.Legion.Index (Tag(Tag, unTag), IndexRecord(IndexRecord, + irTag, irKey), SearchTag(SearchTag, stTag, stKey)) +import Network.Legion.PartitionKey (PartitionKey(K, unKey)) import Network.Legion.PartitionState (PartitionPowerState, infimum, projected) import Network.Legion.PowerState (ApplyDelta(apply)) -import Network.Legion.Runtime (runLegionary, StartupMode(NewCluster, - JoinCluster), forkLegionary) +import Network.Legion.Runtime (StartupMode(NewCluster, JoinCluster), + forkLegionary, Runtime, makeRequest, search) import Network.Legion.Settings (LegionarySettings(LegionarySettings, adminHost, adminPort, peerBindAddr, joinBindAddr)) @@ -73,7 +83,7 @@ -- mainly on the stateful part, and it will do all the heavy lifting on -- that side of things. However, it is worth mentioning a few things about -- the stateless part before we move on. --- +-- -- The unit of state that Legion knows about is called a \"partition\". Each -- partition is identified by a 'PartitionKey', and it is replicated across -- the cluster. Each partition acts as the unit of state for handling @@ -83,20 +93,20 @@ -- the request in the first place. This is a function of the stateless -- part of the application. Generally speaking, the stateless part of -- your application is going to be responsible for --- +-- -- * Starting up the Legion runtime using 'forkLegionary'. -- * Identifying the partition key to which a request should be applied -- (e.g. maybe this is some component of a URL, or else an identifier -- stashed in a browser cookie). -- * Marshalling application requests into requests to the Legion runtime. -- * Marshalling the Legion runtime response into an application response. --- +-- -- Legion doesn't really address any of these things, mainly because there -- are already plenty of great ways to write stateless services. What -- Legion does provide is a runtime that can be embedded in the stateless -- part of your application, that transparently handles all of the hard -- stateful stuff, like replication, rebalancing, request routing, etc. --- +-- -- The only thing required to implement a legion service is to -- provide a request handler and a persistence layer by constructing a -- 'Legionary' value and passing it to 'forkLegionary'. The stateful @@ -111,7 +121,7 @@ -- stands for "output", which is the type of responses your application -- will generate in response to those requests, and @s@ stands for "state", -- which is the application state that each partition can assume. --- +-- -- Implementing a request handler is pretty straight forward, but -- there is a little bit more to it than meets the eye. If you look at -- 'forkLegionary', you will see a constraint named @'LegionConstraints' @@ -121,7 +131,7 @@ -- 'handleRequest', you will see that it is defined in terms of an input, -- an existing state, and an output, but there is no mention of any /new/ -- state that is generated as a result of handling the request. --- +-- -- This is where the 'ApplyDelta' typeclass comes in. Where 'handleRequest' -- takes an input and a state and produces an output, the 'apply' function -- of the 'ApplyDelta' typeclass takes an input and a state and produces @@ -137,7 +147,7 @@ -- only get called once for each input, but 'apply' has a very good -- chance of being called more than once for various reasons including -- re-playing the application of requests to resolve non-determinism. --- +-- -- Taking yet another look at 'handleRequest', you will see that it -- makes no provision for a non-existent partition state (i.e., it is -- written in terms of @s@, not @Maybe s@. Same goes for 'ApplyDelta'). @@ -149,14 +159,31 @@ -- instance of the 'Data.Default.Class.Default' typeclass). This doesn't -- take up infinite disk space because 'Data.Default.Class.def' values -- are cleverly encoded as a zero-length string of bytes. ;-) --- +-- -- The persistence layer provides the framework with a way to store the -- various partition states. This allows you to choose any number of -- persistence strategies, including only in memory, on disk, or in some -- external database. --- +-- -- See 'newMemoryPersistence' and 'diskPersistence' if you need to get -- started quickly with an in-memory persistence layer. + +-------------------------------------------------------------------------------- + +-- $indexing +-- Legion gives you a way to index your partitions so that you can find +-- partitions that have certain characteristics without having to know +-- the partition key a priori. Conceptually, the "index" is a single, +-- global, ordered list of 'IndexRecord's. The 'search' function allows +-- you to scroll forward through this list at will. +-- +-- Each partition may generate zero or more 'IndexRecord's. This +-- is determined by the 'index' function, which is defined by your +-- specific Legion application. For each 'Tag' returned by 'index', an +-- 'IndexRecord' is generated such that: +-- +-- > @IndexRecord {irTag = , irKey = }@ +-- -------------------------------------------------------------------------------- diff --git a/src/Network/Legion/Admin.hs b/src/Network/Legion/Admin.hs --- a/src/Network/Legion/Admin.hs +++ b/src/Network/Legion/Admin.hs @@ -109,7 +109,7 @@ Strip the server header -} stripServerHeader :: Middleware - stripServerHeader = modifyResponse (stripHeader "Server") + stripServerHeader = modifyResponse (stripHeader "Server") {- | Add our own server header. diff --git a/src/Network/Legion/Application.hs b/src/Network/Legion/Application.hs --- a/src/Network/Legion/Application.hs +++ b/src/Network/Legion/Application.hs @@ -7,12 +7,13 @@ LegionConstraints, Legionary(..), Persistence(..), - RequestMsg, ) where import Data.Binary (Binary) import Data.Conduit (Source) import Data.Default.Class (Default) +import Data.Set (Set) +import Network.Legion.Index (Tag) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.PartitionState (PartitionPowerState) import Network.Legion.PowerState (ApplyDelta) @@ -20,7 +21,7 @@ {- | This is a more convenient way to write the somewhat unwieldy set of constraints - + > ( > ApplyDelta i s, Default s, Binary i, Binary o, Binary s, Show i, > Show o, Show s, Eq i @@ -51,10 +52,16 @@ Given a request and a state, returns a response to the request. -} handleRequest :: PartitionKey -> i -> s -> o, + + {- | The user-defined persistence layer implementation. -} + persistence :: Persistence i s, + {- | - The user-defined persistence layer implementation. + A way of indexing partitions so that they can be found without + knowing the partition key. An index entry for the partition will be + created under each of the tags returned by this function. -} - persistence :: Persistence i s + index :: s -> Set Tag } @@ -74,19 +81,5 @@ conduit is terminated without reading the entire list. -} } - - -{- | - 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. - - Unless you know exactly what you are doing, you will have used - 'Network.Legion.forkLegionary' instead of 'Network.Legion.runLegionary' - to run the framework, in which case you can safely ignore the existence - of this type. --} -type RequestMsg i o = ((PartitionKey, i), o -> IO ()) diff --git a/src/Network/Legion/Basics.hs b/src/Network/Legion/Basics.hs --- a/src/Network/Legion/Basics.hs +++ b/src/Network/Legion/Basics.hs @@ -60,7 +60,7 @@ -> IO (Maybe (PartitionPowerState i s)) fetchState cacheT key = atomically $ lookup key <$> readTVar cacheT - + list_ :: TVar (Map PartitionKey (PartitionPowerState i s)) -> Source IO (PartitionKey, PartitionPowerState i s) @@ -106,7 +106,7 @@ list = do keys <- lift $ readHexList <$> getDirectoryContents directory sourceList keys =$= fillData - where + where fillData = awaitForever (\key -> do let path = toPath key state <- lift ((decode . fromStrict) <$> readFile path) diff --git a/src/Network/Legion/ClusterState.hs b/src/Network/Legion/ClusterState.hs --- a/src/Network/Legion/ClusterState.hs +++ b/src/Network/Legion/ClusterState.hs @@ -117,7 +117,7 @@ -> KeySet -> ClusterPropState -> ClusterPropState -claimParticipation peer ks = +claimParticipation peer ks = ClusterPropState . P.delta (Participating peer ks) . unPropState diff --git a/src/Network/Legion/Distribution.hs b/src/Network/Legion/Distribution.hs --- a/src/Network/Legion/Distribution.hs +++ b/src/Network/Legion/Distribution.hs @@ -13,6 +13,7 @@ rebalanceAction, RebalanceAction(..), newPeer, + minimumCompleteServiceSet, ) where import Prelude hiding (null) @@ -40,7 +41,6 @@ newtype Peer = Peer UUID deriving (Show, Binary, Eq, Ord) instance Read Peer where readPrec = Peer <$> readPrec - {- | @@ -56,17 +56,13 @@ ] -{- | - Constuct a distribution that contains no partitions. --} +{- | Constuct a distribution that contains no partitions. -} empty :: ParticipationDefaults empty = D [] -{- | - Find the peers that own the specified partition. --} +{- | Find the peers that own the specified partition. -} findPartition :: PartitionKey -> ParticipationDefaults -> Set Peer findPartition k d = @@ -78,6 +74,15 @@ ++ "via github: " ++ show (k, d) +{- | Find a solution to the minimum complete service set. -} +minimumCompleteServiceSet :: ParticipationDefaults -> Set Peer +minimumCompleteServiceSet defs = Set.fromList [ + p + | (_, peers) <- unD defs + , Just (p, _) <- [Set.minView peers] + ] + + {- | Modify the default participation for the key set. -} @@ -130,7 +135,7 @@ mostUnderserved = sortBy (compare `on` Set.size . snd) underserved in case mostUnderserved of [] -> Nothing - (ks, ps):_ -> + (ks, ps):_ -> let candidateHosts = toList (allPeers Set.\\ ps) bestHosts = sort [(weightOf p, p) | p <- candidateHosts] diff --git a/src/Network/Legion/Index.hs b/src/Network/Legion/Index.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/Index.hs @@ -0,0 +1,42 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{- | This module contains types related to partition indexing. -} +module Network.Legion.Index ( + Tag(..), + IndexRecord(..), + SearchTag(..), +) where + +import Data.Binary (Binary) +import Data.ByteString (ByteString) +import Data.String (IsString) +import GHC.Generics (Generic) +import Network.Legion.PartitionKey (PartitionKey) + + +{- | + A tag is a value associated with a partition state that can be used + to look up a partition key. +-} +newtype Tag = Tag {unTag :: ByteString} + deriving (Eq, Ord, Show, Binary, IsString) + + +{- | This data structure describes a record in the index. -} +data IndexRecord = IndexRecord { + irTag :: Tag, + irKey :: PartitionKey + } + deriving (Eq, Ord, Show, Generic) +instance Binary IndexRecord + + +{- | This data structure describes where in the index to start scrolling. -} +data SearchTag = SearchTag { + stTag :: Tag, + stKey :: Maybe PartitionKey + } + deriving (Show, Eq, Ord, Generic) +instance Binary SearchTag + + diff --git a/src/Network/Legion/KeySet.hs b/src/Network/Legion/KeySet.hs --- a/src/Network/Legion/KeySet.hs +++ b/src/Network/Legion/KeySet.hs @@ -26,7 +26,7 @@ BoundaryAbove, BoundaryAboveAll, BoundaryBelowAll), makeRangedSet, rSetHas, rSetUnion, (-!-), unsafeRangedSet, rSetRanges) import GHC.Generics (Generic) -import Network.Legion.PartitionKey (PartitionKey(K, unkey)) +import Network.Legion.PartitionKey (PartitionKey(K, unKey)) {- | @@ -162,7 +162,7 @@ To help with `rangeSize`. -} toI :: PartitionKey -> Integer -toI = toInteger . unkey +toI = toInteger . unKey {- | diff --git a/src/Network/Legion/LIO.hs b/src/Network/Legion/LIO.hs --- a/src/Network/Legion/LIO.hs +++ b/src/Network/Legion/LIO.hs @@ -9,9 +9,7 @@ import Control.Monad.Logger (LoggingT) -{- | - The logging monad in wich legion operates. --} +{- | The logging monad in wich legion operates. -} type LIO = LoggingT IO diff --git a/src/Network/Legion/PartitionKey.hs b/src/Network/Legion/PartitionKey.hs --- a/src/Network/Legion/PartitionKey.hs +++ b/src/Network/Legion/PartitionKey.hs @@ -20,7 +20,7 @@ {- | This is how partitions are identified and referenced. -} -newtype PartitionKey = K {unkey :: Word256} deriving (Eq, Ord, Show, Bounded) +newtype PartitionKey = K {unKey :: Word256} deriving (Eq, Ord, Show, Bounded) instance Binary PartitionKey where put (K (Word256 (Word128 a b) (Word128 c d))) = put (a, b, c, d) diff --git a/src/Network/Legion/PartitionState.hs b/src/Network/Legion/PartitionState.hs --- a/src/Network/Legion/PartitionState.hs +++ b/src/Network/Legion/PartitionState.hs @@ -19,7 +19,7 @@ projParticipants, projected, infimum, - complete, + idle, ) where import Data.Aeson (ToJSON) @@ -188,7 +188,7 @@ only way more work can happen is if new deltas are applied, either directly or via a merge. -} -complete :: PartitionPropState i s -> Bool -complete = P.complete . unPropState +idle :: PartitionPropState i s -> Bool +idle = P.idle . unPropState diff --git a/src/Network/Legion/PowerState.hs b/src/Network/Legion/PowerState.hs --- a/src/Network/Legion/PowerState.hs +++ b/src/Network/Legion/PowerState.hs @@ -232,7 +232,7 @@ {- | Record the fact that the participant acknowledges the information contained in the powerset. The implication is that the participant - **must** base all future operations on the result of this function. + __must__ base all future operations on the result of this function. -} acknowledge :: (ApplyDelta d s, Ord p) => p diff --git a/src/Network/Legion/Propagation.hs b/src/Network/Legion/Propagation.hs --- a/src/Network/Legion/Propagation.hs +++ b/src/Network/Legion/Propagation.hs @@ -27,7 +27,7 @@ projParticipants, projected, infimum, - complete, + idle, ) where import Prelude hiding (lookup) @@ -141,7 +141,7 @@ {- | Create a new propagation state. -} -new :: (Default s) => o -> p -> Set p -> PropState o s p d +new :: (Default s) => o -> p -> Set p -> PropState o s p d new origin self participants = PropState { powerState = PS.new origin participants, @@ -385,8 +385,8 @@ only way more work can happen is if new deltas are applied, either directly or via a merge. -} -complete :: (Ord p) => PropState o s p d -> Bool -complete PropState {powerState, peerStates} = +idle :: (Ord p) => PropState o s p d -> Bool +idle PropState {powerState, peerStates} = Map.null peerStates && Set.null (divergent powerState) diff --git a/src/Network/Legion/Runtime.hs b/src/Network/Legion/Runtime.hs --- a/src/Network/Legion/Runtime.hs +++ b/src/Network/Legion/Runtime.hs @@ -11,8 +11,10 @@ -} module Network.Legion.Runtime ( forkLegionary, - runLegionary, StartupMode(..), + Runtime, + makeRequest, + search, ) where import Control.Concurrent (forkIO) @@ -30,30 +32,34 @@ import Data.Conduit.Network (sourceSocket) import Data.Conduit.Serialization.Binary (conduitDecode) import Data.Map (Map) +import Data.Set (Set) import Data.Text (pack) import GHC.Generics (Generic) import Network.Legion.Admin (runAdmin, AdminMessage(GetState, GetPart, Eject)) import Network.Legion.Application (LegionConstraints, - Legionary(Legionary), RequestMsg, persistence, getState) + Legionary(Legionary), persistence, getState) 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)) import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.Runtime.ConnectionManager (newConnectionManager, send, ConnectionManager, newPeers) import Network.Legion.Runtime.PeerMessage (PeerMessage(PeerMessage), PeerMessagePayload(ForwardRequest, ForwardResponse, ClusterMerge, - PartitionMerge), MessageId, newSequence, next) + PartitionMerge, Search, SearchResponse), MessageId, newSequence, + nextMessageId) import Network.Legion.Settings (LegionarySettings(LegionarySettings, adminHost, adminPort, peerBindAddr, joinBindAddr)) import Network.Legion.StateMachine (partitionMerge, clusterMerge, NodeState, newNodeState, runSM, UserResponse(Forward, Respond), userRequest, heartbeat, rebalance, migrate, propagate, ClusterAction, - eject) + eject, minimumCompleteServiceSet) import Network.Legion.UUID (getUUID) import Network.Socket (Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN), SocketOption(ReuseAddr), SocketType(Stream), accept, bind, @@ -62,6 +68,7 @@ import Network.Socket.ByteString.Lazy (sendAll) 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.StateMachine as SM @@ -110,7 +117,8 @@ forwarded = Map.empty, nextId = firstMessageId, cm, - self + self, + searches = Map.empty } runConduit $ (joinS `merge` (peerS `merge` (requestSource `merge` adminS))) @@ -141,6 +149,20 @@ 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 i o + = Request PartitionKey i (o -> IO ()) + | SearchDispatch SearchTag (Maybe IndexRecord -> IO ()) +instance (Show i) => Show (RequestMsg i o) where + show (Request k i _) = "(Request " ++ show k ++ " " ++ show i ++ " _)" + show (SearchDispatch s _) = "(SearchDispatch " ++ show s ++ " _)" + + messageSink :: (LegionConstraints i o s) => Legionary i o s -> (RuntimeState i o s, NodeState i s) @@ -205,14 +227,14 @@ rts@RuntimeState {self, nextId, cm} = do send cm peer (PeerMessage self nextId (ClusterMerge ps)) - return rts {nextId = next nextId} + return rts {nextId = nextMessageId nextId} clusterAction (SM.PartitionMerge peer key ps) rts@RuntimeState {self, nextId, cm} = do send cm peer (PeerMessage self nextId (PartitionMerge key ps)) - return rts {nextId = next nextId} + return rts {nextId = nextMessageId nextId} {- | @@ -233,7 +255,7 @@ = do ((), ns2) <- runSM legionary ns (partitionMerge source key ps) return (rts, ns2) - + handleMessage {- Cluster Merge -} legionary (P (PeerMessage source _ (ClusterMerge cs))) @@ -253,11 +275,11 @@ send cm source ( PeerMessage self nextId (ForwardResponse mid response) ) - return (rts {nextId = next nextId}, ns2) + return (rts {nextId = nextMessageId nextId}, ns2) Forward peer -> do send cm peer msg - return (rts {nextId = next nextId}, ns2) - + return (rts {nextId = nextMessageId nextId}, ns2) + handleMessage {- Forward Response -} _legionary (msg@(P (PeerMessage _ _ (ForwardResponse mid response)))) @@ -273,7 +295,7 @@ handleMessage {- User Request -} legionary - (R ((key, request), respond)) + (R (Request key request respond)) (rts@RuntimeState {self, cm, nextId, forwarded}, ns) = do (output, ns2) <- runSM legionary ns (userRequest key request) @@ -288,11 +310,127 @@ return ( rts { forwarded = Map.insert nextId (lift . respond) forwarded, - nextId = next nextId + nextId = nextMessageId nextId }, ns2 ) +handleMessage {- Search Dispatch -} + {- + This is where we send out search request to all the appropriate + nodes in the cluster. + -} + legionary + (R (SearchDispatch searchTag respond)) + (rts@RuntimeState {cm, self, searches}, ns) + = + case Map.lookup searchTag searches of + Nothing -> do + {- + No identical search is currently being executed, kick off a + new one. + -} + (mcss, ns2) <- runSM legionary ns minimumCompleteServiceSet + rts2 <- foldr (>=>) return (sendOne <$> Set.toList mcss) rts + return ( + rts2 { + searches = Map.insert + searchTag + (mcss, Nothing, [lift . respond]) + searches + }, + ns2 + ) + Just (peers, best, responders) -> + {- + A search for this tag is already in progress, just add the + responder to the responder list. + -} + return ( + rts { + searches = Map.insert + searchTag + (peers, best, (lift . respond):responders) + searches + }, + ns + ) + where + sendOne :: Peer -> RuntimeState i o s -> LIO (RuntimeState i o s) + sendOne peer r@RuntimeState {nextId} = do + send cm peer (PeerMessage self nextId (Search searchTag)) + return r {nextId = nextMessageId nextId} + +handleMessage {- Search Execution -} + {- This is where we handle local search execution. -} + legionary + (P (PeerMessage source _ (Search searchTag))) + (rts@RuntimeState {nextId, cm, self}, ns) + = do + (output, ns2) <- runSM legionary ns (SM.search searchTag) + send cm source (PeerMessage self nextId (SearchResponse searchTag output)) + return (rts {nextId = nextMessageId nextId}, ns2) + +handleMessage {- Search Response -} + {- + This is where we gather all the responses from the various peers + to which we dispatched search requests. + -} + _legionary + (msg@(P (PeerMessage source _ (SearchResponse searchTag response)))) + (rts@RuntimeState {searches}, ns) + = + {- TODO: see if this function can't be made more elegant. -} + case Map.lookup searchTag searches of + Nothing -> do + {- There is no search happening. -} + $(logWarn) . pack $ "Unsolicited SearchResponse: " ++ show msg + return (rts, ns) + 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. + -} + mapM_ ($ bestOf best response) responders + return ( + rts {searches = Map.delete searchTag searches}, + ns + ) + else + {- We are still waiting on some outstanding requests. -} + return ( + rts { + searches = Map.insert + searchTag + (peers2, bestOf best response, responders) + searches + }, + ns + ) + else do + {- + There is a search happening, but the peer that responded + is not part of it. + -} + $(logWarn) . pack $ "Unsolicited SearchResponse: " ++ show msg + return (rts, ns) + where + {- | + Figure out which index record returned to us by the various peers + is the most appropriate to return. 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 -} legionary (J (JoinRequest addy, respond)) @@ -474,7 +612,7 @@ sourceSocket so =$= conduitDecode $$ do response <- await case response of - Nothing -> fail + Nothing -> fail $ "Couldn't join a cluster because there was no response " ++ "to our join request!" Just (JoinOk self cps) -> @@ -559,13 +697,13 @@ Forks the legion framework in a background thread, and returns a way to send user requests to it and retrieve the responses to those requests. -} -forkLegionary :: (LegionConstraints i o s, MonadLoggerIO io, MonadIO io2) +forkLegionary :: (LegionConstraints i o s, MonadLoggerIO io) => Legionary i o s {- ^ The user-defined legion application to run. -} -> LegionarySettings {- ^ Settings and configuration of the legionary framework. -} -> StartupMode - -> io (PartitionKey -> i -> io2 o) + -> io (Runtime i o) forkLegionary legionary settings startupMode = do logging <- askLoggerIO @@ -573,13 +711,60 @@ chan <- liftIO newChan forkC "main legion thread" $ runLegionary legionary settings startupMode (chanToSource chan) - return (\ key request -> liftIO $ do - responseVar <- newEmptyMVar - writeChan chan ((key, request), putMVar responseVar) - takeMVar responseVar - ) + 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' to access it. +-} +data Runtime i o = Runtime { + {- | + Send your customized request to the legion runtime, and get back + a response. + -} + rtMakeRequest :: PartitionKey -> i -> 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 i o -> PartitionKey -> i -> 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 i 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 i o s = P (PeerMessage i o s) @@ -588,17 +773,42 @@ | A (AdminMessage i o s) instance (Show i, Show o, Show s) => Show (RuntimeMessage i o s) where show (P m) = "(P " ++ show m ++ ")" - show (R ((p, i), _)) = "(R ((" ++ show p ++ ", " ++ show i ++ "), _))" + show (R m) = "(R " ++ show m ++ ")" show (J (jr, _)) = "(J (" ++ show jr ++ ", _))" show (A a) = "(A (" ++ show a ++ "))" -{- | The runtime state. -} +{- | + 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 identify otherwise identical searches. + +-} data RuntimeState i o s = RuntimeState { self :: Peer, forwarded :: Map MessageId (o -> LIO ()), nextId :: MessageId, - cm :: ConnectionManager i o s + cm :: ConnectionManager i o s, + searches :: Map + SearchTag + (Set Peer, Maybe IndexRecord, [Maybe IndexRecord -> LIO ()]) } diff --git a/src/Network/Legion/Runtime/PeerMessage.hs b/src/Network/Legion/Runtime/PeerMessage.hs --- a/src/Network/Legion/Runtime/PeerMessage.hs +++ b/src/Network/Legion/Runtime/PeerMessage.hs @@ -8,7 +8,7 @@ PeerMessagePayload(..), MessageId, newSequence, - next, + nextMessageId, ) where import Control.Monad.Trans.Class (lift) @@ -18,6 +18,7 @@ import GHC.Generics (Generic) import Network.Legion.ClusterState (ClusterPowerState) import Network.Legion.Distribution (Peer) +import Network.Legion.Index (SearchTag, IndexRecord) import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.PartitionState (PartitionPowerState) @@ -49,6 +50,8 @@ | ForwardRequest PartitionKey i | ForwardResponse MessageId o | ClusterMerge ClusterPowerState + | Search SearchTag + | SearchResponse SearchTag (Maybe IndexRecord) deriving (Generic, Show) instance (Binary i, Binary o, Binary s) => Binary (PeerMessagePayload i o s) @@ -73,7 +76,7 @@ `succ` for this kind of thing, but making `MessageId` an instance of `Enum` really isn't appropriate. -} -next :: MessageId -> MessageId -next (M sequenceId ord) = M sequenceId (ord + 1) +nextMessageId :: MessageId -> MessageId +nextMessageId (M sequenceId ord) = M sequenceId (ord + 1) diff --git a/src/Network/Legion/StateMachine.hs b/src/Network/Legion/StateMachine.hs --- a/src/Network/Legion/StateMachine.hs +++ b/src/Network/Legion/StateMachine.hs @@ -50,6 +50,8 @@ heartbeat, eject, join, + minimumCompleteServiceSet, + search, -- * State machine outputs. ClusterAction(..), @@ -71,16 +73,18 @@ import Data.Default.Class (Default) import Data.Map (Map) import Data.Maybe (fromMaybe) -import Data.Set ((\\)) +import Data.Set (Set, (\\)) import Data.Text (pack, unpack) import Data.Text.Encoding (decodeUtf8) import Data.Time.Clock (getCurrentTime) import Network.Legion.Application (Legionary(Legionary), getState, - saveState, list, persistence, handleRequest) + saveState, list, persistence, handleRequest, index) import Network.Legion.BSockAddr (BSockAddr) import Network.Legion.ClusterState (ClusterPropState, ClusterPowerState) import Network.Legion.Distribution (Peer, rebalanceAction, newPeer, RebalanceAction(Invite)) +import Network.Legion.Index (IndexRecord(IndexRecord), stTag, stKey, + irTag, irKey, SearchTag(SearchTag)) import Network.Legion.KeySet (KeySet, union) import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey) @@ -90,6 +94,7 @@ 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.PartitionState as P @@ -102,7 +107,8 @@ self :: Peer, cluster :: ClusterPropState, partitions :: Map PartitionKey (PartitionPropState i s), - migration :: KeySet + migration :: KeySet, + nsIndex :: Set IndexRecord } instance (Show i, Show s) => Show (NodeState i s) where show = unpack . decodeUtf8 . toStrict . encode @@ -111,12 +117,13 @@ instance is very hard to read. -} instance (Show i, Show s) => ToJSON (NodeState i s) where - toJSON (NodeState self cluster partitions migration) = + toJSON (NodeState self cluster partitions migration nsIndex) = object [ "self" .= show self, "cluster" .= cluster, "partitions" .= Map.mapKeys show partitions, - "migration" .= show migration + "migration" .= show migration, + "nsIndex" .= show nsIndex ] @@ -129,7 +136,8 @@ self, cluster, partitions = Map.empty, - migration = KS.empty + migration = KS.empty, + nsIndex = Set.empty } @@ -236,7 +244,7 @@ peer to a partition. This will cause the data to be transfered in the normal course of propagation. -} -migrate :: (ApplyDelta i s) => SM i o s () +migrate :: (Default s, ApplyDelta i s) => SM i o s () migrate = do NodeState {migration} <- (SM . lift) get Legionary {persistence} <- SM ask @@ -246,7 +254,7 @@ $$ accum (SM . lift) $ modify (\ns -> ns {migration = KS.empty}) where - accum :: (ApplyDelta i s) + accum :: (Default s, ApplyDelta i s) => Sink (PartitionKey, PartitionPowerState i s) (SM i o s) () accum = awaitForever $ \ (key, ps) -> do NodeState {self, cluster, partitions} <- (lift . SM . lift) get @@ -283,7 +291,7 @@ newPartitions = Map.fromAscList [ (key, newPartition) | (key, newPartition, _) <- updates - , not (P.complete newPartition) + , not (P.idle newPartition) ] (lift . put) ns { partitions = newPartitions @@ -353,6 +361,48 @@ {- | + 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 :: SM i o s (Set Peer) +minimumCompleteServiceSet = SM $ do + NodeState {cluster} <- lift get + 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 :: SearchTag -> SM i o s (Maybe IndexRecord) +search SearchTag {stTag, stKey = Nothing} = SM $ do + NodeState {nsIndex} <- lift get + return (Set.lookupGE IndexRecord {irTag = stTag, irKey = minBound} nsIndex) +search SearchTag {stTag, stKey = Just key} = SM $ do + NodeState {nsIndex} <- lift get + return (Set.lookupGT IndexRecord {irTag = stTag, irKey = key} nsIndex) + + +{- | 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. @@ -391,18 +441,35 @@ Just partition -> return partition -{- | Saves a partition state. -} -savePartition :: PartitionKey -> PartitionPropState i s -> SM i o s () +{- | + Saves a partition state. This function automatically handles the cache + for active propagations, as well as reindexing of partitions. +-} +savePartition :: (Default s, ApplyDelta i s) + => PartitionKey + -> PartitionPropState i s + -> SM i o s () savePartition key partition = SM $ do - Legionary {persistence} <- ask - ns@NodeState {partitions} <- lift get + Legionary {persistence, index} <- ask + oldTags <- index . P.ask <$> unSM (getPartition key) + let + currentTags = index (P.ask 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) + + ns@NodeState {partitions, nsIndex} <- lift get lift3 (saveState persistence key ( if P.participating partition then Just (P.getPowerState partition) else Nothing )) lift $ put ns { - partitions = if P.complete partition + partitions = if P.idle partition then {- Remove the partition from the working cache because there @@ -411,7 +478,8 @@ -} Map.delete key partitions else - Map.insert key partition partitions + Map.insert key partition partitions, + nsIndex = (nsIndex \\ obsoleteRecords) `Set.union` newRecords }