Hello World! with this commit trunk will finally be making use of the gorgeous stuff sirius_black last implemented. we want to make sure everything related to this feature is working as intended so for now its being released under a optional define, after the features stability is confirmed the define will be removed along with the previous processings of the feature. By enabling BETA_THREAD_TEST in /src/config/core.h all your mysql logs, query_sql and query_logsql script functions will be handled by a different thread, therefore any slow queries won't have any effect (e.g. slow down) on the game server. Everyone is welcome on helping us test and debug the features, be aware however I DO NOT RECOMMEND YOU TO ENABLE THIS ON PRODUCTION SERVERS AS OF NOW (which is why while we're all testing it still is a optional feature).

git-svn-id: https://rathena.svn.sourceforge.net/svnroot/rathena/trunk@16588 54d463be-8e91-2dee-dedb-b68131a5f0ec

1 files changed, 293 insertions, 11 deletions

diff --git a/src/map/script.c b/src/map/script.c
index 4b8afdb24..7b13787ce 100644
--- a/src/map/script.c
+++ b/src/map/script.c
@@ -61,6 +61,13 @@
 #include <setjmp.h>
 #include <errno.h>
 
+#ifdef BETA_THREAD_TEST
+	#include "../common/atomic.h"
+	#include "../common/spinlock.h"
+	#include "../common/thread.h"
+	#include "../common/mutex.h"
+#endif
+
 
 ///////////////////////////////////////////////////////////////////////////////
 //## TODO possible enhancements: [FlavioJS]
@@ -306,6 +313,30 @@ extern script_function buildin_func[];
 
 static struct linkdb_node* sleep_db;// int oid -> struct script_state*
 
+#ifdef BETA_THREAD_TEST
+/**
+ * MySQL Query Slave
+ **/
+static SPIN_LOCK queryThreadLock;
+static rAthread queryThread = NULL;
+static ramutex	queryThreadMutex = NULL;
+static racond	queryThreadCond = NULL;
+static volatile int32 queryThreadTerminate = 0;
+
+struct queryThreadEntry {
+	bool ok;
+	bool type; /* main db or log db? */
+	struct script_state *st;
+};
+
+/* Ladies and Gentleman the Manager! */
+struct {
+	struct queryThreadEntry **entry;/* array of structs */
+	int count;
+	int timer;/* used to receive processed entries */
+} queryThreadData;
+#endif
+
 /*==========================================
  * ローカルプロトタイプ宣言 (必要な物のみ)
  *------------------------------------------*/
@@ -3006,6 +3037,7 @@ void script_free_state(struct script_state* st)
 	pop_stack(st, 0, st->stack->sp);
 	aFree(st->stack->stack_data);
 	aFree(st->stack);
+	st->stack = NULL;
 	st->pos = -1;
 	aFree(st);
 }
@@ -3634,8 +3666,8 @@ static void script_attach_state(struct script_state* st)
  *------------------------------------------*/
 void run_script_main(struct script_state *st)
 {
-	int cmdcount=script_config.check_cmdcount;
-	int gotocount=script_config.check_gotocount;
+	int cmdcount = script_config.check_cmdcount;
+	int gotocount = script_config.check_gotocount;
 	TBL_PC *sd;
 	struct script_stack *stack=st->stack;
 	struct npc_data *nd;
@@ -3933,8 +3965,175 @@ void script_setarray_pc(struct map_session_data* sd, const char* varname, uint8
 		refcache[0] = key;
 	}
 }
