add tbox_array.hpp and tbox sample.

lib_acl_cpp/include/acl_cpp/lib_acl.hpp

@@ -29,6 +29,7 @@
 #include "stdlib/thread_pool.hpp"
 #include "stdlib/thread_queue.hpp"
 #include "stdlib/tbox.hpp"
+#include "stdlib/tbox_array.hpp"
 #include "stdlib/thread_mutex.hpp"
 #include "stdlib/thread_cond.hpp"
 #include "stdlib/scan_dir.hpp"

lib_acl_cpp/include/acl_cpp/stdlib/tbox.hpp

@@ -75,21 +75,34 @@ public:
 	/**
 	 * 发送消息对象
 	 * @param t {T*} 非空消息对象
+	 * @param notify_first {bool} 如果为 true，则先通知后解锁，否则先解锁
+	 *  后通知，注意二者的区别
 	 * @return {bool}
 	 */
-	bool push(T* t)
+	bool push(T* t, bool notify_first = true)
 	{
 		if (lock_.lock() == false) {
 			abort();
 		}
 		tbox_.push_back(t);
 		size_++;
-		if (lock_.unlock() == false) {
+
-			abort();
+		if (notify_first) {
-		}
+			if (cond_.notify() == false) {
-		if (cond_.notify() == false) {
+				abort();
-			abort();
+			}
+			if (lock_.unlock() == false) {
+				abort();
+			}
+		} else {
+			if (lock_.unlock() == false) {
+				abort();
+			}
+			if (cond_.notify() == false) {
+				abort();
+			}
 		}
+
 		return true;
 	}
 

lib_acl_cpp/include/acl_cpp/stdlib/tbox_array.hpp

@@ -0,0 +1,234 @@
+#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

lib_acl_cpp/samples/tbox/Makefile

@@ -0,0 +1,2 @@
+include ../Makefile.in
+PROG = tbox

lib_acl_cpp/samples/tbox/stdafx.cpp

@@ -0,0 +1,8 @@
+// stdafx.cpp : 只包括标准包含文件的源文件
+// master_threads.pch 将成为预编译头
+// stdafx.obj 将包含预编译类型信息
+
+#include "stdafx.h"
+
+// TODO: 在 STDAFX.H 中
+//引用任何所需的附加头文件，而不是在此文件中引用

lib_acl_cpp/samples/tbox/stdafx.h

@@ -0,0 +1,19 @@
+// stdafx.h : 标准系统包含文件的包含文件，
+// 或是常用但不常更改的项目特定的包含文件
+//
+
+#pragma once
+
+
+//#include <iostream>
+//#include <tchar.h>
+
+// TODO: 在此处引用程序要求的附加头文件
+
+#include "acl_cpp/lib_acl.hpp"
+#include "acl_cpp/stdlib/tbox_array.hpp"
+
+#ifdef	WIN32
+#define	snprintf _snprintf
+#endif
+

lib_acl_cpp/samples/tbox/tbox.cpp

@@ -0,0 +1,49 @@
+#include "stdafx.h"
+
+#if 1
+ #define TBOX	acl::tbox_array
+#else
+ #define TBOX	acl::tbox
+#endif
+
+class producer : public acl::thread
+{
+public:
+	producer(TBOX<int>& box, int max) : box_(box), max_(max) {}
+	~producer(void) {}
+
+protected:
+	void* run(void)
+	{
+		for (int i = 0; i < max_; i++) {
+			int* n = new int;
+			*n = i;
+			box_.push(n, true);
+		}
+
+		return NULL;
+	}
+
+private:
+	TBOX<int>& box_;
+	int max_;
+};
+
+int main(void)
+{
+	int max = 50000000;
+	TBOX<int> box;
+
+	producer thr(box, max);
+	thr.start();
+
+	for (int i = 0; i < max; i++) {
+		int* n = box.pop();
+		assert(*n == i);
+		delete n;
+	}
+
+	printf("All over, max=%d\r\n", max);
+	thr.wait();
+	return 0;
+}