Commit c68fd6e3 authored by Doron Roberts-Kedes's avatar Doron Roberts-Kedes Committed by Facebook Github Bot

BufferedAtomic: add futex extensions, share impl with DeterministicAtomic

Summary:
Make code for futex[Wait/Wake]Impl for DeterministicAtomic templated, rename to deterministicFutex[Wait/Wake]Impl, and move to DeterministicSchedule.h so that it can be shared by BufferedDeterministicAtomic.

Point the futex[Wait/Wake]Impl for DeterministicAtomic at deterministicFutex[Wait/Wake]Impl<DeterministicAtomic>.

Create new futex[Wait/Wake]Impl for BufferedDeterministicAtomic using deterministicFutex[Wait/Wake]Impl<BufferedDeterministicAtomic>

Reviewed By: djwatson

Differential Revision: D13519817

fbshipit-source-id: c792ea9dcd6287236bc772e9aa9662277cc9e642
parent afce0f0b
/*
* Copyright 2018-present 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/test/BufferedAtomic.h>
namespace folly {
namespace test {
// access is protected by futexLock
static std::unordered_map<
const detail::Futex<BufferedDeterministicAtomic>*,
std::list<std::pair<uint32_t, bool*>>>
futexQueues;
static std::mutex futexLock;
int futexWakeImpl(
const folly::detail::Futex<test::BufferedDeterministicAtomic>* futex,
int count,
uint32_t wakeMask) {
return deterministicFutexWakeImpl<test::BufferedDeterministicAtomic>(
futex, futexLock, futexQueues, count, wakeMask);
}
folly::detail::FutexResult futexWaitImpl(
const folly::detail::Futex<test::BufferedDeterministicAtomic>* futex,
uint32_t expected,
std::chrono::system_clock::time_point const* absSystemTime,
std::chrono::steady_clock::time_point const* absSteadyTime,
uint32_t waitMask) {
return deterministicFutexWaitImpl<test::BufferedDeterministicAtomic>(
futex,
futexLock,
futexQueues,
expected,
absSystemTime,
absSteadyTime,
waitMask);
}
} // namespace test
} // namespace folly
...@@ -405,6 +405,19 @@ template <typename T> ...@@ -405,6 +405,19 @@ template <typename T>
using BufferedDeterministicAtomic = using BufferedDeterministicAtomic =
DeterministicAtomicImpl<T, DeterministicSchedule, BufferedAtomic>; DeterministicAtomicImpl<T, DeterministicSchedule, BufferedAtomic>;
/* Futex extensions for DeterministicSchedule based Futexes */
int futexWakeImpl(
const folly::detail::Futex<test::BufferedDeterministicAtomic>* futex,
int count,
uint32_t wakeMask);
folly::detail::FutexResult futexWaitImpl(
const folly::detail::Futex<test::BufferedDeterministicAtomic>* futex,
uint32_t expected,
std::chrono::system_clock::time_point const* absSystemTime,
std::chrono::steady_clock::time_point const* absSteadyTime,
uint32_t waitMask);
} // namespace test } // namespace test
} // namespace folly } // namespace folly
...@@ -482,106 +482,22 @@ detail::FutexResult futexWaitImpl( ...@@ -482,106 +482,22 @@ detail::FutexResult futexWaitImpl(
std::chrono::system_clock::time_point const* absSystemTimeout, std::chrono::system_clock::time_point const* absSystemTimeout,
std::chrono::steady_clock::time_point const* absSteadyTimeout, std::chrono::steady_clock::time_point const* absSteadyTimeout,
uint32_t waitMask) { uint32_t waitMask) {
using namespace test; return deterministicFutexWaitImpl<DeterministicAtomic>(
using namespace std::chrono; futex,
using namespace folly::detail; futexLock,
futexQueues,
bool hasTimeout = absSystemTimeout != nullptr || absSteadyTimeout != nullptr; expected,
bool awoken = false; absSystemTimeout,
FutexResult result = FutexResult::AWOKEN; absSteadyTimeout,
waitMask);
DeterministicSchedule::beforeSharedAccess();
FOLLY_TEST_DSCHED_VLOG(
"futexWait(" << futex << ", " << std::hex << expected << ", .., "
<< std::hex << waitMask << ") beginning..");
futexLock.lock();
// load_direct avoids deadlock on inner call to beforeSharedAccess
if (futex->load_direct() == expected) {
auto& queue = futexQueues[futex];
queue.emplace_back(waitMask, &awoken);
auto ours = queue.end();
ours--;
while (!awoken) {
futexLock.unlock();
DeterministicSchedule::afterSharedAccess();
DeterministicSchedule::beforeSharedAccess();
futexLock.lock();
// Simulate spurious wake-ups, timeouts each time with
// a 10% probability if we haven't been woken up already
if (!awoken && hasTimeout &&
DeterministicSchedule::getRandNumber(100) < 10) {
assert(futexQueues.count(futex) != 0 && &futexQueues[futex] == &queue);
queue.erase(ours);
if (queue.empty()) {
futexQueues.erase(futex);
}
// Simulate ETIMEDOUT 90% of the time and other failures
// remaining time
result = DeterministicSchedule::getRandNumber(100) >= 10
? FutexResult::TIMEDOUT
: FutexResult::INTERRUPTED;
break;
}
}
} else {
result = FutexResult::VALUE_CHANGED;
}
futexLock.unlock();
char const* resultStr = "?";
switch (result) {
case FutexResult::AWOKEN:
resultStr = "AWOKEN";
break;
case FutexResult::TIMEDOUT:
resultStr = "TIMEDOUT";
break;
case FutexResult::INTERRUPTED:
resultStr = "INTERRUPTED";
break;
case FutexResult::VALUE_CHANGED:
resultStr = "VALUE_CHANGED";
break;
}
FOLLY_TEST_DSCHED_VLOG(
"futexWait(" << futex << ", " << std::hex << expected << ", .., "
<< std::hex << waitMask << ") -> " << resultStr);
DeterministicSchedule::afterSharedAccess();
return result;
} }
int futexWakeImpl( int futexWakeImpl(
const detail::Futex<test::DeterministicAtomic>* futex, const detail::Futex<DeterministicAtomic>* futex,
int count, int count,
uint32_t wakeMask) { uint32_t wakeMask) {
using namespace test; return deterministicFutexWakeImpl<DeterministicAtomic>(
using namespace std::chrono; futex, futexLock, futexQueues, count, wakeMask);
int rv = 0;
DeterministicSchedule::beforeSharedAccess();
futexLock.lock();
if (futexQueues.count(futex) > 0) {
auto& queue = futexQueues[futex];
auto iter = queue.begin();
while (iter != queue.end() && rv < count) {
auto cur = iter++;
if ((cur->first & wakeMask) != 0) {
*(cur->second) = true;
rv++;
queue.erase(cur);
}
}
if (queue.empty()) {
futexQueues.erase(futex);
}
}
futexLock.unlock();
FOLLY_TEST_DSCHED_VLOG(
"futexWake(" << futex << ", " << count << ", " << std::hex << wakeMask
<< ") -> " << rv);
DeterministicSchedule::afterSharedAccess();
return rv;
} }
} // namespace test } // namespace test
......
...@@ -22,6 +22,7 @@ ...@@ -22,6 +22,7 @@
#include <glog/logging.h> #include <glog/logging.h>
#include <atomic> #include <atomic>
#include <functional> #include <functional>
#include <list>
#include <mutex> #include <mutex>
#include <queue> #include <queue>
#include <thread> #include <thread>
...@@ -577,6 +578,126 @@ detail::FutexResult futexWaitImpl( ...@@ -577,6 +578,126 @@ detail::FutexResult futexWaitImpl(
std::chrono::steady_clock::time_point const* absSteadyTime, std::chrono::steady_clock::time_point const* absSteadyTime,
uint32_t waitMask); uint32_t waitMask);
/* Generic futex extensions to allow sharing between DeterministicAtomic and
* BufferedDeterministicAtomic.*/
template <template <typename> class Atom>
int deterministicFutexWakeImpl(
const detail::Futex<Atom>* futex,
std::mutex& futexLock,
std::unordered_map<
const detail::Futex<Atom>*,
std::list<std::pair<uint32_t, bool*>>>& futexQueues,
int count,
uint32_t wakeMask) {
using namespace test;
using namespace std::chrono;
int rv = 0;
DeterministicSchedule::beforeSharedAccess();
futexLock.lock();
if (futexQueues.count(futex) > 0) {
auto& queue = futexQueues[futex];
auto iter = queue.begin();
while (iter != queue.end() && rv < count) {
auto cur = iter++;
if ((cur->first & wakeMask) != 0) {
*(cur->second) = true;
rv++;
queue.erase(cur);
}
}
if (queue.empty()) {
futexQueues.erase(futex);
}
}
futexLock.unlock();
FOLLY_TEST_DSCHED_VLOG(
"futexWake(" << futex << ", " << count << ", " << std::hex << wakeMask
<< ") -> " << rv);
DeterministicSchedule::afterSharedAccess();
return rv;
}
template <template <typename> class Atom>
detail::FutexResult deterministicFutexWaitImpl(
const detail::Futex<Atom>* futex,
std::mutex& futexLock,
std::unordered_map<
const detail::Futex<Atom>*,
std::list<std::pair<uint32_t, bool*>>>& futexQueues,
uint32_t expected,
std::chrono::system_clock::time_point const* absSystemTimeout,
std::chrono::steady_clock::time_point const* absSteadyTimeout,
uint32_t waitMask) {
using namespace test;
using namespace std::chrono;
using namespace folly::detail;
bool hasTimeout = absSystemTimeout != nullptr || absSteadyTimeout != nullptr;
bool awoken = false;
FutexResult result = FutexResult::AWOKEN;
DeterministicSchedule::beforeSharedAccess();
FOLLY_TEST_DSCHED_VLOG(
"futexWait(" << futex << ", " << std::hex << expected << ", .., "
<< std::hex << waitMask << ") beginning..");
futexLock.lock();
// load_direct avoids deadlock on inner call to beforeSharedAccess
if (futex->load_direct() == expected) {
auto& queue = futexQueues[futex];
queue.emplace_back(waitMask, &awoken);
auto ours = queue.end();
ours--;
while (!awoken) {
futexLock.unlock();
DeterministicSchedule::afterSharedAccess();
DeterministicSchedule::beforeSharedAccess();
futexLock.lock();
// Simulate spurious wake-ups, timeouts each time with
// a 10% probability if we haven't been woken up already
if (!awoken && hasTimeout &&
DeterministicSchedule::getRandNumber(100) < 10) {
assert(futexQueues.count(futex) != 0 && &futexQueues[futex] == &queue);
queue.erase(ours);
if (queue.empty()) {
futexQueues.erase(futex);
}
// Simulate ETIMEDOUT 90% of the time and other failures
// remaining time
result = DeterministicSchedule::getRandNumber(100) >= 10
? FutexResult::TIMEDOUT
: FutexResult::INTERRUPTED;
break;
}
}
} else {
result = FutexResult::VALUE_CHANGED;
}
futexLock.unlock();
char const* resultStr = "?";
switch (result) {
case FutexResult::AWOKEN:
resultStr = "AWOKEN";
break;
case FutexResult::TIMEDOUT:
resultStr = "TIMEDOUT";
break;
case FutexResult::INTERRUPTED:
resultStr = "INTERRUPTED";
break;
case FutexResult::VALUE_CHANGED:
resultStr = "VALUE_CHANGED";
break;
}
FOLLY_TEST_DSCHED_VLOG(
"futexWait(" << futex << ", " << std::hex << expected << ", .., "
<< std::hex << waitMask << ") -> " << resultStr);
DeterministicSchedule::afterSharedAccess();
return result;
}
/** /**
* Implementations of the atomic_wait API for DeterministicAtomic, these are * Implementations of the atomic_wait API for DeterministicAtomic, these are
* no-ops here. Which for a correct implementation should not make a * no-ops here. Which for a correct implementation should not make a
......
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