let remoteconn = P2PConnection
{ connRepo = Nothing
, connCheckAuth = const False
- , connIhdl = P2PHandleTMVar ihdl iwaitv
- , connOhdl = P2PHandleTMVar ohdl owaitv
+ , connIhdl = P2PHandleTMVar ihdl (Just iwaitv)
+ , connOhdl = P2PHandleTMVar ohdl (Just owaitv)
, connIdent = ConnIdent (Just (Remote.name r))
}
let closeremoteconn = do
-- XXX remove this
when (isNothing (portOption o)) $ do
liftIO $ putStrLn "test begins"
- testGet
+ testPut
giveup "TEST DONE"
withLocalP2PConnections workerpool $ \acquireconn -> liftIO $ do
authenv <- getAuthEnv
Nothing
liftIO $ print res
+testPut = do
+ mgr <- httpManager <$> getUrlOptions
+ burl <- liftIO $ parseBaseUrl "http://localhost:8080/"
+ res <- clientPut (mkClientEnv mgr burl)
+ (P2P.ProtocolVersion 3)
+ (B64Key (fromJust $ deserializeKey ("WORM-s30-m1720547401--foo")))
+ (B64UUID (toUUID ("f11773f0-11e1-45b2-9805-06db16768efe" :: String)))
+ (B64UUID (toUUID ("cu" :: String)))
+ []
+ Nothing
+ Nothing
+ (AssociatedFile (Just "foo"))
+ "foocontent"
+ 30
+ (liftIO (print "validity check") >> return True)
+ liftIO $ print res
+
testRemove = do
mgr <- httpManager <$> getUrlOptions
burl <- liftIO $ parseBaseUrl "http://localhost:8080/"
import P2P.IO
import P2P.Annex
import Annex.WorkerPool
+import Annex.Concurrent
import Types.WorkerPool
import Types.Direction
import Utility.Metered
import qualified Data.ByteString as B
import qualified Data.ByteString.Lazy as L
import qualified Data.ByteString.Lazy.Internal as LI
+import Data.Char
import Control.Concurrent.STM
import Control.Concurrent.Async
import Control.Concurrent
:<|> "git-annex" :> SU :> PV0 :> "remove" :> RemoveAPI RemoveResult
:<|> "git-annex" :> SU :> PV3 :> "remove-before" :> RemoveBeforeAPI
:<|> "git-annex" :> SU :> PV3 :> "gettimestamp" :> GetTimestampAPI
- :<|> "git-annex" :> SU :> PV3 :> "put" :> DataLengthHeader
- :> PutAPI PutResultPlus
- :<|> "git-annex" :> SU :> PV2 :> "put" :> DataLengthHeader
- :> PutAPI PutResultPlus
- :<|> "git-annex" :> SU :> PV1 :> "put" :> DataLengthHeader
- :> PutAPI PutResult
- :<|> "git-annex" :> SU :> PV0 :> "put"
- :> PutAPI PutResult
+ :<|> "git-annex" :> SU :> PV3 :> "put" :> PutAPI PutResultPlus
+ :<|> "git-annex" :> SU :> PV2 :> "put" :> PutAPI PutResultPlus
+ :<|> "git-annex" :> SU :> PV1 :> "put" :> PutAPI PutResult
+ :<|> "git-annex" :> SU :> PV0 :> "put" :> PutAPI PutResult
:<|> "git-annex" :> SU :> PV3 :> "putoffset"
:> PutOffsetAPI PutOffsetResultPlus
:<|> "git-annex" :> SU :> PV2 :> "putoffset"
:<|> servePut st id
:<|> servePut st id
:<|> servePut st dePlus
- :<|> (\su v -> servePut st dePlus su v Nothing)
+ :<|> servePut st dePlus
:<|> servePutOffset st id
:<|> servePutOffset st id
:<|> servePutOffset st dePlus
-> Handler (Headers '[DataLengthHeader] (S.SourceT IO B.ByteString))
serveGetGeneric st su@(B64UUID u) k =
-- Use V0 because it does not alter the returned data to indicate
- -- InValid content.
+ -- Invalid content.
serveGet st su V0 k cu [] Nothing Nothing
where
-- Reuse server UUID as client UUID.
-> Maybe Auth
-> Handler (Headers '[DataLengthHeader] (S.SourceT IO B.ByteString))
serveGet st su apiver (B64Key k) cu bypass baf startat sec auth = do
- conn <- getP2PConnection apiver st cu su bypass sec auth ReadAction
+ conn <- getP2PConnection apiver st cu su bypass sec auth ReadAction id
bsv <- liftIO newEmptyTMVarIO
endv <- liftIO newEmptyTMVarIO
validityv <- liftIO newEmptyTMVarIO
return $ \v -> do
liftIO $ atomically $ putTMVar validityv v
return True
- v <- enteringStage (TransferStage Upload) $
+ enteringStage (TransferStage Upload) $
runFullProto (clientRunState conn) (clientP2PConnection conn) $
void $ receiveContent Nothing nullMeterUpdate
sizer storer getreq
- return v
void $ liftIO $ forkIO $ waitfinal endv finalv conn annexworker
(Len len, bs) <- liftIO $ atomically $ takeTMVar bsv
bv <- liftIO $ newMVar (L.toChunks bs)
-- Make sure the annexworker is not left blocked on endv
-- if the client disconnected early.
void $ liftIO $ atomically $ tryPutTMVar endv ()
- void $ void $ tryNonAsync $ wait annexworker
+ void $ tryNonAsync $ wait annexworker
void $ tryNonAsync $ releaseP2PConnection conn
sizer = pure $ Len $ case startat of
v3 :<|> _ = client p2pHttpAPI
type PutAPI result
- = KeyParam
+ = DataLengthHeaderRequired
+ :> KeyParam
:> CU Required
:> BypassUUIDs
:> AssociatedFileParam
-> (PutResultPlus -> t)
-> B64UUID ServerSide
-> v
- -> Maybe DataLength
+ -> DataLength
-> B64Key
-> B64UUID ClientSide
-> [B64UUID Bypass]
-> IsSecure
-> Maybe Auth
-> Handler t
-servePut st resultmangle su apiver datalen k cu bypass af offset stream sec auth = do
- res <- withP2PConnection apiver st cu su bypass sec auth WriteAction
- $ \conn ->
- liftIO $ proxyClientNetProto conn undefined
+servePut st resultmangle su apiver (DataLength len) (B64Key k) cu bypass baf moffset stream sec auth = do
+ validityv <- liftIO newEmptyTMVarIO
+ let validitycheck = local $ runValidityCheck $
+ liftIO $ atomically $ readTMVar validityv
+ content <- liftIO $ S.unSourceT stream (gather validityv)
+ conn <- getP2PConnection apiver st cu su bypass sec auth WriteAction $
+ \st -> st { connectionWaitVar = False }
+ res <- liftIO $ inAnnexWorker st $
+ enteringStage (TransferStage Download) $
+ runFullProto (clientRunState conn) (clientP2PConnection conn) $
+ protoaction content validitycheck
+
case res of
- Right (stored, plusuuids) -> return $ resultmangle $
- PutResultPlus stored plusuuids
+ Right (Right (Just plusuuids)) -> return $ resultmangle $
+ PutResultPlus True (map B64UUID plusuuids)
+ Right (Right Nothing) -> return $ resultmangle $
+ PutResultPlus False []
+ Right (Left protofail) -> throwError $
+ err500 { errBody = encodeBL (describeProtoFailure protofail) }
Left err -> throwError $
- err500 { errBody = encodeBL err }
+ err500 { errBody = encodeBL (show err) }
+ where
+ protoaction content validitycheck = put' k af $ \offset' ->
+ let offsetdelta = offset' - offset
+ in case compare offset' offset of
+ EQ -> sendContent' nullMeterUpdate (Len len)
+ content validitycheck
+ GT -> sendContent' nullMeterUpdate
+ (Len (len - fromIntegral offsetdelta))
+ (L.drop (fromIntegral offsetdelta) content)
+ validitycheck
+ LT -> sendContent' nullMeterUpdate
+ (Len 0)
+ mempty
+ (return Invalid)
+
+ offset = case moffset of
+ Just (Offset o) -> o
+ Nothing -> 0
+
+ af = AssociatedFile $ case baf of
+ Just (B64FilePath f) -> Just f
+ Nothing -> Nothing
+
+ -- Streams the ByteString from the client. Avoids returning a longer
+ -- or shorter than expected ByteString by truncating or padding;
+ -- in such cases the data is not Valid.
+ gather validityv = unsafeInterleaveIO . go 0
+ where
+ go n S.Stop
+ | n == len = do
+ atomically $ writeTMVar validityv Valid
+ return LI.Empty
+ | otherwise = do
+ atomically $ writeTMVar validityv Invalid
+ padout n
+ go n (S.Error _err) = do
+ atomically $ writeTMVar validityv Invalid
+ padout n
+ go n (S.Skip s) = go n s
+ go n (S.Effect ms) = ms >>= go n
+ go n (S.Yield v s) =
+ let !n' = n + fromIntegral (B.length v)
+ in if n' > len
+ then do
+ atomically $ writeTMVar validityv Invalid
+ return $ LI.Chunk
+ (B.take (fromIntegral (len - n')) v)
+ LI.Empty
+ else LI.Chunk v <$> unsafeInterleaveIO (go n' s)
+
+ padout n =return $ LI.Chunk
+ (B.replicate (fromIntegral (len-n))
+ (fromIntegral (ord 'X')))
+ LI.Empty
clientPut
- :: ProtocolVersion
- -> DataLength
+ :: ClientEnv
+ -> ProtocolVersion
-> B64Key
- -> B64UUID ClientSide
-> B64UUID ServerSide
+ -> B64UUID ClientSide
-> [B64UUID Bypass]
- -> Maybe B64FilePath
- -> Maybe Offset
- -> S.SourceT IO B.ByteString
-> Maybe Auth
- -> ClientM PutResultPlus
-clientPut (ProtocolVersion ver) sz k cu su bypass af o src auth = case ver of
- 3 -> v3 su V3 (Just sz) k cu bypass af o src auth
- 2 -> v2 su V2 (Just sz) k cu bypass af o src auth
- 1 -> plus <$> v1 su V1 (Just sz) k cu bypass af o src auth
- 0 -> plus <$> v0 su V0 k cu bypass af o src auth
- _ -> error "unsupported protocol version"
+ -> Maybe Offset
+ -> AssociatedFile
+ -> FilePath
+ -> FileSize
+ -> Annex Bool
+ -> Annex PutResultPlus
+clientPut clientenv (ProtocolVersion ver) k su cu bypass auth moffset af contentfile contentfilesize validitycheck = do
+ checkv <- liftIO newEmptyTMVarIO
+ checkresultv <- liftIO newEmptyTMVarIO
+ let checker = do
+ liftIO $ atomically $ takeTMVar checkv
+ validitycheck >>= liftIO . atomically . putTMVar checkresultv
+ checkerthread <- liftIO . async =<< forkState checker
+ liftIO (withClientM (cli (stream checkv checkresultv)) clientenv return) >>= \case
+ Left err -> do
+ void $ liftIO $ atomically $ tryPutTMVar checkv ()
+ join $ liftIO (wait checkerthread)
+ throwM err
+ Right res -> do
+ join $ liftIO (wait checkerthread)
+ return res
where
+ stream checkv checkresultv = S.SourceT $ \a -> do
+ bl <- L.readFile contentfile
+ v <- newMVar (0, L.toChunks bl)
+ a (go v)
+ where
+ go v = S.fromActionStep B.null $ do
+ res <- modifyMVar v $ pure . \case
+ (n, []) -> ((n, []), (n, Nothing))
+ (n, (b:bs)) ->
+ let !n' = n + B.length b
+ in ((n', bs), (n, Just b))
+ case res of
+ (_, Just b) -> return b
+ (n, Nothing) -> do
+ void $ liftIO $ atomically $
+ tryPutTMVar checkv ()
+ valid <- liftIO $ atomically $
+ readTMVar checkresultv
+ if not valid
+ then if n == fromIntegral contentfilesize
+ then do
+ modifyMVar_ v $ \(_n, l) ->
+ pure (n+1, l)
+ return "X"
+ else return B.empty
+ else return B.empty
+
+ baf = case af of
+ AssociatedFile Nothing -> Nothing
+ AssociatedFile (Just f) -> Just (B64FilePath f)
+
+ len = DataLength $ case moffset of
+ Nothing -> contentfilesize
+ Just (Offset o) -> contentfilesize - fromIntegral o
+
+ cli src = case ver of
+ 3 -> v3 su V3 len k cu bypass baf moffset src auth
+ 2 -> v2 su V2 len k cu bypass baf moffset src auth
+ 1 -> plus <$> v1 su V1 len k cu bypass baf moffset src auth
+ 0 -> plus <$> v0 su V0 len k cu bypass baf moffset src auth
+ _ -> error "unsupported protocol version"
+
_ :<|> _ :<|> _ :<|> _ :<|>
_ :<|> _ :<|> _ :<|> _ :<|>
_ :<|> _ :<|> _ :<|> _ :<|>
type DataLengthHeader = Header DataLengthHeader' DataLength
+type DataLengthHeaderRequired = Header' '[Required] DataLengthHeader' DataLength
+
type DataLengthHeader' = "X-git-annex-data-length"
type LockIDParam = QueryParam' '[Required] "lockid" LockID
-> (P2PConnectionPair -> Handler (Either ProtoFailure a))
-> Handler a
withP2PConnection apiver st cu su bypass sec auth actionclass connaction = do
- conn <- getP2PConnection apiver st cu su bypass sec auth actionclass
+ conn <- getP2PConnection apiver st cu su bypass sec auth actionclass id
connaction' conn
`finally` liftIO (releaseP2PConnection conn)
where
-> IsSecure
-> Maybe Auth
-> ActionClass
+ -> (ConnectionParams -> ConnectionParams)
-> Handler P2PConnectionPair
-getP2PConnection apiver st cu su bypass sec auth actionclass =
+getP2PConnection apiver st cu su bypass sec auth actionclass fconnparams =
case (getServerMode st sec auth, actionclass) of
(Just P2P.ServeReadWrite, _) -> go P2P.ServeReadWrite
(Just P2P.ServeAppendOnly, RemoveAction) -> throwError err403
throwError err503
Right v -> return v
where
- cp = ConnectionParams
+ cp = fconnparams $ ConnectionParams
{ connectionProtocolVersion = protocolVersion apiver
, connectionServerUUID = fromB64UUID su
, connectionClientUUID = fromB64UUID cu
, connectionBypass = map fromB64UUID bypass
, connectionServerMode = servermode
+ , connectionWaitVar = True
}
basicAuthRequired :: ServerError
, connectionClientUUID :: UUID
, connectionBypass :: [UUID]
, connectionServerMode :: P2P.ServerMode
+ , connectionWaitVar :: Bool
}
deriving (Show, Eq, Ord)
hdl2 <- newEmptyTMVarIO
wait1 <- newEmptyTMVarIO
wait2 <- newEmptyTMVarIO
- let h1 = P2PHandleTMVar hdl1 wait1
- let h2 = P2PHandleTMVar hdl2 wait2
+ let h1 = P2PHandleTMVar hdl1 $
+ if connectionWaitVar connparams
+ then Just wait1
+ else Nothing
+ let h2 = P2PHandleTMVar hdl2 $
+ if connectionWaitVar connparams
+ then Just wait2
+ else Nothing
let serverconn = P2PConnection Nothing
(const True) h1 h2
(ConnIdent (Just "http server"))
data P2PHandle
= P2PHandle Handle
- | P2PHandleTMVar (TMVar (Either L.ByteString Message)) (TMVar ())
+ | P2PHandleTMVar (TMVar (Either L.ByteString Message)) (Maybe (TMVar ()))
signalFullyConsumedByteString :: P2PHandle -> IO ()
signalFullyConsumedByteString (P2PHandle _) = return ()
-signalFullyConsumedByteString (P2PHandleTMVar _ waitv) =
+signalFullyConsumedByteString (P2PHandleTMVar _ Nothing) = return ()
+signalFullyConsumedByteString (P2PHandleTMVar _ (Just waitv)) =
atomically $ putTMVar waitv ()
data P2PConnection = P2PConnection
ifM (atomically (tryPutTMVar mv (Right m)))
( return $ Right ()
, return $ Left $ toException $
- P2PTMVarException "TMVar left full"
+ P2PTMVarException ("TMVar left full " ++ show m)
)
case v of
Left e -> return $ Left $ ProtoFailureException e
liftIO $ atomically $ putTMVar mv (Left b)
-- Wait for the whole bytestring to
-- be processed.
- liftIO $ atomically $ takeTMVar waitv
+ liftIO $ maybe noop (atomically . takeTMVar) waitv
runner next
ReceiveBytes len p next ->
case connIhdl conn of
storer = storeContentTo dest iv
put :: Key -> AssociatedFile -> MeterUpdate -> Proto (Maybe [UUID])
-put key af p = do
+put key af p = put' key af $ \offset ->
+ sendContent key af Nothing offset p
+
+put' :: Key -> AssociatedFile -> (Offset -> Proto (Maybe [UUID])) -> Proto (Maybe [UUID])
+put' key af sender = do
net $ sendMessage (PUT (ProtoAssociatedFile af) key)
r <- net receiveMessage
case r of
- Just (PUT_FROM offset) -> sendContent key af Nothing offset p
+ Just (PUT_FROM offset) -> sender offset
Just ALREADY_HAVE -> return (Just [])
Just (ALREADY_HAVE_PLUS uuids) -> return (Just uuids)
_ -> do
-- Content not available to send. Indicate this by sending
-- empty data and indlicate it's invalid.
go Nothing = sender (Len 0) L.empty (return Invalid)
- sender len content validitycheck = do
- let p' = offsetMeterUpdate p (toBytesProcessed n)
- net $ sendMessage (DATA len)
- net $ sendBytes len content p'
- ver <- net getProtocolVersion
- when (ver >= ProtocolVersion 1) $
- net . sendMessage . VALIDITY =<< validitycheck
- checkSuccessPlus
+
+ sender = sendContent' p'
+
+ p' = offsetMeterUpdate p (toBytesProcessed n)
+
+sendContent' :: MeterUpdate -> Len -> L.ByteString -> Proto Validity -> Proto (Maybe [UUID])
+sendContent' p len content validitycheck = do
+ net $ sendMessage (DATA len)
+ net $ sendBytes len content p
+ ver <- net getProtocolVersion
+ when (ver >= ProtocolVersion 1) $
+ net . sendMessage . VALIDITY =<< validitycheck
+ checkSuccessPlus
receiveContent
:: Observable t
The server may respond with ALREADY-HAVE if it already
had the content of that key.
-In protocol version 2, the server can optionally reply with
+In protocol version 2 and above, the server can optionally reply with
ALREADY-HAVE-PLUS. The subsequent list of UUIDs are additional
UUIDs where the content is stored, in addition to the UUID where
the client was going to send it.
The client then sends a DATA message with content of the file from
the offset to the end of file.
-In protocol version 1, after the data, the client sends an additional
-message, to indicate if the content of the file has changed while it
-was being sent.
+In protocol version 1 and above, after the data, the client sends an
+additional message, to indicate if the content of the file has changed
+while it was being sent.
INVALID
VALID
If the server successfully receives the data and stores the content,
it replies with SUCCESS. Otherwise, FAILURE.
-In protocol version 2, the server can optionally reply with SUCCESS-PLUS
-and a list of UUIDs where the content was stored.
+In protocol version 2 and above, the server can optionally reply with
+SUCCESS-PLUS and a list of UUIDs where the content was stored.
## Getting content from the server
The server then sends a DATA message with the content of the file
from the offset to end of file.
-In protocol version 1, after the data, the server sends an additional
+In protocol version 1 and above, after the data, the server sends an additional
message, to indicate if the content of the file has changed while it
was being sent.
Each request in the protocol is versioned. The versions correspond
to P2P protocol versions.
-The protocol version comes before the request. Eg: `/git-annex/v3/put`
-
If the server does not support a particular protocol version, the
request will fail with a 400 Bad Request, and the client should fall
back to an earlier protocol version.
### POST /git-annex/$uuid/v0/put
-Same as v1, except there is no X-git-annex-data-length header.
-Additional checking client-side will be required to validate the data.
+Same as v1, except additional checking is done to validate the data.
### POST /git-annex/$uuid/v3/putoffset
## work notes
-* Implement: servePut, servePutOffset, serveLockContent
-
-* I have a file `servant.hs` in the httpproto branch that works through some
- of the bytestring streaming issues.
+* Implement: servePutOffset, serveLockContent
* A Locker should expire the lock on its own after 10 minutes initially.
+++ /dev/null
-{-# LANGUAGE BangPatterns #-}
-{-# LANGUAGE DataKinds #-}
-{-# LANGUAGE DeriveGeneric #-}
-{-# LANGUAGE TypeOperators #-}
-{-# LANGUAGE LambdaCase #-}
-module Main (main) where
-
-import Control.Concurrent
- (threadDelay)
-import Control.Monad.IO.Class
- (MonadIO (..))
-import qualified Data.ByteString as BS
-import qualified Data.ByteString.Lazy as BL
-import qualified Data.ByteString.Lazy.Internal as BLI
-import Data.Maybe
- (fromMaybe)
-import Network.HTTP.Client
- (defaultManagerSettings, newManager)
-import Network.Wai
- (Application)
-import System.Environment
- (getArgs, lookupEnv)
-import Text.Read
- (readMaybe)
-
-import Servant
-import Servant.Client.Streaming
-import qualified Servant.Types.SourceT as S
-import Control.Concurrent.MVar
-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 :<|> 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 :<|> writecli = client api
-
-main :: IO ()
-main = do
- args <- getArgs
- case args of
- ("server":_) -> do
- putStrLn "Starting cookbook-basic-streaming at http://localhost:8000"
- port <- fromMaybe 8000 . (>>= readMaybe) <$> lookupEnv "PORT"
- Warp.run port app
- ("client":ns:_) -> do
- mgr <- newManager defaultManagerSettings
- burl <- parseBaseUrl "http://localhost:8000/"
- withClientM (writecli getit) (mkClientEnv mgr burl) $ \me -> case me of
- Left err -> print err
- 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
- bl <- BL.readFile file
- v <- newMVar (BL.toChunks bl)
- return (go v)
- where
- go :: MVar [BS.ByteString] -> S.StepT IO BS.ByteString
- go v = S.fromActionStep BS.null $ do
- print "chunk"
- modifyMVar v $ pure . \case
- [] -> ([], BS.empty)
- (b:bs) -> (bs, b)
-
-gatherbytestring :: S.StepT IO BS.ByteString -> IO BL.ByteString
-gatherbytestring x = do
- l <- unsafeInterleaveIO $ go x
- return l
- where
- go S.Stop = return BLI.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
- BLI.Chunk v <$> unsafeInterleaveIO (go s)
-
projects / git-annex.git / commitdiff