folly::observer_detail::ObserverManager Class Reference

#include <ObserverManager.h>

Classes
class	CurrentQueue
class	DependencyRecorder
class	NextQueue
struct	Singleton

Public Member Functions
	~ObserverManager ()

Static Public Member Functions
static size_t	getVersion ()
static bool	inManagerThread ()
static void	scheduleRefresh (Core::Ptr core, size_t minVersion, bool force=false)
static void	scheduleRefreshNewVersion (Core::WeakPtr coreWeak)
static void	initCore (Core::Ptr core)

Private Types
using	CycleDetector = GraphCycleDetector< const Core * >

Private Member Functions
	ObserverManager ()
void	scheduleCurrent (Function< void()>)
void	scheduleNext (Core::WeakPtr)

Static Private Member Functions
static std::shared_ptr< ObserverManager >	getInstance ()

Private Attributes
std::unique_ptr< CurrentQueue >	currentQueue_
std::unique_ptr< NextQueue >	nextQueue_
SharedMutexReadPriority	versionMutex_
std::atomic< size_t >	version_ {1}
folly::Synchronized< CycleDetector, std::mutex >	cycleDetector_

Static Private Attributes
static FOLLY_TLS bool	inManagerThread_ {false}

Detailed Description

ObserverManager is a singleton which controls the re-computation of all Observers. Such re-computation always happens on the thread pool owned by ObserverManager.

ObserverManager has global current version. All existing Observers may have their version be less (yet to be updated) or equal (up to date) to the global current version.

ObserverManager::CurrentQueue contains all of the Observers which need to be updated to the global current version. Those updates are peformed on the ObserverManager's thread pool, until the queue is empty. If some Observer is updated, all of its dependents are added to ObserverManager::CurrentQueue to be updated.

If some leaf Observer (i.e. created from Observable) is updated, then current version of the ObserverManager should be bumped. All such updated leaf Observers are added to the ObserverManager::NextQueue.

Only when ObserverManager::CurrentQueue is empty, the global current version is bumped and all updates from the ObserverManager::NextQueue are performed. If leaf Observer gets updated more then once before being picked from the ObserverManager::NextQueue, then only the last update is processed.

Definition at line 52 of file ObserverManager.h.

Member Typedef Documentation

using folly::observer_detail::ObserverManager::CycleDetector = GraphCycleDetector<const Core*>

private

Definition at line 241 of file ObserverManager.h.

Constructor & Destructor Documentation

folly::observer_detail::ObserverManager::~ObserverManager

(

)

Definition at line 173 of file ObserverManager.cpp.

References currentQueue_, and nextQueue_.

                                  {
  // Destroy NextQueue, before the rest of this object, since it expects
  // ObserverManager to be alive.
  nextQueue_.reset();
  currentQueue_.reset();
}

folly::observer_detail::ObserverManager::ObserverManager ( )

private

Definition at line 168 of file ObserverManager.cpp.

References currentQueue_, and nextQueue_.

Referenced by folly::observer_detail::ObserverManager::Singleton::createManager().

                                 {
  currentQueue_ = std::make_unique<CurrentQueue>();
  nextQueue_ = std::make_unique<NextQueue>(*this);
}

Member Function Documentation

std::shared_ptr< ObserverManager > folly::observer_detail::ObserverManager::getInstance ( )

staticprivate

Definition at line 201 of file ObserverManager.cpp.

References folly::observer_detail::ObserverManager::Singleton::instance.

Referenced by getVersion(), initCore(), folly::observer_detail::ObserverManager::DependencyRecorder::markRefreshDependency(), scheduleRefresh(), scheduleRefreshNewVersion(), and folly::observer_detail::ObserverManager::DependencyRecorder::unmarkRefreshDependency().

                                                            {
  return Singleton::instance.try_get();
}

static size_t folly::observer_detail::ObserverManager::getVersion ( )

inlinestatic

Definition at line 54 of file ObserverManager.h.

References getInstance().

Referenced by folly::observer_detail::Core::getData().

                             {
    auto instance = getInstance();

    if (!instance) {
      return 1;
    }

    return instance->version_;
  }

static void folly::observer_detail::ObserverManager::initCore ( Core::Ptr  core )

inlinestatic

Definition at line 119 of file ObserverManager.h.

References folly::exchange(), getInstance(), inManagerThread(), inManagerThread_, and SCOPE_EXIT.

Referenced by folly::observer::makeObserver().

                                     {
    DCHECK(core->getVersion() == 0);

    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);
  }

static bool folly::observer_detail::ObserverManager::inManagerThread ( )

inlinestatic

Definition at line 64 of file ObserverManager.h.

References inManagerThread_.

