Fix TSAN failures in DistributedMutex tests (#5684)
Summary:
TSAN was not able to correctly instrument atomic bts and btr instructions, so
when TSAN is enabled implement those with std::atomic::fetch_or and
std::atomic::fetch_and. Also disable tests that fail on TSAN with false
negatives (we know these are false negatives because this other verifiably
correct program fails with the same TSAN error <link>)
```
make clean
TEST_TMPDIR=/dev/shm/rocksdb OPT=-g COMPILE_WITH_TSAN=1 make J=1 -j56 folly_synchronization_distributed_mutex_test
```
This is the code that fails with the same false-negative with TSAN
```
namespace {
class ExceptionWithConstructionTrack : public std::exception {
public:
explicit ExceptionWithConstructionTrack(int id)
: id_{folly::to<std::string>(id)}, constructionTrack_{id} {}
const char* what() const noexcept override {
return id_.c_str();
}
private:
std::string id_;
TestConstruction constructionTrack_;
};
template <typename Storage, typename Atomic>
void transferCurrentException(Storage& storage, Atomic& produced) {
assert(std::current_exception());
new (&storage) std::exception_ptr(std::current_exception());
produced->store(true, std::memory_order_release);
}
void concurrentExceptionPropagationStress(
int numThreads,
std::chrono::milliseconds milliseconds) {
auto&& stop = std::atomic<bool>{false};
auto&& exceptions = std::vector<std::aligned_storage<48, 8>::type>{};
auto&& produced = std::vector<std::unique_ptr<std::atomic<bool>>>{};
auto&& consumed = std::vector<std::unique_ptr<std::atomic<bool>>>{};
auto&& consumers = std::vector<std::thread>{};
for (auto i = 0; i < numThreads; ++i) {
produced.emplace_back(new std::atomic<bool>{false});
consumed.emplace_back(new std::atomic<bool>{false});
exceptions.push_back({});
}
auto producer = std::thread{[&]() {
auto counter = std::vector<int>(numThreads, 0);
for (auto i = 0; true; i = ((i + 1) % numThreads)) {
try {
throw ExceptionWithConstructionTrack{counter.at(i)++};
} catch (...) {
transferCurrentException(exceptions.at(i), produced.at(i));
}
while (!consumed.at(i)->load(std::memory_order_acquire)) {
if (stop.load(std::memory_order_acquire)) {
return;
}
}
consumed.at(i)->store(false, std::memory_order_release);
}
}};
for (auto i = 0; i < numThreads; ++i) {
consumers.emplace_back([&, i]() {
auto counter = 0;
while (true) {
while (!produced.at(i)->load(std::memory_order_acquire)) {
if (stop.load(std::memory_order_acquire)) {
return;
}
}
produced.at(i)->store(false, std::memory_order_release);
try {
auto storage = &exceptions.at(i);
auto exc = folly::launder(
reinterpret_cast<std::exception_ptr*>(storage));
auto copy = std::move(*exc);
exc->std::exception_ptr::~exception_ptr();
std::rethrow_exception(std::move(copy));
} catch (std::exception& exc) {
auto value = std::stoi(exc.what());
EXPECT_EQ(value, counter++);
}
consumed.at(i)->store(true, std::memory_order_release);
}
});
}
std::this_thread::sleep_for(milliseconds);
stop.store(true);
producer.join();
for (auto& thread : consumers) {
thread.join();
}
}
} // namespace
```
Pull Request resolved: https://github.com/facebook/rocksdb/pull/5684
Differential Revision: D16746077
Pulled By: miasantreble
fbshipit-source-id: 8af88dcf9161c05daec1a76290f577918638f79d
77273d4137