REMOVE-BEFORE and GETTIMESTAMP proxying

For clusters, the timestamps have to be translated, since each node can
have its own idea about what time it is. To translate a timestamp, the
proxy remembers what time it asked the node for a timestamp in
GETTIMESTAMP, and applies the delta as an offset in REMOVE-BEFORE.

This does mean that a remove from a cluster has to call GETTIMESTAMP on
every node before dropping from nodes. Not very efficient. Although
currently it tries to drop from every single node anyway, which is also
not very efficient.

I thought about caching the GETTIMESTAMP from the nodes on the first
call. That would improve efficiency. But, since monotonic clocks on
!Linux don't advance when the computer is suspended, consider what might
happen if one node was suspended for a while, then came back. Its
monotonic timestamp would end up behind where the proxying expects it to
be. Would that result in removing when it shouldn't, or refusing to
remove when it should? Have not thought it through. Either way, a
cluster behaving strangly for an extended period of time because one
of its nodes was briefly asleep doesn't seem like good behavior.

diff --git a/Annex/Cluster.hs b/Annex/Cluster.hs
index 3c0bc7e3a275930327be82e73ae2baa6470f7f73..7f20e1486f9c05f94a5002657287d5c6c47a0d84 100644 (file)
--- a/Annex/Cluster.hs
+++ b/Annex/Cluster.hs
@@ -67,7 +67,8 @@ proxyCluster clusteruuid proxydone servermode clientside protoerrhandler = do
                (selectnode, closenodes) <- clusterProxySelector clusteruuid
                        protocolversion bypassuuids
                concurrencyconfig <- getConcurrencyConfig
-               proxy proxydone proxymethods servermode clientside 
+               proxystate <- liftIO mkProxyState
+               proxy proxydone proxymethods proxystate servermode clientside 
                        (fromClusterUUID clusteruuid)
                        selectnode concurrencyconfig protocolversion
                        othermsg (protoerrhandler closenodes)
@@ -107,6 +108,7 @@ clusterProxySelector clusteruuid protocolversion (Bypass bypass) = do
                -- could be out of date, actually try to remove from every
                -- node.
                , proxyREMOVE = const (pure nodes)
+               , proxyGETTIMESTAMP = pure nodes
                -- Content is not locked on the cluster as a whole,
                -- instead it can be locked on individual nodes that are
                -- proxied to the client.

diff --git a/Command/P2PStdIO.hs b/Command/P2PStdIO.hs
index 910d9cb7cc78b7dc1adcd6c2ba219292f62911a5..5563758c8dc5e174fabea77a63eebae9df448273 100644 (file)
--- a/Command/P2PStdIO.hs
+++ b/Command/P2PStdIO.hs
@@ -76,7 +76,9 @@ performProxy clientuuid servermode r = do
                        closeRemoteSide remoteside
                        p2pDone
                let errhandler = p2pErrHandler (closeRemoteSide remoteside)
-               let runproxy othermsg' = proxy closer proxymethods
+               proxystate <- liftIO mkProxyState
+               let runproxy othermsg' = proxy closer
+                       proxymethods proxystate
                        servermode clientside
                        (Remote.uuid r)
                        (singleProxySelector remoteside)

diff --git a/P2P/Proxy.hs b/P2P/Proxy.hs
index 53cc4ff947b1437430adad4131ae23784d17e260..32ef7166fe5f27bc8761680f3a695720fea58f06 100644 (file)
--- a/P2P/Proxy.hs
+++ b/P2P/Proxy.hs
@@ -15,6 +15,7 @@ import qualified Annex
 import P2P.Protocol
 import P2P.IO
 import Utility.Metered
+import Utility.MonotonicClock
 import Git.FilePath
 import Types.Concurrency
 import Annex.Concurrent
@@ -26,16 +27,22 @@ import Control.Concurrent.Async
 import qualified Control.Concurrent.MSem as MSem
 import qualified Data.ByteString.Lazy as L
 import qualified Data.Set as S
+import qualified Data.Map as M
+import Data.Unique
 import GHC.Conc
 
 type ProtoCloser = Annex ()
 
 data ClientSide = ClientSide RunState P2PConnection
 
+newtype RemoteSideId = RemoteSideId Unique
+       deriving (Eq, Ord)
+
 data RemoteSide = RemoteSide
        { remote :: Remote
        , remoteConnect :: Annex (Maybe (RunState, P2PConnection, ProtoCloser))
        , remoteTMVar :: TMVar (RunState, P2PConnection, ProtoCloser)
+       , remoteSideId :: RemoteSideId
        }
 
 mkRemoteSide :: Remote -> Annex (Maybe (RunState, P2PConnection, ProtoCloser)) -> Annex RemoteSide
