From 98f7ccb6398c24c87010dea6a7425b8c41fb4267 Mon Sep 17 00:00:00 2001 From: Xinhao Yuan Date: Thu, 9 Jul 2026 07:04:18 -0700 Subject: [PATCH] Remove batch timeout handling in the runner. Centipede has been able to handle the batch timeout instead. PiperOrigin-RevId: 945095691 --- centipede/BUILD | 2 + centipede/centipede_callbacks.cc | 21 ++++-- centipede/centipede_test.cc | 7 +- centipede/command.cc | 4 +- centipede/engine_worker.cc | 121 +++++++++++++++++++++++++++---- centipede/puzzles/run_puzzle.sh | 1 - centipede/runner.cc | 24 +----- centipede/runner.h | 6 -- centipede/util.cc | 14 ++++ centipede/util.h | 4 + centipede/util_test.cc | 13 ++++ 11 files changed, 164 insertions(+), 53 deletions(-) diff --git a/centipede/BUILD b/centipede/BUILD index 683b2afde..518d7a6d5 100644 --- a/centipede/BUILD +++ b/centipede/BUILD @@ -217,6 +217,7 @@ cc_library( "@abseil-cpp//absl/strings", "@abseil-cpp//absl/strings:str_format", "@abseil-cpp//absl/synchronization", + "@abseil-cpp//absl/time", "@abseil-cpp//absl/types:span", "@com_google_fuzztest//common:defs", "@com_google_fuzztest//common:hash", @@ -1394,6 +1395,7 @@ cc_test( ":thread_pool", ":util", "@abseil-cpp//absl/container:flat_hash_map", + "@abseil-cpp//absl/time", "@com_google_fuzztest//common:defs", "@com_google_fuzztest//common:hash", "@com_google_fuzztest//common:logging", diff --git a/centipede/centipede_callbacks.cc b/centipede/centipede_callbacks.cc index b03b48aad..c615992ef 100644 --- a/centipede/centipede_callbacks.cc +++ b/centipede/centipede_callbacks.cc @@ -62,8 +62,6 @@ namespace fuzztest::internal { -constexpr auto kPollMinimalTimeout = absl::Milliseconds(1); - class CentipedeCallbacks::PersistentModeServer { public: explicit PersistentModeServer(std::string server_path) @@ -194,9 +192,7 @@ class CentipedeCallbacks::PersistentModeServer { int poll_ret = -1; do { poll_fd = {fd, static_cast(event)}; - const int poll_timeout_ms = static_cast(absl::ToInt64Milliseconds( - std::max(deadline - absl::Now(), kPollMinimalTimeout))); - poll_ret = poll(&poll_fd, 1, poll_timeout_ms); + poll_ret = poll(&poll_fd, 1, PollTimeoutMs(deadline - absl::Now())); } while (poll_ret < 0 && errno == EINTR); if (poll_ret == 1 && (poll_fd.revents & (event | POLLHUP)) == event) { return true; @@ -343,7 +339,6 @@ std::string CentipedeCallbacks::ConstructRunnerFlags( std::vector flags = { "CENTIPEDE_RUNNER_FLAGS=", absl::StrCat("timeout_per_input=", env_.timeout_per_input), - absl::StrCat("timeout_per_batch=", env_.timeout_per_batch), absl::StrCat("address_space_limit_mb=", env_.address_space_limit_mb), absl::StrCat("rss_limit_mb=", env_.rss_limit_mb), absl::StrCat("stack_limit_kb=", env_.stack_limit_kb), @@ -565,8 +560,12 @@ int CentipedeCallbacks::ExecuteCentipedeSancovBinaryWithShmem( } // Run. + const auto batch_start_time = absl::Now(); const int exit_code = RunBatchForBinary(binary); inputs_blobseq_.ReleaseSharedMemory(); // Inputs are already consumed. + const bool batch_timed_out = + env_.timeout_per_batch > 0 && + absl::Now() - batch_start_time > absl::Seconds(env_.timeout_per_batch); // Get results. batch_result.exit_code() = exit_code; @@ -591,6 +590,16 @@ int CentipedeCallbacks::ExecuteCentipedeSancovBinaryWithShmem( if (env_.print_runner_log) PrintExecutionLog(); + if (batch_timed_out) { + FUZZTEST_LOG(INFO) + << "Batch timeout detected. Replacing the original exit code: " + << exit_code + << ", failure description: " << batch_result.failure_description(); + batch_result.failure_description() = kExecutionFailurePerBatchTimeout; + batch_result.exit_code() = EXIT_FAILURE; + return EXIT_FAILURE; + } + // TODO: b/467103298 - Handle failures when the exit code is zero, e.g., when // the target exits via `std::_Exit(0)`. if (exit_code != EXIT_SUCCESS) { diff --git a/centipede/centipede_test.cc b/centipede/centipede_test.cc index 3e7a471b6..e242d2d95 100644 --- a/centipede/centipede_test.cc +++ b/centipede/centipede_test.cc @@ -1406,9 +1406,8 @@ TEST_F(CentipedeWithTemporaryLocalDir, EngineWorksInWorkerMode) { fuzztest::internal::DefaultCallbacksFactory< fuzztest::internal::CentipedeDefaultCallbacks> callbacks; - EXPECT_DEATH( - [&] { std::exit(CentipedeMain(env, callbacks)); }(), - ContainsRegex("Failure *: INPUT FAILURE: some_failure_description")); + EXPECT_DEATH([&] { std::exit(CentipedeMain(env, callbacks)); }(), + ContainsRegex("Failure *: some_failure_description")); } TEST_F(CentipedeWithTemporaryLocalDir, EngineWorksInStandaloneMode) { @@ -1428,7 +1427,7 @@ TEST_F(CentipedeWithTemporaryLocalDir, EngineWorksInStandaloneMode) { const int status = std::system(test_command.c_str()); std::exit(WEXITSTATUS(status)); }(), - ContainsRegex("Failure *: INPUT FAILURE: some_failure_description")); + ContainsRegex("Failure *: some_failure_description")); } } // namespace diff --git a/centipede/command.cc b/centipede/command.cc index 4afab0abb..06cc79f0f 100644 --- a/centipede/command.cc +++ b/centipede/command.cc @@ -146,9 +146,7 @@ struct Command::ForkServerProps { /*fd=*/pipe_[1], /*events=*/POLLIN, }; - const int poll_timeout_ms = static_cast(absl::ToInt64Milliseconds( - std::max(deadline - absl::Now(), absl::Milliseconds(1)))); - poll_ret = poll(&poll_fd, 1, poll_timeout_ms); + poll_ret = poll(&poll_fd, 1, PollTimeoutMs(deadline - absl::Now())); // The `poll()` syscall can get interrupted: it sets errno==EINTR in that // case. We should tolerate that. } while (poll_ret < 0 && errno == EINTR); diff --git a/centipede/engine_worker.cc b/centipede/engine_worker.cc index 3270afcc4..d55816e70 100644 --- a/centipede/engine_worker.cc +++ b/centipede/engine_worker.cc @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. #include +#include #include #include #include @@ -26,6 +27,7 @@ #include #include #include +#include #include #include #include @@ -217,6 +219,7 @@ enum class WorkerAction { kTestGetSeeds, kTestMutate, kTestExecute, + kNoOp, }; constexpr std::string_view kWorkerBinaryIdOutputFlagHeader = @@ -237,6 +240,7 @@ constexpr std::string_view kWorkerOutputsBlobSequencePathFlagHeader = constexpr std::string_view kWorkerPersistentModeSocketPathFlagHeader = "persistent_mode_socket="; // TODO: Use better flag names when // standardizing the protocol. +constexpr std::string_view kWorkerCrossOverLevel = "crossover_level="; struct WorkerState { std::atomic has_failure_output = false; @@ -295,14 +299,28 @@ void WorkerEmitError(std::string_view message) { WorkerEmitFailureOutput("SETUP FAILURE: ", message); } +// TODO(xinhaoyuan): Some of them are planned to be removed. +static constexpr std::array kNonFindingPrefixes = { + "SETUP FAILURE:", + "IGNORED FAILURE:", + "SKIPPED TEST:", +}; + void WorkerEmitFinding(std::string_view description, std::string_view signature) { WorkerCheck( GetWorkerState().in_adapter_execute.load(std::memory_order_relaxed), "Must emit finding in adapter execute"); + for (auto bad_prefix : kNonFindingPrefixes) { + if (description.substr(0, bad_prefix.size()) == bad_prefix) { + WorkerLog("Got bad prefix in the finding description: ", description); + WorkerEmitError("Bad prefix in the finding description"); + return; + } + } const bool ignored = GetWorkerState().has_finding.exchange(true); if (!ignored) { - WorkerCheck(WorkerEmitFailureOutput("INPUT FAILURE: ", description), + WorkerCheck(WorkerEmitFailureOutput(/*prefix=*/"", description), "Failed to emit failure output for the finding"); if (const char* finding_signature_path = GetWorkerFlag(kWorkerFailureSignaturePathFlagHeader); @@ -415,8 +433,25 @@ BlobSequence* GetOutputsBlobSequence() { return result; } -WorkerAction GetWorkerAction() { - static WorkerAction worker_action = [] { +int GetCrossOverLevel() { + static int result = []() { + const char* cross_over_level_str = GetWorkerFlag(kWorkerCrossOverLevel); + if (cross_over_level_str != nullptr) { + const int parsed = + atoi(cross_over_level_str); // NOLINT: can't use strto64, etc. + if (0 <= parsed && parsed <= 100) return parsed; + } + // Default + return 50; + }(); + return result; +} + +std::optional GetWorkerAction() { + static auto worker_action = []() -> std::optional { + if (HasWorkerSwitchFlag("dump_configuration")) { + return WorkerAction::kNoOp; + } if (HasWorkerSwitchFlag("dump_binary_id")) { return WorkerAction::kGetBinaryId; } @@ -427,7 +462,9 @@ WorkerAction GetWorkerAction() { return WorkerAction::kTestGetSeeds; } auto* inputs_blobseq = GetInputsBlobSequence(); - WorkerCheck(inputs_blobseq != nullptr, "input blob sequence is not found"); + if (inputs_blobseq == nullptr) { + return std::nullopt; + } auto request_type_blob = inputs_blobseq->Read(); if (IsMutationRequest(request_type_blob)) { inputs_blobseq->Reset(); @@ -437,9 +474,7 @@ WorkerAction GetWorkerAction() { inputs_blobseq->Reset(); return WorkerAction::kTestExecute; } - WorkerCheck(false, "unknown worker action from the flags"); - // should not reach here. - std::abort(); + return std::nullopt; }(); return worker_action; } @@ -587,11 +622,20 @@ void WorkerDoMutate(const FuzzTestAdapter& adapter) { // Assume RAND_MAX is large enough and not worry about fairness for now. const auto origin = rand_r(&seed) % origin_inputs.size(); emitted_inputs.clear(); - adapter.Mutate(adapter.ctx, origin_inputs[origin], /*shrink=*/0, - &input_sink); + const bool do_cross_over = adapter.CrossOver != nullptr && + rand_r(&seed) % 100 < GetCrossOverLevel(); + if (do_cross_over) { + const auto other = rand_r(&seed) % origin_inputs.size(); + adapter.CrossOver(adapter.ctx, origin_inputs[origin], + origin_inputs[other], &input_sink); + } else { + adapter.Mutate(adapter.ctx, origin_inputs[origin], /*shrink=*/0, + &input_sink); + } WORKER_CHECK_FOR_ERROR(); WorkerCheck(emitted_inputs.size() == 1, - "Mutate must emit exactly one input"); + do_cross_over ? "CrossOver must emit exactly one input" + : "Mutate must emit exactly one input"); mutant_bytes.clear(); adapter.SerializeInputContent(adapter.ctx, emitted_inputs[0], &mutant_bytes_sink); @@ -663,6 +707,29 @@ void WorkerDoExecute(const FuzzTestAdapter& adapter) { return true; }(); + // Utils to get execution stats. + + size_t last_time_usec = 0; + auto UsecSinceLast = [&last_time_usec]() { + const uint64_t t = TimeInUsec(); + const uint64_t ret_val = t - last_time_usec; + last_time_usec = t; + return ret_val; + }; + + auto GetPeakRSSMb = []() -> size_t { + struct rusage usage = {}; + if (getrusage(RUSAGE_SELF, &usage) != 0) return 0; +#ifdef __APPLE__ + // On MacOS, the unit seems to be byte according to experiment, while some + // documents mentioned KiB. This could depend on OS variants. + return usage.ru_maxrss >> 20; +#else // __APPLE__ + // On Linux, ru_maxrss is in KiB + return usage.ru_maxrss >> 10; +#endif // __APPLE__ + }; + // In-loop variables declared outside to save allocations. std::vector features; std::vector serialized_metadata; @@ -670,6 +737,16 @@ void WorkerDoExecute(const FuzzTestAdapter& adapter) { const auto input_sink = GetInputSinkTo(emitted_inputs); for (size_t i = 0; i < num_inputs; i++) { + // The stats gathered here are slightly different than + // the original definition: Here the exec_time_usec will cover the time used + // to process coverage feedback, which is usually fine for the purpose of + // corpus scheduling, but coverage processing can be non-deterministic due + // to optimization. E.g. Execute() may perform deep and slow cleanup on + // the beginning of consecutive Execute() calls to reduce noise. + // + // TODO(xinhaoyuan): Consider improving it. E.g. let user override stats. + ExecutionResult::Stats stats = {}; + UsecSinceLast(); auto blob = inputs_blobseq->Read(); if (!blob.IsValid()) return; // no more blobs to read. WorkerCheck(IsDataInput(blob), "Must read data input"); @@ -698,7 +775,9 @@ void WorkerDoExecute(const FuzzTestAdapter& adapter) { }}; GetWorkerState().in_adapter_execute = true; + stats.prep_time_usec = UsecSinceLast(); adapter.Execute(adapter.ctx, input, &feedback_sink); + stats.exec_time_usec = UsecSinceLast(); GetWorkerState().in_adapter_execute = false; WORKER_CHECK_FOR_ERROR(); @@ -748,6 +827,12 @@ void WorkerDoExecute(const FuzzTestAdapter& adapter) { WorkerLog("failed to write input metadata"); break; } + stats.post_time_usec = UsecSinceLast(); + stats.peak_rss_mb = GetPeakRSSMb(); + if (!BatchResult::WriteStats(stats, *outputs_blobseq)) { + WorkerLog("failed to write stats"); + break; + } if (!BatchResult::WriteInputEnd(*outputs_blobseq)) { WorkerLog("failed to write input end"); break; @@ -837,6 +922,12 @@ FuzzTestWorkerStatus WorkerRun(const FuzzTestAdapterManager& manager) { } const auto action = GetWorkerAction(); + WorkerCheck(action.has_value(), "No worker action to run"); + + if (action == WorkerAction::kNoOp) { + return kFuzzTestWorkerSuccess; + } + if (action == WorkerAction::kGetBinaryId) { WorkerDoGetBinaryId(manager); return kFuzzTestWorkerSuccess; @@ -881,8 +972,8 @@ FuzzTestWorkerStatus WorkerRun(const FuzzTestAdapterManager& manager) { WorkerCheck(manager.ConstructAdapter != nullptr, "ConstructAdapter is not defined"); FuzzTestAdapter adapter = {}; - manager.ConstructAdapter(manager.ctx, /*diagnostic_sink=*/&diagnostic_sink, - &adapter); + manager.ConstructAdapter(manager.ctx, + /*diagnostic_sink=*/&diagnostic_sink, &adapter); WORKER_CHECK_FOR_ERROR(); WorkerCheck(adapter.SetUpCoverageDomains != nullptr, "SetUpCoverageDomains must be defined"); @@ -929,7 +1020,11 @@ using ::fuzztest::internal::WorkerRun; } // namespace -int FuzzTestWorkerIsRequired() { return GetWorkerFlags().present; } +int FuzzTestWorkerIsRequired() { + static int result = GetWorkerFlags().present && + fuzztest::internal::GetWorkerAction().has_value(); + return result; +} FuzzTestWorkerStatus FuzzTestWorkerMaybeRun( const FuzzTestAdapterManager* manager) { diff --git a/centipede/puzzles/run_puzzle.sh b/centipede/puzzles/run_puzzle.sh index 0aee6243c..d6d12c853 100755 --- a/centipede/puzzles/run_puzzle.sh +++ b/centipede/puzzles/run_puzzle.sh @@ -94,7 +94,6 @@ function ExpectPerInputTimeout() { # Expects that Centipede found a per-batch timeout. function ExpectPerBatchTimeout() { echo "======= ${FUNCNAME[0]}" - fuzztest::internal::assert_regex_in_file "Per-batch timeout exceeded" "${log}" fuzztest::internal::assert_regex_in_file "Failure.*: per-batch-timeout-exceeded" "${log}" fuzztest::internal::assert_regex_in_file \ "Failure applies to entire batch: not executing inputs one-by-one" "${log}" diff --git a/centipede/runner.cc b/centipede/runner.cc index e085b54e9..82b2b4920 100644 --- a/centipede/runner.cc +++ b/centipede/runner.cc @@ -132,8 +132,7 @@ static void CheckWatchdogLimits() { const char *failure; }; const uint64_t input_start_time = state->input_start_time; - const uint64_t batch_start_time = state->batch_start_time; - if (input_start_time == 0 || batch_start_time == 0) return; + if (input_start_time == 0) return; const Resource resources[] = { {Resource{ /*what=*/"Per-input timeout", @@ -144,15 +143,6 @@ static void CheckWatchdogLimits() { state->run_time_flags.ignore_timeout_reports != 0, /*failure=*/kExecutionFailurePerInputTimeout.data(), }}, - {Resource{ - /*what=*/"Per-batch timeout", - /*units=*/"sec", - /*value=*/curr_time - batch_start_time, - /*limit=*/state->run_time_flags.timeout_per_batch, - /*ignore_report=*/ - state->run_time_flags.ignore_timeout_reports != 0, - /*failure=*/kExecutionFailurePerBatchTimeout.data(), - }}, {Resource{ /*what=*/"RSS limit", /*units=*/"MB", @@ -240,10 +230,10 @@ __attribute__((noinline)) void CheckStackLimit(size_t stack_usage, void GlobalRunnerState::StartWatchdogThread() { fprintf(stderr, "Starting watchdog thread: timeout_per_input: %" PRIu64 - " sec; timeout_per_batch: %" PRIu64 " sec; rss_limit_mb: %" PRIu64 - " MB; stack_limit_kb: %" PRIu64 " KB\n", + " sec; rss_limit_mb: %" PRIu64 " MB; stack_limit_kb: %" PRIu64 + " KB\n", run_time_flags.timeout_per_input.load(), - run_time_flags.timeout_per_batch, run_time_flags.rss_limit_mb.load(), + run_time_flags.rss_limit_mb.load(), state->run_time_flags.stack_limit_kb.load()); pthread_t watchdog_thread; pthread_create(&watchdog_thread, nullptr, WatchdogThread, nullptr); @@ -257,11 +247,6 @@ void GlobalRunnerState::StartWatchdogThread() { void GlobalRunnerState::ResetTimers() { const auto curr_time = time(nullptr); state->input_start_time = curr_time; - // batch_start_time is set only once -- just before the first input of the - // batch is about to start running. - if (batch_start_time == 0) { - batch_start_time = curr_time; - } } // Byte array mutation fallback for a custom mutator, as defined here: @@ -994,7 +979,6 @@ extern "C" void CentipedeEndExecutionBatch() { } in_execution_batch = false; fuzztest::internal::state->input_start_time = 0; - fuzztest::internal::state->batch_start_time = 0; } extern "C" void CentipedePrepareProcessing() { diff --git a/centipede/runner.h b/centipede/runner.h index fbe094166..5ddc3b3e5 100644 --- a/centipede/runner.h +++ b/centipede/runner.h @@ -33,7 +33,6 @@ namespace fuzztest::internal { // Flags derived from CENTIPEDE_RUNNER_FLAGS. struct RunTimeFlags { std::atomic timeout_per_input; - uint64_t timeout_per_batch; std::atomic rss_limit_mb; uint64_t crossover_level; uint64_t ignore_timeout_reports : 1; @@ -67,7 +66,6 @@ struct GlobalRunnerState { // flags can change later (if wrapped with std::atomic). RunTimeFlags run_time_flags = { /*timeout_per_input=*/flag_helper.HasIntFlag(":timeout_per_input=", 0), - /*timeout_per_batch=*/flag_helper.HasIntFlag(":timeout_per_batch=", 0), /*rss_limit_mb=*/flag_helper.HasIntFlag(":rss_limit_mb=", 0), /*crossover_level=*/flag_helper.HasIntFlag(":crossover_level=", 50), /*ignore_timeout_reports=*/ @@ -112,10 +110,6 @@ struct GlobalRunnerState { // time. std::atomic input_start_time; - // Per-batch timer. Initially, zero. ResetInputTimer() sets it to the current - // time before the first input and never resets it. - std::atomic batch_start_time; - // The Watchdog thread sets this to true. std::atomic watchdog_thread_started; }; diff --git a/centipede/util.cc b/centipede/util.cc index 83da9e7f6..ab2eb19ef 100644 --- a/centipede/util.cc +++ b/centipede/util.cc @@ -32,6 +32,7 @@ #include #include #include +#include #include #include #include @@ -50,6 +51,7 @@ #include "absl/strings/str_replace.h" #include "absl/strings/str_split.h" #include "absl/synchronization/mutex.h" +#include "absl/time/time.h" #include "absl/types/span.h" #include "./centipede/feature.h" #include "./common/defs.h" @@ -369,4 +371,16 @@ void Munmap(uint8_t *ptr, size_t size) { FUZZTEST_CHECK_EQ(result, 0); } +int PollTimeoutMs(absl::Duration timeout) { + if (timeout == absl::InfiniteDuration()) { + return -1; + } + const auto ms = absl::ToInt64Milliseconds(absl::Ceil( + std::max(timeout, absl::Milliseconds(1)), absl::Milliseconds(1))); + if (ms > std::numeric_limits::max()) { + return std::numeric_limits::max(); + } + return static_cast(ms); +} + } // namespace fuzztest::internal diff --git a/centipede/util.h b/centipede/util.h index ac5e96392..4905c68b4 100644 --- a/centipede/util.h +++ b/centipede/util.h @@ -23,6 +23,7 @@ #include #include "absl/base/nullability.h" +#include "absl/time/time.h" #include "absl/types/span.h" #include "./centipede/feature.h" #include "./common/defs.h" @@ -192,6 +193,9 @@ class MmapNoReserveArray { uint8_t *array_; }; +// Converts `timeout` to an integer value of milliseconds suitable for `poll()`. +int PollTimeoutMs(absl::Duration timeout); + } // namespace fuzztest::internal #endif // THIRD_PARTY_CENTIPEDE_UTIL_H_ diff --git a/centipede/util_test.cc b/centipede/util_test.cc index 7be77e88c..1629f9212 100644 --- a/centipede/util_test.cc +++ b/centipede/util_test.cc @@ -17,6 +17,7 @@ #include #include #include // NOLINT +#include #include #include #include @@ -24,6 +25,7 @@ #include "gmock/gmock.h" #include "gtest/gtest.h" #include "absl/container/flat_hash_map.h" +#include "absl/time/time.h" #include "./centipede/feature.h" #include "./centipede/thread_pool.h" #include "./common/defs.h" @@ -291,4 +293,15 @@ TEST(UtilTest, RemoveSubset) { testing::ElementsAre(std::vector{1}, std::vector{3})); } +TEST(UtilTest, PollTimeoutMsWorks) { + EXPECT_GT(PollTimeoutMs(absl::ZeroDuration()), 0); + EXPECT_GT(PollTimeoutMs(-absl::InfiniteDuration()), 0); + EXPECT_EQ(PollTimeoutMs(absl::InfiniteDuration()), -1); + EXPECT_EQ(PollTimeoutMs(absl::Milliseconds(123)), 123); + const auto long_finite_duration = + absl::Seconds(std::numeric_limits::max()); + FUZZTEST_CHECK_LT(long_finite_duration, absl::InfiniteDuration()); + EXPECT_GT(PollTimeoutMs(long_finite_duration), 0); +} + } // namespace fuzztest::internal