+#ifdef BETA_THREAD_TEST
+int buildin_query_sql_sub(struct script_state* st, Sql* handle);
+
+/* used to receive items the queryThread has already processed */
+int queryThread_timer(int tid, unsigned int tick, int id, intptr_t data) {
+	int i, cursor = 0;
+	bool allOk = true;
+	
+	EnterSpinLock(&queryThreadLock);
+	
+	for( i = 0; i < queryThreadData.count; i++ ) {
+		struct queryThreadEntry *entry = queryThreadData.entry[i];
+		
+		if( !entry->ok ) {
+			allOk = false;
+			continue;
+		}
+
+		run_script_main(entry->st);
+		
+		entry->st = NULL;/* empty entries */
+		aFree(entry);
+		queryThreadData.entry[i] = NULL;
+	}
+
+	
+	if( allOk ) {
+		/* cancel the repeating timer -- it'll re-create itself when necessary, dont need to remain looping */
+		delete_timer(queryThreadData.timer, queryThread_timer);
+		queryThreadData.timer = INVALID_TIMER;
+	}
+	
+	/* now lets clear the mess. */
+	for( i = 0; i < queryThreadData.count; i++ ) {
+		struct queryThreadEntry *entry = queryThreadData.entry[i];
+		if( entry == NULL )
+			continue;/* entry on hold */
+		
+		/* move */
+		memmove(&queryThreadData.entry[cursor], &queryThreadData.entry[i], sizeof(struct queryThreadEntry*));
+		
+		cursor++;
+	}
+	
+	queryThreadData.count = cursor;
+	
+	LeaveSpinLock(&queryThreadLock);
+	
+	return 0;
+}
+
+void queryThread_add(struct script_state *st, bool type) {
+	int idx = 0;
+	struct queryThreadEntry* entry = NULL;
+	
+	EnterSpinLock(&queryThreadLock);
+	
+	if( queryThreadData.count++ != 0 )
+		RECREATE(queryThreadData.entry, struct queryThreadEntry* , queryThreadData.count);
+	
+	idx = queryThreadData.count-1;
+	
+	CREATE(queryThreadData.entry[idx],struct queryThreadEntry,1);
+	
+	entry = queryThreadData.entry[idx];
+	
+	entry->st = st;
+	entry->ok = false;
+	entry->type = type;
+	if( queryThreadData.timer == INVALID_TIMER ) { /* start the receiver timer */
+		queryThreadData.timer = add_timer_interval(gettick() + 100, queryThread_timer, 0, 0, 100);
+	}
+	
+	LeaveSpinLock(&queryThreadLock);
+	
+	/* unlock the queryThread */
+	racond_signal(queryThreadCond);
+}
+/* adds a new log to the queue */
+void queryThread_log(char * entry, int length) {
+	int idx = logThreadData.count;
+	
+	EnterSpinLock(&queryThreadLock);
+	
+	if( logThreadData.count++ != 0 )
+		RECREATE(logThreadData.entry, char* , logThreadData.count);	
+	
+	CREATE(logThreadData.entry[idx], char, length + 1 );
+	safestrncpy(logThreadData.entry[idx], entry, length + 1 );
+	
+	LeaveSpinLock(&queryThreadLock);
+
+	/* unlock the queryThread */
+	racond_signal(queryThreadCond);
+}
+
+/* queryThread_main */
+static void *queryThread_main(void *x) {
+	Sql *queryThread_handle = Sql_Malloc();
+	int i;
+	
+	if ( SQL_ERROR == Sql_Connect(queryThread_handle, map_server_id, map_server_pw, map_server_ip, map_server_port, map_server_db) )
+		exit(EXIT_FAILURE);
+	
+	if( strlen(default_codepage) > 0 )
+		if ( SQL_ERROR == Sql_SetEncoding(queryThread_handle, default_codepage) )
+			Sql_ShowDebug(queryThread_handle);
 
+	if( log_config.sql_logs ) {
+		logmysql_handle = Sql_Malloc();
+		
+		if ( SQL_ERROR == Sql_Connect(logmysql_handle, log_db_id, log_db_pw, log_db_ip, log_db_port, log_db_db) )
+			exit(EXIT_FAILURE);
+		
+		if( strlen(default_codepage) > 0 )
+			if ( SQL_ERROR == Sql_SetEncoding(logmysql_handle, default_codepage) )
+				Sql_ShowDebug(logmysql_handle);
+	}
+	
+	while( 1 ) {
+		
+		if(queryThreadTerminate > 0)
+			break;
+				
+		EnterSpinLock(&queryThreadLock);
+		
+		/* mess with queryThreadData within the lock */
+		for( i = 0; i < queryThreadData.count; i++ ) {
+			struct queryThreadEntry *entry = queryThreadData.entry[i];
+			
+			if( entry->ok )
+				continue;
+			else if ( !entry->st || !entry->st->stack ) {
+				entry->ok = true;/* dispose */
+				continue;
+			}
+						
+			buildin_query_sql_sub(entry->st, entry->type ? logmysql_handle : queryThread_handle);
+			
+			entry->ok = true;/* we're done with this */
+		}
+		
+		/* also check for any logs in need to be sent */
+		if( log_config.sql_logs ) {
+			for( i = 0; i < logThreadData.count; i++ ) {
+				if( SQL_ERROR == Sql_Query(logmysql_handle, logThreadData.entry[i]) )
+					Sql_ShowDebug(logmysql_handle);
+				aFree(logThreadData.entry[i]);
+			}
+			logThreadData.count = 0;
+		}
+		
+		LeaveSpinLock(&queryThreadLock);
+		
+		ramutex_lock( queryThreadMutex );
+		racond_wait( queryThreadCond,	queryThreadMutex,  -1 );
+		ramutex_unlock( queryThreadMutex );
 
+	}
+		
+	Sql_Free(queryThread_handle);
+	
+	if( log_config.sql_logs ) {
+		Sql_Free(logmysql_handle);
+	}
+	
+	return NULL;
+}
+#endif
 /*==========================================
  * 終了
  *------------------------------------------*/
