fix bug of using post write callback with empty batch (#327)

* fix bug of using post write callback with empty batch Signed-off-by: tabokie <[email protected]> * fix nullptr Signed-off-by: tabokie <[email protected]> Signed-off-by: tabokie <[email protected]> Signed-off-by: Yang Zhang <[email protected]>
tikv · Oct 2, 2024 · 81fbe5e · 81fbe5e
1 parent 4cb9108
commit 81fbe5e
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 2 deletions.
diff --git a/db/db_impl/db_impl_write.cc b/db/db_impl/db_impl_write.cc
@@ -303,6 +303,8 @@ Status DBImpl::MultiBatchWriteImpl(const WriteOptions& write_options,
               next_sequence += count;
               total_count += count;
               memtable_write_cnt++;
+            } else if (w->post_callback) {
+              w->post_callback->Callback(w->sequence);
             }
           }
           total_byte_size = WriteBatchInternal::AppendedByteSize(

diff --git a/db/db_write_test.cc b/db/db_write_test.cc
@@ -922,7 +922,6 @@ TEST_P(DBWriteTest, PostWriteCallback) {
     opts.wait = false;
     can_flush_mutex.Lock();
     ASSERT_OK(dbfull()->Flush(opts));
-    can_flush_mutex.Unlock();
     flushed.store(true, std::memory_order_relaxed);
   }));
 
@@ -931,7 +930,7 @@ TEST_P(DBWriteTest, PostWriteCallback) {
   ASSERT_EQ(flushed.load(std::memory_order_relaxed), false);
 
   the_first_can_exit_write_mutex.Unlock();
-  std::this_thread::sleep_for(std::chrono::milliseconds{10});
+  std::this_thread::sleep_for(std::chrono::milliseconds{50});
   ASSERT_EQ(written.load(std::memory_order_relaxed), 2);
   ASSERT_EQ(flushed.load(std::memory_order_relaxed), true);
 
@@ -940,6 +939,23 @@ TEST_P(DBWriteTest, PostWriteCallback) {
   }
 }
 
+TEST_P(DBWriteTest, PostWriteCallbackEmptyBatch) {
+  Options options = GetOptions();
+  if (options.two_write_queues) {
+    // Not compatible.
+    return;
+  }
+  Reopen(options);
+  WriteBatch batch;
+  WriteOptions opts;
+  opts.sync = false;
+  opts.disableWAL = true;
+  SequenceNumber seq = 0;
+  SimpleCallback callback([&](SequenceNumber s) { seq = s; });
+  ASSERT_OK(dbfull()->Write(opts, &batch, &callback));
+  ASSERT_NE(seq, 0);
+}
+
 INSTANTIATE_TEST_CASE_P(DBWriteTestInstance, DBWriteTest,
                         testing::Values(DBTestBase::kDefault,
                                         DBTestBase::kConcurrentWALWrites,