fcb206b667
Summary: Allow users to give a callback function with parameter using sync point, so more complicated verification can be done in tests. Use it in DBTest.DynamicLevelCompressionPerLevel2 so that failures will be more easy to debug. Test Plan: Run all tests. Run DBTest.DynamicLevelCompressionPerLevel2 with valgrind check. Reviewers: rven, yhchiang, anthony, kradhakrishnan, igor Reviewed By: igor Subscribers: leveldb, dhruba Differential Revision: https://reviews.facebook.net/D36999
90 lines
2.3 KiB
C++
90 lines
2.3 KiB
C++
// Copyright (c) 2014, Facebook, Inc. All rights reserved.
|
|
// This source code is licensed under the BSD-style license found in the
|
|
// LICENSE file in the root directory of this source tree. An additional grant
|
|
// of patent rights can be found in the PATENTS file in the same directory.
|
|
|
|
#include "util/sync_point.h"
|
|
|
|
#ifndef NDEBUG
|
|
namespace rocksdb {
|
|
|
|
SyncPoint* SyncPoint::GetInstance() {
|
|
static SyncPoint sync_point;
|
|
return &sync_point;
|
|
}
|
|
|
|
void SyncPoint::LoadDependency(const std::vector<Dependency>& dependencies) {
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
successors_.clear();
|
|
predecessors_.clear();
|
|
cleared_points_.clear();
|
|
for (const auto& dependency : dependencies) {
|
|
successors_[dependency.predecessor].push_back(dependency.successor);
|
|
predecessors_[dependency.successor].push_back(dependency.predecessor);
|
|
}
|
|
cv_.notify_all();
|
|
}
|
|
|
|
bool SyncPoint::PredecessorsAllCleared(const std::string& point) {
|
|
for (const auto& pred : predecessors_[point]) {
|
|
if (cleared_points_.count(pred) == 0) {
|
|
return false;
|
|
}
|
|
}
|
|
return true;
|
|
}
|
|
|
|
void SyncPoint::SetCallBack(const std::string point,
|
|
std::function<void(void*)> callback) {
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
callbacks_[point] = callback;
|
|
}
|
|
|
|
void SyncPoint::ClearAllCallBacks() {
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
while (num_callbacks_running_ > 0) {
|
|
cv_.wait(lock);
|
|
}
|
|
callbacks_.clear();
|
|
}
|
|
|
|
void SyncPoint::EnableProcessing() {
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
enabled_ = true;
|
|
}
|
|
|
|
void SyncPoint::DisableProcessing() {
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
enabled_ = false;
|
|
}
|
|
|
|
void SyncPoint::ClearTrace() {
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
cleared_points_.clear();
|
|
}
|
|
|
|
void SyncPoint::Process(const std::string& point, void* cb_arg) {
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
|
|
if (!enabled_) return;
|
|
|
|
auto callback_pair = callbacks_.find(point);
|
|
if (callback_pair != callbacks_.end()) {
|
|
num_callbacks_running_++;
|
|
mutex_.unlock();
|
|
callback_pair->second(cb_arg);
|
|
mutex_.lock();
|
|
num_callbacks_running_--;
|
|
cv_.notify_all();
|
|
}
|
|
|
|
while (!PredecessorsAllCleared(point)) {
|
|
cv_.wait(lock);
|
|
}
|
|
|
|
cleared_points_.insert(point);
|
|
cv_.notify_all();
|
|
}
|
|
} // namespace rocksdb
|
|
#endif // NDEBUG
|