diff options
| -rw-r--r-- | Assistant/ThreadedMonad.hs | 10 | ||||
| -rw-r--r-- | Assistant/Threads/Transferrer.hs | 16 | ||||
| -rw-r--r-- | Assistant/TransferQueue.hs | 1 | ||||
| -rw-r--r-- | Logs/Transfer.hs | 15 |
4 files changed, 25 insertions, 17 deletions
diff --git a/Assistant/ThreadedMonad.hs b/Assistant/ThreadedMonad.hs index f32adff43..2fc526599 100644 --- a/Assistant/ThreadedMonad.hs +++ b/Assistant/ThreadedMonad.hs @@ -12,8 +12,6 @@ import qualified Annex import Control.Concurrent import Data.Tuple -import System.Posix.Types -import System.Posix.Process {- The Annex state is stored in a MVar, so that threaded actions can access - it. -} @@ -39,14 +37,14 @@ withThreadState a = do runThreadState :: ThreadState -> Annex a -> IO a runThreadState mvar a = modifyMVar mvar $ \state -> swap <$> Annex.run state a -{- Runs an Annex action in a separate process, using a copy of the state +{- Runs an Annex action in a separate thread, using a copy of the state - from the MVar. - - It's up to the action to perform any necessary shutdown tasks in order - for state to not be lost. And it's up to the caller to resynchronise - with any changes the action makes to eg, the git-annex branch. -} -unsafeForkProcessThreadState :: ThreadState -> Annex a -> IO ProcessID -unsafeForkProcessThreadState mvar a = do +unsafeForkIOThreadState :: ThreadState -> Annex a -> IO ThreadId +unsafeForkIOThreadState mvar a = do state <- readMVar mvar - forkProcess $ void $ Annex.eval state a + forkIO $ void $ Annex.eval state a diff --git a/Assistant/Threads/Transferrer.hs b/Assistant/Threads/Transferrer.hs index 9d3358f54..dd63d4d12 100644 --- a/Assistant/Threads/Transferrer.hs +++ b/Assistant/Threads/Transferrer.hs @@ -18,6 +18,7 @@ import Logs.Location import Annex.Content import qualified Remote +import Data.Time.Clock.POSIX import Data.Time.Clock import qualified Data.Map as M @@ -58,12 +59,12 @@ shouldTransfer dstatus t info = | otherwise = return False key = transferKey t -{- A transfer is run in a separate process, with a *copy* of the Annex +{- A transfer is run in a separate thread, with a *copy* of the Annex - state. This is necessary to avoid blocking the rest of the assistant - on the transfer completing, and also to allow multiple transfers to run - - at once. + - at once. This requires GHC's threaded runtime to work! - - - However, it means that the transfer processes are responsible + - The copy of state means that the transfer processes are responsible - for doing any necessary shutdown cleanups, and that the parent - thread's cache must be invalidated once a transfer completes, as - changes may have been made to the git-annex branch. @@ -73,15 +74,14 @@ runTransfer st dstatus slots t info = case (transferRemote info, associatedFile (Nothing, _) -> noop (_, Nothing) -> noop (Just remote, Just file) -> do - pid <- inTransferSlot slots $ - unsafeForkProcessThreadState st $ + tid <- inTransferSlot slots $ + unsafeForkIOThreadState st $ transferprocess remote file now <- getCurrentTime runThreadState st $ adjustTransfers dstatus $ M.insertWith' const t info - { startedTime = Just now - , transferPid = Just pid - , shouldWait = True + { startedTime = Just $ utcTimeToPOSIXSeconds now + , transferTid = Just tid } where isdownload = transferDirection t == Download diff --git a/Assistant/TransferQueue.hs b/Assistant/TransferQueue.hs index 73e73ca0a..fb7fa87cd 100644 --- a/Assistant/TransferQueue.hs +++ b/Assistant/TransferQueue.hs @@ -24,6 +24,7 @@ stubInfo :: AssociatedFile -> TransferInfo stubInfo f = TransferInfo { startedTime = Nothing , transferPid = Nothing + , transferTid = Nothing , transferRemote = Nothing , bytesComplete = Nothing , associatedFile = f diff --git a/Logs/Transfer.hs b/Logs/Transfer.hs index 260512067..f74d128dc 100644 --- a/Logs/Transfer.hs +++ b/Logs/Transfer.hs @@ -16,6 +16,10 @@ import qualified Fields import System.Posix.Types import Data.Time.Clock +import Data.Time.Clock.POSIX +import Data.Time +import System.Locale +import Control.Concurrent {- Enough information to uniquely identify a transfer, used as the filename - of the transfer information file. -} @@ -33,8 +37,9 @@ data Transfer = Transfer - of some repository, that was acted on to initiate the transfer. -} data TransferInfo = TransferInfo - { startedTime :: Maybe UTCTime + { startedTime :: Maybe POSIXTime , transferPid :: Maybe ProcessID + , transferTid :: Maybe ThreadId , transferRemote :: Maybe Remote , bytesComplete :: Maybe Integer , associatedFile :: Maybe FilePath @@ -76,8 +81,9 @@ transfer t file a = do createAnnexDirectory $ takeDirectory tfile mode <- annexFileMode info <- liftIO $ TransferInfo - <$> (Just <$> getCurrentTime) + <$> (Just . utcTimeToPOSIXSeconds <$> getCurrentTime) <*> pure Nothing -- pid not stored in file, so omitted for speed + <*> pure Nothing -- tid ditto <*> pure Nothing -- not 0; transfer may be resuming <*> pure Nothing <*> pure file @@ -168,13 +174,16 @@ readTransferInfo :: ProcessID -> String -> Maybe TransferInfo readTransferInfo pid s = case bits of [time] -> TransferInfo - <$> readish time + <$> parsetime time <*> pure (Just pid) <*> pure Nothing <*> pure Nothing + <*> pure Nothing <*> pure (if null filename then Nothing else Just filename) <*> pure False _ -> Nothing where (bits, filebits) = splitAt 1 $ lines s filename = join "\n" filebits + parsetime t = Just . utcTimeToPOSIXSeconds + <$> parseTime defaultTimeLocale "%s%Qs" t |