feed-gipeda 0.1.0.2 → 0.1.0.3
raw patch · 6 files changed
+88/−48 lines, 6 filesdep ~asyncdep ~conduit-extra
Dependency ranges changed: async, conduit-extra
Files
- feed-gipeda.cabal +6/−1
- src/FeedGipeda.hs +2/−2
- src/FeedGipeda/Gipeda.hs +5/−1
- src/FeedGipeda/Slave.hs +59/−14
- src/FeedGipeda/THGenerated.hs +7/−6
- src/FeedGipeda/TaskQueue.hs +9/−24
feed-gipeda.cabal view
@@ -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 <https://hackage.haskell.org/package/gipeda gipeda>.@@ -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
src/FeedGipeda.hs view
@@ -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
src/FeedGipeda/Gipeda.hs view
@@ -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 =
src/FeedGipeda/Slave.hs view
@@ -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"
src/FeedGipeda/THGenerated.hs view
@@ -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)
src/FeedGipeda/TaskQueue.hs view
@@ -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