@@ -43,6 +50,7 @@ mkRemoteSide r remoteconnect = RemoteSide
        <$> pure r
        <*> pure remoteconnect
        <*> liftIO (atomically newEmptyTMVar)
+       <*> liftIO (RemoteSideId <$> newUnique)
 
 runRemoteSide :: RemoteSide -> Proto a -> Annex (Either ProtoFailure a)
 runRemoteSide remoteside a = 
@@ -71,6 +79,9 @@ data ProxySelector = ProxySelector
        , proxyUNLOCKCONTENT :: Annex (Maybe RemoteSide)
        , proxyREMOVE :: Key -> Annex [RemoteSide]
        -- ^ remove from all of these remotes
+       , proxyGETTIMESTAMP :: Annex [RemoteSide]
+       -- ^ should send every remote that proxyREMOVE can
+       -- ever return for any key
        , proxyGET :: Key -> Annex (Maybe RemoteSide)
        , proxyPUT :: AssociatedFile -> Key -> Annex [RemoteSide]
        -- ^ put to some/all of these remotes
@@ -82,6 +93,7 @@ singleProxySelector r = ProxySelector
        , proxyLOCKCONTENT = const (pure (Just r))
        , proxyUNLOCKCONTENT = pure (Just r)
        , proxyREMOVE = const (pure [r])
+       , proxyGETTIMESTAMP = pure [r]
        , proxyGET = const (pure (Just r))
        , proxyPUT = const (const (pure [r]))
        }
@@ -178,12 +190,23 @@ getClientBypass _ _ (Just othermsg) cont _ =
        -- Pass along non-BYPASS message from version 0 client.
        cont (Bypass S.empty, (Just othermsg))
 
+data ProxyState = ProxyState
+       { proxyRemoteLatestTimestamps :: TVar (M.Map RemoteSideId MonotonicTimestamp)
+       , proxyRemoteLatestLocalTimestamp :: TVar (Maybe MonotonicTimestamp)
+       }
+
+mkProxyState :: IO ProxyState
+mkProxyState = ProxyState
+       <$> newTVarIO mempty
+       <*> newTVarIO Nothing
+
 {- Proxy between the client and the remote. This picks up after
  - sendClientProtocolVersion.
  -}
 proxy 
        :: Annex r
        -> ProxyMethods
+       -> ProxyState
        -> ServerMode
        -> ClientSide
        -> UUID
@@ -197,7 +220,7 @@ proxy
        -- ^ non-VERSION message that was received from the client when
        -- negotiating protocol version, and has not been responded to yet
        -> ProtoErrorHandled r
-proxy proxydone proxymethods servermode (ClientSide clientrunst clientconn) remoteuuid proxyselector concurrencyconfig (ProtocolVersion protocolversion) othermsg protoerrhandler = do
+proxy proxydone proxymethods proxystate servermode (ClientSide clientrunst clientconn) remoteuuid proxyselector concurrencyconfig (ProtocolVersion protocolversion) othermsg protoerrhandler = do
        case othermsg of
                Nothing -> proxynextclientmessage ()
                Just message -> proxyclientmessage (Just message)
@@ -238,6 +261,13 @@ proxy proxydone proxymethods servermode (ClientSide clientrunst clientconn) remo
                        remotesides <- proxyREMOVE proxyselector k
                        servermodechecker checkREMOVEServerMode $
                                handleREMOVE remotesides k message
+               REMOVE_BEFORE _ k -> do
+                       remotesides <- proxyREMOVE proxyselector k
+                       servermodechecker checkREMOVEServerMode $
+                               handleREMOVE remotesides k message
+               GETTIMESTAMP -> do
+                       remotesides <- proxyGETTIMESTAMP proxyselector
+                       handleGETTIMESTAMP remotesides
                GET _ _ k -> proxyGET proxyselector k >>= \case
                        Just remoteside -> handleGET remoteside message
                        Nothing -> 
@@ -278,6 +308,7 @@ proxy proxydone proxymethods servermode (ClientSide clientrunst clientconn) remo
                PUT_FROM _ -> protoerr
                ALREADY_HAVE -> protoerr
                ALREADY_HAVE_PLUS _ -> protoerr
+               TIMESTAMP _ -> protoerr
                -- Early messages that the client should not send now.
                AUTH _ _ -> protoerr
                VERSION _ -> protoerr
