This old code will now be useful for git-annex beyond the assistant.
git-annex won't use the CheckTransferrer part, and won't run transferkeys
as a batch process, and will want withTransferrer to not shut down
transferkeys processes. Still, the rest of this is a good fit for what I
need now.
Also removed some dead code, and simplified a little bit.
This commit was sponsored by Mark Reidenbach on Patreon.
--- /dev/null
+{- A pool of "git-annex transferkeys" processes
+ -
+ - Copyright 2013-2020 Joey Hess <id@joeyh.name>
+ -
+ - Licensed under the GNU AGPL version 3 or higher.
+ -}
+
+{-# LANGUAGE RankNTypes #-}
+
+module Annex.TransferrerPool where
+
+import Annex.Common
+import Types.TransferrerPool
+import Types.Transfer
+import Utility.Batch
+import Messages.Serialized
+import qualified Command.TransferKeys as T
+
+import Control.Concurrent.STM hiding (check)
+import Control.Concurrent
+import Control.Monad.IO.Class (MonadIO)
+
+{- Runs an action with a Transferrer from the pool.
+ -
+ - When minimizeprocesses is True, only one Transferrer is left running
+ - in the pool at a time. So if this needed to start a new Transferrer,
+ - it's stopped when done. Otherwise, idle processes are left in the pool
+ - for use later.
+ -}
+withTransferrer :: Bool -> FilePath -> BatchCommandMaker -> TransferrerPool -> (Transferrer -> IO a) -> IO a
+withTransferrer minimizeprocesses program batchmaker pool a = do
+ (mi, leftinpool) <- atomically (popTransferrerPool pool)
+ i@(TransferrerPoolItem (Just t) check) <- case mi of
+ Nothing -> mkTransferrerPoolItem pool =<< mkTransferrer program batchmaker
+ Just i -> checkTransferrerPoolItem program batchmaker i
+ a t `finally` returntopool leftinpool check t i
+ where
+ returntopool leftinpool check t i
+ | not minimizeprocesses || leftinpool == 0 =
+ atomically $ pushTransferrerPool pool i
+ | otherwise = do
+ void $ forkIO $ stopTransferrer t
+ atomically $ pushTransferrerPool pool $ TransferrerPoolItem Nothing check
+
+{- Check if a Transferrer from the pool is still ok to be used.
+ - If not, stop it and start a new one. -}
+checkTransferrerPoolItem :: FilePath -> BatchCommandMaker -> TransferrerPoolItem -> IO TransferrerPoolItem
+checkTransferrerPoolItem program batchmaker i = case i of
+ TransferrerPoolItem (Just t) check -> ifM check
+ ( return i
+ , do
+ stopTransferrer t
+ new check
+ )
+ TransferrerPoolItem Nothing check -> new check
+ where
+ new check = do
+ t <- mkTransferrer program batchmaker
+ return $ TransferrerPoolItem (Just t) check
+
+{- Requests that a Transferrer perform a Transfer, and waits for it to
+ - finish. -}
+performTransfer
+ :: (Monad m, MonadIO m, MonadMask m)
+ => Transferrer
+ -> Transfer
+ -> TransferInfo
+ -> (forall a. Annex a -> m a)
+ -- ^ Run an annex action in the monad. Will not be used with
+ -- actions that block for a long time.
+ -> m Bool
+performTransfer transferrer t info runannex = catchBoolIO $ do
+ (liftIO $ T.sendRequest t info (transferrerWrite transferrer))
+ relaySerializedOutput
+ (liftIO $ T.readResponse (transferrerRead transferrer))
+ (liftIO . T.sendSerializedOutputResponse (transferrerWrite transferrer))
+ runannex
+
+{- Starts a new git-annex transferkeys process, setting up handles
+ - that will be used to communicate with it. -}
+mkTransferrer :: FilePath -> BatchCommandMaker -> IO Transferrer
+mkTransferrer program batchmaker = do
+ {- It runs as a batch job. -}
+ let (program', params') = batchmaker (program, [Param "transferkeys"])
+ {- It's put into its own group so that the whole group can be
+ - killed to stop a transfer. -}
+ (Just writeh, Just readh, _, pid) <- createProcess
+ (proc program' $ toCommand params')
+ { create_group = True
+ , std_in = CreatePipe
+ , std_out = CreatePipe
+ }
+ return $ Transferrer
+ { transferrerRead = readh
+ , transferrerWrite = writeh
+ , transferrerHandle = pid
+ }
+
+{- Closing the fds will stop the transferrer, but only when it's in between
+ - transfers. -}
+stopTransferrer :: Transferrer -> IO ()
+stopTransferrer t = do
+ hClose $ transferrerRead t
+ hClose $ transferrerWrite t
+ void $ waitForProcess $ transferrerHandle t
import Utility.ThreadScheduler
import Assistant.Types.TransferSlots
import Assistant.DaemonStatus
-import Assistant.TransferrerPool
+import Annex.TransferrerPool
import Assistant.Types.TransferrerPool
import Assistant.Types.TransferQueue
import Assistant.TransferQueue
runTransferThread' program batchmaker d run = go
where
go = catchPauseResume $
- withTransferrer program batchmaker (transferrerPool d)
+ withTransferrer True program batchmaker (transferrerPool d)
run
pause = catchPauseResume $
runEvery (Seconds 86400) noop
- usual cleanup. However, first check if something else is
- running the transfer, to avoid removing active transfers.
-}
- go remote transferrer = ifM (performTransfer transferrer t info)
+ go remote transferrer = ifM (performTransfer transferrer t info liftAnnex)
( do
case associatedFile info of
AssociatedFile Nothing -> noop
+++ /dev/null
-{- A pool of "git-annex transferkeys" processes
- -
- - Copyright 2013 Joey Hess <id@joeyh.name>
- -
- - Licensed under the GNU AGPL version 3 or higher.
- -}
-
-module Assistant.TransferrerPool where
-
-import Assistant.Common
-import Assistant.Types.TransferrerPool
-import Types.Transfer
-import Utility.Batch
-import Messages.Serialized
-
-import qualified Command.TransferKeys as T
-
-import Control.Concurrent.STM hiding (check)
-import Control.Exception (throw)
-import Control.Concurrent
-
-{- Runs an action with a Transferrer from the pool.
- -
- - Only one Transferrer is left running in the pool at a time.
- - So if this needed to start a new Transferrer, it's stopped when done.
- -}
-withTransferrer :: FilePath -> BatchCommandMaker -> TransferrerPool -> (Transferrer -> IO a) -> IO a
-withTransferrer program batchmaker pool a = do
- (mi, leftinpool) <- atomically (popTransferrerPool pool)
- i@(TransferrerPoolItem (Just t) check) <- case mi of
- Nothing -> mkTransferrerPoolItem pool =<< mkTransferrer program batchmaker
- Just i -> checkTransferrerPoolItem program batchmaker i
- v <- tryNonAsync $ a t
- if leftinpool == 0
- then atomically $ pushTransferrerPool pool i
- else do
- void $ forkIO $ stopTransferrer t
- atomically $ pushTransferrerPool pool $ TransferrerPoolItem Nothing check
- either throw return v
-
-{- Check if a Transferrer from the pool is still ok to be used.
- - If not, stop it and start a new one. -}
-checkTransferrerPoolItem :: FilePath -> BatchCommandMaker -> TransferrerPoolItem -> IO TransferrerPoolItem
-checkTransferrerPoolItem program batchmaker i = case i of
- TransferrerPoolItem (Just t) check -> ifM check
- ( return i
- , do
- stopTransferrer t
- new check
- )
- TransferrerPoolItem Nothing check -> new check
- where
- new check = do
- t <- mkTransferrer program batchmaker
- return $ TransferrerPoolItem (Just t) check
-
-{- Requests that a Transferrer perform a Transfer, and waits for it to
- - finish. -}
-performTransfer :: Transferrer -> Transfer -> TransferInfo -> Assistant Bool
-performTransfer transferrer t info = catchBoolIO $ do
- (liftIO $ T.sendRequest t info (transferrerWrite transferrer))
- relaySerializedOutput
- (liftIO $ T.readResponse (transferrerRead transferrer))
- (liftIO . T.sendSerializedOutputResponse (transferrerWrite transferrer))
- liftAnnex
-
-{- Starts a new git-annex transferkeys process, setting up handles
- - that will be used to communicate with it. -}
-mkTransferrer :: FilePath -> BatchCommandMaker -> IO Transferrer
-mkTransferrer program batchmaker = do
- {- It runs as a batch job. -}
- let (program', params') = batchmaker (program, [Param "transferkeys"])
- {- It's put into its own group so that the whole group can be
- - killed to stop a transfer. -}
- (Just writeh, Just readh, _, pid) <- createProcess
- (proc program' $ toCommand params')
- { create_group = True
- , std_in = CreatePipe
- , std_out = CreatePipe
- }
- return $ Transferrer
- { transferrerRead = readh
- , transferrerWrite = writeh
- , transferrerHandle = pid
- }
-
-{- Checks if a Transferrer is still running. If not, makes a new one. -}
-checkTransferrer :: FilePath -> BatchCommandMaker -> Transferrer -> IO Transferrer
-checkTransferrer program batchmaker t =
- maybe (return t) (const $ mkTransferrer program batchmaker)
- =<< getProcessExitCode (transferrerHandle t)
-
-{- Closing the fds will stop the transferrer. -}
-stopTransferrer :: Transferrer -> IO ()
-stopTransferrer t = do
- hClose $ transferrerRead t
- hClose $ transferrerWrite t
- void $ waitForProcess $ transferrerHandle t
- Licensed under the GNU AGPL version 3 or higher.
-}
-module Assistant.Types.TransferrerPool where
+module Assistant.Types.TransferrerPool (
+ module Types.TransferrerPool,
+ checkNetworkConnections,
+) where
-import Annex.Common
+import Types.TransferrerPool
import Utility.NotificationBroadcaster
import Assistant.Types.DaemonStatus
-import Control.Concurrent.STM hiding (check)
-
-type TransferrerPool = TVar (MkCheckTransferrer, [TransferrerPoolItem])
-
-type CheckTransferrer = IO Bool
-type MkCheckTransferrer = IO (IO Bool)
-
-{- Each item in the pool may have a transferrer running, and has an
- - IO action that can be used to check if it's still ok to use the
- - transferrer. -}
-data TransferrerPoolItem = TransferrerPoolItem (Maybe Transferrer) CheckTransferrer
-
-data Transferrer = Transferrer
- { transferrerRead :: Handle
- , transferrerWrite :: Handle
- , transferrerHandle :: ProcessHandle
- }
-
-newTransferrerPool :: MkCheckTransferrer -> IO TransferrerPool
-newTransferrerPool c = newTVarIO (c, [])
-
-popTransferrerPool :: TransferrerPool -> STM (Maybe TransferrerPoolItem, Int)
-popTransferrerPool p = do
- (c, l) <- readTVar p
- case l of
- [] -> return (Nothing, 0)
- (i:is) -> do
- writeTVar p (c, is)
- return $ (Just i, length is)
-
-pushTransferrerPool :: TransferrerPool -> TransferrerPoolItem -> STM ()
-pushTransferrerPool p i = do
- (c, l) <- readTVar p
- let l' = i:l
- writeTVar p (c, l')
-
-{- Note that making a CheckTransferrer may allocate resources,
- - such as a NotificationHandle, so it's important that the returned
- - TransferrerPoolItem is pushed into the pool, and not left to be
- - garbage collected. -}
-mkTransferrerPoolItem :: TransferrerPool -> Transferrer -> IO TransferrerPoolItem
-mkTransferrerPoolItem p t = do
- mkcheck <- atomically $ fst <$> readTVar p
- check <- mkcheck
- return $ TransferrerPoolItem (Just t) check
+import Control.Concurrent.STM
checkNetworkConnections :: DaemonStatusHandle -> MkCheckTransferrer
checkNetworkConnections dstatushandle = do
--- /dev/null
+{- A pool of "git-annex transferkeys" processes available for use
+ -
+ - Copyright 2013-2020 Joey Hess <id@joeyh.name>
+ -
+ - Licensed under the GNU AGPL version 3 or higher.
+ -}
+
+module Types.TransferrerPool where
+
+import Annex.Common
+
+import Control.Concurrent.STM hiding (check)
+
+type TransferrerPool = TVar (MkCheckTransferrer, [TransferrerPoolItem])
+
+type CheckTransferrer = IO Bool
+type MkCheckTransferrer = IO (IO Bool)
+
+{- Each item in the pool may have a transferrer running, and has an
+ - IO action that can be used to check if it's still ok to use the
+ - transferrer. -}
+data TransferrerPoolItem = TransferrerPoolItem (Maybe Transferrer) CheckTransferrer
+
+data Transferrer = Transferrer
+ { transferrerRead :: Handle
+ , transferrerWrite :: Handle
+ , transferrerHandle :: ProcessHandle
+ }
+
+newTransferrerPool :: MkCheckTransferrer -> IO TransferrerPool
+newTransferrerPool c = newTVarIO (c, [])
+
+popTransferrerPool :: TransferrerPool -> STM (Maybe TransferrerPoolItem, Int)
+popTransferrerPool p = do
+ (c, l) <- readTVar p
+ case l of
+ [] -> return (Nothing, 0)
+ (i:is) -> do
+ writeTVar p (c, is)
+ return $ (Just i, length is)
+
+pushTransferrerPool :: TransferrerPool -> TransferrerPoolItem -> STM ()
+pushTransferrerPool p i = do
+ (c, l) <- readTVar p
+ let l' = i:l
+ writeTVar p (c, l')
+
+{- Note that making a CheckTransferrer may allocate resources,
+ - such as a NotificationHandle, so it's important that the returned
+ - TransferrerPoolItem is pushed into the pool, and not left to be
+ - garbage collected. -}
+mkTransferrerPoolItem :: TransferrerPool -> Transferrer -> IO TransferrerPoolItem
+mkTransferrerPoolItem p t = do
+ mkcheck <- atomically $ fst <$> readTVar p
+ check <- mkcheck
+ return $ TransferrerPoolItem (Just t) check
module Utility.Batch (
batch,
BatchCommandMaker,
+ nonBatchCommandMaker,
getBatchCommandMaker,
toBatchCommand,
batchCommand,
- are available in the path. -}
type BatchCommandMaker = (String, [CommandParam]) -> (String, [CommandParam])
+nonBatchCommandMaker :: BatchCommandMaker
+nonBatchCommandMaker = id
+
getBatchCommandMaker :: IO BatchCommandMaker
getBatchCommandMaker = do
#ifndef mingw32_HOST_OS
Assistant.Threads.Watcher
Assistant.TransferQueue
Assistant.TransferSlots
- Assistant.TransferrerPool
Assistant.Types.Alert
Assistant.Types.BranchChange
Assistant.Types.Changes
Annex.TaggedPush
Annex.Tmp
Annex.Transfer
+ Annex.TransferrerPool
Annex.UntrustedFilePath
Annex.UpdateInstead
Annex.UUID
Types.StoreRetrieve
Types.Test
Types.Transfer
+ Types.TransferrerPool
Types.TrustLevel
Types.UUID
Types.UrlContents
projects / git-annex.git / commitdiff