关于ACE_Task内部消息队列问题

peakzhang

ACE_Task对象中集成了消息队列，对于任务间通讯提供了很容易实现的框架，但我有一点不明白的地方，如下段生产者-消费者模型代码：

2. class Producer : public ACE_Task<ACE_MT_SYNCH>
3. {
4. public:
5.     int svc(void)
6.     {
7.       ACE_Message_Block *mb = 0;
8.       // 初始化mb，此处省略
9.       consumer_.putq(mb);
10.       ACE_OS::sleep(1);  
11.     }
12. private:
13.         Consumer *consumer_;
14. }
16. class Consumer : public ACE_Task<ACE_MT_SYNCH>
17. {
18. public:
19.     int svc(void)
20.     {
21.       ACE_Message_Block *mb = 0;
22.       getq(mb);
23.   
24.       ACE_OS::sleep(60);  // 处理消息，此处处理消息花费时间长
25.       mb->release();
26.     }
27. }

复制代码

Producer产生一条数据后，立即发送给Consumer，但是对于Consumer来说，处理消息速度比Producer产生消息速度慢，所以Consumer需要将消息缓存在消息队列当中（意思是：Producer发送给Consumer 60条消息的时间里，Consumer 只能处理1条，剩下的59条需要缓存在队列里），但是不知为何在ACE_Task中消息无法缓存！不知是不是我的用法不正确，请坛子的各位DX赐教。

peakzhang

消息队列默认的水位尺寸只有16k，你推入的消息有多大？

peakzhang

这段测试的mb=“this is a test”,远远小与16k，现在的问题是Consumer队列中的消息为什么没有缓存？

peakzhang

全部代码如下：

2. #include "ace/OS.h"
3. #include "ace/Task.h"
4. #include "ace/Message_Block.h"
6. //The Consumer Task.
7. class Consumer : public ACE_Task<ACE_MT_SYNCH>
8. {
9. public:
10. int open(void*)
11. {
12.   ACE_DEBUG((LM_DEBUG, "(%t) Producer task opened \n"));  
13.   this->activate();
15.   return 0;
16. }
18. //The Service Processing routine
19. int svc(void)
20. {
21.   int ret = 0;
23.   //Get ready to receive message from Producer
24.   ACE_Message_Block *mb = 0;  
25.   while (1)
26.   {   
27.    ret = this->getq(mb);
28.    if (ret < 0)
29.     ACE_DEBUG((LM_DEBUG, "(%t)error.\n"));
31.    ProcessTask(mb);
33.    mb->release();
34.   }
36.   return 0;
37. }
39. void ProcessTask(ACE_Message_Block *mb)
40. {
41.   char *pBuf = mb->rd_ptr();
42.   ACE_DEBUG((LM_DEBUG, "(%t)Got message: '%s' from remote task\n",pBuf));
44.   ACE_OS::sleep(10);
45. }
47. int close(u_long)
48. {
49.   ACE_DEBUG((LM_DEBUG,"Consumer closes down \n"));
50.   return 0;
51. }
52. };
54. class Producer : public ACE_Task<ACE_MT_SYNCH>
55. {
56. public:
57. Producer(Consumer * consumer) : consumer_(consumer), count_(0)
58. {  
59. }
61. int open(void*)
62. {
63.   ACE_DEBUG((LM_DEBUG, "(%t) Producer task opened \n"));
64.   
65.   this->activate();
66.   return 0;
67. }
69. //The Service Processing routine
70. int svc(void)
71. {
72.   while(count_ < 21)
73.   {
74.    char data[32] = {0};
75.    ACE_OS::sprintf(data, "data form Producer, NO=%d.", data_);   
77.    ACE_NEW_RETURN(mb_, ACE_Message_Block(32, ACE_Message_Block::MB_DATA, 0, data), -1);
78.    mb_->wr_ptr(32);
80.    //Send message to consumer
81.    ACE_DEBUG((LM_DEBUG, "(%t)Sending message: %d to remote task\n", data_));
82.    consumer_->putq(mb_);   
84.    //Go to sleep for a sec.
85.    ACE_OS::sleep(1);
86.    count_++;
87.   }
89.   return 0;
90. }
92. int close(u_long)
93. {  
94.   ACE_DEBUG((LM_DEBUG, "Producer closes down \n"));  
95.   return 0;
96. }
98. private:
99. int   count_;
100. Consumer *consumer_;
101. ACE_Message_Block * mb_;
102. };
105. int main(int argc, char * argv[])
106. {
107. Consumer *consumer = new Consumer;
108. Producer *producer = new Producer(consumer);
110. consumer->open(0);
111. producer->open(0);
113. //Wait for all the tasks to exit.
114. ACE_Thread_Manager::instance()->wait();
116. if (consumer)
117. {
118.   delete consumer;
119.   consumer = NULL;
120. }
122. if (producer)
123. {
124.   delete producer;
125.   producer = NULL;
126. }
128. return 0;
129. }

复制代码

peakzhang

问题找到了，原来还是内存问题，发送到消息队列中的数据，应该事先动态分配内存空间

Producer中svc() char data[32] = {0}，data为局部变量每次分配的内存空间都是相同的，所以运行例子看到Consumer除了第一次收到的正常数据外，其余的数据都是相同的，其实不是缓存区没有缓存，而是缓存的数据都是指向同一个地址空间的数据；若data该成动态分配，即：char *data = new char[32]则运行正常。

看来还是对ACE消息队列的原理理解不深啊，学习ing...

peakzhang

但是不知为何在ACE_Task中消息无法缓存！????
How do u know?
Can you acctually run this code? it looks buggy.
e.g.
consumer_->putq(mb_->clone());//data is in stack.without copy you would end up with core dump.

peakzhang

现在发现问题，不是“无法缓存”引起的。

PS: My en is so so poor, please use Chinese, thank u!

peakzhang

可以认为几乎大部分C++的问题，都和内存管理相关。

renxianfu

学习了
我也碰到这个问题
呵呵