packages feed

git-annex 10.20251029 → 10.20251114

raw patch · 14 files changed

+310/−167 lines, 14 files

Files

Annex/Concurrent.hs view
@@ -97,12 +97,16 @@ 			setConcurrency (ConcurrencyCmdLine (Concurrent 1)) 			Annex.getState id 		_ -> return st-	return $ st'-		-- each thread has its own repoqueue-		{ Annex.repoqueue = Nothing-		-- no errors from this thread yet-		, Annex.errcounter = 0-		}+	return $ dupState' st'++{- Should only be used when concurrency is enabled. -}+dupState' :: AnnexState -> AnnexState+dupState' st = st+	-- each thread has its own repoqueue+	{ Annex.repoqueue = Nothing+	-- no errors from this thread yet+	, Annex.errcounter = 0+	}  {- Merges the passed AnnexState into the current Annex state. -} mergeState :: AnnexState -> Annex ()
Annex/Content.hs view
@@ -147,9 +147,7 @@ #ifndef mingw32_HOST_OS 	lock retention _ (Just lockfile) =  		( posixLocker tryLockShared lockfile >>= \case-			Just lck -> do-				writeretention retention-				return (Just lck)+			Just lck -> writeretention retention lck 			Nothing -> return Nothing 		, Just $ posixLocker tryLockExclusive lockfile >>= \case 			Just lck -> do@@ -166,9 +164,7 @@ 		let (locker, postunlock) = winLocker lockShared obj lckf 		in  			( locker >>= \case-				Just lck -> do-					writeretention retention-					return (Just lck)+				Just lck -> writeretention retention lck 				Nothing -> return Nothing 			, Just $ \lckfile -> do 				maybe noop (\pu -> pu lckfile) postunlock@@ -191,9 +187,20 @@ 					_ -> noop #endif 	-	writeretention Nothing = noop-	writeretention (Just (rt, retentionts)) = -		writeContentRetentionTimestamp key rt retentionts+	writeretention Nothing lck = return (Just lck)+	writeretention (Just (rt, retentionts)) lck = +		-- While writeContentRetentionTimestamp recovers from+		-- races itself, it's still possible for it to fail,+		-- eg if the object directory is no longer writable due+		-- to a sudden permissions change, or the disk going+		-- readonly. In that case, the lock is dropped to avoid+		-- a dangling lock. The lock file is necessarily left on+		-- disk.+		tryNonAsync (writeContentRetentionTimestamp key rt retentionts) >>= \case+			Right () -> return (Just lck)+			Left _ -> do+				liftIO $ dropLock lck+				return Nothing 	 	-- When this is called, an exclusive lock has been taken, so no other 	-- processes can be writing to the retention time stamp file.@@ -1078,14 +1085,24 @@ writeContentRetentionTimestamp :: Key -> OsPath -> POSIXTime -> Annex () writeContentRetentionTimestamp key rt t = do 	lckfile <- calcRepo (gitAnnexContentRetentionTimestampLock key)-	modifyContentDirWhenExists lckfile $ bracket (lock lckfile) unlock $ \_ ->-		readContentRetentionTimestamp rt >>= \case-			Just ts | ts >= t -> return ()-			_ -> replaceFile (const noop) rt $ \tmp ->-				liftIO $ writeFileString tmp $ show t+	modifyContentDirWhenExists lckfile $ epermretry $ +		bracket (lock lckfile) unlock $ \_ ->+			readContentRetentionTimestamp rt >>= \case+				Just ts | ts >= t -> return ()+				_ -> replaceFile (const noop) rt $ \tmp ->+					liftIO $ writeFileString tmp $ show t   where 	lock = takeExclusiveLock 	unlock = liftIO . dropLock++	-- In a race, the directory can get frozen again before the file is+	-- replaced, resulting in permission denied. Retrying will recover+	-- from that. Note that thawContentDir will throw an exception if+	-- it's no longer possible to thaw the content directory, which+	-- avoids this ever being an infinite loop.+	epermretry a = flip catchPermissionDenied a $ \_ -> do+		thawContentDir rt+		epermretry a  {- Does not need locking because the file is written atomically. -} readContentRetentionTimestamp :: OsPath -> Annex (Maybe POSIXTime)
Assistant.hs view
@@ -101,7 +101,7 @@ 		else do 			git_annex <- fromOsPath <$> liftIO programPath 			ps <- gitAnnexDaemonizeParams-			start (Utility.Daemon.daemonize git_annex ps logfd (Just pidfile) False) Nothing+			start (Utility.Daemon.daemonize git_annex ps (Just logfd) (Just pidfile) False) Nothing #else 	-- Windows doesn't daemonize, but does redirect output to the 	-- log file. The only way to do so is to restart the program.
CHANGELOG view
@@ -1,3 +1,17 @@+git-annex (10.20251114) upstream; urgency=medium++  * p2p --pair: Fix to work with external P2P networks.+  * p2phttp: Significant robustness fixes for bugs that caused the +    server to stall.+  * p2phttp: Fix a file descriptor leak.+  * p2phttp: Added the --lockedfiles option.+  * dropunused: Run the annex.secure-erase-command +    (or .git/hooks/secure-erase-annex) when deleting+    temp and bad object files.+  * remotedaemon: Avoid crashing when run with --debug.++ -- Joey Hess <id@joeyh.name>  Fri, 14 Nov 2025 12:46:45 -0400+ git-annex (10.20251029) upstream; urgency=medium    * Support ssh remotes with '#' and '?' in the path to the repository,
Command/DropUnused.hs view
@@ -1,6 +1,6 @@ {- git-annex command  -- - Copyright 2010,2012,2018 Joey Hess <id@joeyh.name>+ - Copyright 2010-2025 Joey Hess <id@joeyh.name>  -  - Licensed under the GNU AGPL version 3 or higher.  -}@@ -17,6 +17,7 @@ import Command.Unused (withUnusedMaps, UnusedMaps(..), startUnused) import Annex.NumCopies import Annex.Content+import Annex.Content.LowLevel  cmd :: Command cmd = withAnnexOptions [jobsOption, jsonOptions] $@@ -79,5 +80,7 @@ performOther :: (Key -> Git.Repo -> OsPath) -> Key -> CommandPerform performOther filespec key = do 	f <- fromRepo $ filespec key-	pruneTmpWorkDirBefore f (liftIO . removeWhenExistsWith removeFile)+	pruneTmpWorkDirBefore f $ \f' -> do+		secureErase f'+		liftIO $ removeWhenExistsWith removeFile f' 	next $ return True
Command/P2P.hs view
@@ -263,7 +263,7 @@ 						Left _e -> return ReceiveFailed 						Right ls -> maybe  							(return ReceiveFailed)-							(finishPairing 100 remotename ourhalf)+							(finishPairing 100 remotename ourhalf ouraddrs) 							(deserializePairData ls)  -- | Allow the peer we're pairing with to authenticate to us,@@ -276,8 +276,8 @@ -- Since we're racing the peer as they do the same, the first try is likely -- to fail to authenticate. Can retry any number of times, to avoid the -- users needing to redo the whole process.-finishPairing :: Int -> RemoteName -> HalfAuthToken -> PairData -> Annex PairingResult-finishPairing retries remotename (HalfAuthToken ourhalf) (PairData (HalfAuthToken theirhalf) theiraddrs) = do+finishPairing :: Int -> RemoteName -> HalfAuthToken -> [P2PAddress] -> PairData -> Annex PairingResult+finishPairing retries remotename (HalfAuthToken ourhalf) ouraddrs (PairData (HalfAuthToken theirhalf) theiraddrs) = do 	case (toAuthToken (ourhalf <> theirhalf), toAuthToken (theirhalf <> ourhalf)) of 		(Just ourauthtoken, Just theirauthtoken) -> do 			liftIO $ putStrLn $ "Successfully exchanged pairing data. Connecting to " ++ remotename ++  "..."@@ -289,9 +289,9 @@ 		liftIO $ threadDelaySeconds (Seconds 2) 		liftIO $ putStrLn $ "Unable to connect to " ++ remotename ++ ". Retrying..." 		go (n-1) theiraddrs theirauthtoken ourauthtoken-	go n (addr:rest) theirauthtoken ourauthtoken = do-		storeP2PAuthToken addr ourauthtoken-		r <- setupLink remotename (P2PAddressAuth addr theirauthtoken)+	go n (theiraddr:rest) theirauthtoken ourauthtoken = do+		forM_ ouraddrs $ \ouraddr -> storeP2PAuthToken ouraddr ourauthtoken+		r <- setupLink remotename (P2PAddressAuth theiraddr theirauthtoken) 		case r of 			LinkSuccess -> return PairSuccess 			_ -> go n rest theirauthtoken ourauthtoken
Command/P2PHttp.hs view
@@ -57,6 +57,7 @@ 	, proxyConnectionsOption :: Maybe Integer 	, jobsOption :: Maybe Concurrency 	, clusterJobsOption :: Maybe Int+	, lockedFilesOption :: Maybe Integer 	, directoryOption :: [FilePath] 	} @@ -119,6 +120,10 @@ 		( long "clusterjobs" <> metavar paramNumber 		<> help "number of concurrent node accesses per connection" 		))+	<*> optional (option auto+		( long "lockedfiles" <> metavar paramNumber+		<> help "number of content files that can be locked"+		)) 	<*> many (strOption 		( long "directory" <> metavar paramPath 		<> help "serve repositories in subdirectories of a directory"@@ -128,8 +133,10 @@ startAnnex o 	| null (directoryOption o) = ifM ((/=) NoUUID <$> getUUID) 		( do+			lockedfilesqsem <- liftIO $ +				mkLockedFilesQSem (lockedFilesOption o) 			authenv <- liftIO getAuthEnv-			st <- mkServerState o authenv+			st <- mkServerState o authenv lockedfilesqsem 			liftIO $ runServer o st 		-- Run in a git repository that is not a git-annex repository. 		, liftIO $ startIO o @@ -146,20 +153,21 @@ 		runServer o st   where 	mkst authenv oldst = do+		lockedfilesqsem <- mkLockedFilesQSem (lockedFilesOption o) 		repos <- findRepos o 		sts <- forM repos $ \r -> do 			strd <- Annex.new r-			Annex.eval strd (mkstannex authenv oldst)+			Annex.eval strd (mkstannex authenv oldst lockedfilesqsem) 		return (mconcat sts) 			{ updateRepos = updaterepos authenv 			} 	-	mkstannex authenv oldst = do+	mkstannex authenv oldst lockedfilesqsem = do 		u <- getUUID 		if u == NoUUID 			then return mempty 			else case M.lookup u (servedRepos oldst) of-				Nothing -> mkServerState o authenv+				Nothing -> mkServerState o authenv lockedfilesqsem 				Just old -> return $ P2PHttpServerState 					{ servedRepos = M.singleton u old 					, serverShutdownCleanup = mempty@@ -213,14 +221,15 @@ 		Socket.listen sock Socket.maxListenQueue 		return sock -mkServerState :: Options -> M.Map Auth P2P.ServerMode -> Annex P2PHttpServerState-mkServerState o authenv = +mkServerState :: Options -> M.Map Auth P2P.ServerMode -> LockedFilesQSem -> Annex P2PHttpServerState+mkServerState o authenv lockedfilesqsem =  	withAnnexWorkerPool (jobsOption o) $ 		mkP2PHttpServerState 			(mkGetServerMode authenv o) 			return 			(fromMaybe 1 $ proxyConnectionsOption o) 			(fromMaybe 1 $ clusterJobsOption o)+			lockedfilesqsem  mkGetServerMode :: M.Map Auth P2P.ServerMode -> Options -> GetServerMode mkGetServerMode _ o _ Nothing
Command/RemoteDaemon.hs view
@@ -1,6 +1,6 @@ {- git-annex command  -- - Copyright 2014-2016 Joey Hess <id@joeyh.name>+ - Copyright 2014-2025 Joey Hess <id@joeyh.name>  -  - Licensed under the GNU AGPL version 3 or higher.  -}@@ -12,6 +12,7 @@ import Command import RemoteDaemon.Core import Utility.Daemon+import qualified Annex #ifndef mingw32_HOST_OS import Annex.Path import Utility.OpenFd@@ -31,9 +32,12 @@ #ifndef mingw32_HOST_OS 		git_annex <- fromOsPath <$> liftIO programPath 		ps <- gitAnnexDaemonizeParams-		let logfd = openFdWithMode (toRawFilePath "/dev/null") ReadOnly Nothing -			defaultFileFlags-			(CloseOnExecFlag True)+		logfd <- ifM (Annex.getRead Annex.debugenabled) +			( return Nothing+			, return $ Just $ openFdWithMode (toRawFilePath "/dev/null") WriteOnly Nothing +				defaultFileFlags+				(CloseOnExecFlag True)+			) 		liftIO $ daemonize git_annex ps logfd Nothing False runNonInteractive #else 		liftIO $ foreground Nothing runNonInteractive
P2P/Http/Server.hs view
@@ -121,12 +121,12 @@ 	-> Maybe Auth 	-> Handler (Headers '[DataLengthHeader] (S.SourceT IO B.ByteString)) serveGet mst su apiver (B64Key k) cu bypass baf startat sec auth = do-	(conn, st) <- getP2PConnection apiver mst cu su bypass sec auth ReadAction id+	(conn, st) <- getP2PConnection apiver WorkerPoolRunner mst cu su bypass sec auth ReadAction id 	bsv <- liftIO newEmptyTMVarIO 	endv <- liftIO newEmptyTMVarIO 	validityv <- liftIO newEmptyTMVarIO 	finalv <- liftIO newEmptyTMVarIO-	annexworker <- liftIO $ async $ inAnnexWorker st $ do+	annexworker <- liftIO $ async $ handleRequestAnnex st $ do 		let storer _offset len = sendContentWith $ \bs -> liftIO $ do 			atomically $ putTMVar bsv (len, bs) 			atomically $ takeTMVar endv@@ -189,7 +189,7 @@ 				validity <- atomically $ takeTMVar validityv 				sz <- takeMVar szv 				atomically $ putTMVar finalv ()-				atomically $ putTMVar endv ()+				void $ atomically $ tryPutTMVar endv () 				return $ case validity of 					Nothing -> True 					Just Valid -> True@@ -232,7 +232,7 @@ 	-> Maybe Auth 	-> Handler CheckPresentResult serveCheckPresent st su apiver (B64Key k) cu bypass sec auth = do-	res <- withP2PConnection apiver st cu su bypass sec auth ReadAction id+	res <- withP2PConnection apiver WorkerPoolRunner st cu su bypass sec auth ReadAction id 		$ \(conn, _) -> liftIO $ proxyClientNetProto conn $ checkPresent k 	case res of 		Right b -> return (CheckPresentResult b)@@ -251,7 +251,7 @@ 	-> Maybe Auth 	-> Handler t serveRemove st resultmangle su apiver (B64Key k) cu bypass sec auth = do-	res <- withP2PConnection apiver st cu su bypass sec auth RemoveAction id+	res <- withP2PConnection apiver WorkerPoolRunner st cu su bypass sec auth RemoveAction id 		$ \(conn, _) -> 			liftIO $ proxyClientNetProto conn $ remove Nothing k 	case res of@@ -273,7 +273,7 @@ 	-> Maybe Auth 	-> Handler RemoveResultPlus serveRemoveBefore st su apiver (B64Key k) cu bypass (Timestamp ts) sec auth = do-	res <- withP2PConnection apiver st cu su bypass sec auth RemoveAction id+	res <- withP2PConnection apiver WorkerPoolRunner st cu su bypass sec auth RemoveAction id 		$ \(conn, _) -> 			liftIO $ proxyClientNetProto conn $ 				removeBeforeRemoteEndTime ts k@@ -294,7 +294,7 @@ 	-> Maybe Auth 	-> Handler GetTimestampResult serveGetTimestamp st su apiver cu bypass sec auth = do-	res <- withP2PConnection apiver st cu su bypass sec auth ReadAction id+	res <- withP2PConnection apiver WorkerPoolRunner st cu su bypass sec auth ReadAction id 		$ \(conn, _) -> 			liftIO $ proxyClientNetProto conn getTimestamp 	case res of@@ -320,7 +320,7 @@ 	-> Maybe Auth 	-> Handler t servePut mst resultmangle su apiver (Just True) _ k cu bypass baf _ _ sec auth = do-	res <- withP2PConnection' apiver mst cu su bypass sec auth WriteAction+	res <- withP2PConnection' apiver WorkerPoolRunner mst cu su bypass sec auth WriteAction 		(\cst -> cst { connectionWaitVar = False }) (liftIO . protoaction) 	servePutResult resultmangle res   where@@ -333,7 +333,7 @@ 		liftIO $ atomically $ readTMVar validityv 	tooshortv <- liftIO newEmptyTMVarIO 	content <- liftIO $ S.unSourceT stream (gather validityv tooshortv)-	res <- withP2PConnection' apiver mst cu su bypass sec auth WriteAction+	res <- withP2PConnection' apiver WorkerPoolRunner mst cu su bypass sec auth WriteAction 		(\cst -> cst { connectionWaitVar = False }) $ \(conn, st) -> do 			liftIO $ void $ async $ checktooshort conn tooshortv 			liftIO (protoaction conn st content validitycheck)@@ -401,7 +401,7 @@ 	-> Maybe B64FilePath 	-> (P2P.Protocol.Offset -> Proto (Maybe [UUID])) 	-> IO (Either SomeException (Either ProtoFailure (Maybe [UUID])))-servePutAction (conn, st) (B64Key k) baf a = inAnnexWorker st $+servePutAction (conn, st) (B64Key k) baf a = handleRequestAnnex st $ 	enteringStage (TransferStage Download) $ 		runFullProto (clientRunState conn) (clientP2PConnection conn) $ 			put' k af a@@ -450,7 +450,7 @@ 	-> Maybe Auth 	-> Handler t servePutOffset st resultmangle su apiver (B64Key k) cu bypass sec auth = do-	res <- withP2PConnection apiver st cu su bypass sec auth WriteAction+	res <- withP2PConnection apiver WorkerPoolRunner st cu su bypass sec auth WriteAction 		(\cst -> cst { connectionWaitVar = False }) $ \(conn, _) -> 			liftIO $ proxyClientNetProto conn $ getPutOffset k af 	case res of@@ -472,14 +472,29 @@ 	-> IsSecure 	-> Maybe Auth 	-> Handler LockResult-serveLockContent mst su apiver (B64Key k) cu bypass sec auth = do-	(conn, st) <- getP2PConnection apiver mst cu su bypass sec auth LockAction id-	let lock = do+serveLockContent mst su apiver (B64Key k) cu bypass sec auth =+	checkAuthActionClass mst su sec auth LockAction $ \st mode ->+		ifM (liftIO $ consumeLockedFilesQSem st)+			( go st mode+			, return $ LockResult False Nothing+			)+  where+	go st mode = do+		-- Uses RequestRunner to avoid using a worker+		-- pool slot for a potentially long-running lock.+		(conn, st') <- getP2PConnection' apiver RequestRunner cu su bypass id st mode+		liftIO $ mkLocker (lock conn st') (unlock conn st') >>= \case+			Just (locker, lockid) -> do+				liftIO $ storeLock lockid locker st+				return $ LockResult True (Just lockid)+			Nothing -> return $ LockResult False Nothing+	+	lock conn st = do 		lockresv <- newEmptyTMVarIO 		unlockv <- newEmptyTMVarIO-		-- A single worker thread takes the lock, and keeps running+		-- A thread takes the lock, and keeps running 		-- until unlock in order to keep the lock held.-		annexworker <- async $ inAnnexWorker st $ do+		lockthread <- async $ handleRequestAnnex st $ do 			lockres <- runFullProto (clientRunState conn) (clientP2PConnection conn) $ do 				net $ sendMessage (LOCKCONTENT k) 				checkSuccess@@ -491,17 +506,14 @@ 						net $ sendMessage UNLOCKCONTENT 				_ -> return () 		atomically (takeTMVar lockresv) >>= \case-			Right True -> return (Just (annexworker, unlockv))+			Right True -> return (Just (lockthread, unlockv)) 			_ -> return Nothing-	let unlock (annexworker, unlockv) = do+	+	unlock conn st (lockthread, unlockv) = do 		atomically $ putTMVar unlockv ()-		void $ wait annexworker+		void $ wait lockthread 		releaseP2PConnection conn-	liftIO $ mkLocker lock unlock >>= \case-		Just (locker, lockid) -> do-			liftIO $ storeLock lockid locker st-			return $ LockResult True (Just lockid)-		Nothing -> return $ LockResult False Nothing+		liftIO $ releaseLockedFilesQSem st  serveKeepLocked 	:: APIVersion v
P2P/Http/State.hs view
@@ -2,7 +2,7 @@  -  - https://git-annex.branchable.com/design/p2p_protocol_over_http/  -- - Copyright 2024 Joey Hess <id@joeyh.name>+ - Copyright 2024-2025 Joey Hess <id@joeyh.name>  -  - Licensed under the GNU AGPL version 3 or higher.  -}@@ -10,6 +10,7 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RankNTypes #-} {-# LANGUAGE TupleSections #-}  module P2P.Http.State where@@ -75,8 +76,11 @@ data PerRepoServerState = PerRepoServerState 	{ acquireP2PConnection :: AcquireP2PConnection 	, annexWorkerPool :: AnnexWorkerPool+	, annexState :: TMVar Annex.AnnexState+	, annexRead :: Annex.AnnexRead 	, getServerMode :: GetServerMode 	, openLocks :: TMVar (M.Map LockID Locker)+	, lockedFilesQSem :: LockedFilesQSem 	}  type AnnexWorkerPool = TMVar (WorkerPool (Annex.AnnexState, Annex.AnnexRead))@@ -91,12 +95,15 @@ 		} 	| CannotServeRequests -mkPerRepoServerState :: AcquireP2PConnection -> AnnexWorkerPool -> GetServerMode -> IO PerRepoServerState-mkPerRepoServerState acquireconn annexworkerpool getservermode = PerRepoServerState+mkPerRepoServerState :: AcquireP2PConnection -> AnnexWorkerPool -> TMVar Annex.AnnexState -> Annex.AnnexRead -> GetServerMode -> LockedFilesQSem -> IO PerRepoServerState+mkPerRepoServerState acquireconn annexworkerpool annexstate annexread getservermode lockedfilesqsem = PerRepoServerState 	<$> pure acquireconn 	<*> pure annexworkerpool+	<*> pure annexstate+	<*> pure annexread 	<*> pure getservermode 	<*> newTMVarIO mempty+	<*> pure lockedfilesqsem  data ActionClass = ReadAction | WriteAction | RemoveAction | LockAction 	deriving (Eq)@@ -104,6 +111,7 @@ withP2PConnection 	:: APIVersion v 	=> v+	-> AnnexActionRunner 	-> TMVar P2PHttpServerState 	-> B64UUID ClientSide 	-> B64UUID ServerSide@@ -114,8 +122,8 @@ 	-> (ConnectionParams -> ConnectionParams) 	-> ((P2PConnectionPair, PerRepoServerState) -> Handler (Either ProtoFailure 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 runner mst cu su bypass sec auth actionclass fconnparams connaction =+	withP2PConnection' apiver runner mst cu su bypass sec auth actionclass fconnparams connaction'   where 	connaction' conn = connaction conn >>= \case 		Right r -> return r@@ -125,6 +133,7 @@ withP2PConnection' 	:: APIVersion v 	=> v+	-> AnnexActionRunner 	-> TMVar P2PHttpServerState 	-> B64UUID ClientSide 	-> B64UUID ServerSide@@ -135,14 +144,15 @@ 	-> (ConnectionParams -> ConnectionParams) 	-> ((P2PConnectionPair, PerRepoServerState) -> Handler a) 	-> Handler a-withP2PConnection' apiver mst cu su bypass sec auth actionclass fconnparams connaction = do-	(conn, st) <- getP2PConnection apiver mst cu su bypass sec auth actionclass fconnparams+withP2PConnection' apiver runner mst cu su bypass sec auth actionclass fconnparams connaction = do+	(conn, st) <- getP2PConnection apiver runner mst cu su bypass sec auth actionclass fconnparams 	connaction (conn, st) 		`finally` liftIO (releaseP2PConnection conn)  getP2PConnection 	:: APIVersion v 	=> v+	-> AnnexActionRunner 	-> TMVar P2PHttpServerState 	-> B64UUID ClientSide 	-> B64UUID ServerSide@@ -152,24 +162,37 @@ 	-> ActionClass 	-> (ConnectionParams -> ConnectionParams) 	-> Handler (P2PConnectionPair, PerRepoServerState)-getP2PConnection apiver mst cu su bypass sec auth actionclass fconnparams =-	checkAuthActionClass mst su sec auth actionclass go-  where-	go st servermode = liftIO (acquireP2PConnection st cp) >>= \case+getP2PConnection apiver runner mst cu su bypass sec auth actionclass fconnparams =+	checkAuthActionClass mst su sec auth actionclass $+		getP2PConnection' apiver runner cu su bypass fconnparams++getP2PConnection'+	:: APIVersion v+	=> v+	-> AnnexActionRunner+	-> B64UUID ClientSide+	-> B64UUID ServerSide+	-> [B64UUID Bypass]+	-> (ConnectionParams -> ConnectionParams)+	-> PerRepoServerState+	-> P2P.ServerMode+	-> Handler (P2PConnectionPair, PerRepoServerState)+getP2PConnection' apiver runner cu su bypass fconnparams st servermode =+	liftIO (acquireP2PConnection st runner cp) >>= \case 		Left (ConnectionFailed err) ->  			throwError err502 { errBody = encodeBL err } 		Left TooManyConnections -> 			throwError err503 		Right v -> return (v, st)-	  where-		cp = fconnparams $ ConnectionParams-			{ connectionProtocolVersion = protocolVersion apiver-			, connectionServerUUID = fromB64UUID su-			, connectionClientUUID = fromB64UUID cu-			, connectionBypass = map fromB64UUID bypass-			, connectionServerMode = servermode-			, connectionWaitVar = True-			}+  where+	cp = fconnparams $ ConnectionParams+		{ connectionProtocolVersion = protocolVersion apiver+		, connectionServerUUID = fromB64UUID su+		, connectionClientUUID = fromB64UUID cu+		, connectionBypass = map fromB64UUID bypass+		, connectionServerMode = servermode+		, connectionWaitVar = True+		}  getPerRepoServerState :: TMVar P2PHttpServerState -> B64UUID ServerSide -> IO (Maybe PerRepoServerState) getPerRepoServerState mstv su = do@@ -251,17 +274,40 @@ 	(clientRunState conn) (clientP2PConnection conn)  type AcquireP2PConnection-	= ConnectionParams+	= AnnexActionRunner+	-> ConnectionParams 	-> IO (Either ConnectionProblem P2PConnectionPair) +type LockedFilesQSem = TMVar Integer++mkLockedFilesQSem :: Maybe Integer -> IO LockedFilesQSem+mkLockedFilesQSem = newTMVarIO . fromMaybe 100++consumeLockedFilesQSem :: PerRepoServerState -> IO Bool+consumeLockedFilesQSem st = atomically $ do+	n <- takeTMVar (lockedFilesQSem st)+	if n < 1+		then do+			putTMVar (lockedFilesQSem st) n+			return False+		else do+			putTMVar (lockedFilesQSem st) (pred n)+			return True++releaseLockedFilesQSem :: PerRepoServerState -> IO ()+releaseLockedFilesQSem st = atomically $ do+	n <- takeTMVar (lockedFilesQSem st)+	putTMVar (lockedFilesQSem st) (succ n)+ mkP2PHttpServerState 	:: GetServerMode 	-> UpdateRepos 	-> ProxyConnectionPoolSize 	-> ClusterConcurrency+	-> LockedFilesQSem 	-> AnnexWorkerPool 	-> Annex P2PHttpServerState-mkP2PHttpServerState getservermode updaterepos proxyconnectionpoolsize clusterconcurrency workerpool = do+mkP2PHttpServerState getservermode updaterepos proxyconnectionpoolsize clusterconcurrency lockedfilesqsem workerpool = do 	enableInteractiveBranchAccess 	myuuid <- getUUID 	myproxies <- M.lookup myuuid <$> getProxies@@ -275,16 +321,22 @@ 		liftIO $ atomically $ putTMVar endv () 		liftIO $ wait asyncservicer 	let servinguuids = myuuid : map proxyRemoteUUID (maybe [] S.toList myproxies)-	st <- liftIO $ mkPerRepoServerState (acquireconn reqv) workerpool getservermode+	annexstate <- liftIO . newTMVarIO =<< dupState+	annexread <- Annex.getRead id+	st <- liftIO $ mkPerRepoServerState +		(acquireconn reqv annexstate annexread)+		workerpool annexstate annexread getservermode lockedfilesqsem 	return $ P2PHttpServerState 		{ servedRepos = M.fromList $ zip servinguuids (repeat st) 		, serverShutdownCleanup = endit 		, updateRepos = updaterepos 		}   where-	acquireconn reqv connparams = do+	acquireconn reqv annexstate annexread runnertype connparams = do+		ready <- newEmptyTMVarIO 		respvar <- newEmptyTMVarIO-		atomically $ putTMVar reqv (connparams, respvar)+		atomically $ putTMVar reqv (runnertype, annexstate, annexread, connparams, ready, respvar)+		() <- atomically $ takeTMVar ready 		atomically $ takeTMVar respvar  	servicer myuuid myproxies proxypool reqv relv endv = do@@ -296,8 +348,8 @@ 					`orElse`  				(Left . Left <$> takeTMVar endv) 		case reqrel of-			Right (connparams, respvar) -> do-				servicereq myuuid myproxies proxypool relv connparams+			Right (runnertype, annexstate, annexread, connparams, ready, respvar) -> do+				servicereq runnertype annexstate annexread myuuid myproxies proxypool relv connparams ready 					>>= atomically . putTMVar respvar 				servicer myuuid myproxies proxypool reqv relv endv 			Left (Right releaseconn) -> do@@ -305,36 +357,40 @@ 				servicer myuuid myproxies proxypool reqv relv endv 			Left (Left ()) -> return () 	-	servicereq myuuid myproxies proxypool relv connparams+	servicereq runnertype annexstate annexread myuuid myproxies proxypool relv connparams ready 		| connectionServerUUID connparams == myuuid =-			localConnection relv connparams workerpool+			localConnection relv connparams runner ready 		| otherwise = 			atomically (getProxyConnectionPool proxypool connparams) >>= \case-				Just conn -> proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool conn-				Nothing -> checkcanproxy myproxies proxypool relv connparams+				Just conn -> +					proxyConnection proxyconnectionpoolsize relv connparams runner proxypool conn ready+				Nothing -> checkcanproxy runnertype annexstate annexread myproxies proxypool relv connparams ready+	  where+		runner = annexActionRunner runnertype workerpool annexstate annexread -	checkcanproxy myproxies proxypool relv connparams = -		inAnnexWorker' workerpool+	checkcanproxy runnertype annexstate annexread myproxies proxypool relv connparams ready =+		runner 			(checkCanProxy' myproxies (connectionServerUUID connparams)) 		>>= \case 			Right (Left reason) -> return $ Left $ 				ConnectionFailed $  					fromMaybe "unknown uuid" reason 			Right (Right (Right proxyremote)) -> proxyconnection $-				openProxyConnectionToRemote workerpool+				runner $ openProxyConnectionToRemote 					(connectionProtocolVersion connparams) 					bypass proxyremote 			Right (Right (Left clusteruuid)) -> proxyconnection $-				openProxyConnectionToCluster workerpool+				runner $ openProxyConnectionToCluster 					(connectionProtocolVersion connparams) 					bypass clusteruuid clusterconcurrency 			Left ex -> return $ Left $ 				ConnectionFailed $ show ex 	  where+		runner = annexActionRunner runnertype workerpool annexstate annexread 		bypass = P2P.Bypass $ S.fromList $ connectionBypass connparams 		proxyconnection openconn = openconn >>= \case 			Right conn -> proxyConnection proxyconnectionpoolsize-				relv connparams workerpool proxypool conn+				relv connparams runner proxypool conn ready 			Left ex -> return $ Left $ 				ConnectionFailed $ show ex @@ -353,11 +409,13 @@ localConnection 	:: TMVar (IO ()) 	-> ConnectionParams-	-> AnnexWorkerPool+	-> (Annex () -> IO (Either SomeException ()))+	-> TMVar () 	-> IO (Either ConnectionProblem P2PConnectionPair)-localConnection relv connparams workerpool = +localConnection relv connparams runner ready =  	localP2PConnectionPair connparams relv $ \serverrunst serverconn ->-		inAnnexWorker' workerpool $+		runner $ do+			liftIO $ atomically $ putTMVar ready () 			void $ runFullProto serverrunst serverconn $ 				P2P.serveOneCommandAuthed 					(connectionServerMode connparams)@@ -428,36 +486,37 @@ 	:: ProxyConnectionPoolSize 	-> TMVar (IO ()) 	-> ConnectionParams-	-> AnnexWorkerPool+	-> (Annex () -> IO (Either SomeException ())) 	-> TMVar ProxyConnectionPool 	-> ProxyConnection+	-> TMVar () 	-> IO (Either ConnectionProblem P2PConnectionPair)-proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool proxyconn = do+proxyConnection proxyconnectionpoolsize relv connparams runner proxypool proxyconn ready = do 	(clientconn, proxyfromclientconn) <-  		mkP2PConnectionPair connparams ("http client", "proxy") 	clientrunst <- mkClientRunState connparams 	proxyfromclientrunst <- mkClientRunState connparams-	asyncworker <- async $-		inAnnexWorker' workerpool $ do-			proxystate <- liftIO Proxy.mkProxyState-			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 = proxyConnectionSelector proxyconn-				, Proxy.proxyConcurrencyConfig = proxyConnectionConcurrency proxyconn-				, Proxy.proxyClientProtocolVersion = connectionProtocolVersion connparams-				}-			let proxy mrequestmessage = case mrequestmessage of-				Just requestmessage -> do-					Proxy.proxyRequest proxydone proxyparams-						requestcomplete requestmessage protoerrhandler-				Nothing -> return ()-			protoerrhandler proxy $-				liftIO $ runNetProto proxyfromclientrunst proxyfromclientconn $-					P2P.net P2P.receiveMessage+	asyncworker <- async $ runner $ do+		liftIO $ atomically $ putTMVar ready ()+		proxystate <- liftIO Proxy.mkProxyState+		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 = proxyConnectionSelector proxyconn+			, Proxy.proxyConcurrencyConfig = proxyConnectionConcurrency proxyconn+			, Proxy.proxyClientProtocolVersion = connectionProtocolVersion connparams+			}+		let proxy mrequestmessage = case mrequestmessage of+			Just requestmessage -> do+				Proxy.proxyRequest proxydone proxyparams+					requestcomplete requestmessage protoerrhandler+			Nothing -> return ()+		protoerrhandler proxy $+			liftIO $ runNetProto proxyfromclientrunst proxyfromclientconn $+				P2P.net P2P.receiveMessage 	 	let closebothsides = do 		liftIO $ closeConnection proxyfromclientconn@@ -495,8 +554,7 @@ 	 	requestcomplete () = return () 	-	closeproxyconnection = -		void . inAnnexWorker' workerpool . proxyConnectionCloser+	closeproxyconnection = void . runner . proxyConnectionCloser  data Locker = Locker 	{ lockerThread :: Async ()@@ -513,6 +571,13 @@ mkLocker lock unlock = do 	lv <- newEmptyTMVarIO 	timeoutdisablev <- newEmptyTMVarIO+	timeouttid <- async $ whenM (atomically $ readTMVar lv) $ do+		threadDelaySeconds $ Seconds $ fromIntegral $+			durationSeconds p2pDefaultLockContentRetentionDuration+		atomically (tryReadTMVar timeoutdisablev) >>= \case+			Nothing -> void $ atomically $+				writeTMVar lv False+			Just () -> noop 	let setlocked = putTMVar lv 	locktid <- async $ lock >>= \case 		Nothing ->@@ -528,13 +593,6 @@ 	locksuccess <- atomically $ readTMVar lv 	if locksuccess 		then do-			timeouttid <- async $ do-				threadDelaySeconds $ Seconds $ fromIntegral $-					durationSeconds p2pDefaultLockContentRetentionDuration-				atomically (tryReadTMVar timeoutdisablev) >>= \case-					Nothing -> void $ atomically $-						writeTMVar lv False-					Just () -> noop 			tid <- async $ do 				wait locktid 				cancel timeouttid@@ -585,11 +643,8 @@ 			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)--inAnnexWorker' :: AnnexWorkerPool -> Annex a -> IO (Either SomeException a)-inAnnexWorker' poolv annexaction = do+inAnnexWorker :: AnnexWorkerPool -> Annex a -> IO (Either SomeException a)+inAnnexWorker poolv annexaction = do 	(workerstrd, workerstage) <- atomically $ waitStartWorkerSlot poolv 	resv <- newEmptyTMVarIO 	aid <- async $ do@@ -611,6 +666,34 @@ 		putTMVar poolv pool' 	return res +handleRequestAnnex :: PerRepoServerState -> Annex a -> IO (Either SomeException a)+handleRequestAnnex st = handleRequestAnnex' (annexState st) (annexRead st)++handleRequestAnnex' :: TMVar Annex.AnnexState -> Annex.AnnexRead -> Annex a -> IO (Either SomeException a)+handleRequestAnnex' stv rd a = do+	ast <- atomically $ readTMVar stv+	let astdup = dupState' ast++	(res, (astdup', _)) <- Annex.run (astdup, rd) $ +		tryNonAsync a+	+	ast' <- atomically $ takeTMVar stv+	((), (ast'', _)) <- Annex.run (ast', rd) $ mergeState astdup'+	atomically $ putTMVar stv ast''++	return res++data AnnexActionRunner+	-- Number of concurrent Annex actions is bounded by the size of the +	-- worker pool+	= WorkerPoolRunner+	-- When using this, must enforce bounds first+	| RequestRunner++annexActionRunner :: AnnexActionRunner -> AnnexWorkerPool -> TMVar Annex.AnnexState -> Annex.AnnexRead -> Annex a -> IO (Either SomeException a)+annexActionRunner WorkerPoolRunner workerpool _ _ = inAnnexWorker workerpool+annexActionRunner RequestRunner _ st rd = handleRequestAnnex' st rd+ data ProxyConnection = ProxyConnection 	{ proxyConnectionRemoteUUID :: UUID 	, proxyConnectionSelector :: Proxy.ProxySelector@@ -644,38 +727,34 @@ 		fastDebug "P2P.Http" ("Closed proxy connection to " ++ desc)  openProxyConnectionToRemote-	:: AnnexWorkerPool-	-> P2P.ProtocolVersion+	:: P2P.ProtocolVersion 	-> P2P.Bypass 	-> Remote-	-> IO (Either SomeException ProxyConnection)-openProxyConnectionToRemote workerpool clientmaxversion bypass remote =-	inAnnexWorker' workerpool $ do-		remoteside <- proxyRemoteSide clientmaxversion bypass remote-		concurrencyconfig <- Proxy.noConcurrencyConfig-		openedProxyConnection (Remote.uuid remote)-			("remote " ++ Remote.name remote)-			(Proxy.singleProxySelector remoteside)-			(Proxy.closeRemoteSide remoteside)-			concurrencyconfig+	-> Annex ProxyConnection+openProxyConnectionToRemote clientmaxversion bypass remote = do+	remoteside <- proxyRemoteSide clientmaxversion bypass remote+	concurrencyconfig <- Proxy.noConcurrencyConfig+	openedProxyConnection (Remote.uuid remote)+		("remote " ++ Remote.name remote)+		(Proxy.singleProxySelector remoteside)+		(Proxy.closeRemoteSide remoteside)+		concurrencyconfig  type ClusterConcurrency = Int  openProxyConnectionToCluster-	:: AnnexWorkerPool-	-> P2P.ProtocolVersion+	:: P2P.ProtocolVersion 	-> P2P.Bypass 	-> ClusterUUID 	-> ClusterConcurrency-	-> IO (Either SomeException ProxyConnection)-openProxyConnectionToCluster workerpool clientmaxversion bypass clusteruuid concurrency =-	inAnnexWorker' workerpool $ do-		(proxyselector, closenodes) <--			clusterProxySelector clusteruuid clientmaxversion bypass-		concurrencyconfig <- Proxy.mkConcurrencyConfig concurrency-		openedProxyConnection (fromClusterUUID clusteruuid)-			("cluster " ++ fromUUID (fromClusterUUID clusteruuid))-			proxyselector closenodes concurrencyconfig+	-> Annex ProxyConnection+openProxyConnectionToCluster clientmaxversion bypass clusteruuid concurrency = do+	(proxyselector, closenodes) <-+		clusterProxySelector clusteruuid clientmaxversion bypass+	concurrencyconfig <- Proxy.mkConcurrencyConfig concurrency+	openedProxyConnection (fromClusterUUID clusteruuid)+		("cluster " ++ fromUUID (fromClusterUUID clusteruuid))+		proxyselector closenodes concurrencyconfig  type ProxyConnectionPool = (Integer, M.Map ProxyConnectionPoolKey [ProxyConnection])
RemoteDaemon/Transport/P2PGeneric.hs view
@@ -38,6 +38,7 @@ import Control.Concurrent.STM.TBMQueue import Control.Concurrent.Async import Network.Socket (Socket)+import System.IO.Error import qualified Data.Set as S  server :: Server@@ -179,6 +180,8 @@ 				P2P.serveAuthed P2P.ServeReadWrite u 			case v' of 				Right () -> return ()+				Left (ProtoFailureIOError e) | isEOFError e ->+					return () 				Left e -> liftIO $ debug' $  					netname ++ " connection error: " ++ describeProtoFailure e
Types/WorkerPool.hs view
@@ -114,8 +114,6 @@  -- | Allocates a WorkerPool that has the specified number of workers -- in it, of each stage.------ The stages are distributed evenly throughout. allocateWorkerPool :: t -> Int -> UsedStages -> WorkerPool t allocateWorkerPool t n u = WorkerPool 	{ usedStages = u
Utility/Daemon.hs view
@@ -44,7 +44,7 @@  - Instead, it runs the cmd with provided params, in the background,  - which the caller should arrange to run this again.  -}-daemonize :: String -> [CommandParam] -> IO Fd -> Maybe OsPath -> Bool -> IO () -> IO ()+daemonize :: String -> [CommandParam] -> Maybe (IO Fd) -> Maybe OsPath -> Bool -> IO () -> IO () daemonize cmd params openlogfd pidfile changedirectory a = do 	maybe noop checkalreadyrunning pidfile 	getEnv envvar >>= \case@@ -55,7 +55,7 @@ 			nullfd <- openFdWithMode (toRawFilePath "/dev/null") ReadOnly Nothing defaultFileFlags  				(CloseOnExecFlag True) 			redir nullfd stdInput-			redirLog =<< openlogfd+			maybe noop (redirLog =<<) openlogfd 			environ <- getEnvironment 			_ <- createProcess $ 				(proc cmd (toCommand params))
git-annex.cabal view
@@ -1,5 +1,5 @@ Name: git-annex-Version: 10.20251029+Version: 10.20251114 Cabal-Version: 1.12 License: AGPL-3 Maintainer: Joey Hess <id@joeyh.name>