packages feed

legion 0.2.0.0 → 0.3.0.0

raw patch · 17 files changed

+455/−107 lines, 17 files

Files

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