cluster support in http API server

Wired it up and it seems to basically work, although the test suite is
not fully passing.

Note that --jobs currently gets multiplied by the number of nodes in the
cluster, which is probably not good.

Annex/Cluster.hs

@@ -58,9 +58,9 @@ proxyCluster clusteruuid proxydone servermode clientside protoerrhandler = do
 			(withclientbypass protocolversion) (protoerrhandler noop)
 
 	withclientbypass protocolversion (bypassuuids, othermsg) = do
-		(selectnode, closenodes) <- clusterProxySelector clusteruuid
+		(selectnode, closenodes, concurrencyconfig) <-
-			protocolversion bypassuuids
+			clusterProxySelector clusteruuid
-		concurrencyconfig <- getConcurrencyConfig
+				protocolversion bypassuuids
 		proxystate <- liftIO mkProxyState
 		let proxyparams = ProxyParams
 			{ proxyMethods = mkProxyMethods
@@ -75,7 +75,11 @@ proxyCluster clusteruuid proxydone servermode clientside protoerrhandler = do
 		proxy proxydone proxyparams othermsg
 			(protoerrhandler closenodes)
 
-clusterProxySelector :: ClusterUUID -> ProtocolVersion -> Bypass -> Annex (ProxySelector, Annex ())
+clusterProxySelector
+	:: ClusterUUID
+	-> ProtocolVersion
+	-> Bypass
+	-> Annex (ProxySelector, Annex (), ConcurrencyConfig)
 clusterProxySelector clusteruuid protocolversion (Bypass bypass) = do
 	nodeuuids <- (fromMaybe S.empty . M.lookup clusteruuid . clusterUUIDs)
 		<$> getClusters
@@ -116,7 +120,8 @@ clusterProxySelector clusteruuid protocolversion (Bypass bypass) = do
 		-- proxied to the client.
 		, proxyLOCKCONTENT = const (pure Nothing)
 		}
-	return (proxyselector, closenodes)
+	concurrencyconfig <- getConcurrencyConfig
+	return (proxyselector, closenodes, concurrencyconfig)
   where
 	-- Nodes of the cluster have remote.name.annex-cluster-node
 	-- containing its name. 

P2P/Http/State.hs

@@ -30,6 +30,7 @@ import CmdLine.Action (startConcurrency)
 import Utility.ThreadScheduler
 import Utility.HumanTime
 import Annex.Proxy
+import Annex.Cluster
 import qualified P2P.Proxy as Proxy
 import qualified Types.Remote as Remote
 
@@ -233,21 +234,21 @@ withP2PConnections workerpool proxyconnectionpoolsize a = do
 			Right (Left reason) -> return $ Left $
 				ConnectionFailed $ 
 					fromMaybe "unknown uuid" reason
-			Right (Right (Right proxyremote)) ->
+			Right (Right (Right proxyremote)) -> proxyconnection $
 				openProxyConnectionToRemote workerpool
 					(connectionProtocolVersion connparams)
-					(connectionBypass connparams)
+					bypass proxyremote
-					proxyremote
+			Right (Right (Left clusteruuid)) -> proxyconnection $
-					>>= \case
+				openProxyConnectionToCluster workerpool
-						Right conn -> proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool conn
+					(connectionProtocolVersion connparams)
-						Left ex -> return $ Left $
+					bypass clusteruuid
-							ConnectionFailed $ show ex
+			Left ex -> return $ Left $
-			Right (Right (Left clusteruuid)) ->
+				ConnectionFailed $ show ex
-				undefined -- XXX todo
+	  where
-				{-
+		bypass = P2P.Bypass $ S.fromList $ connectionBypass connparams
-				openProxyConnectionToCluster clusteruuid
+		proxyconnection openconn = openconn >>= \case
-					>>= proxyConnection clusteruuid relv connparams workerpool
+			Right conn -> proxyConnection proxyconnectionpoolsize
-				-}
+				relv connparams workerpool proxypool conn
 			Left ex -> return $ Left $
 				ConnectionFailed $ show ex
 
@@ -353,16 +354,14 @@ proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool pro
 	asyncworker <- async $
 		inAnnexWorker' workerpool $ do
 			proxystate <- liftIO Proxy.mkProxyState
-			concurrencyconfig <- Proxy.noConcurrencyConfig
 			let proxyparams = Proxy.ProxyParams
 				{ Proxy.proxyMethods = mkProxyMethods
 				, Proxy.proxyState = proxystate
 				, Proxy.proxyServerMode = connectionServerMode connparams
 				, Proxy.proxyClientSide = Proxy.ClientSide proxyfromclientrunst proxyfromclientconn
 				, Proxy.proxyUUID = proxyConnectionRemoteUUID proxyconn
-				, Proxy.proxySelector = Proxy.singleProxySelector $
+				, Proxy.proxySelector = proxyConnectionSelector proxyconn
-					proxyConnectionRemoteSide proxyconn
+				, Proxy.proxyConcurrencyConfig = proxyConnectionConcurrency proxyconn
-				, Proxy.proxyConcurrencyConfig = concurrencyconfig
 				, Proxy.proxyClientProtocolVersion = connectionProtocolVersion connparams
 				}
 			let proxy mrequestmessage = case mrequestmessage of
