foreign_storage::ForeignTableRefreshScheduler Class Reference

#include <ForeignTableRefreshScheduler.h>

Static Public Member Functions
static void	start (std::atomic< bool > &is_program_running)
static void	stop ()
static void	setWaitDuration (int64_t duration_in_seconds)
static bool	isRunning ()
static bool	hasRefreshedTable ()
static void	resetHasRefreshedTable ()

Static Private Member Functions
static void	invalidateQueryEngineCaches ()

Static Private Attributes
static std::atomic< bool >	is_scheduler_running_ {false}
static std::chrono::seconds	thread_wait_duration_ {60}
static std::thread	scheduler_thread_
static std::atomic< bool >	has_refreshed_table_ {false}
static std::mutex	wait_mutex_
static std::condition_variable	wait_condition_

Detailed Description

Definition at line 26 of file ForeignTableRefreshScheduler.h.

Member Function Documentation

bool foreign_storage::ForeignTableRefreshScheduler::hasRefreshedTable ( )

static

Definition at line 101 of file ForeignTableRefreshScheduler.cpp.

References has_refreshed_table_.

                                                     {
  return has_refreshed_table_;
}

void foreign_storage::ForeignTableRefreshScheduler::invalidateQueryEngineCaches ( )

staticprivate

Definition at line 25 of file ForeignTableRefreshScheduler.cpp.

References legacylockmgr::ExecutorOuterLock, legacylockmgr::LockMgr< MutexType, KeyType >::getMutex(), and CacheInvalidator< CACHE_HOLDING_TYPES >::invalidateCaches().

Referenced by start().

                                                               {
  auto execute_write_lock =
      heavyai::unique_lock<legacylockmgr::WrapperType<heavyai::shared_mutex>>(
          *legacylockmgr::LockMgr<heavyai::shared_mutex, bool>::getMutex(
              legacylockmgr::ExecutorOuterLock, true));
  // todo (yoonmin): support per-table invalidation
  UpdateTriggeredCacheInvalidator::invalidateCaches();
  ResultSetCacheInvalidator::invalidateCaches();
}

Here is the call graph for this function:

Here is the caller graph for this function:

bool foreign_storage::ForeignTableRefreshScheduler::isRunning ( )

static

Definition at line 97 of file ForeignTableRefreshScheduler.cpp.

References is_scheduler_running_.

                                             {
  return is_scheduler_running_;
}

void foreign_storage::ForeignTableRefreshScheduler::resetHasRefreshedTable ( )

static

Definition at line 105 of file ForeignTableRefreshScheduler.cpp.

References has_refreshed_table_.

                                                          {
  has_refreshed_table_ = false;
}

void foreign_storage::ForeignTableRefreshScheduler::setWaitDuration ( int64_t  duration_in_seconds )

static

Definition at line 93 of file ForeignTableRefreshScheduler.cpp.

References thread_wait_duration_.

                                                                              {
  thread_wait_duration_ = std::chrono::seconds{duration_in_seconds};
}

void foreign_storage::ForeignTableRefreshScheduler::start ( std::atomic< bool > &  is_program_running )

static

Definition at line 35 of file ForeignTableRefreshScheduler.cpp.

References logger::ERROR, has_refreshed_table_, Catalog_Namespace::SysCatalog::instance(), invalidateQueryEngineCaches(), is_scheduler_running_, LOG, foreign_storage::refresh_foreign_table(), scheduler_thread_, run_benchmark_import::tables, thread_wait_duration_, wait_condition_, and wait_mutex_.

Referenced by startHeavyDBServer().

                                                                            {
  if (is_program_running && !is_scheduler_running_) {
    is_scheduler_running_ = true;
    scheduler_thread_ = std::thread([&is_program_running]() {
      while (is_program_running && is_scheduler_running_) {
        auto& sys_catalog = Catalog_Namespace::SysCatalog::instance();
        // Exit if scheduler has been stopped asynchronously
        if (!is_program_running || !is_scheduler_running_) {
          return;
        }
        bool at_least_one_table_refreshed = false;
        for (const auto& catalog : sys_catalog.getCatalogsForAllDbs()) {
          // Exit if scheduler has been stopped asynchronously
          if (!is_program_running || !is_scheduler_running_) {
            return;
          }
          auto tables = catalog->getAllForeignTablesForRefresh();
          for (auto table : tables) {
            // Exit if scheduler has been stopped asynchronously
            if (!is_program_running || !is_scheduler_running_) {
              return;
            }
            try {
              refresh_foreign_table(*catalog, table->tableName, false);
            } catch (std::runtime_error& e) {
              LOG(ERROR) << "Scheduled refresh for table \"" << table->tableName
                         << "\" resulted in an error. " << e.what();
            }
            has_refreshed_table_ = true;
            at_least_one_table_refreshed = true;
          }
        }
        if (at_least_one_table_refreshed) {
          invalidateQueryEngineCaches();
        }
        // Exit if scheduler has been stopped asynchronously
        if (!is_program_running || !is_scheduler_running_) {
          return;
        }

        // A condition variable is used here (instead of a sleep call)
        // in order to allow for thread wake-up, even in the middle
        // of a wait interval.
        std::unique_lock<std::mutex> wait_lock(wait_mutex_);
        wait_condition_.wait_for(wait_lock, thread_wait_duration_);
      }
    });
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::ForeignTableRefreshScheduler::stop ( )

static

Definition at line 85 of file ForeignTableRefreshScheduler.cpp.

References is_scheduler_running_, scheduler_thread_, and wait_condition_.

Referenced by startHeavyDBServer().

                                        {
  if (is_scheduler_running_) {
    is_scheduler_running_ = false;
    wait_condition_.notify_one();
    scheduler_thread_.join();
  }
}

Here is the caller graph for this function:

Member Data Documentation

std::atomic< bool > foreign_storage::ForeignTableRefreshScheduler::has_refreshed_table_ {false}

staticprivate

Definition at line 42 of file ForeignTableRefreshScheduler.h.

Referenced by hasRefreshedTable(), resetHasRefreshedTable(), and start().

std::atomic< bool > foreign_storage::ForeignTableRefreshScheduler::is_scheduler_running_ {false}

staticprivate

Definition at line 39 of file ForeignTableRefreshScheduler.h.

Referenced by isRunning(), start(), and stop().

std::thread foreign_storage::ForeignTableRefreshScheduler::scheduler_thread_

staticprivate

Definition at line 41 of file ForeignTableRefreshScheduler.h.

Referenced by start(), and stop().

std::chrono::seconds foreign_storage::ForeignTableRefreshScheduler::thread_wait_duration_ {60}

staticprivate

Definition at line 40 of file ForeignTableRefreshScheduler.h.

Referenced by setWaitDuration(), and start().

std::condition_variable foreign_storage::ForeignTableRefreshScheduler::wait_condition_

staticprivate

Definition at line 44 of file ForeignTableRefreshScheduler.h.

Referenced by start(), and stop().

std::mutex foreign_storage::ForeignTableRefreshScheduler::wait_mutex_

staticprivate

Definition at line 43 of file ForeignTableRefreshScheduler.h.

Referenced by start().

---

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

• /home/jenkins-slave/workspace/core-os-doxygen/ThriftHandler/ForeignTableRefreshScheduler.h
• /home/jenkins-slave/workspace/core-os-doxygen/ThriftHandler/ForeignTableRefreshScheduler.cpp