diff --git a/cabal-cache.cabal b/cabal-cache.cabal --- a/cabal-cache.cabal +++ b/cabal-cache.cabal @@ -1,7 +1,7 @@ cabal-version: 2.2 name: cabal-cache -version: 1.0.5.2 +version: 1.0.5.3 synopsis: CI Assistant for Haskell projects description: CI Assistant for Haskell projects. Implements package caching. homepage: https://github.com/haskell-works/cabal-cache diff --git a/src/HaskellWorks/CabalCache/Concurrent/DownloadQueue.hs b/src/HaskellWorks/CabalCache/Concurrent/DownloadQueue.hs --- a/src/HaskellWorks/CabalCache/Concurrent/DownloadQueue.hs +++ b/src/HaskellWorks/CabalCache/Concurrent/DownloadQueue.hs @@ -2,6 +2,7 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TypeApplications #-} module HaskellWorks.CabalCache.Concurrent.DownloadQueue ( DownloadStatus(..) @@ -9,16 +10,19 @@ , runQueue ) where -import Control.Monad.Catch (MonadMask(..)) -import Control.Monad.IO.Class (MonadIO(..)) -import Data.Function ((&)) -import Data.Set ((\\)) +import Control.Monad.Catch (MonadMask(..)) +import Control.Monad.IO.Class (MonadIO(..)) +import Data.Function ((&)) +import Data.Set ((\\)) +import HaskellWorks.CabalCache.Show (tshow) import qualified Control.Concurrent.STM as STM import qualified Control.Monad.Catch as CMC import qualified Data.Relation as R import qualified Data.Set as S import qualified HaskellWorks.CabalCache.Concurrent.Type as Z +import qualified HaskellWorks.CabalCache.IO.Console as CIO +import qualified Network.AWS as AWS import qualified System.IO as IO data DownloadStatus = DownloadSuccess | DownloadFailure deriving (Eq, Show) @@ -26,41 +30,49 @@ createDownloadQueue :: [(Z.ProviderId, Z.ConsumerId)] -> STM.STM Z.DownloadQueue createDownloadQueue dependencies = do tDependencies <- STM.newTVar (R.fromList dependencies) - tUploading <- STM.newTVar S.empty + tDownloading <- STM.newTVar S.empty tFailures <- STM.newTVar S.empty return Z.DownloadQueue {..} takeReady :: Z.DownloadQueue -> STM.STM (Maybe Z.PackageId) takeReady Z.DownloadQueue {..} = do dependencies <- STM.readTVar tDependencies - uploading <- STM.readTVar tUploading + downloading <- STM.readTVar tDownloading failures <- STM.readTVar tFailures - let ready = R.ran dependencies \\ R.dom dependencies \\ uploading \\ failures + -- The packages that need to be downloaded. This set can shrink when packages + -- have been downloaded, or grow when a download unlocks another depdendency for + -- download. When downloads fail, they are not removed from the set, but are + -- tracked separatedly in failures. + let queued = R.ran dependencies \\ R.dom dependencies + -- Packages that are ready for download. These are packages that have been queued + -- but are not currently downloaded nor have failed download. + let ready = queued \\ downloading \\ failures + case S.lookupMin ready of Just packageId -> do - STM.writeTVar tUploading (S.insert packageId uploading) + STM.writeTVar tDownloading (S.insert packageId downloading) return (Just packageId) - Nothing -> if S.null (R.ran dependencies \\ R.dom dependencies \\ failures) + Nothing -> if S.null (queued \\ failures) then return Nothing else STM.retry commit :: Z.DownloadQueue -> Z.PackageId -> STM.STM () commit Z.DownloadQueue {..} packageId = do dependencies <- STM.readTVar tDependencies - uploading <- STM.readTVar tUploading + downloading <- STM.readTVar tDownloading - STM.writeTVar tUploading $ S.delete packageId uploading + STM.writeTVar tDownloading $ S.delete packageId downloading STM.writeTVar tDependencies $ R.withoutRan (S.singleton packageId) dependencies failDownload :: Z.DownloadQueue -> Z.PackageId -> STM.STM () failDownload Z.DownloadQueue {..} packageId = do - uploading <- STM.readTVar tUploading - failures <- STM.readTVar tFailures + downloading <- STM.readTVar tDownloading + failures <- STM.readTVar tFailures - STM.writeTVar tUploading $ S.delete packageId uploading - STM.writeTVar tFailures $ S.insert packageId failures + STM.writeTVar tDownloading $ S.delete packageId downloading + STM.writeTVar tFailures $ S.insert packageId failures runQueue :: (MonadIO m, MonadMask m) => Z.DownloadQueue -> (Z.PackageId -> m DownloadStatus) -> m () runQueue downloadQueue f = do @@ -69,10 +81,13 @@ case maybePackageId of Just packageId -> do downloadStatus <- f packageId - & CMC.handleAll \e -> do - liftIO $ IO.hPutStrLn IO.stderr $ "Exception during download: " <> show e - liftIO $ IO.hFlush IO.stderr - CMC.throwM e + & do CMC.handle @_ @AWS.Error \e -> do + liftIO $ CIO.hPutStrLn IO.stderr $ "Failed download due to exception: " <> tshow e + pure DownloadFailure + & do CMC.handleAll \e -> do + liftIO $ CIO.hPutStrLn IO.stderr $ "Aborting due to unexpected exception during download: " <> tshow e + liftIO $ STM.atomically $ failDownload downloadQueue packageId + CMC.throwM e case downloadStatus of DownloadSuccess -> do liftIO $ STM.atomically $ commit downloadQueue packageId DownloadFailure -> do liftIO $ STM.atomically $ failDownload downloadQueue packageId diff --git a/src/HaskellWorks/CabalCache/Concurrent/Fork.hs b/src/HaskellWorks/CabalCache/Concurrent/Fork.hs --- a/src/HaskellWorks/CabalCache/Concurrent/Fork.hs +++ b/src/HaskellWorks/CabalCache/Concurrent/Fork.hs @@ -1,17 +1,31 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE OverloadedStrings #-} + module HaskellWorks.CabalCache.Concurrent.Fork where -import Control.Exception (finally) +import Control.Exception (finally) import Control.Monad +import HaskellWorks.CabalCache.Show (tshow) -import qualified Control.Concurrent as IO -import qualified Control.Concurrent.STM as STM +import qualified Control.Concurrent as IO +import qualified Control.Concurrent.STM as STM +import qualified HaskellWorks.CabalCache.IO.Console as CIO +import qualified System.IO as IO forkThreadsWait :: Int -> IO () -> IO () forkThreadsWait n f = do tDone <- STM.atomically $ STM.newTVar (0 :: Int) - forM_ [1 .. n] $ \_ -> IO.forkIO $ do - f `finally` (STM.atomically $ STM.modifyTVar tDone (+1)) - - STM.atomically $ do - done <- STM.readTVar tDone - when (done < n) STM.retry + forM_ [1 .. n] $ \_ -> IO.forkFinally + do f `finally` STM.atomically (STM.modifyTVar tDone (+1)) + \case + Left e -> do + CIO.hPutStrLn IO.stderr $ "Fatal exception occurred: " <> tshow e + CIO.hPutStrLn IO.stderr "Waiting for any remaining downlad threads to complete" + STM.atomically $ do + done <- STM.readTVar tDone + when (done < n) STM.retry + Right () -> do + STM.atomically $ do + done <- STM.readTVar tDone + when (done < n) STM.retry diff --git a/src/HaskellWorks/CabalCache/Concurrent/Type.hs b/src/HaskellWorks/CabalCache/Concurrent/Type.hs --- a/src/HaskellWorks/CabalCache/Concurrent/Type.hs +++ b/src/HaskellWorks/CabalCache/Concurrent/Type.hs @@ -20,6 +20,6 @@ data DownloadQueue = DownloadQueue { tDependencies :: STM.TVar (R.Relation ConsumerId ProviderId) - , tUploading :: STM.TVar (S.Set PackageId) + , tDownloading :: STM.TVar (S.Set PackageId) , tFailures :: STM.TVar (S.Set PackageId) } deriving Generic