diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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` diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -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 diff --git a/feed-gipeda.cabal b/feed-gipeda.cabal --- a/feed-gipeda.cabal +++ b/feed-gipeda.cabal @@ -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 . + A service for easy handling of multiple repositories with + . . - 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 diff --git a/src/FeedGipeda.hs b/src/FeedGipeda.hs --- a/src/FeedGipeda.hs +++ b/src/FeedGipeda.hs @@ -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) diff --git a/src/FeedGipeda/Master.hs b/src/FeedGipeda/Master.hs --- a/src/FeedGipeda/Master.hs +++ b/src/FeedGipeda/Master.hs @@ -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 diff --git a/src/FeedGipeda/Master/File.hs b/src/FeedGipeda/Master/File.hs --- a/src/FeedGipeda/Master/File.hs +++ b/src/FeedGipeda/Master/File.hs @@ -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 diff --git a/src/FeedGipeda/Master/Finalize.hs b/src/FeedGipeda/Master/Finalize.hs --- a/src/FeedGipeda/Master/Finalize.hs +++ b/src/FeedGipeda/Master/Finalize.hs @@ -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 diff --git a/src/FeedGipeda/TaskScheduler.hs b/src/FeedGipeda/TaskScheduler.hs --- a/src/FeedGipeda/TaskScheduler.hs +++ b/src/FeedGipeda/TaskScheduler.hs @@ -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" diff --git a/src/FeedGipeda/Types.hs b/src/FeedGipeda/Types.hs --- a/src/FeedGipeda/Types.hs +++ b/src/FeedGipeda/Types.hs @@ -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 diff --git a/tests/Acceptance.hs b/tests/Acceptance.hs --- a/tests/Acceptance.hs +++ b/tests/Acceptance.hs @@ -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 diff --git a/tests/Acceptance/Driver.hs b/tests/Acceptance/Driver.hs --- a/tests/Acceptance/Driver.hs +++ b/tests/Acceptance/Driver.hs @@ -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")