legion 0.5.0.1 → 0.6.0.0
raw patch · 13 files changed
+281/−325 lines, 13 files
Files
- legion.cabal +1/−1
- src/Network/Legion.hs +34/−60
- src/Network/Legion/Admin.hs +2/−2
- src/Network/Legion/Application.hs +8/−42
- src/Network/Legion/Basics.hs +11/−11
- src/Network/Legion/ClusterState.hs +7/−7
- src/Network/Legion/Index.hs +12/−0
- src/Network/Legion/PartitionState.hs +28/−28
- src/Network/Legion/PowerState.hs +43/−44
- src/Network/Legion/Propagation.hs +43/−43
- src/Network/Legion/Runtime.hs +58/−51
- src/Network/Legion/Runtime/PeerMessage.hs +1/−1
- src/Network/Legion/StateMachine.hs +33/−35
legion.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: legion-version: 0.5.0.1+version: 0.6.0.0 synopsis: Distributed, stateful, homogeneous microservice framework. description: Legion is a framework for writing distributed, homogeneous, stateful microservices in Haskell.
src/Network/Legion.hs view
@@ -44,7 +44,7 @@ -- ** Indexing -- $indexing- Legionary(..),+ Indexable(..), LegionConstraints, Persistence(..), ApplyDelta(..),@@ -64,11 +64,11 @@ import Prelude hiding (lookup, readFile, writeFile, null) import Network.Legion.Application (LegionConstraints,- Persistence(Persistence, getState, saveState, list),- Legionary(Legionary, persistence, handleRequest, index))+ Persistence(Persistence, getState, saveState, list)) import Network.Legion.Basics (newMemoryPersistence, diskPersistence) import Network.Legion.Index (Tag(Tag, unTag), IndexRecord(IndexRecord,- irTag, irKey), SearchTag(SearchTag, stTag, stKey))+ irTag, irKey), SearchTag(SearchTag, stTag, stKey),+ Indexable(indexEntries)) import Network.Legion.PartitionKey (PartitionKey(K, unKey)) import Network.Legion.PartitionState (PartitionPowerState) import Network.Legion.PowerState (ApplyDelta(apply))@@ -112,58 +112,27 @@ -- 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--- part of your application will live mostly within the request handler--- 'handleRequest'. If you look at 'handleRequest', you will see that--- it is abstract over the type variables @i@, @o@, and @s@.------ > handleRequest :: PartitionKey -> i -> s -> o------ These are the types your application has to fill in. @i@ stands for--- "input", which is the type of requests your application accepts; @o@--- 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.---+-- The only thing required to implement a legion service is to implement+-- a few typeclasses and call 'forkLegionary'. The state-aware part of+-- your application will live mostly within the request handler, which+-- is implemented via a typeclass `ApplyDelta`.+-- +-- > class ApplyDelta i o s | i s -> o where+-- > apply :: i -> s -> (o, s)+-- +-- If you look at 'apply', you will see that it is abstract over the type+-- variables @i@, @o@, and @s@. These are the types your application+-- has to fill in. @i@ stands for "input", which is the type of requests+-- your application accepts; @o@ 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'--- i o s@, which is short-hand for a long list of typeclasses that--- your @i@, @o@, and @s@ types are going to have to implement. Of--- particular interest is the 'ApplyDelta' typeclass. If you look at--- '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--- a new state.------ > apply :: i -> s -> s------ The reason that Legion splits the definition of what it means to--- fully handle an input into two functions like this is because of the--- approach it takes to solving distributed systems problems. Describing--- this entirely is beyond the scope of this section of documentation--- (TODO link to more info) but the TL;DR is that 'handleRequest' will--- 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').--- This framework takes the somewhat platonic philosophical view that all--- mathematical values exist somewhere and that there is no such thing as--- non-existent partition. When you first spin up a Legion application,--- all of those partitions are going to have a default value, which is--- 'Data.Default.Class.def' (Because your partition state must be an--- 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. ;-)+-- i o s@, which is short-hand for a long list of typeclasses that your+-- @i@, @o@, and @s@ types are going to have to implement. -- -- The persistence layer provides the framework with a way to store the -- various partition states. This allows you to choose any number of@@ -181,14 +150,19 @@ -- 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 = <your tag>, irKey = <partition key>}@+--+-- Indexing is implemented by instantiating the 'Indexable' typeclass+-- for your state type.+--+-- > class Indexable s where+-- > indexEntries :: s -> Set Tag+--+-- The tags returned by 'indexEntries' is used to construct a set of zero+-- or more 'IndexRecord's. For each 'Tag' returned by 'indexEntries',+-- an 'IndexRecord' is generated such that: -- +-- > IndexRecord {irTag = <your tag>, irKey = <partition key>}+ --------------------------------------------------------------------------------
src/Network/Legion/Admin.hs view
@@ -128,8 +128,8 @@ The type of messages sent by the admin service. -} data AdminMessage i o s- = GetState (NodeState i s -> LIO ())- | GetPart PartitionKey (Maybe (PartitionPowerState i s) -> LIO ())+ = GetState (NodeState i o s -> LIO ())+ | GetPart PartitionKey (Maybe (PartitionPowerState i o s) -> LIO ()) | Eject Peer (() -> LIO ()) instance Show (AdminMessage i o s) where
src/Network/Legion/Application.hs view
@@ -5,15 +5,13 @@ -} module Network.Legion.Application ( LegionConstraints,- Legionary(..), Persistence(..), ) 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.Index (Indexable) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.PartitionState (PartitionPowerState) import Network.Legion.PowerState (ApplyDelta)@@ -23,57 +21,25 @@ constraints > (- > ApplyDelta i s, Default s, Binary i, Binary o, Binary s, Show i,+ > ApplyDelta i o s, Default s, Binary i, Binary o, Binary s, Show i, > Show o, Show s, Eq i > ) -} type LegionConstraints i o s = (- ApplyDelta i s, Default s, Binary i, Binary o, Binary s, Show i,- Show o, Show s, Eq i+ ApplyDelta i o s, Indexable s, Default s, Binary i, Binary o, Binary s,+ Show i, Show o, Show s, Eq i ) {- |- This is the type of a user-defined Legion application. Implement this and- allow the Legion framework to manage your cluster.-- - @__i__@ is the type of request your application will handle. @__i__@ stands- for __"input"__.- - @__o__@ is the type of response produced by your application. @__o__@ stands- for __"output"__- - @__s__@ is the type of state maintained by your application. More- precisely, it is the type of the individual partitions that make up- your global application state. @__s__@ stands for __"state"__.--}-data Legionary i o s = Legionary {- {- |- The request handler, implemented by the user to service requests.-- Given a request and a state, returns a response to the request.- -}- handleRequest :: i -> s -> o,-- {- | The user-defined persistence layer implementation. -}- persistence :: Persistence i s,-- {- |- 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.- -}- index :: s -> Set Tag- }---{- | The type of a user-defined persistence strategy used to persist partition states. See 'Network.Legion.newMemoryPersistence' or 'Network.Legion.diskPersistence' if you need to get started quickly. -}-data Persistence i s = Persistence {- getState :: PartitionKey -> IO (Maybe (PartitionPowerState i s)),- saveState :: PartitionKey -> Maybe (PartitionPowerState i s) -> IO (),- list :: Source IO (PartitionKey, PartitionPowerState i s)+data Persistence i o s = Persistence {+ getState :: PartitionKey -> IO (Maybe (PartitionPowerState i o s)),+ saveState :: PartitionKey -> Maybe (PartitionPowerState i o s) -> IO (),+ list :: Source IO (PartitionKey, PartitionPowerState i o s) {- ^ List all the keys known to the persistence layer. It is important that the implementation do the right thing with regard to
src/Network/Legion/Basics.hs view
@@ -33,7 +33,7 @@ A convenient memory-based persistence layer. Good for testing or for applications (like caches) that don't have durability requirements. -}-newMemoryPersistence :: IO (Persistence i s)+newMemoryPersistence :: IO (Persistence i o s) newMemoryPersistence = do cacheT <- atomically (newTVar Map.empty)@@ -44,9 +44,9 @@ } where saveState_- :: TVar (Map PartitionKey (PartitionPowerState i s))+ :: TVar (Map PartitionKey (PartitionPowerState i o s)) -> PartitionKey- -> Maybe (PartitionPowerState i s)+ -> Maybe (PartitionPowerState i o s) -> IO () saveState_ cacheT key (Just state) = (atomically . modifyTVar cacheT . insert key) state@@ -55,15 +55,15 @@ (atomically . modifyTVar cacheT . Map.delete) key fetchState- :: TVar (Map PartitionKey (PartitionPowerState i s))+ :: TVar (Map PartitionKey (PartitionPowerState i o s)) -> PartitionKey- -> IO (Maybe (PartitionPowerState i s))+ -> IO (Maybe (PartitionPowerState i o s)) fetchState cacheT key = atomically $ lookup key <$> readTVar cacheT list_- :: TVar (Map PartitionKey (PartitionPowerState i s))- -> Source IO (PartitionKey, PartitionPowerState i s)+ :: TVar (Map PartitionKey (PartitionPowerState i o s))+ -> Source IO (PartitionKey, PartitionPowerState i o s) list_ cacheT = sourceList . Map.toList =<< lift (atomically (readTVar cacheT)) @@ -72,7 +72,7 @@ diskPersistence :: (Binary i, Binary s) => FilePath -- ^ The directory under which partition states will be stored.- -> Persistence i s+ -> Persistence i o s diskPersistence directory = Persistence { getState,@@ -82,7 +82,7 @@ where getState :: (Binary i, Binary s) => PartitionKey- -> IO (Maybe (PartitionPowerState i s))+ -> IO (Maybe (PartitionPowerState i o s)) getState key = let path = toPath key in doesFileExist path >>= bool@@ -91,7 +91,7 @@ saveState :: (Binary i, Binary s) => PartitionKey- -> Maybe (PartitionPowerState i s)+ -> Maybe (PartitionPowerState i o s) -> IO () saveState key (Just state) = writeFile (toPath key) (toStrict (encode state))@@ -102,7 +102,7 @@ (removeFile path) list :: (Binary i, Binary s)- => Source IO (PartitionKey, PartitionPowerState i s)+ => Source IO (PartitionKey, PartitionPowerState i o s) list = do keys <- lift $ readHexList <$> getDirectoryContents directory sourceList keys =$= fillData
src/Network/Legion/ClusterState.hs view
@@ -75,7 +75,7 @@ A representation of all possible cluster states. -} newtype ClusterPowerState = ClusterPowerState {- unPowerState :: PropPowerState UUID ClusterState Peer Update+ unPowerState :: PropPowerState UUID ClusterState Peer Update () } deriving (Show, Binary) @@ -84,7 +84,7 @@ cluster state. -} newtype ClusterPropState = ClusterPropState {- unPropState :: PropState UUID ClusterState Peer Update+ unPropState :: PropState UUID ClusterState Peer Update () } deriving (Show, ToJSON) @@ -97,16 +97,16 @@ | PeerEjected Peer deriving (Show, Generic) instance Binary Update-instance ApplyDelta Update ClusterState where+instance ApplyDelta Update () ClusterState where apply (PeerJoined peer addr) cs@ClusterState {peers} =- cs {peers = Map.insert peer addr peers}+ ((), cs {peers = Map.insert peer addr peers}) apply (Participating peer ks) cs@ClusterState {distribution} =- cs {distribution = modify (Set.insert peer) ks distribution}+ ((), cs {distribution = modify (Set.insert peer) ks distribution}) apply (PeerEjected peer) cs@ClusterState {distribution, peers} =- cs {+ ((), cs { distribution = modify (Set.delete peer) full distribution, peers = Map.delete peer peers- }+ }) {- |
src/Network/Legion/Index.hs view
@@ -4,14 +4,26 @@ module Network.Legion.Index ( Tag(..), IndexRecord(..),+ Indexable(..), SearchTag(..), ) where import Data.Binary (Binary) import Data.ByteString (ByteString)+import Data.Set (Set) import Data.String (IsString) import GHC.Generics (Generic) import Network.Legion.PartitionKey (PartitionKey)+++{- | This typeclass provides the ability to index partition states. -}+class Indexable s where+ {- |+ 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.+ -}+ indexEntries :: s -> Set Tag {- |
src/Network/Legion/PartitionState.hs view
@@ -43,8 +43,8 @@ You can save these guys to disk in your `Network.Legion.Persistence` layer by using its `Binary` instance. -}-newtype PartitionPowerState i s = PartitionPowerState {- unPowerState :: PropPowerState PartitionKey s Peer i+newtype PartitionPowerState i o s = PartitionPowerState {+ unPowerState :: PropPowerState PartitionKey s Peer i o } deriving (Show, Binary) @@ -52,8 +52,8 @@ A reification of `PropState`, representing the propagation state of the partition state. -}-newtype PartitionPropState i s = PartitionPropState {- unPropState :: PropState PartitionKey s Peer i+newtype PartitionPropState i o s = PartitionPropState {+ unPropState :: PropState PartitionKey s Peer i o } deriving (Eq, Show, ToJSON) @@ -66,18 +66,18 @@ {- | Get the projected partition state value. -}-ask :: (ApplyDelta i s) => PartitionPropState i s -> s+ask :: (ApplyDelta i o s) => PartitionPropState i o s -> s ask = P.ask . unPropState {- | Try to merge two partition states. -}-mergeEither :: (Show i, Show s, ApplyDelta i s)+mergeEither :: (Show i, Show s, ApplyDelta i o s) => Peer- -> PartitionPowerState i s- -> PartitionPropState i s- -> Either String (PartitionPropState i s)+ -> PartitionPowerState i o s+ -> PartitionPropState i o s+ -> Either String (PartitionPropState i o s) mergeEither peer ps prop = PartitionPropState <$> P.mergeEither peer (unPowerState ps) (unPropState prop)@@ -89,8 +89,8 @@ state that is applicable after those actions have been taken. -} actions- :: PartitionPropState i s- -> (Set Peer, PartitionPowerState i s, PartitionPropState i s)+ :: PartitionPropState i o s+ -> (Set Peer, PartitionPowerState i o s, PartitionPropState i o s) actions prop = let (peers, ps, newProp) = P.actions (unPropState prop) in (peers, PartitionPowerState ps, PartitionPropState newProp)@@ -106,7 +106,7 @@ {- ^ self -} -> Set Peer {- ^ The default participation. -}- -> PartitionPropState i s+ -> PartitionPropState i o s new key self = PartitionPropState . P.new key self @@ -114,10 +114,10 @@ Initialize a `PartitionPropState` based on the initial underlying partition power state. -}-initProp :: (ApplyDelta i s)+initProp :: (ApplyDelta i o s) => Peer- -> PartitionPowerState i s- -> PartitionPropState i s+ -> PartitionPowerState i o s+ -> PartitionPropState i o s initProp self = PartitionPropState . P.initProp self . unPowerState @@ -125,7 +125,7 @@ Return `True` if the local peer is participating in the partition power state. -}-participating :: PartitionPropState i s -> Bool+participating :: PartitionPropState i o s -> Bool participating = P.participating . unPropState @@ -133,30 +133,30 @@ Get an opaque encapsulation of the partition power state, for transferring accros the network or whatever. -}-getPowerState :: PartitionPropState i s -> PartitionPowerState i s+getPowerState :: PartitionPropState i o s -> PartitionPowerState i o s getPowerState = PartitionPowerState . P.getPowerState . unPropState {- | Apply a delta to the partition state. -}-delta :: (ApplyDelta i s)+delta :: (ApplyDelta i o s) => i- -> PartitionPropState i s- -> PartitionPropState i s+ -> PartitionPropState i o s+ -> PartitionPropState i o s delta d = PartitionPropState . P.delta d . unPropState {- | Move time forward for the propagation state. -}-heartbeat :: UTCTime -> PartitionPropState i s -> PartitionPropState i s+heartbeat :: UTCTime -> PartitionPropState i o s -> PartitionPropState i o s heartbeat now = PartitionPropState . P.heartbeat now . unPropState {- | Allow a participant to join in the distributed nature of the power state. -}-participate :: (ApplyDelta i s)+participate :: (ApplyDelta i o s) => Peer- -> PartitionPropState i s- -> PartitionPropState i s+ -> PartitionPropState i o s+ -> PartitionPropState i o s participate peer = PartitionPropState . P.participate peer . unPropState @@ -164,21 +164,21 @@ Return the projected peers which are participating in the partition state. -}-projParticipants :: PartitionPropState i s -> Set Peer+projParticipants :: PartitionPropState i o s -> Set Peer projParticipants = P.projParticipants . unPropState {- | Get the projected value of a `PartitionPowerState`. -}-projected :: (ApplyDelta i s) => PartitionPowerState i s -> s+projected :: (ApplyDelta i o s) => PartitionPowerState i o s -> s projected = P.projected . unPowerState {- | Get the infimum value of a `PartitionPowerState` -}-infimum :: PartitionPowerState i s -> s+infimum :: PartitionPowerState i o s -> s infimum = P.infimum . unPowerState @@ -188,7 +188,7 @@ only way more work can happen is if new deltas are applied, either directly or via a merge. -}-idle :: PartitionPropState i s -> Bool+idle :: PartitionPropState i o s -> Bool idle = P.idle . unPropState
src/Network/Legion/PowerState.hs view
@@ -1,10 +1,9 @@ {-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FunctionalDependencies #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-}-{- |- This module contains the fundamental distributed data object.--}+{- | This module contains the fundamental distributed data object. -} module Network.Legion.PowerState ( PowerState, Infimum(..),@@ -46,13 +45,13 @@ This represents the set of all possible future values of @s@, in a distributed, monotonically increasing environment. -}-data PowerState o s p d = PowerState {+data PowerState o s p d r = PowerState { origin :: o, infimum :: Infimum s p, deltas :: Map (StateId p) (Delta p d, Set p) } deriving (Generic, Show, Eq)-instance (Binary o, Binary s, Binary p, Binary d) => Binary (PowerState o s p d)-instance (Show o, Show s, Show p, Show d) => ToJSON (PowerState o s p d) where+instance (Binary o, Binary s, Binary p, Binary d) => Binary (PowerState o s p d r)+instance (Show o, Show s, Show p, Show d) => ToJSON (PowerState o s p d r) where toJSON PowerState {origin, infimum, deltas} = object [ "origin" .= show origin, "infimum" .= infimum,@@ -125,17 +124,17 @@ {- | The class which allows for delta application. -}-class ApplyDelta i s where+class ApplyDelta i o s | i s -> o where {- | Apply a delta to a state value. *This function MUST be total!!!* -}- apply :: i -> s -> s+ apply :: i -> s -> (o, s) {- | Construct a new PowerState with the given origin and initial participants -}-new :: (Default s) => o -> Set p -> PowerState o s p d+new :: (Default s) => o -> Set p -> PowerState o s p d r new origin participants = PowerState { origin,@@ -154,10 +153,10 @@ a lower one. This function is not total. Only `PowerState`s that originated from the same `new` call can be merged. -}-merge :: (Eq o, ApplyDelta d s, Ord p, Show o, Show s, Show p, Show d)- => PowerState o s p d- -> PowerState o s p d- -> PowerState o s p d+merge :: (Eq o, ApplyDelta d r s, Ord p, Show o, Show s, Show p, Show d)+ => PowerState o s p d r+ -> PowerState o s p d r+ -> PowerState o s p d r merge a b = either error id (mergeEither a b) @@ -165,10 +164,10 @@ Like `merge`, but safe. Returns `Nothing` if the two power states do not share the same origin. -}-mergeMaybe :: (Eq o, ApplyDelta d s, Ord p, Show o, Show s, Show p, Show d)- => PowerState o s p d- -> PowerState o s p d- -> Maybe (PowerState o s p d)+mergeMaybe :: (Eq o, ApplyDelta d r s, Ord p, Show o, Show s, Show p, Show d)+ => PowerState o s p d r+ -> PowerState o s p d r+ -> Maybe (PowerState o s p d r) mergeMaybe a b = either (const Nothing) Just (mergeEither a b) @@ -176,10 +175,10 @@ Like `mergeMaybe`, but returns a human-decipherable error message of exactly what went wrong. -}-mergeEither :: (Eq o, ApplyDelta d s, Ord p, Show o, Show s, Show p, Show d)- => PowerState o s p d- -> PowerState o s p d- -> Either String (PowerState o s p d)+mergeEither :: (Eq o, ApplyDelta d r s, Ord p, Show o, Show s, Show p, Show d)+ => PowerState o s p d r+ -> PowerState o s p d r+ -> Either String (PowerState o s p d r) mergeEither (PowerState o1 i1 d1) (PowerState o2 i2 d2) | o1 == o2 = Right . reduce . removeRenegade $ PowerState { origin = o1,@@ -234,10 +233,10 @@ contained in the powerset. The implication is that the participant __must__ base all future operations on the result of this function. -}-acknowledge :: (ApplyDelta d s, Ord p)+acknowledge :: (ApplyDelta d r s, Ord p) => p- -> PowerState o s p d- -> PowerState o s p d+ -> PowerState o s p d r+ -> PowerState o s p d r acknowledge p ps@PowerState {deltas} = reduce ps {deltas = fmap ackOne deltas} where@@ -247,10 +246,10 @@ {- | Allow a participant to join in the distributed nature of the power state. -}-participate :: (ApplyDelta d s, Ord p)+participate :: (ApplyDelta d r s, Ord p) => p- -> PowerState o s p d- -> PowerState o s p d+ -> PowerState o s p d r+ -> PowerState o s p d r participate p ps@PowerState {deltas} = acknowledge p $ ps { deltas = Map.insert (nextId p ps) (Join p, Set.empty) deltas }@@ -260,10 +259,10 @@ Indicate that a participant is removing itself from participating in the distributed power state. -}-disassociate :: (ApplyDelta d s, Ord p)+disassociate :: (ApplyDelta d r s, Ord p) => p- -> PowerState o s p d- -> PowerState o s p d+ -> PowerState o s p d r+ -> PowerState o s p d r disassociate p ps@PowerState {deltas} = acknowledge p $ ps { deltas = Map.insert (nextId p ps) (UnJoin p, Set.empty) deltas }@@ -272,11 +271,11 @@ {- | Introduce a change to the PowerState on behalf of the participant. -}-delta :: (ApplyDelta d s, Ord p)+delta :: (ApplyDelta d r s, Ord p) => p -> d- -> PowerState o s p d- -> PowerState o s p d+ -> PowerState o s p d r+ -> PowerState o s p d r delta p d ps@PowerState {deltas} = acknowledge p $ ps { deltas = Map.insert (nextId p ps) (Delta d, Set.empty) deltas }@@ -285,9 +284,9 @@ {- | Return the current projected value of the power state. -}-projectedValue :: (ApplyDelta d s) => PowerState o s p d -> s+projectedValue :: (ApplyDelta d r s) => PowerState o s p d r -> s projectedValue PowerState {infimum = Infimum {stateValue}, deltas} =- foldr apply stateValue changes+ foldr (\ i s -> snd (apply i s)) stateValue changes where changes = foldr getDeltas [] (toDescList deltas) getDeltas (_, (Delta d, _)) acc = d:acc@@ -297,14 +296,14 @@ {- | Return the current infimum value of the power state. -}-infimumValue :: PowerState o s p d -> s+infimumValue :: PowerState o s p d r -> s infimumValue PowerState {infimum = Infimum {stateValue}} = stateValue {- | Gets the known participants at the infimum. -}-infimumParticipants :: PowerState o s p d -> Set p+infimumParticipants :: PowerState o s p d r -> Set p infimumParticipants PowerState {infimum = Infimum {participants}} = participants @@ -312,7 +311,7 @@ Get all known participants. This includes participants that are projected for removal. -}-allParticipants :: (Ord p) => PowerState o s p d -> Set p+allParticipants :: (Ord p) => PowerState o s p d r -> Set p allParticipants PowerState { infimum = Infimum {participants}, deltas@@ -327,7 +326,7 @@ Get all the projected participants. This does not include participants that are projected for removal. -}-projParticipants :: (Ord p) => PowerState o s p d -> Set p+projParticipants :: (Ord p) => PowerState o s p d r -> Set p projParticipants PowerState { infimum = Infimum {participants}, deltas@@ -344,7 +343,7 @@ context, a peer is "diverging" if there is a delta that the peer has not acknowledged. -}-divergent :: (Ord p) => PowerState o s p d -> Set p+divergent :: (Ord p) => PowerState o s p d r -> Set p divergent PowerState { infimum = Infimum {participants}, deltas@@ -384,7 +383,7 @@ {- | Return the deltas that are unknown to the specified peer. -}-divergences :: (Ord p) => p -> PowerState o s p d -> Map (StateId p) d+divergences :: (Ord p) => p -> PowerState o s p d r -> Map (StateId p) d divergences peer PowerState {deltas} = fromAscList [ (sid, d)@@ -398,7 +397,7 @@ has enough information to derive a new infimum value. In other words, this is where garbage collection happens. -}-reduce :: (ApplyDelta d s, Ord p) => PowerState o s p d -> PowerState o s p d+reduce :: (ApplyDelta d r s, Ord p) => PowerState o s p d r -> PowerState o s p d r reduce ps@PowerState { infimum = infimum@Infimum {participants, stateValue}, deltas@@ -426,7 +425,7 @@ Delta d -> reduce ps { infimum = infimum { stateId = i,- stateValue = apply d stateValue+ stateValue = snd (apply d stateValue) }, deltas = newDeltas }@@ -436,7 +435,7 @@ A utility function that constructs the next `StateId` on behalf of a participant. -}-nextId :: (Ord p) => p -> PowerState o s p d -> StateId p+nextId :: (Ord p) => p -> PowerState o s p d r -> StateId p nextId p PowerState {infimum = Infimum {stateId}, deltas} = case maximum (stateId:keys deltas) of BottomSid -> Sid 0 p
src/Network/Legion/Propagation.hs view
@@ -65,13 +65,13 @@ the power state remains consistent with the state of its propagation throughout the network. -}-data PropState o s p d = PropState {- powerState :: PowerState o s p d,+data PropState o s p d r = PropState {+ powerState :: PowerState o s p d r, peerStates :: Map p PeerStatus, self :: p, now :: Time } deriving (Eq, Show)-instance (Show o, Show s, Show p, Show d) => ToJSON (PropState o s p d) where+instance (Show o, Show s, Show p, Show d) => ToJSON (PropState o s p d r) where toJSON PropState {powerState, peerStates, self, now} = object [ "powerState" .= powerState, "peerStates" .= Map.fromList [@@ -89,25 +89,25 @@ it over the network, but we don't want any code outside of this module to operate on it. -}-newtype PropPowerState o s p d = PropPowerState {- unPowerState :: PowerState o s p d+newtype PropPowerState o s p d r = PropPowerState {+ unPowerState :: PowerState o s p d r } deriving (Show, Binary) {- | Retriev the current projected value of the underlying state. -}-ask :: (ApplyDelta d s) => PropState o s p d -> s+ask :: (ApplyDelta d r s) => PropState o s p d r -> s ask = projectedValue . powerState {- | Create a new propagation state based on an existing power state. -}-initProp :: (ApplyDelta d s, Ord p)+initProp :: (ApplyDelta d r s, Ord p) => p- -> PropPowerState o s p d- -> PropState o s p d+ -> PropPowerState o s p d r+ -> PropState o s p d r initProp self ps = let powerState = acknowledge self (unPowerState ps) in PropState {@@ -125,7 +125,7 @@ Return an opaque representation of the power state, for transfer across the network, or whatever. -}-getPowerState :: PropState o s p d -> PropPowerState o s p d+getPowerState :: PropState o s p d r -> PropPowerState o s p d r getPowerState = PropPowerState . powerState @@ -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 r new origin self participants = PropState { powerState = PS.new origin participants,@@ -155,11 +155,11 @@ Like `merge`, but total. `mergeEither` returns a human readable reason why the foreign powerstate can't be merged in the event of an error. -}-mergeEither :: (Eq o, Ord p, Show o, Show s, Show p, Show d, ApplyDelta d s)+mergeEither :: (Eq o, Ord p, Show o, Show s, Show p, Show d, ApplyDelta d r s) => p- -> PropPowerState o s p d- -> PropState o s p d- -> Either String (PropState o s p d)+ -> PropPowerState o s p d r+ -> PropState o s p d r+ -> Either String (PropState o s p d r) mergeEither source kernel (prop@PropState {powerState, peerStates, self, now}) = let ps = unPowerState kernel in case acknowledge self <$> PS.mergeEither ps powerState of@@ -191,11 +191,11 @@ Like `merge`, but total. `mergeMaybe` returns `Nothing` if the foreign power state can't be merged. -}-mergeMaybe :: (Eq o, Ord p, Show o, Show s, Show p, Show d, ApplyDelta d s)+mergeMaybe :: (Eq o, Ord p, Show o, Show s, Show p, Show d, ApplyDelta d r s) => p- -> PropPowerState o s p d- -> PropState o s p d- -> Maybe (PropState o s p d)+ -> PropPowerState o s p d r+ -> PropState o s p d r+ -> Maybe (PropState o s p d r) mergeMaybe source ps prop = case mergeEither source ps prop of Left _ -> Nothing@@ -208,11 +208,11 @@ precondition is not met, `error` will be called (making this function non-total). Using `mergeMaybe` or `mergeEither` is recommended. -}-merge :: (Eq o, Ord p, Show o, Show s, Show p, Show d, ApplyDelta d s)+merge :: (Eq o, Ord p, Show o, Show s, Show p, Show d, ApplyDelta d r s) => p- -> PropPowerState o s p d- -> PropState o s p d- -> PropState o s p d+ -> PropPowerState o s p d r+ -> PropState o s p d r+ -> PropState o s p d r merge source ps prop = case mergeEither source ps prop of Left err -> error err@@ -222,17 +222,17 @@ {- | Time moves forward. -}-heartbeat :: UTCTime -> PropState o s p d -> PropState o s p d+heartbeat :: UTCTime -> PropState o s p d r -> PropState o s p d r heartbeat newNow prop = prop {now = max (now prop) (Just newNow)} {- | Apply a delta. -}-delta :: (Ord p, ApplyDelta d s)+delta :: (Ord p, ApplyDelta d r s) => d- -> PropState o s p d- -> PropState o s p d+ -> PropState o s p d r+ -> PropState o s p d r delta d prop@PropState {self, powerState, now} = let newPowerState = PS.delta self d powerState in prop {@@ -250,8 +250,8 @@ state that is applicable after those actions have been taken. -} actions :: (Eq p)- => PropState o s p d- -> (Set p, PropPowerState o s p d, PropState o s p d)+ => PropState o s p d r+ -> (Set p, PropPowerState o s p d r, PropState o s p d r) actions prop@PropState {powerState, peerStates, now} = (outOfDatePeers, PropPowerState powerState, newPropState) where@@ -292,10 +292,10 @@ {- | Allow a participant to join in the distributed nature of the power state. -}-participate :: (Ord p, ApplyDelta d s)+participate :: (Ord p, ApplyDelta d r s) => p- -> PropState o s p d- -> PropState o s p d+ -> PropState o s p d r+ -> PropState o s p d r participate peer prop@PropState {powerState, now} = let newPowerState = PS.participate peer powerState in prop {@@ -310,10 +310,10 @@ {- | Eject a participant from the power state. -}-disassociate :: (Ord p, ApplyDelta d s)+disassociate :: (Ord p, ApplyDelta d r s) => p- -> PropState o s p d- -> PropState o s p d+ -> PropState o s p d r+ -> PropState o s p d r disassociate peer prop@PropState {powerState, now} = let newPowerState = PS.disassociate peer powerState in prop {@@ -328,14 +328,14 @@ {- | Return the deltas that are unknown to the specified peer. -}-divergences :: (Ord p) => p -> PropState o s p d -> Map (StateId p) d+divergences :: (Ord p) => p -> PropState o s p d r -> Map (StateId p) d divergences peer = PS.divergences peer . powerState {- | Return self. -}-getSelf :: PropState o s p d -> p+getSelf :: PropState o s p d r -> p getSelf = self @@ -345,7 +345,7 @@ for removal, because until the infimum catches up to that projection, this peer still has an obligation to participate. -}-participating :: (Ord p) => PropState o s p d -> Bool+participating :: (Ord p) => PropState o s p d r -> Bool participating PropState{self, powerState} = self `member` PS.allParticipants powerState @@ -354,28 +354,28 @@ Get all known participants. This includes participants that are projected for removal. -}-allParticipants :: (Ord p) => PropState o s p d -> Set p+allParticipants :: (Ord p) => PropState o s p d r -> Set p allParticipants = PS.allParticipants . powerState {- | Get all of the projected participants. -}-projParticipants :: (Ord p) => PropState o s p d -> Set p+projParticipants :: (Ord p) => PropState o s p d r -> Set p projParticipants = PS.projParticipants . powerState {- | Get the projected value of a PropPowerState. -}-projected :: (ApplyDelta d s) => PropPowerState o s p d -> s+projected :: (ApplyDelta d r s) => PropPowerState o s p d r -> s projected = PS.projectedValue . unPowerState {- | Get the infimum value of the PropPowerState. -}-infimum :: PropPowerState o s p d -> s+infimum :: PropPowerState o s p d r -> s infimum = PS.infimumValue . unPowerState @@ -385,7 +385,7 @@ only way more work can happen is if new deltas are applied, either directly or via a merge. -}-idle :: (Ord p) => PropState o s p d -> Bool+idle :: (Ord p) => PropState o s p d r -> Bool idle PropState {powerState, peerStates} = Map.null peerStates && Set.null (divergent powerState)
src/Network/Legion/Runtime.hs view
@@ -37,8 +37,7 @@ import GHC.Generics (Generic) import Network.Legion.Admin (runAdmin, AdminMessage(GetState, GetPart, Eject))-import Network.Legion.Application (LegionConstraints,- Legionary(Legionary), persistence, getState)+import Network.Legion.Application (LegionConstraints, getState, Persistence) import Network.Legion.BSockAddr (BSockAddr(BSockAddr)) import Network.Legion.ClusterState (ClusterPowerState) import Network.Legion.Conduit (merge, chanToSink, chanToSource)@@ -84,10 +83,10 @@ what you are doing, you probably want to use `forkLegionary` instead. -} runLegionary :: (LegionConstraints i o s)- => Legionary i o s- {- ^ The user-defined legion application to run. -}+ => Persistence i o s+ {- ^ The persistence layer used to back the legion framework. -} -> RuntimeSettings- {- ^ Settings and configuration of the legionary framework. -}+ {- ^ Settings and configuration of the legionframework. -} -> StartupMode -> Source IO (RequestMsg i o) {- ^ A source of requests, together with a way to respond to the requets. -}@@ -98,7 +97,7 @@ -} runLegionary- legionary+ persistence settings@RuntimeSettings {adminHost, adminPort} startupMode requestSource@@ -123,7 +122,7 @@ runConduit $ (joinS `merge` (peerS `merge` (requestSource `merge` adminS))) =$= CL.map toMessage- =$= messageSink legionary (rts, nodeState)+ =$= messageSink persistence (rts, nodeState) where toMessage :: Either@@ -164,19 +163,19 @@ messageSink :: (LegionConstraints i o s)- => Legionary i o s- -> (RuntimeState i o s, NodeState i s)+ => Persistence i o s+ -> (RuntimeState i o s, NodeState i o s) -> Sink (RuntimeMessage i o s) LIO ()-messageSink legionary states =+messageSink persistence states = await >>= \case Nothing -> return () Just msg -> do $(logDebug) . pack $ "Receieved: " ++ show msg- lift . handleMessage legionary msg- >=> lift . updatePeers legionary- >=> lift . clusterHousekeeping legionary- >=> messageSink legionary+ lift . handleMessage persistence msg+ >=> lift . updatePeers persistence+ >=> lift . clusterHousekeeping persistence+ >=> messageSink persistence $ states @@ -185,11 +184,11 @@ joined the cluster. -} updatePeers- :: Legionary i o s- -> (RuntimeState i o s, NodeState i s)- -> LIO (RuntimeState i o s, NodeState i s)-updatePeers legionary (rts, ns) = do- (peers, ns2) <- runSM legionary ns SM.getPeers+ :: Persistence i o s+ -> (RuntimeState i o s, NodeState i o s)+ -> LIO (RuntimeState i o s, NodeState i o s)+updatePeers persistence (rts, ns) = do+ (peers, ns2) <- runSM persistence ns SM.getPeers newPeers (cm rts) peers return (rts, ns2) @@ -199,11 +198,11 @@ appropriately. -} clusterHousekeeping :: (LegionConstraints i o s)- => Legionary i o s- -> (RuntimeState i o s, NodeState i s)- -> LIO (RuntimeState i o s, NodeState i s)-clusterHousekeeping legionary (rts, ns) = do- (actions, ns2) <- runSM legionary ns (+ => Persistence i o s+ -> (RuntimeState i o s, NodeState i o s)+ -> LIO (RuntimeState i o s, NodeState i o s)+clusterHousekeeping persistence (rts, ns) = do+ (actions, ns2) <- runSM persistence ns ( heartbeat >> rebalance >> migrate@@ -218,7 +217,7 @@ machine. -} clusterAction- :: ClusterAction i s+ :: ClusterAction i o s -> RuntimeState i o s -> LIO (RuntimeState i o s) @@ -243,33 +242,33 @@ state and node state. -} handleMessage :: (LegionConstraints i o s)- => Legionary i o s+ => Persistence i o s -> RuntimeMessage i o s- -> (RuntimeState i o s, NodeState i s)- -> LIO (RuntimeState i o s, NodeState i s)+ -> (RuntimeState i o s, NodeState i o s)+ -> LIO (RuntimeState i o s, NodeState i o s) handleMessage {- Partition Merge -}- legionary+ persistence (P (PeerMessage source _ (PartitionMerge key ps))) (rts, ns) = do- ((), ns2) <- runSM legionary ns (partitionMerge source key ps)+ ((), ns2) <- runSM persistence ns (partitionMerge source key ps) return (rts, ns2) handleMessage {- Cluster Merge -}- legionary+ persistence (P (PeerMessage source _ (ClusterMerge cs))) (rts, ns) = do- ((), ns2) <- runSM legionary ns (clusterMerge source cs)+ ((), ns2) <- runSM persistence ns (clusterMerge source cs) return (rts, ns2) handleMessage {- Forward Request -}- legionary+ persistence (P (msg@(PeerMessage source mid (ForwardRequest key request)))) (rts@RuntimeState {nextId, cm, self}, ns) = do- (output, ns2) <- runSM legionary ns (userRequest key request)+ (output, ns2) <- runSM persistence ns (userRequest key request) case output of Respond response -> do send cm source (@@ -294,11 +293,11 @@ return (rts {forwarded = fwd}, ns) handleMessage {- User Request -}- legionary+ persistence (R (Request key request respond)) (rts@RuntimeState {self, cm, nextId, forwarded}, ns) = do- (output, ns2) <- runSM legionary ns (userRequest key request)+ (output, ns2) <- runSM persistence ns (userRequest key request) case output of Respond response -> do lift (respond response)@@ -320,7 +319,7 @@ This is where we send out search request to all the appropriate nodes in the cluster. -}- legionary+ persistence (R (SearchDispatch searchTag respond)) (rts@RuntimeState {cm, self, searches}, ns) =@@ -330,7 +329,7 @@ No identical search is currently being executed, kick off a new one. -}- (mcss, ns2) <- runSM legionary ns minimumCompleteServiceSet + (mcss, ns2) <- runSM persistence ns minimumCompleteServiceSet rts2 <- foldr (>=>) return (sendOne <$> Set.toList mcss) rts return ( rts2 {@@ -363,11 +362,11 @@ handleMessage {- Search Execution -} {- This is where we handle local search execution. -}- legionary+ persistence (P (PeerMessage source _ (Search searchTag))) (rts@RuntimeState {nextId, cm, self}, ns) = do- (output, ns2) <- runSM legionary ns (SM.search searchTag) + (output, ns2) <- runSM persistence ns (SM.search searchTag) send cm source (PeerMessage self nextId (SearchResponse searchTag output)) return (rts {nextId = nextMessageId nextId}, ns2) @@ -432,11 +431,11 @@ bestOf a Nothing = a handleMessage {- Join Request -}- legionary+ persistence (J (JoinRequest addy, respond)) (rts, ns) = do- ((peer, cluster), ns2) <- runSM legionary ns (SM.join addy)+ ((peer, cluster), ns2) <- runSM persistence ns (SM.join addy) respond (JoinOk peer cluster) return (rts, ns2) @@ -448,7 +447,7 @@ respond ns >> return (rts, ns) handleMessage {- Admin Get Partition -}- Legionary {persistence}+ persistence (A (GetPart key respond)) (rts, ns) = do@@ -456,7 +455,7 @@ return (rts, ns) handleMessage {- Admin Eject Peer -}- legionary+ persistence (A (Eject peer respond)) (rts, ns) = do@@ -483,7 +482,7 @@ "next state id" for a peer were global across all power states instead of local to each power state? -}- ((), ns2) <- runSM legionary ns (eject peer)+ ((), ns2) <- runSM persistence ns (eject peer) respond () return (rts, ns2) @@ -576,7 +575,7 @@ makeNodeState :: RuntimeSettings -> StartupMode- -> LIO (Peer, NodeState i s, Map Peer BSockAddr)+ -> LIO (Peer, NodeState i o s, Map Peer BSockAddr) makeNodeState RuntimeSettings {peerBindAddr} NewCluster = do {- Build a brand new node state, for the first node in a cluster. -}@@ -696,21 +695,29 @@ {- | 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.++ - @__i__@ is the type of request your application will handle. @__i__@ stands+ for __"input"__.+ - @__o__@ is the type of response produced by your application. @__o__@ stands+ for __"output"__+ - @__s__@ is the type of state maintained by your application. More+ precisely, it is the type of the individual partitions that make up+ your global application state. @__s__@ stands for __"state"__. -} forkLegionary :: (LegionConstraints i o s, MonadLoggerIO io)- => Legionary i o s- {- ^ The user-defined legion application to run. -}+ => Persistence i o s+ {- ^ The persistence layer used to back the legion framework. -} -> RuntimeSettings- {- ^ Settings and configuration of the legionary framework. -}+ {- ^ Settings and configuration of the legion framework. -} -> StartupMode -> io (Runtime i o) -forkLegionary legionary settings startupMode = do+forkLegionary persistence settings startupMode = do logging <- askLoggerIO liftIO . (`runLoggingT` logging) $ do chan <- liftIO newChan forkC "main legion thread" $- runLegionary legionary settings startupMode (chanToSource chan)+ runLegionary persistence settings startupMode (chanToSource chan) return Runtime { rtMakeRequest = \key request -> liftIO $ do responseVar <- newEmptyMVar
src/Network/Legion/Runtime/PeerMessage.hs view
@@ -46,7 +46,7 @@ cluster and the blacklisting of that node so that it can never re-join. -} data PeerMessagePayload i o s- = PartitionMerge PartitionKey (PartitionPowerState i s)+ = PartitionMerge PartitionKey (PartitionPowerState i o s) | ForwardRequest PartitionKey i | ForwardResponse MessageId o | ClusterMerge ClusterPowerState
src/Network/Legion/StateMachine.hs view
@@ -77,19 +77,18 @@ 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, index)+import Network.Legion.Application (getState, saveState, list, Persistence) 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))+ irTag, irKey, SearchTag(SearchTag), indexEntries, Indexable) import Network.Legion.KeySet (KeySet, union) import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.PartitionState (PartitionPowerState, PartitionPropState)-import Network.Legion.PowerState (ApplyDelta)+import Network.Legion.PowerState (ApplyDelta, apply) import qualified Data.Conduit.List as CL import qualified Data.Map as Map import qualified Data.Set as Set@@ -103,20 +102,20 @@ This is the portion of the local node state that is not persistence related. -}-data NodeState i s = NodeState {+data NodeState i o s = NodeState { self :: Peer, cluster :: ClusterPropState,- partitions :: Map PartitionKey (PartitionPropState i s),+ partitions :: Map PartitionKey (PartitionPropState i o s), migration :: KeySet, nsIndex :: Set IndexRecord }-instance (Show i, Show s) => Show (NodeState i s) where+instance (Show i, Show s) => Show (NodeState i 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 i, Show s) => ToJSON (NodeState i s) where+instance (Show i, Show s) => ToJSON (NodeState i o s) where toJSON (NodeState self cluster partitions migration nsIndex) = object [ "self" .= show self,@@ -130,7 +129,7 @@ {- | Make a new node state. -}-newNodeState :: Peer -> ClusterPropState -> NodeState i s+newNodeState :: Peer -> ClusterPropState -> NodeState i o s newNodeState self cluster = NodeState { self,@@ -152,7 +151,7 @@ we have to do so using a monad. -} newtype SM i o s a = SM {- unSM :: ReaderT (Legionary i o s) (StateT (NodeState i s) LIO) a+ unSM :: ReaderT (Persistence i o s) (StateT (NodeState i o s) LIO) a } deriving (Functor, Applicative, Monad, MonadLogger, MonadIO) @@ -161,27 +160,26 @@ Run an SM action. -} runSM- :: Legionary i o s- -> NodeState i s+ :: Persistence i o s+ -> NodeState i o s -> SM i o s a- -> LIO (a, NodeState i s)-runSM l ns action = runStateT (runReaderT (unSM action) l) ns+ -> LIO (a, NodeState i o s)+runSM p ns action = runStateT (runReaderT (unSM action) p) ns {- | Handle a user request. -}-userRequest :: (ApplyDelta i s, Default s)+userRequest :: (ApplyDelta i o s, Default s, Indexable s) => PartitionKey -> i -> SM i o s (UserResponse o) userRequest key request = SM $ do NodeState {self, cluster} <- lift get- Legionary {handleRequest} <- ask let owners = C.findPartition key cluster if self `Set.member` owners then do partition <- unSM $ getPartition key let- response = handleRequest request (P.ask partition)+ response = fst (apply request (P.ask partition)) partition2 = P.delta request partition unSM $ savePartition key partition2 return (Respond response)@@ -198,10 +196,10 @@ Handle the state transition for a partition merge event. Returns 'Left' if there is an error, and 'Right' if everything went fine. -}-partitionMerge :: (Show i, Show s, ApplyDelta i s, Default s)+partitionMerge :: (Show i, Show s, ApplyDelta i o s, Default s, Indexable s) => Peer -> PartitionKey- -> PartitionPowerState i s+ -> PartitionPowerState i o s -> SM i o s () partitionMerge source key foreignPartition = do partition <- getPartition key@@ -244,18 +242,18 @@ peer to a partition. This will cause the data to be transfered in the normal course of propagation. -}-migrate :: (Default s, ApplyDelta i s) => SM i o s ()+migrate :: (Default s, ApplyDelta i o s, Indexable s) => SM i o s () migrate = do NodeState {migration} <- (SM . lift) get- Legionary {persistence} <- SM ask+ persistence <- SM ask unless (KS.null migration) $ transPipe (SM . lift3) (list persistence) $= CL.filter ((`KS.member` migration) . fst) $$ accum (SM . lift) $ modify (\ns -> ns {migration = KS.empty}) where- accum :: (Default s, ApplyDelta i s)- => Sink (PartitionKey, PartitionPowerState i s) (SM i o s) ()+ accum :: (Default s, ApplyDelta i o s, Indexable s)+ => Sink (PartitionKey, PartitionPowerState i o s) (SM i o s) () accum = awaitForever $ \ (key, ps) -> do NodeState {self, cluster, partitions} <- (lift . SM . lift) get let@@ -270,7 +268,7 @@ Handle all cluster and partition state propagation actions, and return an updated node state. -}-propagate :: SM i o s [ClusterAction i s]+propagate :: SM i o s [ClusterAction i o s] propagate = SM $ do partitionActions <- getPartitionActions clusterActions <- unSM getClusterActions@@ -298,7 +296,7 @@ } return actions - getClusterActions :: SM i o s [ClusterAction i s]+ getClusterActions :: SM i o s [ClusterAction i o s] getClusterActions = SM $ do ns@NodeState {cluster} <- lift get let@@ -413,9 +411,9 @@ with other nodes. It is up to the runtime system to implement the actions. -}-data ClusterAction i s+data ClusterAction i o s = ClusterMerge Peer ClusterPowerState- | PartitionMerge Peer PartitionKey (PartitionPowerState i s)+ | PartitionMerge Peer PartitionKey (PartitionPowerState i o s) {- |@@ -433,11 +431,11 @@ {- | Gets a partition state. -}-getPartition :: (Default s, ApplyDelta i s)+getPartition :: (Default s, ApplyDelta i o s) => PartitionKey- -> SM i o s (PartitionPropState i s)+ -> SM i o s (PartitionPropState i o s) getPartition key = SM $ do- Legionary {persistence} <- ask+ persistence <- ask NodeState {self, partitions, cluster} <- lift get case Map.lookup key partitions of Nothing ->@@ -451,15 +449,15 @@ 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)+savePartition :: (Default s, ApplyDelta i o s, Indexable s) => PartitionKey- -> PartitionPropState i s+ -> PartitionPropState i o s -> SM i o s () savePartition key partition = SM $ do- Legionary {persistence, index} <- ask- oldTags <- index . P.ask <$> unSM (getPartition key)+ persistence <- ask+ oldTags <- indexEntries . P.ask <$> unSM (getPartition key) let- currentTags = index (P.ask partition)+ currentTags = indexEntries (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