mirror of
https://gitee.com/acl-dev/acl.git
synced 2024-12-15 01:10:52 +08:00
235 lines
4.5 KiB
C++
235 lines
4.5 KiB
C++
#pragma once
|
||
#include "../acl_cpp_define.hpp"
|
||
#include <list>
|
||
#include <stdlib.h>
|
||
#include <string.h>
|
||
#include "thread_mutex.hpp"
|
||
#include "thread_cond.hpp"
|
||
#include "noncopyable.hpp"
|
||
|
||
namespace acl
|
||
{
|
||
|
||
/**
|
||
* 用于线程之间的消息通信,通过线程条件变量及线程锁实现
|
||
*
|
||
* 示例:
|
||
*
|
||
* class myobj
|
||
* {
|
||
* public:
|
||
* myobj(void) {}
|
||
* ~myobj(void) {}
|
||
*
|
||
* void test(void) { printf("hello world\r\n"); }
|
||
* };
|
||
*
|
||
* acl::tbox_array<myobj> tbox;
|
||
*
|
||
* void thread_producer(void)
|
||
* {
|
||
* myobj* o = new myobj;
|
||
* tbox.push(o);
|
||
* }
|
||
*
|
||
* void thread_consumer(void)
|
||
* {
|
||
* myobj* o = tbox.pop();
|
||
* o->test();
|
||
* delete o;
|
||
* }
|
||
*/
|
||
|
||
template<typename T>
|
||
class tbox_array : public noncopyable
|
||
{
|
||
public:
|
||
/**
|
||
* 构造方法
|
||
* @param free_obj {bool} 当 tbox_array 销毁时,是否自动检查并释放
|
||
* 未被消费的动态对象
|
||
*/
|
||
tbox_array(bool free_obj = true)
|
||
: capacity_(10000)
|
||
, off_curr_(0)
|
||
, off_next_(0)
|
||
, waiters_(0)
|
||
, free_obj_(free_obj)
|
||
, cond_(&lock_)
|
||
{
|
||
array_ = (T**) malloc(sizeof(T*) * capacity_);
|
||
}
|
||
|
||
~tbox_array(void)
|
||
{
|
||
clear(free_obj_);
|
||
free(array_);
|
||
}
|
||
|
||
/**
|
||
* 清理消息队列中未被消费的消息对象
|
||
* @param free_obj {bool} 释放调用 delete 方法删除消息对象
|
||
*/
|
||
void clear(bool free_obj = false)
|
||
{
|
||
if (free_obj) {
|
||
for (size_t i = off_curr_; i < off_next_; i++) {
|
||
delete array_[i];
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 发送消息对象,本过程是添加完对象后先解锁后通知
|
||
* @param t {T*} 非空消息对象
|
||
* @param notify_first {bool} 如果为 true,则先通知后解锁,否则先解锁
|
||
* 后通知,注意二者的区别
|
||
* @return {bool}
|
||
*/
|
||
bool push(T* t, bool notify_first = false)
|
||
{
|
||
if (lock_.lock() == false) {
|
||
abort();
|
||
}
|
||
|
||
if (off_next_ == capacity_) {
|
||
if (off_curr_ >= 10000) {
|
||
#if 1
|
||
size_t n = 0;
|
||
for (size_t i = off_curr_; i < off_next_; i++) {
|
||
array_[n++] = array_[i];
|
||
}
|
||
#else
|
||
memmove(array_, array_ + off_curr_,
|
||
(off_next_ - off_curr_) * sizeof(T*));
|
||
#endif
|
||
|
||
off_next_ -= off_curr_;
|
||
off_curr_ = 0;
|
||
} else {
|
||
capacity_ += 10000;
|
||
array_ = (T**) realloc(array_, sizeof(T*) * capacity_);
|
||
}
|
||
}
|
||
array_[off_next_++] = t;
|
||
|
||
if (notify_first) {
|
||
if (cond_.notify() == false) {
|
||
abort();
|
||
}
|
||
if (lock_.unlock() == false) {
|
||
abort();
|
||
}
|
||
} else {
|
||
if (lock_.unlock() == false) {
|
||
abort();
|
||
}
|
||
if (cond_.notify() == false) {
|
||
abort();
|
||
}
|
||
}
|
||
|
||
return true;
|
||
}
|
||
|
||
/**
|
||
* 接收消息对象
|
||
* @param wait_ms {int} >= 0 时设置等待超时时间(毫秒级别),
|
||
* 否则永远等待直到读到消息对象或出错
|
||
* @param found {bool*} 非空时用来存放是否获得了一个消息对象,主要用在
|
||
* 当允许传递空对象时的检查
|
||
* @return {T*} 非 NULL 表示获得一个消息对象,返回 NULL 时得需要做进一
|
||
* 步检查,生产者如果 push 了一个空对象(NULL),则消费者也会获得 NULL,
|
||
* 但此时仍然认为获得了一个消息对象,只不过为空对象;如果 wait_ms 参数
|
||
* 为 -1 时返回 NULL 依然认为获得了一个空消息对象,如果 wait_ms 大于
|
||
* 等于 0 时返回 NULL,则应该检查 found 参数的值为 true 还是 false 来
|
||
* 判断是否获得了一个空消息对象
|
||
*/
|
||
T* pop(int wait_ms = -1, bool* found = NULL)
|
||
{
|
||
long long n = ((long long) wait_ms) * 1000;
|
||
bool found_flag;
|
||
|
||
if (lock_.lock() == false) {
|
||
abort();
|
||
}
|
||
while (true) {
|
||
T* t = peek(found_flag);
|
||
if (found_flag) {
|
||
if (lock_.unlock() == false) {
|
||
abort();
|
||
}
|
||
if (found) {
|
||
*found = found_flag;
|
||
}
|
||
return t;
|
||
}
|
||
|
||
// 注意调用顺序,必须先调用 wait 再判断 wait_ms
|
||
waiters_++;
|
||
if (!cond_.wait(n, true) && wait_ms >= 0) {
|
||
waiters_--;
|
||
if (lock_.unlock() == false) {
|
||
abort();
|
||
}
|
||
if (found) {
|
||
*found = false;
|
||
}
|
||
return NULL;
|
||
}
|
||
waiters_--;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 返回当前存在于消息队列中的消息数量
|
||
* @return {size_t}
|
||
*/
|
||
size_t size(void) const
|
||
{
|
||
return off_next_ - off_curr_;
|
||
}
|
||
|
||
public:
|
||
void lock(void)
|
||
{
|
||
if (lock_.lock() == false) {
|
||
abort();
|
||
}
|
||
}
|
||
|
||
void unlock(void)
|
||
{
|
||
if (lock_.unlock() == false) {
|
||
abort();
|
||
}
|
||
}
|
||
|
||
private:
|
||
T** array_;
|
||
size_t capacity_;
|
||
size_t off_curr_;
|
||
size_t off_next_;
|
||
size_t waiters_;
|
||
bool free_obj_;
|
||
thread_mutex lock_;
|
||
thread_cond cond_;
|
||
|
||
T* peek(bool& found_flag)
|
||
{
|
||
if (off_curr_ == off_next_) {
|
||
found_flag = false;
|
||
if (off_curr_ > 0) {
|
||
off_curr_ = off_next_ = 0;
|
||
}
|
||
return NULL;
|
||
}
|
||
|
||
found_flag = true;
|
||
T* t = array_[off_curr_++];
|
||
return t;
|
||
}
|
||
};
|
||
|
||
} // namespace acl
|