p2phttp support --jobs with --directory

--jobs is usually an Annex option setter, but --directory runs in IO, so
would not have that available. So instead moved the option parser into
the command's Options.

CmdLine/GitAnnex/Options.hs

@@ -511,15 +511,19 @@ jsonProgressOption =
 -- action in `allowConcurrentOutput`.
 jobsOption :: [AnnexOption]
 jobsOption = 
-	[ annexOption (setAnnexState . setConcurrency . ConcurrencyCmdLine) $ 
+	[ annexOption (setAnnexState . setConcurrency . ConcurrencyCmdLine)
-		option (maybeReader parseConcurrency)
+		jobsOptionParser
-			( long "jobs" <> short 'J' 
-			<> metavar (paramNumber `paramOr` "cpus")
-			<> help "enable concurrent jobs"
-			<> hidden
-			)
 	]
 
+jobsOptionParser :: Parser Concurrency
+jobsOptionParser = 
+	option (maybeReader parseConcurrency)
+		( long "jobs" <> short 'J' 
+		<> metavar (paramNumber `paramOr` "cpus")
+		<> help "enable concurrent jobs"
+		<> hidden
+		)
+
 timeLimitOption :: [AnnexOption]
 timeLimitOption = 
 	[ annexOption settimelimit $ option (eitherReader parseDuration)

Command/P2PHttp.hs

@@ -11,7 +11,7 @@
 
 module Command.P2PHttp where
 
-import Command
+import Command hiding (jobsOption)
 import P2P.Http.Server
 import P2P.Http.Url
 import qualified P2P.Protocol as P2P
@@ -20,6 +20,7 @@ import Annex.UUID
 import qualified Git
 import qualified Git.Construct
 import qualified Annex
+import Types.Concurrency
 
 import Servant
 import qualified Network.Wai.Handler.Warp as Warp
@@ -29,12 +30,11 @@ import qualified Data.Map as M
 import Data.String
 
 cmd :: Command
-cmd = withAnnexOptions [jobsOption] $
+cmd = noMessages $ dontCheck repoExists $ 
-	noMessages $ dontCheck repoExists $ 
+	noRepo (startIO <$$> optParser) $
-		noRepo (startIO <$$> optParser) $
+		command "p2phttp" SectionPlumbing
-			command "p2phttp" SectionPlumbing
+			"communicate in P2P protocol over http"
-				"communicate in P2P protocol over http"
+			paramNothing (startAnnex <$$> optParser)
-				paramNothing (startAnnex <$$> optParser)
 
 data Options = Options
 	{ portOption :: Maybe PortNumber
@@ -49,6 +49,7 @@ data Options = Options
 	, unauthNoLockingOption :: Bool
 	, wideOpenOption :: Bool
 	, proxyConnectionsOption :: Maybe Integer
+	, jobsOption :: Maybe Concurrency
 	, clusterJobsOption :: Maybe Int
 	, directoryOption :: [FilePath]
 	}
@@ -103,6 +104,7 @@ optParser _ = Options
 		( long "proxyconnections" <> metavar paramNumber
 		<> help "maximum number of idle connections when proxying"
 		))
+	<*> optional jobsOptionParser
 	<*> optional (option auto
 		( long "clusterjobs" <> metavar paramNumber
 		<> help "number of concurrent node accesses per connection"
@@ -124,8 +126,6 @@ startAnnex o
 		)
 	| otherwise = liftIO $ startIO o
 
--- TODO --jobs option only available to startAnnex, not here, need
--- to parse it into Options for this command.
 startIO :: Options -> IO ()
 startIO o
 	| null (directoryOption o) = 
@@ -162,7 +162,7 @@ runServer o mst = go `finally` serverShutdownCleanup mst
 
 mkServerState :: Options -> M.Map Auth P2P.ServerMode -> Annex P2PHttpServerState
 mkServerState o authenv = 
-	getAnnexWorkerPool $
+	withAnnexWorkerPool (jobsOption o) $
 		mkP2PHttpServerState
 			(mkGetServerMode authenv o)
 			(fromMaybe 1 $ proxyConnectionsOption o)

P2P/Http/State.hs

@@ -26,6 +26,8 @@ import Types.NumCopies
 import Types.WorkerPool
 import Annex.WorkerPool
 import Annex.BranchState
+import Annex.Concurrent
+import Types.Concurrency
 import Types.Cluster
 import CmdLine.Action (startConcurrency)
 import Utility.ThreadScheduler
@@ -551,11 +553,13 @@ dropLock lckid st = do
 		Nothing -> return ()
 		Just locker -> wait (lockerThread locker)
 
-getAnnexWorkerPool :: (AnnexWorkerPool -> Annex a) -> Annex a
+withAnnexWorkerPool :: (Maybe Concurrency) -> (AnnexWorkerPool -> Annex a) -> Annex a
-getAnnexWorkerPool a = startConcurrency transferStages $
+withAnnexWorkerPool mc a = do
-	Annex.getState Annex.workers >>= \case
+	maybe noop (setConcurrency . ConcurrencyCmdLine) mc
-		Nothing -> giveup "Use -Jn or set annex.jobs to configure the number of worker threads."
+	startConcurrency transferStages $
-		Just wp -> a wp
+		Annex.getState Annex.workers >>= \case
+			Nothing -> giveup "Use -Jn or set annex.jobs to configure the number of worker threads."
+			Just wp -> a wp
 
 inAnnexWorker :: PerRepoServerState -> Annex a -> IO (Either SomeException a)
 inAnnexWorker st = inAnnexWorker' (annexWorkerPool st)