mirrors/git-annex
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

move the p2phttp server state map into a data type

Browse source
This commit is contained in:
Joey Hess 2024-11-21 12:24:14 -04:00
parent d7ed99a55f
commit 6bdf4a85fb
No known key found for this signature in database
GPG key ID: DB12DB0FF05F8F38
3 changed files with 38 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

7
Command/P2PHttp.hs
View file

@ -110,9 +110,12 @@ seek o = getAnnexWorkerPool $ \workerpool ->
where where
go workerpool servinguuids acquireconn = liftIO $ do go workerpool servinguuids acquireconn = liftIO $ do
authenv <- getAuthEnv authenv <- getAuthEnv
st <- mkP2PHttpServerState acquireconn workerpool $ st <- mkPerRepoServerState acquireconn workerpool $
mkGetServerMode authenv o mkGetServerMode authenv o
let mst = M.fromList $ zip servinguuids (repeat st) let mst = P2PHttpServerState
{ servedRepos = M.fromList $
zip servinguuids (repeat st)
}
let settings = Warp.setPort port $ Warp.setHost host $ let settings = Warp.setPort port $ Warp.setHost host $
Warp.defaultSettings Warp.defaultSettings
case (certFileOption o, privateKeyFileOption o) of case (certFileOption o, privateKeyFileOption o) of

29
P2P/Http/Server.hs
View file

