src/delayed_worker.cc - third_party/github.com/ninja-build/ninja - Git at Google

 // Copyright 2025 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 "delayed_worker.h"

 #include <assert.h>
 #include <inttypes.h>
 #include <string.h>

 #include "disk_interface.h"
 #include "json_parser.h"
 #include "json_writer.h"
 #include "subprocess.h"
 #include "util.h"

 #define DEBUG 0

 #if DEBUG
 #include "scoped_defer.h"
 #define LOG(...) fprintf(stderr, __VA_ARGS__)
 #define LOG_ON_EXIT(...) ScopedDefer([&]() { fprintf(stderr, __VA_ARGS__); })
 #else
 #define LOG(...) (void)0
 #define LOG_ON_EXIT(...) (void)0
 #endif

 namespace {

 // Decode a single RequestInfo::Action. |parser| must wrap its JSON content.
 // On success, set |*action| and return true, on failure call
 // |parser.SetError()| and return false.
 bool DecodeRequestInfoAction(JsonParser& parser,
                              DelayedWorker::RequestInfo::Action* action) {
   JsonParser::KeyValueMap kv;
   if (!parser.ParseObject(&kv))
     return false;

   // Get action id first.
   int64_t id;
   if (!kv.GetRequiredInteger("action_id", &id, parser))
     return false;

   if (id <= 0 || id > INT32_MAX)
     return parser.SetErrorFormat(
         "action_id must be in 1..INT32_MAX range: %" PRId64, id);
   action->action_id = static_cast<uint32_t>(id);

   // Then the other fields.
   if (!kv.GetRequiredString("command", &action->command, parser) ||
       !kv.GetRequiredString("description", &action->description, parser) ||
       !kv.GetRequiredStringList("ninja_outputs", &action->ninja_outputs,
                                 parser) ||
       !kv.GetOptionalOrEmptyString("ninja_depfile", &action->ninja_depfile,
                                    parser))
     return false;

   return true;
 }

 bool DecodeRequest(JsonParser& parser, DelayedWorker::RequestInfo* result) {
   JsonParser::KeyValueMap kv;
   if (!parser.ParseObject(&kv))
     return false;

   // First check version.
   int64_t version = -1;
   if (!kv.GetRequiredInteger("version", &version, parser))
     return false;

   if (version != 2) {
     return parser.SetErrorFormat("Unknown version, expected 2, got %" PRId64,
                                  version);
   }

   int64_t request_id = -1;
   if (!kv.GetRequiredInteger("request_id", &request_id, parser))
     return false;

   if (request_id <= 0 || request_id > INT32_MAX)
     return parser.SetErrorFormat(
         "request_id must be in 1..INT32_MAX range: %" PRId64, request_id);
   result->request_id = static_cast<uint32_t>(request_id);

   // Then get each actions from the 'actions' list.
   JsonParser::ValueInfo vi;
   if (!kv.GetRequiredExpected("actions", JsonParser::kArrayStart, &vi, parser))
     return false;

   JsonParser actions_parser(vi.value);
   JsonParser::ArrayIterator actions_it(actions_parser);
   auto& result_actions = result->actions;
   result_actions.clear();
   while (actions_it.HasItem()) {
     /// Create new instance directly at the end of the vector, then
     /// tell called function to write to it directly.
     result_actions.emplace_back();
     if (!DecodeRequestInfoAction(actions_parser, &result_actions.back())) {
       result_actions.pop_back();
       break;
     }
     if (actions_parser.HasError())
       return parser.SetError(actions_parser.TakeError());
   }

   // Finally, handle the optional metadata.
   BuildMetadataMap& result_metadata = result->build_metadata;
   result_metadata.clear();
   if (kv.Get("build_metadata", &vi)) {
     JsonParser metadata_parser(vi.value);
     JsonParser::ObjectIterator metadata_it(metadata_parser);
     while (metadata_it.HasKeyValue()) {
       auto ret = result_metadata.try_emplace(metadata_it.key().AsString(), "");
       if (!metadata_parser.ExpectString(&ret.first->second)) {
         result_metadata.erase(ret.first);
         break;
       }
     }
     if (metadata_parser.HasError())
       return parser.SetError(metadata_parser.TakeError());
   }
   return true;
 }

 bool DecodeResponse(JsonParser& parser, DelayedWorker::ResponseInfo* result) {
   JsonParser::KeyValueMap top_kv;
   if (!parser.ParseObject(&top_kv))
     return false;

   // Check version first.
   int64_t version = -1;
   if (!top_kv.GetRequiredInteger("version", &version, parser))
     return false;
   if (version != 1) {
     return parser.SetErrorFormat("Invalid version %" PRId64 ", expected 1",
                                  version);
   }

   // Get all other fields, checking them individually.
   int64_t request_id = -1;
   if (!top_kv.GetRequiredInteger("request_id", &request_id, parser))
     return false;

   if (request_id <= 0 || request_id > INT32_MAX)
     return parser.SetErrorFormat(
         "request_id must be in 1..INT32_MAX range: %" PRId64, request_id);
   result->request_id = static_cast<uint32_t>(request_id);

   int64_t status = -1;
   if (!top_kv.GetRequiredInteger("status", &status, parser))
     return false;

   if (status < 0 || status >= 256)
     return parser.SetErrorFormat("status must be in 0..255 range: %" PRId64,
                                  status);
   result->status = static_cast<uint8_t>(status);

   if (!top_kv.GetRequiredString("output", &result->output, parser))
     return false;
   return true;
 }

 }  // namespace

 // static
 std::string DelayedWorker::RequestInfo::ToJson() const {
   std::string output;
   JsonWriter writer;
   {
     auto dict = writer.StartDict(output);
     dict->AddInt("version", 2);
     dict->AddInt("request_id", this->request_id);
     {
       auto actions = dict->AddList("actions");
       int action_id = 1;
       for (const auto& action : this->actions) {
         auto entry = actions->AddDict();
         entry->AddInt("action_id", action_id++);
         entry->AddString("command", action.command);
         entry->AddString("description", action.description);
         {
           auto outputs = entry->AddList("ninja_outputs");
           for (const auto& output : action.ninja_outputs)
             outputs->AddString(output);
         }
         if (!action.ninja_depfile.empty())
           entry->AddString("ninja_depfile", action.ninja_depfile);
       }
     }
     if (!this->build_metadata.empty()) {
       auto metadata = dict->AddDict("build_metadata");
       for (const auto& pair : this->build_metadata)
         metadata->AddString(pair.first, pair.second);
     }
   }
   return output;
 }

 // static
 bool DelayedWorker::RequestInfo::FromJson(StringPiece request_json,
                                           DelayedWorker::RequestInfo* result,
                                           std::string* error) {
   JsonParser parser(request_json);
   if (!DecodeRequest(parser, result)) {
     *error = parser.TakeError();
     return false;
   }
   return true;
 }

 // static
 std::string DelayedWorker::ResponseInfo::ToJson() const {
   std::string output;
   JsonWriter writer;
   {
     auto dict = writer.StartDict(output);
     dict->AddInt("version", 1);
     dict->AddInt("request_id", this->request_id);
     dict->AddInt("status", this->status);
     dict->AddString("output", this->output);
   }
   return output;
 }

 // static
 bool DelayedWorker::ResponseInfo::FromJson(StringPiece response_json,
                                            DelayedWorker::ResponseInfo* result,
                                            std::string* error) {
   JsonParser parser(response_json);
   if (!DecodeResponse(parser, result)) {
     *error = parser.TakeError();
     return false;
   }
   return true;
 }

 /// A DelayedWorkerRunner that runs commands in a SubprocessSet.
 struct SubprocessSetRunner : public DelayedWorkerRunner {
   SubprocessSetRunner(SubprocessSet& subprocess_set)
       : subprocess_set_(subprocess_set) {}

   ExitStatus Run(const std::string& tool, const std::string& tool_input_path,
                  const std::string& tool_output_path,
                  const std::string& tool_description) override {
     // Convenience helper to escape a string for the shell.
     auto escaped = [](const std::string& str) -> std::string {
       std::string result;
       GetShellEscapedString(str, &result);
       return result;
     };

     // Ensure the script is called from the current directory.
     // Only pass two arguments: paths of the input request file, and output
     // response file.
     std::string tool_command = "./" +
         tool + " " + escaped(tool_input_path) + " " + escaped(tool_output_path);

     // Launch single command in console mode.
     bool use_console = true;
     subprocess_set_.Add(tool_command, &tool_description, use_console);

     std::unique_ptr<Subprocess> subproc;
     while (!(subproc = subprocess_set_.NextFinished())) {
       bool interrupted =
           subprocess_set_.DoWork() == SubprocessSet::WorkResult::Interrupted;

       if (interrupted) {
         subprocess_set_.Clear();
         return ExitInterrupted;
       }
     }

     return subproc->Finish();
   }

   SubprocessSet& subprocess_set_;
 };

 // static
 std::unique_ptr<DelayedWorkerRunner> DelayedWorkerRunner::CreateFrom(
     SubprocessSet& subprocess_set) {
   return std::make_unique<SubprocessSetRunner>(subprocess_set);
 }

 // static
 bool DelayedWorker::RunTool(const std::string& tool_path,
                             const std::string& tool_description,
                             const RequestInfo& request,
                             DelayedWorkerRunner& tool_runner,
                             DiskInterface& disk_interface,
                             ResponseInfo* response, std::string* error) {
   static int s_invocation_id = 0;
   int invocation_id = ++s_invocation_id;
   LOG("DelayedWorker::RunTool: Starting new invocation_id=%d\n", invocation_id);
   LOG_ON_EXIT("DelayedWorker::RunTool: Exit for invocation_id=%d\n",
               invocation_id);

   std::string tool_input_path = StringFormat(
       "ninja_delayed_bazel_build_tool_in.%d.json", s_invocation_id);

   std::string tool_output_path = StringFormat(
       "ninja_delayed_bazel_build_tool_out.%d.json", s_invocation_id);

   if (!disk_interface.WriteFile(tool_input_path, request.ToJson(), false)) {
     *error = "failed to write delayed worker input file";
     return false;
   }

   ExitStatus status = tool_runner.Run(tool_path, tool_input_path,
                                       tool_output_path, tool_description);

   // For now, disable this to keep all files for debugging. TODO(digit): Make
   // this a config option.
   if (false) {
     disk_interface.RemoveFile(tool_input_path);
     disk_interface.RemoveFile(tool_output_path);
   }

   if (status == ExitInterrupted) {
     *error = "user interrupt";
     return false;
   }
   if (status != ExitSuccess) {
     *error = StringFormat("delayed worker failed with status %d", status);
     return false;
   }

   std::string json_response;
   auto read_status =
       disk_interface.ReadFile(tool_output_path, &json_response, nullptr);
   if (read_status == FileReader::NotFound) {
     *error = "delayed worker didn't create output file";
     return false;
   }
   if (read_status != FileReader::Okay) {
     *error = "could not read delayed worker's output file";
     return false;
   }

   return DelayedWorker::ResponseInfo::FromJson(json_response, response, error);
 }
	// Copyright 2025 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 "delayed_worker.h"

	#include <assert.h>
	#include <inttypes.h>
	#include <string.h>

	#include "disk_interface.h"
	#include "json_parser.h"
	#include "json_writer.h"
	#include "subprocess.h"
	#include "util.h"

	#define DEBUG 0

	#if DEBUG
	#include "scoped_defer.h"
	#define LOG(...) fprintf(stderr, __VA_ARGS__)
	#define LOG_ON_EXIT(...) ScopedDefer([&]() { fprintf(stderr, __VA_ARGS__); })
	#else
	#define LOG(...) (void)0
	#define LOG_ON_EXIT(...) (void)0
	#endif

	namespace {

	// Decode a single RequestInfo::Action. \|parser\| must wrap its JSON content.
	// On success, set \|*action\| and return true, on failure call
	// \|parser.SetError()\| and return false.
	bool DecodeRequestInfoAction(JsonParser& parser,
	DelayedWorker::RequestInfo::Action* action) {
	JsonParser::KeyValueMap kv;
	if (!parser.ParseObject(&kv))
	return false;

	// Get action id first.
	int64_t id;
	if (!kv.GetRequiredInteger("action_id", &id, parser))
	return false;

	if (id <= 0 \|\| id > INT32_MAX)
	return parser.SetErrorFormat(
	"action_id must be in 1..INT32_MAX range: %" PRId64, id);
	action->action_id = static_cast<uint32_t>(id);

	// Then the other fields.
	if (!kv.GetRequiredString("command", &action->command, parser) \|\|
	!kv.GetRequiredString("description", &action->description, parser) \|\|
	!kv.GetRequiredStringList("ninja_outputs", &action->ninja_outputs,
	parser) \|\|
	!kv.GetOptionalOrEmptyString("ninja_depfile", &action->ninja_depfile,
	parser))
	return false;

	return true;
	}

	bool DecodeRequest(JsonParser& parser, DelayedWorker::RequestInfo* result) {
	JsonParser::KeyValueMap kv;
	if (!parser.ParseObject(&kv))
	return false;

	// First check version.
	int64_t version = -1;
	if (!kv.GetRequiredInteger("version", &version, parser))
	return false;

	if (version != 2) {
	return parser.SetErrorFormat("Unknown version, expected 2, got %" PRId64,
	version);
	}

	int64_t request_id = -1;
	if (!kv.GetRequiredInteger("request_id", &request_id, parser))
	return false;

	if (request_id <= 0 \|\| request_id > INT32_MAX)
	return parser.SetErrorFormat(
	"request_id must be in 1..INT32_MAX range: %" PRId64, request_id);
	result->request_id = static_cast<uint32_t>(request_id);

	// Then get each actions from the 'actions' list.
	JsonParser::ValueInfo vi;
	if (!kv.GetRequiredExpected("actions", JsonParser::kArrayStart, &vi, parser))
	return false;

	JsonParser actions_parser(vi.value);
	JsonParser::ArrayIterator actions_it(actions_parser);
	auto& result_actions = result->actions;
	result_actions.clear();
	while (actions_it.HasItem()) {
	/// Create new instance directly at the end of the vector, then
	/// tell called function to write to it directly.
	result_actions.emplace_back();
	if (!DecodeRequestInfoAction(actions_parser, &result_actions.back())) {
	result_actions.pop_back();
	break;
	}
	if (actions_parser.HasError())
	return parser.SetError(actions_parser.TakeError());
	}

	// Finally, handle the optional metadata.
	BuildMetadataMap& result_metadata = result->build_metadata;
	result_metadata.clear();
	if (kv.Get("build_metadata", &vi)) {
	JsonParser metadata_parser(vi.value);
	JsonParser::ObjectIterator metadata_it(metadata_parser);
	while (metadata_it.HasKeyValue()) {
	auto ret = result_metadata.try_emplace(metadata_it.key().AsString(), "");
	if (!metadata_parser.ExpectString(&ret.first->second)) {
	result_metadata.erase(ret.first);
	break;
	}
	}
	if (metadata_parser.HasError())
	return parser.SetError(metadata_parser.TakeError());
	}
	return true;
	}

	bool DecodeResponse(JsonParser& parser, DelayedWorker::ResponseInfo* result) {
	JsonParser::KeyValueMap top_kv;
	if (!parser.ParseObject(&top_kv))
	return false;

	// Check version first.
	int64_t version = -1;
	if (!top_kv.GetRequiredInteger("version", &version, parser))
	return false;
	if (version != 1) {
	return parser.SetErrorFormat("Invalid version %" PRId64 ", expected 1",
	version);
	}

	// Get all other fields, checking them individually.
	int64_t request_id = -1;
	if (!top_kv.GetRequiredInteger("request_id", &request_id, parser))
	return false;

	if (request_id <= 0 \|\| request_id > INT32_MAX)
	return parser.SetErrorFormat(
	"request_id must be in 1..INT32_MAX range: %" PRId64, request_id);
	result->request_id = static_cast<uint32_t>(request_id);

	int64_t status = -1;
	if (!top_kv.GetRequiredInteger("status", &status, parser))
	return false;

	if (status < 0 \|\| status >= 256)
	return parser.SetErrorFormat("status must be in 0..255 range: %" PRId64,
	status);
	result->status = static_cast<uint8_t>(status);

	if (!top_kv.GetRequiredString("output", &result->output, parser))
	return false;
	return true;
	}

	} // namespace

	// static
	std::string DelayedWorker::RequestInfo::ToJson() const {
	std::string output;
	JsonWriter writer;
	{
	auto dict = writer.StartDict(output);
	dict->AddInt("version", 2);
	dict->AddInt("request_id", this->request_id);
	{
	auto actions = dict->AddList("actions");
	int action_id = 1;
	for (const auto& action : this->actions) {
	auto entry = actions->AddDict();
	entry->AddInt("action_id", action_id++);
	entry->AddString("command", action.command);
	entry->AddString("description", action.description);
	{
	auto outputs = entry->AddList("ninja_outputs");
	for (const auto& output : action.ninja_outputs)
	outputs->AddString(output);
	}
	if (!action.ninja_depfile.empty())
	entry->AddString("ninja_depfile", action.ninja_depfile);
	}
	}
	if (!this->build_metadata.empty()) {
	auto metadata = dict->AddDict("build_metadata");
	for (const auto& pair : this->build_metadata)
	metadata->AddString(pair.first, pair.second);
	}
	}
	return output;
	}

	// static
	bool DelayedWorker::RequestInfo::FromJson(StringPiece request_json,
	DelayedWorker::RequestInfo* result,
	std::string* error) {
	JsonParser parser(request_json);
	if (!DecodeRequest(parser, result)) {
	*error = parser.TakeError();
	return false;
	}
	return true;
	}

	// static
	std::string DelayedWorker::ResponseInfo::ToJson() const {
	std::string output;
	JsonWriter writer;
	{
	auto dict = writer.StartDict(output);
	dict->AddInt("version", 1);
	dict->AddInt("request_id", this->request_id);
	dict->AddInt("status", this->status);
	dict->AddString("output", this->output);
	}
	return output;
	}

	// static
	bool DelayedWorker::ResponseInfo::FromJson(StringPiece response_json,
	DelayedWorker::ResponseInfo* result,
	std::string* error) {
	JsonParser parser(response_json);
	if (!DecodeResponse(parser, result)) {
	*error = parser.TakeError();
	return false;
	}
	return true;
	}

	/// A DelayedWorkerRunner that runs commands in a SubprocessSet.
	struct SubprocessSetRunner : public DelayedWorkerRunner {
	SubprocessSetRunner(SubprocessSet& subprocess_set)
	: subprocess_set_(subprocess_set) {}

	ExitStatus Run(const std::string& tool, const std::string& tool_input_path,
	const std::string& tool_output_path,
	const std::string& tool_description) override {
	// Convenience helper to escape a string for the shell.
	auto escaped = [](const std::string& str) -> std::string {
	std::string result;
	GetShellEscapedString(str, &result);
	return result;
	};

	// Ensure the script is called from the current directory.
	// Only pass two arguments: paths of the input request file, and output
	// response file.
	std::string tool_command = "./" +
	tool + " " + escaped(tool_input_path) + " " + escaped(tool_output_path);

	// Launch single command in console mode.
	bool use_console = true;
	subprocess_set_.Add(tool_command, &tool_description, use_console);

	std::unique_ptr<Subprocess> subproc;
	while (!(subproc = subprocess_set_.NextFinished())) {
	bool interrupted =
	subprocess_set_.DoWork() == SubprocessSet::WorkResult::Interrupted;

	if (interrupted) {
	subprocess_set_.Clear();
	return ExitInterrupted;
	}
	}

	return subproc->Finish();
	}

	SubprocessSet& subprocess_set_;
	};

	// static
	std::unique_ptr<DelayedWorkerRunner> DelayedWorkerRunner::CreateFrom(
	SubprocessSet& subprocess_set) {
	return std::make_unique<SubprocessSetRunner>(subprocess_set);
	}

	// static
	bool DelayedWorker::RunTool(const std::string& tool_path,
	const std::string& tool_description,
	const RequestInfo& request,
	DelayedWorkerRunner& tool_runner,
	DiskInterface& disk_interface,
	ResponseInfo* response, std::string* error) {
	static int s_invocation_id = 0;
	int invocation_id = ++s_invocation_id;
	LOG("DelayedWorker::RunTool: Starting new invocation_id=%d\n", invocation_id);
	LOG_ON_EXIT("DelayedWorker::RunTool: Exit for invocation_id=%d\n",
	invocation_id);

	std::string tool_input_path = StringFormat(
	"ninja_delayed_bazel_build_tool_in.%d.json", s_invocation_id);

	std::string tool_output_path = StringFormat(
	"ninja_delayed_bazel_build_tool_out.%d.json", s_invocation_id);

	if (!disk_interface.WriteFile(tool_input_path, request.ToJson(), false)) {
	*error = "failed to write delayed worker input file";
	return false;
	}

	ExitStatus status = tool_runner.Run(tool_path, tool_input_path,
	tool_output_path, tool_description);

	// For now, disable this to keep all files for debugging. TODO(digit): Make
	// this a config option.
	if (false) {
	disk_interface.RemoveFile(tool_input_path);
	disk_interface.RemoveFile(tool_output_path);
	}

	if (status == ExitInterrupted) {
	*error = "user interrupt";
	return false;
	}
	if (status != ExitSuccess) {
	*error = StringFormat("delayed worker failed with status %d", status);
	return false;
	}

	std::string json_response;
	auto read_status =
	disk_interface.ReadFile(tool_output_path, &json_response, nullptr);
	if (read_status == FileReader::NotFound) {
	*error = "delayed worker didn't create output file";
	return false;
	}
	if (read_status != FileReader::Okay) {
	*error = "could not read delayed worker's output file";
	return false;
	}

	return DelayedWorker::ResponseInfo::FromJson(json_response, response, error);
	}