jabberwocky238/rocksdb
1
0
Fork 0
mirror of https://github.com/facebook/rocksdb.git synced 2026-07-07 14:47:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

Revert trace changes (#14600)

Browse Source 
Summary:
Reverts the two commits https://github.com/facebook/rocksdb/pull/14578 https://github.com/facebook/rocksdb/pull/14575 that changed tracer behavior. The correct behavior is that trace should be written to before WAL.

Although a trace file may not be fully consumed, it is recycles after each crash, so there is no need to worry about applying an uncomited write. The original crash test was a false positive for a different error.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/14600

Reviewed By: xingbowang

Differential Revision: D100397989

Pulled By: joshkang97

fbshipit-source-id: 3da6fb80f682ac6f9529c1a76eaf169e38cf2477
This commit is contained in:
Josh Kang
2026-04-10 16:04:37 -07:00
committed by meta-codesync[bot]
parent c47e35395b
commit f3a20854e0
3 changed files with 50 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/db_impl/db_impl.h
-4
View File
@@ -1747,10 +1747,6 @@ class DBImpl : public DB {
PreReleaseCallback* pre_release_callback, const AssignOrder assign_order,
const PublishLastSeq publish_last_seq, const bool disable_memtable);
void MaybeTraceWriteGroupForPreservedWriteOrder(
const WriteThread::WriteGroup& write_group,
WriteBatchWithIndex* wbwi = nullptr, bool ingest_wbwi_for_commit = true);
// write cached_recoverable_state_ to memtable if it is not empty
// The writer must be the leader in write_thread_ and holding mutex_
Status WriteRecoverableState();
db/db_impl/db_impl_write.cc
+50 -46
View File
@@ -862,37 +862,6 @@ Status DBImpl::SyncBlobDirectWriteManagers(
return Status::OK();
}
void DBImpl::MaybeTraceWriteGroupForPreservedWriteOrder(
const WriteThread::WriteGroup& write_group, WriteBatchWithIndex* wbwi,
bool ingest_wbwi_for_commit) {
// TODO: this use of operator bool on `tracer_` can avoid unnecessary lock
// grabs but does not seem thread-safe.
if (tracer_) {
InstrumentedMutexLock lock(&trace_mutex_);
if (tracer_ != nullptr && tracer_->IsWriteOrderPreserved()) {
WriteBatch* wbwi_trace_batch = nullptr;
if (wbwi != nullptr && !ingest_wbwi_for_commit) {
// For transaction write, preserved-order tracing only needs the
// logical commit marker once, so trace the WBWI batch instead of the
// commit-time batch stored in each writer.
wbwi_trace_batch = wbwi->GetWriteBatch();
}
for (auto* writer : write_group) {
if (writer->CallbackFailed()) {
continue;
}
WriteBatch* trace_batch = wbwi_trace_batch != nullptr
? wbwi_trace_batch
: writer->trace_batch;
// Preserve user-visible ordering while still tracing the logical batch
// for writers whose applied batch was rewritten earlier on the path.
// TODO: maybe handle the tracing status?
tracer_->Write(trace_batch).PermitUncheckedError();
}
}
}
}
Status DBImpl::WriteImpl(
const WriteOptions& write_options, WriteBatch* my_batch,
WriteCallback* callback, UserWriteCallback* user_write_cb,
@@ -1316,6 +1285,26 @@ Status DBImpl::WriteImpl(
}
}
}
// TODO: this use of operator bool on `tracer_` can avoid unnecessary lock
// grabs but does not seem thread-safe.
if (tracer_) {
InstrumentedMutexLock lock(&trace_mutex_);
if (tracer_ && tracer_->IsWriteOrderPreserved()) {
for (auto* writer : write_group) {
if (writer->CallbackFailed()) {
continue;
}
// TODO: maybe handle the tracing status?
if (wbwi && !ingest_wbwi_for_commit) {
// for transaction write, tracer only needs the commit marker which
// is in writer->batch
tracer_->Write(wbwi->GetWriteBatch()).PermitUncheckedError();
} else {
tracer_->Write(writer->trace_batch).PermitUncheckedError();
}
}
}
}
// Note about seq_per_batch_: either disableWAL is set for the entire write
// group or not. In either case we inc seq for each write batch with no
// failed callback. This means that there could be a batch with
@@ -1425,11 +1414,6 @@ Status DBImpl::WriteImpl(
}
}
if (status.ok()) {
MaybeTraceWriteGroupForPreservedWriteOrder(write_group, wbwi.get(),
ingest_wbwi_for_commit);
}
// PreReleaseCallback is called after WAL write and before memtable write
if (status.ok()) {
SequenceNumber next_sequence = current_sequence;
@@ -1634,6 +1618,22 @@ Status DBImpl::PipelinedWriteImpl(const WriteOptions& write_options,
}
}
}
// TODO: this use of operator bool on `tracer_` can avoid unnecessary lock
// grabs but does not seem thread-safe.
if (tracer_) {
InstrumentedMutexLock lock(&trace_mutex_);
if (tracer_ != nullptr && tracer_->IsWriteOrderPreserved()) {
for (auto* writer : wal_write_group) {
if (writer->CallbackFailed()) {
// When optimisitc txn conflict checking fails, we should
// not record to trace.
continue;
}
// TODO: maybe handle the tracing status?
tracer_->Write(writer->trace_batch).PermitUncheckedError();
}
}
}
if (w.disable_wal) {
has_unpersisted_data_.store(true, std::memory_order_relaxed);
}
@@ -1694,9 +1694,6 @@ Status DBImpl::PipelinedWriteImpl(const WriteOptions& write_options,
const ReadOptions read_options;
w.status = ApplyWALToManifest(read_options, write_options, &synced_wals);
}
if (w.status.ok()) {
MaybeTraceWriteGroupForPreservedWriteOrder(wal_write_group);
}
write_thread_.ExitAsBatchGroupLeader(wal_write_group, w.status);
}
@@ -1908,6 +1905,20 @@ Status DBImpl::WriteImplWALOnly(
// Note: no need to update last_batch_group_size_ here since the batch writes
// to WAL only
// TODO: this use of operator bool on `tracer_` can avoid unnecessary lock
// grabs but does not seem thread-safe.
if (tracer_) {
InstrumentedMutexLock lock(&trace_mutex_);
if (tracer_ != nullptr && tracer_->IsWriteOrderPreserved()) {
for (auto* writer : write_group) {
if (writer->CallbackFailed()) {
continue;
}
// TODO: maybe handle the tracing status?
tracer_->Write(writer->trace_batch).PermitUncheckedError();
}
}
}
const bool concurrent_update = true;
// Update stats while we are an exclusive group leader, so we know
@@ -1989,13 +2000,6 @@ Status DBImpl::WriteImplWALOnly(
status = SyncWAL();
}
}
if (status.ok()) {
// Write trace after WAL right. This ensures the replayer does not replay a
// record that has not been recorded in the DB. However, it is not a full
// solution as a crash may still happen before the trace write and after the
// WAL write. TODO: Atomically write WAL and trace.
MaybeTraceWriteGroupForPreservedWriteOrder(write_group);
}
PERF_TIMER_START(write_pre_and_post_process_time);
if (!w.CallbackFailed()) {
db/db_test2.cc
-66
View File
@@ -4172,72 +4172,6 @@ TEST_F(DBTest2, TraceAndManualReplay) {
ASSERT_OK(DestroyDB(dbname2, options));
}
TEST_F(DBTest2, TracePreserveWriteOrderSkipsFailedWrite) {
for (bool enable_pipelined_write : {false, true}) {
SCOPED_TRACE("enable_pipelined_write=" +
std::to_string(enable_pipelined_write));
Options options = CurrentOptions();
options.enable_pipelined_write = enable_pipelined_write;
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(env_));
options.env = fault_env.get();
DestroyAndReopen(options);
TraceOptions trace_opts;
trace_opts.preserve_write_order = true;
const std::string trace_filename = dbname_ +
"/rocksdb.trace_failed_write." +
std::to_string(enable_pipelined_write);
std::unique_ptr<TraceWriter> trace_writer;
ASSERT_OK(
NewFileTraceWriter(env_, EnvOptions(), trace_filename, &trace_writer));
ASSERT_OK(db_->StartTrace(trace_opts, std::move(trace_writer)));
ASSERT_OK(Put("good", "1"));
fault_env->SetFilesystemActive(false);
ASSERT_NOK(Put("bad", "2"));
fault_env->SetFilesystemActive(true);
ASSERT_OK(db_->EndTrace());
Close();
options.env = env_;
Reopen(options);
ASSERT_EQ("1", Get("good"));
ASSERT_EQ("NOT_FOUND", Get("bad"));
Close();
const std::string replay_dbname = test::PerThreadDBPath(
env_, "/db_replay_failed." + std::to_string(enable_pipelined_write));
ASSERT_OK(DestroyDB(replay_dbname, options));
options.create_if_missing = true;
std::unique_ptr<DB> replay_db;
ASSERT_OK(DB::Open(options, replay_dbname, &replay_db));
std::unique_ptr<TraceReader> trace_reader;
ASSERT_OK(
NewFileTraceReader(env_, EnvOptions(), trace_filename, &trace_reader));
std::unique_ptr<Replayer> replayer;
ASSERT_OK(replay_db->NewDefaultReplayer({replay_db->DefaultColumnFamily()},
std::move(trace_reader),
&replayer));
ASSERT_OK(replayer->Prepare());
ASSERT_OK(replayer->Replay(ReplayOptions(), nullptr));
std::string value;
ASSERT_OK(replay_db->Get(ReadOptions(), "good", &value));
ASSERT_EQ("1", value);
ASSERT_TRUE(replay_db->Get(ReadOptions(), "bad", &value).IsNotFound());
replay_db.reset();
ASSERT_OK(DestroyDB(replay_dbname, options));
}
}
TEST_F(DBTest2, TraceWithLimit) {
Options options = CurrentOptions();
options.merge_operator = MergeOperators::CreatePutOperator();