Commit d94bae3a authored by andrey@lmy004's avatar andrey@lmy004

WL#1034 update

QUEUE implementation working now. this should be ready more or less
for testing once the debug output is being cleaned and some things
around DYNAMIC_ARRAY are cleaned
- fix handling in case of errors that lead to crashes, now no more crashes
  in case of table corruption and such.
parent 56db09cf
...@@ -35,6 +35,7 @@ typedef struct st_queue { ...@@ -35,6 +35,7 @@ typedef struct st_queue {
uint offset_to_key; /* compare is done on element+offset */ uint offset_to_key; /* compare is done on element+offset */
int max_at_top; /* Set if queue_top gives max */ int max_at_top; /* Set if queue_top gives max */
int (*compare)(void *, byte *,byte *); int (*compare)(void *, byte *,byte *);
uint auto_extent;
} QUEUE; } QUEUE;
#define queue_top(queue) ((queue)->root[1]) #define queue_top(queue) ((queue)->root[1])
...@@ -49,14 +50,19 @@ typedef int (*queue_compare)(void *,byte *, byte *); ...@@ -49,14 +50,19 @@ typedef int (*queue_compare)(void *,byte *, byte *);
int init_queue(QUEUE *queue,uint max_elements,uint offset_to_key, int init_queue(QUEUE *queue,uint max_elements,uint offset_to_key,
pbool max_at_top, queue_compare compare, pbool max_at_top, queue_compare compare,
void *first_cmp_arg); void *first_cmp_arg);
int init_queue_ex(QUEUE *queue,uint max_elements,uint offset_to_key,
pbool max_at_top, queue_compare compare,
void *first_cmp_arg, uint auto_extent);
int reinit_queue(QUEUE *queue,uint max_elements,uint offset_to_key, int reinit_queue(QUEUE *queue,uint max_elements,uint offset_to_key,
pbool max_at_top, queue_compare compare, pbool max_at_top, queue_compare compare,
void *first_cmp_arg); void *first_cmp_arg);
int resize_queue(QUEUE *queue, uint max_elements); int resize_queue(QUEUE *queue, uint max_elements);
void delete_queue(QUEUE *queue); void delete_queue(QUEUE *queue);
void queue_insert(QUEUE *queue,byte *element); void queue_insert(QUEUE *queue,byte *element);
int queue_insert_safe(QUEUE *queue, byte *element);
byte *queue_remove(QUEUE *queue,uint idx); byte *queue_remove(QUEUE *queue,uint idx);
#define queue_remove_all(queue) { (queue)->elements= 0; } #define queue_remove_all(queue) { (queue)->elements= 0; }
#define queue_is_full(queue) (queue->elements == queue->max_elements)
void _downheap(QUEUE *queue,uint idx); void _downheap(QUEUE *queue,uint idx);
void queue_fix(QUEUE *queue); void queue_fix(QUEUE *queue);
#define is_queue_inited(queue) ((queue)->root != 0) #define is_queue_inited(queue) ((queue)->root != 0)
......
...@@ -19,7 +19,7 @@ ...@@ -19,7 +19,7 @@
Implemention of queues from "Algoritms in C" by Robert Sedgewick. Implemention of queues from "Algoritms in C" by Robert Sedgewick.
An optimisation of _downheap suggested in Exercise 7.51 in "Data An optimisation of _downheap suggested in Exercise 7.51 in "Data
Structures & Algorithms in C++" by Mark Allen Weiss, Second Edition Structures & Algorithms in C++" by Mark Allen Weiss, Second Edition
was implemented by Mikael Ronstrm 2005. Also the O(N) algorithm was implemented by Mikael Ronstrom 2005. Also the O(N) algorithm
of queue_fix was implemented. of queue_fix was implemented.
*/ */
...@@ -67,6 +67,46 @@ int init_queue(QUEUE *queue, uint max_elements, uint offset_to_key, ...@@ -67,6 +67,46 @@ int init_queue(QUEUE *queue, uint max_elements, uint offset_to_key,
} }
/*
Init queue, uses init_queue internally for init work but also accepts
auto_extent as parameter
SYNOPSIS
init_queue_ex()
queue Queue to initialise
max_elements Max elements that will be put in queue
offset_to_key Offset to key in element stored in queue
Used when sending pointers to compare function
max_at_top Set to 1 if you want biggest element on top.
compare Compare function for elements, takes 3 arguments.
first_cmp_arg First argument to compare function
auto_extent When the queue is full and there is insert operation
extend the queue.
NOTES
Will allocate max_element pointers for queue array
RETURN
0 ok
1 Could not allocate memory
*/
int init_queue_ex(QUEUE *queue, uint max_elements, uint offset_to_key,
pbool max_at_top, int (*compare) (void *, byte *, byte *),
void *first_cmp_arg, uint auto_extent)
{
int ret;
DBUG_ENTER("init_queue_ex");
if ((ret= init_queue(queue, max_elements, offset_to_key, max_at_top, compare,
first_cmp_arg)))
DBUG_RETURN(ret);
queue->auto_extent= auto_extent;
DBUG_RETURN(0);
}
/* /*
Reinitialize queue for other usage Reinitialize queue for other usage
...@@ -192,6 +232,31 @@ void queue_insert(register QUEUE *queue, byte *element) ...@@ -192,6 +232,31 @@ void queue_insert(register QUEUE *queue, byte *element)
} }
} }
/*
Does safe insert. If no more space left on the queue resize it.
Return codes:
0 - OK
1 - Cannot allocate more memory
2 - auto_extend is 0, the operation would
*/
int queue_insert_safe(register QUEUE *queue, byte *element)
{
if (queue->elements == queue->max_elements)
{
if (!queue->auto_extent)
return 2;
else if (resize_queue(queue, queue->max_elements + queue->auto_extent))
return 1;
}
queue_insert(queue, element);
return 0;
}
/* Remove item from queue */ /* Remove item from queue */
/* Returns pointer to removed element */ /* Returns pointer to removed element */
......
...@@ -14,16 +14,14 @@ ...@@ -14,16 +14,14 @@
along with this program; if not, write to the Free Software along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include "mysql_priv.h"
#include "event.h"
#include "event_priv.h" #include "event_priv.h"
#include "event.h"
#include "sp.h" #include "sp.h"
/* /*
TODO list : TODO list :
- The default value of created/modified should not be 0000-00-00 because of - The default value of created/modified should not be 0000-00-00 because of
STRICT mode restricions. STRICT mode restricions.
- Remove m_ prefixes of member variables.
- Use timestamps instead of datetime. - Use timestamps instead of datetime.
...@@ -53,9 +51,17 @@ ...@@ -53,9 +51,17 @@
- Consider using conditional variable when doing shutdown instead of - Consider using conditional variable when doing shutdown instead of
waiting till all worker threads end. waiting till all worker threads end.
- Make event_timed::get_show_create_event() work - Make event_timed::get_show_create_event() work
- Add function documentation whenever needed. - Add function documentation whenever needed.
- Add logging to file - Add logging to file
- Move comparison code to class event_timed
- Overload event_timed::new to put the event directly in the DYNAMIC_ARRAY.
This will skip copy operation as well as will simplify the code which is
now aware of events_array DYNAMIC_ARRAY
Warning: Warning:
- For now parallel execution is not possible because the same sp_head cannot be - For now parallel execution is not possible because the same sp_head cannot be
executed few times!!! There is still no lock attached to particular event. executed few times!!! There is still no lock attached to particular event.
...@@ -67,19 +73,60 @@ Warning: ...@@ -67,19 +73,60 @@ Warning:
bool mysql_event_table_exists= 1; bool mysql_event_table_exists= 1;
DYNAMIC_ARRAY events_array; DYNAMIC_ARRAY events_array;
DYNAMIC_ARRAY evex_executing_queue; DYNAMIC_ARRAY EXEC_QUEUE_DARR_NAME;
QUEUE EXEC_QUEUE_QUEUE_NAME;
MEM_ROOT evex_mem_root; MEM_ROOT evex_mem_root;
//extern volatile uint thread_running; //extern volatile uint thread_running;
//////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////
//////////////// Static functions follow /////////////////////////// //////////////// Static functions follow ///////////////////////////
//////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////
void
evex_queue_init(EVEX_QUEUE_TYPE *queue)
{
#ifndef EVEX_USE_QUEUE
VOID(my_init_dynamic_array(queue, sizeof(event_timed *), 50, 100));
#else
if (init_queue_ex(queue, 100 /*num_el*/, 0 /*offset*/,
0 /*smallest_on_top*/, event_timed_compare_q, NULL,
100 /*auto_extent*/))
sql_print_error("Insufficient memory to initialize executing queue.");
#endif
}
int
evex_queue_insert2(EVEX_QUEUE_TYPE *queue, EVEX_PTOQEL element)
{
#ifndef EVEX_USE_QUEUE
VOID(push_dynamic(queue, element));
return 0;
#else
return queue_insert_safe(queue, element);
#endif
}
void
evex_queue_top_updated(EVEX_QUEUE_TYPE *queue)
{
#ifdef EVEX_USE_QUEUE
queue_replaced(queue);
#endif
}
void
evex_queue_sort(EVEX_QUEUE_TYPE *queue)
{
#ifndef EVEX_USE_QUEUE
qsort((gptr) dynamic_element(queue, 0, event_timed**),
queue->elements,
sizeof(event_timed **),
(qsort_cmp) event_timed_compare);
#endif
}
/* NOTE Andrey: Document better /* NOTE Andrey: Document better
Compares two TIME structures. Compares two TIME structures.
...@@ -98,7 +145,7 @@ int sortcmp_lex_string(LEX_STRING s, LEX_STRING t, CHARSET_INFO *cs) ...@@ -98,7 +145,7 @@ int sortcmp_lex_string(LEX_STRING s, LEX_STRING t, CHARSET_INFO *cs)
} }
inline int int
my_time_compare(TIME *a, TIME *b) my_time_compare(TIME *a, TIME *b)
{ {
/* /*
...@@ -107,6 +154,7 @@ my_time_compare(TIME *a, TIME *b) ...@@ -107,6 +154,7 @@ my_time_compare(TIME *a, TIME *b)
*/ */
DBUG_ENTER("my_time_compare"); DBUG_ENTER("my_time_compare");
if (a->year > b->year) if (a->year > b->year)
DBUG_RETURN(1); DBUG_RETURN(1);
...@@ -143,19 +191,53 @@ my_time_compare(TIME *a, TIME *b) ...@@ -143,19 +191,53 @@ my_time_compare(TIME *a, TIME *b)
if (a->second < b->second) if (a->second < b->second)
DBUG_RETURN(-1); DBUG_RETURN(-1);
/*!! second_part is not compared !*/
if (a->second_part > b->second_part)
DBUG_RETURN(1);
if (a->second_part < b->second_part)
DBUG_RETURN(-1);
DBUG_RETURN(0); DBUG_RETURN(0);
} }
int
evex_time_diff(TIME *a, TIME *b)
{
my_bool in_gap;
DBUG_ENTER("my_time_diff");
return sec_since_epoch_TIME(a) - sec_since_epoch_TIME(b);
}
inline int inline int
event_timed_compare(event_timed **a, event_timed **b) event_timed_compare(event_timed **a, event_timed **b)
{ {
return my_time_compare(&(*a)->execute_at, &(*b)->execute_at); my_ulonglong a_t, b_t;
a_t= TIME_to_ulonglong_datetime(&(*a)->execute_at)*100L +
(*a)->execute_at.second_part;
b_t= TIME_to_ulonglong_datetime(&(*b)->execute_at)*100L +
(*b)->execute_at.second_part;
if (a_t > b_t)
return 1;
else if (a_t < b_t)
return -1;
else
return 0;
// return my_time_compare(&(*a)->execute_at, &(*b)->execute_at);
} }
int
event_timed_compare_q(void *vptr, byte* a, byte *b)
{
return event_timed_compare((event_timed **)&a, (event_timed **)&b);
}
/* /*
Open mysql.event table for read Open mysql.event table for read
...@@ -660,7 +742,10 @@ evex_load_and_compile_event(THD * thd, sp_name *spn, bool use_lock) ...@@ -660,7 +742,10 @@ evex_load_and_compile_event(THD * thd, sp_name *spn, bool use_lock)
VOID(push_dynamic(&events_array,(gptr) ett)); VOID(push_dynamic(&events_array,(gptr) ett));
ett_copy= dynamic_element(&events_array, events_array.elements - 1, ett_copy= dynamic_element(&events_array, events_array.elements - 1,
event_timed*); event_timed*);
/**
VOID(push_dynamic(&evex_executing_queue, (gptr) &ett_copy)); VOID(push_dynamic(&evex_executing_queue, (gptr) &ett_copy));
**/
evex_queue_insert(&EVEX_EQ_NAME, (EVEX_PTOQEL) ett_copy);
/* /*
There is a copy in the array which we don't need. sphead won't be There is a copy in the array which we don't need. sphead won't be
...@@ -674,10 +759,13 @@ evex_load_and_compile_event(THD * thd, sp_name *spn, bool use_lock) ...@@ -674,10 +759,13 @@ evex_load_and_compile_event(THD * thd, sp_name *spn, bool use_lock)
qsort of events_array.elements (the current number of elements). qsort of events_array.elements (the current number of elements).
We know that the elements are stored in a contiguous block w/o holes. We know that the elements are stored in a contiguous block w/o holes.
*/ */
/**
qsort((gptr) dynamic_element(&evex_executing_queue, 0, event_timed**), qsort((gptr) dynamic_element(&evex_executing_queue, 0, event_timed**),
evex_executing_queue.elements, evex_executing_queue.elements,
sizeof(event_timed **), sizeof(event_timed **),
(qsort_cmp) event_timed_compare); (qsort_cmp) event_timed_compare);
**/
evex_queue_sort(&EVEX_EQ_NAME);
if (use_lock) if (use_lock)
VOID(pthread_mutex_unlock(&LOCK_event_arrays)); VOID(pthread_mutex_unlock(&LOCK_event_arrays));
...@@ -703,7 +791,7 @@ evex_remove_from_cache(LEX_STRING *db, LEX_STRING *name, bool use_lock) ...@@ -703,7 +791,7 @@ evex_remove_from_cache(LEX_STRING *db, LEX_STRING *name, bool use_lock)
if (use_lock) if (use_lock)
VOID(pthread_mutex_lock(&LOCK_event_arrays)); VOID(pthread_mutex_lock(&LOCK_event_arrays));
/**
for (i= 0; i < evex_executing_queue.elements; ++i) for (i= 0; i < evex_executing_queue.elements; ++i)
{ {
event_timed *et= *dynamic_element(&evex_executing_queue, i, event_timed**); event_timed *et= *dynamic_element(&evex_executing_queue, i, event_timed**);
...@@ -733,6 +821,25 @@ evex_remove_from_cache(LEX_STRING *db, LEX_STRING *name, bool use_lock) ...@@ -733,6 +821,25 @@ evex_remove_from_cache(LEX_STRING *db, LEX_STRING *name, bool use_lock)
goto done; goto done;
} }
} }
**/
for (i= 0; i < evex_queue_num_elements(EVEX_EQ_NAME); ++i)
{
event_timed *et= *evex_queue_element(&EVEX_EQ_NAME, i, event_timed**);
DBUG_PRINT("info", ("[%s.%s]==[%s.%s]?",db->str,name->str, et->dbname.str,
et->name.str));
if (!sortcmp_lex_string(*name, et->name, system_charset_info) &&
!sortcmp_lex_string(*db, et->dbname, system_charset_info))
{
int idx= get_index_dynamic(&events_array, (gptr) et);
//we are lucky the event is in the executing queue, no need of second pass
//destruct first and then remove. the destructor will delete sp_head
et->free_sp();
evex_queue_delete_element(&EVEX_EQ_NAME, idx);
evex_queue_delete_element(&EVEX_EQ_NAME, i);
// ok, we have cleaned
goto done;
}
}
/* /*
ToDo Andrey : Think about whether second pass is needed. All events ToDo Andrey : Think about whether second pass is needed. All events
......
...@@ -16,11 +16,9 @@ ...@@ -16,11 +16,9 @@
#ifndef _EVENT_H_ #ifndef _EVENT_H_
#define _EVENT_H_ #define _EVENT_H_
#include "sp_head.h"
#include "sp.h"
extern ulong opt_event_executor; #include "sp.h"
#include "sp_head.h"
#define EVEX_OK SP_OK #define EVEX_OK SP_OK
#define EVEX_KEY_NOT_FOUND SP_KEY_NOT_FOUND #define EVEX_KEY_NOT_FOUND SP_KEY_NOT_FOUND
......
...@@ -14,9 +14,8 @@ ...@@ -14,9 +14,8 @@
along with this program; if not, write to the Free Software along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include "mysql_priv.h"
#include "event.h"
#include "event_priv.h" #include "event_priv.h"
#include "event.h"
#include "sp.h" #include "sp.h"
...@@ -40,6 +39,7 @@ pthread_mutex_t LOCK_event_arrays, ...@@ -40,6 +39,7 @@ pthread_mutex_t LOCK_event_arrays,
bool evex_is_running= false; bool evex_is_running= false;
ulonglong evex_main_thread_id= 0;
ulong opt_event_executor; ulong opt_event_executor;
my_bool event_executor_running_global_var= false; my_bool event_executor_running_global_var= false;
static my_bool evex_mutexes_initted= false; static my_bool evex_mutexes_initted= false;
...@@ -74,6 +74,7 @@ void evex_init_mutexes() ...@@ -74,6 +74,7 @@ void evex_init_mutexes()
pthread_mutex_init(&LOCK_evex_running, MY_MUTEX_INIT_FAST); pthread_mutex_init(&LOCK_evex_running, MY_MUTEX_INIT_FAST);
} }
int int
init_events() init_events()
{ {
...@@ -109,6 +110,7 @@ shutdown_events() ...@@ -109,6 +110,7 @@ shutdown_events()
VOID(pthread_mutex_lock(&LOCK_evex_running)); VOID(pthread_mutex_lock(&LOCK_evex_running));
VOID(pthread_mutex_unlock(&LOCK_evex_running)); VOID(pthread_mutex_unlock(&LOCK_evex_running));
pthread_mutex_destroy(&LOCK_event_arrays); pthread_mutex_destroy(&LOCK_event_arrays);
pthread_mutex_destroy(&LOCK_workers_count); pthread_mutex_destroy(&LOCK_workers_count);
pthread_mutex_destroy(&LOCK_evex_running); pthread_mutex_destroy(&LOCK_evex_running);
...@@ -200,7 +202,12 @@ event_executor_main(void *arg) ...@@ -200,7 +202,12 @@ event_executor_main(void *arg)
thus data should be freed at later stage. thus data should be freed at later stage.
*/ */
VOID(my_init_dynamic_array(&events_array, sizeof(event_timed), 50, 100)); VOID(my_init_dynamic_array(&events_array, sizeof(event_timed), 50, 100));
/**
VOID(my_init_dynamic_array(&evex_executing_queue, sizeof(event_timed *), 50, 100)); VOID(my_init_dynamic_array(&evex_executing_queue, sizeof(event_timed *), 50, 100));
**/
evex_queue_init(&EVEX_EQ_NAME);
VOID(pthread_mutex_unlock(&LOCK_event_arrays)); VOID(pthread_mutex_unlock(&LOCK_event_arrays));
/* /*
...@@ -217,48 +224,92 @@ event_executor_main(void *arg) ...@@ -217,48 +224,92 @@ event_executor_main(void *arg)
THD_CHECK_SENTRY(thd); THD_CHECK_SENTRY(thd);
/* Read queries from the IO/THREAD until this thread is killed */ /* Read queries from the IO/THREAD until this thread is killed */
evex_main_thread_id= thd->thread_id;
while (!thd->killed) while (!thd->killed)
{ {
TIME time_now; TIME time_now;
my_time_t now; my_time_t now;
my_ulonglong cnt; my_ulonglong cnt;
event_timed *et;
DBUG_PRINT("info", ("EVEX External Loop %d", ++cnt)); DBUG_PRINT("info", ("EVEX External Loop %d", ++cnt));
thd->proc_info = "Sleeping"; thd->proc_info = "Sleeping";
if (!evex_queue_num_elements(EVEX_EQ_NAME) ||
!event_executor_running_global_var)
{
my_sleep(1000000);// sleep 1s my_sleep(1000000);// sleep 1s
if (!event_executor_running_global_var)
continue; continue;
}
{
int t2sleep;
/*
now let's see how much time to sleep, we know there is at least 1
element in the queue.
*/
VOID(pthread_mutex_lock(&LOCK_event_arrays));
if (!evex_queue_num_elements(EVEX_EQ_NAME))
{
VOID(pthread_mutex_unlock(&LOCK_event_arrays));
continue;
}
et= evex_queue_first_element(&EVEX_EQ_NAME, event_timed*);
time(&now); time(&now);
my_tz_UTC->gmt_sec_to_TIME(&time_now, now); my_tz_UTC->gmt_sec_to_TIME(&time_now, now);
t2sleep= evex_time_diff(&et->execute_at, &time_now);
VOID(pthread_mutex_unlock(&LOCK_event_arrays));
if (t2sleep > 0)
{
sql_print_information("Sleeping for %d seconds.", t2sleep);
printf("\nWHEN=%llu NOW=%llu\n", TIME_to_ulonglong_datetime(&et->execute_at), TIME_to_ulonglong_datetime(&time_now));
/*
We sleep t2sleep seconds but we check every second whether this thread
has been killed, or there is new candidate
*/
while (t2sleep-- && !thd->killed &&
evex_queue_num_elements(EVEX_EQ_NAME) &&
(evex_queue_first_element(&EVEX_EQ_NAME, event_timed*) == et))
my_sleep(1000000);
sql_print_information("Finished sleeping");
}
if (!event_executor_running_global_var)
continue;
}
VOID(pthread_mutex_lock(&LOCK_event_arrays)); VOID(pthread_mutex_lock(&LOCK_event_arrays));
for (i= 0; (i < evex_executing_queue.elements) && !thd->killed; ++i)
if (!evex_queue_num_elements(EVEX_EQ_NAME))
{ {
event_timed *et= *dynamic_element(&evex_executing_queue,i,event_timed**); VOID(pthread_mutex_unlock(&LOCK_event_arrays));
// printf("%llu\n", TIME_to_ulonglong_datetime(&et->execute_at)); continue;
if (!event_executor_running_global_var) }
break; et= evex_queue_first_element(&EVEX_EQ_NAME, event_timed*);
thd->proc_info = "Iterating";
THD_CHECK_SENTRY(thd);
/* /*
if this is the first event which is after time_now then no if this is the first event which is after time_now then no
more need to iterate over more elements since the array is sorted. more need to iterate over more elements since the array is sorted.
*/ */
if (et->execute_at.year > 1969 && if (et->execute_at.year > 1969 &&
my_time_compare(&time_now, &et->execute_at) == -1) my_time_compare(&time_now, &et->execute_at) == -1)
break; {
VOID(pthread_mutex_unlock(&LOCK_event_arrays));
continue;
}
if (et->status == MYSQL_EVENT_ENABLED && if (et->status == MYSQL_EVENT_ENABLED)
!check_access(thd, EVENT_ACL, et->dbname.str, 0, 0, 0,
is_schema_db(et->dbname.str)))
{ {
pthread_t th; pthread_t th;
DBUG_PRINT("info", (" Spawning a thread %d", ++iter_num)); DBUG_PRINT("info", (" Spawning a thread %d", ++iter_num));
thd->proc_info = "Starting new thread";
sql_print_information(" Spawning a thread %d", ++iter_num); sql_print_information(" Spawning a thread %d", ++iter_num);
#ifndef DBUG_FAULTY_THR #ifndef DBUG_FAULTY_THR
sql_print_information(" Thread is not debuggable!");
if (pthread_create(&th, NULL, event_executor_worker, (void*)et)) if (pthread_create(&th, NULL, event_executor_worker, (void*)et))
{ {
sql_print_error("Problem while trying to create a thread"); sql_print_error("Problem while trying to create a thread");
...@@ -267,29 +318,18 @@ event_executor_main(void *arg) ...@@ -267,29 +318,18 @@ event_executor_main(void *arg)
#else #else
event_executor_worker((void *) et); event_executor_worker((void *) et);
#endif #endif
printf("[%10s] exec at [%llu]\n", et->name.str,TIME_to_ulonglong_datetime(&et->execute_at));
et->mark_last_executed(); et->mark_last_executed();
thd->proc_info = "Computing next time";
et->compute_next_execution_time(); et->compute_next_execution_time();
printf("[%10s] next at [%llu]\n\n\n", et->name.str,TIME_to_ulonglong_datetime(&et->execute_at));
et->update_fields(thd); et->update_fields(thd);
if ((et->execute_at.year && !et->expression) if ((et->execute_at.year && !et->expression) ||
|| TIME_to_ulonglong_datetime(&et->execute_at) == 0L) TIME_to_ulonglong_datetime(&et->execute_at) == 0L)
et->flags |= EVENT_EXEC_NO_MORE; et->flags |= EVENT_EXEC_NO_MORE;
} }
}
/*
Let's remove elements which won't be executed any more
The number is "i" and it is <= up to evex_executing_queue.elements
*/
j= 0;
while (j < i && j < evex_executing_queue.elements)
{
event_timed *et= *dynamic_element(&evex_executing_queue, j, event_timed**);
if ((et->flags & EVENT_EXEC_NO_MORE) || et->status == MYSQL_EVENT_DISABLED) if ((et->flags & EVENT_EXEC_NO_MORE) || et->status == MYSQL_EVENT_DISABLED)
{ {
delete_dynamic_element(&evex_executing_queue, j); evex_queue_delete_element(&EVEX_EQ_NAME, 1);// 1 is top
DBUG_PRINT("EVEX main thread", ("DELETING FROM EXECUTION QUEUE [%s.%s]",
et->dbname.str, et->name.str));
// nulling the position, will delete later
if (et->dropped) if (et->dropped)
{ {
// we have to drop the event // we have to drop the event
...@@ -299,25 +339,17 @@ event_executor_main(void *arg) ...@@ -299,25 +339,17 @@ event_executor_main(void *arg)
DBUG_ASSERT(idx != -1); DBUG_ASSERT(idx != -1);
delete_dynamic_element(&events_array, idx); delete_dynamic_element(&events_array, idx);
} }
continue; } else
} evex_queue_first_updated(&EVEX_EQ_NAME);
++j;
}
if (evex_executing_queue.elements)
//ToDo Andrey : put a lock here
qsort((gptr) dynamic_element(&evex_executing_queue, 0, event_timed**),
evex_executing_queue.elements,
sizeof(event_timed **),
(qsort_cmp) event_timed_compare
);
VOID(pthread_mutex_unlock(&LOCK_event_arrays)); VOID(pthread_mutex_unlock(&LOCK_event_arrays));
} }// while
err: err:
// First manifest that this thread does not work and then destroy // First manifest that this thread does not work and then destroy
VOID(pthread_mutex_lock(&LOCK_evex_running)); VOID(pthread_mutex_lock(&LOCK_evex_running));
evex_is_running= false; evex_is_running= false;
evex_main_thread_id= 0;
VOID(pthread_mutex_unlock(&LOCK_evex_running)); VOID(pthread_mutex_unlock(&LOCK_evex_running));
sql_print_information("Event scheduler stopping"); sql_print_information("Event scheduler stopping");
...@@ -341,7 +373,8 @@ err: ...@@ -341,7 +373,8 @@ err:
VOID(pthread_mutex_lock(&LOCK_event_arrays)); VOID(pthread_mutex_lock(&LOCK_event_arrays));
// No need to use lock here if EVEX is not running but anyway // No need to use lock here if EVEX is not running but anyway
delete_dynamic(&evex_executing_queue); delete_queue(&executing_queue);
evex_queue_destroy(&EVEX_EQ_NAME);
delete_dynamic(&events_array); delete_dynamic(&events_array);
VOID(pthread_mutex_unlock(&LOCK_event_arrays)); VOID(pthread_mutex_unlock(&LOCK_event_arrays));
...@@ -353,8 +386,10 @@ err: ...@@ -353,8 +386,10 @@ err:
pthread_mutex_lock(&LOCK_thread_count); pthread_mutex_lock(&LOCK_thread_count);
thread_count--; thread_count--;
thread_running--; thread_running--;
#ifndef DBUG_FAULTY_THR
THD_CHECK_SENTRY(thd); THD_CHECK_SENTRY(thd);
delete thd; delete thd;
#endif
pthread_mutex_unlock(&LOCK_thread_count); pthread_mutex_unlock(&LOCK_thread_count);
...@@ -366,8 +401,10 @@ err_no_thd: ...@@ -366,8 +401,10 @@ err_no_thd:
free_root(&evex_mem_root, MYF(0)); free_root(&evex_mem_root, MYF(0));
sql_print_information("Event scheduler stopped"); sql_print_information("Event scheduler stopped");
#ifndef DBUG_FAULTY_THR
my_thread_end(); my_thread_end();
pthread_exit(0); pthread_exit(0);
#endif
DBUG_RETURN(0);// Can't return anything here DBUG_RETURN(0);// Can't return anything here
} }
...@@ -386,6 +423,7 @@ event_executor_worker(void *event_void) ...@@ -386,6 +423,7 @@ event_executor_worker(void *event_void)
init_alloc_root(&worker_mem_root, MEM_ROOT_BLOCK_SIZE, MEM_ROOT_PREALLOC); init_alloc_root(&worker_mem_root, MEM_ROOT_BLOCK_SIZE, MEM_ROOT_PREALLOC);
#ifndef DBUG_FAULTY_THR
my_thread_init(); my_thread_init();
if (!(thd = new THD)) // note that contructor of THD uses DBUG_ ! if (!(thd = new THD)) // note that contructor of THD uses DBUG_ !
...@@ -411,6 +449,9 @@ event_executor_worker(void *event_void) ...@@ -411,6 +449,9 @@ event_executor_worker(void *event_void)
thread_count++; thread_count++;
thread_running++; thread_running++;
VOID(pthread_mutex_unlock(&LOCK_thread_count)); VOID(pthread_mutex_unlock(&LOCK_thread_count));
#else
thd= current_thd;
#endif
// thd->security_ctx->priv_host is char[MAX_HOSTNAME] // thd->security_ctx->priv_host is char[MAX_HOSTNAME]
...@@ -420,6 +461,8 @@ event_executor_worker(void *event_void) ...@@ -420,6 +461,8 @@ event_executor_worker(void *event_void)
thd->security_ctx->priv_user= event->definer_user.str; thd->security_ctx->priv_user= event->definer_user.str;
thd->db= event->dbname.str; thd->db= event->dbname.str;
if (!check_access(thd, EVENT_ACL, event->dbname.str, 0, 0, 0,
is_schema_db(event->dbname.str)))
{ {
char exec_time[200]; char exec_time[200];
int ret; int ret;
...@@ -434,6 +477,7 @@ event_executor_worker(void *event_void) ...@@ -434,6 +477,7 @@ event_executor_worker(void *event_void)
err: err:
VOID(pthread_mutex_lock(&LOCK_thread_count)); VOID(pthread_mutex_lock(&LOCK_thread_count));
#ifndef DBUG_FAULTY_THR
thread_count--; thread_count--;
thread_running--; thread_running--;
/* /*
...@@ -451,6 +495,7 @@ err: ...@@ -451,6 +495,7 @@ err:
VOID(pthread_mutex_lock(&LOCK_thread_count)); VOID(pthread_mutex_lock(&LOCK_thread_count));
THD_CHECK_SENTRY(thd); THD_CHECK_SENTRY(thd);
delete thd; delete thd;
#endif
VOID(pthread_mutex_unlock(&LOCK_thread_count)); VOID(pthread_mutex_unlock(&LOCK_thread_count));
err_no_thd: err_no_thd:
...@@ -502,6 +547,12 @@ evex_load_events_from_db(THD *thd) ...@@ -502,6 +547,12 @@ evex_load_events_from_db(THD *thd)
"Table probably corrupted"); "Table probably corrupted");
goto end; goto end;
} }
if (et->status != MYSQL_EVENT_ENABLED)
{
DBUG_PRINT("evex_load_events_from_db",("Event %s is disabled", et->name.str));
delete et;
continue;
}
DBUG_PRINT("evex_load_events_from_db", DBUG_PRINT("evex_load_events_from_db",
("Event %s loaded from row. Time to compile", et->name.str)); ("Event %s loaded from row. Time to compile", et->name.str));
...@@ -515,8 +566,7 @@ evex_load_events_from_db(THD *thd) ...@@ -515,8 +566,7 @@ evex_load_events_from_db(THD *thd)
// let's find when to be executed // let's find when to be executed
et->compute_next_execution_time(); et->compute_next_execution_time();
DBUG_PRINT("evex_load_events_from_db", DBUG_PRINT("evex_load_events_from_db", ("Adding to the exec list."));
("Adding %s to the executor list.", et->name.str));
VOID(push_dynamic(&events_array,(gptr) et)); VOID(push_dynamic(&events_array,(gptr) et));
/* /*
We always add at the end so the number of elements - 1 is the place We always add at the end so the number of elements - 1 is the place
...@@ -526,23 +576,21 @@ evex_load_events_from_db(THD *thd) ...@@ -526,23 +576,21 @@ evex_load_events_from_db(THD *thd)
*/ */
et_copy= dynamic_element(&events_array, events_array.elements - 1, et_copy= dynamic_element(&events_array, events_array.elements - 1,
event_timed*); event_timed*);
VOID(push_dynamic(&evex_executing_queue,(gptr) &et_copy));
evex_queue_insert(&EVEX_EQ_NAME, (EVEX_PTOQEL) et_copy);
printf("%p %s\n", et_copy, et_copy->name.str);
et->free_sphead_on_delete= false; et->free_sphead_on_delete= false;
delete et; delete et;
} }
end_read_record(&read_record_info);
qsort((gptr) dynamic_element(&evex_executing_queue, 0, event_timed**), ret= 0;
evex_executing_queue.elements,
sizeof(event_timed **), end:
(qsort_cmp) event_timed_compare
);
VOID(pthread_mutex_unlock(&LOCK_event_arrays)); VOID(pthread_mutex_unlock(&LOCK_event_arrays));
end_read_record(&read_record_info);
thd->version--; // Force close to free memory thd->version--; // Force close to free memory
ret= 0;
end:
close_thread_tables(thd); close_thread_tables(thd);
DBUG_PRINT("info", ("Finishing with status code %d", ret)); DBUG_PRINT("info", ("Finishing with status code %d", ret));
......
...@@ -16,8 +16,11 @@ ...@@ -16,8 +16,11 @@
#ifndef _EVENT_PRIV_H_ #ifndef _EVENT_PRIV_H_
#define _EVENT_PRIV_H_ #define _EVENT_PRIV_H_
#include "mysql_priv.h"
#define EVEX_USE_QUEUE
#define UNLOCK_MUTEX_AND_BAIL_OUT(__mutex, __label) \ #define UNLOCK_MUTEX_AND_BAIL_OUT(__mutex, __label) \
{ VOID(pthread_mutex_unlock(&__mutex)); goto __label; } { VOID(pthread_mutex_unlock(&__mutex)); goto __label; }
...@@ -41,14 +44,6 @@ enum ...@@ -41,14 +44,6 @@ enum
EVEX_FIELD_COUNT /* a cool trick to count the number of fields :) */ EVEX_FIELD_COUNT /* a cool trick to count the number of fields :) */
}; };
extern bool evex_is_running;
extern bool mysql_event_table_exists;
extern DYNAMIC_ARRAY events_array;
extern DYNAMIC_ARRAY evex_executing_queue;
extern MEM_ROOT evex_mem_root;
extern pthread_mutex_t LOCK_event_arrays,
LOCK_workers_count,
LOCK_evex_running;
int int
...@@ -60,4 +55,71 @@ evex_db_find_event_aux(THD *thd, const LEX_STRING dbname, ...@@ -60,4 +55,71 @@ evex_db_find_event_aux(THD *thd, const LEX_STRING dbname,
TABLE * TABLE *
evex_open_event_table(THD *thd, enum thr_lock_type lock_type); evex_open_event_table(THD *thd, enum thr_lock_type lock_type);
int
event_timed_compare_q(void *vptr, byte* a, byte *b);
int
evex_time_diff(TIME *a, TIME *b);
#define EXEC_QUEUE_QUEUE_NAME executing_queue
#define EXEC_QUEUE_DARR_NAME evex_executing_queue
#ifdef EVEX_USE_QUEUE
#define EVEX_QUEUE_TYPE QUEUE
#define EVEX_PTOQEL byte *
#define EVEX_EQ_NAME executing_queue
#define evex_queue_first_element(queue, __cast) ((__cast)queue_top(queue))
#define evex_queue_element(queue, idx, __cast) ((__cast)queue_top(queue))
#define evex_queue_delete_element(queue, idx) queue_remove(queue, idx)
#define evex_queue_destroy(queue) delete_queue(queue)
#define evex_queue_first_updated(queue) queue_replaced(queue)
#define evex_queue_insert(queue, element) queue_insert_safe(queue, element);
#else
#define EVEX_QUEUE_TYPE DYNAMIC_ARRAY
#define EVEX_PTOQEL gptr
#define EVEX_EQ_NAME evex_executing_queue
#define evex_queue_element(queue, idx, __cast) dynamic_element(queue,idx, __cast)
#define evex_queue_delete_element(queue, idx) delete_dynamic_element(queue, idx);
#define evex_queue_destroy(queue) delete_dynamic(queue)
/*
push_dynamic() expects ptr to the memory to put in, to make things fast
so when a pointer has to be put inside a ptr-to-ptr is being passed
*/
#define evex_queue_first_updated(queue)
#define evex_queue_insert(queue, element) VOID(push_dynamic(queue, &element))
#endif
void
evex_queue_init(EVEX_QUEUE_TYPE *queue);
int
evex_queue_insert2(EVEX_QUEUE_TYPE *queue, EVEX_PTOQEL element);
void
evex_queue_sort(EVEX_QUEUE_TYPE *queue);
#define evex_queue_num_elements(queue) queue.elements
extern bool evex_is_running;
extern bool mysql_event_table_exists;
extern DYNAMIC_ARRAY events_array;
extern DYNAMIC_ARRAY EXEC_QUEUE_DARR_NAME;
extern QUEUE EXEC_QUEUE_QUEUE_NAME;
extern MEM_ROOT evex_mem_root;
extern pthread_mutex_t LOCK_event_arrays,
LOCK_workers_count,
LOCK_evex_running;
extern ulonglong evex_main_thread_id;
#endif /* _EVENT_PRIV_H_ */ #endif /* _EVENT_PRIV_H_ */
...@@ -14,9 +14,8 @@ ...@@ -14,9 +14,8 @@
along with this program; if not, write to the Free Software along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include "mysql_priv.h"
#include "event.h"
#include "event_priv.h" #include "event_priv.h"
#include "event.h"
#include "sp.h" #include "sp.h"
...@@ -789,7 +788,7 @@ event_timed::get_show_create_event(THD *thd, uint *length) ...@@ -789,7 +788,7 @@ event_timed::get_show_create_event(THD *thd, uint *length)
+ strlen("DO ") + + strlen("DO ") +
+ body.length + strlen(";"); + body.length + strlen(";");
ret= dst= (char*) alloc_root(thd->mem_root, len); ret= dst= (char*) alloc_root(thd->mem_root, len + 1);
memcpy(dst, "CREATE EVENT ", tmp_len= strlen("CREATE EVENT ")); memcpy(dst, "CREATE EVENT ", tmp_len= strlen("CREATE EVENT "));
dst+= tmp_len; dst+= tmp_len;
memcpy(dst, dbname.str, tmp_len=dbname.length); memcpy(dst, dbname.str, tmp_len=dbname.length);
...@@ -832,7 +831,7 @@ event_timed::get_show_create_event(THD *thd, uint *length) ...@@ -832,7 +831,7 @@ event_timed::get_show_create_event(THD *thd, uint *length)
*dst= '\0'; *dst= '\0';
*length= len; *length= len;
dst[len]= '\0';
return ret; return ret;
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment