Re: 一道count frequency of all words的面试题 (转载) - 未名空间MITBBS历史存档

Re: 一道count frequency of all words的面试题 (转载)# JobHunting - 待字闺中

s*w2013-09-23 07:09

1 楼

要有人觉得有所帮助，给发个包子
【以下文字转载自 Programming 讨论区】
发信人: saw (句子熊), 信区: Programming
标题: Re: 一道count frequency of all words的面试题 (转载)
发信站: BBS 未名空间站 (Mon Sep 23 00:28:12 2013, 美东)
又琢磨了两天，看了不少相关东西，终于搞定了，觉得自己对这个多线程的理解加强了
很多。思考比单纯的看人说原理更刻骨铭心。
这个问题我设计的用一个 producer 多个 consumer 的架构，上个版本的是两头用
condition_variable, 一个通知 producer 有空间了开始干活生产，另一个通知
consumer 有库存了开始消费。参见上篇里面的 wait 和　notify_all,notify_one 语
句。　这个思路对于单 producer 单　consumer 没问题，可是不适用于多 consumer.
因为所有的 consumer 可能同时睡觉（没空存），同时醒来（有库存）,结果只有一个
能抢占mutex(拿到库存)，其他的只好继续睡觉（while 循环 wait）.　如果无限制的
生产下去，每个睡觉的都有机会醒来干活。可是在有限生产的情况下，producer 干完
活了后，总有睡觉的　consumer 无人唤醒导致死锁。解决的办法就是用　non-block
的　try_lock, lock 不上就返回　false，这样有机会检查是否需要(还在生产或是有
数据没处理)继续　try_lock
还有些零碎的改动：任何数据，只要有两个线程用，其中一个写，就是　shared data.
我最开始的版本中，都是只想到了保存中间数据的　deque data, 而忘记了
保存最终结果的　map 也是被多个 consumer 同时写的。
用过　valgrind 的　helgrind，　没发现咋用
贴一下现在的　code
运行环境：linux gcc 4.7.2
编译：g++ -pthread -std=c++11 testcc.cpp
运行：./a.out < 任意有几行文本的文件
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
using namespace std;
/*
hide concurrency inside the data structure
to minimize critical sections
*/
class concurDeque {
int capacity;
deque data;
mutex mx;
condition_variable room2produce;
public:
concurDeque(int cap) : capacity(cap) {}
int size() {
unique_lock lck(mx);
return data.size();
}
void put(const string &line) {
{
unique_lock lck(mx);
//cout << line << endl;
// while loop against spurious wakeup
while(data.size()>=capacity)
room2produce.wait(lck);
data.push_back(line);
// do not rely on lck out of scope for unlocking mx
// coz notify may be too fast and the other party may wake up to
find
// that this lck is not out of scope yet
//lck.unlock();
}
}
bool get(string &line) {
bool consumed = false;
if(mx.try_lock()) {
if(!data.empty()) {
line = data.front();
data.pop_front();
consumed = true;
}
mx.unlock();
}
if(consumed)
room2produce.notify_all();
return consumed;
}
};
class concurMap {
map freq;
mutex mx;
public:
void insert(string line) {
lock_guard lck(mx);
istringstream iss(line);
string word;
while( iss >> word)
freq[word]++;
}
void print() {
lock_guard lck(mx);
cout << "counts as following:" << endl;
for(auto kvpair : freq)
cout << kvpair.first << ": " << kvpair.second << endl;
}
};
concurDeque ccd(5);
concurMap ccm;
atomic stillproducing{true};
void consumer() {
// cout here may be messed up coz no mutex production
// but it is just for debug anyway
while(stillproducing || ccd.size()>0) {
string line;
if(ccd.get(line)) // get should not block
ccm.insert(line);
/* {// just for debug
mutex mtx;
lock_guard lck(mtx);
cout << "consumer " << this_thread::get_id() << ":" << line << endl;
}
*/
}
}
void producer() {
string line;
while(getline(cin,line)) {
ccd.put(line);
}
//cout << "producer " << this_thread::get_id() << ": end!" << endl;
stillproducing.store(false);
}
int main() {
vector ts;
int nThread = 3; // now works for multiple consumers
// launch consumers in threads
for(int i=0;its.push_back(thread(consumer));
// a single producer in main, since IO is bottleneck, no need for
concurrent read of input
producer();
// wait consumers finish all work
for(int i=0;its[i].join();
//
ccm.print();
}

p*22013-09-23 07:09

2 楼

都啥年代了，还用这么原始的方式？

s*w2013-09-23 07:09

3 楼

大牛说的对，这个是很原始，应该尽量用　async, packaged_task，我理解你的意思,
接下来就用 async 来试试
不过那是高级阶段了，学习的时候能想明白多线程里面的很多问题还是挺好的，总不能
面试的时候回答我不知道为啥他们醒不来

【在 p*****2 的大作中提到】

: 都啥年代了，还用这么原始的方式？

p*22013-09-23 07:09

4 楼

,
嗯。学习不错。我还基本没用过地层的多线程呢。

【在 s*w 的大作中提到】

: 大牛说的对，这个是很原始，应该尽量用　async, packaged_task，我理解你的意思,
: 接下来就用 async 来试试
: 不过那是高级阶段了，学习的时候能想明白多线程里面的很多问题还是挺好的，总不能
: 面试的时候回答我不知道为啥他们醒不来

w*m2013-09-23 07:09

5 楼

我十年前用过

【在 p*****2 的大作中提到】

:
: ,
: 嗯。学习不错。我还基本没用过地层的多线程呢。