#Centipede Use PeriodicAction for stats reporting in main fuzzing loop

PiperOrigin-RevId: 630513916

centipede/BUILD

@@ -342,7 +342,6 @@ cc_library(
     deps = [
         "@com_google_absl//absl/base:core_headers",
         "@com_google_absl//absl/functional:any_invocable",
-        "@com_google_absl//absl/log:check",
         "@com_google_absl//absl/synchronization",
         "@com_google_absl//absl/time",
     ],
@@ -794,6 +793,7 @@ cc_library(
         ":environment",
         ":minimize_crash",
         ":pc_info",
+        ":periodic_action",
         ":runner_result",
         ":seed_corpus_config_cc_proto",
         ":seed_corpus_maker_lib",

centipede/centipede_interface.cc

@@ -55,6 +55,7 @@
 #include "./centipede/environment.h"
 #include "./centipede/minimize_crash.h"
 #include "./centipede/pc_info.h"
+#include "./centipede/periodic_action.h"
 #include "./centipede/runner_result.h"
 #include "./centipede/seed_corpus_config.pb.h"
 #include "./centipede/seed_corpus_maker_lib.h"
@@ -139,36 +140,6 @@ int ForEachBlob(const Environment &env) {
   return EXIT_SUCCESS;
 }
 
-// Runs in a dedicated thread, periodically calls PrintExperimentStats
-// on `stats_vec` and `envs`.
-// Stops when `continue_running` becomes false.
-// Exits immediately if --experiment flag is not used.
-void ReportStatsThread(const std::atomic<bool> &continue_running,
-                       const std::vector<std::atomic<Stats>> &stats_vec,
-                       const std::vector<Environment> &envs) {
-  CHECK(!envs.empty());
-
-  std::vector<std::unique_ptr<StatsReporter>> reporters;
-  reporters.emplace_back(
-      std::make_unique<StatsCsvFileAppender>(stats_vec, envs));
-  if (!envs.front().experiment.empty() || ABSL_VLOG_IS_ON(1)) {
-    reporters.emplace_back(std::make_unique<StatsLogger>(stats_vec, envs));
-  }
-
-  // TODO(ussuri): Use constant time increments for CSV generation?
-  for (int i = 0; continue_running; ++i) {
-    // Sleep at least a few seconds, and at most 600.
-    int seconds_to_sleep = std::clamp(i, 5, 600);
-    // Sleep(1) in a loop so that we check continue_running once a second.
-    while (--seconds_to_sleep && continue_running) {
-      absl::SleepFor(absl::Seconds(1));
-    }
-    for (auto &reporter : reporters) {
-      reporter->ReportCurrStats();
-    }
-  }
-}
-
 // Loads corpora from work dirs provided in `env.args`, if there are two args
 // provided, analyzes differences. If there is one arg provided, reports the
 // function coverage. Returns EXIT_SUCCESS on success, EXIT_FAILURE otherwise.
@@ -230,11 +201,31 @@ int Fuzz(const Environment &env, const BinaryInfo &binary_info,
 
   std::vector<Environment> envs(env.num_threads, env);
   std::vector<std::atomic<Stats>> stats_vec(env.num_threads);
-  std::atomic<bool> stats_thread_continue_running = true;
 
-  std::thread stats_thread(ReportStatsThread,
-                           std::ref(stats_thread_continue_running),
-                           std::ref(stats_vec), std::ref(envs));
+  // Start periodic stats dumping and, optionally, logging.
+  std::vector<PeriodicAction> stats_reporters;
+  stats_reporters.emplace_back(
+      [csv_appender = StatsCsvFileAppender{stats_vec, envs}]() mutable {
+        csv_appender.ReportCurrStats();
+      },
+      PeriodicAction::Options{
+          .sleep_before_each =
+              [](size_t iteration) {
+                return absl::Minutes(std::clamp(iteration, 0UL, 10UL));
+              },
+      });
+  if (!envs.front().experiment.empty() || ABSL_VLOG_IS_ON(1)) {
+    stats_reporters.emplace_back(
+        [logger = StatsLogger{stats_vec, envs}]() mutable {
+          logger.ReportCurrStats();
+        },
+        PeriodicAction::Options{
+            .sleep_before_each =
+                [](size_t iteration) {
+                  return absl::Seconds(std::clamp(iteration, 5UL, 600UL));
+                },
+        });
+  }
 
   auto fuzzing_worker = [&](Environment &my_env, std::atomic<Stats> &stats,
                             bool create_tmpdir) {
@@ -274,8 +265,9 @@ int Fuzz(const Environment &env, const BinaryInfo &binary_info,
     }
   }
 
-  stats_thread_continue_running = false;
-  stats_thread.join();
+  for (auto &reporter : stats_reporters) {
+    reporter.Stop();
+  }
 
   if (!env.knobs_file.empty()) PrintRewardValues(stats_vec, std::cerr);
 

centipede/distill.cc

@@ -421,12 +421,11 @@ int Distill(const Environment &env, const DistillOptions &opts) {
                     << " unique: " << stats.num_byte_unique_elts
                     << " distilled: " << stats.num_feature_unique_elts;
         },
-        {
-            // Seeing 0's at the beginning is not interesting, unless debugging.
-            .delay = absl::Seconds(ABSL_VLOG_IS_ON(1) ? 0 : 60),
-            // Again, increase the frequency with --v >= 1 to aid debugging.
-            .interval = absl::Seconds(ABSL_VLOG_IS_ON(1) ? 10 : 60),
-        },
+        // Seeing 0's at the beginning is not interesting, unless debugging.
+        // Likewise, increase the frequency --v >= 1 to aid debugging.
+        PeriodicAction::ConstDelayConstInterval(
+            absl::Seconds(ABSL_VLOG_IS_ON(1) ? 0 : 60),
+            absl::Seconds(ABSL_VLOG_IS_ON(1) ? 10 : 60)),
     };
     // The RAM pool shared between all the `DistillToOneOutputShard()` threads.
     perf::ResourcePool ram_pool{kRamQuota};

centipede/periodic_action.cc

@@ -14,13 +14,13 @@
 
 #include "./centipede/periodic_action.h"
 
+#include <cstdint>
 #include <memory>
 #include <thread>
 #include <utility>
 
 #include "absl/base/thread_annotations.h"
 #include "absl/functional/any_invocable.h"
-#include "absl/log/check.h"
 #include "absl/synchronization/mutex.h"
 #include "absl/synchronization/notification.h"
 #include "absl/time/time.h"
@@ -32,12 +32,7 @@ class PeriodicAction::Impl {
   Impl(absl::AnyInvocable<void()> action, PeriodicAction::Options options)
       : action_{std::move(action)},
         options_{std::move(options)},
-        thread_{[this]() { RunLoop(); }} {
-    // NOTE: Allow `options_.delay` to be `absl::InfiniteDuration()`: that's a
-    // valid use case, where the run-loop actually starts looping only after a
-    // first explicit nudge.
-    CHECK_GT(options_.interval, absl::ZeroDuration());
-  }
+        thread_{[this]() { RunLoop(); }} {}
 
   void Stop() {
     StopAsync();
@@ -54,8 +49,8 @@ class PeriodicAction::Impl {
     if (!stop_.HasBeenNotified()) {
       // Prime the run-loop to exit next time it re-checks `stop_`.
       stop_.Notify();
-      // Nudge the run-loop out of the sleeping phase, if it's there: the loop
-      // immediately goes to re-check `stop_` and exits.
+      // Nudge the run-loop out of the sleeping phase, if it's currently idling
+      // there: the loop immediately goes to re-check `stop_` and exits.
       {
         absl::MutexLock lock{&nudge_mu_};
         nudge_ = true;
@@ -70,10 +65,13 @@ class PeriodicAction::Impl {
 
  private:
   void RunLoop() {
-    SleepUnlessWokenByNudge(options_.delay);
+    uint64_t iteration = 0;
     while (!stop_.HasBeenNotified()) {
-      action_();
-      SleepUnlessWokenByNudge(options_.interval);
+      SleepUnlessWokenByNudge(options_.sleep_before_each(iteration));
+      if (!stop_.HasBeenNotified()) {
+        action_();
+      }
+      ++iteration;
     }
   }
 

centipede/periodic_action.h

@@ -33,6 +33,7 @@
 #ifndef FUZZTEST_CENTIPEDE_PERIODIC_ACTION_H_
 #define FUZZTEST_CENTIPEDE_PERIODIC_ACTION_H_
 
+#include <cstdint>
 #include <memory>
 
 #include "absl/functional/any_invocable.h"
@@ -43,16 +44,42 @@ namespace centipede {
 class PeriodicAction {
  public:
   struct Options {
-    // A delay before the first invocation of the callback. Allowed to be
-    // `absl::IniniteDuration()`: in that case, the periodic invocations are
-    // initiated only by a first explicit `Nudge()` call.
-    absl::Duration delay = absl::ZeroDuration();
-    // The interval between the end of one invocation of the callback and the
-    // start of another. Note that a nudge triggers an out-of-schedule
-    // invocation and resets the timer (see `Nudge()`).
-    absl::Duration interval = absl::InfiniteDuration();
+    // The interval to sleep for before a given iteration. Iteration numbers are
+    // 0-based.
+    //
+    // Thus, the interval before `iteration == 0` is the delay before the first
+    // invocation of the action, the interval before `iteration == 1` is the
+    // interval between the first and the second invocation, etc.
+    //
+    // This is a functor and not a fixed value to enable dynamic intervals (the
+    // caller can use static functor state for that). Note that
+    // `PeriodicAction::Nudge()` calls trigger out-of-schedule invocations and
+    // count as iterations (therefore incrementing the internal iteration
+    // counter and resetting the timer).
+    //
+    // If `sleep_before_each()` ever returns an `absl::InfiniteDuration()`, then
+    // periodic action execution will be paused and resumed only by the next
+    // `Nudge()` call.
+    absl::AnyInvocable<absl::Duration(uint64_t iter_num)> sleep_before_each;
   };
 
+  // Convenience factory methods for common options.
+  static Options ConstDelayConstInterval(  //
+      absl::Duration delay, absl::Duration interval) {
+    return {
+        [delay, interval](uint64_t i) { return i == 0 ? delay : interval; },
+    };
+  }
+  static Options ZeroDelayZeroInterval() {
+    return ConstDelayConstInterval(absl::ZeroDuration(), absl::ZeroDuration());
+  }
+  static Options ZeroDelayConstInterval(absl::Duration interval) {
+    return ConstDelayConstInterval(absl::ZeroDuration(), interval);
+  }
+  static Options ConstDelayZeroInterval(absl::Duration delay) {
+    return ConstDelayConstInterval(delay, absl::ZeroDuration());
+  }
+
   PeriodicAction(absl::AnyInvocable<void()> action, Options options);
 
   // Movable, but not copyable.

centipede/periodic_action_test.cc

@@ -16,6 +16,7 @@
 
 #include <algorithm>
 #include <cmath>
+#include <cstddef>
 
 #include "gtest/gtest.h"
 #include "absl/time/clock.h"
@@ -33,7 +34,7 @@ TEST(PeriodicActionTest, OnlyPeriodicInvocations) {
   int count = 0;
   PeriodicAction action{
       [&count]() { ++count; },
-      {.delay = absl::ZeroDuration(), .interval = kPeriodicInterval},
+      PeriodicAction::ZeroDelayConstInterval(kPeriodicInterval),
   };
   absl::SleepFor(kDuration);
   action.Stop();
@@ -47,16 +48,18 @@ TEST(PeriodicActionTest, OnlyNudgedInvocations) {
   int count = 0;
   PeriodicAction action{
       [&count]() { ++count; },
-      // Effectively disable invocations done periodically: only explicit
-      // `Nudge()` calls below will trigger them.
-      {.delay = absl::InfiniteDuration(), .interval = absl::InfiniteDuration()},
+      {
+          // Effectively disable periodic invocations: only `Nudge()` calls
+          // below will trigger them.
+          .sleep_before_each = [](size_t) { return absl::InfiniteDuration(); },
+      },
   };
   int expected_count = 0;
   const absl::Time end_time = absl::Now() + kDuration;
   while (absl::Now() < end_time) {
     action.Nudge();
-    // Sleep after a nudge, not before, to guarantee that the action has time to
-    // finish and increment `count`.
+    // Sleep after a nudge, not before, to guarantee that the action has time
+    // to finish and increment `count`.
     absl::SleepFor(kNudgeInterval);
     ++expected_count;
   }
@@ -82,7 +85,7 @@ TEST(PeriodicActionTest, PeriodicAndNudgedInvocations) {
   int count = 0;
   PeriodicAction action{
       [&count]() { ++count; },
-      {.delay = absl::ZeroDuration(), .interval = kPeriodicInterval},
+      PeriodicAction::ZeroDelayConstInterval(kPeriodicInterval),
   };
   const absl::Time end_time = absl::Now() + kDuration;
   while (absl::Now() < end_time) {
@@ -111,7 +114,7 @@ TEST(PeriodicActionTest, ClashingPeriodicAndNudgedInvocations) {
   int count = 0;
   PeriodicAction action{
       [&count]() { ++count; },
-      {.delay = absl::ZeroDuration(), .interval = kPeriodicInterval},
+      PeriodicAction::ZeroDelayConstInterval(kPeriodicInterval),
   };
   const absl::Time end_time = absl::Now() + kDuration;
   while (absl::Now() < end_time) {

centipede/rusage_profiler.cc

@@ -430,10 +430,7 @@ void RUsageProfiler::StartTimelapse(  //
         const auto& s = TakeSnapshot(loc, title);
         if (also_log) s.Log();
       },
-      PeriodicAction::Options{
-          .delay = absl::ZeroDuration(),
-          .interval = interval,
-      });
+      PeriodicAction::ZeroDelayConstInterval(interval));
 }
 
 void RUsageProfiler::StopTimelapse() {

centipede/stats.h

@@ -472,6 +472,9 @@ class StatsLogger : public StatsReporter {
   using StatsReporter::StatsReporter;
   ~StatsLogger() override = default;
 
+  StatsLogger(StatsLogger &&) = default;
+  StatsLogger &operator=(StatsLogger &&) = default;
+
  private:
   bool ShouldReportThisField(const Stats::FieldInfo &field) override;
   void PreAnnounceFields(
@@ -507,6 +510,9 @@ class StatsCsvFileAppender : public StatsReporter {
   using StatsReporter::StatsReporter;
   ~StatsCsvFileAppender() override;
 
+  StatsCsvFileAppender(StatsCsvFileAppender &&) = default;
+  StatsCsvFileAppender &operator=(StatsCsvFileAppender &&) = default;
+
  private:
   struct BufferedRemoteFile {
     RemoteFile *file = nullptr;