feed-gipeda 0.2.0.0 → 0.3.0.0
raw patch · 11 files changed
+173/−143 lines, 11 filesdep +distributed-process-p2pdep +extradep −distributed-process-simplelocalnetdep ~tasty-hspec
Dependencies added: distributed-process-p2p, extra
Dependencies removed: distributed-process-simplelocalnet
Dependency ranges changed: tasty-hspec
Files
- README.md +6/−4
- app/Main.hs +30/−23
- feed-gipeda.cabal +11/−5
- src/FeedGipeda.hs +44/−48
- src/FeedGipeda/Master.hs +1/−1
- src/FeedGipeda/Master/File.hs +2/−0
- src/FeedGipeda/Master/Finalize.hs +2/−2
- src/FeedGipeda/TaskScheduler.hs +55/−40
- src/FeedGipeda/Types.hs +12/−10
- tests/Acceptance.hs +6/−6
- tests/Acceptance/Driver.hs +4/−4
README.md view
@@ -88,9 +88,11 @@ Enter watch mode. Watch for changes to config and re-fetch repositories every 5 seconds. Read config from the default location, act as both master and slave node - `feed-gipeda --check --config=~/.feed-gipeda/feed-gipeda.yaml` Enter check mode. Check the specified config file for syntax errors. Useful in a CI setting.-- `feed-gipeda --master=localhost:12345` -Enter default mode. Dispatch benchmark requests on registered slave nodes, don't work on them in this process-- `feed-gipeda --slave=localhost:12346` -Enter slave-only mode. Listen via multicast for master nodes which request you to do benchmarks.+- `feed-gipeda --master=12345` +Enter default mode. Dispatch benchmark requests on registered slave nodes, don't work on them in this process.+- `feed-gipeda --slave=12346:127.0.0.1:12345` +Enter slave-only mode, listen on 127.0.0.1:12346. Periodically try+to register at the master node located at 127.0.0.1:12345 which requests you to+do benchmarks. - `feed-gipeda --deploy-to=deploymentDir/` Enter default mode. Deploy changes via `rsync` to the local `deploymentDir`
app/Main.hs view
@@ -6,7 +6,9 @@ import Control.Logging as Logging import Control.Monad (join) import Data.Functor-import Data.List (elemIndex)+import Data.List (intercalate, last, init)+import Data.List.Extra (split)+import Data.Monoid ((<>)) import FeedGipeda import Options.Applicative import System.Directory (getAppUserDataDirectory)@@ -14,18 +16,6 @@ import Text.Read (readMaybe) -endpoint :: ReadM Endpoint-endpoint = do- s <- str- case elemIndex ':' s of- Nothing -> readerError "Expected a colon separator"- Just idx -> do- let (host, port') = splitAt idx s- case readMaybe (drop 1 port') of- Just port -> return (Endpoint host port)- Nothing -> readerError "Port was not integral"-- paths :: FilePath -> Parser Paths paths defaultConfig = Paths@@ -73,22 +63,39 @@ <> help "ssh or local path under which to deploy site/ folders with rsync") +slave :: ReadM ProcessRole+slave = do+ s <- str+ case split (== ':') s of+ sport : rest | length rest > 1 -> -- support IPv6+ case (readMaybe sport, readMaybe mport) of+ (Just sp, Just mp) -> return (Slave sp (Endpoint mhost mp))+ (Nothing, _) -> readerError "Slave port was not integral"+ (_, Nothing) -> readerError "Master port was not integral"+ where+ mport = last rest+ mhost = intercalate ":" (init rest)+ _ -> readerError "Expected 3 sections separated by a colon"+ processRole :: Parser ProcessRole processRole = impl- <$> optional (option endpoint+ <$> optional (option auto (long "master"- <> metavar "ENDPOINT"- <> help "Start in master mode, distributing work items. Identified via the given TCP endpoint (ipadress:portnumber)."))- <*> optional (option endpoint+ <> metavar "PORT"+ <> help "Start in master mode, distributing work items. Identified via the given TCP port number."))+ <*> optional (option slave (long "slave"- <> metavar "ENDPOINT"- <> help "Start in slave mode, requesting work items from a master node. Identified via the given TCP endpoint (ipadress:portnumber)."))+ <> metavar "PORT:HOST:PORT"+ <> help "Start in slave mode, requesting work items from a master node. Identified via the given local slave port and the name and port of the master node.")) where- impl Nothing Nothing = Both (Endpoint "localhost" 1337) (Endpoint "localhost" 1338)- impl (Just ep) Nothing = Master ep- impl Nothing (Just ep) = Slave ep- impl (Just mep) (Just sep) = Both mep sep+ impl Nothing Nothing = Both 1337 1338+ impl (Just port) Nothing = Master port+ impl Nothing (Just s) = s+ impl (Just master) (Just (Slave slave ep)) =+ if port ep == master && host ep `elem` ["localhost", "127.0.0.1", "::1"]+ then Both master slave+ else error "Contradiction in --master and --slave args!" -- not proud of this verbosity :: Parser Verbosity
feed-gipeda.cabal view
@@ -1,10 +1,12 @@ name: feed-gipeda-version: 0.2.0.0+version: 0.3.0.0 synopsis: CI service around gipeda description:- A service for easy handling of multiple repositories with <https://hackage.haskell.org/package/gipeda gipeda>.+ A service for easy handling of multiple repositories with+ <https://hackage.haskell.org/package/gipeda gipeda>. .- See @--help@ for usage. Example invocation for benchmarking the whole of the @Pipes@ library:+ See @--help@ for usage. Example invocation for benchmarking the whole of the+ @Pipes@ library: . @ $ cat > feed-gipeda.yaml@@ -25,6 +27,7 @@ tested-with: GHC == 7.8.4 , GHC == 7.10.2 , GHC == 7.10.3+ , GHC == 8.0.1 extra-source-files: assets/default_index.html , README.md @@ -69,7 +72,7 @@ , distributed-process-extras >= 0.1 && < 0.3 , distributed-process-client-server >= 0.1 && < 0.3 , distributed-process-async >= 0.1 && < 0.3- , distributed-process-simplelocalnet >= 0.2 && < 0.3+ , distributed-process-p2p >= 0.1 && < 0.2 -- we need makeRelativeToProject for CI builds , file-embed >= 0.0.10 , reactive-banana >= 1 && < 1.2@@ -90,6 +93,8 @@ ghc-options: -threaded build-depends: base >= 4.6 && < 5 , feed-gipeda+ -- @Data.List.Extra.split@+ , extra < 2 , directory >= 1.2.0.0 && < 1.3.0.0 , filepath >= 1.4.0.0 && < 1.5.0.0 , optparse-applicative >= 0.11@@ -111,7 +116,8 @@ , tasty-quickcheck , tasty-smallcheck , tasty-hunit- , tasty-hspec+ -- tasty-hspec-1.1 has broken bounds on hspec+ , tasty-hspec >= 1.1.1 , HUnit , feed-gipeda , temporary >= 1.1
src/FeedGipeda.hs view
@@ -9,43 +9,36 @@ ) where -import Control.Arrow (second)-import Control.Concurrent (forkIO)-import Control.Concurrent.Chan (Chan,- newChan,- readChan,- writeChan)-import Control.Distributed.Process (Process,- RemoteTable,- liftIO,- say,- spawnLocal)-import qualified Control.Distributed.Process.Backend.SimpleLocalnet as SLN-import Control.Distributed.Process.Node (initRemoteTable,- runProcess)-import Control.Logging as Logging-import Control.Monad (forever,- void, when)-import Data.List (elemIndex)-import Data.Maybe (fromMaybe,- isJust)-import Data.Set (Set)-import Data.Time (NominalDiffTime)-import qualified FeedGipeda.Config as Config-import qualified FeedGipeda.Gipeda as Gipeda-import FeedGipeda.GitShell (SHA)-import qualified FeedGipeda.Master as Master-import qualified FeedGipeda.Master.CommitQueue as CommitQueue-import qualified FeedGipeda.Master.File as Master.File+import Control.Arrow (second)+import Control.Concurrent (forkIO)+import Control.Concurrent.Chan (Chan, newChan, readChan,+ writeChan)+import qualified Control.Distributed.Backend.P2P as P2P+import Control.Distributed.Process (Process, RemoteTable,+ getSelfNode, liftIO, say,+ spawnLocal)+import Control.Distributed.Process.Node (initRemoteTable, runProcess)+import Control.Logging as Logging+import Control.Monad (forever, void, when)+import Data.List (elemIndex)+import Data.Maybe (fromMaybe, isJust)+import Data.Set (Set)+import Data.Time (NominalDiffTime)+import qualified FeedGipeda.Config as Config+import qualified FeedGipeda.Gipeda as Gipeda+import FeedGipeda.GitShell (SHA)+import qualified FeedGipeda.Master as Master+import qualified FeedGipeda.Master.CommitQueue as CommitQueue+import qualified FeedGipeda.Master.File as Master.File import FeedGipeda.Prelude-import FeedGipeda.Repo (Repo)-import qualified FeedGipeda.TaskScheduler as TaskScheduler-import qualified FeedGipeda.THGenerated as THGenerated+import FeedGipeda.Repo (Repo)+import qualified FeedGipeda.TaskScheduler as TaskScheduler+import qualified FeedGipeda.THGenerated as THGenerated import FeedGipeda.Types-import Network.URI (parseURI)-import System.Directory (getAppUserDataDirectory)-import System.Exit (exitSuccess)-import System.FilePath ((</>))+import Network.URI (parseURI)+import System.Directory (getAppUserDataDirectory)+import System.Exit (exitSuccess)+import System.FilePath ((</>)) remoteTable :: RemoteTable@@ -69,7 +62,7 @@ -> IO () feedGipeda paths cmd deployment role_ verbosity = do case verbosity of- Verbose -> Logging.setLogLevel Logging.LevelDebug+ Verbose -> Logging.setLogLevel Logging.LevelDebug NotVerbose -> Logging.setLogLevel Logging.LevelWarn case cmd of@@ -78,26 +71,29 @@ Config.checkFile (configFile paths) >>= maybe exitSuccess error Build mode timeout -> do case slaveEndpoint role_ of- Just (Endpoint host port) -> do+ Just (Endpoint shost sport) -> do let run = if isBoth role_ then void . forkIO else id- run $ do- backend <- SLN.initializeBackend host (show port) remoteTable- TaskScheduler.work backend+ Endpoint mhost mport = masterEndpoint role_+ master = P2P.makeNodeId (mhost ++ ":" ++ show mport)+ run (TaskScheduler.work shost (show sport) master remoteTable) _ -> return () - case masterEndpoint role_ of- Nothing -> return ()- Just (Endpoint host port) -> do- backend <- SLN.initializeBackend host (show port) remoteTable- node <- SLN.newLocalNode backend+ case (role_, masterEndpoint role_) of+ (Slave _ _, _) -> return ()+ (_, Endpoint host port) -> do queue <- CommitQueue.new- runProcess node $ do+ P2P.bootstrap host (show port) [] remoteTable $ do+ -- We supply no seeds, so the node won't have been created yet.+ -- I think this is a bug.+ _ <- getSelfNode let toTask :: (Repo, SHA) -> IO (TaskScheduler.Task String) toTask (repo, commit) = do script <- Gipeda.determineBenchmarkScript repo- return (THGenerated.stringDict, THGenerated.benchmarkClosure script repo commit timeout, Master.File.writeBenchmarkCSV repo commit . fromMaybe "")+ let closure = THGenerated.benchmarkClosure script repo commit timeout+ let finalize = Master.File.writeBenchmarkCSV repo commit . fromMaybe ""+ return (THGenerated.stringDict, closure, finalize) - TaskScheduler.start backend (CommitQueue.dequeue queue >>= toTask)+ TaskScheduler.start (CommitQueue.dequeue queue >>= toTask) liftIO (Master.checkForNewCommits paths deployment mode queue)
src/FeedGipeda/Master.hs view
@@ -250,5 +250,5 @@ exit <- (&&) <$> Event.isSet notBenchmarking <*> Event.isSet notFinalizing unless exit detectIdle case mode of- Once -> detectIdle+ Once -> detectIdle WatchForChanges _ -> Event.new >>= Event.wait -- block indefinitely
src/FeedGipeda/Master/File.hs view
@@ -9,6 +9,7 @@ ) where +import Control.Concurrent (threadDelay) import Data.Char (toLower) import Data.Functor import Data.List (isSuffixOf)@@ -59,6 +60,7 @@ cwd <- getCurrentDirectory resultsDir <- Repo.resultsDir repo createDirectoryIfMissing True resultsDir+ threadDelay 1000 writeFile (resultsDir </> commit <.> "csv") result
src/FeedGipeda/Master/Finalize.hs view
@@ -48,7 +48,7 @@ logDebug (takeBaseName executable ++ ": " ++ show exitCode) case exitCode of ExitFailure _ -> logDebug stderr >> logDebug stdout- _ -> return ()+ _ -> return () -- That's too much even for debug --logDebug "stdout:" --logDebug stdout@@ -74,7 +74,7 @@ if isJust first then do saveSettings repo- executeIn (Just project) gipeda ["-j", "--keep-going", "--always-make"]+ executeIn (Just project) gipeda ["--always-make"] rsyncSite repos repo deployment return () else
src/FeedGipeda/TaskScheduler.hs view
@@ -16,35 +16,35 @@ ) where -import Control.Concurrent (threadDelay)-import Control.Concurrent.MSemN (MSemN)-import qualified Control.Concurrent.MSemN as MSemN-import Control.Distributed.Process hiding- (call)+import Control.Concurrent (threadDelay)+import Control.Concurrent.MSemN (MSemN)+import qualified Control.Concurrent.MSemN as MSemN+import qualified Control.Distributed.Backend.P2P as P2P+import Control.Distributed.Process hiding (call) import Control.Distributed.Process.Async-import Control.Distributed.Process.Backend.SimpleLocalnet (Backend (..),- findSlaves,- startSlave)-import Control.Distributed.Process.Extras hiding- (call,- send)+import Control.Distributed.Process.Extras hiding (call, send) import Control.Distributed.Process.Extras.Time-import Control.Distributed.Process.ManagedProcess+import Control.Distributed.Process.ManagedProcess hiding (forkProcess,+ runProcess)+import Control.Distributed.Process.Node (forkProcess,+ runProcess) import Control.Distributed.Process.Serializable-import Control.Monad (forever)-import Data.Binary (Binary)+import Control.Monad (forever)+import Data.Binary (Binary)+import qualified Data.ByteString as BS import Data.Functor-import Data.Map (Map)-import qualified Data.Map as Map-import Data.Maybe (isNothing)-import Data.Sequence (Seq,- ViewL (..),- (<|), (|>))-import qualified Data.Sequence as Seq-import Data.Set (Set)-import qualified Data.Set as Set-import Data.Typeable (Typeable)-import GHC.Generics (Generic)+import Data.Map (Map)+import qualified Data.Map as Map+import Data.Maybe (isNothing)+import Data.Sequence (Seq, ViewL (..),+ (<|), (|>))+import qualified Data.Sequence as Seq+import Data.Set (Set)+import qualified Data.Set as Set+import Data.String (fromString)+import Data.Typeable (Typeable)+import FeedGipeda.Prelude+import GHC.Generics (Generic) {-| A @Task@ contains all information to execute a process on a remote@@ -99,13 +99,12 @@ -} start :: forall a r . Serializable a- => Backend- -> IO (Task a)+ => IO (Task a) -> Process ()-start backend awaitTask = do+start awaitTask = do idleSlaves <- liftIO (MSemN.new 0) q <- spawnLocal (queue idleSlaves)- spawnLocal (slaveDiscovery backend q)+ spawnLocal (slaveDiscovery q) spawnLocal (dispatch idleSlaves q) return () where@@ -151,7 +150,7 @@ assignment = do node <- fst <$> Set.minView idle (ref, task) <- case Seq.viewl onHold of- EmptyL -> Nothing+ EmptyL -> Nothing head :< _ -> Just head return (node, ref, task) in@@ -201,7 +200,7 @@ continue qs' AsyncPending -> fail "Waited for an async task, but still pending" _ -> do- say (show node ++ " failed. Reassigning task.")+ logInfo (show node ++ " failed. Reassigning task.") -- (temporarily) blacklist the failing node liftIO (MSemN.signal idleSlaves (-1)) qs' <- assignTasks qs@@ -225,14 +224,30 @@ newSlaves = Map.fromSet (const Nothing) slaveSet - slaveDiscovery :: Backend -> ProcessId -> Process ()- slaveDiscovery backend queue = forever $ do+ slaveDiscovery :: ProcessId -> Process ()+ slaveDiscovery queue = forever $ do self <- getSelfNode- slaves <- Set.fromList . map processNodeId <$> findSlaves backend- --say $ show slaves- cast queue (SlaveListChanged (Set.delete self slaves))-+ slaves <- Set.delete self . Set.fromList <$> P2P.getPeers+ cast queue (SlaveListChanged slaves)+ liftIO (threadDelay 2000000) --- | Register as a slave node and request tasks from the master node. Blocks.-work :: Backend -> IO ()-work = startSlave+{-| Register as a slave node and request tasks from the master node. Blocks.+ Slave discovery is done in a P2P fashion. We have a star topology with the+ master node at the center.+-}+work+ :: String+ -- ^ Host name of the local node, e.g. its IP address.+ -> String+ -- ^ Port number of the local node.+ -> NodeId+ -- ^ @NodeId@ of the master node.+ -> RemoteTable+ -> IO ()+work host service master rt =+ P2P.bootstrap host service [master] rt . forever $ do+ liftIO (threadDelay 2000000)+ -- This will try to reach and register with the master node. It's necessary+ -- in case we lost our master.+ -- A little unfortunate that we have to hardcode this...+ whereisRemoteAsync master "P2P:Controller"
src/FeedGipeda/Types.hs view
@@ -58,21 +58,23 @@ -- | Whether the current feed-gipeda process is a master node, slave node, or both. data ProcessRole- = Master Endpoint- | Slave Endpoint- | Both Endpoint Endpoint+ = Master Int+ -- ^ local port+ | Slave Int Endpoint+ -- ^ local port, remote endpoint of the master node+ | Both Int Int+ -- ^ local port for master, local port for slave deriving (Show, Eq) --masterEndpoint :: ProcessRole -> Maybe Endpoint-masterEndpoint (Master ep) = Just ep-masterEndpoint (Both ep _) = Just ep-masterEndpoint _ = Nothing+masterEndpoint :: ProcessRole -> Endpoint+masterEndpoint (Master port) = Endpoint "127.0.0.1" port+masterEndpoint (Both masterPort _) = Endpoint "127.0.0.1" masterPort+masterEndpoint (Slave _ ep) = ep slaveEndpoint :: ProcessRole -> Maybe Endpoint-slaveEndpoint (Slave ep) = Just ep-slaveEndpoint (Both _ ep) = Just ep+slaveEndpoint (Slave port _) = Just (Endpoint "127.0.0.1" port)+slaveEndpoint (Both _ slavePort) = Just (Endpoint "127.0.0.1" slavePort) slaveEndpoint _ = Nothing
tests/Acceptance.hs view
@@ -134,15 +134,15 @@ (path, stdout, stderr, handle) <- Files.withWellFormedConfig >>= Driver.withMasterInTmpDir 12345 withAssertNotExit handle- spawnSlave 12346- spawnSlave 12347- spawnSlave 12348- spawnSlave 12349+ spawnSlave 12345 12346+ spawnSlave 12345 12347+ spawnSlave 12345 12348+ spawnSlave 12345 12349 assertCsvFilesChangeWithin waitSeconds path ] where- spawnSlave port = do- (_, _, h) <- Driver.withSlave port+ spawnSlave mport sport = do+ (_, _, h) <- Driver.withSlave mport sport withAssertNotExit h
tests/Acceptance/Driver.hs view
@@ -86,9 +86,9 @@ } -withSlave :: Int -> Managed (Source IO ByteString, Source IO ByteString, StreamingProcessHandle)-withSlave port =- withProcess (proc "feed-gipeda" ["--slave", "localhost:" ++ show port])+withSlave :: Int -> Int -> Managed (Source IO ByteString, Source IO ByteString, StreamingProcessHandle)+withSlave mp sp =+ withProcess (proc "feed-gipeda" ["--slave", show sp ++ ":127.0.0.1:" ++ show mp]) withProcess :: CreateProcess -> Managed (Source IO ByteString, Source IO ByteString, StreamingProcessHandle)@@ -117,7 +117,7 @@ when check (tell ["--check"]) whenJust deploymentDir $ \r -> tell ["--deploy-to", r] whenJust watch $ \dt -> tell ["--watch", show dt]- whenJust masterPort $ \p -> tell ["--master", "localhost:" ++ show p]+ whenJust masterPort $ \p -> tell ["--master", show p] return () path <- managed (withSystemTempDirectory "feed-gipeda")