From 1e0f92a5a1ccf7ff4c51c67c27a826709a99301b Mon Sep 17 00:00:00 2001 From: Joey Hess Date: Wed, 10 Jul 2024 16:06:39 -0400 Subject: [PATCH] implemented serveGet and clientGet Both are only at bare proof of concept stage. Still need to deal with signaling validity and invalidity, and checking it. And there's a bad bug: After -JN*2 requests, another request hangs! So, I think it's failing to free up the Annex worker and end of request lifetime. Perhaps I need to use this: https://docs.servant.dev/en/stable/cookbook/managed-resource/ManagedResource.html --- Command/P2PHttp.hs | 16 +++++- P2P/Annex.hs | 20 ++++++++ P2P/Http.hs | 124 ++++++++++++++++++++++++++++++++++++++------- P2P/Http/State.hs | 44 +++++++++++----- P2P/Protocol.hs | 4 ++ 5 files changed, 177 insertions(+), 31 deletions(-) diff --git a/Command/P2PHttp.hs b/Command/P2PHttp.hs index 4b141adbe1..c7b0d05e7a 100644 --- a/Command/P2PHttp.hs +++ b/Command/P2PHttp.hs @@ -73,7 +73,7 @@ seek o = getAnnexWorkerPool $ \workerpool -> do -- XXX remove this when (isNothing (portOption o)) $ do liftIO $ putStrLn "test begins" - testGetTimestamp + testGet giveup "TEST DONE" withLocalP2PConnections $ \acquireconn -> liftIO $ do authenv <- getAuthEnv @@ -158,6 +158,20 @@ testCheckPresent = do Nothing liftIO $ print res +testGet = do + mgr <- httpManager <$> getUrlOptions + burl <- liftIO $ parseBaseUrl "http://localhost:8080/" + res <- liftIO $ clientGet (mkClientEnv mgr burl) + (P2P.ProtocolVersion 3) + (B64Key (fromJust $ deserializeKey ("WORM-s3218-m1720641607--passwd" :: String))) + (B64UUID (toUUID ("cu" :: String))) + (B64UUID (toUUID ("f11773f0-11e1-45b2-9805-06db16768efe" :: String))) + [] + Nothing + Nothing + Nothing + liftIO $ print res + testRemove = do mgr <- httpManager <$> getUrlOptions burl <- liftIO $ parseBaseUrl "http://localhost:8080/" diff --git a/P2P/Annex.hs b/P2P/Annex.hs index 8d7348e36f..59428db8c9 100644 --- a/P2P/Annex.hs +++ b/P2P/Annex.hs @@ -101,6 +101,26 @@ runLocal runst runner a = case a of Left e -> return $ Left $ ProtoFailureException e Right (Left e) -> return $ Left e Right (Right ok) -> runner (next ok) + SendContentWith consumer getb validitycheck next -> do + v <- tryNonAsync $ do + let fallback = return $ Left $ + ProtoFailureMessage "Transfer failed" + let consumer' b ti = do + validator <- consumer b + indicatetransferred ti + return validator + runner getb >>= \case + Left e -> giveup $ describeProtoFailure e + Right b -> checktransfer (\ti -> Right <$> consumer' b ti) fallback >>= \case + Left e -> return (Left e) + Right validator -> + runner validitycheck >>= \case + Right v -> Right <$> validator v + _ -> Right <$> validator Nothing + case v of + Left e -> return $ Left $ ProtoFailureException e + Right (Left e) -> return $ Left e + Right (Right ok) -> runner (next ok) SetPresent k u next -> do v <- tryNonAsync $ logChange k u InfoPresent case v of diff --git a/P2P/Http.hs b/P2P/Http.hs index 4fa0dab043..19d43332ae 100644 --- a/P2P/Http.hs +++ b/P2P/Http.hs @@ -10,6 +10,7 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE TypeOperators #-} {-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE TypeApplications #-} {-# LANGUAGE OverloadedStrings #-} module P2P.Http ( @@ -23,6 +24,7 @@ import P2P.Http.Types import P2P.Http.State import P2P.Protocol hiding (Offset, Bypass, auth) import P2P.IO +import P2P.Annex import Annex.WorkerPool import Types.WorkerPool import Types.Direction @@ -30,9 +32,15 @@ import Utility.Metered import Servant import Servant.Client.Streaming +import Servant.API import qualified Servant.Types.SourceT as S import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as L +import qualified Data.ByteString.Lazy.Internal as LI import Control.Concurrent.STM +import Control.Concurrent.Async +import Control.Concurrent +import System.IO.Unsafe type P2PHttpAPI = "git-annex" :> PV3 :> "key" :> CaptureKey :> GetAPI @@ -142,23 +150,66 @@ serveGet -> Maybe Auth -> Handler (Headers '[DataLengthHeader] (S.SourceT IO B.ByteString)) serveGet st apiver (B64Key k) cu su bypass baf startat sec auth = do - res <- withP2PConnection apiver st cu su bypass sec auth ReadAction - $ \runst conn -> do - liftIO $ inAnnexWorker st $ - enteringStage (TransferStage Upload) $ do - liftIO $ print "IN ANNEX WORKER!" - {- - let storer offset len getdata checkvalidity = do - undefined -- FIXME - -- XXX needs to run in annex monad to runFullProto - liftIO $ runNetProto runst conn $ - receiveContent Nothing nullMeterUpdate + (runst, conn, releaseconn) <- + getP2PConnection apiver st cu su bypass sec auth ReadAction + bsv <- liftIO newEmptyTMVarIO + endv <- liftIO newEmptyTMVarIO + validityv <- liftIO newEmptyTMVarIO + aid <- liftIO $ async $ inAnnexWorker st $ do + let consumer bs = do + liftIO $ atomically $ putTMVar bsv bs + liftIO $ atomically $ takeTMVar endv + return $ \v -> do + liftIO $ atomically $ + putTMVar validityv v + return True + let storer _offset _len getdata checkvalidity = + sendContentWith consumer getdata checkvalidity + enteringStage (TransferStage Upload) $ + runFullProto runst conn $ + void $ receiveContent Nothing nullMeterUpdate sizer storer getreq - -} - undefined - undefined -- XXX fixme streaming out + bs <- liftIO $ atomically $ takeTMVar bsv + bv <- liftIO $ newMVar (L.toChunks bs) + let streamer = S.SourceT $ \s -> s =<< return + (stream (releaseconn, bv, endv, validityv, aid)) + return $ addHeader 111111 streamer where - sizer = Len $ case startat of + stream (releaseconn, bv, endv, validityv, aid) = + S.fromActionStep B.null $ do + print "chunk" + modifyMVar bv $ \case + (b:bs) -> return (bs, b) + [] -> do + endbit <- cleanup (releaseconn, endv, validityv, aid) + return ([], endbit) + + cleanup (releaseconn, endv, validityv, aid) = + ifM (atomically $ isEmptyTMVar endv) + ( pure mempty + , do + atomically $ putTMVar endv () + validity <- atomically $ takeTMVar validityv + print ("got validity", validity) + wait aid >>= \case + Left ex -> throwM ex + Right (Left err) -> error $ + describeProtoFailure err + Right (Right ()) -> return () + () <- releaseconn + -- When the key's content is invalid, + -- indicate that to the client by padding + -- the response, so it is not the same + -- length indicated by the DataLengthHeader. + return $ case validity of + Nothing -> mempty + Just Valid -> mempty + Just Invalid -> "XXXXXXX" + -- FIXME: need to count bytes and emit + -- something to make it invalid + ) + + sizer = pure $ Len $ case startat of Just (Offset o) -> fromIntegral o Nothing -> 0 @@ -169,6 +220,43 @@ serveGet st apiver (B64Key k) cu su bypass baf startat sec auth = do Nothing -> Nothing clientGet + :: ClientEnv + -> ProtocolVersion + -> B64Key + -> B64UUID ClientSide + -> B64UUID ServerSide + -> [B64UUID Bypass] + -> Maybe B64FilePath + -> Maybe Offset + -> Maybe Auth + -> IO () +clientGet clientenv ver k cu su bypass af o auth = + withClientM (clientGet' ver k cu su bypass af o auth) clientenv $ \case + Left err -> throwM err + Right respheaders -> do + let dl = case lookupResponseHeader @DataLengthHeader' respheaders of + Header h -> h + _ -> error "missing data length header" + liftIO $ print ("datalength", dl :: Integer) + b <- S.unSourceT (getResponse respheaders) gatherbytestring + liftIO $ print "got it all, writing to file 'got'" + L.writeFile "got" b + +gatherbytestring :: S.StepT IO B.ByteString -> IO L.ByteString +gatherbytestring x = do + l <- unsafeInterleaveIO $ go x + return l + where + go S.Stop = return LI.Empty + go (S.Error err) = error $ show ("ERROR", err) + go (S.Skip s) = do + go s + go (S.Effect ms) = do + ms >>= go + go (S.Yield v s) = do + LI.Chunk v <$> unsafeInterleaveIO (go s) + +clientGet' :: ProtocolVersion -> B64Key -> B64UUID ClientSide @@ -178,7 +266,7 @@ clientGet -> Maybe Offset -> Maybe Auth -> ClientM (Headers '[DataLengthHeader] (S.SourceT IO B.ByteString)) -clientGet (ProtocolVersion ver) = case ver of +clientGet' (ProtocolVersion ver) = case ver of 3 -> v3 V3 2 -> v2 V2 1 -> v1 V1 @@ -647,7 +735,9 @@ type AssociatedFileParam = QueryParam "associatedfile" B64FilePath type OffsetParam = QueryParam "offset" Offset -type DataLengthHeader = Header "X-git-annex-data-length" Integer +type DataLengthHeader = Header DataLengthHeader' Integer + +type DataLengthHeader' = "X-git-annex-data-length" type LockIDParam = QueryParam' '[Required] "lockid" LockID diff --git a/P2P/Http/State.hs b/P2P/Http/State.hs index 7c42b2ea93..d3e00dfa21 100644 --- a/P2P/Http/State.hs +++ b/P2P/Http/State.hs @@ -9,6 +9,7 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE FlexibleContexts #-} module P2P.Http.State where @@ -63,7 +64,29 @@ withP2PConnection -> ActionClass -> (RunState -> P2PConnection -> Handler (Either ProtoFailure a)) -> Handler a -withP2PConnection apiver st cu su bypass sec auth actionclass connaction = +withP2PConnection apiver st cu su bypass sec auth actionclass connaction = do + (runst, conn, releaseconn) <- + getP2PConnection apiver st cu su bypass sec auth actionclass + connaction' runst conn + `finally` liftIO releaseconn + where + connaction' runst conn = connaction runst conn >>= \case + Right r -> return r + Left err -> throwError $ + err500 { errBody = encodeBL (describeProtoFailure err) } + +getP2PConnection + :: APIVersion v + => v + -> P2PHttpServerState + -> B64UUID ClientSide + -> B64UUID ServerSide + -> [B64UUID Bypass] + -> IsSecure + -> Maybe Auth + -> ActionClass + -> Handler (RunState, P2PConnection, ReleaseP2PConnection) +getP2PConnection apiver st cu su bypass sec auth actionclass = case (getServerMode st sec auth, actionclass) of (Just P2P.ServeReadWrite, _) -> go P2P.ServeReadWrite (Just P2P.ServeAppendOnly, RemoveAction) -> throwError err403 @@ -77,9 +100,7 @@ withP2PConnection apiver st cu su bypass sec auth actionclass connaction = throwError err502 { errBody = encodeBL err } Left TooManyConnections -> throwError err503 - Right (runst, conn, releaseconn) -> - connaction' runst conn - `finally` liftIO releaseconn + Right v -> return v where cp = ConnectionParams { connectionProtocolVersion = protocolVersion apiver @@ -88,11 +109,6 @@ withP2PConnection apiver st cu su bypass sec auth actionclass connaction = , connectionBypass = map fromB64UUID bypass , connectionServerMode = servermode } - - connaction' runst conn = connaction runst conn >>= \case - Right r -> return r - Left err -> throwError $ - err500 { errBody = encodeBL (describeProtoFailure err) } basicAuthRequired :: ServerError basicAuthRequired = err401 { errHeaders = [(h, v)] } @@ -119,11 +135,13 @@ type AcquireP2PConnection = ( Either ConnectionProblem ( RunState , P2PConnection - , IO () -- ^ release connection + , ReleaseP2PConnection -- ^ release connection ) ) -{- Runs P2P actions in the local repository only. -} +type ReleaseP2PConnection = IO () + +{- Acquire P2P connections to the local repository. -} -- TODO need worker pool, this can only service a single request at -- a time. -- TODO proxies @@ -137,8 +155,8 @@ withLocalP2PConnections a = do where acquireconn reqv connparams = do respvar <- newEmptyTMVarIO - liftIO $ atomically $ putTMVar reqv (connparams, respvar) - liftIO $ atomically $ takeTMVar respvar + atomically $ putTMVar reqv (connparams, respvar) + atomically $ takeTMVar respvar servicer reqv relv = do reqrel <- liftIO $ diff --git a/P2P/Protocol.hs b/P2P/Protocol.hs index 9f8131d3b8..ca63aabac4 100644 --- a/P2P/Protocol.hs +++ b/P2P/Protocol.hs @@ -310,6 +310,10 @@ data LocalF c -- content been transferred. | StoreContentTo FilePath (Maybe IncrementalVerifier) Offset Len (Proto L.ByteString) (Proto (Maybe Validity)) ((Bool, Verification) -> c) -- ^ Like StoreContent, but stores the content to a temp file. + | SendContentWith (L.ByteString -> Annex (Maybe Validity -> Annex Bool)) (Proto L.ByteString) (Proto (Maybe Validity)) (Bool -> c) + -- ^ Reads content from the Proto L.ByteString and sends it to the + -- callback. The callback must consume the whole lazy ByteString, + -- before it returns a validity checker. | SetPresent Key UUID c | CheckContentPresent Key (Bool -> c) -- ^ Checks if the whole content of the key is locally present. -- 2.30.2