ACE队列，入队失败

xiaokousky

ACE_Message_Queue<ACE_NULL_SYNCH> *mq_=new ACE_Message_Queue<ACE_NULL_SYNCH>

// 我定义下面入队函数
int put(MessageBlock* mb)
{
        ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, guard, lock_, -1);
        int rtn = mq_->enqueue_head(mb);
        if (rtn<0)
           printf("put MessageBlock enqueue_head rtn=%d", rtn);
         return rtn;
}

通常情况下不会出现问题
但有时会出现rtn=-1情况，即入队失败。请问这可能由哪些原因导致的？

winston

记得队列满就会失败。你把错误原因打印出来就知道了。lasterror()

xiaokousky

lasterror() 为 10035

xiaokousky

10035 这是什么错误啊
在哪里可以看到错误码对应的解释

winston

应该是SOCKET的错误代码。
10035—WSAEWOULDBLOCK
Resource temporarily unavailable. This error is most commonly returned on nonblocking sockets in which the requested operation cannot complete immediately. For example, calling connect on a nonblocking socket returns this error because the connection request cannot be completed immediately.

贴上比较完整的代码出来，否则不好分析。

xiaokousky

class MessageBlock : public ACE_Message_Block
{
public:
    MessageBlock(size_t size) : ACE_Message_Block(size) {};

    template <class T>
    int read_by_type(T& x)
    {
        return read_data((char*)&x, sizeof(T));
    };

    template <class T>
    int write_by_type(const T& x)
    {
        return write_data((char*)&x, sizeof(T));
    };

    int read_data(char* data, size_t len)
    {
        if (length() < len)
            return -1;
        memcpy(data, rd_ptr(), len);
        rd_ptr(len);
        return 0;
    };

    int write_data(const char* data, size_t len)
    {
        return copy(data, len);
    };
};

MessageBlock派生于ACE_Message_Block,封装了一下读写操作，这个应该不会产生问题。
由于代码比较多，贴上来可能不容易暴露出问题，我把我想的说一下吧。
整个逻辑可以看做是一个线程put，另外一个线程从队尾取走消息块
int put(MessageBlock* mb)
{
        ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, guard, lock_, -1);
        int rtn = mq_->enqueue_head(mb);
        if (rtn<0)
           printf("put MessageBlock enqueue_head rtn=%d eque_full=%d", rtn, mq_->isfull());
         return rtn;
}
打印队列始终为“满”。即mq_->isfull()总是返回1
我的消息块大小为16k，我看了下ACE_Message_Queue默认的低水位标和高水位标都为16K
当超过高水位标时就不入队了。这个时候没有入队的MessageBlock是不是要自己写代码回收？
理想状态是所有MessageBlock都能入队列，所以我在队列初始化的时候，调用
mq_->low_water_mark(16*16*1024);
mq_->high_water_mark(16*1024*1024);
把高、低水位标设置大很多。

但是还是会发现进入下面语句
if (rtn<0)
           printf("put MessageBlock enqueue_head rtn=%d eque_full=%d", rtn, mq_->isfull());
其中rtn=-1, eque_full=1

xiaokousky

问题终于解决了
我每个块大小为16K，设定低水位标为16M，高水位标32M。当超过高水位标，等待50ms再次入队，若入队任然不成功，再等待50ms入队，这样经过几次后，最后还是不能入队，就主动选择丢块回收内存。先前设定的等待时间太长，导致出现丢块的现象比较明显。现在OK了，目前还没有一次丢块。
mq_->low_water_mark(16*1024*1024);
mq_->high_water_mark(32*1024*1024);

非常感谢版主

winston

嗯。跟我猜测的差不多，队列水位的问题。

newzai

我看还是消费者那个线程的处理速度问题，这个是一个瓶颈。

winston

嗯。满了报错，是因为处理不够快，或者同步不对头。