optimize and test fiber_sbox

lib_fiber/cpp/include/fiber/fiber.hpp

@@ -265,8 +265,9 @@ public:
 	 * @param ctx {void*} 传递给协程执行函数的参数
 	 * @param size {size_t} 协程栈大小
 	 * @param share_stack {bool} 是否创建共享栈协程
+	 * @return {ACL_FIBER*}
 	 */
-	static void fiber_create(void (*fn)(ACL_FIBER*, void*),
+	static ACL_FIBER* fiber_create(void (*fn)(ACL_FIBER*, void*),
 			void* ctx, size_t size, bool share_stack = false);
 
 	/**

lib_fiber/cpp/include/fiber/fiber_sem.hpp

@@ -49,7 +49,9 @@ private:
 template<typename T>
 class fiber_sbox {
 public:
-	fiber_sbox(bool free_obj = true) : sem_(0), free_obj_(free_obj) {}
+	fiber_sbox(bool free_obj = true, bool async = true)
+	: sem_(0, async ? fiber_sem_t_async : fiber_sem_t_def)
+	, free_obj_(free_obj) {}
 
 	~fiber_sbox(void) { clear(free_obj_); }
 
@@ -58,10 +60,19 @@ public:
 		sem_.post();
 	}
 
-	T* pop(void) {
+	T* pop(bool* found = NULL) {
-		sem_.wait();
+		if (sem_.wait() < 0) {
+			if (found) {
+				*found = false;
+			}
+			return NULL;
+		}
+
 		T* t = sbox_.front();
 		sbox_.pop_front();
+		if (found) {
+			*found = true;
+		}
 		return t;
 	}
 
@@ -88,7 +99,9 @@ private:
 template<typename T>
 class fiber_sbox2 {
 public:
-	fiber_sbox2(void): sem_(0) {}
+	fiber_sbox2(bool async = true)
+	: sem_(0, async ? fiber_sem_t_async : fiber_sem_t_def) {}
+
 	~fiber_sbox2(void) {}
 
 	void push(T t) {
@@ -96,11 +109,14 @@ public:
 		sem_.post();
 	}
 
-	T pop(void) {
+	bool pop(T& t) {
-		sem_.wait();
+		if (sem_.wait() < 0) {
-		T t = sbox_.front();
+			return false;
+		}
+
+		t = sbox_.front();
 		sbox_.pop_front();
-		return t;
+		return true;
 	}
 
 	size_t size(void) const {

lib_fiber/cpp/include/fiber/go_fiber.hpp

@@ -35,9 +35,9 @@ public:
 	go_fiber(void) {}
 	go_fiber(size_t stack_size, bool on) : stack_size_(stack_size), stack_share_(on) {}
 
-	void operator > (std::function<void()> fn) {
+	ACL_FIBER* operator > (std::function<void()> fn) {
 		fiber_ctx* ctx = new fiber_ctx(fn);
-		fiber::fiber_create(fiber_main, (void*) ctx, stack_size_, stack_share_);
+		return fiber::fiber_create(fiber_main, (void*) ctx, stack_size_, stack_share_);
 	}
 
 	void operator < (std::function<void()> fn) {

lib_fiber/cpp/src/fiber.cpp

@@ -308,7 +308,7 @@ void fiber::stdout_open(bool on)
 	acl_fiber_msg_stdout_enable(on ? 1 : 0);
 }
 
-void fiber::fiber_create(void (*fn)(ACL_FIBER*, void*), void* ctx,
+ACL_FIBER* fiber::fiber_create(void (*fn)(ACL_FIBER*, void*), void* ctx,
 	size_t stack_size, bool share_stack /* false */)
 {
 	ACL_FIBER_ATTR attr;
@@ -316,7 +316,7 @@ void fiber::fiber_create(void (*fn)(ACL_FIBER*, void*), void* ctx,
 	acl_fiber_attr_setstacksize(&attr, stack_size);
 	acl_fiber_attr_setsharestack(&attr, share_stack ? 1 : 0);
 
-	acl_fiber_create2(&attr, fn, ctx);
+	return acl_fiber_create2(&attr, fn, ctx);
 }
 
 void fiber::stacktrace(const fiber& fb, std::vector<fiber_frame>& out, size_t max)

lib_fiber/samples-c++1x/fiber_sbox/Makefile

@@ -0,0 +1,3 @@
+include ../Makefile_cpp.in
+CFLAGS += -std=c++11
+PROG = fiber_sbox

lib_fiber/samples-c++1x/fiber_sbox/main.cpp

@@ -0,0 +1,112 @@
+#include "stdafx.h"
+#include <memory>
+#include <atomic>
+
+class message {
+public:
+	message(std::atomic<long>& nmsgs, int id)
+	: nmsgs_(nmsgs), id_(id) {}
+
+	~message(void) { --nmsgs_; }
+
+	int get_id(void) const {
+		return id_;
+	}
+
+private:
+	std::atomic<long>& nmsgs_;
+	int id_;
+};
+
+using shared_message = std::shared_ptr<message>;
+
+static void usage(const char* procname) {
+	printf("usage: %s -h[help]\r\n"
+		" -c fiber_pool_count [default: 5] \r\n"
+		" -n message_count [default: 100]\r\n"
+		" -S [if post in sync mode, default: async mode]\r\n"
+		, procname);
+}
+
+int main(int argc, char* argv[]) {
+	bool sync = false;
+	int  ch, nfibers = 5, count = 100;
+
+	while ((ch = getopt(argc, argv, "hc:n:S")) > 0) {
+		switch (ch) {
+		case 'h':
+			usage(argv[0]);
+			return 0;
+		case 'c':
+			nfibers = atoi(optarg);
+			break;
+		case 'n':
+			count = atoi(optarg);
+			break;
+		case 'S':
+			sync = true;
+			break;
+		default:
+			break;
+		}
+	}
+
+	acl::fiber::stdout_open(true);
+	acl::log::stdout_open(true);
+
+	acl::fiber_sbox2<shared_message> box(!sync);
+	std::vector<ACL_FIBER*> fibers;
+	std::atomic<int> nfibers_left(nfibers);
+
+	for (int i = 0; i < nfibers; i++) {
+		auto fb = go[&box, &fibers, &nfibers_left] {
+			while (true) {
+				shared_message msg;
+				if (!box.pop(msg)) {
+					std::cout << "POP end!" << std::endl;
+					break;
+				}
+
+				int id = msg->get_id();
+				std::cout << "fiber-" << acl::fiber::self()
+					<< ", id=" << id << std::endl;
+			}
+
+			std::cout << "fiber-" << acl::fiber::self()
+				<< " exited!" << std::endl;
+			--nfibers_left;
+		};
+
+		fibers.push_back(fb);
+	}
+
+	std::atomic<long> nmsgs(0);
+
+	go[&nmsgs, &nfibers_left] {
+		while (nfibers_left > 0) {
+			std::cout << "message count: " << nmsgs << std::endl;
+			acl::fiber::delay(500);
+		}
+		std::cout << "All consumers exited!" << std::endl;
+	};
+
+	go[&box, &fibers, &nmsgs, nfibers, count] {
+		for (int i = 0; i < count; i++) {
+			auto msg = std::make_shared<message>(nmsgs, i);
+			++nmsgs;
+			box.push(msg);
+			if (i > 0 && i % 10 == 0) {
+				::sleep(1);
+			}
+		}
+
+		for (auto fb : fibers) {
+			std::cout << "Begin kill fiber-"
+				<< acl_fiber_id(fb) << std::endl;
+			acl_fiber_kill(fb);
+		}
+	};
+
+	acl::fiber::schedule();
+	return 0;
+}

lib_fiber/samples-c++1x/fiber_sbox/stdafx.h

@@ -0,0 +1,19 @@
+// stdafx.h : 标准系统包含文件的包含文件，
+// 或是常用但不常更改的项目特定的包含文件
+//
+
+#pragma once
+
+
+//#include <iostream>
+//#include <tchar.h>
+
+// TODO: 在此处引用程序要求的附加头文件
+#include <assert.h>
+#include <iostream>
+
+#include "lib_acl.h" // just for getopt on Windows
+#include "acl_cpp/lib_acl.hpp"
+#include "fiber/libfiber.h"
+#include "fiber/libfiber.hpp"
+#include "fiber/go_fiber.hpp"