Rewrite and fixed bug in pthread_patch.c for windows platform in fiber module.

Author: 郑树新

lib_fiber/c/src/common/pthread_patch.c

@@ -47,7 +47,7 @@ int pthread_key_create(pthread_key_t* key, void (*destructor)(void*))
     if (!key)
         return -1;
 
-    DWORD k = FlsAlloc((PFLS_CALLBACK_FUNCTION)destructor);
+    DWORD k = FlsAlloc((PFLS_KEY_FUNCTION)destructor);
 
     if (k == FLS_OUT_OF_INDEXES)
         return -1;
@@ -122,69 +122,108 @@ int pthread_once(pthread_once_t *once_control, void (*init_routine)(void))
 
 #include "common/htable.h"
 
-typedef struct FLS_KEY FLS_KEY;
-typedef struct TLS_KEY TLS_KEY;
-
-struct FLS_KEY {
-	HTABLE *keys;
-};
-
-struct TLS_KEY {
+typedef struct TLS_KEY {
 	pthread_key_t key;
 	void (*destructor)(void *);
-	FIFO *objs;
-};
+} TLS_KEY;
+
+typedef struct TLS_OBJ {
+	pthread_key_t key;
+	void *value;
+} TLS_OBJ;
 
-static pthread_key_t   __tls_key  = TLS_OUT_OF_INDEXES;
-static pthread_key_t   __fls_key  = FLS_OUT_OF_INDEXES;
+typedef struct FLS_KEY {
+	FIFO *locals; // Hold TLS_OBJ
+} FLS_KEY;
+
+typedef struct PROC_KEYS {
+	HTABLE *keys; // tkey => TLS_KEY
+} PROC_KEYS;
+
+static pthread_key_t __tls_key = TLS_OUT_OF_INDEXES;
+static pthread_key_t __fls_key = FLS_OUT_OF_INDEXES;
+static PROC_KEYS      *__pkeys = NULL;
+static pthread_mutex_t __lock;
 
 extern int acl_fiber_scheduled(void);
 
+static void hash_key(pthread_key_t key, char *buf, size_t n)
+{
+	assert(n > 10);
+	_snprintf(buf, n, "%d", key);
+	buf[n - 1] = '\0';
+}
+
 static void thread_exit(void *ctx)
 {
 	FLS_KEY *fkey = (FLS_KEY *) ctx;
-	void *ctx;
+	assert(fkey && fkey->locals);
+
+	// The first one is set in thread_init() in fiber.c, it should be called
+	// in the end, because the original fiber will be freed that it means
+	// the current thread will be end and the thread context will freed.
+	TLS_OBJ *first = fifo_pop_front(fkey->locals), *obj;
+	assert(first);
 
-	assert(fkey);
-	assert(fkey->keys);
+	char key[32];
 
-	ITER iter;
-	foreach(iter, fkey->keys) {
-		TLS_KEY *tkey = (TLS_KEY *) iter.data;
-		assert(tkey);
+	pthread_mutex_lock(&__lock);
 
-		if (tkey->destructor == NULL) {
+	while ((obj = fifo_pop_front(fkey->locals))) {
+		hash_key(obj->key, key, sizeof(key));
+		TLS_KEY *tkey = (TLS_KEY*) htable_find(__pkeys->keys, key);
+		if (tkey == NULL || tkey->destructor == NULL) {
 			continue;
 		}
+		tkey->destructor(obj->value);
+		mem_free(obj);
+	}
 
-		while ((ctx = tkey->objs->pop_front(tkey->objs))) {
-			tkey->destructor(ctx);
-		}
+	fifo_free(fkey->locals, NULL);
+	mem_free(fkey);
 
-		fifo_free(tkey->objs, NULL);
-	}
+	hash_key(first->key, key, sizeof(key));
+	TLS_KEY *tkey = (TLS_KEY*) htable_find(__pkeys->keys, key);
+	assert(tkey);
 
-	htable_free(tkey->keys, mem_free);
-	mem_free(tkey);
-}
+	pthread_mutex_unlock(&__lock);
+
+	void *value = first->value;
+	mem_free(first);
 
-/* 每个进程的唯一初始化函数 */
+	// Must the __fls_key's value to NULL to avoid the thread_exit
+	// to be triggered again for the current thread when the original
+	// fiber is deleted.
+	FlsSetValue(__fls_key, NULL);
+	tkey->destructor(value);
+}
 
 static void thread_once(void)
 {
 	__tls_key = TlsAlloc();
 	__fls_key = FlsAlloc(thread_exit);
+	__pkeys   = (PROC_KEYS *) mem_calloc(1, sizeof(PROC_KEYS));
+	__pkeys->keys = htable_create(10);
+	pthread_mutex_init(&__lock, NULL);
 }
 