@@ -4021,25 +4220,88 @@ int do_final_script() {
 	
 	if( atcmd_binding_count != 0 )
 		aFree(atcmd_binding);
+#ifdef BETA_THREAD_TEST
+	/* QueryThread */
+	InterlockedIncrement(&queryThreadTerminate);
+	racond_signal(queryThreadCond);
+	rathread_wait(queryThread, NULL);
+	
+	// Destroy cond var and mutex.
+	racond_destroy( queryThreadCond );
+	ramutex_destroy( queryThreadMutex );
+	
+	/* Clear missing vars */
+	for( i = 0; i < queryThreadData.count; i++ ) {
+		aFree(queryThreadData.entry[i]);
+	}
+	
+	aFree(queryThreadData.entry);
+	
+	for( i = 0; i < logThreadData.count; i++ ) {
+		aFree(logThreadData.entry[i]);
+	}
+	
+	aFree(logThreadData.entry);	
+#endif
 	
 	return 0;
 }
 /*==========================================
  * 初期化
  *------------------------------------------*/
-int do_init_script()
-{
+int do_init_script() {
 	userfunc_db=strdb_alloc(DB_OPT_DUP_KEY,0);
 	scriptlabel_db=strdb_alloc(DB_OPT_DUP_KEY,50);
 	autobonus_db = strdb_alloc(DB_OPT_DUP_KEY,0);
 
 	mapreg_init();
+#ifdef BETA_THREAD_TEST
+	CREATE(queryThreadData.entry, struct queryThreadEntry*, 1);
+	queryThreadData.count = 0;
+	CREATE(logThreadData.entry, char *, 1);
+	logThreadData.count = 0;
+	/* QueryThread Start */
+	
+	InitializeSpinLock(&queryThreadLock);
 	
+	queryThreadData.timer = INVALID_TIMER;
+	queryThreadTerminate = 0;
+	queryThreadMutex = ramutex_create();
+	queryThreadCond = racond_create();
+	
+	queryThread = rathread_create(queryThread_main, NULL);
+	
+	if(queryThread == NULL){
+		ShowFatalError("do_init_script: cannot spawn Query Thread.\n");
+		exit(EXIT_FAILURE);
+	}
+
+	add_timer_func_list(queryThread_timer, "queryThread_timer");
+#endif
 	return 0;
 }
 
 int script_reload() {
 	int i;
+
+#ifdef BETA_THREAD_TEST
+	/* we're reloading so any queries undergoing should be...exterminated. */
+	EnterSpinLock(&queryThreadLock);
+	
+	for( i = 0; i < queryThreadData.count; i++ ) {
+		aFree(queryThreadData.entry[i]);
+	}
+	queryThreadData.count = 0;
+	
+	if( queryThreadData.timer != INVALID_TIMER ) {
+		delete_timer(queryThreadData.timer, queryThread_timer);
+		queryThreadData.timer = INVALID_TIMER;
+	}
+	
+	LeaveSpinLock(&queryThreadLock);
+#endif
+
+	
 	userfunc_db->clear(userfunc_db, db_script_free_code_sub);
 	db_clear(scriptlabel_db);
 
@@ -13960,6 +14222,7 @@ int buildin_query_sql_sub(struct script_state* st, Sql* handle)
 
 	// Execute the query
 	query = script_getstr(st,2);
+
 	if( SQL_ERROR == Sql_QueryStr(handle, query) )
 	{
 		Sql_ShowDebug(handle);
@@ -14014,24 +14277,43 @@ int buildin_query_sql_sub(struct script_state* st, Sql* handle)
 	// Free data
 	Sql_FreeResult(handle);
 	script_pushint(st, i);
+		
 	return 0;
 }
 
-BUILDIN_FUNC(query_sql)
-{
+BUILDIN_FUNC(query_sql) {
+#ifdef BETA_THREAD_TEST
+	if( st->state != RERUNLINE ) {
+		queryThread_add(st,false);
+
+		st->state = RERUNLINE;/* will continue when the query is finished running. */
+	} else
+		st->state = RUN;
+		
+	return 0;
+#else
 	return buildin_query_sql_sub(st, mmysql_handle);
+#endif
 }
 
-BUILDIN_FUNC(query_logsql)
-{
-	if( !log_config.sql_logs )
-	{// logmysql_handle == NULL
+BUILDIN_FUNC(query_logsql) {
+	if( !log_config.sql_logs ) {// logmysql_handle == NULL
 		ShowWarning("buildin_query_logsql: SQL logs are disabled, query '%s' will not be executed.\n", script_getstr(st,2));
 		script_pushint(st,-1);
 		return 1;
 	}
-
+#ifdef BETA_THREAD_TEST
+	if( st->state != RERUNLINE ) {
+		queryThread_add(st,true);
+		
+		st->state = RERUNLINE;/* will continue when the query is finished running. */
+	} else
+		st->state = RUN;
+	
+	return 0;	
+#else
 	return buildin_query_sql_sub(st, logmysql_handle);
+#endif
 }
 
 //Allows escaping of a given string.