Commit 69fa3511 authored by Adrien Béraud's avatar Adrien Béraud

add thread pool

A simple thread pool.
It will start a new thread if all current threads
are busy, up to the number of cores on the host.
Keeps pending tasks on a queue.

Adapts few uses in the daemon.

Change-Id: Ia94a64bcdd4d8d8b16cf84e5a735c5246b08c13d
parent b6771a0c
......@@ -93,6 +93,7 @@ libring_la_SOURCES = \
fileutils.cpp \
archiver.cpp \
threadloop.cpp \
thread_pool.cpp \
ip_utils.h \
ip_utils.cpp \
utf8_utils.cpp \
......@@ -104,6 +105,7 @@ libring_la_SOURCES = \
plugin_loader.h \
plugin_manager.h \
threadloop.h \
thread_pool.h \
conference.h \
account_factory.h \
call_factory.h \
......
......@@ -25,6 +25,8 @@
#include "config.h"
#endif
#include "thread_pool.h"
#include "sip/sdp.h"
#include "sip/sipvoiplink.h"
#include "sip/sipcall.h"
......@@ -1293,11 +1295,7 @@ RingAccount::generateDhParams()
{
//make sure cachePath_ is writable
fileutils::check_dir(cachePath_.c_str(), 0700);
std::packaged_task<decltype(loadDhParams)> task(loadDhParams);
dhParams_ = task.get_future();
std::thread task_td(std::move(task), cachePath_ + DIR_SEPARATOR_STR "dhParams");
task_td.detach();
dhParams_ = ThreadPool::instance().get<tls::DhParams>(std::bind(loadDhParams, cachePath_ + DIR_SEPARATOR_STR "dhParams"));
}
MatchRank
......
......@@ -22,6 +22,7 @@
#include "client/ring_signal.h"
#include "thread_pool.h"
#include "fileutils.h"
#include "logger.h"
......@@ -179,7 +180,7 @@ readCertificates(const std::string& path)
void
CertificateStore::pinCertificatePath(const std::string& path, std::function<void(const std::vector<std::string>&)> cb)
{
std::thread([&, path, cb]() {
ThreadPool::instance().run([&, path, cb]() {
auto certs = readCertificates(path);
std::vector<std::string> ids;
std::vector<std::weak_ptr<crypto::Certificate>> scerts;
......@@ -201,7 +202,7 @@ CertificateStore::pinCertificatePath(const std::string& path, std::function<void
if (cb)
cb(ids);
emitSignal<DRing::ConfigurationSignal::CertificatePathPinned>(path, ids);
}).detach();
});
}
unsigned
......
/*
* Copyright (C) 2016 Savoir-faire Linux Inc.
*
* Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "thread_pool.h"
#include "logger.h"
#include <atomic>
#include <thread>
namespace ring {
struct ThreadPool::ThreadState
{
std::thread thread {};
std::atomic_bool run {true};
};
ThreadPool::ThreadPool()
: maxThreads_(std::thread::hardware_concurrency())
{
threads_.reserve(maxThreads_);
}
ThreadPool::~ThreadPool()
{
join();
}
void
ThreadPool::run(std::function<void()>&& cb)
{
std::unique_lock<std::mutex> l(lock_);
// launch new thread if necessary
if (not readyThreads_ && threads_.size() < maxThreads_) {
threads_.emplace_back(new ThreadState());
auto& t = *threads_.back();
t.thread = std::thread([&]() {
while (t.run) {
std::function<void()> task;
// pick task from queue
{
std::unique_lock<std::mutex> l(lock_);
readyThreads_++;
cv_.wait(l, [&](){
return not t.run or not tasks_.empty();
});
readyThreads_--;
if (not t.run)
break;
task = std::move(tasks_.front());
tasks_.pop();
}
// run task
try {
if (task)
task();
} catch (const std::exception& e) {
RING_ERR("Exception running task: %s", e.what());
}
}
});
}
// push task to queue
tasks_.emplace(std::move(cb));
// notify thread
l.unlock();
cv_.notify_one();
}
void
ThreadPool::join()
{
for (auto& t : threads_)
t->run = false;
cv_.notify_all();
for (auto& t : threads_)
t->thread.join();
threads_.clear();
}
}
/*
* Copyright (C) 2016 Savoir-faire Linux Inc.
*
* Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#pragma once
#include <condition_variable>
#include <vector>
#include <queue>
#include <future>
namespace ring {
class ThreadPool {
public:
static ThreadPool& instance() {
static ThreadPool pool;
return pool;
}
ThreadPool();
~ThreadPool();
void run(std::function<void()>&& cb);
template<class T>
std::future<T> get(std::function<T()>&& cb) {
auto ret = std::make_shared<std::promise<T>>();
run(std::bind([=](std::function<T()>& mcb) mutable {
ret->set_value(mcb());
}, std::move(cb)));
return ret->get_future();
}
void join();
private:
struct ThreadState;
std::queue<std::function<void()>> tasks_ {};
std::vector<std::unique_ptr<ThreadState>> threads_;
unsigned readyThreads_ {0};
std::mutex lock_ {};
std::condition_variable cv_ {};
const unsigned maxThreads_;
};
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment