Add more UT for FileBasedWAL (vesoft-inc#15)

* Fixes vesoft-inc#14 Add more UT for FileBasedWAL
tong-hao · Sep 13, 2018 · 7b2658f · 7b2658f
1 parent d383bb8
commit 7b2658f
Show file tree

Hide file tree

Showing 6 changed files with 71 additions and 7 deletions.
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -17,7 +17,7 @@ set(CMAKE_CXX_STANDARD 14)
 set(CMAKE_EXE_LINKER_FLAGS "-static-libstdc++")
 
 # Possible values are Debug, Release, RelWithDebInfo, MinSizeRel
-set(CMAKE_BUILD_TYPE "RelWithDebInfo")
+set(CMAKE_BUILD_TYPE "Debug")
 
 set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY "_build")
 set(CMAKE_LIBRARY_OUTPUT_DIRECTORY "_build")

diff --git a/common/base/Base.h b/common/base/Base.h
@@ -55,6 +55,7 @@
 #include <folly/ThreadLocal.h>
 #include <folly/Varint.h>
 
+#include "thread/NamedThread.h"
 //#include "base/StringUnorderedMap.h"
 
 #define VE_MUST_USE_RESULT              __attribute__((warn_unused_result))

diff --git a/consensus/FileBasedWal.cpp b/consensus/FileBasedWal.cpp
@@ -13,7 +13,7 @@ namespace consensus {
 
 using WalFileInfoPair = FileBasedWal::WalFiles::value_type;
 using namespace vesoft::fs;
-
+using namespace vesoft::thread;
 
 namespace internal {
 
@@ -115,6 +115,7 @@ class FileBasedWalIterator final : public LogIterator {
                     idRanges_.push_back(
                         std::make_pair(it->second.firstLogId,
                                        it->second.lastLogId));
+                    ++it;
                 }
             } else {
                 LOG(ERROR) << "LogID " << currId_
@@ -302,7 +303,8 @@ FileBasedWal::FileBasedWal(const folly::StringPiece dir,
     }
 
     bufferFlushThread_.reset(
-        new std::thread(std::bind(&FileBasedWal::bufferFlushLoop, this)));
+        new NamedThread("wal_buffer_flush_t",
+                        std::bind(&FileBasedWal::bufferFlushLoop, this)));
 }
 
 
@@ -597,12 +599,11 @@ void FileBasedWal::flushBuffer(std::shared_ptr<Buffer> buffer) {
 
     std::lock_guard<std::mutex> guard(walFilesLock_);
 
-    auto& info = walFiles_.back();
     Cord cord;
     int64_t logId = buffer->firstLogId();
     auto it = buffer->logs().begin();
     while (it != buffer->logs().end()) {
-        if (info.second.size + cord.size()
+        if (walFiles_.back().second.size + cord.size()
             + it->size()
             + sizeof(LogID)
             + sizeof(int32_t) * 2
@@ -614,7 +615,6 @@ void FileBasedWal::flushBuffer(std::shared_ptr<Buffer> buffer) {
             // Need to close the current file and create a new file
             closeCurrFile();
             prepareNewFile(logId);
-            info = walFiles_.back();
         }
 
         cord << logId << int32_t(it->size());

diff --git a/consensus/FileBasedWal.h b/consensus/FileBasedWal.h
@@ -150,7 +150,7 @@ class FileBasedWal final : public Wal,
     std::condition_variable bufferReadyCV_;
     std::condition_variable slotReadyCV_;
 
-    std::unique_ptr<std::thread> bufferFlushThread_;
+    std::unique_ptr<thread::NamedThread> bufferFlushThread_;
 
 private:  // Methods
     // Private constructor to prevent instantiating the class directly

diff --git a/consensus/test/CMakeLists.txt b/consensus/test/CMakeLists.txt
@@ -3,6 +3,7 @@ add_executable(
     FileBasedWalTest.cpp
     $<TARGET_OBJECTS:consensus_obj>
     $<TARGET_OBJECTS:base_obj>
+    $<TARGET_OBJECTS:thread_obj>
     $<TARGET_OBJECTS:fs_obj>
 )
 target_link_libraries(
@@ -15,6 +16,7 @@ target_link_libraries(
     event
     double-conversion
     dl
+    ev
     -pthread
 )
 add_test(NAME file_based_wal_test COMMAND file_based_wal_test)

diff --git a/consensus/test/FileBasedWalTest.cpp b/consensus/test/FileBasedWalTest.cpp
@@ -44,6 +44,67 @@ TEST(FileBasedWal, AppendLogs) {
         ++(*it);
         ++id;
     }
+    EXPECT_EQ(10, id);
+}
+
+
+TEST(FileBasedWal, CacheOverflow) {
+    // Make a message which length > 1KB
+    static const char* kMsg =
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz123456789"
+        "abcdefghijklmnopqrstuvwxyz123456789abcdefghijklmnopqrstuvwxyz-%06d";
+
+    // Force to make each file 1MB, each buffer is 1MB, and there are two
+    // buffers at most
+    FileBasedWalPolicy policy;
+    policy.fileSize = 1;
+    policy.bufferSize = 1;
+    policy.numBuffers = 2;
+
+    TempDir walDir("/tmp/testWal.XXXXXX");
+
+    // Create a new WAL, add one log and close it
+    auto wal = FileBasedWal::getWal(walDir.path(), policy);
+    EXPECT_EQ(-1, wal->lastLogId());
+
+    // Append > 10MB logs in total
+    for (int i = 0; i < 10000; i++) {
+        ASSERT_TRUE(wal->appendLog(i, folly::stringPrintf(kMsg, i)));
+    }
+    ASSERT_EQ(9999, wal->lastLogId());
+    // Close the wal
+    wal.reset();
+
+    // Check the number of files
+    auto files = FileUtils::listAllFilesInDir(walDir.path());
+    ASSERT_EQ(11, files.size());
+
+    // Now let's open it to read
+    wal = FileBasedWal::getWal(walDir.path(), policy);
+    EXPECT_EQ(9999, wal->lastLogId());
+
+    auto it = wal->iterator(0);
+    LogID id = 0;
+    while (it->valid()) {
+        ASSERT_EQ(id, it->logId());
+        ASSERT_EQ(folly::stringPrintf(kMsg, id), it->logMsg());
+        ++(*it);
+        ++id;
+    }
+    EXPECT_EQ(10000, id);
 }
 
 }  // namespace consensus