diff --git a/Annex/Concurrent.hs b/Annex/Concurrent.hs --- a/Annex/Concurrent.hs +++ b/Annex/Concurrent.hs @@ -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 () diff --git a/Annex/Content.hs b/Annex/Content.hs --- a/Annex/Content.hs +++ b/Annex/Content.hs @@ -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) diff --git a/Assistant.hs b/Assistant.hs --- a/Assistant.hs +++ b/Assistant.hs @@ -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. diff --git a/CHANGELOG b/CHANGELOG --- a/CHANGELOG +++ b/CHANGELOG @@ -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 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, diff --git a/Command/DropUnused.hs b/Command/DropUnused.hs --- a/Command/DropUnused.hs +++ b/Command/DropUnused.hs @@ -1,6 +1,6 @@ {- git-annex command - - - Copyright 2010,2012,2018 Joey Hess + - Copyright 2010-2025 Joey Hess - - 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 diff --git a/Command/P2P.hs b/Command/P2P.hs --- a/Command/P2P.hs +++ b/Command/P2P.hs @@ -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 diff --git a/Command/P2PHttp.hs b/Command/P2PHttp.hs --- a/Command/P2PHttp.hs +++ b/Command/P2PHttp.hs @@ -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 diff --git a/Command/RemoteDaemon.hs b/Command/RemoteDaemon.hs --- a/Command/RemoteDaemon.hs +++ b/Command/RemoteDaemon.hs @@ -1,6 +1,6 @@ {- git-annex command - - - Copyright 2014-2016 Joey Hess + - Copyright 2014-2025 Joey Hess - - 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 diff --git a/P2P/Http/Server.hs b/P2P/Http/Server.hs --- a/P2P/Http/Server.hs +++ b/P2P/Http/Server.hs @@ -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 diff --git a/P2P/Http/State.hs b/P2P/Http/State.hs --- a/P2P/Http/State.hs +++ b/P2P/Http/State.hs @@ -2,7 +2,7 @@ - - https://git-annex.branchable.com/design/p2p_protocol_over_http/ - - - Copyright 2024 Joey Hess + - Copyright 2024-2025 Joey Hess - - 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]) diff --git a/RemoteDaemon/Transport/P2PGeneric.hs b/RemoteDaemon/Transport/P2PGeneric.hs --- a/RemoteDaemon/Transport/P2PGeneric.hs +++ b/RemoteDaemon/Transport/P2PGeneric.hs @@ -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 diff --git a/Types/WorkerPool.hs b/Types/WorkerPool.hs --- a/Types/WorkerPool.hs +++ b/Types/WorkerPool.hs @@ -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 diff --git a/Utility/Daemon.hs b/Utility/Daemon.hs --- a/Utility/Daemon.hs +++ b/Utility/Daemon.hs @@ -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)) diff --git a/git-annex.cabal b/git-annex.cabal --- a/git-annex.cabal +++ b/git-annex.cabal @@ -1,5 +1,5 @@ Name: git-annex -Version: 10.20251029 +Version: 10.20251114 Cabal-Version: 1.12 License: AGPL-3 Maintainer: Joey Hess