多线程OR多进程-访存密集型应用

JohnBull

hy036630 发表于 2011-12-21 14:58
不好意思 可能我上面没有说清楚
我用的是下面这个环形队列  无锁的实现一个进程读，一个进程写，没有牵涉 ...

不妨设置一个计数器，在“push时满”和"pop时空"的时候++下，统计下一分钟发生多少次，再决定下一步怎办

hy036630

这倒是个好主意
我试试 谢谢了{:3_188:}

zylthinking

hy036630 发表于 2011-12-21 14:58
不好意思 可能我上面没有说清楚
我用的是下面这个环形队列  无锁的实现一个进程读，一个进程写，没有牵涉 ...

看代码， 流水线中几个线程都会处理同一份数据？
是不是这样

1. thread1(char* p){
   
2.     static int n = 0;
   
3.    n += p[1];
   
4. }
   
5. 
   
6. thread2(char* p){
   
7.     static int n = 0;
   
8.    n += p[2];
   
9. }
   
10. 
    
11. 。。。。。
    
12. 
    
13. thread_i(char* p){
    
14.     static int n = 0;
    
15.    n += p[i];
    
16. }

复制代码

然而 p 是相同的？

sonicling

用semaphore把队列指针包起来，一来可以保护指针，二来可以及时唤醒等待线程。如果忙等的话，生产者线程更新了队列，等到消费者线程那边发现的时候，黄花菜都凉了。

zylthinking

sonicling 发表于 2011-12-21 17:12
用semaphore把队列指针包起来，一来可以保护指针，二来可以及时唤醒等待线程。如果忙等的话，生产者线程更新 ...

它很显然是多核， 倒不是这个原因， 很有可能是多处理器缓存同步

sonicling

也可以不用semaphore，但是要及时唤醒等待线程！否则等待线程会等到下一次自然调度才得到通知。

sonicling

CPU间缓存同步会影响效率，但是不至于数量级的差别。我以前也做过无锁循环队列，不过是内核级的，用atomic_t来表示位置，每次更新队列，强制唤醒消费线程，吞吐量上60w/s无压力。

hy036630

的确是这样的，我觉得也不应该是忙等的问题，应为机器是16线程的机器，按照道理至少有8个独立线程同时跑，
缓存我开始以为是这样的原因

不过设置了CPU亲和了以后也没有效果。

hy036630

我所谓的忙等是这个意思
从环形队列取
while(NULL!=(event = queue.pop()));
doSomething();

插入到环形队列里
while(queue.push(event));
doSomething();


在多核的机器上面，应该不存在比较大的问题吧。

hy036630

sonicling 发表于 2011-12-21 17:31
CPU间缓存同步会影响效率，但是不至于数量级的差别。我以前也做过无锁循环队列，不过是内核级的，用atomic_ ...

其实我觉得不是队列慢了导致整体性能慢，
我觉得可能是增加流水线的长度会导致处理数据的速度变慢。
我给大家贴一段处理的代码吧。

1. string GroupPE::getEventKey(eventPtr msg) {
   
2.     vector<string> keys;
   
3.     keys.push_back("UchAPN"); // apn
   
4.     keys.push_back("Wlac"); // 位置区编号：lacid
   
5.     keys.push_back("Cell"); // 小区信息：sac
   
6.     keys.push_back("InterfaceType"); // 接口类型：interfacetype
   
7.     keys.push_back("Protocoltype"); // 协议类型：protocolid
   
8.     keys.push_back("Eventtype"); // 事件类型：eventtype
   
9.     string key = Utils::encoding(*msg, keys.begin(), keys.end(), 2000);
   
10.     return key;
    
11. }
    
12. 
    
13.    /*每个线程有一个统计map 用来存放统计数据 每次来一个event就根据getEventKey这个方法得到key，然后查看key是不是在统计map中，如果在统计map中，那么更新这个map的数据，如果没有，那么new一个event，设置一些值，然后放到统计map中*/
    
