packages feed

legion 0.8.0.2 → 0.8.0.3

raw patch · 7 files changed

+136/−69 lines, 7 files

Files

legion.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/  name:                legion-version:             0.8.0.2+version:             0.8.0.3 synopsis:            Distributed, stateful, homogeneous microservice framework. description:         Legion is a framework for writing distributed,                      homogeneous, stateful microservices in Haskell.
src/Network/Legion/ClusterState.hs view
@@ -47,6 +47,7 @@ import qualified Data.Set as Set import qualified Data.Text as T import qualified Network.Legion.Distribution as D+import qualified Network.Legion.KeySet as KS import qualified Network.Legion.PowerState as PS import qualified Network.Legion.PowerState.Monad as PM @@ -71,12 +72,14 @@       rebalanceOrd = minBound     } instance ToJSON ClusterState where-  toJSON ClusterState {distribution, peers} = object [-      "distribution" .= distribution,+  toJSON (ClusterState distribution_ peers_ updates_ rebalanceOrd_) = object [+      "distribution" .= distribution_,       "peers" .= Map.fromList [           (show p, show a)-          | (p, a) <- Map.toList peers-        ]+          | (p, a) <- Map.toList peers_+        ],+      "updates" .= (show <$> updates_),+      "rebalanceOrd" .= show rebalanceOrd_     ] instance Show ClusterState where   show = T.unpack . decodeUtf8 . LBS.toStrict . encode@@ -124,12 +127,17 @@ popUpdate :: ClusterState -> ClusterState popUpdate cs@ClusterState {updates, distribution, peers} =   case (updates, rebalanceAction (Map.keysSet peers) distribution) of-    (u:moreUpdates, (NoAction, _)) -> popUpdate cs {-        peers = case u of-          PeerJoined peer addr -> Map.insert peer addr peers-          PeerEjected peer -> Map.delete peer peers,-        updates = moreUpdates-      }+    (u:moreUpdates, (NoAction, _)) -> popUpdate $+      case u of+        PeerJoined peer addr -> cs {+            peers = Map.insert peer addr peers,+            updates = moreUpdates+          }+        PeerEjected peer -> cs {+            distribution = D.modify (Set.delete peer) KS.full distribution,+            peers = Map.delete peer peers,+            updates = moreUpdates+          }     _ -> cs
src/Network/Legion/Distribution.hs view
@@ -138,7 +138,6 @@       in         unD (modify (Set.\\ defunct) KS.full distribution) -     {- |       Figure out if there are any under-served partitions and also figure       out if this peer is the best candidate to service them . "Under
src/Network/Legion/PowerState/Monad.hs view
@@ -15,7 +15,10 @@   event,   merge,   acknowledge,+  acknowledgeAs, +  getPowerState,+   participate,   disassociate, ) where@@ -127,18 +130,30 @@ -} acknowledge :: (Monad m, Ord p, Event e r s, Eq e, Eq o)   => PowerStateT o s p e r m ()-acknowledge = PowerStateT $ do+acknowledge = PowerStateT (lift2 ask) >>= acknowledgeAs+++{- | Like 'acknowledge', but for an arbigrary participant. -}+acknowledgeAs :: (Monad m, Ord p, Event e r s, Eq e, Eq o)+  => p+  -> PowerStateT o s p e r m ()+acknowledgeAs p = PowerStateT $ do   ps <- get   prop <- lift get-  self <- lift2 ask   let-    (ps2, outputs) = PS.acknowledge self ps+    (ps2, outputs) = PS.acknowledge p ps     prop2 = if ps2 /= ps       then Send       else prop   put ps2   (lift . put) prop2   (lift3 . tell) outputs+++{- | Return the current value of the power state. -}+getPowerState :: (Monad m, Ord p)+  => PowerStateT o s p e r m (PowerState o s p e r)+getPowerState = PowerStateT get   {- |
src/Network/Legion/Runtime.hs view
@@ -399,29 +399,6 @@ handleMessage {- Admin Eject Peer -}     (A (Eject peer respond))   = do-    {--      TODO: we should attempt to notify the ejected peer that it has-      been ejected instead of just cutting it off and washing our hands-      of it. I have a vague notion that maybe ejected peers should be-      permanently recorded in the cluster state so that if they ever-      reconnect then we can notify them that they are no longer welcome-      to participate.--      On a related note, we need to think very hard about the split brain-      problem. A random thought about that is that we should consider the-      extreme case where the network just fails completely and every node-      believes that every other node should be or has been ejected. This-      would obviously be catastrophic in terms of data durability unless-      we have some way to reintegrate an ejected node. So, either we-      have to guarantee that such a situation can never happen, or else-      implement a reintegration strategy.  It might be acceptable for-      the reintegration strategy to be very costly if it is characterized-      as an extreme recovery scenario.--      Question: would a reintegration strategy become less costly if the-      "next state id" for a peer were global across all power states-      instead of local to each power state?-    -}     eject peer     lift2 $ respond ()
src/Network/Legion/Runtime/ConnectionManager.hs view
@@ -15,7 +15,7 @@ import Prelude hiding (lookup)  import Control.Concurrent (Chan, writeChan, newChan, readChan)-import Control.Exception (try, SomeException)+import Control.Exception (try, SomeException, bracketOnError) import Control.Monad (void) import Control.Monad.Logger (logInfo, logWarn) import Control.Monad.Trans.Class (lift)@@ -102,14 +102,20 @@     handle chan so =       lift (readChan chan) >>= sendWithRetry so . encode >>= handle chan -    {- |-      Open a socket.-    -}+    {- | Open a socket. -}     openSocket :: IO Socket-    openSocket = do-      so <- socket (fam addr) Stream defaultProtocol-      connect so addr-      return so+    openSocket =+      {-+        Make sure to close the socket if an error happens during+        connection, because if not, we could easily run out of file+        descriptors in the case where we rapidly try to send thousands+        of message to the same peer, which could happen when one object+        is a hotspot.+      -}+      bracketOnError+        (socket (fam addr) Stream defaultProtocol)+        close+        (\so -> connect so addr >> return so)      {- |       Try to send the payload over the socket, and if that fails, then try to@@ -122,7 +128,7 @@         Left err -> do           $(logWarn) . pack             $ "Can't connect to: " ++ show addr ++ ". Dropping message on "-            ++ "the floor: " ++ show payload ++ ". The error was: "+            ++ "the floor. The error was: "             ++ show (err :: SomeException)           return Nothing         Right so -> do@@ -132,8 +138,7 @@               $ "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+              ++ "are not retrying. The message is being dropped on the floor."             Right _ -> return ()           return (Just so)     sendWithRetry (Just so) payload =
src/Network/Legion/StateMachine.hs view
@@ -226,9 +226,40 @@  {- | Eject a peer from the cluster.  -} eject :: (MonadLogger m, MonadThrow m) => Peer -> SM e o s m ()-eject peer = runClusterPowerStateT (C.eject peer)+eject peer = do+  {-+    We need to think very hard about the split brain problem. A random+    thought about that is that we should consider the extreme case where+    the network just fails completely and every node believes that every+    other node should be or has been ejected. This would obviously be+    catastrophic in terms of data durability unless we have some way to+    reintegrate an ejected node. So, either we have to guarantee that+    such a situation can never happen, or else implement a reintegration+    strategy.  It might be acceptable for the reintegration strategy to+    be very costly if it is characterized as an extreme recovery scenario. +    Question: would a reintegration strategy become less costly if the+    "next state id" for a peer were global across all power states+    instead of local to each power state?+  -}+  runClusterPowerStateT (C.eject peer)+  {-+    'runClusterPowerStateT (C.eject peer)' will cause us to attempt to+    notify the peer that they have been ejected, but that notification+    is almost certainly going to go unacknowledged because the peer+    is probably down.+    +    This call to 'eject' was presumably invoked as a result of user+    action, and we must therefore trust the user to know that the peer+    is really down and not coming back. This "guarantee" allows us to+    acknowledge the ejection on the peer's behalf. +    This call will acknowledge the drop on behalf of the peer, and also+    remove that peer from the keyspace distribution map.+  -}+  runClusterPowerStateTAs peer (return ())++ {- | Handle a peer join request.  -} join :: (MonadIO m, MonadThrow m)   => BSockAddr@@ -492,40 +523,72 @@   -> PartitionPowerStateT e o s (SM e o s m) a   -> SM e o s m (a, PartitionPowerState e o s) runPartitionPowerStateT key m = do+    NodeState {self} <- getNodeState+    partition <- getPartition key+    PM.runPowerStateT self partition (+        m <* (removeObsolete >> PM.acknowledge)+      ) >>= \case+        Left err -> throwM err+        Right (a, action, partition2, _infOutputs) -> do+          case action of+            Send -> pushActions [+                PartitionMerge p key partition2+                | p <- Set.toList (PS.allParticipants partition2)+                , p /= self+              ]+            DoNothing -> return ()+          $(logDebug) . pack+            $ "Partition update: " ++ show partition+            ++ " --> " ++ show partition2 ++ " :: " ++ show action+          savePartition key partition2+          return (a, partition2)+  where+    {- |+      Remove obsolete peers. Obsolete peers are peers that are no longer+      participating in the replication of this partition, due to a+      rebalance. Such peers are removed lazily here at read time.+    -}+    removeObsolete :: (Eq e, Event e o s, Monad m)+      => PartitionPowerStateT e o s (SM e o s m) ()+    removeObsolete = do+      owners <- C.findOwners key . cluster <$> lift getNodeState+      peers <- PS.projParticipants <$> PM.getPowerState+      let obsolete = peers \\ owners+      mapM_+        (\peer -> PM.disassociate peer >> PM.acknowledgeAs peer)+        (Set.toList obsolete)+++{- | Like 'runClusterPowerStateTAs', but run as the local peer. -}+runClusterPowerStateT :: (MonadThrow m)+  => ClusterPowerStateT (SM e o s m) a+  -> SM e o s m a+runClusterPowerStateT m = do   NodeState {self} <- getNodeState-  partition <- getPartition key-  PM.runPowerStateT self partition (m <* PM.acknowledge) >>= \case-    Left err -> throwM err-    Right (a, action, partition2, _infOutputs) -> do-      case action of-        Send -> pushActions [-            PartitionMerge p key partition2-            | p <- Set.toList (PS.allParticipants partition2)-          ]-        DoNothing -> return ()-      $(logDebug) . pack-        $ "Partition update: " ++ show partition-        ++ " --> " ++ show partition2 ++ " :: " ++ show action-      savePartition key partition2-      return (a, partition2)+  runClusterPowerStateTAs self m   {- |   Run a clusterstate-flavored 'PowerStateT' in the 'SM' monad,   automatically acknowledging the resulting power state.++  Generalized to run as any peer, in order to support exceptional cases+  like 'eject'. -}-runClusterPowerStateT :: (MonadThrow m)-  => ClusterPowerStateT (SM e o s m) a+runClusterPowerStateTAs :: (MonadThrow m)+  => Peer {- ^ The peer to run as. -}+  -> ClusterPowerStateT (SM e o s m) a   -> SM e o s m a-runClusterPowerStateT m = do+runClusterPowerStateTAs as m = do   NodeState {cluster, self} <- getNodeState-  PM.runPowerStateT self cluster (m <* PM.acknowledge) >>= \case+  PM.runPowerStateT as cluster (m <* PM.acknowledge) >>= \case     Left err -> throwM err     Right (a, action, cluster2, _outputs) -> do       case action of         Send -> pushActions [             ClusterMerge p cluster2             | p <- Set.toList (PS.allParticipants cluster2)+            , p /= self           ]         DoNothing -> return ()       modifyNodeState (\ns -> ns {cluster = cluster2})