Commit f43ce6d6 authored by Andrii Grynenko's avatar Andrii Grynenko Committed by Facebook Github Bot

Change makeObserver to compute the first value inline

Summary:
Current thread was previously blocked waiting on the Future to be completed anyway, so there's no reason to not treat it like one of the ObserverManager threads (if it isn't one already) and run the computation inline.
Without this change nested makeObserver calls could result in a deadlock if recursion depth was greater than the number of ObserverManager threads.

Reviewed By: yfeldblum

Differential Revision: D10097363

fbshipit-source-id: eb781ed5aea0ad9415da40fbc7f1c15ed481e92f
parent a7937a66
......@@ -64,33 +64,25 @@ class ObserverManager {
return inManagerThread_;
}
static Future<Unit>
static void
scheduleRefresh(Core::Ptr core, size_t minVersion, bool force = false) {
if (core->getVersion() >= minVersion) {
return makeFuture<Unit>(Unit());
return;
}
auto instance = getInstance();
if (!instance) {
return makeFuture<Unit>(
std::logic_error("ObserverManager requested during shutdown"));
return;
}
Promise<Unit> promise;
auto future = promise.getFuture();
SharedMutexReadPriority::ReadHolder rh(instance->versionMutex_);
instance->scheduleCurrent([core = std::move(core),
promise = std::move(promise),
instancePtr = instance.get(),
rh = std::move(rh),
force]() mutable {
promise.setWith([&]() { core->refresh(instancePtr->version_, force); });
});
return future;
instance->scheduleCurrent(
[core = std::move(core),
instancePtr = instance.get(),
rh = std::move(rh),
force]() { core->refresh(instancePtr->version_, force); });
}
static void scheduleRefreshNewVersion(Core::WeakPtr coreWeak) {
......@@ -105,7 +97,20 @@ class ObserverManager {
static void initCore(Core::Ptr core) {
DCHECK(core->getVersion() == 0);
scheduleRefresh(std::move(core), 1).get();
auto instance = getInstance();
if (!instance) {
throw std::logic_error("ObserverManager requested during shutdown");
}
auto inManagerThread = std::exchange(inManagerThread_, true);
SCOPE_EXIT {
inManagerThread_ = inManagerThread;
};
SharedMutexReadPriority::ReadHolder rh(instance->versionMutex_);
core->refresh(instance->version_, false);
}
class DependencyRecorder {
......
......@@ -350,3 +350,14 @@ TEST(Observer, SetCallback) {
EXPECT_EQ(43, callbackValue);
EXPECT_EQ(2, callbackCallsCount);
}
int makeObserverRecursion(int n) {
if (n == 0) {
return 0;
}
return **makeObserver([=] { return makeObserverRecursion(n - 1) + 1; });
}
TEST(Observer, NestedMakeObserver) {
EXPECT_EQ(32, makeObserverRecursion(32));
}
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