@ -40,16 +40,15 @@ import qualified Servant.Types.SourceT as S
import qualified Data.ByteString as B import qualified Data.ByteString as B
import qualified Data.ByteString.Lazy as L import qualified Data.ByteString.Lazy as L
import qualified Data.ByteString.Lazy.Internal as LI import qualified Data.ByteString.Lazy.Internal as LI
import qualified Data.Map as M
import Control.Concurrent.Async import Control.Concurrent.Async
import Control.Concurrent import Control.Concurrent
import System.IO.Unsafe import System.IO.Unsafe
import Data.Either import Data.Either
p2pHttpApp :: M.Map UUID P2PHttpServerState -> Application p2pHttpApp :: P2PHttpServerState -> Application
p2pHttpApp = serve p2pHttpAPI . serveP2pHttp p2pHttpApp = serve p2pHttpAPI . serveP2pHttp
serveP2pHttp :: M.Map UUID P2PHttpServerState -> Server P2PHttpAPI serveP2pHttp :: P2PHttpServerState -> Server P2PHttpAPI
serveP2pHttp st serveP2pHttp st
= serveGet st = serveGet st
:<|> serveGet st :<|> serveGet st
@ -92,7 +91,7 @@ serveP2pHttp st
:<|> serveGetGeneric st :<|> serveGetGeneric st
serveGetGeneric serveGetGeneric
:: M.Map UUID P2PHttpServerState :: P2PHttpServerState
-> B64UUID ServerSide -> B64UUID ServerSide
-> B64Key -> B64Key
-> Maybe (B64UUID ClientSide) -> Maybe (B64UUID ClientSide)
@ -110,7 +109,7 @@ serveGetGeneric st su@(B64UUID u) k mcu bypass =
serveGet serveGet
:: APIVersion v :: APIVersion v
=> M.Map UUID P2PHttpServerState => P2PHttpServerState
-> B64UUID ServerSide -> B64UUID ServerSide
-> v -> v
-> B64Key -> B64Key
@ -223,7 +222,7 @@ serveGet mst su apiver (B64Key k) cu bypass baf startat sec auth = do
serveCheckPresent serveCheckPresent
:: APIVersion v :: APIVersion v
=> M.Map UUID P2PHttpServerState => P2PHttpServerState
-> B64UUID ServerSide -> B64UUID ServerSide
-> v -> v
-> B64Key -> B64Key
@ -241,7 +240,7 @@ serveCheckPresent st su apiver (B64Key k) cu bypass sec auth = do
serveRemove serveRemove
:: APIVersion v :: APIVersion v
=> M.Map UUID P2PHttpServerState => P2PHttpServerState
-> (RemoveResultPlus -> t) -> (RemoveResultPlus -> t)
-> B64UUID ServerSide -> B64UUID ServerSide
-> v -> v
@ -263,7 +262,7 @@ serveRemove st resultmangle su apiver (B64Key k) cu bypass sec auth = do
serveRemoveBefore serveRemoveBefore
:: APIVersion v :: APIVersion v
=> M.Map UUID P2PHttpServerState => P2PHttpServerState
-> B64UUID ServerSide -> B64UUID ServerSide
-> v -> v
-> B64Key -> B64Key
@ -286,7 +285,7 @@ serveRemoveBefore st su apiver (B64Key k) cu bypass (Timestamp ts) sec auth = do
serveGetTimestamp serveGetTimestamp
:: APIVersion v :: APIVersion v
=> M.Map UUID P2PHttpServerState => P2PHttpServerState
-> B64UUID ServerSide -> B64UUID ServerSide
-> v -> v
-> B64UUID ClientSide -> B64UUID ClientSide
@ -305,7 +304,7 @@ serveGetTimestamp st su apiver cu bypass sec auth = do
servePut servePut
:: APIVersion v :: APIVersion v
=> M.Map UUID P2PHttpServerState => P2PHttpServerState
-> (PutResultPlus -> t) -> (PutResultPlus -> t)
-> B64UUID ServerSide -> B64UUID ServerSide
-> v -> v
@ -397,7 +396,7 @@ servePut mst resultmangle su apiver _datapresent (DataLength len) k cu bypass ba
closeP2PConnection conn closeP2PConnection conn
servePutAction servePutAction
:: (P2PConnectionPair, P2PHttpServerState) :: (P2PConnectionPair, PerRepoServerState)
-> B64Key -> B64Key
-> Maybe B64FilePath -> Maybe B64FilePath
-> (P2P.Protocol.Offset -> Proto (Maybe [UUID])) -> (P2P.Protocol.Offset -> Proto (Maybe [UUID]))
@ -422,7 +421,7 @@ servePutResult resultmangle res = case res of
servePut' servePut'
:: APIVersion v :: APIVersion v
=> M.Map UUID P2PHttpServerState => P2PHttpServerState
-> (PutResultPlus -> t) -> (PutResultPlus -> t)
-> B64UUID ServerSide -> B64UUID ServerSide
-> v -> v
@ -440,7 +439,7 @@ servePut' st resultmangle su v = servePut st resultmangle su v Nothing
servePutOffset servePutOffset
:: APIVersion v :: APIVersion v
=> M.Map UUID P2PHttpServerState => P2PHttpServerState
-> (PutOffsetResultPlus -> t) -> (PutOffsetResultPlus -> t)
-> B64UUID ServerSide -> B64UUID ServerSide
-> v -> v
@ -464,7 +463,7 @@ servePutOffset st resultmangle su apiver (B64Key k) cu bypass sec auth = do
serveLockContent serveLockContent
:: APIVersion v :: APIVersion v
=> M.Map UUID P2PHttpServerState => P2PHttpServerState
-> B64UUID ServerSide -> B64UUID ServerSide
-> v -> v
-> B64Key -> B64Key
@ -501,7 +500,7 @@ serveLockContent mst su apiver (B64Key k) cu bypass sec auth = do
serveKeepLocked serveKeepLocked
:: APIVersion v :: APIVersion v
=> M.Map UUID P2PHttpServerState => P2PHttpServerState
-> B64UUID ServerSide -> B64UUID ServerSide
-> v -> v
-> LockID -> LockID

34
P2P/Http/State.hs
View file

