通过同步延长线程的生命周期(c++ 11)
Extend the life of threads with synchronization (C++11)
我有一个程序,其中有一个函数,它接受一个指针作为参数,和一个main。主要是创建n个线程,每个线程根据传递的arg
在不同的内存区域上运行函数。然后将线程连接起来,主线程在区域之间执行一些数据混合,并创建n新线程,这些新线程执行与旧线程相同的操作。
为了改进程序,我想让线程保持活动状态,消除创建它们所需的长时间。线程应该在主线程工作时处于睡眠状态,并在必须再次启动时得到通知。以同样的方式,主线程应该在线程工作时等待,就像join一样。
我不能用一个强大的实现来结束这个,总是陷入死锁。
简单的基线代码,任何关于如何修改它的提示将不胜感激
#include <thread>
#include <climits>
...
void myfunc(void * p) {
do_something(p);
}
int main(){
void * myp[n_threads] {a_location, another_location,...};
std::thread mythread[n_threads];
for (unsigned long int j=0; j < ULONG_MAX; j++) {
for (unsigned int i=0; i < n_threads; i++) {
mythread[i] = std::thread(myfunc, myp[i]);
}
for (unsigned int i=0; i < n_threads; i++) {
mythread[i].join();
}
mix_data(myp);
}
return 0;
}
这是一种仅使用c++ 11标准库中的类的可能方法。基本上,您创建的每个线程都有一个相关的命令队列(封装在std::packaged_task<>
对象中),它会不断地检查该队列。如果队列是空的,线程将等待一个条件变量(std::condition_variable
)。
虽然通过使用std::mutex
和std::unique_lock<>
RAII包装器可以避免数据竞争,但是主线程可以通过存储与每个提交的std::packaged_tast<>
相关联的std::future<>
对象并调用wait()
来等待特定作业的终止。
下面是一个遵循这种设计的简单程序。注释应该足以解释它的作用:
#include <thread>
#include <iostream>
#include <sstream>
#include <future>
#include <queue>
#include <condition_variable>
#include <mutex>
// Convenience type definition
using job = std::packaged_task<void()>;
// Some data associated to each thread.
struct thread_data
{
int id; // Could use thread::id, but this is filled before the thread is started
std::thread t; // The thread object
std::queue<job> jobs; // The job queue
std::condition_variable cv; // The condition variable to wait for threads
std::mutex m; // Mutex used for avoiding data races
bool stop = false; // When set, this flag tells the thread that it should exit
};
// The thread function executed by each thread
void thread_func(thread_data* pData)
{
std::unique_lock<std::mutex> l(pData->m, std::defer_lock);
while (true)
{
l.lock();
// Wait until the queue won't be empty or stop is signaled
pData->cv.wait(l, [pData] () {
return (pData->stop || !pData->jobs.empty());
});
// Stop was signaled, let's exit the thread
if (pData->stop) { return; }
// Pop one task from the queue...
job j = std::move(pData->jobs.front());
pData->jobs.pop();
l.unlock();
// Execute the task!
j();
}
}
// Function that creates a simple task
job create_task(int id, int jobNumber)
{
job j([id, jobNumber] ()
{
std::stringstream s;
s << "Hello " << id << "." << jobNumber << std::endl;
std::cout << s.str();
});
return j;
}
int main()
{
const int numThreads = 4;
const int numJobsPerThread = 10;
std::vector<std::future<void>> futures;
// Create all the threads (will be waiting for jobs)
thread_data threads[numThreads];
int tdi = 0;
for (auto& td : threads)
{
td.id = tdi++;
td.t = std::thread(thread_func, &td);
}
//=================================================
// Start assigning jobs to each thread...
for (auto& td : threads)
{
for (int i = 0; i < numJobsPerThread; i++)
{
job j = create_task(td.id, i);
futures.push_back(j.get_future());
std::unique_lock<std::mutex> l(td.m);
td.jobs.push(std::move(j));
}
// Notify the thread that there is work do to...
td.cv.notify_one();
}
// Wait for all the tasks to be completed...
for (auto& f : futures) { f.wait(); }
futures.clear();
//=================================================
// Here the main thread does something...
std::cin.get();
// ...done!
//=================================================
//=================================================
// Posts some new tasks...
for (auto& td : threads)
{
for (int i = 0; i < numJobsPerThread; i++)
{
job j = create_task(td.id, i);
futures.push_back(j.get_future());
std::unique_lock<std::mutex> l(td.m);
td.jobs.push(std::move(j));
}
// Notify the thread that there is work do to...
td.cv.notify_one();
}
// Wait for all the tasks to be completed...
for (auto& f : futures) { f.wait(); }
futures.clear();
// Send stop signal to all threads and join them...
for (auto& td : threads)
{
std::unique_lock<std::mutex> l(td.m);
td.stop = true;
td.cv.notify_one();
}
// Join all the threads
for (auto& td : threads) { td.t.join(); }
}
您想要的概念是线程池。这个问题涉及到现有的实现。
这个想法是为许多线程实例提供一个容器。每个实例都与一个函数相关联,该函数轮询任务队列,当任务可用时,拉出它并运行它。一旦任务结束(如果它终止,但这是另一个问题),线程就会循环到任务队列。
所以你需要一个同步队列,一个线程类来实现队列上的循环,一个任务对象的接口,也许还有一个类来驱动整个事情(池类)。
或者,您可以为它必须执行的任务创建一个非常专门的线程类(例如,仅将内存区域作为参数)。这需要为线程提供通知机制,以指示它们已完成当前迭代。
线程main函数将是该特定任务的循环,在一次迭代结束时,线程发出结束信号,并等待条件变量开始下一个循环。从本质上讲,您将在线程内内联任务代码,从而完全放弃对队列的需求。
using namespace std;
// semaphore class based on C++11 features
class semaphore {
private:
mutex mMutex;
condition_variable v;
int mV;
public:
semaphore(int v): mV(v){}
void signal(int count=1){
unique_lock lock(mMutex);
mV+=count;
if (mV > 0) mCond.notify_all();
}
void wait(int count = 1){
unique_lock lock(mMutex);
mV-= count;
while (mV < 0)
mCond.wait(lock);
}
};
template <typename Task>
class TaskThread {
thread mThread;
Task *mTask;
semaphore *mSemStarting, *mSemFinished;
volatile bool mRunning;
public:
TaskThread(Task *task, semaphore *start, semaphore *finish):
mTask(task), mRunning(true),
mSemStart(start), mSemFinished(finish),
mThread(&TaskThread<Task>::psrun){}
~TaskThread(){ mThread.join(); }
void run(){
do {
(*mTask)();
mSemFinished->signal();
mSemStart->wait();
} while (mRunning);
}
void finish() { // end the thread after the current loop
mRunning = false;
}
private:
static void psrun(TaskThread<Task> *self){ self->run();}
};
classcMyTask {
public:
MyTask(){}
void operator()(){
// some code here
}
};
int main(){
MyTask task1;
MyTask task2;
semaphore start(2), finished(0);
TaskThread<MyTask> t1(&task1, &start, &finished);
TaskThread<MyTask> t2(&task2, &start, &finished);
for (int i = 0; i < 10; i++){
finished.wait(2);
start.signal(2);
}
t1.finish();
t2.finish();
}
上面建议的(粗略的)实现依赖于Task
类型,它必须提供operator()
(即。类的函子)。我之前说过可以将任务代码直接合并到线程函数体中,但由于我不了解它,所以我尽量保持它的抽象。一个条件变量用于线程的开始,一个条件变量用于线程的结束,这两个条件变量都封装在信号量实例中。
看到另一个建议使用boost::barrier
的答案,我只能支持这个想法:确保用那个类替换我的信号量类,如果可能的话,原因是更好地依赖于经过良好测试和维护的外部代码,而不是为相同的功能集自行实现的解决方案。
总而言之,这两种方法都是有效的,但是前者为了灵活性而放弃了一点性能。如果要执行的任务需要足够长的时间,则管理和队列同步成本可以忽略不计。
更新:代码修复和测试。将一个简单的条件变量替换为一个信号量
可以很容易地使用屏障(只是一个条件变量和计数器的方便包装)来实现。它基本上会阻塞,直到所有N个线程都到达"屏障"。然后再次"循环"。Boost提供了一个实现
void myfunc(void * p, boost::barrier& start_barrier, boost::barrier& end_barrier) {
while (!stop_condition) // You'll need to tell them to stop somehow
{
start_barrier.wait ();
do_something(p);
end_barrier.wait ();
}
}
int main(){
void * myp[n_threads] {a_location, another_location,...};
boost::barrier start_barrier (n_threads + 1); // child threads + main thread
boost::barrier end_barrier (n_threads + 1); // child threads + main thread
std::thread mythread[n_threads];
for (unsigned int i=0; i < n_threads; i++) {
mythread[i] = std::thread(myfunc, myp[i], start_barrier, end_barrier);
}
start_barrier.wait (); // first unblock the threads
for (unsigned long int j=0; j < ULONG_MAX; j++) {
end_barrier.wait (); // mix_data must not execute before the threads are done
mix_data(myp);
start_barrier.wait (); // threads must not start new iteration before mix_data is done
}
return 0;
}
下面是一个简单的编译和工作代码,执行一些随机的东西。它实现了阿莱古纳的屏障概念。每个线程的任务长度是不同的,所以有必要有一个强大的同步机制。我将尝试在相同的任务上做一个池并对结果进行基准测试,然后可能会像Andy Prowl指出的那样使用期货。
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include <complex>
#include <random>
const unsigned int n_threads=4; //varying this will not (almost) change the total amount of work
const unsigned int task_length=30000/n_threads;
const float task_length_variation=task_length/n_threads;
unsigned int rep=1000; //repetitions of tasks
class t_chronometer{
private:
std::chrono::steady_clock::time_point _t;
public:
t_chronometer(): _t(std::chrono::steady_clock::now()) {;}
void reset() {_t = std::chrono::steady_clock::now();}
double get_now() {return std::chrono::duration_cast<std::chrono::duration<double>>(std::chrono::steady_clock::now() - _t).count();}
double get_now_ms() {return
std::chrono::duration_cast<std::chrono::duration<double,std::milli>>(std::chrono::steady_clock::now() - _t).count();}
};
class t_barrier {
private:
std::mutex m_mutex;
std::condition_variable m_cond;
unsigned int m_threshold;
unsigned int m_count;
unsigned int m_generation;
public:
t_barrier(unsigned int count):
m_threshold(count),
m_count(count),
m_generation(0) {
}
bool wait() {
std::unique_lock<std::mutex> lock(m_mutex);
unsigned int gen = m_generation;
if (--m_count == 0)
{
m_generation++;
m_count = m_threshold;
m_cond.notify_all();
return true;
}
while (gen == m_generation)
m_cond.wait(lock);
return false;
}
};
using namespace std;
void do_something(complex<double> * c, unsigned int max) {
complex<double> a(1.,0.);
complex<double> b(1.,0.);
for (unsigned int i = 0; i<max; i++) {
a *= polar(1.,2.*M_PI*i/max);
b *= polar(1.,4.*M_PI*i/max);
*(c)+=a+b;
}
}
bool done=false;
void task(complex<double> * c, unsigned int max, t_barrier* start_barrier, t_barrier* end_barrier) {
while (!done) {
start_barrier->wait ();
do_something(c,max);
end_barrier->wait ();
}
cout << "task finished" << endl;
}
int main() {
t_chronometer t;
std::default_random_engine gen;
std::normal_distribution<double> dis(.0,1000.0);
complex<double> cpx[n_threads];
for (unsigned int i=0; i < n_threads; i++) {
cpx[i] = complex<double>(dis(gen), dis(gen));
}
t_barrier start_barrier (n_threads + 1); // child threads + main thread
t_barrier end_barrier (n_threads + 1); // child threads + main thread
std::thread mythread[n_threads];
unsigned long int sum=0;
for (unsigned int i=0; i < n_threads; i++) {
unsigned int max = task_length + i * task_length_variation;
cout << i+1 << "th task length: " << max << endl;
mythread[i] = std::thread(task, &cpx[i], max, &start_barrier, &end_barrier);
sum+=max;
}
cout << "total task length " << sum << endl;
complex<double> c(0,0);
for (unsigned long int j=1; j < rep+1; j++) {
start_barrier.wait (); //give to the threads the missing call to start
if (j==rep) done=true;
end_barrier.wait (); //wait for the call from each tread
if (j%100==0) cout << "cycle: " << j << endl;
for (unsigned int i=0; i<n_threads; i++) {
c+=cpx[i];
}
}
for (unsigned int i=0; i < n_threads; i++) {
mythread[i].join();
}
cout << "result: " << c << " it took: " << t.get_now() << " s." << endl;
return 0;
}
- Pisano 周期生成器在 3 位周期内行为不正确
- 如何在共享库的整个生命周期内存储数据
- 提升图形库:资源受限的最短周期
- 我必须找到给定数组中所有数字的周期,就像有很多解决方案,但数组的大小是 10^5
- 如何理解句子的生命周期始于对e的评估
- 它解决了什么问题,对于非真空初始化,生命周期在初始化之前就开始了
- 找到包括特定边的最短周期
- Go/C++gRPC客户端通道和存根生命周期
- 如何将"this"的生命周期移动到C++中的另一个对象中?
- 是否可以通过使用移动/交换 c++11 来延长返回的临时变量的生命周期
- 如何在没有同步的情况下使用多个线程(2、4,8、16 个线程)在循环(10,100、1000 个周期)中打印字符串?
- C/C++ 经过的流程周期,不包括断点处
- 标准::计时::时钟、硬件时钟和周期计数
- 使用对象的生命周期作为设置器的安全性
- time_t的时钟周期和获取时间问题
- 查询时间/周期是针对所有内核/线程的序列化请求还是并行请求?
- 临时人员的生命周期传递给函数
- 使用贝尔曼福特检测产品超过阈值的周期
- 为什么这个循环每次迭代需要 1.32 个周期
- 我想知道在构造函数中初始化变量时的生命周期