Implement GNU jobserver posix client support

CMakeLists.txt

@@ -169,7 +169,10 @@ if(WIN32)
 	# errors by telling windows.h to not define those two.
 	add_compile_definitions(NOMINMAX)
 else()
-	target_sources(libninja PRIVATE src/subprocess-posix.cc)
+	target_sources(libninja PRIVATE
+		src/jobserver-posix.cc
+		src/subprocess-posix.cc
+	)
 	if(CMAKE_SYSTEM_NAME STREQUAL "OS400" OR CMAKE_SYSTEM_NAME STREQUAL "AIX")
 		target_sources(libninja PRIVATE src/getopt.c)
 		# Build getopt.c, which can be compiled as either C or C++, as C++

configure.py

@@ -564,6 +564,7 @@ def has_re2c() -> bool:
         objs += cxx('minidump-win32', variables=cxxvariables)
     objs += cc('getopt')
 else:
+    objs += cxx('jobserver-posix')
     objs += cxx('subprocess-posix')
 if platform.is_aix():
     objs += cc('getopt')

src/build.cc

@@ -163,6 +163,16 @@ Edge* Plan::FindWork() {
     return NULL;
 
   Edge* work = ready_.top();
+
+  // Only initiate work if the jobserver client can acquire a token.
+  if (builder_ && builder_->jobserver_ &&
+      builder_->jobserver_->Enabled()) {
+    int job_tokens = builder_->jobserver_->Tokens();
+    work->job_token_ = builder_->jobserver_->Acquire();
+    if (job_tokens == builder_->jobserver_->Tokens())
+      return NULL;
+  }
+
   ready_.pop();
   return work;
 }
@@ -199,6 +209,10 @@ bool Plan::EdgeFinished(Edge* edge, EdgeResult result, string* err) {
     edge->pool()->EdgeFinished(*edge);
   edge->pool()->RetrieveReadyEdges(&ready_);
 
+  // If jobserver is used, return the token for this job.
+  if (builder_ && builder_->jobserver_)
+    builder_->jobserver_->Release(&edge->job_token_);
+
   // The rest of this function only applies to successful commands.
   if (result != kEdgeSucceeded)
     return true;
@@ -592,14 +606,18 @@ void Plan::Dump() const {
 }
 
 struct RealCommandRunner : public CommandRunner {
-  explicit RealCommandRunner(const BuildConfig& config) : config_(config) {}
+  explicit RealCommandRunner(const BuildConfig& config, Jobserver* jobserver) :
+      config_(config), jobserver_(jobserver) {}
+
   size_t CanRunMore() const override;
   bool StartCommand(Edge* edge) override;
   bool WaitForCommand(Result* result) override;
   vector<Edge*> GetActiveEdges() override;
+  void ClearJobTokens(const std::vector<Edge*>&) override;
   void Abort() override;
 
   const BuildConfig& config_;
+  Jobserver* jobserver_;
   SubprocessSet subprocs_;
   map<const Subprocess*, Edge*> subproc_to_edge_;
 };
@@ -612,7 +630,13 @@ vector<Edge*> RealCommandRunner::GetActiveEdges() {
   return edges;
 }
 
+void RealCommandRunner::ClearJobTokens(const std::vector<Edge*> &edges) {
+  for (Edge* edge : edges)
+    jobserver_->Release(&edge->job_token_);
+}
+
 void RealCommandRunner::Abort() {
+  ClearJobTokens(GetActiveEdges());
   subprocs_.Clear();
 }
 
@@ -628,6 +652,14 @@ size_t RealCommandRunner::CanRunMore() const {
       capacity = load_capacity;
   }
 
+  // When initialized, behave as if the implicit token is acquired already.
+  // Otherwise, this happens after a token is released but before it is replaced,
+  // so the base capacity is represented by job_tokens + 1 when positive.
+  // Add an extra loop on capacity for each job in order to get an extra token.
+  int job_tokens = jobserver_->Tokens();
+  if (job_tokens)
+    capacity = abs(job_tokens) - subproc_number + 2;
+
   if (capacity < 0)
     capacity = 0;
 
@@ -667,10 +699,10 @@ bool RealCommandRunner::WaitForCommand(Result* result) {
   return true;
 }
 
-Builder::Builder(State* state, const BuildConfig& config, BuildLog* build_log,
-                 DepsLog* deps_log, DiskInterface* disk_interface,
-                 Status* status, int64_t start_time_millis)
-    : state_(state), config_(config), plan_(this), status_(status),
+Builder::Builder(State* state, const BuildConfig& config, Jobserver* jobserver,
+                 BuildLog* build_log, DepsLog* deps_log, DiskInterface* disk_interface,
+                 Status* status, int64_t start_time_millis) : state_(state),
+      config_(config), jobserver_(jobserver), plan_(this), status_(status),
       start_time_millis_(start_time_millis), disk_interface_(disk_interface),
       explanations_(g_explaining ? new Explanations() : nullptr),
       scan_(state, build_log, deps_log, disk_interface,
@@ -775,7 +807,7 @@ bool Builder::Build(string* err) {
     if (config_.dry_run)
       command_runner_.reset(new DryRunCommandRunner);
     else
-      command_runner_.reset(new RealCommandRunner(config_));
+      command_runner_.reset(new RealCommandRunner(config_, jobserver_));
   }
 
   // We are about to start the build process.

src/build.h

@@ -24,6 +24,7 @@
 #include "depfile_parser.h"
 #include "exit_status.h"
 #include "graph.h"
+#include "jobserver.h"
 #include "util.h"  // int64_t
 
 struct BuildLog;
@@ -161,6 +162,7 @@ struct CommandRunner {
   virtual bool WaitForCommand(Result* result) = 0;
 
   virtual std::vector<Edge*> GetActiveEdges() { return std::vector<Edge*>(); }
+  virtual void ClearJobTokens(const std::vector<Edge*>&) {}
   virtual void Abort() {}
 };
 
@@ -187,9 +189,9 @@ struct BuildConfig {
 
 /// Builder wraps the build process: starting commands, updating status.
 struct Builder {
-  Builder(State* state, const BuildConfig& config, BuildLog* build_log,
-          DepsLog* deps_log, DiskInterface* disk_interface, Status* status,
-          int64_t start_time_millis);
+  Builder(State* state, const BuildConfig& config, Jobserver* jobserver,
+          BuildLog* build_log, DepsLog* deps_log, DiskInterface* disk_interface,
+          Status* status, int64_t start_time_millis);
   ~Builder();
 
   /// Clean up after interrupted commands by deleting output files.
@@ -224,6 +226,7 @@ struct Builder {
 
   State* state_;
   const BuildConfig& config_;
+  Jobserver* jobserver_;
   Plan plan_;
   std::unique_ptr<CommandRunner> command_runner_;
   Status* status_;