多線程OR多進(jìn)程-訪存密集型應(yīng)用

JohnBull

hy036630 發(fā)表于 2011-12-21 14:58
不好意思 可能我上面沒有說清楚
我用的是下面這個(gè)環(huán)形隊(duì)列  無鎖的實(shí)現(xiàn)一個(gè)進(jìn)程讀，一個(gè)進(jìn)程寫，沒有牽涉 ...

不妨設(shè)置一個(gè)計(jì)數(shù)器，在“push時(shí)滿”和"pop時(shí)空"的時(shí)候++下，統(tǒng)計(jì)下一分鐘發(fā)生多少次，再?zèng)Q定下一步怎辦

hy036630

這倒是個(gè)好主意
我試試 謝謝了{(lán):3_188:}

zylthinking

hy036630 發(fā)表于 2011-12-21 14:58
不好意思 可能我上面沒有說清楚
我用的是下面這個(gè)環(huán)形隊(duì)列  無鎖的實(shí)現(xiàn)一個(gè)進(jìn)程讀，一個(gè)進(jìn)程寫，沒有牽涉 ...

看代碼， 流水線中幾個(gè)線程都會(huì)處理同一份數(shù)據(jù)？
是不是這樣

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. }

復(fù)制代碼

然而 p 是相同的？

sonicling

用semaphore把隊(duì)列指針包起來，一來可以保護(hù)指針，二來可以及時(shí)喚醒等待線程。如果忙等的話，生產(chǎn)者線程更新了隊(duì)列，等到消費(fèi)者線程那邊發(fā)現(xiàn)的時(shí)候，黃花菜都涼了。

zylthinking

sonicling 發(fā)表于 2011-12-21 17:12
用semaphore把隊(duì)列指針包起來，一來可以保護(hù)指針，二來可以及時(shí)喚醒等待線程。如果忙等的話，生產(chǎn)者線程更新 ...

它很顯然是多核， 倒不是這個(gè)原因， 很有可能是多處理器緩存同步

sonicling

也可以不用semaphore，但是要及時(shí)喚醒等待線程！否則等待線程會(huì)等到下一次自然調(diào)度才得到通知。

sonicling

CPU間緩存同步會(huì)影響效率，但是不至于數(shù)量級(jí)的差別。我以前也做過無鎖循環(huán)隊(duì)列，不過是內(nèi)核級(jí)的，用atomic_t來表示位置，每次更新隊(duì)列，強(qiáng)制喚醒消費(fèi)線程，吞吐量上60w/s無壓力。

hy036630

的確是這樣的，我覺得也不應(yīng)該是忙等的問題，應(yīng)為機(jī)器是16線程的機(jī)器，按照道理至少有8個(gè)獨(dú)立線程同時(shí)跑，
緩存我開始以為是這樣的原因

不過設(shè)置了CPU親和了以后也沒有效果。

hy036630

我所謂的忙等是這個(gè)意思
從環(huán)形隊(duì)列取
while(NULL!=(event = queue.pop()));
doSomething();

插入到環(huán)形隊(duì)列里
while(queue.push(event));
doSomething();


在多核的機(jī)器上面，應(yīng)該不存在比較大的問題吧。

hy036630

sonicling 發(fā)表于 2011-12-21 17:31
CPU間緩存同步會(huì)影響效率，但是不至于數(shù)量級(jí)的差別。我以前也做過無鎖循環(huán)隊(duì)列，不過是內(nèi)核級(jí)的，用atomic_ ...

其實(shí)我覺得不是隊(duì)列慢了導(dǎo)致整體性能慢，
我覺得可能是增加流水線的長(zhǎng)度會(huì)導(dǎo)致處理數(shù)據(jù)的速度變慢。
我給大家貼一段處理的代碼吧。

1. string GroupPE::getEventKey(eventPtr msg) {
   
2.     vector<string> keys;
   
3.     keys.push_back("UchAPN"); // apn
   
4.     keys.push_back("Wlac"); // 位置區(qū)編號(hào)：lacid
   
5.     keys.push_back("Cell"); // 小區(qū)信息：sac
   
6.     keys.push_back("InterfaceType"); // 接口類型：interfacetype
   
7.     keys.push_back("Protocoltype"); // 協(xié)議類型：protocolid
   
8.     keys.push_back("Eventtype"); // 事件類型：eventtype
   
9.     string key = Utils::encoding(*msg, keys.begin(), keys.end(), 2000);
   
10.     return key;
    
11. }
    
12. 
    
13.    /*每個(gè)線程有一個(gè)統(tǒng)計(jì)map 用來存放統(tǒng)計(jì)數(shù)據(jù) 每次來一個(gè)event就根據(jù)getEventKey這個(gè)方法得到key，然后查看key是不是在統(tǒng)計(jì)map中，如果在統(tǒng)計(jì)map中，那么更新這個(gè)map的數(shù)據(jù)，如果沒有，那么new一個(gè)event，設(shè)置一些值，然后放到統(tǒng)計(jì)map中*/
    
14. void GroupPE::process(eventPtr msg) {//統(tǒng)計(jì)
    
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 { // 進(jìn)行新統(tǒng)計(jì)結(jié)果記錄
    
50.         // 如果單個(gè)map數(shù)量太大超過閥值，需要進(jìn)行清理
    
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; // 修改事件的類型為“統(tǒng)計(jì)事件對(duì)象”
    
67.         groupMapEvent->getHead()->data_len = 150;
    
68.         // 進(jìn)行業(yè)務(wù)信息初始化
    
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.         //統(tǒng)計(jì)維度的設(shè)置  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. }

復(fù)制代碼

下面是Event的代碼 不好意思 有點(diǎn)長(zhǎng) 我截取一點(diǎn)給大家 CEventMessage.h

1. class CEventMessage {
   
2. public:
   
3. 
   
4.     /*下面這些函數(shù)主要是根據(jù)類型找到配置文件然后在body的指定位置設(shè)置值*/
   
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;    //元數(shù)據(jù) 就是配置文件 指示body中的數(shù)據(jù)是如何組成的
    
81.     typedef hash_map<string, string> TempValueMap;
    
82.     TempValueMap mytempvalmap;//用來存放臨時(shí)值 用于豐富維度用
    
83. 
    
84. public:
    
85.     CMetaData * metaData;
    
86. private:
    
87. 
    
88.     EVENT_HEAD_MESSAGE *head;  //用于網(wǎng)絡(luò)傳輸  head
    
89.     char *body;                //實(shí)際數(shù)據(jù)
    
90. };

復(fù)制代碼

CEventMessage.cpp太長(zhǎng)了就不放上來了 其實(shí)很簡(jiǎn)單  根據(jù)metaData找到配置，根據(jù)配置在body的指定位置設(shè)置值。