Cyber: cyber_recorder file writer threading bug fix

sxwanzi · Oct 27, 2020 · 3bf0934 · 3bf0934
1 parent 87127ff
commit 3bf0934
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 5 deletions.
diff --git a/cyber/record/file/record_file_base.h b/cyber/record/file/record_file_base.h
@@ -45,7 +45,7 @@ class RecordFileBase {
   std::string path_;
   proto::Header header_;
   proto::Index index_;
-  int fd_;
+  int fd_ = -1;
 };
 
 }  // namespace record

diff --git a/cyber/record/file/record_file_writer.cc b/cyber/record/file/record_file_writer.cc
@@ -35,7 +35,7 @@ using apollo::cyber::proto::Header;
 using apollo::cyber::proto::SectionType;
 using apollo::cyber::proto::SingleIndex;
 
-RecordFileWriter::RecordFileWriter() {}
+RecordFileWriter::RecordFileWriter() : is_writing_(false) {}
 
 RecordFileWriter::~RecordFileWriter() { Close(); }
 
@@ -66,7 +66,13 @@ bool RecordFileWriter::Open(const std::string& path) {
 void RecordFileWriter::Close() {
   if (is_writing_) {
     // wait for the flush operation that may exist now
-    while (!chunk_flush_->empty()) {
+    while (1) {
+      {
+        std::unique_lock<std::mutex> flush_lock(flush_mutex_);
+        if (chunk_flush_->empty()) {
+          break;
+        }
+      }
       std::this_thread::sleep_for(std::chrono::milliseconds(100));
     }
 
@@ -78,7 +84,13 @@ void RecordFileWriter::Close() {
     }
 
     // wait for the last flush operation
-    while (!chunk_flush_->empty()) {
+    while (1) {
+      {
+        std::unique_lock<std::mutex> flush_lock(flush_mutex_);
+        if (chunk_flush_->empty()) {
+          break;
+        }
+      }
       std::this_thread::sleep_for(std::chrono::milliseconds(100));
     }
 

diff --git a/cyber/record/file/record_file_writer.h b/cyber/record/file/record_file_writer.h
@@ -92,7 +92,7 @@ class RecordFileWriter : public RecordFileBase {
   bool WriteSection(const T& message);
   bool WriteIndex();
   void Flush();
-  bool is_writing_ = false;
+  std::atomic_bool is_writing_;
   std::unique_ptr<Chunk> chunk_active_ = nullptr;
   std::unique_ptr<Chunk> chunk_flush_ = nullptr;
   std::shared_ptr<std::thread> flush_thread_ = nullptr;