14. void GroupPE::process(eventPtr msg) {//统计
    
15.   
    
16.     string key = this->getEventKey(msg);
    
17.     map<string, eventPtr>::iterator iter4map = this->groupMap.find(key);
    
18. 
    
19. 
    
20.     if (iter4map != groupMap.end()) {
    
21.         int times = iter4map->second->getValueByName("times", 0);
    
22.         ++times;
    
23.         iter4map->second->setValueByName("times", times);
    
24.         if (msg->getValueByName("cause", -1) == 255) { // response success
    
25.             times = iter4map->second->getValueByName("succtimes", 0);
    
26.             ++times;
    
27.             iter4map->second->setValueByName("succtimes", times);
    
28.         }
    
29.         string str = msg->getValueByName("Btime", "1970-01-01 00:00:00.000");
    
30.         long btime = Utils::millisecs(str.c_str());
    
31.         str = msg->getValueByName("RspTime", "1970-01-01 00:00:00.000");
    
32.         long rsptime = Utils::millisecs(str.c_str());
    
33.         double delays = ((double) rsptime - (double) btime) / 1000;
    
34.         double d = iter4map->second->getValueByName("delays", (double) 0);
    
35.         iter4map->second->setValueByName("delays", delays + d);
    
36. 
    
37.         d = iter4map->second->getValueByName("maxdelays", (double) 0);
    
38.         if (d < delays)
    
39.             iter4map->second->setValueByName("maxdelays", delays);
    
40.         d = iter4map->second->getValueByName("mindelays", (double) 0);
    
41.         if (d > delays)
    
42.             iter4map->second->setValueByName("mindelays", delays);
    
43.         if (btime != rsptime || msg->getValueByName("Tdrtype", (char) 0) != 1) {
    
44.             times = iter4map->second->getValueByName("rsptimes", 0);
    
45.             ++times;
    
46.             iter4map->second->setValueByName("rsptimes", times);
    
47.         } else {
    
48.         }
    
49.     } else { // 进行新统计结果记录
    
50.         // 如果单个map数量太大超过阀值，需要进行清理
    
51.         //begin use pointer add by lxy 7-19
    
52. #ifndef NEW_ALLOC
    
53.         CEventMessage *groupMapEvent = this->event_alloc.allocate(1);
    
54.         groupMapEvent->metaData = NULL;
    
55.         char * body = this->mt_allocator.allocate(150);
    
56.         groupMapEvent->setBody(body);
    
57.         EVENT_HEAD_MESSAGE * head = this->head_alloc.allocate(1); //new EVENT_HEAD_MESSAGE;
    
58. #else
    
59.         CEventMessage *groupMapEvent = new CEventMessage();
    
60.         char * body = new char[150];
    
61.         groupMapEvent->setBody(body);
    
62.         EVENT_HEAD_MESSAGE * head = new EVENT_HEAD_MESSAGE(); //new EVENT_HEAD_MESSAGE;
    
63. #endif
    
64.         groupMapEvent->reset(head);
    
65.         groupMapEvent->setHead(head);
    
66.         groupMapEvent->getHead()->type = 2005; // 修改事件的类型为“统计事件对象”
    
67.         groupMapEvent->getHead()->data_len = 150;
    
68.         // 进行业务信息初始化
    
69.         groupMapEvent->setValueByName("times", 1);
    
70.         //  groupMapEvent->setValueByName("cause", msg->getValueByName("cause", -1));
    
71.         if (msg->getValueByName("Cause", -1) == 255)
    
72.             groupMapEvent->setValueByName("succtimes", 1);
    
73.         else
    
74.             groupMapEvent->setValueByName("succtimes", 0);
    
75.         string Btime_str = msg->getValueByName("Btime", "1970-01-01 00:00:00.000");
    
76.         long btime = Utils::millisecs(Btime_str.c_str());
    
77.         groupMapEvent->setValueByName("sttime", msg->getValueByName("Btime", "1970-01-01 00:00:00.000"));
    
78.         string RspTime_str = msg->getValueByName("RspTime", "1970-01-01 00:00:00.000");
    
79.         long rsptime = Utils::millisecs(RspTime_str.c_str());
    
80.         double delays = ((double) rsptime - (double) btime) / 1000;
    
81.         groupMapEvent->setValueByName("delays", delays);
    
82.         groupMapEvent->setValueByName("maxdelays", delays);
    
83.         groupMapEvent->setValueByName("mindelays", delays);
    
84.         //  groupMapEvent->setValueByName("tdrtype", msg->getValueByName("tdrtype", 0));
    
85.         if (btime != rsptime || msg->getValueByName("Tdrtype", (char) 0) != 1) {
    
86.             groupMapEvent->setValueByName("rsptimes", 1);
    
87.         } else {
    
88.             groupMapEvent->setValueByName("rsptimes", 0);
    
89.         }
    
90.         //统计维度的设置  add by lxy 7-21
    
91.         groupMapEvent->setValueByName("apn", msg->getValueByName("UchAPN", "*"));
    
92.         groupMapEvent->setValueByName("lacid", msg->getValueByName("Wlac", 0));
    
93.         groupMapEvent->setValueByName("sac", msg->getValueByName("Cell", 0));
    
94.         groupMapEvent->setValueByName("interfacetype", msg->getValueByName("InterfaceType", "*"));
    
95.         groupMapEvent->setValueByName("protocolid", msg->getValueByName("Protocoltype", "*"));
    
96.         groupMapEvent->setValueByName("eventtype", msg->getValueByName("Eventtype", 0));
    
97.         this->groupMap.insert(map<string, eventPtr, std::less<string>, __gnu_cxx::__mt_alloc<string> >::value_type(key, groupMapEvent));
    
98.     }
    
99. }

