[并发编程]使用线程安全队列和条件变量的notify来安排分步骤任务

 1 // 方法1：直接构建N个THread来run
 2 foreach (i, size)
 3 {
 4     thread trd(&Instance::doWork, &inst);
 5     lstTrd.push_back(trd);
 6 }
 7 
 8 foreach(lstTrd, join);
 9 
10 
11 // 方法2：使用notify的方式来处理分步骤任务
12 
13 class threadsafe_queue
14 {
15     queue<int> q;
16     mutex m;
17     condition_variable cond;
18 public:
19     void push(int i)
20     {
21         lock_guard<mutex> lk(m);
22         q.push_back(i);
23         cond.notify_one();
24     }
25     
26     void wait_and_pop(int &i)
27     {
28         unique_lock<mutex> lk(m);
29         cond.wait(lk, [this]{ return !q.empty(); } );
30         i = q.front();
31         q.pop();
32     }
33 }
34 
35 class Instance
36 {
37     // 存储原始的用于处理的n个数目的数据
38     List<GDSLayer> lstInfo;
39     ...
40     // 存储最终的数据（其实应该每个步骤都有中间过程数据产生）
41     List<Result>   lstResult;
42     
43     // 每个步骤完成就往对应的queue里面把完成的数据的index插入，而下一个线程就在等待这个队列有元素
44     // 于是对于数据index = 0就会进行：A->B->C...
45     threadsafe_queue merge_q;
46     threadsafe_queue match_q;
47     threadsafe_queue GDSPinGen_q;
48     threadsafe_queue OBSGen_q;
49     threadsafe_queue LEFPin_q;
50 public:
51     void merge_thread()
52     {
53         // 如果发生错误，那么是否要符合没有到size也可以退出
54         // 事先是否要把所有可能的操作数的空间都分配好，免得如果先操作0，结果notify到了3，然后操作3，虽然原始数据有，但存储过程数据还没有
55         // 同时，Log信息处理是否可行？中间检查信息的操作是安全的吗？
56         // 但这样，每个thread只负责一个步骤，处理完一个数据才处理下一个（是否如此？）按理来说是不会存在跨越的问题
57         foreach (i, lstInfo.size())
58         {
59             bool b_state = doMerge(lstInfo[i]);
60             if (b_state)
61                 merge_q.push_back(i);
62         }
63     }
64     
65     void match_thread()
66     {
67         foreach(i, lstInfo.size())
68         {
69             int key = -1;
70             merge_q.wait_and_pop(key);
71             bool b_state = doMatch( lstInfo[ key ] );
72             if (b_state)
73             {
74                 match_q.push_back( key );
75             }
76         }
77     }
78     
79     void GDSPinGen_thread()
80     {}
81     
82     void OBSGen_thread()
83     {}
84     
85     void LEFPin_thread()
86     {}
87 }