fix: Fix bugs around rdb save and improve memory accounting (#495)

Signed-off-by: Roman Gershman <roman@dragonflydb.io>

src/server/error.h

@@ -11,21 +11,21 @@
 
 namespace dfly {
 
-using facade::kWrongTypeErr;
+using facade::kDbIndOutOfRangeErr;
+using facade::kInvalidDbIndErr;
 using facade::kInvalidIntErr;
 using facade::kSyntaxErr;
-using facade::kInvalidDbIndErr;
-using facade::kDbIndOutOfRangeErr;
+using facade::kWrongTypeErr;
 
 #ifndef RETURN_ON_ERR
 
-#define RETURN_ON_ERR(x) \
-  do {                   \
-    auto __ec = (x);       \
-    if (__ec) {            \
-      VLOG(1) << "Error " << __ec << " while calling " #x; \
-      return __ec;         \
-    }                      \
+#define RETURN_ON_ERR(x)                                      \
+  do {                                                        \
+    auto __ec = (x);                                          \
+    if (__ec) {                                               \
+      LOG(ERROR) << "Error " << __ec << " while calling " #x; \
+      return __ec;                                            \
+    }                                                         \
   } while (0)
 
 #endif  // RETURN_ON_ERR

src/server/memory_cmd.cc

@@ -52,7 +52,7 @@ void MemoryCmd::Run(CmdArgList args) {
       return (*cntx_)->SendError(kInvalidIntErr);
     }
     tid = tid % shard_set->pool()->size();
-    string res = shard_set->pool()->at(tid)->AwaitBrief([this] { return MallocStats(); });
+    string res = shard_set->pool()->at(tid)->AwaitBrief([&] { return MallocStats(tid); });
 
     return (*cntx_)->SendBulkString(res);
   }
@@ -61,28 +61,33 @@ void MemoryCmd::Run(CmdArgList args) {
   return (*cntx_)->SendError(err, kSyntaxErrType);
 }
 
-string MemoryCmd::MallocStats() {
+string MemoryCmd::MallocStats(unsigned tid) {
   string str;
 
   uint64_t start = absl::GetCurrentTimeNanos();
   absl::StrAppend(&str, "___ Begin mimalloc statistics ___\n");
   mi_stats_print_out(MiStatsCallback, &str);
 
-  absl::StrAppend(&str, "\nArena statistics from a single thread:\n");
+  absl::StrAppend(&str, "\nArena statistics from thread:", tid, "\n");
   absl::StrAppend(&str, "Count BlockSize Reserved Committed Used\n");
 
   mi_heap_t* data_heap = ServerState::tlocal()->data_heap();
   BlockMap block_map;
 
   mi_heap_visit_blocks(data_heap, false /* visit all blocks*/, MiArenaVisit, &block_map);
-
+  uint64_t reserved = 0, committed = 0, used = 0;
   for (const auto& k_v : block_map) {
     absl::StrAppend(&str, k_v.second, " ", get<0>(k_v.first), " ", get<1>(k_v.first), " ",
                     get<2>(k_v.first), " ", get<3>(k_v.first), "\n");
+    reserved += k_v.second * get<1>(k_v.first);
+    committed += k_v.second * get<2>(k_v.first);
+    used += k_v.second * get<3>(k_v.first);
   }
 
   uint64_t delta = (absl::GetCurrentTimeNanos() - start) / 1000;
   absl::StrAppend(&str, "--- End mimalloc statistics, took ", delta, "us ---\n");
+  absl::StrAppend(&str, "total reserved: ", reserved, ", comitted: ", committed, ", used: ", used,
+                  "\n");
 
   return str;
 }

src/server/memory_cmd.h

@@ -17,7 +17,7 @@ class MemoryCmd {
   void Run(CmdArgList args);
 
  private:
-  std::string MallocStats();
+  std::string MallocStats(unsigned tid);
 
   ServerFamily& sf_;
   ConnectionContext* cntx_;

src/server/rdb_save.cc

@@ -718,6 +718,9 @@ io::Result<size_t> AlignedBuffer::WriteSome(const iovec* v, uint32_t len) {
 // to the nearest page boundary.
 error_code AlignedBuffer::Flush() {
   size_t len = (buf_offs_ + kAmask) & (~kAmask);
+  if (len == 0)
+    return error_code{};
+
   iovec ivec{.iov_base = aligned_buf_, .iov_len = len};
   buf_offs_ = 0;
 
@@ -922,7 +925,7 @@ error_code RdbSaver::SaveBody(RdbTypeFreqMap* freq_map) {
 
   error_code io_error = impl_->ConsumeChannel();
   if (io_error) {
-    VLOG(1) << "io error " << io_error;
+    LOG(ERROR) << "io error " << io_error;
     return io_error;
   }