@@ -397,9 +396,7 @@ proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool pro
 		}
   where
 	protoerrhandler cont a = a >>= \case
-		Left _ -> 
+		Left _ -> proxyConnectionCloser proxyconn
-			Proxy.closeRemoteSide $
-				proxyConnectionRemoteSide proxyconn
 		Right v -> cont v
 	
 	proxydone = return ()
@@ -407,8 +404,7 @@ proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool pro
 	requestcomplete () = return ()
 	
 	closeproxyconnection = 
-		void . inAnnexWorker' workerpool .
+		void . inAnnexWorker' workerpool . proxyConnectionCloser
-			Proxy.closeRemoteSide . proxyConnectionRemoteSide
 
 data Locker = Locker
 	{ lockerThread :: Async ()
@@ -523,29 +519,56 @@ inAnnexWorker' poolv annexaction = do
 
 data ProxyConnection = ProxyConnection
 	{ proxyConnectionRemoteUUID :: UUID
-	, proxyConnectionRemoteSide :: Proxy.RemoteSide
+	, proxyConnectionSelector :: Proxy.ProxySelector
+	, proxyConnectionCloser :: Annex ()
+	, proxyConnectionConcurrency :: Proxy.ConcurrencyConfig
 	, proxyConnectionLastUsed :: POSIXTime
 	}
-	deriving (Show)
+
+instance Show ProxyConnection where
+	show pc = unwords
+		[ "ProxyConnection"
+		, show (proxyConnectionRemoteUUID pc)
+		, show (proxyConnectionLastUsed pc)
+		]
+
+openedProxyConnection
+	:: UUID
+	-> Proxy.ProxySelector
+	-> Annex ()
+	-> Proxy.ConcurrencyConfig
+	-> IO ProxyConnection
+openedProxyConnection u selector closer concurrency = do
+	now <- getPOSIXTime
+	return $ ProxyConnection u selector closer concurrency now
 
 openProxyConnectionToRemote
 	:: AnnexWorkerPool
 	-> P2P.ProtocolVersion
-	-> [UUID]
+	-> P2P.Bypass
 	-> Remote
 	-> IO (Either SomeException ProxyConnection)
 openProxyConnectionToRemote workerpool clientmaxversion bypass remote =
-	inAnnexWorker' workerpool (proxyRemoteSide clientmaxversion bypass' remote) >>= \case
+	inAnnexWorker' workerpool $ do
-		Left ex -> return (Left ex)
+		remoteside <- proxyRemoteSide clientmaxversion bypass remote
-		Right remoteside -> do
+		concurrencyconfig <- Proxy.noConcurrencyConfig
-			now <- getPOSIXTime
+		liftIO $ openedProxyConnection (Remote.uuid remote)
-			return $ Right $
+			(Proxy.singleProxySelector remoteside)
-				ProxyConnection (Remote.uuid remote) remoteside now
+			(Proxy.closeRemoteSide remoteside)
-  where
+			concurrencyconfig
-	bypass' = P2P.Bypass (S.fromList bypass)
 
-openProxyConnectionToCluster :: ClusterUUID -> IO ProxyConnection
+openProxyConnectionToCluster
-openProxyConnectionToCluster cu = error "XXX" -- TODO
+	:: AnnexWorkerPool
+	-> P2P.ProtocolVersion
+	-> P2P.Bypass
+	-> ClusterUUID
+	-> IO (Either SomeException ProxyConnection)
+openProxyConnectionToCluster workerpool clientmaxversion bypass clusteruuid =
+	inAnnexWorker' workerpool $ do
+		(proxyselector, closenodes, concurrencyconfig) <-
+			clusterProxySelector clusteruuid clientmaxversion bypass
+		liftIO $ openedProxyConnection (fromClusterUUID clusteruuid)
+			proxyselector closenodes concurrencyconfig
 
 type ProxyConnectionPool = (Integer, M.Map ProxyConnectionPoolKey [ProxyConnection])
 

doc/todo/git-annex_proxies.mdwn

@@ -28,7 +28,9 @@ Planned schedule of work:
 
 ## work notes
 
-* Make http server support clusters.
+* http proxying for a local git remote seems to probably not work
+
+* git-annex testremote cluster
 
 * Support proxying to git remotes using annex+http urls.
   (Current documentation says proxying only works with ssh remotes,