implement proxy connection pool
removeOldestProxyConnectionPool will be innefficient the larger the pool is. A better data structure could be more efficient. Eg, make each value in the pool include the timestamp of its oldest element, then the oldest value can be found and modified, rather than rebuilding the whole Map. But, for pools of a few hundred items, this should be fine. It's O(n*n log n) or so. Also, when more than 1 connection with the same pool key exists, it's efficient even for larger pools, since removeOldestProxyConnectionPool is not needed. The default of 1 idle connection could perhaps be larger.. like the number of jobs? Otoh, it seems good to ramp up and down the number of connections, which does happen. With 1, there is at most one stale connection, which might cause a request to fail.
This commit is contained in:
parent
fb43b7ea3f
commit
d1faa13d6a
5 changed files with 114 additions and 38 deletions
|
@ -40,6 +40,7 @@ data Options = Options
|
||||||
, unauthReadOnlyOption :: Bool
|
, unauthReadOnlyOption :: Bool
|
||||||
, unauthAppendOnlyOption :: Bool
|
, unauthAppendOnlyOption :: Bool
|
||||||
, wideOpenOption :: Bool
|
, wideOpenOption :: Bool
|
||||||
|
, proxyConnectionsOption :: Maybe Integer
|
||||||
}
|
}
|
||||||
|
|
||||||
optParser :: CmdParamsDesc -> Parser Options
|
optParser :: CmdParamsDesc -> Parser Options
|
||||||
|
@ -84,10 +85,17 @@ optParser _ = Options
|
||||||
( long "wideopen"
|
( long "wideopen"
|
||||||
<> help "give unauthenticated users full read+write access"
|
<> help "give unauthenticated users full read+write access"
|
||||||
)
|
)
|
||||||
|
<*> optional (option auto
|
||||||
|
( long "proxyconnections" <> metavar paramNumber
|
||||||
|
<> help "maximum number of idle connections when proxying"
|
||||||
|
))
|
||||||
|
|
||||||
seek :: Options -> CommandSeek
|
seek :: Options -> CommandSeek
|
||||||
seek o = getAnnexWorkerPool $ \workerpool ->
|
seek o = getAnnexWorkerPool $ \workerpool ->
|
||||||
withP2PConnections workerpool $ \acquireconn -> liftIO $ do
|
withP2PConnections workerpool (fromMaybe 1 $ proxyConnectionsOption o)
|
||||||
|
(go workerpool)
|
||||||
|
where
|
||||||
|
go workerpool acquireconn = liftIO $ do
|
||||||
authenv <- getAuthEnv
|
authenv <- getAuthEnv
|
||||||
st <- mkP2PHttpServerState acquireconn workerpool $
|
st <- mkP2PHttpServerState acquireconn workerpool $
|
||||||
mkGetServerMode authenv o
|
mkGetServerMode authenv o
|
||||||
|
@ -100,7 +108,7 @@ seek o = getAnnexWorkerPool $ \workerpool ->
|
||||||
certfile (chainFileOption o) privatekeyfile
|
certfile (chainFileOption o) privatekeyfile
|
||||||
Warp.runTLS tlssettings settings (p2pHttpApp st)
|
Warp.runTLS tlssettings settings (p2pHttpApp st)
|
||||||
_ -> giveup "You must use both --certfile and --privatekeyfile options to enable HTTPS."
|
_ -> giveup "You must use both --certfile and --privatekeyfile options to enable HTTPS."
|
||||||
where
|
|
||||||
port = maybe
|
port = maybe
|
||||||
(fromIntegral defaultP2PHttpProtocolPort)
|
(fromIntegral defaultP2PHttpProtocolPort)
|
||||||
fromIntegral
|
fromIntegral
|
||||||
|
|
|
@ -10,6 +10,7 @@
|
||||||
{-# LANGUAGE BangPatterns #-}
|
{-# LANGUAGE BangPatterns #-}
|
||||||
{-# LANGUAGE OverloadedStrings #-}
|
{-# LANGUAGE OverloadedStrings #-}
|
||||||
{-# LANGUAGE FlexibleContexts #-}
|
{-# LANGUAGE FlexibleContexts #-}
|
||||||
|
{-# LANGUAGE TupleSections #-}
|
||||||
|
|
||||||
module P2P.Http.State where
|
module P2P.Http.State where
|
||||||
|
|
||||||
|
@ -33,10 +34,11 @@ import qualified P2P.Proxy as Proxy
|
||||||
import qualified Types.Remote as Remote
|
import qualified Types.Remote as Remote
|
||||||
|
|
||||||
import Servant
|
import Servant
|
||||||
import qualified Data.Map as M
|
import qualified Data.Map.Strict as M
|
||||||
import qualified Data.Set as S
|
import qualified Data.Set as S
|
||||||
import Control.Concurrent.Async
|
import Control.Concurrent.Async
|
||||||
import Control.Concurrent.STM
|
import Control.Concurrent.STM
|
||||||
|
import Data.Time.Clock.POSIX
|
||||||
|
|
||||||
data P2PHttpServerState = P2PHttpServerState
|
data P2PHttpServerState = P2PHttpServerState
|
||||||
{ acquireP2PConnection :: AcquireP2PConnection
|
{ acquireP2PConnection :: AcquireP2PConnection
|
||||||
|
@ -175,13 +177,17 @@ type AcquireP2PConnection
|
||||||
= ConnectionParams
|
= ConnectionParams
|
||||||
-> IO (Either ConnectionProblem P2PConnectionPair)
|
-> IO (Either ConnectionProblem P2PConnectionPair)
|
||||||
|
|
||||||
withP2PConnections :: AnnexWorkerPool -> (AcquireP2PConnection -> Annex a) -> Annex a
|
withP2PConnections
|
||||||
withP2PConnections workerpool a = do
|
:: AnnexWorkerPool
|
||||||
|
-> ProxyConnectionPoolSize
|
||||||
|
-> (AcquireP2PConnection -> Annex a)
|
||||||
|
-> Annex a
|
||||||
|
withP2PConnections workerpool proxyconnectionpoolsize a = do
|
||||||
myuuid <- getUUID
|
myuuid <- getUUID
|
||||||
reqv <- liftIO newEmptyTMVarIO
|
reqv <- liftIO newEmptyTMVarIO
|
||||||
relv <- liftIO newEmptyTMVarIO
|
relv <- liftIO newEmptyTMVarIO
|
||||||
endv <- liftIO newEmptyTMVarIO
|
endv <- liftIO newEmptyTMVarIO
|
||||||
proxypool <- liftIO $ newTMVarIO mempty
|
proxypool <- liftIO $ newTMVarIO (0, mempty)
|
||||||
asyncservicer <- liftIO $ async $
|
asyncservicer <- liftIO $ async $
|
||||||
servicer myuuid proxypool reqv relv endv
|
servicer myuuid proxypool reqv relv endv
|
||||||
let endit = do
|
let endit = do
|
||||||
|
@ -216,8 +222,8 @@ withP2PConnections workerpool a = do
|
||||||
| connectionServerUUID connparams == myuuid =
|
| connectionServerUUID connparams == myuuid =
|
||||||
localConnection relv connparams workerpool
|
localConnection relv connparams workerpool
|
||||||
| otherwise =
|
| otherwise =
|
||||||
atomically (getProxyConnectionFromPool proxypool connparams) >>= \case
|
atomically (getProxyConnectionPool proxypool connparams) >>= \case
|
||||||
Just conn -> proxyConnection relv connparams workerpool conn
|
Just conn -> proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool conn
|
||||||
Nothing -> checkcanproxy myuuid proxypool relv connparams
|
Nothing -> checkcanproxy myuuid proxypool relv connparams
|
||||||
|
|
||||||
checkcanproxy myuuid proxypool relv connparams =
|
checkcanproxy myuuid proxypool relv connparams =
|
||||||
|
@ -233,7 +239,7 @@ withP2PConnections workerpool a = do
|
||||||
(connectionBypass connparams)
|
(connectionBypass connparams)
|
||||||
proxyremote
|
proxyremote
|
||||||
>>= \case
|
>>= \case
|
||||||
Right conn -> proxyConnection relv connparams workerpool conn
|
Right conn -> proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool conn
|
||||||
Left ex -> return $ Left $
|
Left ex -> return $ Left $
|
||||||
ConnectionFailed $ show ex
|
ConnectionFailed $ show ex
|
||||||
Right (Right (Left clusteruuid)) ->
|
Right (Right (Left clusteruuid)) ->
|
||||||
|
@ -332,12 +338,14 @@ mkClientRunState connparams = do
|
||||||
mkRunState $ const $ Client prototvar
|
mkRunState $ const $ Client prototvar
|
||||||
|
|
||||||
proxyConnection
|
proxyConnection
|
||||||
:: TMVar (IO ())
|
:: ProxyConnectionPoolSize
|
||||||
|
-> TMVar (IO ())
|
||||||
-> ConnectionParams
|
-> ConnectionParams
|
||||||
-> AnnexWorkerPool
|
-> AnnexWorkerPool
|
||||||
|
-> TMVar ProxyConnectionPool
|
||||||
-> ProxyConnection
|
-> ProxyConnection
|
||||||
-> IO (Either ConnectionProblem P2PConnectionPair)
|
-> IO (Either ConnectionProblem P2PConnectionPair)
|
||||||
proxyConnection relv connparams workerpool proxyconn = do
|
proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool proxyconn = do
|
||||||
(clientconn, proxyfromclientconn) <-
|
(clientconn, proxyfromclientconn) <-
|
||||||
mkP2PConnectionPair connparams ("http client", "proxy")
|
mkP2PConnectionPair connparams ("http client", "proxy")
|
||||||
clientrunst <- mkClientRunState connparams
|
clientrunst <- mkClientRunState connparams
|
||||||
|
@ -371,9 +379,13 @@ proxyConnection relv connparams workerpool proxyconn = do
|
||||||
r <- liftIO $ wait asyncworker
|
r <- liftIO $ wait asyncworker
|
||||||
liftIO $ closeConnection proxyfromclientconn
|
liftIO $ closeConnection proxyfromclientconn
|
||||||
liftIO $ closeConnection clientconn
|
liftIO $ closeConnection clientconn
|
||||||
inAnnexWorker' workerpool $
|
if returntopool
|
||||||
Proxy.closeRemoteSide $
|
then liftIO $ do
|
||||||
proxyConnectionRemoteSide proxyconn
|
now <- getPOSIXTime
|
||||||
|
evicted <- atomically $ putProxyConnectionPool proxypool proxyconnectionpoolsize connparams $
|
||||||
|
proxyconn { proxyConnectionLastUsed = now }
|
||||||
|
maybe noop closeproxyconnection evicted
|
||||||
|
else closeproxyconnection proxyconn
|
||||||
either throwM return r
|
either throwM return r
|
||||||
|
|
||||||
return $ Right $ P2PConnectionPair
|
return $ Right $ P2PConnectionPair
|
||||||
|
@ -385,13 +397,19 @@ proxyConnection relv connparams workerpool proxyconn = do
|
||||||
}
|
}
|
||||||
where
|
where
|
||||||
protoerrhandler cont a = a >>= \case
|
protoerrhandler cont a = a >>= \case
|
||||||
Left err ->
|
Left _ ->
|
||||||
Proxy.closeRemoteSide $
|
Proxy.closeRemoteSide $
|
||||||
proxyConnectionRemoteSide proxyconn
|
proxyConnectionRemoteSide proxyconn
|
||||||
Right v -> cont v
|
Right v -> cont v
|
||||||
|
|
||||||
proxydone = return ()
|
proxydone = return ()
|
||||||
|
|
||||||
requestcomplete () = return ()
|
requestcomplete () = return ()
|
||||||
|
|
||||||
|
closeproxyconnection =
|
||||||
|
void . inAnnexWorker' workerpool .
|
||||||
|
Proxy.closeRemoteSide . proxyConnectionRemoteSide
|
||||||
|
|
||||||
data Locker = Locker
|
data Locker = Locker
|
||||||
{ lockerThread :: Async ()
|
{ lockerThread :: Async ()
|
||||||
, lockerVar :: TMVar Bool
|
, lockerVar :: TMVar Bool
|
||||||
|
@ -506,7 +524,9 @@ inAnnexWorker' poolv annexaction = do
|
||||||
data ProxyConnection = ProxyConnection
|
data ProxyConnection = ProxyConnection
|
||||||
{ proxyConnectionRemoteUUID :: UUID
|
{ proxyConnectionRemoteUUID :: UUID
|
||||||
, proxyConnectionRemoteSide :: Proxy.RemoteSide
|
, proxyConnectionRemoteSide :: Proxy.RemoteSide
|
||||||
|
, proxyConnectionLastUsed :: POSIXTime
|
||||||
}
|
}
|
||||||
|
deriving (Show)
|
||||||
|
|
||||||
openProxyConnectionToRemote
|
openProxyConnectionToRemote
|
||||||
:: AnnexWorkerPool
|
:: AnnexWorkerPool
|
||||||
|
@ -517,36 +537,74 @@ openProxyConnectionToRemote
|
||||||
openProxyConnectionToRemote workerpool clientmaxversion bypass remote =
|
openProxyConnectionToRemote workerpool clientmaxversion bypass remote =
|
||||||
inAnnexWorker' workerpool (proxyRemoteSide clientmaxversion bypass' remote) >>= \case
|
inAnnexWorker' workerpool (proxyRemoteSide clientmaxversion bypass' remote) >>= \case
|
||||||
Left ex -> return (Left ex)
|
Left ex -> return (Left ex)
|
||||||
Right remoteside -> return $ Right $
|
Right remoteside -> do
|
||||||
ProxyConnection (Remote.uuid remote) remoteside
|
now <- getPOSIXTime
|
||||||
|
return $ Right $
|
||||||
|
ProxyConnection (Remote.uuid remote) remoteside now
|
||||||
where
|
where
|
||||||
bypass' = P2P.Bypass (S.fromList bypass)
|
bypass' = P2P.Bypass (S.fromList bypass)
|
||||||
|
|
||||||
openProxyConnectionToCluster :: ClusterUUID -> IO ProxyConnection
|
openProxyConnectionToCluster :: ClusterUUID -> IO ProxyConnection
|
||||||
openProxyConnectionToCluster cu = error "XXX" -- TODO
|
openProxyConnectionToCluster cu = error "XXX" -- TODO
|
||||||
|
|
||||||
type ProxyConnectionPool =
|
type ProxyConnectionPool = (Integer, M.Map ProxyConnectionPoolKey [ProxyConnection])
|
||||||
M.Map (UUID, UUID, P2P.ProtocolVersion) [ProxyConnection]
|
|
||||||
|
|
||||||
getProxyConnectionFromPool
|
type ProxyConnectionPoolSize = Integer
|
||||||
|
|
||||||
|
-- Returns any older ProxyConnection that was evicted from the pool.
|
||||||
|
putProxyConnectionPool
|
||||||
|
:: TMVar ProxyConnectionPool
|
||||||
|
-> ProxyConnectionPoolSize
|
||||||
|
-> ConnectionParams
|
||||||
|
-> ProxyConnection
|
||||||
|
-> STM (Maybe ProxyConnection)
|
||||||
|
putProxyConnectionPool proxypool maxsz connparams conn = do
|
||||||
|
(sz, m) <- takeTMVar proxypool
|
||||||
|
let ((sz', m'), evicted) = case M.lookup k m of
|
||||||
|
Nothing -> ((succ sz, M.insert k [conn] m), Nothing)
|
||||||
|
Just [] -> ((succ sz, M.insert k [conn] m), Nothing)
|
||||||
|
Just cs -> if sz >= maxsz
|
||||||
|
then ((sz, M.insert k (conn : dropFromEnd 1 cs) m), lastMaybe cs)
|
||||||
|
else ((sz, M.insert k (conn : cs) m), Nothing)
|
||||||
|
let ((sz'', m''), evicted') = if sz' > maxsz
|
||||||
|
then removeOldestProxyConnectionPool (sz', m')
|
||||||
|
else ((sz', m'), Nothing)
|
||||||
|
putTMVar proxypool (sz'', m'')
|
||||||
|
return (evicted <|> evicted')
|
||||||
|
where
|
||||||
|
k = proxyConnectionPoolKey connparams
|
||||||
|
|
||||||
|
removeOldestProxyConnectionPool :: ProxyConnectionPool -> (ProxyConnectionPool, Maybe ProxyConnection)
|
||||||
|
removeOldestProxyConnectionPool (sz, m) =
|
||||||
|
((pred sz, m'), snd <$> headMaybe l)
|
||||||
|
where
|
||||||
|
m' = M.fromListWith (++) $ map (\(k', v) -> (k', [v])) (drop 1 l)
|
||||||
|
l = sortOn (proxyConnectionLastUsed . snd) $
|
||||||
|
concatMap (\(k', pl) -> map (k', ) pl) $
|
||||||
|
M.toList m
|
||||||
|
|
||||||
|
getProxyConnectionPool
|
||||||
:: TMVar ProxyConnectionPool
|
:: TMVar ProxyConnectionPool
|
||||||
-> ConnectionParams
|
-> ConnectionParams
|
||||||
-> STM (Maybe ProxyConnection)
|
-> STM (Maybe ProxyConnection)
|
||||||
getProxyConnectionFromPool proxypool connparams = do
|
getProxyConnectionPool proxypool connparams = do
|
||||||
m <- takeTMVar proxypool
|
(sz, m) <- takeTMVar proxypool
|
||||||
case M.lookup k m of
|
case M.lookup k m of
|
||||||
Nothing -> do
|
|
||||||
putTMVar proxypool m
|
|
||||||
return Nothing
|
|
||||||
Just [] -> do
|
|
||||||
putTMVar proxypool $ M.insert k [] m
|
|
||||||
return Nothing
|
|
||||||
Just (c:cs) -> do
|
Just (c:cs) -> do
|
||||||
putTMVar proxypool $ M.insert k cs m
|
putTMVar proxypool (sz-1, M.insert k cs m)
|
||||||
return (Just c)
|
return (Just c)
|
||||||
|
_ -> do
|
||||||
|
putTMVar proxypool (sz, m)
|
||||||
|
return Nothing
|
||||||
where
|
where
|
||||||
k =
|
k = proxyConnectionPoolKey connparams
|
||||||
( connectionServerUUID connparams
|
|
||||||
, connectionClientUUID connparams
|
type ProxyConnectionPoolKey = (UUID, UUID, [UUID], P2P.ProtocolVersion)
|
||||||
, connectionProtocolVersion connparams
|
|
||||||
)
|
proxyConnectionPoolKey :: ConnectionParams -> ProxyConnectionPoolKey
|
||||||
|
proxyConnectionPoolKey connparams =
|
||||||
|
( connectionServerUUID connparams
|
||||||
|
, connectionClientUUID connparams
|
||||||
|
, connectionBypass connparams
|
||||||
|
, connectionProtocolVersion connparams
|
||||||
|
)
|
||||||
|
|
|
@ -45,6 +45,9 @@ data RemoteSide = RemoteSide
|
||||||
, remoteSideId :: RemoteSideId
|
, remoteSideId :: RemoteSideId
|
||||||
}
|
}
|
||||||
|
|
||||||
|
instance Show RemoteSide where
|
||||||
|
show rs = show (remote rs)
|
||||||
|
|
||||||
mkRemoteSide :: Remote -> Annex (Maybe (RunState, P2PConnection, ProtoCloser)) -> Annex RemoteSide
|
mkRemoteSide :: Remote -> Annex (Maybe (RunState, P2PConnection, ProtoCloser)) -> Annex RemoteSide
|
||||||
mkRemoteSide r remoteconnect = RemoteSide
|
mkRemoteSide r remoteconnect = RemoteSide
|
||||||
<$> pure r
|
<$> pure r
|
||||||
|
|
|
@ -39,6 +39,14 @@ convenient way to download the content of any key, by using the path
|
||||||
|
|
||||||
A good choice is one worker per CPU core: `--jobs=cpus`
|
A good choice is one worker per CPU core: `--jobs=cpus`
|
||||||
|
|
||||||
|
* `--proxyconnections=N`
|
||||||
|
|
||||||
|
When is command is run in a repository that is configured to act as a
|
||||||
|
proxy for some of its remotes, this is the maximum number of idle
|
||||||
|
connections to keep open to proxied remotes.
|
||||||
|
|
||||||
|
The default is 1.
|
||||||
|
|
||||||
* `--port=N`
|
* `--port=N`
|
||||||
|
|
||||||
Port to listen on. The default is port 9417, which is the default
|
Port to listen on. The default is port 9417, which is the default
|
||||||
|
@ -112,6 +120,8 @@ git-http-backend(1)
|
||||||
|
|
||||||
[[git-annex-shell]](1)
|
[[git-annex-shell]](1)
|
||||||
|
|
||||||
|
[[git-annex-updateproxy]](1)
|
||||||
|
|
||||||
<https://git-annex.branchable.com/design/p2p_protocol_over_http/>
|
<https://git-annex.branchable.com/design/p2p_protocol_over_http/>
|
||||||
|
|
||||||
# AUTHOR
|
# AUTHOR
|
||||||
|
|
|
@ -32,9 +32,6 @@ Planned schedule of work:
|
||||||
|
|
||||||
* test http server proxying with special remotes
|
* test http server proxying with special remotes
|
||||||
|
|
||||||
* http server proxying needs to reuse connections to special remotes,
|
|
||||||
keeping a pool of open ones. Question: How many to keep in the pool?
|
|
||||||
|
|
||||||
* Make http server support clusters.
|
* Make http server support clusters.
|
||||||
|
|
||||||
* Support proxying to git remotes using annex+http urls.
|
* Support proxying to git remotes using annex+http urls.
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue