From: Joey Hess Date: Wed, 10 Jul 2024 21:48:48 +0000 (-0400) Subject: a little progress on serveGet hang X-Git-Tag: archive/raspbian/10.20250416-2+rpi1~1^2~21^2~228^2~122 X-Git-Url: https://dgit.raspbian.org/?a=commitdiff_plain;h=80fb5445b5b13c60f9b63f4571331365e7487074;p=git-annex.git a little progress on serveGet hang Now it gets to the validity checker, but it seems it never runs it. --- diff --git a/P2P/Annex.hs b/P2P/Annex.hs index 59428db8c9..d922b28e50 100644 --- a/P2P/Annex.hs +++ b/P2P/Annex.hs @@ -107,16 +107,23 @@ runLocal runst runner a = case a of ProtoFailureMessage "Transfer failed" let consumer' b ti = do validator <- consumer b + liftIO $ print "got validator" indicatetransferred ti + liftIO $ print "indicatetransferred ti done" 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 -> + Right validator -> do + liftIO $ print "running validity check" runner validitycheck >>= \case - Right v -> Right <$> validator v - _ -> Right <$> validator Nothing + Right v -> do + liftIO $ print ("calling validator 1", v) + Right <$> validator v + _ -> do + liftIO $ print "calling validator nothing" + Right <$> validator Nothing case v of Left e -> return $ Left $ ProtoFailureException e Right (Left e) -> return $ Left e diff --git a/P2P/Http.hs b/P2P/Http.hs index 19d43332ae..7e5092fdb4 100644 --- a/P2P/Http.hs +++ b/P2P/Http.hs @@ -158,8 +158,11 @@ serveGet st apiver (B64Key k) cu su bypass baf startat sec auth = do aid <- liftIO $ async $ inAnnexWorker st $ do let consumer bs = do liftIO $ atomically $ putTMVar bsv bs + liftIO $ print "consumer waiting for endv" liftIO $ atomically $ takeTMVar endv + liftIO $ print "consumer took endv" return $ \v -> do + liftIO $ print "consumer put validityv" liftIO $ atomically $ putTMVar validityv v return True @@ -178,17 +181,22 @@ serveGet st apiver (B64Key k) cu su bypass baf startat sec auth = do 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) + modifyMVar bv $ nextchunk $ + cleanup (releaseconn, endv, validityv, aid) + nextchunk atend (b:bs) + | not (B.null b) = return (bs, b) + | otherwise = nextchunk atend bs + nextchunk atend [] = do + endbit <- atend + return ([], endbit) + cleanup (releaseconn, endv, validityv, aid) = ifM (atomically $ isEmptyTMVar endv) - ( pure mempty - , do + ( do + print "at end" atomically $ putTMVar endv () + print "signaled end" validity <- atomically $ takeTMVar validityv print ("got validity", validity) wait aid >>= \case @@ -207,6 +215,7 @@ serveGet st apiver (B64Key k) cu su bypass baf startat sec auth = do Just Invalid -> "XXXXXXX" -- FIXME: need to count bytes and emit -- something to make it invalid + , pure mempty ) sizer = pure $ Len $ case startat of diff --git a/servant.hs b/servant.hs index 9cf0dd0b13..c6765a7d79 100644 --- a/servant.hs +++ b/servant.hs @@ -32,23 +32,33 @@ import System.IO.Unsafe import qualified Network.Wai.Handler.Warp as Warp type API = "readme" :> StreamGet NoFraming OctetStream (SourceIO BS.ByteString) + :<|> "writeme" :> StreamBody NoFraming OctetStream (SourceIO BS.ByteString) :> Post '[JSON] Bool api :: Proxy API api = Proxy server :: Server API -server = readme where +server = readme :<|> writeme where readme = liftIO $ do putStrLn "/proxy" return $ S.SourceT $ \k -> do k =<< readfilelazy "README.md" k =<< readfilelazy "another" +writeme :: SourceIO BS.ByteString -> Handler Bool +writeme src = do + liftIO $ print "gathering lazy bytestring" + b <- liftIO $ S.unSourceT src gatherbytestring + liftIO $ print "got lazy bytestring, writing to file" + liftIO$ BL.writeFile "writem" b + liftIO$ print "write complete" + return True + app :: Application app = serve api server cli :: ClientM (S.SourceT IO BS.ByteString) -cli = client api +cli :<|> writecli = client api main :: IO () main = do @@ -61,17 +71,17 @@ main = do ("client":ns:_) -> do mgr <- newManager defaultManagerSettings burl <- parseBaseUrl "http://localhost:8000/" - withClientM cli (mkClientEnv mgr burl) $ \me -> case me of + withClientM (writecli getit) (mkClientEnv mgr burl) $ \me -> case me of Left err -> print err - Right src -> do - b <- S.unSourceT src gatherbytestring - liftIO $ print "got it all, writing" - BL.writeFile "got" (BL.init b) + Right src -> print src _ -> do putStrLn "Try:" putStrLn "cabal new-run cookbook-basic-streaming server" putStrLn "cabal new-run cookbook-basic-streaming client 10" putStrLn "time curl -H 'Accept: application/json' localhost:8000/slow/5" + where + getit = S.SourceT $ \k -> do + k =<< readfilelazy "/home/joey/README.md" readfilelazy :: FilePath -> IO (S.StepT IO BS.ByteString) readfilelazy file = do