@@ -318,15 +349,70 @@ proxy proxydone proxymethods servermode (ClientSide clientrunst clientconn) remo
                _ <- client $ net $ sendMessage (ERROR "protocol error X")
                giveup "protocol error M"
        
+       -- When there is a single remote, reply with its timestamp,
+       -- to avoid needing timestamp translation.
+       handleGETTIMESTAMP (remoteside:[]) = do
+               liftIO $ hPutStrLn stderr "!!!! single remote side"
+               liftIO $ atomically $ do
+                       writeTVar (proxyRemoteLatestTimestamps proxystate)
+                               mempty
+                       writeTVar (proxyRemoteLatestLocalTimestamp proxystate)
+                               Nothing
+               proxyresponse remoteside GETTIMESTAMP
+                       (const proxynextclientmessage)
+       -- When there are multiple remotes, reply with our local timestamp,
+       -- and do timestamp translation when sending REMOVE-FROM.
+       handleGETTIMESTAMP remotesides = do
+               -- Order of getting timestamps matters.
+               -- Getting the local time after the time of the remotes
+               -- means that if there is some delay in getting the time
+               -- from a remote, that is reflected in the local time,
+               -- and so reduces the allowed time.
+               remotetimes <- (M.fromList . mapMaybe join) <$> getremotetimes 
+               localtime <- liftIO currentMonotonicTimestamp
+               liftIO $ atomically $ do
+                       writeTVar (proxyRemoteLatestTimestamps proxystate)
+                               remotetimes
+                       writeTVar (proxyRemoteLatestLocalTimestamp proxystate)
+                               (Just localtime)
+               protoerrhandler proxynextclientmessage $
+                       client $ net $ sendMessage (TIMESTAMP localtime)
+         where
+               getremotetimes = forMC concurrencyconfig remotesides $ \r ->
+                       runRemoteSideOrSkipFailed r $ do
+                               net $ sendMessage GETTIMESTAMP
+                               net receiveMessage >>= return . \case
+                                       Just (TIMESTAMP ts) ->
+                                               Just (remoteSideId r, ts)
+                                       _ -> Nothing
+
+       proxyTimestamp ts _ _ Nothing = ts -- not proxying timestamps
+       proxyTimestamp ts r tsm (Just correspondinglocaltime) =
+               case M.lookup (remoteSideId r) tsm of
+                       Just oldts -> oldts + (ts - correspondinglocaltime)
+                       Nothing -> ts -- not reached
+
        handleREMOVE [] _ _ =
                -- When no places are provided to remove from,
                -- don't report a successful remote.
                protoerrhandler proxynextclientmessage $
                        client $ net $ sendMessage FAILURE      
        handleREMOVE remotesides k message = do
+               tsm <- liftIO $ readTVarIO $ 
+                       proxyRemoteLatestTimestamps proxystate
+               oldlocaltime <- liftIO $ readTVarIO $
+                       proxyRemoteLatestLocalTimestamp proxystate
                v <- forMC concurrencyconfig remotesides $ \r ->
                        runRemoteSideOrSkipFailed r $ do
-                               net $ sendMessage message
+                               case message of
+                                       REMOVE_BEFORE ts _ -> do 
+                                               v <- net getProtocolVersion
+                                               if v < ProtocolVersion 3
+                                                       then net $ sendMessage $
+                                                               REMOVE k
+                                                       else net $ sendMessage $
+                                                               REMOVE_BEFORE (proxyTimestamp ts r tsm oldlocaltime) k
+                                       _ -> net $ sendMessage message
                                net receiveMessage >>= return . \case
                                        Just SUCCESS ->
                                                Just ((True, Nothing), [Remote.uuid (remote r)])

diff --git a/doc/todo/P2P_locking_connection_drop_safety.mdwn b/doc/todo/P2P_locking_connection_drop_safety.mdwn
index 18ae8015a48d1d2fad412279b90e2a5ea8a21802..b5a40ca0f0f8cc987c3fa77b171149e5b58ed291 100644 (file)
--- a/doc/todo/P2P_locking_connection_drop_safety.mdwn
+++ b/doc/todo/P2P_locking_connection_drop_safety.mdwn
@@ -73,7 +73,7 @@ Then REMOVE Key Timestamp can have the timestamp adjusted when it's sent
 out to each client, by calling GETTIMESTAMP again and applying the offsets
 between the cluster's clock and each node's clock.
 
-TODO
+> done
 
 ## future flag day
 
@@ -94,3 +94,5 @@ those remotes. This problem is not likely enough to occur to seem worth
 that disruption.
 
 A flag day might be worth doing in a couple of years though. --[[Joey]]
+
+> [[done]] --[[Joey]]