@ -44,6 +44,10 @@ import Control.Concurrent.Async
import Data.Time.Clock.POSIX import Data.Time.Clock.POSIX
data P2PHttpServerState = P2PHttpServerState data P2PHttpServerState = P2PHttpServerState
{ servedRepos :: M.Map UUID PerRepoServerState
}
data PerRepoServerState = PerRepoServerState
{ acquireP2PConnection :: AcquireP2PConnection { acquireP2PConnection :: AcquireP2PConnection
, annexWorkerPool :: AnnexWorkerPool , annexWorkerPool :: AnnexWorkerPool
, getServerMode :: GetServerMode , getServerMode :: GetServerMode
@ -62,8 +66,8 @@ data ServerMode
} }
| CannotServeRequests | CannotServeRequests
mkP2PHttpServerState :: AcquireP2PConnection -> AnnexWorkerPool -> GetServerMode -> IO P2PHttpServerState mkPerRepoServerState :: AcquireP2PConnection -> AnnexWorkerPool -> GetServerMode -> IO PerRepoServerState
mkP2PHttpServerState acquireconn annexworkerpool getservermode = P2PHttpServerState mkPerRepoServerState acquireconn annexworkerpool getservermode = PerRepoServerState
<$> pure acquireconn <$> pure acquireconn
<*> pure annexworkerpool <*> pure annexworkerpool
<*> pure getservermode <*> pure getservermode
@ -75,7 +79,7 @@ data ActionClass = ReadAction | WriteAction | RemoveAction | LockAction
withP2PConnection withP2PConnection
:: APIVersion v :: APIVersion v
=> v => v
-> M.Map UUID P2PHttpServerState -> P2PHttpServerState
-> B64UUID ClientSide -> B64UUID ClientSide
-> B64UUID ServerSide -> B64UUID ServerSide
-> [B64UUID Bypass] -> [B64UUID Bypass]
@ -83,7 +87,7 @@ withP2PConnection
-> Maybe Auth -> Maybe Auth
-> ActionClass -> ActionClass
-> (ConnectionParams -> ConnectionParams) -> (ConnectionParams -> ConnectionParams)
-> ((P2PConnectionPair, P2PHttpServerState) -> Handler (Either ProtoFailure a)) -> ((P2PConnectionPair, PerRepoServerState) -> Handler (Either ProtoFailure a))
-> Handler a -> Handler a
withP2PConnection apiver mst cu su bypass sec auth actionclass fconnparams connaction = withP2PConnection apiver mst cu su bypass sec auth actionclass fconnparams connaction =
withP2PConnection' apiver mst cu su bypass sec auth actionclass fconnparams connaction' withP2PConnection' apiver mst cu su bypass sec auth actionclass fconnparams connaction'
@ -96,7 +100,7 @@ withP2PConnection apiver mst cu su bypass sec auth actionclass fconnparams conna
withP2PConnection' withP2PConnection'
:: APIVersion v :: APIVersion v
=> v => v
-> M.Map UUID P2PHttpServerState -> P2PHttpServerState
-> B64UUID ClientSide -> B64UUID ClientSide
-> B64UUID ServerSide -> B64UUID ServerSide
-> [B64UUID Bypass] -> [B64UUID Bypass]
@ -104,7 +108,7 @@ withP2PConnection'
-> Maybe Auth -> Maybe Auth
-> ActionClass -> ActionClass
-> (ConnectionParams -> ConnectionParams) -> (ConnectionParams -> ConnectionParams)
-> ((P2PConnectionPair, P2PHttpServerState) -> Handler a) -> ((P2PConnectionPair, PerRepoServerState) -> Handler a)
-> Handler a -> Handler a
withP2PConnection' apiver mst cu su bypass sec auth actionclass fconnparams connaction = do withP2PConnection' apiver mst cu su bypass sec auth actionclass fconnparams connaction = do
(conn, st) <- getP2PConnection apiver mst cu su bypass sec auth actionclass fconnparams (conn, st) <- getP2PConnection apiver mst cu su bypass sec auth actionclass fconnparams
@ -114,7 +118,7 @@ withP2PConnection' apiver mst cu su bypass sec auth actionclass fconnparams conn
getP2PConnection getP2PConnection
:: APIVersion v :: APIVersion v
=> v => v
-> M.Map UUID P2PHttpServerState -> P2PHttpServerState
-> B64UUID ClientSide -> B64UUID ClientSide
-> B64UUID ServerSide -> B64UUID ServerSide
-> [B64UUID Bypass] -> [B64UUID Bypass]
@ -122,7 +126,7 @@ getP2PConnection
-> Maybe Auth -> Maybe Auth
-> ActionClass -> ActionClass
-> (ConnectionParams -> ConnectionParams) -> (ConnectionParams -> ConnectionParams)
-> Handler (P2PConnectionPair, P2PHttpServerState) -> Handler (P2PConnectionPair, PerRepoServerState)
getP2PConnection apiver mst cu su bypass sec auth actionclass fconnparams = getP2PConnection apiver mst cu su bypass sec auth actionclass fconnparams =
checkAuthActionClass mst su sec auth actionclass go checkAuthActionClass mst su sec auth actionclass go
where where
@ -143,15 +147,15 @@ getP2PConnection apiver mst cu su bypass sec auth actionclass fconnparams =
} }
checkAuthActionClass checkAuthActionClass
:: M.Map UUID P2PHttpServerState :: P2PHttpServerState
-> B64UUID ServerSide -> B64UUID ServerSide
-> IsSecure -> IsSecure
-> Maybe Auth -> Maybe Auth
-> ActionClass -> ActionClass
-> (P2PHttpServerState -> P2P.ServerMode -> Handler a) -> (PerRepoServerState -> P2P.ServerMode -> Handler a)
-> Handler a -> Handler a
checkAuthActionClass mst su sec auth actionclass go = checkAuthActionClass mst su sec auth actionclass go =
case M.lookup (fromB64UUID su) mst of case M.lookup (fromB64UUID su) (servedRepos mst) of
Just st -> select st Just st -> select st
Nothing -> throwError err404 Nothing -> throwError err404
where where
@ -493,13 +497,13 @@ mkLocker lock unlock = do
wait locktid wait locktid
return Nothing return Nothing
storeLock :: LockID -> Locker -> P2PHttpServerState -> IO () storeLock :: LockID -> Locker -> PerRepoServerState -> IO ()
storeLock lckid locker st = atomically $ do storeLock lckid locker st = atomically $ do
m <- takeTMVar (openLocks st) m <- takeTMVar (openLocks st)
let !m' = M.insert lckid locker m let !m' = M.insert lckid locker m
putTMVar (openLocks st) m' putTMVar (openLocks st) m'
keepingLocked :: LockID -> P2PHttpServerState -> IO () keepingLocked :: LockID -> PerRepoServerState -> IO ()
keepingLocked lckid st = do keepingLocked lckid st = do
m <- atomically $ readTMVar (openLocks st) m <- atomically $ readTMVar (openLocks st)
case M.lookup lckid m of case M.lookup lckid m of
@ -508,7 +512,7 @@ keepingLocked lckid st = do
atomically $ void $ atomically $ void $
tryPutTMVar (lockerTimeoutDisable locker) () tryPutTMVar (lockerTimeoutDisable locker) ()
dropLock :: LockID -> P2PHttpServerState -> IO () dropLock :: LockID -> PerRepoServerState -> IO ()
dropLock lckid st = do dropLock lckid st = do
v <- atomically $ do v <- atomically $ do
m <- takeTMVar (openLocks st) m <- takeTMVar (openLocks st)
@ -532,7 +536,7 @@ getAnnexWorkerPool a = startConcurrency transferStages $
Nothing -> giveup "Use -Jn or set annex.jobs to configure the number of worker threads." Nothing -> giveup "Use -Jn or set annex.jobs to configure the number of worker threads."
Just wp -> a wp Just wp -> a wp
inAnnexWorker :: P2PHttpServerState -> Annex a -> IO (Either SomeException a) inAnnexWorker :: PerRepoServerState -> Annex a -> IO (Either SomeException a)
inAnnexWorker st = inAnnexWorker' (annexWorkerPool st) inAnnexWorker st = inAnnexWorker' (annexWorkerPool st)
inAnnexWorker' :: AnnexWorkerPool -> Annex a -> IO (Either SomeException a) inAnnexWorker' :: AnnexWorkerPool -> Annex a -> IO (Either SomeException a)