Commit 465c86b5 authored by Tudor Bosman's avatar Tudor Bosman

EventCount: "condition variable" for lock-free algorithms

Summary:
This allows you to convert non-blocking lock-free algorithms into
blocking versions.

See
http://www.1024cores.net/home/lock-free-algorithms/eventcounts,
http://software.intel.com/en-us/forums/showthread.php?t=62364,
http://www.akkadia.org/drepper/futex.pdf

Test Plan: new test

Reviewed By: delong.j@fb.com

FB internal diff: D474570
parent 9c292425
/*
* Copyright 2012 Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#ifndef FOLLY_EXPERIMENTAL_EVENTCOUNT_H_
#define FOLLY_EXPERIMENTAL_EVENTCOUNT_H_
#include <syscall.h>
#include <linux/futex.h>
#include <sys/time.h>
#include <climits>
#include <atomic>
#include <thread>
namespace folly {
namespace detail {
int futex(int* uaddr, int op, int val, const struct timespec* timeout,
int* uaddr2, int val3) noexcept {
return syscall(SYS_futex, uaddr, op, val, timeout, uaddr2, val3);
}
} // namespace detail
/**
* Event count: a condition variable for lock free algorithms.
*
* See http://www.1024cores.net/home/lock-free-algorithms/eventcounts for
* details.
*
* Event counts allow you to convert a non-blocking lock-free / wait-free
* algorithm into a blocking one, by isolating the blocking logic. You call
* prepareWait() before checking your condition and then either cancelWait()
* or wait() depending on whether the condition was true. When another
* thread makes the condition true, it must call notify() / notifyAll() just
* like a regular condition variable.
*
* If "<" denotes the happens-before relationship, consider 2 threads (T1 and
* T2) and 3 events:
* - E1: T1 returns from prepareWait
* - E2: T1 calls wait
* (obviously E1 < E2, intra-thread)
* - E3: T2 calls notifyAll
*
* If E1 < E3, then E2's wait will complete (and T1 will either wake up,
* or not block at all)
*
* This means that you can use an EventCount in the following manner:
*
* Waiter:
* if (!condition()) { // handle fast path first
* for (;;) {
* auto key = eventCount.prepareWait();
* if (condition()) {
* eventCount.cancelWait();
* break;
* } else {
* eventCount.wait(key);
* }
* }
* }
*
* (This pattern is encapsulated in await())
*
* Poster:
* make_condition_true();
* eventCount.notifyAll();
*
* Note that, just like with regular condition variables, the waiter needs to
* be tolerant of spurious wakeups and needs to recheck the condition after
* being woken up. Also, as there is no mutual exclusion implied, "checking"
* the condition likely means attempting an operation on an underlying
* data structure (push into a lock-free queue, etc) and returning true on
* success and false on failure.
*/
class EventCount {
public:
EventCount() noexcept : epoch_(0), waiters_(0) { }
class Key {
friend class EventCount;
explicit Key(int e) noexcept : epoch_(e) { }
int epoch_;
};
void notify() noexcept;
void notifyAll() noexcept;
Key prepareWait() noexcept;
void cancelWait() noexcept;
void wait(Key key) noexcept;
/**
* Wait for condition() to become true. Will clean up appropriately if
* condition() throws, and then rethrow.
*/
template <class Condition>
void await(Condition condition);
private:
void doNotify(int n) noexcept;
EventCount(const EventCount&) = delete;
EventCount(EventCount&&) = delete;
EventCount& operator=(const EventCount&) = delete;
EventCount& operator=(EventCount&&) = delete;
std::atomic<int> epoch_;
std::atomic<int> waiters_;
};
inline void EventCount::notify() noexcept {
doNotify(1);
}
inline void EventCount::notifyAll() noexcept {
doNotify(INT_MAX);
}
inline void EventCount::doNotify(int n) noexcept {
// The order is important: epoch_ is incremented before waiters_ is checked.
// prepareWait() increments waiters_ before checking epoch_, so it is
// impossible to miss a wakeup.
++epoch_;
if (waiters_ != 0) {
detail::futex(reinterpret_cast<int*>(&epoch_), FUTEX_WAKE, n, nullptr,
nullptr, 0);
}
}
inline EventCount::Key EventCount::prepareWait() noexcept {
++waiters_;
return Key(epoch_);
}
inline void EventCount::cancelWait() noexcept {
--waiters_;
}
inline void EventCount::wait(Key key) noexcept {
while (epoch_ == key.epoch_) {
detail::futex(reinterpret_cast<int*>(&epoch_), FUTEX_WAIT, key.epoch_,
nullptr, nullptr, 0);
}
--waiters_;
}
template <class Condition>
void EventCount::await(Condition condition) {
if (condition()) return; // fast path
// condition() is the only thing that may throw, everything else is
// noexcept, so we can hoist the try/catch block outside of the loop
try {
for (;;) {
auto key = prepareWait();
if (condition()) {
cancelWait();
break;
} else {
wait(key);
}
}
} catch (...) {
cancelWait();
throw;
}
}
} // namespace folly
#endif /* FOLLY_EXPERIMENTAL_EVENTCOUNT_H_ */
/*
* Copyright 2012 Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "folly/experimental/EventCount.h"
#include <algorithm>
#include <random>
#include <glog/logging.h>
#include <gtest/gtest.h>
#include "folly/Random.h"
#include "folly/Benchmark.h"
using namespace folly;
namespace {
class Semaphore {
public:
explicit Semaphore(int v=0) : value_(v) { }
void down() {
ec_.await([this] { return tryDown(); });
}
void up() {
++value_;
ec_.notifyAll();
}
int value() const {
return value_;
}
private:
bool tryDown() {
for (;;) {
int v = value_;
if (v == 0) {
return false;
}
if (value_.compare_exchange_weak(v, v-1)) {
return true;
}
}
}
std::atomic<int> value_;
EventCount ec_;
};
template <class T, class Random>
void randomPartition(Random& random, T key, int n,
std::vector<std::pair<T, int>>& out) {
while (n != 0) {
int m = std::min(n, 1000);
std::uniform_int_distribution<uint32_t> u(1, m);
int cut = u(random);
out.push_back(std::make_pair(key, cut));
n -= cut;
}
}
} // namespace
TEST(EventCount, Simple) {
// We're basically testing for no deadlock.
static const size_t count = 300000;
enum class Op {
UP,
DOWN
};
std::vector<std::pair<Op, int>> ops;
std::mt19937 rnd(randomNumberSeed());
randomPartition(rnd, Op::UP, count, ops);
size_t uppers = ops.size();
randomPartition(rnd, Op::DOWN, count, ops);
size_t downers = ops.size() - uppers;
VLOG(1) << "Using " << ops.size() << " threads: uppers=" << uppers
<< " downers=" << downers << " sem_count=" << count;
std::random_shuffle(ops.begin(), ops.end());
std::vector<std::thread> threads;
threads.reserve(ops.size());
Semaphore sem;
for (auto& op : ops) {
int n = op.second;
if (op.first == Op::UP) {
auto fn = [&sem, n] () mutable {
while (n--) {
sem.up();
}
};
threads.push_back(std::thread(fn));
} else {
auto fn = [&sem, n] () mutable {
while (n--) {
sem.down();
}
};
threads.push_back(std::thread(fn));
}
}
for (auto& thread : threads) {
thread.join();
}
EXPECT_EQ(0, sem.value());
}
int main(int argc, char *argv[]) {
testing::InitGoogleTest(&argc, argv);
google::ParseCommandLineFlags(&argc, &argv, true);
auto ret = RUN_ALL_TESTS();
if (!ret) {
folly::runBenchmarksOnFlag();
}
return ret;
}
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment