Enhance/fix FileBasedWAL (vesoft-inc#27)

* Enhance/fix FileBasedWAL

1) Support log rollback
2) Separate buffer flusher logic, so that the number of flushers is independent from the number of WALs/Shards
3) Improve the multiple threads support, but still not support multiple threads calling appendLogs()/rollbackToLog()
4) Add UT for rollback
5) Fix the flaky UTs

* Rebased and addressed dutor's comments

CMakeLists.txt

@@ -37,8 +37,9 @@ set(CMAKE_EXE_LINKER_FLAGS "-static-libstdc++")
 
 # Possible values are Debug, Release, RelWithDebInfo, MinSizeRel
 if(NOT CMAKE_BUILD_TYPE)
- set(CMAKE_BUILD_TYPE "Debug" FORCE)
+ set(CMAKE_BUILD_TYPE "Debug")
 endif(NOT CMAKE_BUILD_TYPE)
+message("== Build type is " ${CMAKE_BUILD_TYPE} " ==")
 
 set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY "_build")
 set(CMAKE_LIBRARY_OUTPUT_DIRECTORY "_build")

common/base/Base.h

@@ -34,6 +34,7 @@
 #include <list>
 #include <unordered_map>
 #include <unordered_set>
+#include <queue>
 #include <deque>
 
 #include <cstdio>
@@ -59,6 +60,7 @@
 #include <folly/Varint.h>
 #include <folly/dynamic.h>
 #include <folly/json.h>
+#include <folly/RWSpinLock.h>
 
 #include "thread/NamedThread.h"
 //#include "base/StringUnorderedMap.h"

common/fs/TempDir.cpp

@@ -17,7 +17,7 @@ TempDir::TempDir(const char* pathTemplate, bool deleteOnDestroy)
  strncpy(name.get(), pathTemplate, len);
  name.get()[len] = '\0';
 
- VLOG(2) << "Trying to create the temp directory \""
+ VLOG(2) << "Trying to create the temp directory with pattern \""
  << name.get() << "\"";
 
  if (!mkdtemp(name.get())) {
@@ -27,6 +27,7 @@ TempDir::TempDir(const char* pathTemplate, bool deleteOnDestroy)
  << strerror(errno);
  } else {
  dirPath_ = std::move(name);
+ VLOG(2) << "Created temporary directory " << dirPath_.get();
  }
 }
 

dataman/RowSetWriter.cpp

@@ -20,7 +20,7 @@ RowSetWriter::RowSetWriter(const SchemaProviderIf* schema,
 void RowSetWriter::writeRowLength(int64_t len) {
  uint8_t buf[10];
  size_t lenBytes = folly::encodeVarint(len, buf);
- DCHECK_GT(lenBytes, 0);
+ DCHECK_GT(lenBytes, 0UL);
  data_.append(reinterpret_cast<char*>(buf), lenBytes);
 }
 

raftex/BufferFlusher.cpp

@@ -0,0 +1,86 @@
+/* Copyright (c) 2018 - present, VE Software Inc. All rights reserved
+ *
+ * This source code is licensed under Apache 2.0 License
+ * (found in the LICENSE.Apache file in the root directory)
+ */
+
+#include "base/Base.h"
+#include "raftex/BufferFlusher.h"
+#include "raftex/FileBasedWal.h"
+
+namespace vesoft {
+namespace vgraph {
+namespace raftex {
+
+BufferFlusher::BufferFlusher()
+ : flushThread_("Buffer flusher",
+ std::bind(&BufferFlusher::flushLoop, this)) {
+}
+
+
+BufferFlusher::~BufferFlusher() {
+ {
+ std::lock_guard<std::mutex> g(buffersLock_);
+ stopped_ = true;
+ }
+
+ bufferReadyCV_.notify_one();
+
+ flushThread_.join();
+ CHECK(buffers_.empty());
+}
+
+
+bool BufferFlusher::flushBuffer(std::shared_ptr<FileBasedWal> wal,
+ BufferPtr buffer) {
+ {
+ std::lock_guard<std::mutex> g(buffersLock_);
+
+ if (stopped_) {
+ LOG(ERROR) << "Buffer flusher has stopped";
+ return false;
+ }
+
+ buffers_.emplace(std::move(wal), std::move(buffer));
+ }
+
+ // Notify the loop thread
+ bufferReadyCV_.notify_one();
+
+ return true;
+}
+
+
+void BufferFlusher::flushLoop() {
+ LOG(INFO) << "Buffer flusher loop started";
+
+ while(true) {
+ decltype(buffers_)::value_type bufferPair;
+ {
+ std::unique_lock<std::mutex> g(buffersLock_);
+ if (buffers_.empty()) {
+ if (stopped_) {
+ VLOG(1) << "The buffer flusher has stopped,"
+ " so exiting the flush loop";
+ break;
+ }
+ // Otherwise need to wait
+ bufferReadyCV_.wait(g, [this] {
+ return !buffers_.empty() || stopped_;
+ });
+ } else {
+ bufferPair = std::move(buffers_.front());
+ buffers_.pop();
+
+ bufferPair.first->flushBuffer(bufferPair.second);
+ }
+ }
+ }
+
+ LOG(INFO) << "Buffer flusher loop finished";
+}
+
+} // namespace raftex
+} // namespace vgraph
+} // namespace vesoft
+

raftex/BufferFlusher.h

@@ -0,0 +1,47 @@
+/* Copyright (c) 2018 - present, VE Software Inc. All rights reserved
+ *
+ * This source code is licensed under Apache 2.0 License
+ * (found in the LICENSE.Apache file in the root directory)
+ */
+
+#ifndef RAFTEX_BUFFERFLUSHER_H_
+#define RAFTEX_BUFFERFLUSHER_H_
+
+#include "base/Base.h"
+#include "thread/NamedThread.h"
+#include "raftex/InMemoryLogBuffer.h"
+
+namespace vesoft {
+namespace vgraph {
+namespace raftex {
+
+class FileBasedWal;
+
+class BufferFlusher final {
+public:
+ BufferFlusher();
+ ~BufferFlusher();
+
+ bool flushBuffer(std::shared_ptr<FileBasedWal> wal, BufferPtr buffer);
+
+private:
+ void flushLoop();
+
+private:
+ bool stopped_{false};
+
+ std::queue<
+ std::pair<std::shared_ptr<FileBasedWal>, BufferPtr>
+ > buffers_;
+ std::mutex buffersLock_;
+ std::condition_variable bufferReadyCV_;
+
+ thread::NamedThread flushThread_;
+};
+
+} // namespace raftex
+} // namespace vgraph
+} // namespace vesoft
+
+#endif // RAFTEX_BUFFERFLUSHER_H_
+

raftex/CMakeLists.txt

@@ -1,5 +1,8 @@
 add_library(
  raftex_obj OBJECT
+ BufferFlusher.cpp
+ InMemoryLogBuffer.cpp
+ FileBasedWalIterator.cpp
  FileBasedWal.cpp
 )
 add_dependencies(raftex_obj common)