diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,202 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/legion.cabal b/legion.cabal new file mode 100644 --- /dev/null +++ b/legion.cabal @@ -0,0 +1,76 @@ +-- Initial keyspace-partition.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/ + +name: legion +version: 0.1.0.0 +synopsis: Distributed, stateful, homogeneous microservice framework. +description: Legion is a framework for writing distributed, + homogeneous, stateful microservices in Haskell. +homepage: https://github.com/taphu/legion +license: Apache-2.0 +license-file: LICENSE +author: Rick Owens +maintainer: rick@owenssoftware.com +copyright: 2015-2016 Rick Owens +category: Concurrency, Network +build-type: Simple +-- extra-source-files: +cabal-version: >=1.10 + +source-repository head + type: git + location: git@github.com:taphu/legion.git + +library + exposed-modules: + Network.Legion + other-modules: + Network.Legion.Admin + Network.Legion.Application + Network.Legion.BSockAddr + Network.Legion.Basics + Network.Legion.ClusterState + Network.Legion.Conduit + Network.Legion.ConnectionManager + Network.Legion.Distribution + Network.Legion.Fork + Network.Legion.KeySet + Network.Legion.LIO + Network.Legion.PartitionKey + Network.Legion.PartitionState + Network.Legion.PowerState + Network.Legion.Propagation + Network.Legion.Runtime + Network.Legion.Settings + Network.Legion.StateMachine + Network.Legion.UUID + -- other-extensions: + build-depends: + Ranged-sets >= 0.3.0 && < 0.4, + attoparsec >= 0.13.0.1 && < 0.14, + base >= 4.8 && < 4.9, + binary >= 0.7.5 && < 0.9, + binary-conduit >= 1.2.3 && < 1.3, + bytestring >= 0.10.4.0 && < 0.11, + conduit >= 1.2.4 && < 1.3, + conduit-extra >= 1.1.9 && < 1.2, + containers >= 0.5.5.1 && < 0.6, + data-default-class >= 0.0.1 && < 0.2, + data-dword >= 0.3 && < 0.4, + directory >= 1.2.1.0 && < 1.3, + exceptions >= 0.8 && < 0.9, + monad-logger >= 0.3.17 && < 0.4, + network >= 2.6.2.1 && < 2.7, + scotty >= 0.11.0 && < 0.12, + scotty-resource >= 0.1 && < 0.2, + stm >= 2.4.4.1 && < 2.5, + text >= 1.2.2.0 && < 1.3, + time >= 1.4.2 && < 1.6, + transformers >= 0.3.0.0 && < 0.5, + unix >= 2.7 && < 2.8, + uuid >= 1.3.11 && < 1.4, + warp >= 3.2 && < 3.3 + hs-source-dirs: src + default-language: Haskell2010 + ghc-options: -Wall -W -fwarn-incomplete-uni-patterns + diff --git a/src/Network/Legion.hs b/src/Network/Legion.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion.hs @@ -0,0 +1,173 @@ +{- | + 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. + + In addition to scalability problems, which modern databases (especially + NoSQL databases) have done a good job of solving, there is another, + more fundamental problem facing these architectures: The application + is not really stateless. + + 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. +-} + +module Network.Legion ( + -- * Service Implementation + -- $service-implementaiton + Legionary(..), + LegionConstraints, + Persistence(..), + ApplyDelta(..), + RequestMsg, + -- * Invoking Legion + -- $invocation + forkLegionary, + runLegionary, + StartupMode(..), + -- * Fundamental Types + PartitionKey(..), + PartitionPowerState, + projected, + infimum, + -- * Framework Configuration + -- $framework-config + LegionarySettings(..), + -- * Utils + newMemoryPersistence, + diskPersistence, +) where + +import Prelude hiding (lookup, readFile, writeFile, null) + +import Network.Legion.Application (LegionConstraints, + Persistence(Persistence, getState, saveState, list), + Legionary(Legionary, persistence, handleRequest), RequestMsg) +import Network.Legion.Basics (newMemoryPersistence, diskPersistence) +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.Settings (LegionarySettings(LegionarySettings, + adminHost, adminPort, peerBindAddr, joinBindAddr)) + +-------------------------------------------------------------------------------- + +-- $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. +-- +-- 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 +-- +-- * 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 +-- 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. +-- +-- 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. ;-) +-- +-- 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. + +-------------------------------------------------------------------------------- + +-- $invocation +-- Notes on invocation. + +-------------------------------------------------------------------------------- + +-- $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. + diff --git a/src/Network/Legion/Admin.hs b/src/Network/Legion/Admin.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/Admin.hs @@ -0,0 +1,76 @@ +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{- | + This module contains the admin interface code. +-} +module Network.Legion.Admin ( + runAdmin, +) where + +import Control.Concurrent (forkIO, newChan, newEmptyMVar, writeChan, + putMVar, takeMVar, Chan) +import Control.Monad (void) +import Control.Monad.Logger (askLoggerIO, runLoggingT) +import Control.Monad.Trans.Class (lift) +import Data.Conduit (Source) +import Data.Default.Class (def) +import Data.Text.Lazy (Text, pack) +import Network.Legion.Application (LegionConstraints) +import Network.Legion.Conduit (chanToSource) +import Network.Legion.LIO (LIO) +import Network.Legion.PartitionKey (PartitionKey(K)) +import Network.Legion.StateMachine (AdminMessage(GetState, GetPart)) +import Network.Wai.Handler.Warp (HostPreference, defaultSettings, Port, + setHost, setPort) +import Web.Scotty.Resource.Trans (resource, get) +import Web.Scotty.Trans (Options, scottyOptsT, settings, ScottyT, text, + ActionT, param) + +{- | + Start the admin service in a background thread. +-} +runAdmin :: (LegionConstraints i o s) + => Port + -> HostPreference + -> LIO (Source LIO (AdminMessage i o s)) +runAdmin addr host = do + logging <- askLoggerIO + chan <- lift newChan + void . lift . forkIO . (`runLoggingT` logging) $ + let + website :: ScottyT Text LIO () + website = do + resource "/clusterstate" $ + get $ do + val <- send chan GetState + text (pack (show val)) + resource "/propstate/:key" $ + get $ do + key <- K . read <$> param "key" + val <- send chan (GetPart key) + text (pack (show val)) + in scottyOptsT (options addr host) (`runLoggingT` logging) website + return (chanToSource chan) + where + send + :: Chan (AdminMessage i o s) + -> ((a -> LIO ()) -> AdminMessage i o s) + -> ActionT Text LIO a + send chan msg = lift . lift $ do + mvar <- newEmptyMVar + writeChan chan (msg (lift . putMVar mvar)) + takeMVar mvar + + +{- | + Build some warp settings based on the configured socket address. +-} +options :: Port -> HostPreference -> Options +options port host = def { + settings = + setPort port + . setHost host + $ defaultSettings + } + + diff --git a/src/Network/Legion/Application.hs b/src/Network/Legion/Application.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/Application.hs @@ -0,0 +1,85 @@ +{-# LANGUAGE MultiParamTypeClasses #-} +{- | + This module contains the data types necessary for implementing the + user application. +-} +module Network.Legion.Application ( + LegionConstraints, + Legionary(..), + Persistence(..), + RequestMsg, +) where + +import Data.Binary (Binary) +import Data.Conduit (Source) +import Data.Default.Class (Default) +import Network.Legion.PartitionKey (PartitionKey) +import Network.Legion.PartitionState (PartitionPowerState) +import Network.Legion.PowerState (ApplyDelta) + +{- | + 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 + > ) +-} +class ( + ApplyDelta i s, Default s, Binary i, Binary o, Binary s, Show i, + Show o, Show s, Eq i + ) => LegionConstraints i o s where + + +{- | + This is the type of a user-defined Legion application. Implement this and + allow the Legion framework to manage your cluster. +-} +data Legionary i o s = Legionary { + {- | + The request handler, implemented by the user to service requests. + + Returns a response to the request, together with the new partition + state. + -} + handleRequest :: PartitionKey -> i -> s -> o, + {- | + The user-defined persistence layer implementation. + -} + persistence :: Persistence i s + } + + +{- | + 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) + {- ^ + 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. + -} + } + + +{- | + 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/BSockAddr.hs b/src/Network/Legion/BSockAddr.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/BSockAddr.hs @@ -0,0 +1,53 @@ +{- | + This module contains the BSockAddr data type. +-} +module Network.Legion.BSockAddr ( + BSockAddr(..) +) where + +import Data.Binary (Binary(put, get)) +import Data.Word (Word8) +import Network.Socket (SockAddr(SockAddrInet, SockAddrInet6, SockAddrUnix, + SockAddrCan)) + + +{- | + A type useful only for creating a `Binary` instance of `SockAddr`. +-} +newtype BSockAddr = BSockAddr {getAddr :: SockAddr} deriving (Show, Eq) + +instance Binary BSockAddr where + put (BSockAddr addr) = + case addr of + SockAddrInet p h -> do + put (0 :: Word8) + put (fromEnum p, h) + SockAddrInet6 p f h s -> do + put (1 :: Word8) + put (fromEnum p, f, h, s) + SockAddrUnix s -> do + put (2 :: Word8) + put s + SockAddrCan a -> do + put (3 :: Word8) + put a + + get = BSockAddr <$> do + c <- get + case (c :: Word8) of + 0 -> do + (p, h) <- get + return (SockAddrInet (toEnum p) h) + 1 -> do + (p, f, h, s) <- get + return (SockAddrInet6 (toEnum p) f h s) + 2 -> SockAddrUnix <$> get + 3 -> SockAddrCan <$> get + _ -> + fail + $ "Can't decode BSockAddr because the constructor tag " + ++ "was not understood. Probably this data is representing " + ++ "something else." + + + diff --git a/src/Network/Legion/Basics.hs b/src/Network/Legion/Basics.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/Basics.hs @@ -0,0 +1,124 @@ +{-# 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 i 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 i s)) + -> PartitionKey + -> Maybe (PartitionPowerState i 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 i s)) + -> PartitionKey + -> 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) + list_ cacheT = + sourceList . Map.toList =<< lift (atomically (readTVar cacheT)) + + +{- | A convenient way to persist partition states to disk. -} +diskPersistence :: (Binary i, Binary s) + => FilePath + -- ^ The directory under which partition states will be stored. + -> Persistence i s + +diskPersistence directory = Persistence { + getState, + saveState, + list + } + where + getState :: (Binary i, Binary s) + => PartitionKey + -> IO (Maybe (PartitionPowerState i s)) + getState key = + let path = toPath key in + doesFileExist path >>= bool + (return Nothing) + ((Just . decode . fromStrict) <$> readFile path) + + saveState :: (Binary i, Binary s) + => PartitionKey + -> Maybe (PartitionPowerState i 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 i, Binary s) + => Source IO (PartitionKey, PartitionPowerState i 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 + + diff --git a/src/Network/Legion/ClusterState.hs b/src/Network/Legion/ClusterState.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/ClusterState.hs @@ -0,0 +1,225 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE NamedFieldPuns #-} +{- | + This module contains the data types related to the distributed cluster state. +-} +module Network.Legion.ClusterState ( + ClusterState, + ClusterPowerState, + ClusterPropState, + claimParticipation, + new, + initProp, + getPowerState, + getPeers, + findPartition, + getDistribution, + joinCluster, + mergeEither, + actions, + allParticipants, + heartbeat, +) where + +import Data.Binary (Binary) +import Data.Default.Class (Default(def)) +import Data.Map (Map) +import Data.Set (Set) +import Data.Time.Clock (UTCTime) +import Data.UUID (UUID) +import GHC.Generics (Generic) +import Network.Legion.BSockAddr (BSockAddr(BSockAddr)) +import Network.Legion.Distribution (ParticipationDefaults, modify, Peer) +import Network.Legion.KeySet (KeySet, full, unions) +import Network.Legion.PartitionKey (PartitionKey) +import Network.Legion.PowerState (ApplyDelta(apply)) +import Network.Legion.Propagation (PropState, PropPowerState) +import Network.Socket (SockAddr) +import qualified Data.Map as Map +import qualified Data.Set as Set +import qualified Network.Legion.Distribution as D +import qualified Network.Legion.Propagation as P + + +{- | + An opaque data type, representing the cluster state that is shared + between all peers. +-} +data ClusterState = ClusterState { + distribution :: ParticipationDefaults, + peers :: Map Peer BSockAddr + } + deriving (Show, Generic) +instance Binary ClusterState +instance Default ClusterState where + def = ClusterState { + distribution = D.empty, + peers = Map.empty + } + + +{- | + A representation of all possible cluster states. +-} +newtype ClusterPowerState = ClusterPowerState { + unPowerState :: PropPowerState UUID ClusterState Peer Update + } deriving (Show, Binary) + + +{- | + A reification of `PropState`, representing the propagation state of the + cluster state. +-} +newtype ClusterPropState = ClusterPropState { + unPropState :: PropState UUID ClusterState Peer Update + } deriving (Show) + + +{- | + The kinds of updates that can be applied to the cluster state. +-} +data Update + = PeerJoined Peer BSockAddr + | Participating Peer KeySet + deriving (Show, Generic) +instance Binary Update +instance ApplyDelta Update ClusterState where + apply (PeerJoined peer addr) cs@ClusterState {peers} = + cs {peers = Map.insert peer addr peers} + apply (Participating peer ks) cs@ClusterState {distribution} = + cs {distribution = modify (Set.insert peer) ks distribution} + + +{- | + Helper function, for easily claiming participation in a key set. +-} +claimParticipation + :: Peer + -> KeySet + -> ClusterPropState + -> ClusterPropState +claimParticipation peer ks = + ClusterPropState + . P.delta (Participating peer ks) + . unPropState + + +{- | + Create the cluster state appropriate for a brand-new cluster. +-} +new :: UUID -> Peer -> SockAddr -> ClusterPropState +new clusterId self addy = + claimParticipation self full + . ClusterPropState + . P.delta (PeerJoined self (BSockAddr addy)) + $ P.new clusterId self (Set.singleton self) + + +{- | + Initialize a `ClusterPropState` based on the initial underlying cluster power + state. +-} +initProp :: Peer -> ClusterPowerState -> ClusterPropState +initProp self = ClusterPropState . P.initProp self . unPowerState + + +{- | + Return an opaque representation of the underling power state, for transfer + across the network, or whatever. +-} +getPowerState :: ClusterPropState -> ClusterPowerState +getPowerState = ClusterPowerState . P.getPowerState . unPropState + + +{- | + Get the cluster peers. +-} +getPeers :: ClusterPropState -> Map Peer BSockAddr +getPeers = peers . P.ask . unPropState + + +{- | + get the cluster distribution. +-} +getDistribution :: ClusterPropState -> ParticipationDefaults +getDistribution = distribution . P.ask . unPropState + + +{- | + Find the nodes that own a given partition. +-} +findPartition :: PartitionKey -> ClusterPropState -> Set Peer +findPartition key = + D.findPartition key . distribution . P.ask . unPropState + + +{- | + Allow a new peer to join the cluster. +-} +joinCluster + :: Peer + {- ^ The peer that is joining. -} + -> BSockAddr + {- ^ The cluster address of the new peer. -} + -> ClusterPropState + {- ^ The current cluster propagation state. -} + -> ClusterPropState +joinCluster peer addy = + ClusterPropState + . P.delta (PeerJoined peer addy) + . P.participate peer + . unPropState + + +{- | + Merge a foreign cluster state with our own cluster state. This function + returns the new cluster propagation state, along with a set of partition keys + for which the default participation has changed (aka, a rebalance happened), + indicating that some action should be taken to migrate the indicated + partitions. +-} +mergeEither + :: Peer + -> ClusterPowerState + -> ClusterPropState + -> Either String (ClusterPropState, KeySet) +mergeEither otherPeer (ClusterPowerState otherPS) (ClusterPropState prop) = + let + self = P.getSelf prop + divergences = P.divergences self (P.initProp otherPeer otherPS) + migrating = unions [ + ks + | (_, Participating _ ks) <- Map.toList divergences + ] + in case P.mergeEither otherPeer otherPS prop of + Left err -> Left err + Right newProp -> Right (ClusterPropState newProp, migrating) + + +{- | + Get the peers which require action (i.e. Send), if any, and the + powerstate version to send to those peers, and the new propagation + state that is applicable after those actions have been taken. +-} +actions :: ClusterPropState -> (Set Peer, ClusterPowerState, ClusterPropState) +actions prop = + let (peers, ps, newProp) = P.actions (unPropState prop) + in (peers, ClusterPowerState ps, ClusterPropState newProp) + + +{- | + Return all cluster participants. +-} +allParticipants :: ClusterPropState -> Set Peer +allParticipants = P.allParticipants . unPropState + + +{- | + Move time forward for the propagation state. +-} +heartbeat :: UTCTime -> ClusterPropState -> ClusterPropState +heartbeat now = ClusterPropState . P.heartbeat now . unPropState + + diff --git a/src/Network/Legion/Conduit.hs b/src/Network/Legion/Conduit.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/Conduit.hs @@ -0,0 +1,53 @@ +{- | + This module contains some handy conduit abstractions. +-} +module Network.Legion.Conduit ( + chanToSource, + chanToSink, + merge, +) where + +import Control.Concurrent (forkIO) +import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan) +import Control.Monad (void, forever) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Conduit (Source, Sink, ($$), await, ($=), yield, await) +import qualified Data.Conduit.List as CL (map) + +{- | + Convert a chanel into a Source. +-} +chanToSource :: (MonadIO io) => Chan a -> Source io a +chanToSource chan = forever $ yield =<< liftIO (readChan chan) + + +{- | + Convert an chanel into a Sink. +-} +chanToSink :: (MonadIO io) => Chan a -> Sink a io () +chanToSink chan = do + val <- await + case val of + Nothing -> return () + Just v -> do + liftIO (writeChan chan v) + chanToSink chan + + +{- | + Merge two sources into one source. This is a concurrency abstraction. + The resulting source will produce items from either of the input sources + as they become available. As you would expect from a multi-producer, + single-consumer concurrency abstraction, the ordering of items produced + by each source is consistent relative to other items produced by + that same source, but the interleaving of items from both sources + is nondeterministic. +-} +merge :: (MonadIO io) => Source IO a -> Source IO b -> Source io (Either a b) +merge left right = do + chan <- liftIO newChan + (liftIO . void . forkIO) (left $= CL.map Left $$ chanToSink chan) + (liftIO . void . forkIO) (right $= CL.map Right $$ chanToSink chan) + chanToSource chan + + diff --git a/src/Network/Legion/ConnectionManager.hs b/src/Network/Legion/ConnectionManager.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/ConnectionManager.hs @@ -0,0 +1,209 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE TemplateHaskell #-} +{- | + This module manages connections to other nodes in the cluster. +-} +module Network.Legion.ConnectionManager ( + ConnectionManager, + newConnectionManager, + send, + newPeers, +) where + +import Prelude hiding (lookup) + +import Control.Concurrent (Chan, writeChan, newChan, readChan) +import Control.Exception (try, SomeException) +import Control.Monad (void) +import Control.Monad.Logger (logInfo, logWarn) +import Control.Monad.Trans.Class (lift) +import Data.Binary (Binary, encode) +import Data.ByteString.Lazy (ByteString) +import Data.Map (toList, insert, empty, Map, lookup) +import Data.Text (pack) +import Network.Legion.BSockAddr (BSockAddr(BSockAddr)) +import Network.Legion.Distribution (Peer) +import Network.Legion.Fork (forkC) +import Network.Legion.LIO (LIO) +import Network.Legion.StateMachine (PeerMessage) +import Network.Socket (SockAddr, Socket, socket, SocketType(Stream), + defaultProtocol, connect, close, SockAddr(SockAddrInet, SockAddrInet6, + SockAddrUnix, SockAddrCan), Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN)) +import Network.Socket.ByteString.Lazy (sendAll) + +{- | + A handle on the connection manager +-} +data ConnectionManager i o s = C (Chan (Message i o s)) +instance Show (ConnectionManager i o s) where + show _ = "ConnectionManager" + + +{- | + Create a new connection manager. +-} +newConnectionManager :: (Binary i, Binary o, Binary s) + => Map Peer BSockAddr + -> LIO (ConnectionManager i o s) +newConnectionManager initPeers = do + chan <- lift newChan + forkC "connection manager thread" $ + manager chan S {connections = empty} + let cm = C chan + newPeers cm initPeers + return cm + where + manager :: (Binary s, Binary o, Binary i) + => Chan (Message i o s) + -> State i o s + -> LIO () + manager chan state = lift (readChan chan) >>= handle state >>= manager chan + + handle :: (Binary i, Binary o, Binary s) + => State i o s + -> Message i o s + -> LIO (State i o s) + handle s@S {connections} (NewPeer peer addr) = + case lookup peer connections of + Nothing -> do + conn <- connection addr + return s { + connections = insert peer conn connections + } + Just _ -> + return s + + handle s@S {connections} (Send peer msg) = do + case lookup peer connections of + Nothing -> $(logWarn) . pack $ "unknown peer: " ++ show peer + Just conn -> lift $ writeChan conn msg + return s + + +{- | + Build a new connection. +-} +connection :: (Binary i, Binary o, Binary s) + => SockAddr + -> LIO (Chan (PeerMessage i o s)) + +connection addr = do + chan <- lift newChan + forkC ("connection to: " ++ show addr) $ + handle chan Nothing + return chan + where + handle :: (Binary i, Binary o, Binary s) + => Chan (PeerMessage i o s) + -> Maybe Socket + -> LIO () + handle chan so = + lift (readChan chan) >>= sendWithRetry so . encode >>= handle chan + + {- | + Open a socket. + -} + openSocket :: IO Socket + openSocket = do + so <- socket (fam addr) Stream defaultProtocol + connect so addr + return so + + {- | + Try to send the payload over the socket, and if that fails, then try to + create a new socket and retry sending the payload. Return whatever the + "working" socket is. + -} + sendWithRetry :: Maybe Socket -> ByteString -> LIO (Maybe Socket) + sendWithRetry Nothing payload = do + result <- (lift . try) openSocket + case result of + Left err -> do + $(logWarn) . pack + $ "Can't connect to: " ++ show addr ++ ". Dropping message on " + ++ "the floor: " ++ show payload ++ ". The error was: " + ++ show (err :: SomeException) + return Nothing + Right so -> do + result2 <- (lift . try) (sendAll so payload) + case result2 of + Left err -> $(logWarn) . pack + $ "An error happend when trying to send a payload over a socket " + ++ "to the address: " ++ show addr ++ ". The error was: " + ++ show (err :: SomeException) ++ ". This is the last straw, we " + ++ "are not retrying. The message is being dropped on the floor. " + ++ "The message was: " ++ show payload + Right _ -> return () + return (Just so) + sendWithRetry (Just so) payload = do + result <- (lift . try) (sendAll so payload) + case result of + Left err -> do + $(logInfo) . pack + $ "Socket to " ++ show addr ++ " died. Retrying on a new " + ++ "socket. The error was: " ++ show (err :: SomeException) + (lift . void) (try (close so) :: IO (Either SomeException ())) + sendWithRetry Nothing payload + Right _ -> + return (Just so) + + +{- | + Send a message to a peer. +-} +send + :: ConnectionManager i o s + -> Peer + -> PeerMessage i o s + -> LIO () +send (C chan) peer = lift . writeChan chan . Send peer + + +{- | + Tell the connection manager about a new peer. +-} +newPeer + :: ConnectionManager i o s + -> Peer + -> SockAddr + -> LIO () +newPeer (C chan) peer addr = lift $ writeChan chan (NewPeer peer addr) + + +{- | + Tell the connection manager about all the peers known to the cluster state. +-} +newPeers :: ConnectionManager i o s -> Map Peer BSockAddr -> LIO () +newPeers cm peers = + mapM_ oneNewPeer (toList peers) + where + oneNewPeer (peer, BSockAddr addy) = newPeer cm peer addy + + +{- | + The internal state of the connection manager. +-} +data State i o s = S { + connections :: Map Peer (Chan (PeerMessage i o s)) + } + + +{- | + The types of messages that the ConnectionManager understands. +-} +data Message i o s + = NewPeer Peer SockAddr + | Send Peer (PeerMessage i o s) + + +{- | + Guess the family of a `SockAddr`. +-} +fam :: SockAddr -> Family +fam SockAddrInet {} = AF_INET +fam SockAddrInet6 {} = AF_INET6 +fam SockAddrUnix {} = AF_UNIX +fam SockAddrCan {} = AF_CAN + + diff --git a/src/Network/Legion/Distribution.hs b/src/Network/Legion/Distribution.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/Distribution.hs @@ -0,0 +1,189 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{- | + This module defines the data structures and functions used for handling the + key space distribution. +-} +module Network.Legion.Distribution ( + ParticipationDefaults, + Peer, + empty, + modify, + findPartition, + rebalanceAction, + RebalanceAction(..), + newPeer, +) where + +import Prelude hiding (null) + +import Data.Binary (Binary) +import Data.Function (on) +import Data.List (sort, sortBy) +import Data.Set (Set, toList) +import Data.UUID (UUID) +import GHC.Generics (Generic) +import Network.Legion.KeySet (KeySet, member, (\\), null) +import Network.Legion.LIO (LIO) +import Network.Legion.PartitionKey (PartitionKey) +import Network.Legion.UUID (getUUID) +import qualified Data.Set as Set +import qualified Network.Legion.KeySet as KS + + +{- | + The way to identify a peer. +-} +newtype Peer = Peer UUID deriving (Show, Binary, Eq, Ord) + + +{- | + The distribution of partitions and partition replicas among the cluster. +-} +newtype ParticipationDefaults = D { + unD :: [(KeySet, Set Peer)] + } deriving (Show, Binary) + + +{- | + Constuct a distribution that contains no partitions. +-} +empty :: ParticipationDefaults + +empty = D [] + + +{- | + Find the peers that own the specified partition. +-} +findPartition :: PartitionKey -> ParticipationDefaults -> Set Peer + +findPartition k d = + case [ps | (ks, ps) <- unD d, k `member` ks] of + [ps] -> ps + _ -> error + $ "No exact mach for key in distribution. This means there is a bug in " + ++ "the module `Network.Legion.Distribution`. Please report this bug " + ++ "via github: " ++ show (k, d) + + +{- | + Modify the default participation for the key set. +-} +modify + :: (Set Peer -> Set Peer) + -> KeySet + -> ParticipationDefaults + -> ParticipationDefaults + +modify fun keyset = + {- + doModify can produce key ranges that contain zero keys, which is + why the `filter`. + -} + D . filter (not . null . fst) . doModify keyset . unD + where + doModify ks [] = [(ks, fun Set.empty)] + doModify ks ((r, ps):dist) = + let { + unaffected = r \\ ks; + affected = r \\ unaffected; + remaining = ks \\ affected; + } in + (unaffected, ps):(affected, fun ps):doModify remaining dist + + +{- | + Return the best action, if any, that the indicated peer should take to + rebalance an unbalanced distribution. +-} +rebalanceAction + :: Peer + -> Set Peer + -> ParticipationDefaults + -> Maybe RebalanceAction +rebalanceAction self allPeers (D dist) = + rebuild + {- TODO rebalance -} + where + _rebalance :: a + _rebalance = error "rebalance undefined" + rebuild = + let + underserved = [ + (ks, ps) + | (ks, ps) <- dist + , Set.size ps < 3 + , not (self `Set.member` ps) + ] + mostUnderserved = sortBy (compare `on` Set.size . snd) underserved + in case mostUnderserved of + [] -> Nothing + (ks, ps):_ -> + let + candidateHosts = toList (allPeers Set.\\ ps) + bestHosts = sort [(weightOf p, p) | p <- candidateHosts] + in case bestHosts of + {- we are the best host -} + (_, candidate):_ | candidate == self -> Just (Invite ks) + _ -> Nothing + + weightOf p = sum [KS.size ks | (ks, ps) <- dist, p `Set.member` ps] + +-- -- TODO: first figure out if any replicas need re-building. +-- case sortBy (weight `on` snd) (toList dist) of +-- (p, keyspace):remaining | p == peer -> +-- case reverse remaining of +-- [] -> Nothing +-- (target, targetSpace):_ -> +-- let +-- {- | +-- Keys that already exist at the target are not eligible +-- for being moved. +-- -} +-- eligibleSpace = keyspace \\ targetSpace +-- migrationSize = (size keyspace - size targetSpace) `div` 2 +-- migrants = pickMigrants migrationSize eligibleSpace +-- in +-- case migrants of +-- Just keys -> Just (Move target keys) +-- Nothing -> Nothing +-- _ -> Nothing +-- where +-- weight +-- :: KeySet +-- -> KeySet +-- -> Ordering +-- weight = flip compare `on` size +-- +-- pickMigrants :: Integer -> KeySet -> Maybe KeySet +-- pickMigrants n keyspace = +-- let migrants = take n keyspace in +-- if size migrants > 0 +-- then Just migrants +-- else Nothing + + +{- | + The actions that are taken in order to build a balanced cluster. +-} +data RebalanceAction + = Invite KeySet + deriving (Show, Generic) +instance Binary RebalanceAction + + +{- | + Create a new peer. +-} +newPeer :: LIO Peer +newPeer = Peer <$> getUUID + + +-- {- | +-- Trace helper +-- -} +-- t :: (Show a) => String -> a -> a +-- t msg a = trace (msg ++ ": " ++ show a) a + + diff --git a/src/Network/Legion/Fork.hs b/src/Network/Legion/Fork.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/Fork.hs @@ -0,0 +1,49 @@ +{-# LANGUAGE TemplateHaskell #-} +{- | + This module holds `forkC`, because we use it in at least two other modules. +-} +module Network.Legion.Fork ( + forkC +) where + +import Control.Concurrent (forkIO) +import Control.Exception (SomeException, try) +import Control.Monad (void) +import Control.Monad.Logger (logError, askLoggerIO, runLoggingT) +import Control.Monad.Trans.Class (lift) +import Data.Text (pack) +import Network.Legion.LIO (LIO) +import System.Exit (ExitCode(ExitFailure)) +import System.IO (hPutStrLn, stderr) +import System.Posix.Process (exitImmediately) + +{- | + Forks a critical thread. "Critical" in this case means that if the thread + crashes for whatever reason, then the program cannot continue correctly, so + we should crash the program instead of running in some kind of zombie broken + state. +-} +forkC + :: String + -- ^ The name of the critical thread, used for logging. + -> LIO () + -- ^ The IO to execute. + -> LIO () +forkC name io = do + logging <- askLoggerIO + lift . void . forkIO $ do + result <- try (runLoggingT io logging) + case result of + Left err -> do + let msg = + "Exception caught in critical thread " ++ show name + ++ ". We are crashing the entire program because we can't " + ++ "continue without this thread. The error was: " + ++ show (err :: SomeException) + -- write the message to every place we can think of. + (`runLoggingT` logging) . $(logError) . pack $ msg + putStrLn msg + hPutStrLn stderr msg + exitImmediately (ExitFailure 1) + Right v -> return v + diff --git a/src/Network/Legion/KeySet.hs b/src/Network/Legion/KeySet.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/KeySet.hs @@ -0,0 +1,205 @@ +{-# LANGUAGE DeriveGeneric #-} +{- | + This module contains the `KeySet` data type and operations. `KeySet`, + conceptually, has the same meaning as @Set PartitionKey@, but it is optimized + for non-sparse continuous ranges of included values that could never fit into + memory in an actual @Set PartitionKey@. +-} +module Network.Legion.KeySet ( + KeySet, + take, + size, + union, + unions, + member, + (\\), + empty, + null, + fromRange, + full +) where + +import Prelude hiding (take, null) + +import Data.Binary (Binary(put, get)) +import Data.Ranged (Range(Range), RSet, rSetEmpty, Boundary(BoundaryBelow, + BoundaryAbove, BoundaryAboveAll, BoundaryBelowAll), makeRangedSet, + rSetHas, rSetUnion, (-!-), unsafeRangedSet, rSetRanges) +import GHC.Generics (Generic) +import Network.Legion.PartitionKey (PartitionKey(K, unkey)) + + +{- | + Represents a set of partition keys. This type is intended to have set + semantics, but unlike `Data.Set.Set`, it performs well with dense sets + because it only stores the set of continuous ranges in memory. +-} +newtype KeySet = S {unS :: RSet PartitionKey} deriving (Show, Eq) + +instance Binary KeySet where + put = + put . fmap encodeRange . rSetRanges . unS + where + encodeRange + :: Range PartitionKey + -> (BinBoundary PartitionKey, BinBoundary PartitionKey) + encodeRange (Range a b) = (boundaryToBin a, boundaryToBin b) + + boundaryToBin :: Boundary a -> BinBoundary a + boundaryToBin (BoundaryBelow a) = BinBelow a + boundaryToBin (BoundaryAbove a) = BinAbove a + boundaryToBin BoundaryBelowAll = BinBelowAll + boundaryToBin BoundaryAboveAll = BinAboveAll + get = + (S . unsafeRangedSet . fmap decodeRange) <$> get + where + decodeRange + :: (BinBoundary PartitionKey, BinBoundary PartitionKey) + -> Range PartitionKey + decodeRange (a, b) = Range (binToBoundary a) (binToBoundary b) + + binToBoundary :: BinBoundary a -> Boundary a + binToBoundary (BinBelow a) = BoundaryBelow a + binToBoundary (BinAbove a) = BoundaryAbove a + binToBoundary BinBelowAll = BoundaryBelowAll + binToBoundary BinAboveAll = BoundaryAboveAll + + + +{- | + Construct the set of all partition keys within the specified range. Both the + start element and the end element are inclusive. +-} +fromRange :: PartitionKey -> PartitionKey -> KeySet +fromRange a b + | a > b = fromRange b a + | otherwise = S (makeRangedSet [Range (BoundaryBelow a) (BoundaryAbove b)]) + + +{- | + Construct an empty `KeySet`. +-} +empty :: KeySet +empty = S rSetEmpty + + +{- | + Construct a KeySet containing all keys. +-} +full :: KeySet +full = fromRange minBound maxBound + + +{- | + Check if a key range is empty or not. +-} +null :: KeySet -> Bool +null = (0 >=) . size + + +{- | + Take the difference of the two sets. +-} +(\\) :: KeySet -> KeySet -> KeySet +S a \\ S b = S (a -!- b) + + +{- | + Test for set membership. +-} +member :: PartitionKey -> KeySet -> Bool +member k = flip rSetHas k . unS + + +{- | + Take the union of the two sets. +-} +union :: KeySet -> KeySet -> KeySet +union (S a) (S b) = S (a `rSetUnion` b) + + +{- | + Take the union of a list of sets. +-} +unions :: [KeySet] -> KeySet +unions = foldr union empty + + +{- | + Used to help with the Binary instance of KeySet. +-} +data BinBoundary a + = BinAbove a + | BinBelow a + | BinAboveAll + | BinBelowAll + deriving (Generic) +instance (Binary a) => Binary (BinBoundary a) + + +{- | + Figure out how large a `KeySet` is. +-} +size :: KeySet -> Integer +size = sum . fmap rangeSize . rSetRanges . unS + + +{- | + Figure out how large a particular range is. +-} +rangeSize :: Range PartitionKey -> Integer +rangeSize (Range BoundaryBelowAll b) = rangeSize (Range (BoundaryBelow minBound) b) +rangeSize (Range BoundaryAboveAll b) = rangeSize (Range (BoundaryAbove maxBound) b) +rangeSize (Range a BoundaryBelowAll) = rangeSize (Range a (BoundaryBelow minBound)) +rangeSize (Range a BoundaryAboveAll) = rangeSize (Range a (BoundaryAbove maxBound)) +rangeSize (Range (BoundaryAbove a) (BoundaryAbove b)) = toI b - toI a +rangeSize (Range (BoundaryBelow a) (BoundaryBelow b)) = toI b - toI a +rangeSize (Range (BoundaryAbove a) (BoundaryBelow b)) = (toI b - toI a) - 1 +rangeSize (Range (BoundaryBelow a) (BoundaryAbove b)) = (toI b - toI a) + 1 + + +{- | + To help with `rangeSize`. +-} +toI :: PartitionKey -> Integer +toI = toInteger . unkey + + +{- | + Opposite of `toI` +-} +fromI :: Integer -> PartitionKey +fromI = K . fromInteger + + +{- | + Take the first n values from a KeySet. +-} +take :: Integer -> KeySet -> KeySet +take num set = + S $ doTake num [] (rSetRanges (unS set)) + where + doTake 0 acc _ = makeRangedSet acc + doTake _ acc [] = makeRangedSet acc + doTake n acc (first:remaining) + | firstSize <= n = + doTake (n - firstSize) (acc ++ [first]) remaining + | otherwise = + makeRangedSet (acc ++ [takeRange n first]) + where + firstSize = rangeSize first + + takeRange + :: Integer + -> Range PartitionKey + -> Range PartitionKey + takeRange n (Range BoundaryBelowAll b) = + takeRange n (Range (BoundaryBelow minBound) b) + takeRange n (Range BoundaryAboveAll b) = + takeRange n (Range (BoundaryAbove minBound) b) + takeRange n (Range (BoundaryAbove a) _) = + Range (BoundaryAbove a) (BoundaryAbove (fromI (toI a + n))) + takeRange n (Range (BoundaryBelow a) _) = + Range (BoundaryBelow a) (BoundaryBelow (fromI (toI a + n))) + + diff --git a/src/Network/Legion/LIO.hs b/src/Network/Legion/LIO.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/LIO.hs @@ -0,0 +1,17 @@ +{- | + This module defines the specialized logging monad in which legion + opperates. +-} +module Network.Legion.LIO ( + LIO +) where + +import Control.Monad.Logger (LoggingT) + + +{- | + 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 new file mode 100644 --- /dev/null +++ b/src/Network/Legion/PartitionKey.hs @@ -0,0 +1,95 @@ +{- | + This module contains the PartitionKey type. +-} +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. +-} +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) + get = do + (a, b, c, d) <- get + return (K (Word256 (Word128 a b) (Word128 c d))) + +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) + + diff --git a/src/Network/Legion/PartitionState.hs b/src/Network/Legion/PartitionState.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/PartitionState.hs @@ -0,0 +1,182 @@ +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{- | + This module contains types related to the partition state. +-} +module Network.Legion.PartitionState ( + PartitionPropState, + PartitionPowerState, + ask, + mergeEither, + actions, + new, + initProp, + participating, + getPowerState, + delta, + heartbeat, + participate, + projParticipants, + projected, + infimum, +) where + +import Data.Binary (Binary) +import Data.Default.Class (Default) +import Data.Set (Set) +import Data.Time.Clock (UTCTime) +import Network.Legion.Distribution (Peer) +import Network.Legion.PartitionKey (PartitionKey) +import Network.Legion.PowerState (ApplyDelta) +import Network.Legion.Propagation (PropState, PropPowerState) +import qualified Network.Legion.Propagation as P + +{- | + This is an opaque representation of your application's partition state. + Internally, this represents the complete, nondeterministic set of states the + partition can be in as a result of concurrency, eventual consistency, and all + the other distributed systems reasons your partition state might have more + than one value. + + 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 + } deriving (Show, Binary) + + +{- | + A reification of `PropState`, representing the propagation state of the + partition state. +-} +newtype PartitionPropState i s = PartitionPropState { + unPropState :: PropState PartitionKey s Peer i + } deriving (Eq, Show) + + +-- {- | +-- A convenient alias for the partition state infimum. +-- -} +-- type PartitionInfimum s = Infimum s Peer + + +{- | + Get the projected partition state value. +-} +ask :: (ApplyDelta i s) => PartitionPropState i s -> s +ask = P.ask . unPropState + + +{- | + Try to merge two partition states. +-} +mergeEither :: (Show i, Show s, ApplyDelta i s) + => Peer + -> PartitionPowerState i s + -> PartitionPropState i s + -> Either String (PartitionPropState i s) +mergeEither peer ps prop = + PartitionPropState <$> + P.mergeEither peer (unPowerState ps) (unPropState prop) + + +{- | + Get the peers which require action (i.e. Send), if any, and the + powerstate version to send to those peers, and the new propagation + state that is applicable after those actions have been taken. +-} +actions + :: PartitionPropState i s + -> (Set Peer, PartitionPowerState i s, PartitionPropState i s) +actions prop = + let (peers, ps, newProp) = P.actions (unPropState prop) + in (peers, PartitionPowerState ps, PartitionPropState newProp) + + +{- | + Create a new, default, PartitionPropState. +-} +new :: (Default s) + => PartitionKey + {- ^ The power state origin, which is the partition key. -} + -> Peer + {- ^ self -} + -> Set Peer + {- ^ The default participation. -} + -> PartitionPropState i s +new key self = PartitionPropState . P.new key self + + +{- | + Initialize a `PartitionPropState` based on the initial underlying + partition power state. +-} +initProp :: (ApplyDelta i s) + => Peer + -> PartitionPowerState i s + -> PartitionPropState i s +initProp self = PartitionPropState . P.initProp self . unPowerState + + +{- | + Return `True` if the local peer is participating in the partition + power state. +-} +participating :: PartitionPropState i s -> Bool +participating = P.participating . unPropState + + +{- | + Get an opaque encapsulation of the partition power state, for + transferring accros the network or whatever. +-} +getPowerState :: PartitionPropState i s -> PartitionPowerState i s +getPowerState = PartitionPowerState . P.getPowerState . unPropState + + +{- | Apply a delta to the partition state. -} +delta :: (ApplyDelta i s) + => i + -> PartitionPropState i s + -> PartitionPropState i s +delta d = PartitionPropState . P.delta d . unPropState + + +{- | Move time forward for the propagation state. -} +heartbeat :: UTCTime -> PartitionPropState i s -> PartitionPropState i 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) + => Peer + -> PartitionPropState i s + -> PartitionPropState i s +participate peer = PartitionPropState . P.participate peer . unPropState + + +{- | + Return the projected peers which are participating in the partition + state. +-} +projParticipants :: PartitionPropState i s -> Set Peer +projParticipants = P.projParticipants . unPropState + + +{- | + Get the projected value of a `PartitionPowerState`. +-} +projected :: (ApplyDelta i s) => PartitionPowerState i s -> s +projected = P.projected . unPowerState + + +{- | + Get the infimum value of a `PartitionPowerState` +-} +infimum :: PartitionPowerState i s -> s +infimum = P.infimum . unPowerState + + diff --git a/src/Network/Legion/PowerState.hs b/src/Network/Legion/PowerState.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/PowerState.hs @@ -0,0 +1,392 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE NamedFieldPuns #-} +{- | + This module contains the fundamental distributed data object. +-} +module Network.Legion.PowerState ( + PowerState, + Infimum(..), + ApplyDelta(..), + StateId, + new, + merge, + mergeMaybe, + mergeEither, + acknowledge, + participate, + disassociate, + projectedValue, + infimumValue, + infimumParticipants, + allParticipants, + projParticipants, + divergent, + divergences, + delta, +) where + +import Prelude hiding (null) + +import Data.Binary (Binary(put, get)) +import Data.Default.Class (Default(def)) +import Data.DoubleWord (Word256(Word256), Word128(Word128)) +import Data.Map (Map, filterWithKey, unionWith, minViewWithKey, keys, + toDescList, toAscList, fromAscList) +import Data.Set (Set, union, (\\), null, member) +import Data.Word (Word64) +import GHC.Generics (Generic) +import qualified Data.Map as Map (insert, empty) +import qualified Data.Set as Set (insert, empty, delete) + + +{- | + This represents the set of all possible future values of @s@, in a + distributed, monotonically increasing environment. +-} +data PowerState o s p d = 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) + + +{- | + `Infimum` is the infimum, or greatest lower bound, of the possible + values of @s@. +-} +data Infimum s p = Infimum { + stateId :: StateId p, + participants :: Set p, + stateValue :: s + } deriving (Generic, Show) +instance (Binary s, Binary p) => Binary (Infimum s p) +instance (Eq p) => Eq (Infimum s p) where + Infimum s1 _ _ == Infimum s2 _ _ = s1 == s2 +instance (Ord p) => Ord (Infimum s p) where + compare (Infimum s1 _ _) (Infimum s2 _ _) = compare s1 s2 + + +{- | + `StateId` is a monotonically increasing, totally ordered identification + value which allows us to lend the attribute of monotonicity to state + operations which would not naturally be monotonic. +-} +data StateId p + = BottomSid + | Sid Word256 p + deriving (Generic, Eq, Ord, Show) +instance (Binary p) => Binary (StateId p) where + put = put . toMaybe + where + toMaybe :: StateId p -> Maybe (Word64, Word64, Word64, Word64, p) + toMaybe BottomSid = + Nothing + toMaybe (Sid (Word256 (Word128 a b) (Word128 c d)) p) = + Just (a, b, c, d, p) + get = do + theThing <- get + return $ case theThing of + Nothing -> BottomSid + Just (a, b, c, d, p) -> Sid (Word256 (Word128 a b) (Word128 c d)) p +instance Default (StateId p) where + def = BottomSid + + +{- | + `Delta` is how we represent mutations to the power state. +-} +data Delta p d + = Join p + | UnJoin p + | Delta d + deriving (Generic, Show, Eq) +instance (Binary p, Binary d) => Binary (Delta p d) + + +{- | + The class which allows for delta application. +-} +class ApplyDelta i s where + {- | + Apply a delta to a state value. *This function MUST be total!!!* + -} + apply :: i -> s -> s + + +{- | + Construct a new PowerState with the given origin and initial participants +-} +new :: (Default s) => o -> Set p -> PowerState o s p d +new origin participants = + PowerState { + origin, + infimum = Infimum { + stateId = def, + participants, + stateValue = def + }, + deltas = Map.empty + } + + +{- | + Monotonically merge the information in two power states. The resulting + power state may have a higher infimum value, but it will never have + 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 a b = either error id (mergeEither a b) + + +{- | + 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 a b = either (const Nothing) Just (mergeEither a b) + + +{- | + 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 (PowerState o1 i1 d1) (PowerState o2 i2 d2) | o1 == o2 = + Right . reduce $ PowerState {origin = o1, infimum, deltas} + where + infimum = max i1 i2 + deltas = removeObsolete (unionWith mergeKnowns d1 d2) + removeObsolete = filterWithKey (\k _ -> k > stateId infimum) + mergeKnowns (d, s1) (_, s2) = (d, s1 `union` s2) + +mergeEither a b = Left + $ "PowerStates " ++ show a ++ " and " ++ show b ++ " do not share the " + ++ "same origin, and cannot be merged." + + +{- | + 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. +-} +acknowledge :: (ApplyDelta d s, Ord p) + => p + -> PowerState o s p d + -> PowerState o s p d +acknowledge p ps@PowerState {deltas} = + reduce ps {deltas = fmap ackOne deltas} + where + ackOne (d, acks) = (d, Set.insert p acks) + + +{- | + Allow a participant to join in the distributed nature of the power state. +-} +participate :: (ApplyDelta d s, Ord p) + => p + -> PowerState o s p d + -> PowerState o s p d +participate p ps@PowerState {deltas} = reduce ps { + deltas = Map.insert (nextId p ps) (Join p, Set.empty) deltas + } + + +{- | + Indicate that a participant is removing itself from participating in + the distributed power state. +-} +disassociate :: (ApplyDelta d s, Ord p) + => p + -> PowerState o s p d + -> PowerState o s p d +disassociate p ps@PowerState {deltas} = reduce ps { + deltas = Map.insert (nextId p ps) (UnJoin p, Set.empty) deltas + } + + +{- | + Introduce a change to the PowerState on behalf of the participant. +-} +delta :: (ApplyDelta d s, Ord p) + => p + -> d + -> PowerState o s p d + -> PowerState o s p d +delta p d ps@PowerState {deltas} = reduce ps { + deltas = Map.insert (nextId p ps) (Delta d, Set.empty) deltas + } + + +{- | + Return the current projected value of the power state. +-} +projectedValue :: (ApplyDelta d s) => PowerState o s p d -> s +projectedValue PowerState {infimum = Infimum {stateValue}, deltas} = + foldr apply stateValue changes + where + changes = foldr getDeltas [] (toDescList deltas) + getDeltas (_, (Delta d, _)) acc = d:acc + getDeltas _ acc = acc + + +{- | + Return the current infimum value of the power state. +-} +infimumValue :: PowerState o s p d -> s +infimumValue PowerState {infimum = Infimum {stateValue}} = stateValue + + +{- | + Gets the known participants at the infimum. +-} +infimumParticipants :: PowerState o s p d -> Set p +infimumParticipants PowerState {infimum = Infimum {participants}} = participants + + +{- | + Get all known participants. This includes participants that are + projected for removal. +-} +allParticipants :: (Ord p) => PowerState o s p d -> Set p +allParticipants PowerState { + infimum = Infimum {participants}, + deltas + } = + foldr updateParticipants participants (toDescList deltas) + where + updateParticipants (_, (Join p, _)) = Set.insert p + updateParticipants _ = id + + +{- | + 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 PowerState { + infimum = Infimum {participants}, + deltas + } = + foldr updateParticipants participants (toDescList deltas) + where + updateParticipants (_, (Join p, _)) = Set.insert p + updateParticipants (_, (UnJoin p, _)) = Set.delete p + updateParticipants _ = id + + +{- | + Returns the participants that we think might be diverging. In this + 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 PowerState { + infimum = Infimum {participants}, + deltas + } = + accum participants Set.empty (toAscList deltas) + where + {- | + `accum` mnemonics: + j = pro(J)ected participants + d = (D)ivergent participants + a = peers that have (A)cknowledged an update. + p = (P)eer that is joining or unjoining + -} + accum _ d [] = d + + accum j d ((_, (Join p, a)):moreDeltas) = + let + j2 = Set.insert p j + d2 = (j2 \\ a) `union` d + in + accum j2 d2 moreDeltas + + accum j d ((_, (UnJoin p, a)):moreDeltas) = + let + j2 = Set.delete p j + d2 = (j2 \\ a) `union` d + in + accum j2 d2 moreDeltas + + accum j d ((_, (Delta _, a)):moreDeltas) = + let + d2 = (j \\ a) `union` d + in + accum j d2 moreDeltas + + +{- | + Return the deltas that are unknown to the specified peer. +-} +divergences :: (Ord p) => p -> PowerState o s p d -> Map (StateId p) d +divergences peer PowerState {deltas} = + fromAscList [ + (sid, d) + | (sid, (Delta d, p)) <- toAscList deltas + , not (peer `member` p) + ] + + +{- | + This helper function is responsible for figuring out if the power state + 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 ps@PowerState { + infimum = infimum@Infimum {participants, stateValue}, + deltas + } = + case minViewWithKey deltas of + Nothing -> ps + Just ((i, (update, acks)), newDeltas) -> + if not . null $ participants \\ acks + then ps + else case update of + Join p -> reduce ps { + infimum = infimum { + stateId = i, + participants = Set.insert p participants + }, + deltas = newDeltas + } + UnJoin p -> reduce ps { + infimum = infimum { + stateId = i, + participants = Set.delete p participants + }, + deltas = newDeltas + } + Delta d -> reduce ps { + infimum = infimum { + stateId = i, + stateValue = apply d stateValue + }, + deltas = newDeltas + } + + +{- | + 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 p PowerState {infimum = Infimum {stateId}, deltas} = + case maximum (stateId:keys deltas) of + BottomSid -> Sid 0 p + Sid ord _ -> Sid (succ ord) p + + diff --git a/src/Network/Legion/Propagation.hs b/src/Network/Legion/Propagation.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/Propagation.hs @@ -0,0 +1,348 @@ +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE NamedFieldPuns #-} +{- | + This module defines how to propagate a PowerState amoung its participants. +-} +module Network.Legion.Propagation ( + PropState, + PropPowerState, + merge, + mergeMaybe, + mergeEither, + heartbeat, + delta, + actions, + new, + initProp, + getPowerState, + ask, + participate, + getSelf, + divergences, + participating, + allParticipants, + projParticipants, + projected, + infimum, +) where + +import Prelude hiding (lookup) + +import Data.Binary (Binary) +import Data.Default.Class (Default) +import Data.Map (Map, lookup) +import Data.Maybe (fromMaybe) +import Data.Set (member, Set) +import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime) +import Data.Time.Format () -- For `instance Show UTCTime` +import Network.Legion.PowerState (PowerState, divergent, ApplyDelta, + acknowledge, projectedValue, StateId) +import qualified Data.Map as Map +import qualified Data.Set as Set +import qualified Network.Legion.PowerState as PS + + +{- | + Internally, we use `Maybe UTCTime` to represent the current time, so that we + have a convenient way to represent "now" (i.e. `Nothing`) without using `IO`. + This type aliases gives us a convenient way to spell `Maybe UTCTime`. +-} +type Time = Maybe UTCTime + + +{- | + Opaque Propagation State. Values of this type encapsulate the + current value of a power state along with state having to do with + the distribution of that powerstate among its participants. The + power state is not directly accessible, but rather must be accessed + through functions provided by this module. In addition to providing + a more coherent hierarchy of abstraction, this also helps ensure that + 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, + peerStates :: Map p PeerStatus, + self :: p, + now :: Time + } deriving (Eq, Show) + + +{- | + This type is an opaque representation of the underlying power state. It + exists because we sometimes want to pack up the power state and ship + 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 + } deriving (Show, Binary) + + +{- | + Retriev the current projected value of the underlying state. +-} +ask :: (ApplyDelta d s) => PropState o s p d -> s +ask = projectedValue . powerState + + +{- | + Create a new propagation state based on an existing power state. +-} +initProp :: (ApplyDelta d s, Eq p, Ord p) + => p + -> PropPowerState o s p d + -> PropState o s p d +initProp self ps = + let powerState = acknowledge self (unPowerState ps) + in PropState { + powerState = powerState, + peerStates = Map.fromAscList [ + (p, NeedsSendAt Nothing) + | p <- Set.toAscList (divergent powerState) + ], + self, + now = Nothing + } + + +{- | + 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 = PropPowerState . powerState + + +{- | + The propagation state of a single remote participant. +-} +data PeerStatus + = NeedsSendAt Time + | NeedsAck + deriving (Show, Eq) + + +{- | + Create a new propagation state. +-} +new :: (Default s) => o -> p -> Set p -> PropState o s p d +new origin self participants = + PropState { + powerState = PS.new origin participants, + peerStates = Map.empty, + self, + now = Nothing + } + + +{- | + Like `merge`, but total. `mergeEither` returns a human readable reason why + the foreign powerstate can't be merged in the event of an error. +-} +{- + This algorithm is weaksauce. We need to find someone who knows a lot about + gossip protocols to fix this. +-} +mergeEither :: (Eq o, Ord p, Show o, Show s, Show p, Show d, ApplyDelta d s) + => p + -> PropPowerState o s p d + -> PropState o s p d + -> Either String (PropState o s p d) +mergeEither source kernel (prop@PropState {powerState, peerStates, self, now}) = + let ps = unPowerState kernel + in case acknowledge self <$> PS.mergeEither ps powerState of + Left err -> Left err + Right merged -> Right prop { + powerState = merged, + peerStates = + Map.fromList $ [ + (p, ns) + | p <- Set.toList (divergent merged) + , let ns = fromMaybe (NeedsSendAt now) (lookup p peerStates) + ] + ++ + {- + If the source of the foreign powerstate believes we + are divergent, then it is going to keep sending updates + until someone clues it in. That someone is us for now. + -} + [(source, NeedsAck) | self `member` divergent ps] + } + + +{- | + 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) + => p + -> PropPowerState o s p d + -> PropState o s p d + -> Maybe (PropState o s p d) +mergeMaybe source ps prop = + case mergeEither source ps prop of + Left _ -> Nothing + Right v -> Just v + + +{- | + Try to merge a foreign powerstate. The precondition is that the foreign + powerstate shares the same origin as the local powerstate. If this + 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) + => p + -> PropPowerState o s p d + -> PropState o s p d + -> PropState o s p d +merge source ps prop = + case mergeEither source ps prop of + Left err -> error err + Right v -> v + + +{- | + Time moves forward. +-} +heartbeat :: UTCTime -> PropState o s p d -> PropState o s p d +heartbeat newNow prop = prop {now = max (now prop) (Just newNow)} + + +{- | + Apply a delta. +-} +delta :: (Ord p, ApplyDelta d s) + => d + -> PropState o s p d + -> PropState o s p d +delta d prop@PropState {self, powerState, now} = + let newPowerState = acknowledge self (PS.delta self d powerState) + in prop { + powerState = newPowerState, + peerStates = Map.fromAscList [ + (p, NeedsSendAt now) + | p <- Set.toAscList (divergent newPowerState) + ] + } + + +{- | + Get the peers which require action (i.e. Send), if any, and the + powerstate version to send to those peers, and the new propagation + 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) +actions prop@PropState {powerState, peerStates, now} = + (outOfDatePeers, PropPowerState powerState, newPropState) + where + outOfDatePeers = Set.fromAscList [ + p + | (p, status) <- Map.toAscList peerStates + , shouldSendNow status + ] + + shouldSendNow NeedsAck = True + shouldSendNow (NeedsSendAt time) = now > time + + newPropState = prop { + peerStates = Map.fromAscList [ + (p, ns) + {- Careful, this pattern omits `NeedsAck`. This is intentional. -} + | (p, NeedsSendAt time) <- Map.toAscList peerStates + , let ns = NeedsSendAt (nextTime time) + ] + } + + nextTime :: Time -> Time + nextTime time = + if now > time + then addUTCTime gracePeriod <$> now + else time + + +{- | + The grace period for receiving some response to an action. +-} +gracePeriod :: NominalDiffTime +gracePeriod = oneMinute + where + oneMinute = 60 + + +{- | + Allow a participant to join in the distributed nature of the power state. +-} +participate :: (Ord p, ApplyDelta d s) + => p + -> PropState o s p d + -> PropState o s p d +participate peer prop@PropState {powerState, now} = + let newPowerState = PS.participate peer powerState + in prop { + powerState = newPowerState, + peerStates = Map.fromAscList [ + (p, NeedsSendAt now) + | p <- Set.toAscList (divergent newPowerState) + ] + } + + +{- | + Return the deltas that are unknown to the specified peer. +-} +divergences :: (Ord p) => p -> PropState o s p d -> Map (StateId p) d +divergences peer = PS.divergences peer . powerState + + +{- | + Return self. +-} +getSelf :: PropState o s p d -> p +getSelf = self + + +{- | + Return `True` if the local peer is participating in the underlying + power state. This will return `True` even if the peer is projected + 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 PropState{self, powerState} = + self `member` PS.allParticipants powerState + + +{- | + Get all known participants. This includes participants that are + projected for removal. +-} +allParticipants :: (Ord p) => PropState o s p d -> Set p +allParticipants = PS.allParticipants . powerState + + +{- | + Get all of the projected participants. +-} +projParticipants :: (Ord p) => PropState o s p d -> Set p +projParticipants = PS.projParticipants . powerState + + +{- | + Get the projected value of a PropPowerState. +-} +projected :: (ApplyDelta d s) => PropPowerState o s p d -> s +projected = PS.projectedValue . unPowerState + + +{- | + Get the infimum value of the PropPowerState. +-} +infimum :: PropPowerState o s p d -> s +infimum = PS.infimumValue . unPowerState + + diff --git a/src/Network/Legion/Runtime.hs b/src/Network/Legion/Runtime.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/Runtime.hs @@ -0,0 +1,358 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TemplateHaskell #-} +{- | + This module is responsible for the runtime operation of the legion + framework. This mostly means opening sockets and piping data around to the + various connected pieces. +-} +module Network.Legion.Runtime ( + forkLegionary, + runLegionary, + StartupMode(..), +) where + +import Control.Concurrent (forkIO) +import Control.Concurrent.Chan (writeChan, newChan, Chan) +import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar) +import Control.Monad (void, forever, join) +import Control.Monad.Catch (catchAll, try, SomeException, throwM) +import Control.Monad.IO.Class (liftIO) +import Control.Monad.Logger (logWarn, logError, logInfo, LoggingT, + MonadLoggerIO, runLoggingT, askLoggerIO) +import Control.Monad.Trans.Class (lift) +import Data.Binary (encode) +import Data.Conduit (Source, ($$), (=$=), yield, await, awaitForever, + transPipe, ConduitM, runConduit) +import Data.Conduit.Network (sourceSocket) +import Data.Conduit.Serialization.Binary (conduitDecode) +import Data.Map (Map) +import Data.Text (pack) +import Network.Legion.Admin (runAdmin) +import Network.Legion.Application (LegionConstraints, Legionary, + RequestMsg) +import Network.Legion.BSockAddr (BSockAddr(BSockAddr)) +import Network.Legion.ClusterState (ClusterPowerState) +import Network.Legion.Conduit (merge, chanToSink, chanToSource) +import Network.Legion.ConnectionManager (newConnectionManager, send, + newPeers) +import Network.Legion.Distribution (Peer, newPeer) +import Network.Legion.Fork (forkC) +import Network.Legion.LIO (LIO) +import Network.Legion.PartitionKey (PartitionKey) +import Network.Legion.Settings (LegionarySettings(LegionarySettings, + adminHost, adminPort, peerBindAddr, joinBindAddr)) +import Network.Legion.StateMachine (stateMachine, LInput(J, P, R, + A), JoinRequest(JoinRequest), JoinResponse(JoinOk, JoinRejected), + LOutput(Send, NewPeers), AdminMessage, NodeState, PeerMessage, + newNodeState) +import Network.Legion.UUID (getUUID) +import Network.Socket (Family(AF_INET, AF_INET6, AF_UNIX, AF_CAN), + SocketOption(ReuseAddr), SocketType(Stream), accept, bindSocket, + defaultProtocol, listen, setSocketOption, socket, SockAddr(SockAddrInet, + SockAddrInet6, SockAddrUnix, SockAddrCan), connect, getPeerName, Socket) +import Network.Socket.ByteString.Lazy (sendAll) +import qualified Data.Conduit.List as CL +import qualified Network.Legion.ClusterState as C + + +{- | + 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 i o s) + => Legionary i o s + -- ^ The user-defined legion application to run. + -> LegionarySettings + -- ^ Settings and configuration of the legionary framework. + -> StartupMode + -> Source IO (RequestMsg i o) + -- ^ A source of requests, together with a way to respond to the requets. + {- + We don't use `LIO` in the type signature here because we don't + export the `LIO` symbol. + -} + -> LoggingT IO () + +runLegionary + legionary + settings@LegionarySettings {adminHost, adminPort} + startupMode + requestSource + = do + peerS <- loggingC =<< startPeerListener settings + (nodeState, peers) <- makeNodeState settings startupMode + cm <- newConnectionManager peers + $(logInfo) . pack + $ "The initial node state is: " ++ show nodeState + adminS <- loggingC =<< runAdmin adminPort adminHost + joinS <- loggingC (joinMsgSource settings) + runConduit $ + (joinS `merge` (peerS `merge` (requestSource `merge` adminS))) + =$= CL.map toMessage + =$= stateMachine legionary nodeState + =$= handleOutput cm + where + handleOutput cm = awaitForever (lift . \case + Send peer message -> send cm peer message + NewPeers peers -> newPeers cm peers + ) + + toMessage + :: Either + (JoinRequest, JoinResponse -> LIO ()) + (Either + (PeerMessage i o s) + (Either (RequestMsg i o) (AdminMessage i o s))) + -> LInput i o s + toMessage (Left m) = J m + toMessage (Right (Left m)) = P m + toMessage (Right (Right (Left m))) = R m + toMessage (Right (Right (Right m))) = A m + + {- | + Turn an LIO-based conduit into an IO-based conduit, so that it + will work with `merge`. + -} + loggingC :: ConduitM i o LIO r -> LIO (ConduitM i o IO r) + loggingC c = do + logging <- askLoggerIO + return (transPipe (`runLoggingT` logging) c) + + +{- | 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. Future plans include + -- implementing some safeguards to make sure only one node in + -- a cluster was started using this startup mode, but for now, + -- we are counting on you, the user, to do the right thing. + | JoinCluster SockAddr + -- ^ Indicates that the node should try to join an existing cluster, + -- either by starting fresh, or by recovering from a shutdown + -- or crash. + deriving (Show, Eq) + + +{- | + Construct a source of incoming peer messages. We have to start the + peer listener first before we spin up the cluster management, which + is why this is an @LIO (Source LIO PeerMessage)@ instead of a + @Source LIO PeerMessage@. +-} +startPeerListener :: (LegionConstraints i o s) + => LegionarySettings + -> LIO (Source LIO (PeerMessage i o s)) + +startPeerListener LegionarySettings {peerBindAddr} = + catchAll (do + (inputChan, so) <- lift $ do + inputChan <- newChan + so <- socket (fam peerBindAddr) Stream defaultProtocol + setSocketOption so ReuseAddr 1 + bindSocket so peerBindAddr + listen so 5 + return (inputChan, so) + forkC "peer socket acceptor" $ acceptLoop so inputChan + return (chanToSource inputChan) + ) (\err -> do + $(logError) . pack + $ "Couldn't start incomming peer message service, because of: " + ++ show (err :: SomeException) + throwM err + ) + where + acceptLoop :: (LegionConstraints i o s) + => Socket + -> Chan (PeerMessage i o s) + -> LIO () + acceptLoop so inputChan = + catchAll ( + forever $ do + (conn, _) <- lift $ accept so + remoteAddr <- lift $ getPeerName conn + logging <- askLoggerIO + let runSocket = + sourceSocket conn + =$= conduitDecode + $$ msgSink + void + . lift + . forkIO + . (`runLoggingT` logging) + . logErrors remoteAddr + $ runSocket + ) (\err -> do + $(logError) . pack + $ "error in peer message accept loop: " + ++ show (err :: SomeException) + throwM err + ) + where + msgSink = chanToSink inputChan + + logErrors :: SockAddr -> LIO () -> LIO () + logErrors remoteAddr io = do + result <- try io + case result of + Left err -> + $(logWarn) . pack + $ "Incomming peer connection (" ++ show remoteAddr + ++ ") crashed because of: " ++ show (err :: SomeException) + Right v -> return v + + +{- | Figure out how to construct the initial node state. -} +makeNodeState :: (LegionConstraints i o s) + => LegionarySettings + -> StartupMode + -> LIO (NodeState i o s, Map Peer BSockAddr) + +makeNodeState LegionarySettings {peerBindAddr} NewCluster = do + {- Build a brand new node state, for the first node in a cluster. -} + self <- newPeer + clusterId <- getUUID + let cluster = C.new clusterId self peerBindAddr + nodeState <- newNodeState self cluster + return (nodeState, C.getPeers cluster) + +makeNodeState LegionarySettings {peerBindAddr} (JoinCluster addr) = do + {- + Join a cluster by either starting fresh, or recovering from a + shutdown or crash. + -} + $(logInfo) "Trying to join an existing cluster." + (self, clusterPS) <- joinCluster (JoinRequest (BSockAddr peerBindAddr)) + let cluster = C.initProp self clusterPS + nodeState <- newNodeState self cluster + return (nodeState, C.getPeers cluster) + where + joinCluster :: JoinRequest -> LIO (Peer, ClusterPowerState) + joinCluster joinMsg = liftIO $ do + so <- socket (fam addr) Stream defaultProtocol + connect so addr + sendAll so (encode joinMsg) + {- + using sourceSocket and conduitDecode is easier than building + a recive/decode state loop, even though we only read a single + response. + -} + sourceSocket so =$= conduitDecode $$ do + response <- await + case response of + Nothing -> fail + $ "Couldn't join a cluster because there was no response " + ++ "to our join request!" + Just (JoinOk self cps) -> + return (self, cps) + Just (JoinRejected reason) -> fail + $ "The cluster would not allow us to re-join. " + ++ "The reason given was: " ++ show reason + + +{- | A source of cluster join request messages. -} +joinMsgSource + :: LegionarySettings + -> Source LIO (JoinRequest, JoinResponse -> LIO ()) + +joinMsgSource LegionarySettings {joinBindAddr} = join . lift $ + catchAll (do + (chan, so) <- lift $ do + chan <- newChan + so <- socket (fam joinBindAddr) Stream defaultProtocol + setSocketOption so ReuseAddr 1 + bindSocket so joinBindAddr + listen so 5 + return (chan, so) + forkC "join socket acceptor" $ acceptLoop so chan + return (chanToSource chan) + ) (\err -> do + $(logError) . pack + $ "Couldn't start join request service, because of: " + ++ show (err :: SomeException) + throwM err + ) + where + acceptLoop :: Socket -> Chan (JoinRequest, JoinResponse -> LIO ()) -> LIO () + acceptLoop so chan = + catchAll ( + forever $ do + (conn, _) <- lift $ accept so + logging <- askLoggerIO + (void . lift . forkIO . (`runLoggingT` logging) . logErrors) ( + sourceSocket conn + =$= conduitDecode + =$= attachResponder conn + $$ chanToSink chan + ) + ) (\err -> do + $(logError) . pack + $ "error in join request accept loop: " + ++ show (err :: SomeException) + throwM err + ) + where + logErrors :: LIO () -> LIO () + logErrors io = do + result <- try io + case result of + Left err -> + $(logWarn) . pack + $ "Incomming join connection crashed because of: " + ++ show (err :: SomeException) + Right v -> return v + + attachResponder + :: Socket + -> ConduitM JoinRequest (JoinRequest, JoinResponse -> LIO ()) LIO () + attachResponder conn = awaitForever (\msg -> do + mvar <- liftIO newEmptyMVar + yield (msg, lift . putMVar mvar) + response <- liftIO $ takeMVar mvar + liftIO $ sendAll conn (encode response) + ) + + +{- | Guess the family of a `SockAddr`. -} +fam :: SockAddr -> Family +fam SockAddrInet {} = AF_INET +fam SockAddrInet6 {} = AF_INET6 +fam SockAddrUnix {} = AF_UNIX +fam SockAddrCan {} = AF_CAN + + +{- | + Forks the legion framework in a background thread, and returns a way to + send user requests to it and retrieve the responses to those requests. +-} +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 -> IO o) + +forkLegionary legionary settings startupMode = do + logging <- askLoggerIO + liftIO . (`runLoggingT` logging) $ do + chan <- liftIO newChan + forkC "main legion thread" $ + runLegionary legionary settings startupMode (chanToSource chan) + return (\ key request -> do + responseVar <- newEmptyMVar + writeChan chan ((key, request), putMVar responseVar) + takeMVar responseVar + ) + + diff --git a/src/Network/Legion/Settings.hs b/src/Network/Legion/Settings.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/Settings.hs @@ -0,0 +1,33 @@ +{- | + This module contains the user settings. +-} +module Network.Legion.Settings ( + LegionarySettings(..), +) where + +import Network.Socket (SockAddr) +import Network.Wai.Handler.Warp (HostPreference, Port) + +{- | Settings used when starting up the legion framework. -} +data LegionarySettings = LegionarySettings { + peerBindAddr :: SockAddr, + {- ^ + The address on which the legion framework will listen for + rebalancing and cluster management commands. + -} + joinBindAddr :: SockAddr, + {- ^ + The address on which the legion framework will listen for cluster + join requests. + -} + adminHost :: HostPreference, + {- ^ + The host address on which the admin service should run. + -} + adminPort :: Port + {- ^ + The host port on which the admin service should run. + -} + } + + diff --git a/src/Network/Legion/StateMachine.hs b/src/Network/Legion/StateMachine.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/StateMachine.hs @@ -0,0 +1,644 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TemplateHaskell #-} +{- | + This module contains the state machine implementation of a legion node. +-} +module Network.Legion.StateMachine ( + stateMachine, + LInput(..), + LOutput(..), + JoinRequest(..), + JoinResponse(..), + AdminMessage(..), + NodeState, + Forwarded(..), + PeerMessage(..), + PeerMessagePayload(..), + MessageId, + next, + newNodeState, +) where + +import Prelude hiding (lookup) + +import Control.Exception (throw) +import Control.Monad (unless) +import Control.Monad.Catch (try, SomeException, MonadCatch, MonadThrow) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Logger (logDebug, logWarn, logError, logInfo, + MonadLogger) +import Control.Monad.Trans.Class (MonadTrans, lift) +import Control.Monad.Trans.State (StateT, runStateT, get, put) +import Data.Binary (Binary) +import Data.Conduit (Source, Conduit, ($$), await, awaitForever, + transPipe, ConduitM, yield, ($=)) +import Data.Default.Class (Default) +import Data.Map (Map, insert, lookup) +import Data.Maybe (fromMaybe) +import Data.Set (member, minView, (\\)) +import Data.Text (pack) +import Data.Time.Clock (getCurrentTime) +import Data.UUID (UUID) +import Data.Word (Word64) +import GHC.Generics (Generic) +import Network.Legion.Application (Legionary, LegionConstraints, + Persistence(getState, saveState, list), Legionary(Legionary, + persistence, handleRequest), RequestMsg) +import Network.Legion.BSockAddr (BSockAddr) +import Network.Legion.ClusterState (claimParticipation, ClusterPropState, + getPeers, getDistribution, ClusterPowerState) +import Network.Legion.Distribution (rebalanceAction, RebalanceAction( + Invite), Peer, newPeer) +import Network.Legion.KeySet (union, KeySet) +import Network.Legion.LIO (LIO) +import Network.Legion.PartitionKey (PartitionKey) +import Network.Legion.PartitionState (PartitionPowerState, PartitionPropState) +import Network.Legion.PowerState (ApplyDelta) +import Network.Legion.UUID (getUUID) +import qualified Data.Conduit.List as CL +import qualified Data.Map as Map +import qualified Data.Set as Set +import qualified Network.Legion.ClusterState as C +import qualified Network.Legion.KeySet as KS +import qualified Network.Legion.PartitionState as P + + +{- | This conduit houses the main legionary state machine. -} +stateMachine :: (LegionConstraints i o s) + => Legionary i o s + -> NodeState i o s + -> Conduit (LInput i o s) LIO (LOutput i o s) +stateMachine l n = awaitForever (\msg -> do + newState <- runStateMT n $ do + handleMessage l msg + heartbeat + migrate l + propagate + rebalance l + logState + stateMachine l newState + ) + where + logState = lift . logNodeState =<< getS + + +{- | Handle one incomming message. -} +handleMessage :: (LegionConstraints i o s) + => Legionary i o s + -> LInput i o s + -> StateM i o s () + +handleMessage l msg = do + NodeState {cluster} <- getS + let + {- | Return `True` if the peer is a known peer, false otherwise. -} + known peer = peer `member` C.allParticipants cluster + $(logDebug) . pack $ "Receiving: " ++ show msg + case msg of + P peerMsg@PeerMessage {source} -> + if known source + then handlePeerMessage l peerMsg + else + $(logWarn) . pack + $ "Dropping message from unknown peer: " ++ show source + R ((key, request), respond) -> + case minView (C.findPartition key cluster) of + Nothing -> + $(logError) . pack + $ "Keyspace does not contain key: " ++ show key ++ ". This " + ++ "is a very bad thing and probably means there is a bug, " + ++ "or else this node has not joined a cluster yet." + Just (peer, _) -> + forward peer key request respond + J m -> handleJoinRequest m + A m -> lift . handleAdminMessage l m =<< getS + + +{- | Handles one incomming message from a peer. -} +handlePeerMessage :: (LegionConstraints i o s) + => Legionary i o s + -> PeerMessage i o s + -> StateM i o s () + +handlePeerMessage -- PartitionMerge + Legionary { + persistence + } + msg@PeerMessage { + source, + payload = PartitionMerge key ps + } + = do + nodeState@NodeState {self, propStates, cluster} <- getS + propState <- lift $ maybe + (getStateL persistence self cluster key) + return + (lookup key propStates) + let + owners = C.findPartition key cluster + case P.mergeEither source ps propState of + Left err -> + $(logWarn) . pack + $ "Can't apply incomming partition action message " + ++ show msg ++ "because of: " ++ show err + Right newPropState -> do + $(logDebug) "Saving because of PartitionMerge" + lift $ saveStateL persistence key ( + if P.participating newPropState + then Just (P.getPowerState newPropState) + else Nothing + ) + putS nodeState { + propStates = if newPropState == P.new key self owners + then Map.delete key propStates + else insert key newPropState propStates + } + +handlePeerMessage -- ForwardRequest + Legionary {handleRequest, persistence} + msg@PeerMessage { + payload = ForwardRequest key request, + source, + messageId + } + = do + ns@NodeState {self, cluster, propStates} <- getS + let owners = C.findPartition key cluster + if self `member` owners + then do + let + respond = send source . ForwardResponse messageId + + -- TODO + -- - figure out some slick concurrency here, by maintaining + -- a map of keys that are currently being accessed or + -- something + -- + either (respond . rethrow) respond =<< try (do + prop <- lift $ getStateL persistence self cluster key + let response = handleRequest key request (P.ask prop) + newProp = P.delta request prop + $(logDebug) "Saving because of ForwardRequest" + lift $ saveStateL persistence key (Just (P.getPowerState newProp)) + $(logInfo) . pack + $ "Handling user request: " ++ show request + $(logDebug) . pack + $ "Request details request: " ++ show prop ++ " ++ " + ++ show request ++ " --> " ++ show (response, newProp) + putS ns {propStates = insert key newProp propStates} + return response + ) + else + {- + we don't own the key after all, someone was wrong to forward + us this request. + -} + case minView owners of + Nothing -> $(logError) . pack + $ "Can't find any owners for the key: " ++ show key + Just (peer, _) -> + emit (Send peer msg) + where + {- | + rethrow is just a reification of `throw`. + -} + rethrow :: SomeException -> a + rethrow = throw + +handlePeerMessage -- ForwardResponse + Legionary {} + msg@PeerMessage { + payload = ForwardResponse messageId response + } + = do + nodeState@NodeState {forwarded} <- getS + case lookup messageId (unF forwarded) of + Nothing -> $(logWarn) . pack + $ "This peer received a response for a forwarded request that it " + ++ "didn't send. The only time you might expect to see this is if " + ++ "this peer recently crashed and was immediately restarted. If " + ++ "you are seeing this in other circumstances then probably " + ++ "something is very wrong at the network level. The message was: " + ++ show msg + Just respond -> + lift $ respond response + putS nodeState { + forwarded = F . Map.delete messageId . unF $ forwarded + } + +handlePeerMessage -- ClusterMerge + Legionary {} + msg@PeerMessage { + source, + payload = ClusterMerge ps + } + = do + nodeState@NodeState {migration, cluster} <- getS + case C.mergeEither source ps cluster of + Left err -> + $(logWarn) . pack + $ "Can't apply incomming cluster action message " + ++ show msg ++ "because of: " ++ show err + Right (newCluster, newMigration) -> + putS nodeState { + migration = migration `union` newMigration, + cluster = newCluster + } + + +{- | Handle a join request message -} +handleJoinRequest + :: (JoinRequest, JoinResponse -> LIO ()) + -> StateM i o s () + +handleJoinRequest (JoinRequest peerAddr, respond) = do + ns@NodeState {cluster} <- getS + peer <- lift newPeer + let newCluster = C.joinCluster peer peerAddr cluster + emit . NewPeers . getPeers $ newCluster + lift $ respond (JoinOk peer (C.getPowerState newCluster)) + putS ns {cluster = newCluster} + + +{- | + Handle a message from the admin service. +-} +handleAdminMessage + :: Legionary i o s + -> AdminMessage i o s + -> NodeState i o s + -> LIO () +handleAdminMessage _ (GetState respond) ns = + respond ns +handleAdminMessage Legionary {persistence} (GetPart key respond) _ = do + partitionVal <- lift (getState persistence key) + respond partitionVal + + +{- | Update all of the propagation states with the current time. -} +heartbeat :: StateM i o s () +heartbeat = do + now <- liftIO getCurrentTime + ns@NodeState {cluster, propStates} <- getS + putS ns { + cluster = C.heartbeat now cluster, + propStates = Map.fromAscList [ + (k, P.heartbeat now p) + | (k, p) <- Map.toAscList propStates + ] + } + + +{- | + Migrate partitions based on new cluster state information. + + TODO: this migration algorithm is super naive. It just goes ahead + and migrates everything in one pass, which is going to be terrible + for performance. + + Also, it is important to remember that "migrate" in this context does + not mean "transfer data". Rather, "migrate" means to add a participating + peer to a partition. This will cause the data to be transfered in the + normal course of propagation. +-} +migrate :: (LegionConstraints i o s) => Legionary i o s -> StateM i o s () +migrate Legionary{persistence} = do + ns@NodeState {migration} <- getS + unless (KS.null migration) $ + putS =<< lift ( + listL persistence + $= CL.filter ((`KS.member` migration) . fst) + $$ accum ns {migration = KS.empty} + ) + where + accum ns@NodeState {self, cluster, propStates} = await >>= \case + Nothing -> return ns + Just (key, ps) -> + let + origProp = fromMaybe (P.initProp self ps) (lookup key propStates) + newPeers_ = C.findPartition key cluster \\ P.projParticipants origProp + {- This 'P.participate' is where the magic happens. -} + newProp = foldr P.participate origProp (Set.toList newPeers_) + in do + $(logDebug) . pack $ "Migrating: " ++ show key + lift (saveStateL persistence key (Just (P.getPowerState newProp))) + accum ns { + propStates = Map.insert key newProp propStates + } + + +{- | + Handle all cluster and partition state propagation actions, and return + an updated node state. +-} +propagate :: (LegionConstraints i o s) => StateM i o s () +propagate = do + ns@NodeState {cluster, propStates, self} <- getS + let (peers, ps, cluster2) = C.actions cluster + $(logDebug) . pack $ "Cluster Actions: " ++ show (peers, ps) + mapM_ (doClusterAction ps) (Set.toList peers) + propStates2 <- mapM doPartitionActions (Map.toList propStates) + putS ns { + cluster = cluster2, + propStates = Map.fromAscList [ + (k, p) + | (k, p) <- propStates2 + , p /= P.initProp self (P.getPowerState p) + ] + } + where + doClusterAction ps peer = + send peer (ClusterMerge ps) + + doPartitionActions (key, propState) = do + let (peers, ps, propState2) = P.actions propState + mapM_ (perform ps) (Set.toList peers) + return (key, propState2) + where + perform ps peer = + send peer (PartitionMerge key ps) + + +{- | + Figure out if any rebalancing actions must be taken by this node, and kick + them off if so. +-} +rebalance :: (LegionConstraints i o s) => Legionary i o s -> StateM i o s () +rebalance _ = do + ns@NodeState {self, cluster} <- getS + let + allPeers = (Set.fromList . Map.keys . getPeers) cluster + dist = getDistribution cluster + action = rebalanceAction self allPeers dist + $(logDebug) . pack $ "The rebalance action is: " ++ show action + putS ns { + cluster = case action of + Nothing -> cluster + Just (Invite ks) -> claimParticipation self ks cluster + } + + +{- | This is the type of input accepted by the legionary state machine. -} +data LInput i o s + = P (PeerMessage i o s) + | R (RequestMsg i o) + | J (JoinRequest, JoinResponse -> LIO ()) + | A (AdminMessage i o s) + +instance (Show i, Show o, Show s) => Show (LInput i o s) where + show (P m) = "(P " ++ show m ++ ")" + show (R ((p, i), _)) = "(R ((" ++ show p ++ ", " ++ show i ++ "), _))" + show (J (jr, _)) = "(J (" ++ show jr ++ ", _))" + show (A a) = "(A (" ++ show a ++ "))" + + +{- | This is the type of output produced by the legionary state machine. -} +data LOutput i o s + = Send Peer (PeerMessage i o s) + | NewPeers (Map Peer BSockAddr) + + +{- | A helper function to log the state of the node: -} +logNodeState :: (LegionConstraints i o s) => NodeState i o s -> LIO () +logNodeState ns = $(logDebug) . pack + $ "The current node state is: " ++ show ns + + +{- | Like `getState`, but in LIO, and provides the correct bottom value. -} +getStateL :: (ApplyDelta i s, Default s) + => Persistence i s + -> Peer + -> ClusterPropState + -> PartitionKey + -> LIO (PartitionPropState i s) + +getStateL p self cluster key = + {- dp == default participants -} + let dp = C.findPartition key cluster + in maybe + (P.new key self dp) + (P.initProp self) + <$> lift (getState p key) + + +{- | Like `saveState`, but in LIO. -} +saveStateL + :: Persistence i s + -> PartitionKey + -> Maybe (PartitionPowerState i s) + -> LIO () +saveStateL p k = lift . saveState p k + + +{- | Like `list`, but in LIO. -} +listL :: Persistence i s -> Source LIO (PartitionKey, PartitionPowerState i s) +listL p = transPipe lift (list p) + + +{- | This is the type of a join request message. -} +data JoinRequest = JoinRequest BSockAddr + deriving (Generic, Show) +instance Binary JoinRequest + + +{- | The response to a JoinRequst message -} +data JoinResponse + = JoinOk Peer ClusterPowerState + | JoinRejected String + deriving (Generic) +instance Binary JoinResponse + + +{- | + The type of messages sent by the admin service. +-} +data AdminMessage i o s + = GetState (NodeState i o s -> LIO ()) + | GetPart PartitionKey (Maybe (PartitionPowerState i s) -> LIO ()) + +instance Show (AdminMessage i o s) where + show (GetState _) = "(GetState _)" + show (GetPart k _) = "(GetPart " ++ show k ++ " _)" + + +{- | Defines the local state of a node in the cluster. -} +data NodeState i o s = NodeState { + self :: Peer, + cluster :: ClusterPropState, + forwarded :: Forwarded o, + propStates :: Map PartitionKey (PartitionPropState i s), + migration :: KeySet, + nextId :: MessageId + } + deriving (Show) + + +{- | A set of forwardmed messages. -} +newtype Forwarded o = F {unF :: Map MessageId (o -> LIO ())} +instance Show (Forwarded o) where + show = show . Map.keys . unF + + +{- | + The type of messages sent to us from other peers. +-} +data PeerMessage i o s = PeerMessage { + source :: Peer, + messageId :: MessageId, + payload :: PeerMessagePayload i o s + } + deriving (Generic, Show) +instance (Binary i, Binary o, Binary s) => Binary (PeerMessage i o s) + + +{- | + The data contained within a peer message. + + When we get around to implementing durability and data replication, + the sustained inability to confirm that a node has received one of + these messages should result in the ejection of that node from the + cluster and the blacklisting of that node so that it can never re-join. +-} +data PeerMessagePayload i o s + = PartitionMerge PartitionKey (PartitionPowerState i s) + | ForwardRequest PartitionKey i + | ForwardResponse MessageId o + | ClusterMerge ClusterPowerState + deriving (Generic, Show) +instance (Binary i, Binary o, Binary s) => Binary (PeerMessagePayload i o s) + + +data MessageId = M UUID Word64 deriving (Generic, Show, Eq, Ord) +instance Binary MessageId + + +{- | + Generate the next message id in the sequence. We would normally use + `succ` for this kind of thing, but making `MessageId` an instance of + `Enum` really isn't appropriate. +-} +next :: MessageId -> MessageId +next (M sequenceId ord) = M sequenceId (ord + 1) + + +{- | + Initialize a new sequence of messageIds +-} +newSequence :: LIO MessageId +newSequence = lift $ do + sid <- getUUID + return (M sid 0) + + +{- | + Make a new node state. +-} +newNodeState :: Peer -> ClusterPropState -> LIO (NodeState i o s) +newNodeState self cluster = do + nextId <- newSequence + return NodeState { + self, + nextId, + cluster, + forwarded = F Map.empty, + propStates = Map.empty, + migration = KS.empty + } + + +send :: Peer -> PeerMessagePayload i o s -> StateM i o s () +send peer payload = do + ns@NodeState {self, nextId} <- getS + emit (Send peer PeerMessage { + source = self, + messageId = nextId, + payload + }) + putS ns {nextId = next nextId} + + +{- | + Forward a user request to a peer for handling, being sure to do all + the node state accounting. +-} +forward + :: Peer + -> PartitionKey + -> i + -> (o -> IO ()) + -> StateM i o s () +forward peer key request respond = do + ns@NodeState {nextId, self, forwarded} <- getS + emit (Send peer PeerMessage { + source = self, + messageId = nextId, + payload = ForwardRequest key request + }) + putS ns { + nextId = next nextId, + forwarded = F . insert nextId (lift . respond) . unF $ forwarded + } + + +{- | + The monad in which the internals of the state machine run. This is really + just a conduit, but we wrap it because we only want to allow `yield`, which + we have re-named `emit`. +-} +newtype StateMT i o s m r = StateMT { + unStateMT :: + StateT + (NodeState i o s) + (ConduitM (LInput i o s) (LOutput i o s) m) + r + } deriving ( + Functor, Applicative, Monad, MonadLogger, MonadCatch, + MonadThrow, MonadIO + ) +instance MonadTrans (StateMT i o s) where + lift = StateMT . lift . lift + + +{- | + The state machine monad, in LIO. +-} +type StateM i o s r = StateMT i o s LIO r + + +{- | + Run the state machine monad, starting with the initial node state. +-} +runStateMT + :: NodeState i o s + -> StateMT i o s m () + -> ConduitM (LInput i o s) (LOutput i o s) m (NodeState i o s) +runStateMT ns = fmap snd . (`runStateT` ns) . unStateMT + + +{- | + Emit some output from the state machine. +-} +emit :: LOutput i o s -> StateM i o s () +emit = StateMT . lift . yield + + +{- | + Get the node State. +-} +getS :: StateMT i o s m (NodeState i o s) +getS = StateMT get + + +{- | + Put the node state. +-} +putS :: NodeState i o s -> StateMT i o s m () +putS = StateMT . put + + diff --git a/src/Network/Legion/UUID.hs b/src/Network/Legion/UUID.hs new file mode 100644 --- /dev/null +++ b/src/Network/Legion/UUID.hs @@ -0,0 +1,22 @@ +{- | + Contains common UUID functionality. +-} +module Network.Legion.UUID ( + getUUID, +) where + +import Control.Concurrent (threadDelay) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.UUID (UUID) +import Data.UUID.V1 (nextUUID) + + +{- | A utility function that makes a UUID, no matter what. -} +getUUID :: (MonadIO io) => io UUID + +getUUID = liftIO nextUUID >>= maybe (wait >> getUUID) return + where + wait = liftIO (threadDelay oneMillisecond) + oneMillisecond = 1000 + +