复制代码

下面是Event的代码 不好意思 有点长 我截取一点给大家 CEventMessage.h

1. class CEventMessage {
   
2. public:
   
3. 
   
4.     /*下面这些函数主要是根据类型找到配置文件然后在body的指定位置设置值*/
   
5.     int getValueByName(const string &targetname, const int &defval) const; //
   
6.     unsigned int getValueByName(const string &targetname, const unsigned int &defval) const;
   
7.     long getValueByName(const string &targetname, const long &defval) const;
   
8.     unsigned long getValueByName(const string &targetname, const unsigned long &defval) const;
   
9.     char getValueByName(const string &targetname, const char &defval) const;
   
10.     unsigned char getValueByName(const string &targetname, const unsigned char &defval) const;
    
11.     short getValueByName(const string &targetname, const short &defval) const;
    
12.     unsigned short getValueByName(const string &targetname, const unsigned short &defval) const;
    
13.     string getValueByName(const string &targetname, const string &defval) const;
    
14.     double getValueByName(const string &targetname, const double &defval) const;
    
15.     float getValueByName(const string &targetname, const float &defval) const;
    
16. 
    
17. 
    
18.     void setValueByName(const string &targetname, const int &val);
    
19.     void setValueByName(const string &targetname, const unsigned int &val);
    
20.     void setValueByName(const string &targetname, const long &val);
    
21.     void setValueByName(const string &targetname, const unsigned long &val);
    
22.     void setValueByName(const string &targetname, const short &val);
    
23.     void setValueByName(const string &targetname, const unsigned short &val);
    
24.     void setValueByName(const string &targetname, const char &val);
    
25.     void setValueByName(const string &targetname, const unsigned char &val);
    
26.     void setValueByName(const string &targetname, const float &val);
    
27.     void setValueByName(const string &targetname, const string &val);
    
28.     void setValueByName(const string &targetname, const double &val);
    
29. 
    
30. public:
    
31. 
    
32.     CEventMessage();
    
33.     CEventMessage(const unsigned short& type);
    
34.     CEventMessage(const CEventMessage& event);
    
35. 
    
36.     CEventMessage& operator=(const CEventMessage& event);
    
37. 
    
38.     virtual ~CEventMessage();
    
39. 
    
40.     char * getBytes(int &length);
    
41. 
    
42.     void setBody(char* body);
    
43. 
    
44.     char* getBody() const;
    
45. 
    
46.     void setHead(EVENT_HEAD_MESSAGE* head);
    
47. 
    
48.     EVENT_HEAD_MESSAGE* getHead() const;
    
49. 
    
50.     static void reset(EVENT_HEAD_MESSAGE *event_head_message) {
    
51.         event_head_message->pkg_len = 1;
    
52.         event_head_message->flag = 0;
    
53.         event_head_message->format = 0;
    
54.         event_head_message->ver_info = 1;
    
55.         event_head_message->peid = 0;
    
56.     }
    
57. 
    
58.     unsigned int getPrevPEId() const {
    
59.         if (head != NULL)
    
60.             return head->peid;
    
61.         return 0;
    
62.     }
    
63. 
    
64.     /**
    
65.      * ??socket?жs???????
    
66.      * @param conn
    
67.      * @return
    
68.      */
    
69. 
    
70. private:
    
71. 
    
72.     template<class T>
    
73.     const EVENT_TYPE_ENTY * getCheckedEventTypeEntity(const string&name)const;
    
74.     template<class T>
    
75.     const EVENT_TYPE_ENTY * getCheckedEventTypeEntityAllocBody(const string&name);
    
76.    
    
77.     void resetFields(const CEventMessage& event);
    
78.     string getStringForNum(EVENT_TYPE_ENTY *eventType)const;
    
79.     void setStringFormNum(EVENT_TYPE_ENTY *eventType, const string&str_val) const;
    
80.     CMetaData * getMetaData()const;    //元数据 就是配置文件 指示body中的数据是如何组成的
    
81.     typedef hash_map<string, string> TempValueMap;
    
82.     TempValueMap mytempvalmap;//用来存放临时值 用于丰富维度用
    
83. 
    
84. public:
    
85.     CMetaData * metaData;
    
86. private:
    
87. 
    
88.     EVENT_HEAD_MESSAGE *head;  //用于网络传输  head
    
89.     char *body;                //实际数据
    
90. };

复制代码

CEventMessage.cpp太长了就不放上来了 其实很简单  根据metaData找到配置，根据配置在body的指定位置设置值。