Fixed bugs for fiber_mutex and fiber_cond module that the FILE_EVENT maybe freed when using it.

Author: zhengshuxin

lib_fiber/c/src/fiber_io.c

@@ -863,8 +863,16 @@ FILE_EVENT *fiber_file_cache_get(socket_t fd)
 
 void fiber_file_cache_put(FILE_EVENT *fe)
 {
-	fiber_file_del(fe, fe->fd);
-	fe->fd = INVALID_SOCKET;
+	// If the fe is being refered by more than one fibers, don't put it
+	// back to cache or free it, because the fe is still aliving now.
+	if (fe->refer > 1) {
+		return;
+	}
+
+	if (fe->fd != INVALID_SOCKET) {
+		fiber_file_del(fe, fe->fd);
+		fe->fd = INVALID_SOCKET;
+	}
 
 	if (array_size(__thread_fiber->cache) < __thread_fiber->cache_max) {
 		if (!(fe->status & STATUS_BUFFED)) {

lib_fiber/c/src/sync/sync_timer.c

@@ -179,8 +179,8 @@ void sync_timer_wakeup(SYNC_TIMER *timer, SYNC_OBJ *obj)
 				fe->mbox_wsem = acl_fiber_sem_create(1);
 			}
 		} else if (fe->mbox_wsem == NULL) {
-			msg_fatal("%s(%d): fe's mbox_wsem NULL, out=%d",
-				 __FUNCTION__, __LINE__, (int) out);
+			msg_fatal("%s(%d): mbox_wsem NULL, out=%d, fd=%d, refer=%d",
+				__FUNCTION__, __LINE__, (int) out, fe->fd, fe->refer);
 		}
 
 		file_event_refer(fe);
@@ -190,11 +190,29 @@ void sync_timer_wakeup(SYNC_TIMER *timer, SYNC_OBJ *obj)
 		acl_fiber_sem_wait(fe->mbox_wsem);
 
 		fe->mask |= EVENT_SYSIO;
+
+		// The fe maybe be used again in mbox_send->acl_fiber_write
+		// ->fiber_file_open->fiber_file_get.
 		mbox_send(timer->box, msg);
 
+#ifdef	DEBUG
+		FILE_EVENT *f = fiber_file_get(out);
+		if (f == NULL) {
+			msg_fatal("%s(%d): fiber_file_get NULL, fe=%p, out=%d",
+				__FUNCTION__, __LINE__, fe, out);
+		} else if (f->mbox_wsem == NULL) {
+			msg_fatal("%s(%d): mbox_wsem NULL, f=%p, fe=%p, out=%d",
+				__FUNCTION__, __LINE__, f, fe, out);
+		} else if (f != fe) {
+			msg_fatal("%s(%d): invalid f=%p, fe=%p, out=%d",
+				__FUNCTION__, __LINE__, f, fe, out);
+		}
+#endif
+
 		if (acl_fiber_sem_post(fe->mbox_wsem) == 1) {
 			fiber_file_cache_put(fe);
 		}
+
 		file_event_unrefer(fe);
 	} else {
 		// If the current notifier is a fiber in the same thread with

lib_fiber/c/src/sync/sync_waiter.c

@@ -124,8 +124,8 @@ void sync_waiter_wakeup(SYNC_WAITER *waiter, ACL_FIBER *fb)
 				fe->mbox_wsem = acl_fiber_sem_create(1);
 			}
 		} else if (fe->mbox_wsem == NULL) {
-			msg_fatal("%s(%d): fe's mbox_wsem NULL, out=%d",
-				  __FUNCTION__, __LINE__, (int) out);
+			msg_fatal("%s(%d): mbox_wsem NULL, out=%d, fd=%d, refer=%d",
+				__FUNCTION__, __LINE__, (int) out, fe->fd, fe->refer);
 		}
 
 		file_event_refer(fe);
@@ -139,10 +139,25 @@ void sync_waiter_wakeup(SYNC_WAITER *waiter, ACL_FIBER *fb)
 		// ->fiber_file_open->fiber_file_get.
 		mbox_send(waiter->box, fb);
 
+#ifdef	DEBUG
+		FILE_EVENT *f = fiber_file_get(out);
+		if (f == NULL) {
+			msg_fatal("%s(%d): fiber_file_get NULL, fe=%p, out=%d",
+				__FUNCTION__, __LINE__, fe, out);
+		} else if (f->mbox_wsem == NULL) {
+			msg_fatal("%s(%d): mbox_wsem NULL, f=%p, fe=%p, out=%d",
+				__FUNCTION__, __LINE__, f, fe, out);
+		} else if (f != fe) {
+			msg_fatal("%s(%d): invalid f=%p, fe=%p, out=%d",
+				__FUNCTION__, __LINE__, f, fe, out);
+		}
+#endif
+
 		// If no other fiber is suspended by the sem, then release it.
 		if (acl_fiber_sem_post(fe->mbox_wsem) == 1) {
 			fiber_file_cache_put(fe);
 		}
+
 		file_event_unrefer(fe);
 	} else {
 		// If the current notifier is a fiber in the same thread with

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

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

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

@@ -0,0 +1,55 @@
+#include "stdafx.h"
+#include <vector>
+#include <thread>
+#include <iostream>
+
+int main() {
+	std::vector<std::thread*> threads;
+	std::vector<acl::fiber_mutex*> locks;
+	std::vector<long long*> cnts;
+	size_t number = 2;
+
+	for (size_t i = 0; i < number; i++) {
+		auto* lock = new acl::fiber_mutex;
+		locks.push_back(lock);
+		auto* cnt = new long long;
+		*cnt = 0;
+		cnts.push_back(cnt);
+	}
+
+	acl::fiber::stdout_open(true);
+
+	for (size_t i = 0; i < 10; i++) {
+		auto* thr = new std::thread([&] {
+			for (size_t j = 0; j < 100; j++) {
+				go[&] {
+					acl::fiber::delay(5);
+					for (size_t m = 0; m < 100; m++) {
+						size_t k = m % number;
+						locks[k]->lock();
+						*cnts[k] += 1;
+						acl::fiber::yield();
+						locks[k]->unlock();
+					}
+				};
+			}
+
+			acl::fiber::schedule();
+		});
+		threads.push_back(thr);
+	}
+
+	for (auto it : threads) {
+		it->join();
+		delete it;
+	}
+
+	for (size_t i = 0; i < number; i++) {
+		printf("cnts[%zd]=%lld\r\n", i, *cnts[i]);
+		delete cnts[i];
+		delete locks[i];
+	}
+
+	printf("All over!\r\n");
+	return 0;
+}

lib_fiber/samples-c++1x/fiber_mutex/stdafx.cpp

@@ -0,0 +1 @@
+#include "stdafx.h"

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

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