You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "fgerlits (via GitHub)" <gi...@apache.org> on 2023/08/08 15:53:54 UTC
[GitHub] [nifi-minifi-cpp] fgerlits commented on a diff in pull request #1606: MINIFICPP-2129 Refactor threadpool
fgerlits commented on code in PR #1606:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1606#discussion_r1287294806
##########
libminifi/test/unit/BackTraceTests.cpp:
##########
@@ -24,76 +24,60 @@
#include "utils/BackTrace.h"
#include "utils/Monitors.h"
#include "utils/ThreadPool.h"
+#include "range/v3/algorithm/any_of.hpp"
using namespace std::literals::chrono_literals;
-class WorkerNumberExecutions : public utils::AfterExecute<int> {
- public:
- explicit WorkerNumberExecutions(int tasks)
- : tasks(tasks) {
- }
-
- WorkerNumberExecutions(WorkerNumberExecutions && other) noexcept
- : runs(other.runs),
- tasks(other.tasks) {
- }
-
- bool isFinished(const int &result) override {
- return result <= 0 || ++runs >= tasks;
- }
- bool isCancelled(const int& /*result*/) override {
- return false;
- }
-
- std::chrono::steady_clock::duration wait_time() override {
- return 50ms;
- }
-
- protected:
- int runs{0};
- int tasks;
-};
-
TEST_CASE("BT1", "[TPT1]") {
const BackTrace trace = TraceResolver::getResolver().getBackTrace("BT1");
#ifdef HAS_EXECINFO
- REQUIRE(!trace.getTraces().empty());
+ CHECK(!trace.getTraces().empty());
#endif
}
-TEST_CASE("BT2", "[TPT2]") {
- std::atomic<int> counter = 0;
- utils::ThreadPool<int> pool(4);
- pool.start();
- std::this_thread::sleep_for(std::chrono::milliseconds(150));
- for (int i = 0; i < 3; i++) {
- std::unique_ptr<utils::AfterExecute<int>> after_execute = std::unique_ptr<utils::AfterExecute<int>>(new WorkerNumberExecutions(5));
- utils::Worker<int> functor([&counter]() { return ++counter; }, "id", std::move(after_execute));
+void inner_function(std::atomic_flag& ready_to_check, std::atomic_flag& checking_done) {
+ ready_to_check.test_and_set();
+ ready_to_check.notify_all();
+ checking_done.wait(false);
+}
- std::future<int> fut;
- pool.execute(std::move(functor), fut); // NOLINT(bugprone-use-after-move)
- }
+void outer_function(std::atomic_flag& ready_to_check, std::atomic_flag& checking_done) {
+ inner_function(ready_to_check, checking_done);
+}
- std::unique_ptr<utils::AfterExecute<int>> after_execute = std::unique_ptr<utils::AfterExecute<int>>(new WorkerNumberExecutions(5));
- utils::Worker<int> functor([&counter]() { return ++counter; }, "id", std::move(after_execute));
- std::future<int> fut;
- pool.execute(std::move(functor), fut); // NOLINT(bugprone-use-after-move)
+TEST_CASE("BT2", "[TPT2]") {
+ std::atomic_flag ready_for_checking;
+ std::atomic_flag done_with_checking;
- std::vector<BackTrace> traces = pool.getTraces();
- for (const auto &trace : traces) {
- std::cerr << "Thread name: " << trace.getName() << std::endl;
- const auto &trace_strings = trace.getTraces();
-#ifdef HAS_EXECINFO
- REQUIRE(trace_strings.size() > 2);
- for (const auto& trace_string : trace_strings) {
- std::cerr << " - " << trace_string << std::endl;
- }
- if (trace_strings.at(0).find("sleep_for") != std::string::npos) {
- REQUIRE(trace_strings.at(1).find("counterFunction") != std::string::npos);
+ constexpr std::string_view thread_pool_name = "Winnie the pool";
+ constexpr size_t number_of_worker_threads = 3;
+ utils::ThreadPool pool(number_of_worker_threads, nullptr, thread_pool_name.data());
+ utils::Worker worker([&]() -> utils::TaskRescheduleInfo {
+ outer_function(ready_for_checking, done_with_checking);
+ return utils::TaskRescheduleInfo::Done();
+ }, "id");
+ std::future<utils::TaskRescheduleInfo> future;
+ pool.execute(std::move(worker), future);
+
+ pool.start();
+ {
+ ready_for_checking.wait(false);
+ std::vector<BackTrace> traces = pool.getTraces();
+ CHECK(traces.size() <= number_of_worker_threads);
Review Comment:
I think we should check (or rather REQUIRE, because of line 72) that `traces.size() > 0`
##########
libminifi/test/unit/SocketTests.cpp:
##########
@@ -175,30 +179,28 @@ bool createSocket() {
socketA->initialize();
}
- return true;
+ co_return true;
}
/**
* MINIFI-320 was created to address reallocations within TLSContext
* This test will create 20 threads that attempt to create contexts
* to ensure we no longer see the segfaults.
*/
Review Comment:
I think this comment can be retired now.
##########
libminifi/test/unit/BackTraceTests.cpp:
##########
@@ -24,76 +24,60 @@
#include "utils/BackTrace.h"
#include "utils/Monitors.h"
#include "utils/ThreadPool.h"
+#include "range/v3/algorithm/any_of.hpp"
using namespace std::literals::chrono_literals;
-class WorkerNumberExecutions : public utils::AfterExecute<int> {
- public:
- explicit WorkerNumberExecutions(int tasks)
- : tasks(tasks) {
- }
-
- WorkerNumberExecutions(WorkerNumberExecutions && other) noexcept
- : runs(other.runs),
- tasks(other.tasks) {
- }
-
- bool isFinished(const int &result) override {
- return result <= 0 || ++runs >= tasks;
- }
- bool isCancelled(const int& /*result*/) override {
- return false;
- }
-
- std::chrono::steady_clock::duration wait_time() override {
- return 50ms;
- }
-
- protected:
- int runs{0};
- int tasks;
-};
-
TEST_CASE("BT1", "[TPT1]") {
const BackTrace trace = TraceResolver::getResolver().getBackTrace("BT1");
#ifdef HAS_EXECINFO
- REQUIRE(!trace.getTraces().empty());
+ CHECK(!trace.getTraces().empty());
#endif
}
-TEST_CASE("BT2", "[TPT2]") {
- std::atomic<int> counter = 0;
- utils::ThreadPool<int> pool(4);
- pool.start();
- std::this_thread::sleep_for(std::chrono::milliseconds(150));
- for (int i = 0; i < 3; i++) {
- std::unique_ptr<utils::AfterExecute<int>> after_execute = std::unique_ptr<utils::AfterExecute<int>>(new WorkerNumberExecutions(5));
- utils::Worker<int> functor([&counter]() { return ++counter; }, "id", std::move(after_execute));
+void inner_function(std::atomic_flag& ready_to_check, std::atomic_flag& checking_done) {
+ ready_to_check.test_and_set();
+ ready_to_check.notify_all();
+ checking_done.wait(false);
+}
- std::future<int> fut;
- pool.execute(std::move(functor), fut); // NOLINT(bugprone-use-after-move)
- }
+void outer_function(std::atomic_flag& ready_to_check, std::atomic_flag& checking_done) {
+ inner_function(ready_to_check, checking_done);
+}
- std::unique_ptr<utils::AfterExecute<int>> after_execute = std::unique_ptr<utils::AfterExecute<int>>(new WorkerNumberExecutions(5));
- utils::Worker<int> functor([&counter]() { return ++counter; }, "id", std::move(after_execute));
- std::future<int> fut;
- pool.execute(std::move(functor), fut); // NOLINT(bugprone-use-after-move)
+TEST_CASE("BT2", "[TPT2]") {
+ std::atomic_flag ready_for_checking;
+ std::atomic_flag done_with_checking;
- std::vector<BackTrace> traces = pool.getTraces();
- for (const auto &trace : traces) {
- std::cerr << "Thread name: " << trace.getName() << std::endl;
- const auto &trace_strings = trace.getTraces();
-#ifdef HAS_EXECINFO
- REQUIRE(trace_strings.size() > 2);
- for (const auto& trace_string : trace_strings) {
- std::cerr << " - " << trace_string << std::endl;
- }
- if (trace_strings.at(0).find("sleep_for") != std::string::npos) {
- REQUIRE(trace_strings.at(1).find("counterFunction") != std::string::npos);
+ constexpr std::string_view thread_pool_name = "Winnie the pool";
+ constexpr size_t number_of_worker_threads = 3;
+ utils::ThreadPool pool(number_of_worker_threads, nullptr, thread_pool_name.data());
+ utils::Worker worker([&]() -> utils::TaskRescheduleInfo {
+ outer_function(ready_for_checking, done_with_checking);
+ return utils::TaskRescheduleInfo::Done();
+ }, "id");
+ std::future<utils::TaskRescheduleInfo> future;
+ pool.execute(std::move(worker), future);
+
+ pool.start();
+ {
+ ready_for_checking.wait(false);
+ std::vector<BackTrace> traces = pool.getTraces();
+ CHECK(traces.size() <= number_of_worker_threads);
Review Comment:
can `traces.size()` be anything other than 1? we only schedule a single task
also: should we schedule multiple tasks? the old test looks like it scheduled 3 of them, although it's not clear to me if they could run in parallel
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org