Skip to content

M
mariadb
Commit 04a99458 authored by malff/marcsql@weblab.(none)'s avatar malff/marcsql@weblab.(none)
Browse files
Options

Merge malff@bk-internal.mysql.com:/home/bk/mysql-5.1-runtime 

into  weblab.(none):/home/marcsql/TREE/mysql-5.1-24392
parents 85fb7c8d c06c7356
Expand all Show whitespace changes
Inline Side-by-side
Showing with 404 additions and 368 deletions
sql/event_data_objects.cc
View file @ 04a99458
...@@ -20,6 +20,8 @@ ...@@ -20,6 +20,8 @@
#include "event_db_repository.h" #include "event_db_repository.h"
#include "sp_head.h" #include "sp_head.h"
/* That's a provisional solution */
extern Event_db_repository events_event_db_repository;
#define EVEX_MAX_INTERVAL_VALUE 1000000000L #define EVEX_MAX_INTERVAL_VALUE 1000000000L
...@@ -30,6 +32,47 @@ event_change_security_context(THD *thd, LEX_STRING user, LEX_STRING host, ...@@ -30,6 +32,47 @@ event_change_security_context(THD *thd, LEX_STRING user, LEX_STRING host,
static void static void
event_restore_security_context(THD *thd, Security_context *backup); event_restore_security_context(THD *thd, Security_context *backup);
/*
Initiliazes dbname and name of an Event_queue_element_for_exec
object
SYNOPSIS
Event_queue_element_for_exec::init()
RETURN VALUE
FALSE OK
TRUE Error (OOM)
*/
bool
Event_queue_element_for_exec::init(LEX_STRING db, LEX_STRING n)
{
if (!(dbname.str= my_strndup(db.str, dbname.length= db.length, MYF(MY_WME))))
return TRUE;
if (!(name.str= my_strndup(n.str, name.length= n.length, MYF(MY_WME))))
{
my_free((gptr) dbname.str, MYF(0));
return TRUE;
}
return FALSE;
}
/*
Destructor
SYNOPSIS
Event_queue_element_for_exec::~Event_queue_element_for_exec()
*/
Event_queue_element_for_exec::~Event_queue_element_for_exec()
{
my_free((gptr) dbname.str, MYF(0));
my_free((gptr) name.str, MYF(0));
}
/* /*
Returns a new instance Returns a new instance
...@@ -743,7 +786,7 @@ Event_timed::~Event_timed() ...@@ -743,7 +786,7 @@ Event_timed::~Event_timed()
*/ */
Event_job_data::Event_job_data() Event_job_data::Event_job_data()
:thd(NULL), sphead(NULL), sql_mode(0) :sphead(NULL), sql_mode(0)
{ {
} }
...@@ -1239,6 +1282,7 @@ Event_queue_element::compute_next_execution_time() ...@@ -1239,6 +1282,7 @@ Event_queue_element::compute_next_execution_time()
DBUG_PRINT("info", ("Dropped: %d", dropped)); DBUG_PRINT("info", ("Dropped: %d", dropped));
status= Event_queue_element::DISABLED; status= Event_queue_element::DISABLED;
status_changed= TRUE; status_changed= TRUE;
dropped= TRUE;
goto ret; goto ret;
} }
...@@ -1446,32 +1490,6 @@ Event_queue_element::mark_last_executed(THD *thd) ...@@ -1446,32 +1490,6 @@ Event_queue_element::mark_last_executed(THD *thd)
} }
/*
Drops the event
SYNOPSIS
Event_queue_element::drop()
thd thread context
RETURN VALUE
0 OK
-1 Cannot open mysql.event
-2 Cannot find the event in mysql.event (already deleted?)
others return code from SE in case deletion of the event row
failed.
*/
int
Event_queue_element::drop(THD *thd)
{
DBUG_ENTER("Event_queue_element::drop");
DBUG_RETURN(Events::get_instance()->
drop_event(thd, dbname, name, FALSE, TRUE));
}
/* /*
Saves status and last_executed_at to the disk if changed. Saves status and last_executed_at to the disk if changed.
...@@ -1503,13 +1521,13 @@ Event_queue_element::update_timing_fields(THD *thd) ...@@ -1503,13 +1521,13 @@ Event_queue_element::update_timing_fields(THD *thd)
thd->reset_n_backup_open_tables_state(&backup); thd->reset_n_backup_open_tables_state(&backup);
if (Events::get_instance()->open_event_table(thd, TL_WRITE, &table)) if (events_event_db_repository.open_event_table(thd, TL_WRITE, &table))
{ {
ret= TRUE; ret= TRUE;
goto done; goto done;
} }
fields= table->field; fields= table->field;
if ((ret= Events::get_instance()->db_repository-> if ((ret= events_event_db_repository.
find_named_event(thd, dbname, name, table))) find_named_event(thd, dbname, name, table)))
goto done; goto done;
......
sql/event_data_objects.h
View file @ 04a99458
...@@ -27,6 +27,27 @@ class sp_head; ...@@ -27,6 +27,27 @@ class sp_head;
class Sql_alloc; class Sql_alloc;
class Event_queue_element_for_exec
{
public:
Event_queue_element_for_exec(){};
~Event_queue_element_for_exec();
bool
init(LEX_STRING dbname, LEX_STRING name);
LEX_STRING dbname;
LEX_STRING name;
bool dropped;
THD *thd;
private:
/* Prevent use of these */
Event_queue_element_for_exec(const Event_queue_element_for_exec &);
void operator=(Event_queue_element_for_exec &);
};
class Event_basic class Event_basic
{ {
protected: protected:
...@@ -96,9 +117,6 @@ public: ...@@ -96,9 +117,6 @@ public:
bool bool
compute_next_execution_time(); compute_next_execution_time();
int
drop(THD *thd);
void void
mark_last_executed(THD *thd); mark_last_executed(THD *thd);
...@@ -160,7 +178,6 @@ public: ...@@ -160,7 +178,6 @@ public:
class Event_job_data : public Event_basic class Event_job_data : public Event_basic
{ {
public: public:
THD *thd;
sp_head *sphead; sp_head *sphead;
LEX_STRING body; LEX_STRING body;
......
sql/event_queue.cc
View file @ 04a99458
This diff is collapsed.
sql/event_queue.h
View file @ 04a99458
...@@ -16,12 +16,10 @@ ...@@ -16,12 +16,10 @@
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
class Event_basic; class Event_basic;
class Event_db_repository;
class Event_job_data;
class Event_queue_element; class Event_queue_element;
class Event_queue_element_for_exec;
class THD; class THD;
class Event_scheduler;
class Event_queue class Event_queue
{ {
...@@ -35,19 +33,19 @@ public: ...@@ -35,19 +33,19 @@ public:
deinit_mutexes(); deinit_mutexes();
bool bool
init_queue(THD *thd, Event_db_repository *db_repo); init_queue(THD *thd);
void void
deinit_queue(); deinit_queue();
/* Methods for queue management follow */ /* Methods for queue management follow */
int void
create_event(THD *thd, LEX_STRING dbname, LEX_STRING name); create_event(THD *thd, Event_queue_element *new_element);
int void
update_event(THD *thd, LEX_STRING dbname, LEX_STRING name, update_event(THD *thd, LEX_STRING dbname, LEX_STRING name,
LEX_STRING *new_schema, LEX_STRING *new_name); Event_queue_element *new_element);
void void
drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name); drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name);
...@@ -59,14 +57,15 @@ public: ...@@ -59,14 +57,15 @@ public:
recalculate_activation_times(THD *thd); recalculate_activation_times(THD *thd);
bool bool
get_top_for_execution_if_time(THD *thd, Event_job_data **job_data); get_top_for_execution_if_time(THD *thd,
Event_queue_element_for_exec **event_name);
void void
dump_internal_status(); dump_internal_status();
int void
load_events_from_db(THD *thd); empty_queue();
protected: protected:
void void
find_n_remove_event(LEX_STRING db, LEX_STRING name); find_n_remove_event(LEX_STRING db, LEX_STRING name);
...@@ -76,8 +75,6 @@ protected: ...@@ -76,8 +75,6 @@ protected:
drop_matching_events(THD *thd, LEX_STRING pattern, drop_matching_events(THD *thd, LEX_STRING pattern,
bool (*)(LEX_STRING, Event_basic *)); bool (*)(LEX_STRING, Event_basic *));
void
empty_queue();
void void
dbug_dump_queue(time_t now); dbug_dump_queue(time_t now);
...@@ -86,11 +83,7 @@ protected: ...@@ -86,11 +83,7 @@ protected:
pthread_mutex_t LOCK_event_queue; pthread_mutex_t LOCK_event_queue;
pthread_cond_t COND_queue_state; pthread_cond_t COND_queue_state;
Event_db_repository *db_repository; /* The sorted queue with the Event_queue_element objects */
Event_scheduler *scheduler;
/* The sorted queue with the Event_job_data objects */
QUEUE queue; QUEUE queue;
TIME next_activation_at; TIME next_activation_at;
......
sql/event_scheduler.cc
View file @ 04a99458
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#include "event_data_objects.h" #include "event_data_objects.h"
#include "event_scheduler.h" #include "event_scheduler.h"
#include "event_queue.h" #include "event_queue.h"
#include "event_db_repository.h"
#ifdef __GNUC__ #ifdef __GNUC__
#if __GNUC__ >= 2 #if __GNUC__ >= 2
...@@ -34,6 +35,11 @@ ...@@ -34,6 +35,11 @@
extern pthread_attr_t connection_attrib; extern pthread_attr_t connection_attrib;
Event_db_repository *Event_worker_thread::db_repository;
Events *Event_worker_thread::events_facade;
static static
const LEX_STRING scheduler_states_names[] = const LEX_STRING scheduler_states_names[] =
{ {
...@@ -60,8 +66,8 @@ struct scheduler_param { ...@@ -60,8 +66,8 @@ struct scheduler_param {
et The event itself et The event itself
*/ */
static void void
evex_print_warnings(THD *thd, Event_job_data *et) Event_worker_thread::print_warnings(THD *thd, Event_job_data *et)
{ {
MYSQL_ERROR *err; MYSQL_ERROR *err;
DBUG_ENTER("evex_print_warnings"); DBUG_ENTER("evex_print_warnings");
...@@ -253,49 +259,97 @@ event_worker_thread(void *arg) ...@@ -253,49 +259,97 @@ event_worker_thread(void *arg)
{ {
/* needs to be first for thread_stack */ /* needs to be first for thread_stack */
THD *thd; THD *thd;
Event_job_data *event= (Event_job_data *)arg; Event_queue_element_for_exec *event= (Event_queue_element_for_exec *)arg;
int ret;
thd= event->thd; thd= event->thd;
thd->thread_stack= (char *) &thd; // remember where our stack is thd->thread_stack= (char *) &thd; // remember where our stack is
DBUG_ENTER("event_worker_thread");
if (!post_init_event_thread(thd)) Event_worker_thread worker_thread;
worker_thread.run(thd, (Event_queue_element_for_exec *)arg);
deinit_event_thread(thd);
return 0; // Can't return anything here
}
/*
Function that executes an event in a child thread. Setups the
environment for the event execution and cleans after that.
SYNOPSIS
Event_worker_thread::run()
thd Thread context
event The Event_queue_element_for_exec object to be processed
*/
void
Event_worker_thread::run(THD *thd, Event_queue_element_for_exec *event)
{
int ret;
Event_job_data *job_data= NULL;
DBUG_ENTER("Event_worker_thread::run");
DBUG_PRINT("info", ("Baikonur, time is %d, BURAN reporting and operational."
"THD=0x%lx", time(NULL), thd));
if (post_init_event_thread(thd))
goto end;
if (!(job_data= new Event_job_data()))
goto end;
else if ((ret= db_repository->
load_named_event(thd, event->dbname, event->name, job_data)))
{ {
DBUG_PRINT("info", ("Baikonur, time is %ld, BURAN reporting and operational." DBUG_PRINT("error", ("Got %d from load_named_event", ret));
"THD: 0x%lx", goto end;
(long) time(NULL), (long) thd)); }
sql_print_information("SCHEDULER: [%s.%s of %s] executing in thread %lu. " sql_print_information("SCHEDULER: [%s.%s of %s] executing in thread %lu. ",
"Execution %u", job_data->dbname.str, job_data->name.str,
event->dbname.str, event->name.str, job_data->definer.str, thd->thread_id);
event->definer.str, thd->thread_id,
event->execution_count);
thd->enable_slow_log= TRUE; thd->enable_slow_log= TRUE;
ret= event->execute(thd); ret= job_data->execute(thd);
evex_print_warnings(thd, event); print_warnings(thd, job_data);
sql_print_information("SCHEDULER: [%s.%s of %s] executed in thread %lu. " sql_print_information("SCHEDULER: [%s.%s of %s] executed in thread %lu. "
"RetCode=%d", event->dbname.str, event->name.str, "RetCode=%d", job_data->dbname.str, job_data->name.str,
event->definer.str, thd->thread_id, ret); job_data->definer.str, thd->thread_id, ret);
if (ret == EVEX_COMPILE_ERROR) if (ret == EVEX_COMPILE_ERROR)
sql_print_information("SCHEDULER: COMPILE ERROR for event %s.%s of %s", sql_print_information("SCHEDULER: COMPILE ERROR for event %s.%s of %s",
event->dbname.str, event->name.str, job_data->dbname.str, job_data->name.str,
event->definer.str); job_data->definer.str);
else if (ret == EVEX_MICROSECOND_UNSUP) else if (ret == EVEX_MICROSECOND_UNSUP)
sql_print_information("SCHEDULER: MICROSECOND is not supported"); end:
delete job_data;
if (event->dropped)
{
sql_print_information("SCHEDULER: Dropping %s.%s", event->dbname.str,
event->name.str);
/*
Using db_repository can lead to a race condition because we access
the table without holding LOCK_metadata.
Scenario:
1. CREATE EVENT xyz AT ... (conn thread)
2. execute xyz (worker)
3. CREATE EVENT XYZ EVERY ... (conn thread)
4. drop xyz (worker)
5. XYZ was just created on disk but `drop xyz` of the worker dropped it.
A consequent load to create Event_queue_element will fail.
If all operations are performed under LOCK_metadata there is no such
problem. However, this comes at the price of introduction bi-directional
association between class Events and class Event_worker_thread.
*/
events_facade->drop_event(thd, event->dbname, event->name, FALSE);
} }
DBUG_PRINT("info", ("BURAN %s.%s is landing!", event->dbname.str, DBUG_PRINT("info", ("BURAN %s.%s is landing!", event->dbname.str,
event->name.str)); event->name.str));
delete event;
deinit_event_thread(thd); delete event;
DBUG_RETURN(0); // Can't return anything here
} }
...@@ -441,7 +495,6 @@ bool ...@@ -441,7 +495,6 @@ bool
Event_scheduler::run(THD *thd) Event_scheduler::run(THD *thd)
{ {
int res= FALSE; int res= FALSE;
Event_job_data *job_data;
DBUG_ENTER("Event_scheduler::run"); DBUG_ENTER("Event_scheduler::run");
sql_print_information("SCHEDULER: Manager thread started with id %lu", sql_print_information("SCHEDULER: Manager thread started with id %lu",
...@@ -454,18 +507,20 @@ Event_scheduler::run(THD *thd) ...@@ -454,18 +507,20 @@ Event_scheduler::run(THD *thd)
while (is_running()) while (is_running())
{ {
Event_queue_element_for_exec *event_name;
/* Gets a minimized version */ /* Gets a minimized version */
if (queue->get_top_for_execution_if_time(thd, &job_data)) if (queue->get_top_for_execution_if_time(thd, &event_name))
{ {
sql_print_information("SCHEDULER: Serious error during getting next " sql_print_information("SCHEDULER: Serious error during getting next "
"event to execute. Stopping"); "event to execute. Stopping");
break; break;
} }
DBUG_PRINT("info", ("get_top returned job_data: 0x%lx", (long) job_data)); DBUG_PRINT("info", ("get_top returned job_data=0x%lx", event_name));
if (job_data) if (event_name)
{ {
if ((res= execute_top(thd, job_data))) if ((res= execute_top(thd, event_name)))
break; break;
} }
else else
...@@ -499,7 +554,7 @@ Event_scheduler::run(THD *thd) ...@@ -499,7 +554,7 @@ Event_scheduler::run(THD *thd)
*/ */
bool bool
Event_scheduler::execute_top(THD *thd, Event_job_data *job_data) Event_scheduler::execute_top(THD *thd, Event_queue_element_for_exec *event_name)
{ {
THD *new_thd; THD *new_thd;
pthread_t th; pthread_t th;
...@@ -510,13 +565,13 @@ Event_scheduler::execute_top(THD *thd, Event_job_data *job_data) ...@@ -510,13 +565,13 @@ Event_scheduler::execute_top(THD *thd, Event_job_data *job_data)
pre_init_event_thread(new_thd); pre_init_event_thread(new_thd);
new_thd->system_thread= SYSTEM_THREAD_EVENT_WORKER; new_thd->system_thread= SYSTEM_THREAD_EVENT_WORKER;
job_data->thd= new_thd; event_name->thd= new_thd;
DBUG_PRINT("info", ("BURAN %s@%s ready for start t-3..2..1..0..ignition", DBUG_PRINT("info", ("BURAN %s@%s ready for start t-3..2..1..0..ignition",
job_data->dbname.str, job_data->name.str)); event_name->dbname.str, event_name->name.str));
/* Major failure */ /* Major failure */
if ((res= pthread_create(&th, &connection_attrib, event_worker_thread, if ((res= pthread_create(&th, &connection_attrib, event_worker_thread,
job_data))) event_name)))
goto error; goto error;
++started_events; ++started_events;
...@@ -537,7 +592,7 @@ error: ...@@ -537,7 +592,7 @@ error:
delete new_thd; delete new_thd;
pthread_mutex_unlock(&LOCK_thread_count); pthread_mutex_unlock(&LOCK_thread_count);
} }
delete job_data; delete event_name;
DBUG_RETURN(TRUE); DBUG_RETURN(TRUE);
} }
......
sql/event_scheduler.h
View file @ 04a99458
...@@ -15,8 +15,11 @@ ...@@ -15,8 +15,11 @@
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 */
class Event_queue; class Event_queue;
class Event_job_data; class Event_job_data;
class Event_db_repository;
class Events;
void void
pre_init_event_thread(THD* thd); pre_init_event_thread(THD* thd);
...@@ -27,6 +30,29 @@ post_init_event_thread(THD* thd); ...@@ -27,6 +30,29 @@ post_init_event_thread(THD* thd);
void void
deinit_event_thread(THD *thd); deinit_event_thread(THD *thd);
class Event_worker_thread
{
public:
static void
init(Events *events, Event_db_repository *db_repo)
{
db_repository= db_repo;
events_facade= events;
}
void
run(THD *thd, Event_queue_element_for_exec *event);
private:
void
print_warnings(THD *thd, Event_job_data *et);
static Event_db_repository *db_repository;
static Events *events_facade;
};
class Event_scheduler class Event_scheduler
{ {
public: public:
...@@ -71,10 +97,9 @@ private: ...@@ -71,10 +97,9 @@ private:
uint uint
workers_count(); workers_count();
/* helper functions */ /* helper functions */
bool bool
execute_top(THD *thd, Event_job_data *job_data); execute_top(THD *thd, Event_queue_element_for_exec *event_name);
/* helper functions for working with mutexes & conditionals */ /* helper functions for working with mutexes & conditionals */
void void
......
sql/events.cc
View file @ 04a99458
...@@ -97,7 +97,7 @@ Event_queue events_event_queue; ...@@ -97,7 +97,7 @@ Event_queue events_event_queue;
static static
Event_scheduler events_event_scheduler; Event_scheduler events_event_scheduler;
static
Event_db_repository events_event_db_repository; Event_db_repository events_event_db_repository;
Events Events::singleton; Events Events::singleton;
...@@ -295,29 +295,6 @@ Events::Events() ...@@ -295,29 +295,6 @@ Events::Events()
} }
/*
Opens mysql.event table with specified lock
SYNOPSIS
Events::open_event_table()
thd Thread context
lock_type How to lock the table
table We will store the open table here
RETURN VALUE
1 Cannot lock table
2 The table is corrupted - different number of fields
0 OK
*/
int
Events::open_event_table(THD *thd, enum thr_lock_type lock_type,
TABLE **table)
{
return db_repository->open_event_table(thd, lock_type, table);
}
/* /*
The function exported to the world for creating of events. The function exported to the world for creating of events.
...@@ -351,16 +328,24 @@ Events::create_event(THD *thd, Event_parse_data *parse_data, bool if_not_exists) ...@@ -351,16 +328,24 @@ Events::create_event(THD *thd, Event_parse_data *parse_data, bool if_not_exists)
/* On error conditions my_error() is called so no need to handle here */ /* On error conditions my_error() is called so no need to handle here */
if (!(ret= db_repository->create_event(thd, parse_data, if_not_exists))) if (!(ret= db_repository->create_event(thd, parse_data, if_not_exists)))
{ {
if ((ret= event_queue->create_event(thd, parse_data->dbname, Event_queue_element *new_element;
parse_data->name)))
if (!(new_element= new Event_queue_element()))
ret= TRUE; // OOM
else if ((ret= db_repository->load_named_event(thd, parse_data->dbname,
parse_data->name,
new_element)))
{ {
DBUG_ASSERT(ret == OP_LOAD_ERROR); DBUG_ASSERT(ret == OP_LOAD_ERROR);
my_error(ER_EVENT_MODIFY_QUEUE_ERROR, MYF(0)); delete new_element;
} }
else
event_queue->create_event(thd, new_element);
} }
pthread_mutex_unlock(&LOCK_event_metadata); pthread_mutex_unlock(&LOCK_event_metadata);
DBUG_RETURN(ret); DBUG_RETURN(ret);
} }
...@@ -387,6 +372,7 @@ bool ...@@ -387,6 +372,7 @@ bool
Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to) Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to)
{ {
int ret; int ret;
Event_queue_element *new_element;
DBUG_ENTER("Events::update_event"); DBUG_ENTER("Events::update_event");
LEX_STRING *new_dbname= rename_to ? &rename_to->m_db : NULL; LEX_STRING *new_dbname= rename_to ? &rename_to->m_db : NULL;
LEX_STRING *new_name= rename_to ? &rename_to->m_name : NULL; LEX_STRING *new_name= rename_to ? &rename_to->m_name : NULL;
...@@ -400,12 +386,20 @@ Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to) ...@@ -400,12 +386,20 @@ Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to)
/* On error conditions my_error() is called so no need to handle here */ /* On error conditions my_error() is called so no need to handle here */
if (!(ret= db_repository->update_event(thd, parse_data, new_dbname, new_name))) if (!(ret= db_repository->update_event(thd, parse_data, new_dbname, new_name)))
{ {
if ((ret= event_queue->update_event(thd, parse_data->dbname, LEX_STRING dbname= new_dbname ? *new_dbname : parse_data->dbname;
parse_data->name, new_dbname, new_name))) LEX_STRING name= new_name ? *new_name : parse_data->name;
if (!(new_element= new Event_queue_element()))
ret= TRUE; // OOM
else if ((ret= db_repository->load_named_event(thd, dbname, name,
new_element)))
{ {
DBUG_ASSERT(ret == OP_LOAD_ERROR); DBUG_ASSERT(ret == OP_LOAD_ERROR);
my_error(ER_EVENT_MODIFY_QUEUE_ERROR, MYF(0)); delete new_element;
} }
else
event_queue->update_event(thd, parse_data->dbname, parse_data->name,
new_element);
} }
pthread_mutex_unlock(&LOCK_event_metadata); pthread_mutex_unlock(&LOCK_event_metadata);
...@@ -423,10 +417,6 @@ Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to) ...@@ -423,10 +417,6 @@ Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to)
name [in] Event's name name [in] Event's name
if_exists [in] When set and the event does not exist => if_exists [in] When set and the event does not exist =>
warning onto the stack warning onto the stack
only_from_disk [in] Whether to remove the event from the queue too.
In case of Event_job_data::drop() it's needed to
do only disk drop because Event_queue will handle
removal from memory queue.
RETURN VALUE RETURN VALUE
FALSE OK FALSE OK
...@@ -434,8 +424,7 @@ Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to) ...@@ -434,8 +424,7 @@ Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to)
*/ */
bool bool
Events::drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists, Events::drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists)
bool only_from_disk)
{ {
int ret; int ret;
DBUG_ENTER("Events::drop_event"); DBUG_ENTER("Events::drop_event");
...@@ -448,10 +437,7 @@ Events::drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists, ...@@ -448,10 +437,7 @@ Events::drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists,
pthread_mutex_lock(&LOCK_event_metadata); pthread_mutex_lock(&LOCK_event_metadata);
/* On error conditions my_error() is called so no need to handle here */ /* On error conditions my_error() is called so no need to handle here */
if (!(ret= db_repository->drop_event(thd, dbname, name, if_exists))) if (!(ret= db_repository->drop_event(thd, dbname, name, if_exists)))
{
if (!only_from_disk)
event_queue->drop_event(thd, dbname, name); event_queue->drop_event(thd, dbname, name);
}
pthread_mutex_unlock(&LOCK_event_metadata); pthread_mutex_unlock(&LOCK_event_metadata);
DBUG_RETURN(ret); DBUG_RETURN(ret);
} }
...@@ -655,11 +641,12 @@ Events::init() ...@@ -655,11 +641,12 @@ Events::init()
} }
check_system_tables_error= FALSE; check_system_tables_error= FALSE;
if (event_queue->init_queue(thd, db_repository)) if (event_queue->init_queue(thd) || load_events_from_db(thd))
{ {
sql_print_error("SCHEDULER: Error while loading from disk."); sql_print_error("SCHEDULER: Error while loading from disk.");
goto end; goto end;
} }
scheduler->init_scheduler(event_queue); scheduler->init_scheduler(event_queue);
DBUG_ASSERT(opt_event_scheduler == Events::EVENTS_ON || DBUG_ASSERT(opt_event_scheduler == Events::EVENTS_ON ||
...@@ -667,6 +654,7 @@ Events::init() ...@@ -667,6 +654,7 @@ Events::init()
if (opt_event_scheduler == Events::EVENTS_ON) if (opt_event_scheduler == Events::EVENTS_ON)
res= scheduler->start(); res= scheduler->start();
Event_worker_thread::init(this, db_repository);
end: end:
delete thd; delete thd;
/* Remember that we don't have a THD */ /* Remember that we don't have a THD */
...@@ -903,3 +891,131 @@ Events::check_system_tables(THD *thd) ...@@ -903,3 +891,131 @@ Events::check_system_tables(THD *thd)
DBUG_RETURN(ret); DBUG_RETURN(ret);
} }
/*
Loads all ENABLED events from mysql.event into the prioritized
queue. Called during scheduler main thread initialization. Compiles
the events. Creates Event_queue_element instances for every ENABLED event
from mysql.event.
SYNOPSIS
Events::load_events_from_db()
thd Thread context. Used for memory allocation in some cases.
RETURN VALUE
0 OK
!0 Error (EVEX_OPEN_TABLE_FAILED, EVEX_MICROSECOND_UNSUP,
EVEX_COMPILE_ERROR) - in all these cases mysql.event was
tampered.
NOTES
Reports the error to the console
*/
int
Events::load_events_from_db(THD *thd)
{
TABLE *table;
READ_RECORD read_record_info;
int ret= -1;
uint count= 0;
bool clean_the_queue= TRUE;
DBUG_ENTER("Events::load_events_from_db");
DBUG_PRINT("enter", ("thd=0x%lx", thd));
if ((ret= db_repository->open_event_table(thd, TL_READ, &table)))
{
sql_print_error("SCHEDULER: Table mysql.event is damaged. Can not open");
DBUG_RETURN(EVEX_OPEN_TABLE_FAILED);
}
init_read_record(&read_record_info, thd, table ,NULL,1,0);
while (!(read_record_info.read_record(&read_record_info)))
{
Event_queue_element *et;
if (!(et= new Event_queue_element))
{
DBUG_PRINT("info", ("Out of memory"));
break;
}
DBUG_PRINT("info", ("Loading event from row."));
if ((ret= et->load_from_row(table)))
{
sql_print_error("SCHEDULER: Error while loading from mysql.event. "
"Table probably corrupted");
break;
}
if (et->status != Event_queue_element::ENABLED)
{
DBUG_PRINT("info",("%s is disabled",et->name.str));
delete et;
continue;
}
/* let's find when to be executed */
if (et->compute_next_execution_time())
{
sql_print_error("SCHEDULER: Error while computing execution time of %s.%s."
" Skipping", et->dbname.str, et->name.str);
continue;
}
{
Event_job_data temp_job_data;
DBUG_PRINT("info", ("Event %s loaded from row. ", et->name.str));
temp_job_data.load_from_row(table);
/*
We load only on scheduler root just to check whether the body
compiles.
*/
switch (ret= temp_job_data.compile(thd, thd->mem_root)) {
case EVEX_MICROSECOND_UNSUP:
sql_print_error("SCHEDULER: mysql.event is tampered. MICROSECOND is not "
"supported but found in mysql.event");
break;
case EVEX_COMPILE_ERROR:
sql_print_error("SCHEDULER: Error while compiling %s.%s. Aborting load",
et->dbname.str, et->name.str);
break;
default:
break;
}
thd->end_statement();
thd->cleanup_after_query();
}
if (ret)
{
delete et;
goto end;
}
DBUG_PRINT("load_events_from_db", ("Adding 0x%lx to the exec list."));
event_queue->create_event(thd, et);
count++;
}
clean_the_queue= FALSE;
end:
end_read_record(&read_record_info);
if (clean_the_queue)
{
event_queue->empty_queue();
ret= -1;
}
else
{
ret= 0;
sql_print_information("SCHEDULER: Loaded %d event%s", count,
(count == 1)?"":"s");
}
close_thread_tables(thd);
DBUG_PRINT("info", ("Status code %d. Loaded %d event(s)", ret, count));
DBUG_RETURN(ret);
}
sql/events.h
View file @ 04a99458
...@@ -42,13 +42,6 @@ sortcmp_lex_string(LEX_STRING s, LEX_STRING t, CHARSET_INFO *cs); ...@@ -42,13 +42,6 @@ sortcmp_lex_string(LEX_STRING s, LEX_STRING t, CHARSET_INFO *cs);
class Events class Events
{ {
public: public:
/*
Quite NOT the best practice and will be removed once
Event_timed::drop() and Event_timed is fixed not do drop directly
or other scheme will be found.
*/
friend class Event_queue_element;
/* The order should match the order in opt_typelib */ /* The order should match the order in opt_typelib */
enum enum_opt_event_scheduler enum enum_opt_event_scheduler
{ {
...@@ -92,15 +85,11 @@ public: ...@@ -92,15 +85,11 @@ public:
update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to); update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to);
bool bool
drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists, drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists);
bool only_from_disk);
void void
drop_schema_events(THD *thd, char *db); drop_schema_events(THD *thd, char *db);
int
open_event_table(THD *thd, enum thr_lock_type lock_type, TABLE **table);
bool bool
show_create_event(THD *thd, LEX_STRING dbname, LEX_STRING name); show_create_event(THD *thd, LEX_STRING dbname, LEX_STRING name);
...@@ -119,6 +108,9 @@ private: ...@@ -119,6 +108,9 @@ private:
bool bool
check_system_tables(THD *thd); check_system_tables(THD *thd);
int
load_events_from_db(THD *thd);
/* Singleton DP is used */ /* Singleton DP is used */
Events(); Events();
~Events(){} ~Events(){}
......
sql/sql_parse.cc
View file @ 04a99458
...@@ -4047,8 +4047,7 @@ end_with_restore_list: ...@@ -4047,8 +4047,7 @@ end_with_restore_list:
if (!(res= Events::get_instance()->drop_event(thd, if (!(res= Events::get_instance()->drop_event(thd,
lex->spname->m_db, lex->spname->m_db,
lex->spname->m_name, lex->spname->m_name,
lex->drop_if_exists, lex->drop_if_exists)))
FALSE)))
send_ok(thd); send_ok(thd);
} }
break; break;
......
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
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7