src/real_command_runner.cc

// Copyright 2011 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "build.h"
#include "build_config.h"
#include "build_result.h"
#include "disk_interface.h"
#include "exit_status.h"
#include "jobserver.h"
#include "limits.h"
#include "subprocess.h"

struct RealCommandRunner : public CommandRunner {
  explicit RealCommandRunner(const BuildConfig& config,
                             DiskInterface& disk_interface,
                             Jobserver::Client* jobserver)
      : config_(config), disk_interface_(disk_interface), jobserver_(jobserver),
        subprocs_(AsyncLoop::Get(), &config_.environment) {}
  size_t CanRunMore() const override;
  bool StartCommand(Edge* edge) override;
  BuildResult WaitForCommand() override;
  BuildResult WaitForCommandOrJobserverToken(bool watch_jobserver) override;
  std::vector<Edge*> GetActiveEdges() override;
  void Abort() override;

  SubprocessSet* GetSubprocessSet() override { return &subprocs_; }

  void ClearJobTokens() {
    if (jobserver_) {
      for (Edge* edge : GetActiveEdges()) {
        jobserver_->Release(std::move(edge->job_slot_));
      }
    }
  }

  const BuildConfig& config_;
  DiskInterface& disk_interface_;
  Jobserver::Client* jobserver_ = nullptr;
  SubprocessSet subprocs_;
  std::map<const Subprocess*, Edge*> subproc_to_edge_;
};

std::vector<Edge*> RealCommandRunner::GetActiveEdges() {
  std::vector<Edge*> edges;
  edges.reserve(subproc_to_edge_.size());
  for (const auto& pair : subproc_to_edge_)
    edges.push_back(pair.second);
  return edges;
}

void RealCommandRunner::Abort() {
  ClearJobTokens();
  subprocs_.Clear();
}

size_t RealCommandRunner::CanRunMore() const {
  size_t subproc_number =
      subprocs_.running_.size() + subprocs_.finished_.size();

  int64_t capacity = config_.parallelism - subproc_number;

  if (jobserver_) {
    // When a jobserver token pool is used, make the
    // capacity infinite, and let FindWork() limit jobs
    // through token acquisitions instead.
    capacity = INT_MAX;
  }

  if (config_.max_load_average > 0.0f) {
    int load_capacity = config_.max_load_average - GetLoadAverage();
    if (load_capacity < capacity)
      capacity = load_capacity;
  }

  if (capacity < 0)
    capacity = 0;

  if (capacity == 0 && subprocs_.running_.empty())
    // Ensure that we make progress.
    capacity = 1;

  return capacity;
}

bool RealCommandRunner::StartCommand(Edge* edge) {
  std::string command = edge->EvaluateCommand();
  std::string description = edge->GetBinding("description");
  if (description.empty())
    description = command;
  const Subprocess* subproc =
      subprocs_.Add(command, &description, edge->use_console());
  if (!subproc)
    return false;

  subproc_to_edge_.insert(std::make_pair(subproc, edge));
  return true;
}

BuildResult RealCommandRunner::WaitForCommand() {
  return WaitForCommandOrJobserverToken(false);
}

BuildResult RealCommandRunner::WaitForCommandOrJobserverToken(
    bool watch_jobserver) {
#ifndef _WIN32
  // Jobserver mode is enabled and runner is watching for tokens.
  if (jobserver_ && watch_jobserver) {
    subprocs_.SetJobserverFD(jobserver_->GetJobserverFD());
  } else {
    subprocs_.SetJobserverFD(-1);
  }
#endif  // !_WIN32

  SubprocessSet::WorkResult work_result = SubprocessSet::WorkResult::NoWork;
  if (subprocs_.HasFinished()) {
    work_result = SubprocessSet::WorkResult::SubprocFinished;
  }

  // Wait for DoWork() to report activity
  while (work_result == SubprocessSet::WorkResult::NoWork) {
    work_result = subprocs_.DoWork();
  }

  // Address interrupts first, then subprocesses finishing, then finally
  // jobserver token availability
  BuildResult build_result;
  switch (work_result) {
  case SubprocessSet::WorkResult::Interrupted:
    build_result = BuildResult::Interrupted{};
    break;
  case SubprocessSet::WorkResult::SubprocFinished: {
    std::unique_ptr<Subprocess> subproc = subprocs_.NextFinished();
    ExitStatus status = subproc->Finish();
    std::string output = subproc->GetOutput();

    auto e = subproc_to_edge_.find(subproc.get());
    Edge* edge = e->second;
    subproc_to_edge_.erase(e);
    build_result =
        BuildResult::CommandCompleted{ edge, status, std::move(output) };
    break;
  }
  case SubprocessSet::WorkResult::JobserverTokenAvailable:
    build_result = BuildResult::JobserverTokenAvailable{};
    break;
  default:
    break;
  }

  return build_result;
}

// static
CommandRunner* CommandRunner::factory(const BuildConfig& config,
                                      DiskInterface& disk_interface,
                                      Jobserver::Client* jobserver) {
  return new RealCommandRunner(config, disk_interface, jobserver);
}