diff options
| -rw-r--r-- | CMakeLists.txt | 1 | ||||
| -rw-r--r-- | sql/my_apc.cc | 158 | ||||
| -rw-r--r-- | sql/my_apc.h | 15 | ||||
| -rw-r--r-- | unittest/sql/CMakeLists.txt | 3 | ||||
| -rw-r--r-- | unittest/sql/my_apc-t.cc | 201 |
5 files changed, 221 insertions, 157 deletions
diff --git a/CMakeLists.txt b/CMakeLists.txt index 8b0c4898785..ec8c602761a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -279,6 +279,7 @@ IF(WITH_UNIT_TESTS) ADD_SUBDIRECTORY(unittest/strings) ADD_SUBDIRECTORY(unittest/examples) ADD_SUBDIRECTORY(unittest/mysys) + ADD_SUBDIRECTORY(unittest/sql) ENDIF() IF(NOT WITHOUT_SERVER) diff --git a/sql/my_apc.cc b/sql/my_apc.cc index 212fbd0f6cb..48d539aed78 100644 --- a/sql/my_apc.cc +++ b/sql/my_apc.cc @@ -15,22 +15,13 @@ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ -#ifdef MY_APC_STANDALONE - -#include <my_global.h> -#include <my_pthread.h> -#include <my_sys.h> - -#include "my_apc.h" - -#else +#ifndef MY_APC_STANDALONE #include "sql_priv.h" #include "sql_class.h" #endif - /* Standalone testing: g++ -c -DMY_APC_STANDALONE -g -I.. -I../include -o my_apc.o my_apc.cc @@ -140,12 +131,19 @@ void Apc_target::dequeue_request(Call_request *qe) /* Make an APC (Async Procedure Call) to another thread. + + @detail + Make an APC call: schedule it for execution and wait until the target + thread has executed it. - - The caller is responsible for making sure he's not calling to the same - thread. + - The caller is responsible for making sure he's not posting request + to the thread he's calling this function from. - - The caller should have locked target_thread_mutex. + - The caller must have locked target_mutex. The function will release it. + @retval FALSE - Ok, the call has been made + @retval TRUE - Call wasnt made (either the target is in disabled state or + timeout occured) psergey-todo: Should waits here be KILLable? (it seems one needs to use thd->enter_cond() calls to be killable) @@ -251,137 +249,3 @@ void Apc_target::process_apc_requests() } } -/***************************************************************************** - * Testing - *****************************************************************************/ -#ifdef MY_APC_STANDALONE - -volatile bool started= FALSE; -volatile bool service_should_exit= FALSE; -volatile bool requestors_should_exit=FALSE; - -volatile int apcs_served= 0; -volatile int apcs_missed=0; -volatile int apcs_timed_out=0; - -Apc_target apc_target; -mysql_mutex_t target_mutex; - -int int_rand(int size) -{ - return round (((double)rand() / RAND_MAX) * size); -} - -/* An APC-serving thread */ -void *test_apc_service_thread(void *ptr) -{ - my_thread_init(); - mysql_mutex_init(0, &target_mutex, MY_MUTEX_INIT_FAST); - apc_target.init(&target_mutex); - apc_target.enable(); - started= TRUE; - fprintf(stderr, "# test_apc_service_thread started\n"); - while (!service_should_exit) - { - //apc_target.disable(); - usleep(10000); - //apc_target.enable(); - for (int i = 0; i < 10 && !service_should_exit; i++) - { - apc_target.process_apc_requests(); - usleep(int_rand(30)); - } - } - apc_target.disable(); - apc_target.destroy(); - my_thread_end(); - pthread_exit(0); -} - -class Apc_order -{ -public: - int value; // The value - int *where_to; // Where to write it - Apc_order(int a, int *b) : value(a), where_to(b) {} -}; - -void test_apc_func(void *arg) -{ - Apc_order *order=(Apc_order*)arg; - usleep(int_rand(1000)); - *(order->where_to) = order->value; - __sync_fetch_and_add(&apcs_served, 1); -} - -void *test_apc_requestor_thread(void *ptr) -{ - my_thread_init(); - fprintf(stderr, "# test_apc_requestor_thread started\n"); - while (!requestors_should_exit) - { - int dst_value= 0; - int src_value= int_rand(4*1000*100); - /* Create APC to do dst_value= src_value */ - Apc_order apc_order(src_value, &dst_value); - bool timed_out; - - bool res= apc_target.make_apc_call(test_apc_func, (void*)&apc_order, 60, &timed_out); - if (res) - { - if (timed_out) - __sync_fetch_and_add(&apcs_timed_out, 1); - else - __sync_fetch_and_add(&apcs_missed, 1); - - if (dst_value != 0) - fprintf(stderr, "APC was done even though return value says it wasnt!\n"); - } - else - { - if (dst_value != src_value) - fprintf(stderr, "APC was not done even though return value says it was!\n"); - } - //usleep(300); - } - fprintf(stderr, "# test_apc_requestor_thread exiting\n"); - my_thread_end(); -} - -const int N_THREADS=23; -int main(int args, char **argv) -{ - pthread_t service_thr; - pthread_t request_thr[N_THREADS]; - int i, j; - my_thread_global_init(); - - pthread_create(&service_thr, NULL, test_apc_service_thread, (void*)NULL); - while (!started) - usleep(1000); - for (i = 0; i < N_THREADS; i++) - pthread_create(&request_thr[i], NULL, test_apc_requestor_thread, (void*)NULL); - - for (i = 0; i < 15; i++) - { - usleep(500*1000); - fprintf(stderr, "# %d APCs served %d missed\n", apcs_served, apcs_missed); - } - fprintf(stderr, "# Shutting down requestors\n"); - requestors_should_exit= TRUE; - for (i = 0; i < N_THREADS; i++) - pthread_join(request_thr[i], NULL); - - fprintf(stderr, "# Shutting down service\n"); - service_should_exit= TRUE; - pthread_join(service_thr, NULL); - fprintf(stderr, "# Done.\n"); - my_thread_end(); - my_thread_global_end(); - return 0; -} - -#endif // MY_APC_STANDALONE - - - diff --git a/sql/my_apc.h b/sql/my_apc.h index 3d9a2154af2..99861ca3194 100644 --- a/sql/my_apc.h +++ b/sql/my_apc.h @@ -49,22 +49,17 @@ public: void disable(); void process_apc_requests(); - + + /* Functor class for calls you can schedule */ class Apc_call { public: + /* This function will be called in the target thread */ virtual void call_in_target_thread()= 0; virtual ~Apc_call() {} }; - /* - Make an APC call: schedule it for execution and wait until the target - thread has executed it. This function must not be called from a thread - that's different from the target thread. - - @retval FALSE - Ok, the call has been made - @retval TRUE - Call wasnt made (either the target is in disabled state or - timeout occured) - */ + + /* Make a call in the target thread (see function definition for details) */ bool make_apc_call(Apc_call *call, int timeout_sec, bool *timed_out); #ifndef DBUG_OFF diff --git a/unittest/sql/CMakeLists.txt b/unittest/sql/CMakeLists.txt new file mode 100644 index 00000000000..36b14e9a08a --- /dev/null +++ b/unittest/sql/CMakeLists.txt @@ -0,0 +1,3 @@ + +MY_ADD_TESTS(my_apc LINK_LIBRARIES mysys EXT cc) + diff --git a/unittest/sql/my_apc-t.cc b/unittest/sql/my_apc-t.cc new file mode 100644 index 00000000000..ada09a1675c --- /dev/null +++ b/unittest/sql/my_apc-t.cc @@ -0,0 +1,201 @@ +/* + Copyright (c) 2012, Monty Program Ab + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; version 2 of the License. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ + +/* + This file does standalone APC system tests. +*/ +#include <stdio.h> +#include <my_global.h> +#include <my_pthread.h> +#include <my_sys.h> + +#include <tap.h> + +#include "../sql/my_apc.h" + +#define MY_APC_STANDALONE 1 +#include "../sql/my_apc.cc" + +volatile bool started= FALSE; +volatile bool service_should_exit= FALSE; +volatile bool requestors_should_exit=FALSE; + +/* Counters for APC calls */ +int apcs_served= 0; +int apcs_missed=0; +int apcs_timed_out=0; +mysql_mutex_t apc_counters_mutex; + +inline void increment_counter(int *var) +{ + mysql_mutex_lock(&apc_counters_mutex); + *var= *var+1; + mysql_mutex_unlock(&apc_counters_mutex); +} + +volatile bool have_errors= false; + +Apc_target apc_target; +mysql_mutex_t target_mutex; + +int int_rand(int size) +{ + return round (((double)rand() / RAND_MAX) * size); +} + +/* + APC target thread (the one that will serve the APC requests). We will have + one target. +*/ +void *test_apc_service_thread(void *ptr) +{ + my_thread_init(); + mysql_mutex_init(0, &target_mutex, MY_MUTEX_INIT_FAST); + apc_target.init(&target_mutex); + apc_target.enable(); + started= TRUE; + fprintf(stderr, "# test_apc_service_thread started\n"); + while (!service_should_exit) + { + //apc_target.disable(); + usleep(10000); + //apc_target.enable(); + for (int i = 0; i < 10 && !service_should_exit; i++) + { + apc_target.process_apc_requests(); + usleep(int_rand(30)); + } + } + apc_target.disable(); + apc_target.destroy(); + mysql_mutex_destroy(&target_mutex); + my_thread_end(); + pthread_exit(0); +} + + +/* + One APC request (to write 'value' into *where_to) +*/ +class Apc_order : public Apc_target::Apc_call +{ +public: + int value; // The value + int *where_to; // Where to write it + Apc_order(int a, int *b) : value(a), where_to(b) {} + + void call_in_target_thread() + { + usleep(int_rand(1000)); + *where_to = value; + increment_counter(&apcs_served); + } +}; + + +/* + APC requestor thread. It makes APC requests, and checks if they were actually + executed. +*/ +void *test_apc_requestor_thread(void *ptr) +{ + my_thread_init(); + fprintf(stderr, "# test_apc_requestor_thread started\n"); + while (!requestors_should_exit) + { + int dst_value= 0; + int src_value= int_rand(4*1000*100); + /* Create an APC to do "dst_value= src_value" assignment */ + Apc_order apc_order(src_value, &dst_value); + bool timed_out; + + mysql_mutex_lock(&target_mutex); + bool res= apc_target.make_apc_call(&apc_order, 60, &timed_out); + if (res) + { + if (timed_out) + increment_counter(&apcs_timed_out); + else + increment_counter(&apcs_missed); + + if (dst_value != 0) + { + fprintf(stderr, "APC was done even though return value says it wasnt!\n"); + have_errors= true; + } + } + else + { + if (dst_value != src_value) + { + fprintf(stderr, "APC was not done even though return value says it was!\n"); + have_errors= true; + } + } + //usleep(300); + } + fprintf(stderr, "# test_apc_requestor_thread exiting\n"); + my_thread_end(); + return NULL; +} + +/* Number of APC requestor threads */ +const int N_THREADS=23; + + +int main(int args, char **argv) +{ + pthread_t service_thr; + pthread_t request_thr[N_THREADS]; + int i; + + my_thread_global_init(); + + mysql_mutex_init(0, &apc_counters_mutex, MY_MUTEX_INIT_FAST); + + plan(1); + diag("Testing APC delivery and execution"); + + pthread_create(&service_thr, NULL, test_apc_service_thread, (void*)NULL); + while (!started) + usleep(1000); + for (i = 0; i < N_THREADS; i++) + pthread_create(&request_thr[i], NULL, test_apc_requestor_thread, (void*)NULL); + + for (i = 0; i < 15; i++) + { + usleep(500*1000); + fprintf(stderr, "# %d APCs served %d missed\n", apcs_served, apcs_missed); + } + fprintf(stderr, "# Shutting down requestors\n"); + requestors_should_exit= TRUE; + for (i = 0; i < N_THREADS; i++) + pthread_join(request_thr[i], NULL); + + fprintf(stderr, "# Shutting down service\n"); + service_should_exit= TRUE; + pthread_join(service_thr, NULL); + + mysql_mutex_destroy(&apc_counters_mutex); + + fprintf(stderr, "# Done.\n"); + my_thread_end(); + my_thread_global_end(); + + ok1(!have_errors); + return exit_status(); +} + |