Browse Source
MWL#182: Explain running statements
MWL#182: Explain running statements
First code - "Asynchronous procedure call" system - new THD::check_killed() that serves APC request is called from within most important loops - EXPLAIN code is now able to generate EXPLAIN output on-the-fly [incomplete] Parts that are still missing: - put THD::check_killed() call into every loop where we could spend significant amount of time - Make sure EXPLAIN code works for group-by queries that replace JOIN::join_tab with make_simple_join() and other such cases. - User interface: what error code to use, where to get timeout settings from, etc.pull/57/head
Sergey Petrunya
14 years ago
18 changed files with 915 additions and 36 deletions
-
2libmysqld/Makefile.am
-
7sql/Makefile.am
-
355sql/my_apc.cc
-
98sql/my_apc.h
-
2sql/mysql_priv.h
-
1sql/mysqld.cc
-
19sql/protocol.h
-
1sql/sp_head.cc
-
119sql/sql_class.cc
-
94sql/sql_class.h
-
34sql/sql_lex.cc
-
6sql/sql_lex.h
-
27sql/sql_parse.cc
-
1sql/sql_prepare.cc
-
81sql/sql_select.cc
-
7sql/sql_select.h
-
91sql/sql_show.cc
-
6sql/sql_yacc.yy
@ -0,0 +1,355 @@ |
|||
/*
|
|||
TODO: MP AB Copyright |
|||
*/ |
|||
|
|||
|
|||
#ifdef MY_APC_STANDALONE
|
|||
|
|||
#include <my_global.h>
|
|||
#include <my_pthread.h>
|
|||
#include <my_sys.h>
|
|||
|
|||
#else
|
|||
|
|||
#include "mysql_priv.h"
|
|||
|
|||
#endif
|
|||
|
|||
//#include "my_apc.h"
|
|||
|
|||
/*
|
|||
Standalone testing: |
|||
g++ -c -DMY_APC_STANDALONE -g -I.. -I../include -o my_apc.o my_apc.cc |
|||
g++ -L../mysys -L../dbug -L../strings my_apc.o -lmysys -ldbug -lmystrings -lpthread -lrt |
|||
*/ |
|||
|
|||
|
|||
void Apc_target::init() |
|||
{ |
|||
// todo: should use my_pthread_... functions instead?
|
|||
DBUG_ASSERT(!enabled); |
|||
(void)pthread_mutex_init(&LOCK_apc_queue, MY_MUTEX_INIT_SLOW); |
|||
} |
|||
|
|||
|
|||
void Apc_target::destroy() |
|||
{ |
|||
DBUG_ASSERT(!enabled); |
|||
pthread_mutex_destroy(&LOCK_apc_queue); |
|||
} |
|||
|
|||
|
|||
void Apc_target::enable() |
|||
{ |
|||
pthread_mutex_lock(&LOCK_apc_queue); |
|||
enabled++; |
|||
pthread_mutex_unlock(&LOCK_apc_queue); |
|||
} |
|||
|
|||
|
|||
void Apc_target::disable() |
|||
{ |
|||
bool process= FALSE; |
|||
pthread_mutex_lock(&LOCK_apc_queue); |
|||
if (!(--enabled)) |
|||
process= TRUE; |
|||
pthread_mutex_unlock(&LOCK_apc_queue); |
|||
if (process) |
|||
process_apc_requests(); |
|||
} |
|||
|
|||
void Apc_target::enqueue_request(Call_request *qe) |
|||
{ |
|||
//call_queue_size++;
|
|||
if (apc_calls) |
|||
{ |
|||
Call_request *after= apc_calls->prev; |
|||
qe->next= apc_calls; |
|||
apc_calls->prev= qe; |
|||
|
|||
qe->prev= after; |
|||
after->next= qe; |
|||
} |
|||
else |
|||
{ |
|||
apc_calls= qe; |
|||
qe->next= qe->prev= qe; |
|||
} |
|||
} |
|||
|
|||
void Apc_target::dequeue_request(Call_request *qe) |
|||
{ |
|||
//call_queue_size--;
|
|||
if (apc_calls == qe) |
|||
{ |
|||
if ((apc_calls= apc_calls->next) == qe) |
|||
{ |
|||
//DBUG_ASSERT(!call_queue_size);
|
|||
apc_calls= NULL; |
|||
} |
|||
} |
|||
|
|||
qe->prev->next= qe->next; |
|||
qe->next->prev= qe->prev; |
|||
} |
|||
|
|||
|
|||
/*
|
|||
Make an apc call in another thread. The caller is responsible so |
|||
that we're not calling to ourselves. |
|||
|
|||
*/ |
|||
|
|||
bool Apc_target::make_apc_call(apc_func_t func, void *func_arg, |
|||
int timeout_sec, bool *timed_out) |
|||
{ |
|||
bool res= TRUE; |
|||
*timed_out= FALSE; |
|||
|
|||
pthread_mutex_lock(&LOCK_apc_queue); |
|||
if (enabled) |
|||
{ |
|||
/* Create and post the request */ |
|||
Call_request apc_request; |
|||
apc_request.func= func; |
|||
apc_request.func_arg= func_arg; |
|||
apc_request.done= FALSE; |
|||
(void)pthread_cond_init(&apc_request.COND_request, NULL); |
|||
(void)pthread_mutex_init(&apc_request.LOCK_request, MY_MUTEX_INIT_SLOW); |
|||
pthread_mutex_lock(&apc_request.LOCK_request); |
|||
enqueue_request(&apc_request); |
|||
apc_request.what="enqueued by make_apc_call"; |
|||
pthread_mutex_unlock(&LOCK_apc_queue); |
|||
|
|||
struct timespec abstime; |
|||
const int timeout= timeout_sec; |
|||
set_timespec(abstime, timeout); |
|||
|
|||
int wait_res= 0; |
|||
/* todo: how about processing other errors here? */ |
|||
while (!apc_request.done && (wait_res != ETIMEDOUT)) |
|||
{ |
|||
wait_res= pthread_cond_timedwait(&apc_request.COND_request, |
|||
&apc_request.LOCK_request, &abstime); |
|||
} |
|||
|
|||
if (!apc_request.done) |
|||
{ |
|||
/* We timed out */ |
|||
apc_request.done= TRUE; |
|||
*timed_out= TRUE; |
|||
pthread_mutex_unlock(&apc_request.LOCK_request); |
|||
|
|||
pthread_mutex_lock(&LOCK_apc_queue); |
|||
dequeue_request(&apc_request); |
|||
pthread_mutex_unlock(&LOCK_apc_queue); |
|||
res= TRUE; |
|||
} |
|||
else |
|||
{ |
|||
/* Request was successfully executed and dequeued by the target thread */ |
|||
pthread_mutex_unlock(&apc_request.LOCK_request); |
|||
res= FALSE; |
|||
} |
|||
|
|||
/* Destroy all APC request data */ |
|||
pthread_mutex_destroy(&apc_request.LOCK_request); |
|||
pthread_cond_destroy(&apc_request.COND_request); |
|||
} |
|||
else |
|||
{ |
|||
pthread_mutex_unlock(&LOCK_apc_queue); |
|||
} |
|||
return res; |
|||
} |
|||
|
|||
|
|||
/*
|
|||
Process all APC requests |
|||
*/ |
|||
|
|||
void Apc_target::process_apc_requests() |
|||
{ |
|||
while (1) |
|||
{ |
|||
Call_request *request; |
|||
|
|||
pthread_mutex_lock(&LOCK_apc_queue); |
|||
if (!(request= get_first_in_queue())) |
|||
{ |
|||
pthread_mutex_unlock(&LOCK_apc_queue); |
|||
break; |
|||
} |
|||
|
|||
request->what="seen by process_apc_requests"; |
|||
pthread_mutex_lock(&request->LOCK_request); |
|||
|
|||
if (request->done) |
|||
{ |
|||
/*
|
|||
We can get here when |
|||
- the requestor thread has been waiting for this request |
|||
- the wait has timed out |
|||
- it has set request->done=TRUE |
|||
- it has released LOCK_request, because its next action |
|||
will be to remove the request from the queue, however, |
|||
it could not attempt to lock the queue while holding the lock on |
|||
request, because that would deadlock with this function |
|||
(we here first lock the queue and then lock the request) |
|||
*/ |
|||
pthread_mutex_unlock(&request->LOCK_request); |
|||
pthread_mutex_unlock(&LOCK_apc_queue); |
|||
fprintf(stderr, "Whoa rare event #1!\n"); |
|||
continue; |
|||
} |
|||
/*
|
|||
Remove the request from the queue (we're holding its lock so we can be |
|||
sure that request owner won't try to remove it) |
|||
*/ |
|||
request->what="dequeued by process_apc_requests"; |
|||
dequeue_request(request); |
|||
request->done= TRUE; |
|||
|
|||
pthread_mutex_unlock(&LOCK_apc_queue); |
|||
|
|||
request->func(request->func_arg); |
|||
request->what="func called by process_apc_requests"; |
|||
|
|||
pthread_cond_signal(&request->COND_request); |
|||
|
|||
pthread_mutex_unlock(&request->LOCK_request); |
|||
} |
|||
} |
|||
|
|||
/*****************************************************************************
|
|||
* 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; |
|||
|
|||
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(); |
|||
apc_target.init(); |
|||
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
|
|||
|
|||
|
|||
|
|||
@ -0,0 +1,98 @@ |
|||
/* |
|||
TODO: MP AB Copyright |
|||
*/ |
|||
|
|||
/* |
|||
Design |
|||
- Mutex-guarded request queue (it belongs to the target), which can be enabled/ |
|||
disabled (when empty). |
|||
|
|||
- After the request has been put into queue, the requestor waits for request |
|||
to be satisfied. The worker satisifes the request and signals the |
|||
requestor. |
|||
*/ |
|||
|
|||
/* |
|||
Target for asynchronous calls. |
|||
*/ |
|||
class Apc_target |
|||
{ |
|||
public: |
|||
Apc_target() : enabled(0), apc_calls(NULL) /*, call_queue_size(0)*/ {} |
|||
~Apc_target() { DBUG_ASSERT(!enabled && !apc_calls);} |
|||
|
|||
/* |
|||
Initialize the target. This must be called before anything else. Right |
|||
after initialization, the target is disabled. |
|||
*/ |
|||
void init(); |
|||
|
|||
/* |
|||
Destroy the target. The target must be disabled when this call is made. |
|||
*/ |
|||
void destroy(); |
|||
|
|||
/* |
|||
Enter into state where this target will be serving APC requests |
|||
*/ |
|||
void enable(); |
|||
|
|||
/* |
|||
Leave the state where we could serve APC requests (will serve all already |
|||
enqueued requests) |
|||
*/ |
|||
void disable(); |
|||
|
|||
/* |
|||
This should be called periodically to serve observation requests. |
|||
*/ |
|||
void process_apc_requests(); |
|||
|
|||
typedef void (*apc_func_t)(void *arg); |
|||
|
|||
/* |
|||
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) |
|||
*/ |
|||
bool make_apc_call(apc_func_t func, void *func_arg, |
|||
int timeout_sec, bool *timed_out); |
|||
|
|||
private: |
|||
class Call_request; |
|||
int enabled; |
|||
|
|||
Call_request *apc_calls; |
|||
pthread_mutex_t LOCK_apc_queue; |
|||
//int call_queue_size; |
|||
|
|||
class Call_request |
|||
{ |
|||
public: |
|||
apc_func_t func; |
|||
void *func_arg; |
|||
bool done; |
|||
|
|||
pthread_mutex_t LOCK_request; |
|||
pthread_cond_t COND_request; |
|||
|
|||
Call_request *next; |
|||
Call_request *prev; |
|||
|
|||
const char *what; |
|||
}; |
|||
|
|||
void enqueue_request(Call_request *qe); |
|||
void dequeue_request(Call_request *qe); |
|||
Call_request *get_first_in_queue() |
|||
{ |
|||
return apc_calls; |
|||
} |
|||
}; |
|||
|
|||
/////////////////////////////////////////////////////////////////////// |
|||
|
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue