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

waitForAllUpdates()

Summary: This is useful for unit tests which need to wait for observer updates.

Reviewed By: yfeldblum

Differential Revision: D13133404

fbshipit-source-id: ff36f55ec537490a763865cea2125617b58dc9c0
parent 23020412
...@@ -15,6 +15,8 @@ ...@@ -15,6 +15,8 @@
*/ */
#include <folly/experimental/observer/detail/ObserverManager.h> #include <folly/experimental/observer/detail/ObserverManager.h>
#include <future>
#include <folly/ExceptionString.h> #include <folly/ExceptionString.h>
#include <folly/Format.h> #include <folly/Format.h>
#include <folly/MPMCQueue.h> #include <folly/MPMCQueue.h>
...@@ -143,6 +145,18 @@ class ObserverManager::NextQueue { ...@@ -143,6 +145,18 @@ class ObserverManager::NextQueue {
for (auto& core : cores) { for (auto& core : cores) {
manager_.scheduleRefresh(std::move(core), manager_.version_, true); manager_.scheduleRefresh(std::move(core), manager_.version_, true);
} }
{
auto wEmptyWaiters = emptyWaiters_.wlock();
// We don't want any new waiters to be added while we are checking the
// queue.
if (queue_.isEmpty()) {
for (auto& promise : *wEmptyWaiters) {
promise.set_value();
}
wEmptyWaiters->clear();
}
}
} }
}); });
} }
...@@ -158,11 +172,23 @@ class ObserverManager::NextQueue { ...@@ -158,11 +172,23 @@ class ObserverManager::NextQueue {
thread_.join(); thread_.join();
} }
void waitForEmpty() {
std::promise<void> promise;
auto future = promise.get_future();
emptyWaiters_.wlock()->push_back(std::move(promise));
// Write to the queue to notify the thread.
queue_.blockingWrite(Core::WeakPtr());
future.get();
}
private: private:
ObserverManager& manager_; ObserverManager& manager_;
MPMCQueue<Core::WeakPtr> queue_; MPMCQueue<Core::WeakPtr> queue_;
std::thread thread_; std::thread thread_;
std::atomic<bool> stop_{false}; std::atomic<bool> stop_{false};
folly::Synchronized<std::vector<std::promise<void>>> emptyWaiters_;
}; };
ObserverManager::ObserverManager() { ObserverManager::ObserverManager() {
...@@ -185,6 +211,18 @@ void ObserverManager::scheduleNext(Core::WeakPtr core) { ...@@ -185,6 +211,18 @@ void ObserverManager::scheduleNext(Core::WeakPtr core) {
nextQueue_->add(std::move(core)); nextQueue_->add(std::move(core));
} }
void ObserverManager::waitForAllUpdates() {
auto instance = getInstance();
if (!instance) {
return;
}
instance->nextQueue_->waitForEmpty();
// Wait for all readers to release the lock.
SharedMutexReadPriority::WriteHolder wh(instance->versionMutex_);
}
struct ObserverManager::Singleton { struct ObserverManager::Singleton {
static folly::Singleton<ObserverManager> instance; static folly::Singleton<ObserverManager> instance;
// MSVC 2015 doesn't let us access ObserverManager's constructor if we // MSVC 2015 doesn't let us access ObserverManager's constructor if we
......
...@@ -134,6 +134,8 @@ class ObserverManager { ...@@ -134,6 +134,8 @@ class ObserverManager {
core->refresh(instance->version_, false); core->refresh(instance->version_, false);
} }
static void waitForAllUpdates();
class DependencyRecorder { class DependencyRecorder {
public: public:
using DependencySet = std::unordered_set<Core::Ptr>; using DependencySet = std::unordered_set<Core::Ptr>;
......
...@@ -361,3 +361,20 @@ int makeObserverRecursion(int n) { ...@@ -361,3 +361,20 @@ int makeObserverRecursion(int n) {
TEST(Observer, NestedMakeObserver) { TEST(Observer, NestedMakeObserver) {
EXPECT_EQ(32, makeObserverRecursion(32)); EXPECT_EQ(32, makeObserverRecursion(32));
} }
TEST(Observer, WaitForAllUpdates) {
folly::observer::SimpleObservable<int> observable{42};
auto observer = makeObserver([o = observable.getObserver()] {
std::this_thread::sleep_for(std::chrono::milliseconds{100});
return **o;
});
EXPECT_EQ(42, **observer);
observable.setValue(43);
folly::observer_detail::ObserverManager::waitForAllUpdates();
EXPECT_EQ(43, **observer);
}
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