-static void hash_key(pthread_key_t key, char *buf, size_t n)
+void pthread_fkey_create(void)
 {
-	assert(n > 10);
-	_snprintf(buf, n, "%d", key);
-	buf[n - 1] = '\0';
+	pthread_once(&__control_once, thread_once);
+
+	FLS_KEY *fkey = (FLS_KEY*) TlsGetValue(__tls_key);
+	if (fkey == NULL) {
+		assert(!acl_fiber_scheduled());
+		fkey = (FLS_KEY*) mem_calloc(sizeof(FLS_KEY), 1);
+		fkey->locals = fifo_new();
+		FlsSetValue(__fls_key, fkey);
+		TlsSetValue(__tls_key, fkey);
+	}
 }
 
 int pthread_key_create(pthread_key_t *key_ptr, void (*destructor)(void*))
 {
+	char kbuf[32];
 	pthread_once(&__control_once, thread_once);
 
 	assert(__tls_key != TLS_OUT_OF_INDEXES);
@@ -193,73 +232,72 @@ int pthread_key_create(pthread_key_t *key_ptr, void (*destructor)(void*))
 	if (*key_ptr <= 0 && *key_ptr != TLS_OUT_OF_INDEXES) {
 		*key_ptr = TlsAlloc();
 		assert(*key_ptr != TLS_OUT_OF_INDEXES);
+	} else {
+		hash_key(*key_ptr, kbuf, sizeof(kbuf));
+		pthread_mutex_lock(&__lock);
+		if (htable_find(__pkeys->keys, kbuf)) {
+			pthread_mutex_unlock(&__lock);
+			return 0;
+		}
+		*key_ptr = TlsAlloc();
+		assert(*key_ptr != TLS_OUT_OF_INDEXES);
 	}
 
-	TLS_KEY *tkey    = (TLS_KEY*) mem_calloc(sizeof(TLS_KEY), 1);
+	TLS_KEY *tkey = (TLS_KEY*) mem_calloc(sizeof(TLS_KEY), 1);
 	tkey->key        = *key_ptr;
 	tkey->destructor = destructor;
-	tkey->objs       = fifo_new();
 
-	char kbuf[32];
 	hash_key(*key_ptr, kbuf, sizeof(kbuf));
-	FLS_KEY *fkey;
-
-	if ((fkey = (FLS_KEY *) TlsGetValue(__tls_key)) == NULL) {
-		if (acl_fiber_scheduled()) {
-			msg_fatal("%s(%d): should be in thread mode",
-				__FUNCTION__, __LINE__);
-		}
 
-		fkey = (FLS_KEY *) mem_calloc(sizeof(FLS_KEY), 1);
-		fkey->keys = htable_create(10);
-
-		TlsSetValue(__tls_key, fkey);
-		FlsSetValue(__fls_key, fkey);
-	}
-
-	htable_enter(fkey->keys, kbuf, tkey);
+	pthread_mutex_lock(&__lock);
+	htable_enter(__pkeys->keys, kbuf, tkey);
+	pthread_mutex_unlock(&__lock);
 	return 0;
 }
 
-void *pthread_getspecific(pthread_key_t key)
-{
-	return TlsGetValue(key);
-}
-
 int pthread_setspecific(pthread_key_t key, void *value)
 {
 	if (key < 0 || key == TLS_OUT_OF_INDEXES) {
 		msg_error("%s(%d): key(%d) invalid", __FUNCTION__, __LINE__, key);
 		return EINVAL;
 	}
 
-	FLS_KEY *fkey = (FLS_KEY *) TlsGetValue(__tls_key);
-	if (fkey == NULL) {
-		msg_error("%s(%d): no FLS_KEY for __tls_key=%d",
-			__FUNCTION__, __LINE__, __fls_key);
-		return EINVAL;
-	}
+	assert(__tls_key != TLS_OUT_OF_INDEXES);
+	assert(__pkeys);
 
 	char kbuf[32];
 	hash_key(key, kbuf, sizeof(kbuf));
 
-	TLS_KEY *tkey = (TLS_KEY *) htable_find(fkey->keys, kbuf);
+	pthread_mutex_lock(&__lock);
+	TLS_KEY *tkey = (TLS_KEY*) htable_find(__pkeys->keys, kbuf);
+	pthread_mutex_unlock(&__lock);
+
 	if (tkey == NULL) {
-		msg_error("%s(%d): no TLS_KEY for key=%d, __tls_key=%d",
-			__FUNCTION__, __LINE__, key, __tls_key);
+		msg_error("%s(%d): no TLS_KEY for key=%d", __FUNCTION__, __LINE__, key);
 		return EINVAL;
 	}
 
+	FLS_KEY *fkey = (FLS_KEY*) TlsGetValue(__tls_key);
+	assert(fkey && fkey->locals);
+	TLS_OBJ *obj = (TLS_OBJ*) mem_calloc(1, sizeof(TLS_OBJ));
+	obj->key = key;
+	obj->value = value;
+	fkey->locals->push_back(fkey->locals, obj);
+
 	if (!TlsSetValue(key, value)) {
 		msg_error("%s(%d): TlsSetValue(key=%d) error(%s)",
 			__FUNCTION__, __LINE__, key, last_serror());
 		return -1;
 	}
 
-	tkey->objs->push_back(tkey->objs, value);
 	return 0;
 }
 
