test poll.c

lib_fiber/c/src/hook/poll.c

@@ -81,6 +81,7 @@ static void read_callback(EVENT *ev, FILE_EVENT *fe)
 		pfd = ring_to_appl(iter, POLLFD, me);
 		if (pfd->pfd->events & POLLIN) {
 			handle_poll_read(ev, fe, pfd);
+			break;
 		}
 	}
 }
@@ -133,6 +134,7 @@ static void write_callback(EVENT *ev, FILE_EVENT *fe)
 		pfd = ring_to_appl(iter, POLLFD, me);
 		if (pfd->pfd->events & POLLOUT) {
 			handle_poll_write(ev, fe, pfd);
+			break;
 		}
 	}
 }
@@ -215,8 +217,6 @@ static void poll_event_clean(EVENT *ev, POLL_EVENT *pe)
 			pfd->fe->fiber_w = NULL;
 		}
 
-		ring_init(&pfd->fe->pfds);
-
 		// Unrefer the fe because we don't need it again.
 		//file_event_unrefer(pfd->fe);
 		pfd->fe      = NULL;

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

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

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

@@ -5,7 +5,8 @@ static void usage(const char* procname) {
 	printf("usage: %s -h[help]\r\n"
 		" -s ip:port, default: 127.0.0.1:9001\r\n"
 		" -c fiber_pool_count [default: 100] \r\n"
-		" -r timeout\r\n"
+		" -r read_timeout\r\n"
+		" -w write_timeout\r\n"
 		" -S [if in sync mode, default: false]\r\n"
 		, procname);
 }
@@ -13,9 +14,9 @@ static void usage(const char* procname) {
 int main(int argc, char* argv[]) {
 	acl::string addr("127.0.0.1:9001");
 	bool sync = false, use_unique = false;
-	int  ch, nfibers = 100;
+	int  ch, nfibers = 100, rtimeo = -1, wtimeo = -1;
 
-	while ((ch = getopt(argc, argv, "hs:Sc:r:U")) > 0) {
+	while ((ch = getopt(argc, argv, "hs:Sc:r:w:U")) > 0) {
 		switch (ch) {
 		case 'h':
 			usage(argv[0]);
@@ -32,6 +33,12 @@ int main(int argc, char* argv[]) {
 		case 'U':
 			use_unique = true;
 			break;
+		case 'r':
+			rtimeo = atoi(optarg);
+			break;
+		case 'w':
+			wtimeo = atoi(optarg);
+			break;
 		default:
 			break;
 		}
@@ -47,7 +54,7 @@ int main(int argc, char* argv[]) {
 		std::cout << "unique_ptr should be used for c++14" << std::endl;
 #endif
 	} else {
-		server_pool_run(addr, sync, nfibers);
+		server_pool_run(addr, sync, nfibers, rtimeo, wtimeo);
 	}
 
 	return 0;

lib_fiber/samples-c++1x/fiber_pool/server_pool.cpp

@@ -3,6 +3,7 @@
 #include <atomic>
 #include "server_pool.h"
 
+static int __rtimeo = 0, __wtimeo = 0;
 using shared_stream = std::shared_ptr<acl::socket_stream>;
 
 class socket_client {
@@ -49,13 +50,17 @@ private:
 
 using shared_message = std::shared_ptr<message>;
 
-void server_pool_run(const char* addr, bool sync, int nfibers) {
+void server_pool_run(const char* addr, bool sync, int nfibers,
+		int rtimeo, int wtimeo) {
 	acl::server_socket ss;
 	if (!ss.open(addr)) {
 		printf("listen %s error %s\r\n", addr, acl::last_serror());
 		return;
 	}
 
+	__rtimeo = rtimeo;
+	__wtimeo = wtimeo;
+
 	printf("listen on %s, fiber pool: %d\r\n", addr, nfibers);
 
 	acl::fiber_sbox2<shared_message> box;
@@ -70,6 +75,15 @@ void server_pool_run(const char* addr, bool sync, int nfibers) {
 				}
 				auto client = msg->get_client();
 				auto data = msg->get_data();
+
+				if (__wtimeo > 0) {
+					int fd = client->get_conn().sock_handle();
+					if (acl_write_wait(fd , __wtimeo) < 0) {
+						printf("write wait error\r\n");
+						break;
+					}
+				}
+
 				if (client->get_conn().write(data.c_str(), data.size()) == -1) {
 					printf("write error: %s\r\n", acl::last_serror());
 					break;
@@ -80,6 +94,7 @@ void server_pool_run(const char* addr, bool sync, int nfibers) {
 
 	std::atomic<long> nusers(0), nmsgs(0);
 
+#if 0
 	go[&nusers, &nmsgs] {
 		while (true) {
 			std::cout << "client count: " << nusers
@@ -87,6 +102,7 @@ void server_pool_run(const char* addr, bool sync, int nfibers) {
 			::sleep(1);
 		}
 	};
+#endif
 
 	go[&ss, &box, &nusers, &nmsgs, sync] {
 		while (true) {
@@ -103,6 +119,14 @@ void server_pool_run(const char* addr, bool sync, int nfibers) {
 			go[&box, &nmsgs, client, sync] {
 				char buf[4096];
 				while (true) {
+					if (__rtimeo > 0) {
+						int fd = client->get_conn().sock_handle();
+						if (acl_read_wait(fd, __rtimeo) < 0) {
+							printf("read wait error\r\n");
+							break;
+						}
+					}
+
 					int ret = client->get_conn().read(buf, sizeof(buf), false);
 					if (ret <= 0) {
 						break;

lib_fiber/samples-c++1x/fiber_pool/server_pool.h

@@ -1,4 +1,4 @@
 #pragma once
 
-void server_pool_run(const char* addr, bool sync, int nfibers);
+void server_pool_run(const char* addr, bool sync, int nfibers, int rtimeo, int wtimeo);
 void server_pool2_run(const char* addr, bool sync, int nfibers);