-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathQueue.cpp
More file actions
125 lines (103 loc) · 2.64 KB
/
Queue.cpp
File metadata and controls
125 lines (103 loc) · 2.64 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
//
// Queue.cpp
// MessageLoop
//
// Created by Lymons on 14-3-20.
// Copyright (c) 2014年 Lymons. All rights reserved.
//
#include "Queue.h"
#include "Message.h"
#include "SocketNotifier.h"
#include "Mutex.h"
#define PRIORITY_LEN 3
MMutex qMutex;
Queue::Queue()
{
PriorityManager::iterator iter = mPriorityMgr.begin();
for (; iter != mPriorityMgr.end(); ++ iter) {
delete iter->second;
}
}
Message* Queue::Next()
{
SocketNotifier* notifier = &SocketNotifier::Instance();
Message* message = Pop();
if (!message) {
notifier->StartMonitor();
message = Pop();
}
return message;
}
bool Queue::Push(Message* m)
{
MLock l(qMutex);
if ((m->mStatus & PRIORITY_HIGHT) == 1) {
int count = 0;
bool allow = true;
pair<bool, int>* prior = mPriorityMgr[m->mThreadID];
if (prior) {
count = prior->second;
allow = prior->first;
}
if (allow) {
mPriorityMessages.push(m);
if (++ count >= PRIORITY_LEN) {
// 达到最多容纳的数目限制后就不允许放入新的消息
allow = false;
}
if (prior == NULL) {
prior = new pair<bool, int>();
mPriorityMgr[m->mThreadID] = prior;
}
prior->first = allow;
prior->second = count;
} else {
return false;
}
} else {
mMessageQueue.push(m);
}
return true;
}
void Queue::Push(queue<Message*>& q)
{
size_t len = q.size();
for (int i=0; i < len; ++i) {
Message* m = q.front();
if (Push(m)) {
q.pop();
}
}
}
Message* Queue::Pop()
{
Message* message = NULL;
MLock l(qMutex);
if (!mPriorityMessages.empty()) {
message = mPriorityMessages.front();
mPriorityMessages.pop();
pair<bool, int>* prior = mPriorityMgr[message->mThreadID];
if (prior) {
// 只要是被处理，就必须等到所有线程的处理完成后才能放入新的消息
prior->first = false;
}
return message;
}
if (!mMessageQueue.empty()) {
message = mMessageQueue.top();
mMessageQueue.pop();
}
return message;
}
void Queue::Recycle(Message* m)
{
if ((m->mStatus & PRIORITY_HIGHT) == 1) {
MLock l(qMutex);
pair<bool, int>* prior = mPriorityMgr[m->mThreadID];
if (prior && --prior->second <= 0) {
delete prior;
mPriorityMgr.erase(m->mThreadID);
}
}
m->Recycle();
}