+void *pthread_getspecific(pthread_key_t key)
+{
+	return TlsGetValue(key);
+}
+
 #endif /* USE_FLS */
 
 /* Free the mutex */

lib_fiber/c/src/common/pthread_patch.h

@@ -30,6 +30,7 @@ struct pthread_mutexattr_t {
 
 FIBER_API int pthread_once(pthread_once_t *once_control, void (*init_routine)(void));
 int pthread_key_create(pthread_key_t *key_ptr, void (*destructor)(void*));
+void pthread_fkey_create(void);
 
 void *pthread_getspecific(pthread_key_t key);
 int pthread_setspecific(pthread_key_t key, void *value);

lib_fiber/c/src/fiber.c

@@ -120,6 +120,7 @@ static void thread_free(THREAD *tf)
 	fiber = tf->original;
 	mem_free(tf);
 
+	// Make sure the original fiber is the last one to be freed.
 	fiber_real_free(fiber);
 }
 
@@ -182,6 +183,8 @@ static void thread_init(void)
 	ring_init(&local->dead);
 
 #ifdef THREAD_LOCAL_DYNAMIC
+	pthread_fkey_create();
+
 	if (pthread_setspecific(__fiber_key, local) != 0) {
 		printf("pthread_setspecific error!\r\n");
 		abort();
@@ -845,13 +848,16 @@ void acl_fiber_schedule_with(int event_mode)
 	acl_fiber_schedule();
 }
 
-#if defined(_WIN32) || defined(_WIN64)
+#ifdef _TEST
+# if defined(_WIN32) || defined(_WIN64)
 static pthread_key_t dummy_key;
 
 static void thread_exit_dummy(void* ctx fiber_unused)
 {
-	msg_info("%s(%d): schedule end now, key=%d\r\n", __FUNCTION__, __LINE__, dummy_key);
+	msg_info("%s(%d): thread=%ld, schedule end now, key=%d\r\n",
+		__FUNCTION__, __LINE__, thread_self(), dummy_key);
 }
+# endif
 #endif
 
 void acl_fiber_schedule(void)
@@ -867,11 +873,15 @@ void acl_fiber_schedule(void)
 	set_time_metric(1000);
 #endif
 
-#if defined(_WIN32) || defined(_WIN64)
+#ifdef _TEST
+# if defined(_WIN32) || defined(_WIN64)
 	// On windows, the __tls_key in pthread_key_create() must be created in no
 	// fiber mode, because the destructor set in FlsAlloc must in no fiber mode.
 	// See: https://github.com/acl-dev/acl/issues/363
+	pthread_fkey_create();
 	pthread_key_create(&dummy_key, thread_exit_dummy);
+	pthread_setspecific(dummy_key, NULL);
+# endif
 #endif
 
 	fiber_check();
@@ -897,7 +907,10 @@ void acl_fiber_schedule(void)
 	}
 
 	/* Release dead fiber */
-	while ((head = ring_pop_head(&__thread_local->dead)) != NULL) {
+	THREAD *local = __thread_local;
+	assert(local);
+
+	while ((head = ring_pop_head(&local->dead)) != NULL) {
 		fiber = RING_TO_APPL(head, ACL_FIBER, me);
 		fiber_free(fiber);
 	}

lib_fiber/c/src/fiber/fiber_win.c

@@ -56,8 +56,9 @@ void acl_fiber_stackfree(ACL_FIBER_STACK *stack)
 void fiber_real_free(ACL_FIBER *fiber)
 {
 	FIBER_WIN *fb = (FIBER_WIN *) fiber;
-	DeleteFiber(fb->context);
+	LPVOID ctx = fb->context;
 	stack_free(fb);
+	DeleteFiber(ctx);
 }
 
 void fiber_real_swap(ACL_FIBER *from, ACL_FIBER *to)

lib_fiber/c/src/fiber_io.c

@@ -316,7 +316,6 @@ static void fiber_io_loop(ACL_FIBER *self fiber_unused, void *ctx)
 
 	// Don't set ev_fiber NULL here, using fiber_io_clear() to set it NULL
 	// in acl_fiber_schedule() after scheduling finished.
-	// 
 }
 
 void fiber_io_clear(void)

lib_fiber/samples-c++1x/httpc/httpc.cpp

@@ -51,6 +51,7 @@ static void usage(const char* procname) {
 
 static void fiber_main(ACL_FIBER *, void*) {
 		printf("hello world!\r\n");
+		acl::fiber::delay(100);
 }
 
 static void test() {
@@ -68,7 +69,7 @@ int main(int argc, char *argv[]) {
 	acl::acl_cpp_init();
 	acl::log::stdout_open(true);
 
-	test(); return 0;
+	test();
 
 	while ((ch = getopt(argc, argv, "he:s:c:n:")) > 0) {
 		switch (ch) {