legion 0.8.0.3 → 0.9.0.0
raw patch · 11 files changed
+327/−382 lines, 11 filesdep +timedep −attoparsecdep −stm
Dependencies added: time
Dependencies removed: attoparsec, stm
Files
- legion.cabal +2/−4
- src/Network/Legion.hs +168/−111
- src/Network/Legion/Admin.hs +7/−15
- src/Network/Legion/Application.hs +33/−15
- src/Network/Legion/Basics.hs +0/−124
- src/Network/Legion/Distribution.hs +3/−1
- src/Network/Legion/PartitionKey.hs +0/−62
- src/Network/Legion/PowerState.hs +3/−3
- src/Network/Legion/Runtime.hs +103/−39
- src/Network/Legion/Runtime/ConnectionManager.hs +2/−2
- src/Network/Legion/StateMachine.hs +6/−6
legion.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: legion-version: 0.8.0.3+version: 0.9.0.0 synopsis: Distributed, stateful, homogeneous microservice framework. description: Legion is a framework for writing distributed, homogeneous, stateful microservices in Haskell.@@ -30,7 +30,6 @@ Network.Legion.Admin Network.Legion.Application Network.Legion.BSockAddr- Network.Legion.Basics Network.Legion.ClusterState Network.Legion.Conduit Network.Legion.Distribution@@ -55,7 +54,6 @@ build-depends: Ranged-sets >= 0.3.0 && < 0.4, aeson >= 0.11.2.0 && < 0.12,- attoparsec >= 0.13.0.1 && < 0.14, base >= 4.8 && < 4.10, binary >= 0.7.5 && < 0.9, binary-conduit >= 1.2.3 && < 1.3,@@ -73,8 +71,8 @@ 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, unix >= 2.7 && < 2.8, uuid >= 1.3.11 && < 1.4,
src/Network/Legion.hs view
@@ -1,12 +1,13 @@ {- | Legion is a mathematically sound framework for writing horizontally- scalable user applications. Historically, horizontal scalability has- been achieved via the property of statelessness. Programmers would- design their applications to be free of any kind of persistent state,- avoiding the problem of distributed state management. This almost never- turns out to really be possible, so programmers achieve "statelessness"- by delegating application state management to some kind of external,- shared database -- which ends up having its own scalability problems.+ scalable business logic, or applications. Historically,+ horizontal scalability has been achieved via the property of+ statelessness. Programmers would design their applications to be free+ of any kind of persistent state, avoiding the problem of distributed+ state management. This almost never turns out to really be possible,+ so programmers achieve "statelessness" by delegating application state+ management to some kind of external, shared database (which ends up+ having its own scalability problems). In addition to scalability problems, which modern databases (especially NoSQL databases) have done a good job of solving, there is another,@@ -15,172 +16,228 @@ Legion is a Haskell framework that abstracts state partitioning, data replication, request routing, and cluster rebalancing, making it easy- to implement large and robust distributed data applications.-- Examples of services that rely on partitioning include ElasticSearch,- Riak, DynamoDB, and others. In other words, almost all scalable- databases.+ to implement large and robust distributed stateful applications. -} module Network.Legion (- -- * Using Legion+ -- * API Reference -- ** Starting the Legion Runtime- -- $startup forkLegionary, StartupMode(..), Runtime, - -- ** Runtime Configuration- -- $framework-config- RuntimeSettings(..),- -- ** Making Runtime Requests makeRequest, search, - -- * Implementing a Legion Application- -- $service-implementaiton-- -- ** Indexing- -- $indexing- Indexable(..),+ -- ** Application Definition LegionConstraints, Persistence(..), Event(..),- Tag(..), - -- * Other Types+ -- ** Runtime Configuration+ -- $framework-config+ RuntimeSettings(..),++ -- ** Indexing+ Indexable(..),+ Tag(..), SearchTag(..), IndexRecord(..),++ -- ** Other Types+ Peer, PartitionKey(..), PartitionPowerState,+ ClusterPowerState, - -- * Utils- newMemoryPersistence,- diskPersistence,+ -- * Implementing a Legion Application+ -- $service-implementaiton++ -- ** Typclasses to Implement+ -- $constraints++ -- *** Event+ -- $event++ -- *** Indexable+ -- $indexable++ -- ** Exposing Your Application+ -- $expose++ -- ** Partitions, Explained+ -- $partitions++ -- ** The Persistence Layer+ -- $persistence+ ) where import Prelude hiding (lookup, readFile, writeFile, null) import Network.Legion.Application (LegionConstraints,- Persistence(Persistence, getState, saveState, list))-import Network.Legion.Basics (newMemoryPersistence, diskPersistence)+ Persistence(Persistence, getState, saveState, list, saveCluster))+import Network.Legion.ClusterState (ClusterPowerState)+import Network.Legion.Distribution (Peer) import Network.Legion.Index (Tag(Tag, unTag), IndexRecord(IndexRecord, irTag, irKey), SearchTag(SearchTag, stTag, stKey), Indexable(indexEntries)) 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),+import Network.Legion.Runtime (StartupMode(NewCluster, JoinCluster, Recover), forkLegionary, Runtime, makeRequest, search) import Network.Legion.Settings (RuntimeSettings(RuntimeSettings, adminHost, adminPort, peerBindAddr, joinBindAddr)) -------------------------------------------------------------------------------- +-- $framework-config+-- The legion framework has several operational parameters which can be+-- controlled using configuration. These include the address binding used+-- to expose the cluster management service endpoint and what file to use+-- for cluster state journaling. To get started quickly, consider using+-- @[legion-extra:Network.Legion.Config](https://hackage.haskell.org/package/legion-extra/docs/Network-Legion-Config.html)@++--------------------------------------------------------------------------------+ -- $service-implementaiton--- Whenever you use Legion to develop a distributed application, your--- application is going to be divided into two major parts, the state/less/--- part, and the state/ful/ part. The stateless part is going to be the--- context in which a legion node is running -- probably a web server if you--- are exposing your application as a web service. Legion itself is focused--- 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.+-- Implementing a Legion application boils down to+-- two things: providing a persistence layer (see the+-- [legion-extra](https://hackage.haskell.org/package/legion-extra)+-- package for some pre-packaged persistence layers), and implementing+-- all of the typeclasses in 'LegionConstraints', of which 'Event'+-- is the most important. ----- 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--- stateful user requests which are routed to it based on the `PartitionKey`--- associated with the request. What the stateful part of Legion is--- /not/ able to do is figure out what partition key is associated with--- 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+-- Take a look at+-- @[Network.Legion.Discovery.LegionApp](https://github.com/owensmurray/legion-discovery/blob/master/src/Network/Legion/Discovery/LegionApp.hs)@+-- for a good example of how to build a core Legion application.++--------------------------------------------------------------------------------++-- $constraints+-- Lets take a look at 'LegionConstraints': ----- * 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.+-- > (+-- > Event e o s,+-- > Indexable s,+-- > +-- > Binary e, Binary o, Binary s, Default s, Eq e, Show e, Show o, Show s+-- > ) ----- 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.+-- First, a note on the type variables uesd: ----- 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 `Event`.--- +-- - @__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"__.+--+-- The two most important typeclasses here are 'Event' and+-- 'Indexable'. The rest are mainly used for implementation details, like+-- packaging up data to send over the network, and constructing log messages,+-- etc.+ +--------------------------------------------------------------------------------++-- $event+-- Your 'Event' instance will serve as the core of your application. If you+-- think of your application as a large state machine, with inputs, state+-- changes and outputs, then the 'Event' typeclass acts as the main state+-- transition function.+-- -- > class Event e o s | e -> s o where -- > apply :: e -> s -> (o, s)--- --- If you look at 'apply', you will see that it is abstract over the type--- variables @e@, @o@, and @s@. These are the types your application--- has to fill in. @e@ stands for "event", 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'--- e o s@, which is short-hand for a long list of typeclasses that your--- @e@, @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--- persistence strategies, including only in memory, on disk, or in some--- external database.+-- The 'apply' function acts like a state transition function. It+-- handles application inputs (which are the events themselves), state+-- transitions, and outputs. In other words, your 'Event' instance __is__+-- your legion application. ----- See 'newMemoryPersistence' and 'diskPersistence' if you need to get--- started quickly with an in-memory persistence layer.+-- You will notice that the 'apply' function is totally pure. The+-- idea is that Legion will handle all of the necessary IO. It will+-- push events around on the network. It will retrieve state from the+-- persistence layer and automatically invoke your 'apply' function+-- where appropriate. This purity is necessary because it is the nature+-- of distributed, replicated systems that the order of events may+-- sometimes need to be rearranged, and the events themselves will have+-- to be replicated, and therefore applied more than once (at least one+-- time for each replica). -------------------------------------------------------------------------------- --- $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.------ Indexing is implemented by instantiating the 'Indexable' typeclass--- for your state type.+-- $indexable+-- The next important typeclass is 'Indexable'. -- -- > class Indexable s where -- > indexEntries :: s -> Set Tag ----- The tags returned by 'indexEntries' is used to construct a set of zero+-- For handling regular requests using 'makeRequest', you must know+-- the key of the partition you are looking for a priori. Sometimes,+-- though, you want to look up some unknown set of partitions based+-- on another attribute. The most basic example is when you want to do a+-- simple listing of all the partition keys in the system.+--+-- This is where the indexing system and the 'search' function come+-- in. The indexing system is exposed at a relatively low level of+-- abstraction because the use cases for which it is needed will vary+-- wildly from application to application. There is only a single global+-- index, but each partition may produce zero to many records in that+-- index. This is what the @Set Tag@ portion of the type signature above+-- is all about.+--+-- 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.+--+-- Indexing is implemented by instantiating the 'Indexable' typeclass+-- for your state type.+--+-- The tags returned by 'indexEntries' are 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>}- + -------------------------------------------------------------------------------- --- $startup--- While this section is being worked on, you can check out the--- [legion-discovery](https://github.com/owensmurray/legion-discovery)--- project for an example of a stateful web services that advantage of--- Legion's ability to define your own operations on your data. Take a look at--- [`Network.Legion.Discovery.App`](https://github.com/owensmurray/legion-discovery/blob/master/src/Network/Legion/Discovery/App.hs)--- to see where the magic of defining a Legion application happens. The rest--- of the code is mostly just standard HTTP-interface-written-in-Haskell,--- and requests sent to the Legion runtime.+-- $expose+-- The interface to your new application is a Haskell+-- api, which isn't very useful on its own. You are+-- usually going to want to provide a wrapper around your+-- core Legion app so that it is accessible to the outside world. The+-- [Legion-Discovery](https://github.com/owensmurray/legion-discovery/blob/master/src/Network/Legion/Discovery/Server.hs)+-- application uses [Servant](https://hackage.haskell.org/package/servant)+-- to expose its core Legion application via a web interface. -------------------------------------------------------------------------------- --- $framework-config--- The legion framework has several operational parameters which can--- be controlled using configuration. These include the address binding--- used to expose the cluster management service endpoint and what file--- to use for cluster state journaling.+-- $partitions+-- Coming soon.++-- 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+-- stateful user requests which are routed to it based on the `PartitionKey`+-- associated with the request. What the stateful part of Legion is+-- /not/ able to do is figure out what partition key is associated with+-- the request in the first place.++--------------------------------------------------------------------------------++-- $persistence+-- Coming soon.++-- 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.
src/Network/Legion/Admin.hs view
@@ -22,6 +22,7 @@ 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)@@ -71,11 +72,10 @@ get $ json =<< send chan GetIndex resource "/divergent" $ get $- json . Map.mapKeys (show . toInteger . unKey) =<< send chan GetDivergent+ 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"@@ -101,9 +101,7 @@ takeMVar mvar -{- |- Build some warp settings based on the configured socket address.--}+{- | Build some warp settings based on the configured socket address. -} options :: Port -> HostPreference -> Options options port host = def { settings =@@ -116,21 +114,15 @@ setServer :: Middleware setServer = addServerHeader . stripServerHeader where- {- |- Strip the server header- -}+ {- | Strip the server header -} stripServerHeader :: Middleware stripServerHeader = modifyResponse (stripHeader "Server") - {- |- Add our own server header.- -}+ {- | Add our own server header. -} addServerHeader :: Middleware addServerHeader = addHeaders [("Server", serverValue)] - {- |- The value of the @Server:@ header.- -}+ {- | The value of the @Server:@ header. -} serverValue = encodeUtf8 (T.pack ("legion-admin/" ++ showVersion version)) @@ -143,7 +135,7 @@ | GetPart PartitionKey (PartitionPowerState e o s -> LIO ()) | Eject Peer (() -> LIO ()) | GetIndex (Set IndexRecord -> LIO ())- | GetDivergent (Map PartitionKey (PartitionPowerState e o s) -> LIO ())+ | GetDivergent (Map Peer (Maybe UTCTime) -> LIO ()) | GetStates (Map PartitionKey (PartitionPowerState e o s) -> LIO ()) instance Show (AdminMessage e o s) where
src/Network/Legion/Application.hs view
@@ -8,10 +8,11 @@ Persistence(..), ) where -import Data.Aeson (ToJSON) import Data.Binary (Binary) import Data.Conduit (Source) import Data.Default.Class (Default)+import Network.Legion.ClusterState (ClusterPowerState)+import Network.Legion.Distribution (Peer) import Network.Legion.Index (Indexable) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.PartitionState (PartitionPowerState)@@ -22,16 +23,32 @@ constraints > (- > Binary e, Binary o, Binary s, Default s, Eq e, Event e o s, Indexable s,- > Show e, Show o, Show s, ToJSON s+ > Event e o s,+ > Indexable s,+ > Binary e,+ > Binary o,+ > Binary s,+ > Default s,+ > Eq e,+ > Show e,+ > Show o,+ > Show s > ) The @ToJSON s@ requirement is strictly for servicing the admin web endpoints. -} type LegionConstraints e o s = (- Binary e, Binary o, Binary s, Default s, Eq e, Event e o s, Indexable s,- Show e, Show o, Show s, ToJSON s+ Event e o s,+ Indexable s,+ Binary e,+ Binary o,+ Binary s,+ Default s,+ Eq e,+ Show e,+ Show o,+ Show s ) @@ -41,16 +58,17 @@ 'Network.Legion.diskPersistence' if you need to get started quickly. -} data Persistence e o s = Persistence {- getState :: PartitionKey -> IO (Maybe (PartitionPowerState e o s)),- saveState :: PartitionKey -> Maybe (PartitionPowerState e o s) -> IO (),- list :: Source IO (PartitionKey, PartitionPowerState e o s)- {- ^- List all the keys known to the persistence layer. It is- important that the implementation do the right thing- with regard to `Data.Conduit.addCleanup`, because- there are cases where the conduit is terminated- without reading the entire list.- -}+ saveCluster :: Peer -> ClusterPowerState -> IO (),+ getState :: PartitionKey -> IO (Maybe (PartitionPowerState e o s)),+ saveState :: PartitionKey -> Maybe (PartitionPowerState e o s) -> IO (),+ list :: Source IO (PartitionKey, PartitionPowerState e o s)+ {- ^+ List all the keys known to the persistence layer. It is+ important that the implementation do the right thing+ with regard to `Data.Conduit.addCleanup`, because+ there are cases where the conduit is terminated+ without reading the entire list.+ -} }
− src/Network/Legion/Basics.hs
@@ -1,124 +0,0 @@-{-# LANGUAGE NamedFieldPuns #-}-{- |- This module contains some basis persistence strategies useful for- testing, or getting started.--}-module Network.Legion.Basics (- newMemoryPersistence,- diskPersistence,-) where--import Prelude hiding (lookup, readFile, writeFile)--import Control.Concurrent.STM (atomically, newTVar, modifyTVar', readTVar,- TVar)-import Control.Monad.Trans.Class (lift)-import Data.Binary (Binary, encode, decode)-import Data.Bool (bool)-import Data.ByteString (readFile, writeFile)-import Data.ByteString.Lazy (toStrict, fromStrict)-import Data.Conduit (Source, (=$=), awaitForever, yield)-import Data.Conduit.List (sourceList)-import Data.Either (rights)-import Data.Map (Map, insert, lookup)-import Network.Legion.Application (Persistence(Persistence, getState,- saveState, list))-import Network.Legion.PartitionKey (PartitionKey, toHex, fromHex)-import Network.Legion.PartitionState(PartitionPowerState)-import System.Directory (removeFile, doesFileExist, getDirectoryContents)-import qualified Data.Map as Map---{- |- A convenient memory-based persistence layer. Good for testing or for- applications (like caches) that don't have durability requirements.--}-newMemoryPersistence :: IO (Persistence e o s)--newMemoryPersistence = do- cacheT <- atomically (newTVar Map.empty)- return Persistence {- getState = fetchState cacheT,- saveState = saveState_ cacheT,- list = list_ cacheT- }- where- saveState_- :: TVar (Map PartitionKey (PartitionPowerState e o s))- -> PartitionKey- -> Maybe (PartitionPowerState e o s)- -> IO ()- saveState_ cacheT key (Just state) =- (atomically . modifyTVar' cacheT . insert key) state-- saveState_ cacheT key Nothing =- (atomically . modifyTVar' cacheT . Map.delete) key-- fetchState- :: TVar (Map PartitionKey (PartitionPowerState e o s))- -> PartitionKey- -> IO (Maybe (PartitionPowerState e o s))- fetchState cacheT key = atomically $- lookup key <$> readTVar cacheT-- list_- :: TVar (Map PartitionKey (PartitionPowerState e o s))- -> Source IO (PartitionKey, PartitionPowerState e o s)- list_ cacheT =- sourceList . Map.toList =<< lift (atomically (readTVar cacheT))---{- | A convenient way to persist partition states to disk. -}-diskPersistence :: (Binary e, Binary s)- => FilePath- -- ^ The directory under which partition states will be stored.- -> Persistence e o s--diskPersistence directory = Persistence {- getState,- saveState,- list- }- where- getState :: (Binary e, Binary s)- => PartitionKey- -> IO (Maybe (PartitionPowerState e o s))- getState key =- let path = toPath key in- doesFileExist path >>= bool- (return Nothing)- ((Just . decode . fromStrict) <$> readFile path)-- saveState :: (Binary e, Binary s)- => PartitionKey- -> Maybe (PartitionPowerState e o s)- -> IO ()- saveState key (Just state) =- writeFile (toPath key) (toStrict (encode state))- saveState key Nothing =- let path = toPath key in- doesFileExist path >>= bool- (return ())- (removeFile path)-- list :: (Binary e, Binary s)- => Source IO (PartitionKey, PartitionPowerState e o s)- list = do- keys <- lift $ readHexList <$> getDirectoryContents directory- sourceList keys =$= fillData- where- fillData = awaitForever (\key -> do- let path = toPath key- state <- lift ((decode . fromStrict) <$> readFile path)- yield (key, state)- )- readHexList = rights . fmap fromHex . filter notSys- notSys = not . (`elem` [".", ".."])-- {- |- Convert a key to a path- -}- toPath :: PartitionKey -> FilePath- toPath key = directory ++ "/" ++ toHex key--
src/Network/Legion/Distribution.hs view
@@ -41,9 +41,11 @@ {- | The way to identify a peer. -}-newtype Peer = Peer UUID deriving (Show, Binary, Eq, Ord)+newtype Peer = Peer UUID deriving (Binary, Eq, Ord) instance Read Peer where readPrec = Peer <$> readPrec+instance Show Peer where+ showsPrec n (Peer uuid) = showsPrec n uuid {- |
src/Network/Legion/PartitionKey.hs view
@@ -4,20 +4,12 @@ -} module Network.Legion.PartitionKey ( PartitionKey(..),- toHex,- fromHex ) where -import Data.Attoparsec.ByteString (parseOnly, atEnd)-import Data.Attoparsec.ByteString.Char8 (hexadecimal) import Data.Binary (Binary(put, get))-import Data.Bits (testBit)-import Data.Bool (bool)-import Data.ByteString.Char8 (pack) import Data.DoubleWord (Word256(Word256), Word128(Word128)) import Data.Ranged (DiscreteOrdered(adjacent, adjacentBelow))-import Data.Word (Word64) {- | This is how partitions are identified and referenced. -}@@ -35,59 +27,5 @@ instance DiscreteOrdered PartitionKey where adjacent (K a) (K b) = a < b && succ a == b adjacentBelow (K k) = if k == minBound then Nothing else Just (K (pred k))---{- | Convert a `PartitionKey` into a hex string. -}-toHex :: PartitionKey -> String-toHex (K (Word256 (Word128 a b) (Word128 c d))) =- concatMap toHex64 [a, b, c, d]---{- |- Convert a `Word64` into a hex string.-- I know I'm going to hell for this, but I just can't abide the- @hexstring@ package pulling @aeson@ into our dependency tree.--}-toHex64 :: Word64 -> String-toHex64 w = fmap (digit . quad) [15, 14..0]- where- quad :: Int -> (Int, Int, Int, Int)- quad n = let base = n * 4 in (base + 3, base + 2, base + 1, base)-- digit :: (Int, Int, Int, Int) -> Char- digit (a, b, c, d) =- case (testBit w a, testBit w b, testBit w c, testBit w d) of- (False, False, False, False) -> '0'- (False, False, False, True) -> '1'- (False, False, True, False) -> '2'- (False, False, True, True) -> '3'- (False, True, False, False) -> '4'- (False, True, False, True) -> '5'- (False, True, True, False) -> '6'- (False, True, True, True) -> '7'- (True, False, False, False) -> '8'- (True, False, False, True) -> '9'- (True, False, True, False) -> 'a'- (True, False, True, True) -> 'b'- (True, True, False, False) -> 'c'- (True, True, False, True) -> 'd'- (True, True, True, False) -> 'e'- (True, True, True, True) -> 'f'---{- | Maybe convert a hex string into a partition key -}-fromHex :: String -> Either String PartitionKey-fromHex str- | length str > 64 =- Left "trailing characters while parsing hex PartitionKey"- | otherwise =- K <$> parseOnly parser (pack str)- where- parser = do- w <- hexadecimal- atEnd >>= bool- (fail "not a valid hex string")- (return w)
src/Network/Legion/PowerState.hs view
@@ -63,7 +63,7 @@ events :: Map (StateId p) (Delta p e, Set p) } deriving (Generic, Show, Eq) instance (Binary o, Binary s, Binary p, Binary e) => Binary (PowerState o s p e r)-instance (Show o, ToJSON s, Show p, Show e) => ToJSON (PowerState o s p e r) where+instance (Show o, Show s, Show p, Show e) => ToJSON (PowerState o s p e r) where toJSON PowerState {origin, infimum, events} = object [ "origin" .= show origin, "infimum" .= infimum,@@ -88,11 +88,11 @@ Infimum s1 _ _ == Infimum s2 _ _ = s1 == s2 instance (Ord p) => Ord (Infimum s p) where compare (Infimum s1 _ _) (Infimum s2 _ _) = compare s1 s2-instance (ToJSON s, Show p) => ToJSON (Infimum s p) where+instance (Show s, Show p) => ToJSON (Infimum s p) where toJSON Infimum {stateId, participants, stateValue} = object [ "stateId" .= show stateId, "participants" .= Set.map show participants,- "stateValue" .= stateValue+ "stateValue" .= show stateValue ]
src/Network/Legion/Runtime.hs view
@@ -34,11 +34,14 @@ 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)+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)@@ -70,10 +73,12 @@ 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@@ -83,11 +88,6 @@ 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).-- For the vast majority of service implementations, you are going to need- to implement some halfway complex concurrency in order to populate the- request source, and to handle the responses. Unless you know exactly- what you are doing, you probably want to use `forkLegionary` instead. -} runLegionary :: (LegionConstraints e o s) => Persistence e o s@@ -109,12 +109,12 @@ startupMode requestSource = do- {- Start the various messages sources. -}+ {- Start the various messages sources. -} peerS <- loggingC =<< startPeerListener settings adminS <- loggingC =<< runAdmin adminPort adminHost joinS <- loggingC (joinMsgSource settings) - (self, nodeState, peers) <- makeNodeState settings startupMode+ (self, nodeState, peers) <- makeNodeState persistence settings startupMode rts <- newRuntimeState self peers let messageSource = transPipe lift (@@ -139,6 +139,7 @@ nextId = firstMessageId, cm, self,+ commClock = Map.empty, searches = Map.empty } @@ -171,6 +172,10 @@ messageSink = awaitForever (\msg -> do $(logDebug) . pack $ "Receieved: " ++ show msg lift $ do+ case msg of+ P (PeerMessage source _ _) ->+ updateRecvClock source+ _ -> return () handleMessage msg updatePeers clusterActions@@ -198,8 +203,8 @@ 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.@@ -409,8 +414,18 @@ handleMessage {- Admin Get Divergent -} (A (GetDivergent respond))- =- lift2 . respond =<< SMM.partitions <$> SMM.getNodeState+ = 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))@@ -425,15 +440,13 @@ {- | This defines the various ways a node can be spun up. -} data StartupMode = NewCluster- {- ^- Indicates that we should bootstrap a new cluster at startup. The- persistence layer may be safely pre-populated because the new node- will claim the entire keyspace.- -}+ {- ^ Indicates that we should bootstrap a new cluster at startup. -} | JoinCluster SockAddr+ {- ^ Indicates that the node should try to join an existing cluster. -}+ | Recover Peer ClusterPowerState {- ^- Indicates that the node should try to join an existing cluster,- either by starting fresh, or by recovering from a shutdown or crash.+ Recover from a crash as the given peer, using the given cluster+ state. -} deriving (Show, Eq) @@ -508,29 +521,37 @@ {- | Figure out how to construct the initial node state. -} makeNodeState- :: RuntimeSettings+ :: Persistence e o s+ -> RuntimeSettings -> StartupMode -> LIO (Peer, NodeState e o s, Map Peer BSockAddr) -makeNodeState RuntimeSettings {peerBindAddr} NewCluster = do- {- Build a brand new node state, for the first node in a cluster. -}- self <- newPeer- clusterId <- getUUID- let- cluster = C.new clusterId self peerBindAddr- nodeState = newNodeState self cluster- return (self, nodeState, C.getPeers cluster)+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 RuntimeSettings {peerBindAddr} (JoinCluster addr) = do+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))- let- nodeState = newNodeState self cluster- return (self, nodeState, C.getPeers cluster)+ makeNodeState persistence settings (Recover self cluster) where joinCluster :: JoinRequest -> LIO (Peer, ClusterPowerState) joinCluster joinMsg = liftIO $ do@@ -550,11 +571,37 @@ ++ "to our join request!" Just (JoinOk self cps) -> return (self, cps)- Just (JoinRejected reason) -> fail- $ "The cluster would not allow us to re-join. "- ++ "The reason given was: " ++ show reason +makeNodeState persistence _ (Recover self cluster) = do+ let+ nodeState = newNodeState self cluster+ liftIO $ saveCluster persistence self cluster+ return (self, nodeState, C.getPeers cluster) ++{- |+ 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@@ -674,7 +721,8 @@ Legion application. This allows you to make requests and access the partition index. - 'Runtime' is an opaque structure. Use 'makeRequest' to access it.+ 'Runtime' is an opaque structure. Use 'makeRequest' and 'search' to+ access it. -} data Runtime e o = Runtime { {- |@@ -746,14 +794,16 @@ 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 ()])+ SearchTag+ (Set Peer, Maybe IndexRecord, [Maybe IndexRecord -> LIO ()]) } {- | This is the type of a join request message. -}-data JoinRequest = JoinRequest BSockAddr+newtype JoinRequest = JoinRequest BSockAddr deriving (Generic, Show) instance Binary JoinRequest @@ -761,7 +811,6 @@ {- | The response to a JoinRequst message -} data JoinResponse = JoinOk Peer ClusterPowerState- | JoinRejected String deriving (Generic) instance Binary JoinResponse @@ -810,5 +859,20 @@ 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+ }+ )
src/Network/Legion/Runtime/ConnectionManager.hs view
@@ -36,7 +36,7 @@ {- | A handle on the connection manager -}-data ConnectionManager e o s = C (Chan (Message e o s))+newtype ConnectionManager e o s = C (Chan (Message e o s)) instance Show (ConnectionManager e o s) where show _ = "ConnectionManager" @@ -188,7 +188,7 @@ {- | The internal state of the connection manager. -}-data State e o s = S {+newtype State e o s = S { connections :: Map Peer (Chan (PeerMessage e o s)) }
src/Network/Legion/StateMachine.hs view
@@ -60,8 +60,7 @@ import Control.Monad (void, unless) import Control.Monad.Catch (throwM, MonadThrow) import Control.Monad.IO.Class (MonadIO, liftIO)-import Control.Monad.Logger (MonadLogger, logDebug, logError,- MonadLoggerIO, logWarn)+import Control.Monad.Logger (logDebug, logError, MonadLoggerIO, logWarn) import Control.Monad.Trans.Class (lift) import Data.Bool (bool) import Data.Conduit ((=$=), runConduit, transPipe, awaitForever)@@ -70,7 +69,7 @@ import Data.Maybe (fromMaybe) import Data.Set (Set, (\\), member) import Data.Text (pack)-import Network.Legion.Application (getState, saveState, list)+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,@@ -225,7 +224,7 @@ {- | Eject a peer from the cluster. -}-eject :: (MonadLogger m, MonadThrow m) => Peer -> SM e o s m ()+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@@ -560,7 +559,7 @@ {- | Like 'runClusterPowerStateTAs', but run as the local peer. -}-runClusterPowerStateT :: (MonadThrow m)+runClusterPowerStateT :: (MonadThrow m, MonadIO m) => ClusterPowerStateT (SM e o s m) a -> SM e o s m a runClusterPowerStateT m = do@@ -575,7 +574,7 @@ Generalized to run as any peer, in order to support exceptional cases like 'eject'. -}-runClusterPowerStateTAs :: (MonadThrow m)+runClusterPowerStateTAs :: (MonadThrow m, MonadIO m) => Peer {- ^ The peer to run as. -} -> ClusterPowerStateT (SM e o s m) a -> SM e o s m a@@ -584,6 +583,7 @@ 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