From 414564705a9bb8fb2f1f81938e7f607b4010e78a Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Sun, 16 Aug 2020 23:26:18 +0200 Subject: [PATCH] possibility to wait for a migration to complete --- src/Hash2Pub/DHTProtocol.hs | 2 +- src/Hash2Pub/FediChord.hs | 2 +- src/Hash2Pub/FediChordTypes.hs | 3 ++ src/Hash2Pub/PostService.hs | 50 ++++++++++++++++++++++++---------- 4 files changed, 41 insertions(+), 16 deletions(-) diff --git a/src/Hash2Pub/DHTProtocol.hs b/src/Hash2Pub/DHTProtocol.hs index f5fcdbd..52ea5ba 100644 --- a/src/Hash2Pub/DHTProtocol.hs +++ b/src/Hash2Pub/DHTProtocol.hs @@ -460,7 +460,7 @@ respondJoin nsSTM msgSet = do writeTVar nsSTM joinedNS ownService <- nodeService <$> readTVar (parentRealNode nsSnap) let - serviceDataMigrator = migrateData ownService lowerKeyBound (getNid senderNS) (getDomain senderNS, fromIntegral $ getServicePort senderNS) + serviceDataMigrator = migrateData ownService (getNid nsSnap) lowerKeyBound (getNid senderNS) (getDomain senderNS, fromIntegral $ getServicePort senderNS) lowerKeyBound = maybe (getNid nsSnap) getNid $ headMay (predecessors nsSnap) pure (Just serviceDataMigrator, joinResponse) -- otherwise respond with empty payload diff --git a/src/Hash2Pub/FediChord.hs b/src/Hash2Pub/FediChord.hs index f3a482c..8d25186 100644 --- a/src/Hash2Pub/FediChord.hs +++ b/src/Hash2Pub/FediChord.hs @@ -297,7 +297,7 @@ fediChordVserverLeave ns = do ownService <- nodeService <$> (liftIO . readTVarIO $ parentRealNode ns) -- previously held data is the one between the immediate predecessor and -- the own ID - migrationResult <- liftIO $ migrateData ownService lowerKeyBound (getNid ns) (getDomain migrateToNode, fromIntegral $ getServicePort migrateToNode) + migrationResult <- liftIO $ migrateData ownService (getNid ns) lowerKeyBound (getNid ns) (getDomain migrateToNode, fromIntegral $ getServicePort migrateToNode) liftEither migrationResult diff --git a/src/Hash2Pub/FediChordTypes.hs b/src/Hash2Pub/FediChordTypes.hs index 214ece2..cbd3a58 100644 --- a/src/Hash2Pub/FediChordTypes.hs +++ b/src/Hash2Pub/FediChordTypes.hs @@ -426,10 +426,13 @@ class Service s d where getListeningPortFromService :: (Integral i) => s d -> i -- | trigger a service data migration of data between the two given keys migrateData :: s d + -> NodeID -- ^ source/ sender node ID -> NodeID -- ^ start key -> NodeID -- ^ end key -> (String, Int) -- ^ hostname and port of target service -> IO (Either String ()) -- ^ success or failure + -- | Wait for an incoming migration from a given node to succeed, may block forever + waitForMigrationFrom :: s d -> NodeID -> IO () instance Hashable.Hashable NodeID where hashWithSalt salt = Hashable.hashWithSalt salt . getNodeID diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 71998df..548469e 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -10,6 +10,7 @@ module Hash2Pub.PostService where import Control.Concurrent import Control.Concurrent.Async +import Control.Concurrent.MVar import Control.Concurrent.STM import Control.Concurrent.STM.TChan import Control.Concurrent.STM.TChan @@ -46,20 +47,21 @@ import Hash2Pub.RingMap data PostService d = PostService - { serviceConf :: ServiceConf + { serviceConf :: ServiceConf -- queues, other data structures - , baseDHT :: (DHT d) => d - , serviceThread :: TVar ThreadId - , subscribers :: TVar RelayTags + , baseDHT :: (DHT d) => d + , serviceThread :: TVar ThreadId + , subscribers :: TVar RelayTags -- ^ for each tag store the subscribers + their queue - , ownSubscriptions :: TVar (HMap.HashMap NodeID POSIXTime) + , ownSubscriptions :: TVar (HMap.HashMap NodeID POSIXTime) -- ^ tags subscribed by the own node have an assigned lease time - , ownPosts :: TVar (HSet.HashSet Txt.Text) + , ownPosts :: TVar (HSet.HashSet Txt.Text) -- ^ just store the existence of posts for saving memory, - , relayInQueue :: TQueue (Hashtag, PostID, PostContent) + , relayInQueue :: TQueue (Hashtag, PostID, PostContent) -- ^ Queue for processing incoming posts of own instance asynchronously - , postFetchQueue :: TQueue PostID - , httpMan :: HTTP.Manager + , postFetchQueue :: TQueue PostID + , migrationsInProgress :: TVar (HMap.HashMap NodeID (MVar ())) + , httpMan :: HTTP.Manager } deriving (Typeable) @@ -86,6 +88,7 @@ instance DHT d => Service PostService d where ownPostVar <- newTVarIO HSet.empty relayInQueue' <- newTQueueIO postFetchQueue' <- newTQueueIO + migrationsInProgress' <- newTVarIO HMap.empty httpMan' <- HTTP.newManager HTTP.defaultManagerSettings let thisService = PostService { @@ -97,6 +100,7 @@ instance DHT d => Service PostService d where , ownPosts = ownPostVar , relayInQueue = relayInQueue' , postFetchQueue = postFetchQueue' + , migrationsInProgress = migrationsInProgress' , httpMan = httpMan' } port' = fromIntegral (confServicePort conf) @@ -117,6 +121,17 @@ instance DHT d => Service PostService d where migrateData = clientDeliverSubscriptions + waitForMigrationFrom serv fromID = do + migrationSynchroniser <- atomically $ do + syncPoint <- HMap.lookup fromID <$> readTVar (migrationsInProgress serv) + maybe + -- decision: this function blocks until it gets an incoming migration from given ID + retry + pure + syncPoint + -- block until migration finished + takeMVar migrationSynchroniser + -- | return a WAI application postServiceApplication :: DHT d => PostService d -> Application @@ -136,7 +151,7 @@ placeholderPost = Txt.take 5120 . Txt.repeat $ 'O' -- size 5KiB type PostServiceAPI = "relay" :> "inbox" :> Capture "hashtag" Txt.Text :> ReqBody '[PlainText] Txt.Text :> PutCreated '[PlainText] NoContent -- delivery endpoint of newly published posts of the relay's instance - :<|> "relay" :> "subscribers" :> ReqBody '[PlainText] Txt.Text :> PostNoContent '[PlainText] Txt.Text + :<|> "relay" :> "subscribers" :> Capture "senderID" Integer :> ReqBody '[PlainText] Txt.Text :> PostNoContent '[PlainText] Txt.Text -- endpoint for delivering the subscriptions and outstanding queue :<|> "post" :> Capture "postid" Txt.Text :> Get '[PlainText] Txt.Text -- fetch endpoint for posts, full post ID is http://$domain/post/$postid @@ -194,10 +209,14 @@ newtype UnhandledTagException = UnhandledTagException String instance Exception UnhandledTagException -subscriptionDelivery :: DHT d => PostService d -> Txt.Text -> Handler Txt.Text -subscriptionDelivery serv subList = do +subscriptionDelivery :: DHT d => PostService d -> Integer -> Txt.Text -> Handler Txt.Text +subscriptionDelivery serv senderID subList = do let tagSubs = Txt.lines subList + -- signal that the migration is in progress + syncMVar <- liftIO newEmptyMVar + liftIO . atomically $ modifyTVar' (migrationsInProgress serv) $ + HMap.insert (fromInteger senderID) syncMVar -- In favor of having the convenience of rolling back the transaction once a -- not-handled tag occurs, this results in a single large transaction. -- Hopefully the performance isn't too bad. @@ -211,6 +230,8 @@ subscriptionDelivery serv subList = do `catchSTM` (\e -> pure . Left $ show (e :: UnhandledTagException)) -- TODO: potentially log this :: STM (Either String ())) + -- TODO: should this always signal migration finished to avoid deadlocksP + liftIO $ putMVar syncMVar () case res of Left err -> throwError err410 {errBody = BSUL.fromString err} Right _ -> pure "" @@ -322,11 +343,12 @@ relayInboxClient :<|> subscriptionDeliveryClient :<|> postFetchClient :<|> postM -- and their outstanding delivery queue to another instance. -- If the transfer succeeds, the transfered subscribers are removed from the local list. clientDeliverSubscriptions :: PostService d + -> NodeID -- ^ sender node ID -> NodeID -- ^ fromTag -> NodeID -- ^ toTag -> (String, Int) -- ^ hostname and port of instance to deliver to -> IO (Either String ()) -- Either signals success or failure -clientDeliverSubscriptions serv fromKey toKey (toHost, toPort) = do +clientDeliverSubscriptions serv fromNode fromKey toKey (toHost, toPort) = do -- collect tag intearval intervalTags <- takeRMapSuccessorsFromTo fromKey toKey <$> readTVarIO (subscribers serv) -- returns a [ (TagSubscribersSTM, TChan PostID, Hashtag) ] @@ -350,7 +372,7 @@ clientDeliverSubscriptions serv fromKey toKey (toHost, toPort) = do "" intervalTags -- send subscribers - resp <- runClientM (subscriptionDeliveryClient subscriberData) (mkClientEnv (httpMan serv) (BaseUrl Http toHost (fromIntegral toPort) "")) + resp <- runClientM (subscriptionDeliveryClient (getNodeID fromNode) subscriberData) (mkClientEnv (httpMan serv) (BaseUrl Http toHost (fromIntegral toPort) "")) -- on failure return a Left, otherwise delete subscription entry case resp of Left err -> pure . Left . show $ err