I have written a simple Worker Pool program. When I compile it without -Ox flag it runs as expected, but when i use -Ox (x>0) or -Og threads do not start.
worker_pool.h
#ifndef WORKER_POOL_H
#define WORKER_POOL_H
#include <sema.h>
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <optional>
#include <queue>
#include <thread>
#include <vector>
using namespace std;
class worker_pool {
private:
class worker {
private:
worker_pool *wp;
long id;
public:
worker(worker_pool *_wp, long _id) : wp(_wp), id(_id){};
void operator()() {
while (!wp->stop) {
printf("monkey %d doing job\n", int(id));
auto t = wp->fetch();
if (!t.has_value())
continue;
else
t.value()();
}
};
};
std::vector<std::thread> workers;
std::queue<std::function<void(void)>> q;
std::mutex mx;
Semaphore sm;
std::condition_variable cv_empty;
std::mutex mx_empty;
std::optional<std::function<void(void)>> fetch() {
sm.wait();
std::unique_lock l(mx);
if (stop)
return {};
auto res = std::move(q.front());
q.pop();
if (q.empty())
cv_empty.notify_all();
return std::move(res);
};
public:
worker_pool(long tcount = std::thread::hardware_concurrency()) {
for (long i = 0; i < tcount; i++) {
workers.push_back(std::move(std::thread(worker(this, i))));
}
}
~worker_pool() { terminate(); }
worker_pool(worker_pool const &) = delete;
worker_pool &operator=(worker_pool const &) = delete;
worker_pool(worker_pool &&) = delete;
worker_pool &operator=(worker_pool &&) = delete;
bool stop;
template <typename F, typename... ARGS>
auto submit(F &&f, ARGS &&... args) -> std::future<decltype(f(args...))> {
std::lock_guard l(mx);
auto func = std::bind(std::forward<F>(f), std::forward(args)...);
auto task_ptr =
std::make_shared<std::packaged_task<decltype(f(args...))()>>(func);
q.push([task_ptr] { (*task_ptr)(); });
sm.signal();
return task_ptr->get_future();
}
void terminate() {
stop = true;
sm.signal(workers.size());
for (size_t i = 0; i < workers.capacity(); i++) {
if (workers[i].joinable())
workers[i].join();
}
}
long jobs_remaining() {
std::lock_guard l(mx);
return q.size();
}
void wait_until_empty() {
std::unique_lock l(mx_empty);
while (!(q.empty() || stop))
cv_empty.wait(l, [&] { return q.empty() || stop; });
}
};
#endif // WORKER_POOL_H
main.cpp
#include <sema.h>
#include <worker_pool.h>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <string>
using namespace std;
int main(int argc, char *argv[]) {
printf("creating worker pool\n");
worker_pool wp(4);
vector<future<string>> futures;
printf("creating jobs\n");
for (int i = 0; i < 10; i++) {
futures.push_back(
wp.submit([]() { return std::move(string("Hello world")); }));
}
printf("waiting for jobs to finish\n");
wp.wait_until_empty();
for (auto &f : futures) {
auto str = f.get();
cout << str << endl;
}
printf("finished jobs\n");
wp.terminate();
cout << "program ended" << endl;
return 0;
}
Compilation commands:
clang++ -std=c++17 -pthread -g -Iinclude -Llib src/main.cpp -o bin/gmain
clang++ -std=c++17 -pthread -Iinclude -Llib -O3 src/main.cpp -o bin/main
Clang version 8, GCC version 8.2.1.
Aucun commentaire:
Enregistrer un commentaire