legion (empty) → 0.1.0.0

raw patch · 23 files changed

+3812/−0 lines, 23 filesdep +Ranged-setsdep +attoparsecdep +basesetup-changed

Dependencies added: Ranged-sets, attoparsec, base, binary, binary-conduit, bytestring, conduit, conduit-extra, containers, data-default-class, data-dword, directory, exceptions, monad-logger, network, scotty, scotty-resource, stm, text, time, transformers, unix, uuid, warp

Files

• LICENSE +202/−0
• Setup.hs +2/−0
• legion.cabal +76/−0
• src/Network/Legion.hs +173/−0
• src/Network/Legion/Admin.hs +76/−0
• src/Network/Legion/Application.hs +85/−0
• src/Network/Legion/BSockAddr.hs +53/−0
• src/Network/Legion/Basics.hs +124/−0
• src/Network/Legion/ClusterState.hs +225/−0
• src/Network/Legion/Conduit.hs +53/−0
• src/Network/Legion/ConnectionManager.hs +209/−0
• src/Network/Legion/Distribution.hs +189/−0
• src/Network/Legion/Fork.hs +49/−0
• src/Network/Legion/KeySet.hs +205/−0
• src/Network/Legion/LIO.hs +17/−0
• src/Network/Legion/PartitionKey.hs +95/−0
• src/Network/Legion/PartitionState.hs +182/−0
• src/Network/Legion/PowerState.hs +392/−0
• src/Network/Legion/Propagation.hs +348/−0
• src/Network/Legion/Runtime.hs +358/−0
• src/Network/Legion/Settings.hs +33/−0
• src/Network/Legion/StateMachine.hs +644/−0
• src/Network/Legion/UUID.hs +22/−0

@@ -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.

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -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+

@@ -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.+

@@ -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+  }++

@@ -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 ())++

@@ -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."+++

@@ -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++

@@ -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++

@@ -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++

@@ -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++

@@ -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++

@@ -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+

@@ -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)))++

@@ -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++

@@ -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)++

@@ -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++

@@ -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++

@@ -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++

@@ -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+      )++

@@ -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.+      -}+  }++

@@ -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++

@@ -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++