szaszm commented on code in PR #1581:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1581#discussion_r1211136629
##########
libminifi/test/unit/ThreadPoolTests.cpp:
##########
Review Comment:
What is the function called `function` used for at line 28? Can we remove it
if it's unused?
##########
libminifi/src/core/Processor.cpp:
##########
@@ -372,26 +372,26 @@ void Processor::setMaxConcurrentTasks(const uint8_t
tasks) {
}
void Processor::yield() {
- yield_expiration_ = std::chrono::system_clock::now() +
yield_period_msec_.load();
+ yield_expiration_ = std::chrono::steady_clock::now() +
yield_period_msec_.load();
}
void Processor::yield(std::chrono::milliseconds delta_time) {
- yield_expiration_ = std::chrono::system_clock::now() + delta_time;
+ yield_expiration_ = std::chrono::steady_clock::now() + delta_time;
}
bool Processor::isYield() {
- return yield_expiration_.load() >= std::chrono::system_clock::now();
+ return getYieldTime() > 0ms;
}
void Processor::clearYield() {
- yield_expiration_ = std::chrono::system_clock::time_point();
+ yield_expiration_ = std::chrono::steady_clock::time_point();
}
std::chrono::milliseconds Processor::getYieldTime() const {
auto yield_expiration = yield_expiration_.load();
- auto current_time = std::chrono::system_clock::now();
+ auto current_time = std::chrono::steady_clock::now();
Review Comment:
Now that timestamps are changed from system_clock to steady_clock, does this
affect serialized timestamps? I don't think `steady_clock` is guaranteed to be
steadily increasing in sync with real time between executions or reboots, so
reading persisted timestamps anywhere may result in incorrect behavior.
##########
libminifi/src/utils/ThreadPool.cpp:
##########
@@ -151,10 +153,10 @@ void ThreadPool<T>::manageWorkers() {
if (nullptr != thread_manager_) {
while (running_) {
- auto waitperiod = std::chrono::milliseconds(500);
+ auto wait_period = 500ms;
{
- std::unique_lock<std::recursive_mutex> lock(manager_mutex_,
std::try_to_lock);
- if (!lock.owns_lock()) {
+ std::unique_lock<std::recursive_mutex> manager_lock(manager_mutex_,
std::try_to_lock);
+ if (!manager_lock.owns_lock()) {
Review Comment:
It looks like this mutex lock should be retried after waiting. Currently it
just proceeds to access the shared data without synchronization, after waiting
a bit, if the first lock attempt fails.
##########
libminifi/include/utils/ThreadPool.h:
##########
@@ -94,7 +94,7 @@ class Worker {
promise->set_value(result);
return false;
}
- next_exec_time_ = std::max(next_exec_time_ +
run_determinant_->wait_time(), std::chrono::steady_clock::now());
+ next_exec_time_ = std::max(next_exec_time_,
std::chrono::steady_clock::now() + run_determinant_->wait_time());
Review Comment:
This seems to change behavior, especially for long running processors: If a
processor is scheduled to run once every 5 seconds, and it takes 4 seconds to
run, then I'd expect the next execution to happen 1 second after finishing
execution, not 5 seconds later. The latter would mean effectively running every
9 seconds instead of the configured 5.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
