diff --git a/feed-gipeda.cabal b/feed-gipeda.cabal --- a/feed-gipeda.cabal +++ b/feed-gipeda.cabal @@ -1,5 +1,5 @@ name: feed-gipeda -version: 0.1.0.2 +version: 0.1.0.3 synopsis: Simple project template from stack description: A service for easy handling of multiple repositories with . @@ -71,11 +71,16 @@ , logging >= 3.0.4 && < 3.1 , temporary >= 1.1 && < 1.3 , concurrent-extra < 0.8 + , conduit + , conduit-extra >= 1.1.2 && < 1.2 + -- Concurrently + , async >= 2.0.1 && < 2.2 executable feed-gipeda hs-source-dirs: app main-is: Main.hs default-language: Haskell2010 + ghc-options: -threaded build-depends: base >= 4.6 && < 5 , feed-gipeda , directory >= 1.2.0.0 && < 1.3.0.0 diff --git a/src/FeedGipeda.hs b/src/FeedGipeda.hs --- a/src/FeedGipeda.hs +++ b/src/FeedGipeda.hs @@ -89,14 +89,14 @@ node <- SLN.newLocalNode backend tasks <- newChan runProcess node $ do - taskQueue <- TaskQueue.start backend timeout + taskQueue <- TaskQueue.start backend spawnLocal $ forever $ do (finalize, benchmarkScript, repo, commit) <- liftIO (readChan tasks) spawnLocal $ do result <- TaskQueue.execute taskQueue THGenerated.stringDict - (THGenerated.benchmarkClosure benchmarkScript repo commit) + (THGenerated.benchmarkClosure benchmarkScript repo commit timeout) liftIO (finalize result) let diff --git a/src/FeedGipeda/Gipeda.hs b/src/FeedGipeda/Gipeda.hs --- a/src/FeedGipeda/Gipeda.hs +++ b/src/FeedGipeda/Gipeda.hs @@ -178,7 +178,11 @@ settingsForRepo repo = do clone <- Repo.cloneDir repo firstCommit <- GitShell.firstCommit clone - gipedaYaml <- GitShell.showHead clone "gipeda.yaml" + gipedaYaml <- + GitShell.showHead clone ".gipeda.yaml" + <|> GitShell.showHead clone ".gipeda.yml" + <|> GitShell.showHead clone "gipeda.yaml" + <|> GitShell.showHead clone "gipeda.yml" let yaml :: Yaml.Value yaml = diff --git a/src/FeedGipeda/Slave.hs b/src/FeedGipeda/Slave.hs --- a/src/FeedGipeda/Slave.hs +++ b/src/FeedGipeda/Slave.hs @@ -1,4 +1,4 @@ -{-| @benchmark@ contains the logic to be executed on slave nodes. +{-| @benchmark@ contains the logic to be executed on slave nodes. -} @@ -7,17 +7,53 @@ ) where -import Control.Logging as Logging -import qualified Data.Text as Text -import FeedGipeda.GitShell (SHA) -import FeedGipeda.Repo (Repo) -import qualified FeedGipeda.Repo as Repo -import System.Exit (ExitCode (..)) -import System.IO.Temp (withSystemTempDirectory) -import System.Process (cwd, proc, readCreateProcessWithExitCode, - shell, showCommandForUser) +import Control.Concurrent.Async (Concurrently (..)) +import Control.Exception (bracket) +import Control.Logging as Logging +import Data.Conduit (($$)) +import qualified Data.Conduit.List as CL +import Data.Conduit.Process (ClosedStream (..), + CreateProcess (..), + interruptProcessGroupOf, proc, + readCreateProcessWithExitCode, shell, + showCommandForUser, streamingProcess, + streamingProcessHandleRaw, + waitForStreamingProcess) +import Data.Monoid +import qualified Data.Text as Text +import qualified Data.Text.Encoding as Text +import FeedGipeda.GitShell (SHA) +import FeedGipeda.Repo (Repo) +import qualified FeedGipeda.Repo as Repo +import FeedGipeda.Types (Timeout) +import System.Exit (ExitCode (..)) +import System.IO.Temp (withSystemTempDirectory) +import qualified System.Timeout +-- We have to roll our own, because the provided functions use @terminateProcess@ +-- instead of @interruptProcessGroupOf@, so that in case of shelling out we only +-- kill the shell process but not its children. +readCreateProcessGroupWithExitCode :: CreateProcess -> IO (ExitCode, String, String) +readCreateProcessGroupWithExitCode cp = + bracket + (streamingProcess cp { create_group = True }) + cleanup + captureStreams + where + cleanup (_, _, _, sph) = + interruptProcessGroupOf (streamingProcessHandleRaw sph) + + captureStreams (ClosedStream, out, err, sph) = + runConcurrently $ (,,) + <$> Concurrently (waitForStreamingProcess sph) + <*> Concurrently (out $$ stringSink) + <*> Concurrently (err $$ stringSink) + + stringSink = + Text.unpack . Text.decodeUtf8 <$> CL.fold (<>) mempty + + procReportingError :: Repo -> SHA -> Maybe FilePath -> FilePath -> [String] -> IO String procReportingError repo commit cwd cmd args = do (exitCode, stdout, stderr) <- @@ -29,7 +65,7 @@ shellReportingError :: Repo -> SHA -> Maybe FilePath -> FilePath -> IO String shellReportingError repo commit cwd cmd = do (exitCode, stdout, stderr) <- - readCreateProcessWithExitCode (shell cmd) { cwd = cwd } "" + readCreateProcessGroupWithExitCode (shell cmd) { cwd = cwd } reportError repo commit cmd exitCode stderr return stdout @@ -60,10 +96,19 @@ Will be executed on slave nodes. -} -benchmark :: String -> Repo -> SHA -> IO String -benchmark benchmarkScript repo commit = do +benchmark :: String -> Repo -> SHA -> Timeout -> IO String +benchmark benchmarkScript repo commit timeout = do clone <- Repo.cloneDir repo Logging.log (Text.pack ("Benchmarking " ++ Repo.uri repo ++ "@" ++ commit)) withSystemTempDirectory "feed-gipeda" $ \cloneDir -> do cloneRecursiveAndCheckout repo commit cloneDir - shellReportingError repo commit (Just cloneDir) benchmarkScript + res <- System.Timeout.timeout (ceiling (timeout * 10^6)) $ + shellReportingError repo commit (Just cloneDir) benchmarkScript + case res of + Just res -> return res + Nothing -> do + Logging.warn . Text.pack . unlines $ + [ "Benchmark script timed out (--timeout is " ++ show timeout ++ ")" + , "At commit " ++ Repo.uri repo ++ "@" ++ commit + ] + return "build/timeout;1.0" diff --git a/src/FeedGipeda/THGenerated.hs b/src/FeedGipeda/THGenerated.hs --- a/src/FeedGipeda/THGenerated.hs +++ b/src/FeedGipeda/THGenerated.hs @@ -19,17 +19,18 @@ import FeedGipeda.GitShell (SHA) import FeedGipeda.Repo (Repo) import qualified FeedGipeda.Slave as Slave +import FeedGipeda.Types (Timeout) -benchmarkProcess :: (String, Repo, SHA) -> Process String -benchmarkProcess (benchmarkScript, repo, sha) = - liftIO (Slave.benchmark benchmarkScript repo sha) +benchmarkProcess :: (String, Repo, SHA, Rational) -> Process String +benchmarkProcess (benchmarkScript, repo, sha, timeout) = + liftIO (Slave.benchmark benchmarkScript repo sha (fromRational timeout)) remotable ['benchmarkProcess] -benchmarkClosure :: String -> Repo -> SHA -> Closure (Process String) -benchmarkClosure benchmarkScript repo commit = - $(mkClosure 'benchmarkProcess) (benchmarkScript, repo, commit) +benchmarkClosure :: String -> Repo -> SHA -> Timeout -> Closure (Process String) +benchmarkClosure benchmarkScript repo commit timeout = + $(mkClosure 'benchmarkProcess) (benchmarkScript, repo, commit, toRational timeout) stringDict :: Static (SerializableDict String) diff --git a/src/FeedGipeda/TaskQueue.hs b/src/FeedGipeda/TaskQueue.hs --- a/src/FeedGipeda/TaskQueue.hs +++ b/src/FeedGipeda/TaskQueue.hs @@ -43,7 +43,6 @@ import qualified Data.Sequence as Seq import Data.Set (Set) import qualified Data.Set as Set -import Data.Time (NominalDiffTime) import Data.Typeable (Typeable) import GHC.Generics (Generic) @@ -77,14 +76,13 @@ data QueueState a = QueueState - { slaves :: Map NodeId (Maybe MonitorRef) - , active :: Map MonitorRef (NodeId, Async a, CallRef (Maybe a), Task a) - , onHold :: Seq (CallRef (Maybe a), Task a) - , timeout :: NominalDiffTime + { slaves :: Map NodeId (Maybe MonitorRef) + , active :: Map MonitorRef (NodeId, Async a, CallRef (Maybe a), Task a) + , onHold :: Seq (CallRef (Maybe a), Task a) } -initialQueueState :: NominalDiffTime -> QueueState a +initialQueueState :: QueueState a initialQueueState = QueueState Map.empty Map.empty Seq.empty @@ -95,10 +93,9 @@ start :: forall a . Serializable a => Backend - -> NominalDiffTime -> Process (TaskQueue a) -start backend timeout = do - queue <- TaskQueue <$> spawnLocal (queue (Proxy :: Proxy a) timeout) +start backend = do + queue <- TaskQueue <$> spawnLocal (queue (Proxy :: Proxy a)) spawnLocal (slaveDiscovery backend queue) return queue @@ -120,13 +117,12 @@ queue :: forall a . Serializable a => Proxy a - -> NominalDiffTime -> Process () -queue _ timeout = serve () init process +queue _ = serve () init process where init :: InitHandler () (QueueState a) init () = - return (InitOk (initialQueueState timeout) Infinity) + return (InitOk initialQueueState Infinity) process :: ProcessDefinition (QueueState a) process = defaultProcess @@ -141,7 +137,7 @@ } assignTasks :: QueueState a -> Process (QueueState a) - assignTasks qs@(QueueState slaves active onHold timeout) = do + assignTasks qs@(QueueState slaves active onHold) = do let idle :: Set NodeId idle = @@ -160,10 +156,6 @@ Just (node, callRef, (dict, closure)) -> do handle <- async (remoteTask dict node closure) monitorRef <- monitorAsync handle - spawnLocal $ do - -- Not sure if this is the way to go - liftIO (threadDelay (ceiling (timeout * 1000000))) - cancel handle -- we will handle this in onTaskCompleted return qs { slaves = Map.insert node (Just monitorRef) slaves , onHold = Seq.drop 1 onHold @@ -200,13 +192,6 @@ , active = withoutRef } replyTo callRef (Just ret) - continue qs' - AsyncCancelled -> do -- we cancelled the task because of timeout. Don't reassign - qs' <- assignTasks qs - { slaves = Map.adjust (const Nothing) node (slaves qs) - , active = withoutRef - } - replyTo callRef Nothing continue qs' AsyncPending -> fail "Waited for an async task, but still pending" _ -> do