#Centipede Atomically write CSV lines to fuzzing-stats*.csv reports

Before, each field sample would be immediately appended to the report. That created problems:

1. External readers had to be able to tolerate incomplete last lines in reports.
1. If the process was killed in the middle of a line and restarted in the same workdir, the first new line would be appended to the unfinished line. The resulting line was malformed.

Also, with network file systems, writing a line at a time means fewer write requests sent to the file server (however, the benefit is likely minor, given the size and frequency of writes here).

PiperOrigin-RevId: 617693512

centipede/BUILD

diff --git a/centipede/BUILD b/centipede/BUILD
index 9ee4f7a..81a80a6 100644
--- a/centipede/BUILD
+++ b/centipede/BUILD
@@ -378,6 +378,7 @@
         ":logging",
         ":remote_file",
         ":workdir",
+        "@com_google_absl//absl/base:nullability",
         "@com_google_absl//absl/container:btree",
         "@com_google_absl//absl/container:flat_hash_map",
         "@com_google_absl//absl/log",

centipede/stats.cc

diff --git a/centipede/stats.cc b/centipede/stats.cc
index ee5b771..db93370 100644
--- a/centipede/stats.cc
+++ b/centipede/stats.cc
@@ -172,7 +172,7 @@
 
 StatsCsvFileAppender::~StatsCsvFileAppender() {
   for (const auto &[group_name, file] : files_) {
-    RemoteFileClose(file);
+    RemoteFileClose(file.file);
   }
 }
 
@@ -194,8 +194,8 @@
 }
 
 void StatsCsvFileAppender::SetCurrGroup(const Environment &master_env) {
-  RemoteFile *&file = files_[master_env.experiment_name];
-  if (file == nullptr) {
+  BufferedRemoteFile &file = files_[master_env.experiment_name];
+  if (file.file == nullptr) {
     const std::string filename =
         WorkDir{master_env}.FuzzingStatsPath(master_env.experiment_name);
     // If a non-empty file already exists and has the same CVS header, then
@@ -215,11 +215,11 @@
         RemoteFileSetContents(GetBackupFilename(filename), contents);
       }
     }
-    file = RemoteFileOpen(filename, append ? "a" : "w");
-    CHECK(file != nullptr) << VV(filename);
-    if (!append) RemoteFileAppend(file, csv_header_);
+    file.file = RemoteFileOpen(filename, append ? "a" : "w");
+    CHECK(file.file != nullptr) << VV(filename);
+    if (!append) RemoteFileAppend(file.file, csv_header_);
   }
-  curr_file_ = file;
+  curr_file_ = &file;
 }
 
 void StatsCsvFileAppender::SetCurrField(const Stats::FieldInfo &field_info) {
@@ -238,7 +238,8 @@
   }
   double avg = !values.empty() ? (1.0 * sum / values.size()) : 0;
 
-  std::string values_str;
+  CHECK(curr_file_ != nullptr);
+  std::string &values_str = curr_file_->buffer;
   if (curr_field_info_.traits & TraitBits::kMin)
     absl::StrAppendFormat(&values_str, "%" PRIu64 ",", min);
   if (curr_field_info_.traits & TraitBits::kMax)
@@ -247,8 +248,6 @@
     absl::StrAppendFormat(&values_str, "%.1lf,", avg);
   if (curr_field_info_.traits & TraitBits::kSum)
     absl::StrAppendFormat(&values_str, "%" PRIu64 ",", sum);
-
-  RemoteFileAppend(curr_file_, values_str);
 }
 
 void StatsCsvFileAppender::ReportFlags(const GroupToFlags &group_to_flags) {
@@ -257,8 +256,9 @@
 }
 
 void StatsCsvFileAppender::DoneFieldSamplesBatch() {
-  for (const auto &[group_name, file] : files_) {
-    RemoteFileAppend(file, "\n");
+  for (auto &&[group_name, file] : files_) {
+    RemoteFileAppend(file.file, absl::StrCat(file.buffer, "\n"));
+    file.buffer.clear();
   }
 }
 

centipede/stats.h

diff --git a/centipede/stats.h b/centipede/stats.h
index faf14a6..cd870f7 100644
--- a/centipede/stats.h
+++ b/centipede/stats.h
@@ -15,6 +15,7 @@
 #ifndef THIRD_PARTY_CENTIPEDE_STATS_H_
 #define THIRD_PARTY_CENTIPEDE_STATS_H_
 
+#include <atomic>
 #include <cstdint>
 #include <cstdlib>
 #include <initializer_list>
@@ -24,6 +25,7 @@
 #include <string_view>
 #include <vector>
 
+#include "absl/base/nullability.h"
 #include "absl/container/btree_map.h"
 #include "absl/container/flat_hash_map.h"
 #include "absl/types/span.h"
@@ -379,6 +381,11 @@
   ~StatsCsvFileAppender() override;
 
  private:
+  struct BufferedRemoteFile {
+    RemoteFile *file = nullptr;
+    std::string buffer;
+  };
+
   void PreAnnounceFields(
       std::initializer_list<Stats::FieldInfo> fields) override;
   void SetCurrGroup(const Environment &master_env) override;
@@ -392,8 +399,8 @@
   virtual std::string GetBackupFilename(const std::string &filename) const;
 
   std::string csv_header_;
-  absl::flat_hash_map<std::string /*group_name*/, RemoteFile *> files_;
-  RemoteFile *curr_file_;
+  absl::flat_hash_map<std::string /*group_name*/, BufferedRemoteFile> files_;
+  absl::Nullable<BufferedRemoteFile *> curr_file_ = nullptr;
   Stats::FieldInfo curr_field_info_;
 };