Referenced by folly::observer_detail::ObserverManager::CurrentQueue::add(), folly::observer_detail::ObserverManager::DependencyRecorder::DependencyRecorder(), folly::observer_detail::Core::getData(), folly::observer::TLObserver< T >::getSnapshotRef(), initCore(), folly::observer_detail::ObserverManager::DependencyRecorder::markDependency(), and folly::observer_detail::Core::refresh().

                                {
    return inManagerThread_;
  }

void folly::observer_detail::ObserverManager::scheduleCurrent ( Function< void()>  task )

private

Definition at line 180 of file ObserverManager.cpp.

References currentQueue_, and folly::gen::move.

                                                           {
  currentQueue_->add(std::move(task));
}

void folly::observer_detail::ObserverManager::scheduleNext ( Core::WeakPtr  core )

private

Definition at line 184 of file ObserverManager.cpp.

References folly::gen::move, and nextQueue_.

                                                   {
  nextQueue_->add(std::move(core));
}

static void folly::observer_detail::ObserverManager::scheduleRefresh ( Core::Ptr  core, size_t  minVersion, bool  force = false  )

inlinestatic

Definition at line 69 of file ObserverManager.h.

References folly::annotate_rwlock_acquired(), folly::annotate_rwlock_released(), getInstance(), folly::gen::move, and folly::rdlock.

Referenced by folly::observer_detail::Core::refresh().

                                                                       {
    if (core->getVersion() >= minVersion) {
      return;
    }

    auto instance = getInstance();

    if (!instance) {
      return;
    }

    SharedMutexReadPriority::ReadHolder rh(instance->versionMutex_);

    // TSAN assumes that the thread that locks the mutex must
    // be the one that unlocks it. However, we are passing ownership of
    // the read lock into the lambda, and the thread that performs the async
    // work will be the one that unlocks it. To avoid noise with TSAN,
    // annotate that the thread has released the mutex, and then annotate
    // the async thread as acquiring the mutex.
    annotate_rwlock_released(
        &instance->versionMutex_,
        annotate_rwlock_level::rdlock,
        __FILE__,
        __LINE__);

    instance->scheduleCurrent([core = std::move(core),
                               instancePtr = instance.get(),
                               rh = std::move(rh),
                               force]() {
      // Make TSAN know that the current thread owns the read lock now.
      annotate_rwlock_acquired(
          &instancePtr->versionMutex_,
          annotate_rwlock_level::rdlock,
          __FILE__,
          __LINE__);

      core->refresh(instancePtr->version_, force);
    });
  }

static void folly::observer_detail::ObserverManager::scheduleRefreshNewVersion ( Core::WeakPtr  coreWeak )

inlinestatic

Definition at line 109 of file ObserverManager.h.

References getInstance(), and folly::gen::move.

Referenced by folly::observer::detail::ObserverCreatorContext< Observable, Traits >::update().

                                                              {
    auto instance = getInstance();

    if (!instance) {
      return;
    }

    instance->scheduleNext(std::move(coreWeak));
  }

Member Data Documentation

std::unique_ptr<CurrentQueue> folly::observer_detail::ObserverManager::currentQueue_

private

Definition at line 221 of file ObserverManager.h.

Referenced by ObserverManager(), scheduleCurrent(), and ~ObserverManager().

folly::Synchronized<CycleDetector, std::mutex> folly::observer_detail::ObserverManager::cycleDetector_

private

Definition at line 242 of file ObserverManager.h.

FOLLY_TLS bool folly::observer_detail::ObserverManager::inManagerThread_ {false}

staticprivate

Definition at line 227 of file ObserverManager.h.

Referenced by folly::observer_detail::ObserverManager::CurrentQueue::CurrentQueue(), initCore(), and inManagerThread().

std::unique_ptr<NextQueue> folly::observer_detail::ObserverManager::nextQueue_

private

Definition at line 224 of file ObserverManager.h.

Referenced by ObserverManager(), scheduleNext(), and ~ObserverManager().

std::atomic<size_t> folly::observer_detail::ObserverManager::version_ {1}

private

Definition at line 239 of file ObserverManager.h.

SharedMutexReadPriority folly::observer_detail::ObserverManager::versionMutex_

private

Version mutex is used to make sure all updates are processed from the CurrentQueue, before bumping the version and moving to the NextQueue.

To achieve this every task added to CurrentQueue holds a reader lock. NextQueue grabs a writer lock before bumping the version, so it can only happen if CurrentQueue is empty (notice that we use read-priority shared mutex).

Definition at line 238 of file ObserverManager.h.

---

The documentation for this class was generated from the following files:

• proxygen/folly/folly/experimental/observer/detail/ObserverManager.h
• proxygen/folly/folly/experimental/observer/detail/ObserverManager.cpp