}
}
+void
+HAService::checkPermissionsClientAndListener() {
+ // Since we're used as CS callback we need to suppress
+ // any exceptions, unlikely though they may be.
+ try {
+ if (client_) {
+ client_->checkPermissions();
+ }
+
+ if (listener_) {
+ listener_->checkPermissions();
+ }
+ } catch (const isc::MultiThreadingInvalidOperation& ex) {
+ LOG_ERROR(ha_logger, HA_PAUSE_CLIENT_LISTENER_ILLEGAL)
+ .arg(ex.what());
+ // The exception needs to be propagated to the caller of the
+ // @ref MultiThreadingCriticalSection constructor.
+ throw;
+ } catch (const std::exception& ex) {
+ LOG_ERROR(ha_logger, HA_PAUSE_CLIENT_LISTENER_FAILED)
+ .arg(ex.what());
+ }
+}
+
void
HAService::startClientAndListener() {
// Add critical section callbacks.
MultiThreadingMgr::instance().addCriticalSectionCallbacks("HA_MT",
+ std::bind(&HAService::checkPermissionsClientAndListener, this),
std::bind(&HAService::pauseClientAndListener, this),
std::bind(&HAService::resumeClientAndListener, this));
// Since we're used as CS callback we need to suppress
// any exceptions, unlikely though they may be.
try {
- // The listener is the only one handling commands, so if any command
- // uses @ref MultiThreadingCriticalSection, it should throw first.
- // In this situation there is no need to resume the client's
- // @ref HttpThreadPool because it does not get paused in the first place.
- if (listener_) {
- listener_->pause();
- }
-
if (client_) {
client_->pause();
}
- } catch (const isc::MultiThreadingInvalidOperation& ex) {
- LOG_ERROR(ha_logger, HA_PAUSE_CLIENT_LISTENER_ILLEGAL)
- .arg(ex.what());
- // The exception needs to be propagated to the caller of the
- // @ref MultiThreadingCriticalSection constructor.
- throw;
+ if (listener_) {
+ listener_->pause();
+ }
} catch (const std::exception& ex) {
LOG_ERROR(ha_logger, HA_PAUSE_CLIENT_LISTENER_FAILED)
.arg(ex.what());
// Remove critical section callbacks.
MultiThreadingMgr::instance().removeCriticalSectionCallbacks("HA_MT");
- if (listener_) {
- listener_->stop();
- }
-
if (client_) {
client_->stop();
}
+
+ if (listener_) {
+ listener_->stop();
+ }
}
// Explicit instantiations.
/// @return Pointer to the response to the ha-maintenance-cancel.
data::ConstElementPtr processMaintenanceCancel();
+ /// @brief Check client and(or) listener current thread permissions to
+ /// perform thread pool state transition.
+ ///
+ /// @throw MultiThreadingInvalidOperation if the state transition is done on
+ /// any of the owned threads
+ void checkPermissionsClientAndListener();
+
/// @brief Start the client and(or) listener instances.
///
/// When HA+MT is enabled it starts the client's thread pool
}
}
+void
+CmdHttpListener::checkPermissions() {
+ if (thread_pool_) {
+ thread_pool_->checkPausePermissions();
+ }
+}
+
void
CmdHttpListener::pause() {
if (thread_pool_) {
/// @brief Destructor
virtual ~CmdHttpListener();
+ /// @brief Check if the current thread can perform thread pool state
+ /// transition.
+ ///
+ /// @throw MultiThreadingInvalidOperation if the state transition is done on
+ /// any of the owned threads
+ void checkPermissions();
+
/// @brief Starts running the listener's thread pool.
void start();
stop();
}
+ /// @brief Check if the current thread can perform thread pool state
+ /// transition.
+ ///
+ /// @throw MultiThreadingInvalidOperation if the state transition is done on
+ /// any of the owned threads
+ void checkPermissions() {
+ if (thread_pool_) {
+ thread_pool_->checkPausePermissions();
+ }
+ }
+
/// @brief Starts running the client's thread pool, if multi-threaded.
void start() {
if (thread_pool_) {
impl_->start();
}
+void
+HttpClient::checkPermissions() {
+ impl_->checkPermissions();
+}
+
void
HttpClient::pause() {
impl_->pause();
const CloseHandler& close_callback =
CloseHandler());
+ /// @brief Check if the current thread can perform thread pool state
+ /// transition.
+ ///
+ /// @throw MultiThreadingInvalidOperation if the state transition is done on
+ /// any of the owned threads
+ void checkPermissions();
+
/// @brief Starts running the client's thread pool, if multi-threaded.
void start();
return (std::string("unknown-state"));
}
+void
+HttpThreadPool::checkPausePermissions() {
+ checkPermissions(State::PAUSED);
+}
+
+void
+HttpThreadPool::checkPermissions(State new_state) {
+ auto id = std::this_thread::get_id();
+ if (checkThreadId(id)) {
+ isc_throw(MultiThreadingInvalidOperation, "invalid thread pool state change to "
+ << HttpThreadPool::stateToText(new_state) << " performed by owned thread");
+ }
+}
+
bool
HttpThreadPool::checkThreadId(std::thread::id id) {
for (auto thread : threads_) {
void
HttpThreadPool::setState(State new_state) {
- auto id = std::this_thread::get_id();
- if (checkThreadId(id)) {
- isc_throw(MultiThreadingInvalidOperation, "invalid thread pool state change to "
- << HttpThreadPool::stateToText(new_state) << " performed by owned thread");
- }
+ checkPermissions(new_state);
std::unique_lock<std::mutex> main_lck(mutex_);
return (getState() == State::STOPPED);
}
+ /// @brief Check current thread permissions to transition to the new PAUSED
+ /// state.
+ /// @throw MultiThreadingInvalidOperation if the state transition is done on
+ /// any of the owned threads
+ void checkPausePermissions();
+
private:
+ /// @brief Check current thread permissions to transition to the new state.
+ ///
+ /// @param state The new transition state for the pool.
+ /// @throw MultiThreadingInvalidOperation if the state transition is done on
+ /// any of the owned threads
+ void checkPermissions(State state);
+
/// @brief Check specified thread id against own threads.
///
/// @return true if thread is owned, false otherwise.
/// -# Waits until all threads have paused.
/// -# Returns to caller.
///
- /// @param state new state for the pool.
+ /// @param state The new transition state for the pool.
/// @throw MultiThreadingInvalidOperation if the state transition is done on
/// any of the owned threads.
void setState(State state);
void
MultiThreadingMgr::enterCriticalSection() {
+ checkCallbacksPermissions();
std::lock_guard<std::mutex> lk(mutex_);
stopProcessing();
++critical_section_count_;
void
MultiThreadingMgr::exitCriticalSection() {
+ checkCallbacksPermissions();
std::lock_guard<std::mutex> lk(mutex_);
if (critical_section_count_ == 0) {
isc_throw(InvalidOperation, "invalid negative value for override");
bool
MultiThreadingMgr::isInCriticalSection() {
+ checkCallbacksPermissions();
std::lock_guard<std::mutex> lk(mutex_);
return (isInCriticalSectionInternal());
}
}
void
-MultiThreadingMgr::stopProcessing() {
- if (getMode() && !isInCriticalSectionInternal()) {
- // First call the registered callback for entering the critical section
- // so that if any exception is thrown, there is no need to stop and then
- // start the service threads.
+MultiThreadingMgr::checkCallbacksPermissions() {
+ if (getMode()) {
for (const auto& cb : cs_callbacks_.getCallbackPairs()) {
try {
- (cb.entry_cb_)();
+ (cb.check_cb_)();
} catch (const isc::MultiThreadingInvalidOperation& ex) {
// If any registered callback throws, the exception needs to be
// propagated to the caller of the
// Because this function is called by the
// @ref MultiThreadingCriticalSection constructor, throwing here
// is safe.
- // @note should we call the exit_cb_ here for all successful
- // entry_cb_ already run?
throw;
} catch (...) {
// We can't log it and throwing could be chaos.
// must be exception-proof
}
}
-
- if (getThreadPoolSize()) {
- thread_pool_.stop();
- }
}
}
void
-MultiThreadingMgr::startProcessing() {
- if (getMode() && !isInCriticalSectionInternal()) {
- if (getThreadPoolSize()) {
- thread_pool_.start(getThreadPoolSize());
+MultiThreadingMgr::callEntryCallbacks() {
+ if (getMode()) {
+ const auto& callbacks = cs_callbacks_.getCallbackPairs();
+ for (auto cb_it = callbacks.begin(); cb_it != callbacks.end(); cb_it++) {
+ try {
+ (cb_it->entry_cb_)();
+ } catch (...) {
+ // We can't log it and throwing could be chaos.
+ // We'll swallow it and tell people their callbacks
+ // must be exception-proof
+ }
}
+ }
+}
- for (const auto& cb : cs_callbacks_.getCallbackPairs()) {
+void
+MultiThreadingMgr::callExitCallbacks() {
+ if (getMode()) {
+ const auto& callbacks = cs_callbacks_.getCallbackPairs();
+ for (auto cb_it = callbacks.rbegin(); cb_it != callbacks.rend(); cb_it++) {
try {
- (cb.exit_cb_)();
+ (cb_it->exit_cb_)();
} catch (...) {
// We can't log it and throwing could be chaos.
// We'll swallow it and tell people their callbacks
// Because this function is called by the
// @ref MultiThreadingCriticalSection destructor, throwing here
// is not safe and will cause the process to crash.
- // @note should we call the exit_cb_ here in reverse order of
- // the entry_cb_ calls?
}
}
}
}
+void
+MultiThreadingMgr::stopProcessing() {
+ if (getMode() && !isInCriticalSectionInternal()) {
+ if (getThreadPoolSize()) {
+ thread_pool_.stop();
+ }
+ callEntryCallbacks();
+ }
+}
+
+void
+MultiThreadingMgr::startProcessing() {
+ if (getMode() && !isInCriticalSectionInternal()) {
+ if (getThreadPoolSize()) {
+ thread_pool_.start(getThreadPoolSize());
+ }
+ callExitCallbacks();
+ }
+}
+
void
MultiThreadingMgr::addCriticalSectionCallbacks(const std::string& name,
+ const CSCallbackPair::Callback& check_cb,
const CSCallbackPair::Callback& entry_cb,
const CSCallbackPair::Callback& exit_cb) {
- cs_callbacks_.addCallbackPair(name, entry_cb, exit_cb);
+ cs_callbacks_.addCallbackPair(name, check_cb, entry_cb, exit_cb);
}
void
void
CSCallbackPairList::addCallbackPair(const std::string& name,
+ const CSCallbackPair::Callback& check_cb,
const CSCallbackPair::Callback& entry_cb,
const CSCallbackPair::Callback& exit_cb) {
if (name.empty()) {
isc_throw(BadValue, "CSCallbackPairList - name cannot be empty");
}
+ if (!check_cb) {
+ isc_throw(BadValue, "CSCallbackPairList - check callback for " << name
+ << " cannot be empty");
+ }
+
if (!entry_cb) {
isc_throw(BadValue, "CSCallbackPairList - entry callback for " << name
<< " cannot be empty");
}
}
- cb_pairs_.push_back(CSCallbackPair(name, entry_cb, exit_cb));
+ cb_pairs_.push_back(CSCallbackPair(name, check_cb, entry_cb, exit_cb));
}
void
/// @brief Embodies a named pair of CriticalSection callbacks.
///
-/// This class associates a pair of callbacks, one to be invoked
-/// upon CriticalSection entry and the other invoked upon
-/// CriticalSection exit, with name. The name allows the pair
-/// to be uniquely identified such that they can be added and
-/// removed as needed.
+/// This class associates with a name, a pair of callbacks, one to be invoked
+/// before CriticalSection entry and exit callbacks to validate current thread
+/// permissions to perform such actions, one to be invoked upon CriticalSection
+/// entry and one to be invoked upon CriticalSection exit,
+/// The name allows the pair to be uniquely identified such that they can be
+/// added and removed as needed.
struct CSCallbackPair {
/// @brief Defines a callback as a simple void() functor.
typedef std::function<void()> Callback;
/// @brief Constructor
///
/// @param name Name by which the callbacks can be found.
+ /// @param entry_cb Callback to check current thread permissions to call
+ /// the CriticalSection entry and exit callbacks.
/// @param entry_cb Callback to invoke upon CriticalSection entry.
/// @param entry_cb Callback to invoke upon CriticalSection exit.
- CSCallbackPair(const std::string& name, const Callback& entry_cb,
- const Callback& exit_cb)
- : name_(name), entry_cb_(entry_cb), exit_cb_(exit_cb) {}
+ CSCallbackPair(const std::string& name, const Callback& check_cb,
+ const Callback& entry_cb, const Callback& exit_cb)
+ : name_(name), check_cb_(check_cb), entry_cb_(entry_cb),
+ exit_cb_(exit_cb) {}
/// @brief Name by which the callback can be found.
std::string name_;
+ /// @brief Check permissions callback associated with name.
+ Callback check_cb_;
+
/// @brief Entry point callback associated with name.
Callback entry_cb_;
/// @brief Adds a callback pair to the list.
///
/// @param name Name of the callback to add.
- /// @param entry_cb Callback to add.
- /// @param exit_cb Callback to add.
+ /// @param check_cb The check permissions callback to add.
+ /// @param entry_cb The CriticalSection entry callback to add.
+ /// @param exit_cb The CriticalSection exit callback to add.
///
/// @throw BadValue if the name is already in the list,
/// the name is blank, or either callback is empty.
void addCallbackPair(const std::string& name,
+ const CSCallbackPair::Callback& check_cb,
const CSCallbackPair::Callback& entry_cb,
const CSCallbackPair::Callback& exit_cb);
std::list<CSCallbackPair> cb_pairs_;
};
+class MultiThreadingCriticalSection;
+
/// @brief Multi Threading Manager.
///
/// This singleton class holds the multi-threading mode.
/// @endcode
class MultiThreadingMgr : public boost::noncopyable {
public:
-
/// @brief Returns a single instance of Multi Threading Manager.
///
/// MultiThreadingMgr is a singleton and this method is the only way
///
/// @param name Name of the set of callbacks. This value is used by the
/// callback owner to add and remove them. Duplicates are not allowed.
+ /// @param entry_cb Callback to check current thread permissions to call
+ /// the CriticalSection entry and exit callbacks.
/// @param entry_cb Callback to invoke upon CriticalSection entry. Cannot be
/// empty.
/// @param exit_cb Callback to invoke upon CriticalSection exit. Cannot be
/// empty.
void addCriticalSectionCallbacks(const std::string& name,
+ const CSCallbackPair::Callback& check_cb,
const CSCallbackPair::Callback& entry_cb,
const CSCallbackPair::Callback& exit_cb);
/// @return The critical section flag.
bool isInCriticalSectionInternal();
+ /// @brief Class method tests if current thread is allowed to enter the
+ /// @ref MultiThreadingCriticalSection and to invoke the stop and start
+ /// callbacks.
+ ///
+ /// Has no effect in single-threaded mode.
+ ///
+ /// @note This function swallows exceptions thrown by validate
+ /// callbacks without logging to avoid breaking the CS
+ /// chain.
+ /// @throw MultiThreadingInvalidOperation if current thread has no
+ /// permission to enter CriticalSection.
+ void checkCallbacksPermissions();
+
+ /// @brief Class method which invokes CriticalSection entry callbacks.
+ ///
+ /// Has no effect in single-threaded mode.
+ ///
+ /// @note This function swallows exceptions thrown by validate
+ /// callbacks without logging to avoid breaking the CS
+ /// chain.
+ void callEntryCallbacks();
+
+ /// @brief Class method which invokes CriticalSection entry callbacks.
+ ///
+ /// Has no effect in single-threaded mode.
+ ///
+ /// @note This function swallows exceptions thrown by validate
+ /// callbacks without logging to avoid breaking the CS
+ /// chain.
+ void callExitCallbacks();
+
/// @brief Class method stops non-critical processing.
///
/// Stops the DHCP thread pool if it's running and invokes
invocations_.push_back(4);
}
- /// @brief A callback that adds the value 5 to invocations lists and throws
- /// isc::Exception which is ignored.
+ /// @brief A callback that throws @ref isc::Exception which is ignored.
void ignoredException() {
- invocations_.push_back(5);
isc_throw(isc::Exception, "ignored");
}
- /// @brief A callback that adds the value 6 to invocations lists and throws
- /// isc::MultiThreadingInvalidOperation which is propagated to the scope of
- /// the @ref MultiThreadingCriticalSection constructor.
+ /// @brief A callback that throws @ref isc::MultiThreadingInvalidOperation
+ /// which is propagated to the scope of the
+ /// @ref MultiThreadingCriticalSection constructor.
void observedException() {
- invocations_.push_back(6);
isc_throw(isc::MultiThreadingInvalidOperation, "observed");
}
if (entries.size()) {
// We expect entry invocations.
ASSERT_EQ(invocations_.size(), entries.size());
- ASSERT_TRUE(invocations_ == entries);
+ ASSERT_EQ(invocations_, entries);
} else {
// We do not expect entry invocations.
ASSERT_FALSE(invocations_.size());
if (entries.size()) {
// We expect entry invocations.
ASSERT_EQ(invocations_.size(), entries.size());
- ASSERT_TRUE(invocations_ == entries);
+ ASSERT_EQ(invocations_, entries);
} else {
// We do not expect entry invocations.
ASSERT_FALSE(invocations_.size());
if (exits.size()) {
// We expect exit invocations.
- ASSERT_TRUE(invocations_ == exits);
+ ASSERT_EQ(invocations_, exits);
} else {
// We do not expect exit invocations.
ASSERT_FALSE(invocations_.size());
auto& mgr = MultiThreadingMgr::instance();
// Cannot add with a blank name.
- ASSERT_THROW_MSG(mgr.addCriticalSectionCallbacks("", [](){}, [](){}),
+ ASSERT_THROW_MSG(mgr.addCriticalSectionCallbacks("", [](){}, [](){}, [](){}),
BadValue, "CSCallbackPairList - name cannot be empty");
- // Cannot add with an empty entry callback.
- ASSERT_THROW_MSG(mgr.addCriticalSectionCallbacks("bad", nullptr, [](){}),
+ // Cannot add with an empty check callback.
+ ASSERT_THROW_MSG(mgr.addCriticalSectionCallbacks("bad", nullptr, [](){}, [](){}),
+ BadValue, "CSCallbackPairList - check callback for bad cannot be empty");
+
+ // Cannot add with an empty exit callback.
+ ASSERT_THROW_MSG(mgr.addCriticalSectionCallbacks("bad", [](){}, nullptr, [](){}),
BadValue, "CSCallbackPairList - entry callback for bad cannot be empty");
// Cannot add with an empty exit callback.
- ASSERT_THROW_MSG(mgr.addCriticalSectionCallbacks("bad", [](){}, nullptr),
+ ASSERT_THROW_MSG(mgr.addCriticalSectionCallbacks("bad", [](){}, [](){}, nullptr),
BadValue, "CSCallbackPairList - exit callback for bad cannot be empty");
// Should be able to add foo.
- ASSERT_NO_THROW_LOG(mgr.addCriticalSectionCallbacks("foo", [](){}, [](){}));
+ ASSERT_NO_THROW_LOG(mgr.addCriticalSectionCallbacks("foo", [](){}, [](){}, [](){}));
// Should not be able to add foo twice.
- ASSERT_THROW_MSG(mgr.addCriticalSectionCallbacks("foo", [](){}, [](){}),
+ ASSERT_THROW_MSG(mgr.addCriticalSectionCallbacks("foo", [](){}, [](){}, [](){}),
BadValue, "CSCallbackPairList - callbacks for foo already exist");
// Should be able to add bar.
- ASSERT_NO_THROW_LOG(mgr.addCriticalSectionCallbacks("bar", [](){}, [](){}));
+ ASSERT_NO_THROW_LOG(mgr.addCriticalSectionCallbacks("bar", [](){}, [](){}, [](){}));
// Should be able to remove foo.
ASSERT_NO_THROW_LOG(mgr.removeCriticalSectionCallbacks("foo"));
// Add two sets of CriticalSection call backs.
MultiThreadingMgr::instance().addCriticalSectionCallbacks("oneAndTwo",
+ std::bind(&CriticalSectionCallbackTest::ignoredException, this),
std::bind(&CriticalSectionCallbackTest::one, this),
std::bind(&CriticalSectionCallbackTest::two, this));
MultiThreadingMgr::instance().addCriticalSectionCallbacks("threeAndFour",
+ std::bind(&CriticalSectionCallbackTest::ignoredException, this),
std::bind(&CriticalSectionCallbackTest::three, this),
std::bind(&CriticalSectionCallbackTest::four, this));
// callbacks execute at the appropriate times and we can do
// so repeatedly.
for (int i = 0; i < 3; ++i) {
- runCriticalSections({1,3}, {2,4});
+ runCriticalSections({1 ,3}, {2, 4});
}
// Now remove the first set of callbacks.
// Add two sets of CriticalSection call backs.
MultiThreadingMgr::instance().addCriticalSectionCallbacks("observed",
std::bind(&CriticalSectionCallbackTest::observedException, this),
- std::bind(&CriticalSectionCallbackTest::observedException, this));
+ std::bind(&CriticalSectionCallbackTest::one, this),
+ std::bind(&CriticalSectionCallbackTest::two, this));
MultiThreadingMgr::instance().addCriticalSectionCallbacks("ignored",
std::bind(&CriticalSectionCallbackTest::ignoredException, this),
- std::bind(&CriticalSectionCallbackTest::ignoredException, this));
+ std::bind(&CriticalSectionCallbackTest::three, this),
+ std::bind(&CriticalSectionCallbackTest::four, this));
// Apply multi-threading configuration with 16 threads and queue size 256.
MultiThreadingMgr::instance().apply(true, 16, 256);
// callbacks execute at the appropriate times and we can do
// so repeatedly.
for (int i = 0; i < 3; ++i) {
- runCriticalSections({6}, {}, true);
+ runCriticalSections({}, {}, true);
}
// Now remove the first set of callbacks.
MultiThreadingMgr::instance().removeCriticalSectionCallbacks("observed");
// Retest CriticalSections.
- runCriticalSections({5}, {5});
+ runCriticalSections({3}, {4});
// Now remove the remaining callbacks.
MultiThreadingMgr::instance().removeAllCriticalSectionCallbacks();
projects / thirdparty / kea.git / commitdiff
