legion 0.8.0.1 → 0.8.0.2
raw patch · 7 files changed
+79/−23 lines, 7 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- legion.cabal +1/−1
- src/Network/Legion/Admin.hs +21/−3
- src/Network/Legion/Basics.hs +3/−3
- src/Network/Legion/Conduit.hs +20/−2
- src/Network/Legion/Index.hs +7/−1
- src/Network/Legion/Runtime.hs +26/−13
- src/Network/Legion/StateMachine.hs +1/−0
legion.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: legion-version: 0.8.0.1+version: 0.8.0.2 synopsis: Distributed, stateful, homogeneous microservice framework. description: Legion is a framework for writing distributed, homogeneous, stateful microservices in Haskell.
src/Network/Legion/Admin.hs view
@@ -18,6 +18,8 @@ import Control.Monad.Trans.Class (lift) import Data.Conduit (Source) import Data.Default.Class (def)+import Data.Map (Map)+import Data.Set (Set) import Data.Text.Encoding (encodeUtf8) import Data.Text.Lazy (Text) import Data.Version (showVersion)@@ -25,9 +27,10 @@ import Network.Legion.Application (LegionConstraints) import Network.Legion.Conduit (chanToSource) import Network.Legion.Distribution (Peer)+import Network.Legion.Index (IndexRecord) import Network.Legion.LIO (LIO) import Network.Legion.Lift (lift2)-import Network.Legion.PartitionKey (PartitionKey(K))+import Network.Legion.PartitionKey (PartitionKey(K), unKey) import Network.Legion.PartitionState (PartitionPowerState) import Network.Legion.StateMachine.Monad (NodeState) import Network.Wai (Middleware, modifyResponse)@@ -40,6 +43,7 @@ import Web.Scotty.Resource.Trans (resource, get, delete) import Web.Scotty.Trans (Options, scottyOptsT, settings, ScottyT, ActionT, param, middleware, status, json)+import qualified Data.Map as Map import qualified Data.Text as T {- |@@ -62,9 +66,17 @@ . logExceptionsAndContinue logging resource "/clusterstate" $+ get $ json =<< send chan GetState+ resource "/index" $+ get $ json =<< send chan GetIndex+ resource "/divergent" $ get $- json =<< send chan GetState- resource "/propstate/:key" $+ json . Map.mapKeys (show . toInteger . unKey) =<< send chan GetDivergent+ resource "/partitions" $+ get $+ json . Map.mapKeys (show . toInteger . unKey) =<< send chan GetStates+ + resource "/partitions/:key" $ get $ do key <- K . read <$> param "key" json =<< send chan (GetPart key)@@ -130,10 +142,16 @@ = GetState (NodeState e o s -> LIO ()) | GetPart PartitionKey (PartitionPowerState e o s -> LIO ()) | Eject Peer (() -> LIO ())+ | GetIndex (Set IndexRecord -> LIO ())+ | GetDivergent (Map PartitionKey (PartitionPowerState e o s) -> LIO ())+ | GetStates (Map PartitionKey (PartitionPowerState e o s) -> LIO ()) instance Show (AdminMessage e o s) where show (GetState _) = "(GetState _)" show (GetPart k _) = "(GetPart " ++ show k ++ " _)" show (Eject p _) = "(Eject " ++ show p ++ " _)"+ show (GetIndex _) = "(GetIndex _)"+ show (GetDivergent _) = "(GetDivergent _)"+ show (GetStates _) = "(GetStates _)"
src/Network/Legion/Basics.hs view
@@ -10,7 +10,7 @@ import Prelude hiding (lookup, readFile, writeFile) -import Control.Concurrent.STM (atomically, newTVar, modifyTVar, readTVar,+import Control.Concurrent.STM (atomically, newTVar, modifyTVar', readTVar, TVar) import Control.Monad.Trans.Class (lift) import Data.Binary (Binary, encode, decode)@@ -49,10 +49,10 @@ -> Maybe (PartitionPowerState e o s) -> IO () saveState_ cacheT key (Just state) =- (atomically . modifyTVar cacheT . insert key) state+ (atomically . modifyTVar' cacheT . insert key) state saveState_ cacheT key Nothing =- (atomically . modifyTVar cacheT . Map.delete) key+ (atomically . modifyTVar' cacheT . Map.delete) key fetchState :: TVar (Map PartitionKey (PartitionPowerState e o s))
src/Network/Legion/Conduit.hs view
@@ -10,7 +10,7 @@ import Control.Concurrent (forkIO) import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan)-import Control.Monad (void, forever)+import Control.Monad (void) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Conduit (Source, Sink, ($$), ($=), yield, awaitForever) import qualified Data.Conduit.List as CL (map)@@ -19,7 +19,25 @@ Convert a channel into a Source. -} chanToSource :: (MonadIO io) => Chan a -> Source io a-chanToSource chan = forever $ yield =<< liftIO (readChan chan)+chanToSource chan = do+ {-+ Don't use 'Control.Monad.forever' here. For some reason that is unclear to+ me, use of 'forever' creates a space leak, despite the comments in the+ 'forever' source code.+ + The code:++ > forever $ yield =<< liftIO (readChan chan)++ will reliably leak several megabytes of memory over the course of 10k+ messages when tested using the 'legion-discovery' project. This was+ discovered by @-hr@ heap profiling, which pointed to 'chanToSource'+ as the retainer. I think it didn't point to 'forever' as the retainer+ because 'forever' is inlined, and thus does not have a cost-centre+ associated with it.+ -}+ yield =<< liftIO (readChan chan)+ chanToSource chan {- |
src/Network/Legion/Index.hs view
@@ -8,12 +8,14 @@ SearchTag(..), ) where +import Data.Aeson (ToJSON, toJSON, object, (.=)) import Data.Binary (Binary) import Data.ByteString (ByteString) import Data.Set (Set) import Data.String (IsString)+import Data.Text.Encoding (decodeUtf8) import GHC.Generics (Generic)-import Network.Legion.PartitionKey (PartitionKey)+import Network.Legion.PartitionKey (PartitionKey, unKey) {- | This typeclass provides the ability to index partition states. -}@@ -41,6 +43,10 @@ } deriving (Eq, Ord, Show, Generic) instance Binary IndexRecord+instance ToJSON IndexRecord where+ toJSON (IndexRecord tag key) = object [+ (decodeUtf8 . unTag) tag .= toInteger (unKey key)+ ] {- | This data structure describes where in the index to start scrolling. -}
src/Network/Legion/Runtime.hs view
@@ -37,8 +37,8 @@ import Data.Text (pack) import GHC.Generics (Generic) import Network.Legion.Admin (runAdmin, AdminMessage(GetState, GetPart,- Eject))-import Network.Legion.Application (LegionConstraints, Persistence)+ Eject, GetIndex, GetDivergent, GetStates))+import Network.Legion.Application (LegionConstraints, Persistence, list) import Network.Legion.BSockAddr (BSockAddr(BSockAddr)) import Network.Legion.ClusterState (ClusterPowerState) import Network.Legion.Conduit (merge, chanToSink, chanToSource)@@ -113,17 +113,15 @@ peerS <- loggingC =<< startPeerListener settings adminS <- loggingC =<< runAdmin adminPort adminHost joinS <- loggingC (joinMsgSource settings)- loopChan <- lift newChan (self, nodeState, peers) <- makeNodeState settings startupMode- rts <- newRuntimeState self peers (writeChan loopChan)+ rts <- newRuntimeState self peers let messageSource = transPipe lift ( (joinS =$= CL.map J) `merge` (peerS =$= CL.map P) `merge` (requestSource =$= CL.map R) `merge`- (adminS =$= CL.map A) `merge`- chanToSource loopChan+ (adminS =$= CL.map A) ) void . runRTS persistence nodeState rts . runConduit $ messageSource@@ -132,9 +130,8 @@ newRuntimeState :: (Binary e, Binary o, Binary s) => Peer -> Map Peer BSockAddr- -> (RuntimeMessage e o s -> IO ()) -> LoggingT IO (RuntimeState e o s)- newRuntimeState self peers loop = do+ newRuntimeState self peers = do cm <- newConnectionManager peers firstMessageId <- newSequence return RuntimeState {@@ -142,8 +139,7 @@ nextId = firstMessageId, cm, self,- searches = Map.empty,- loop+ searches = Map.empty } {- |@@ -429,7 +425,26 @@ eject peer lift2 $ respond () +handleMessage {- Admin Get Index -}+ (A (GetIndex respond))+ =+ lift2 . respond =<< SMM.nsIndex <$> SMM.getNodeState +handleMessage {- Admin Get Divergent -}+ (A (GetDivergent respond))+ =+ lift2 . respond =<< SMM.partitions <$> SMM.getNodeState++handleMessage {- Admin Get States -}+ (A (GetStates respond))+ = do+ persistence <- SMM.getPersistence+ lift2 . respond . Map.fromList =<< runConduit (+ transPipe liftIO (list persistence)+ =$= CL.consume+ )++ {- | This defines the various ways a node can be spun up. -} data StartupMode = NewCluster@@ -756,9 +771,7 @@ cm :: ConnectionManager e o s, searches :: Map SearchTag- (Set Peer, Maybe IndexRecord, [Maybe IndexRecord -> LIO ()]),- loop :: RuntimeMessage e o s -> IO ()- {- ^ A way to send messages back into the message handler. -}+ (Set Peer, Maybe IndexRecord, [Maybe IndexRecord -> LIO ()]) }
src/Network/Legion/StateMachine.hs view
@@ -448,6 +448,7 @@ else Nothing )) modifyNodeState (\ns@NodeState {partitions, nsIndex} ->+ nsIndex `seq` ns { partitions = if Set.null (PS.divergent partition) then