mirror of
https://github.com/MariaDB/server.git
synced 2025-01-16 12:02:42 +01:00
61430c77a4
into siva.hindu.god:/home/tsmith/m/bk/mrg-jan17/maint/51 BUILD/check-cpu: Auto merged configure.in: Auto merged include/my_global.h: Auto merged mysql-test/mysql-test-run.pl: Auto merged mysql-test/r/sp.result: Auto merged mysql-test/r/type_enum.result: Auto merged mysql-test/r/udf.result: Auto merged mysql-test/r/view.result: Auto merged mysql-test/t/mysqladmin.test: Auto merged mysql-test/t/sp.test: Auto merged mysql-test/t/type_enum.test: Auto merged mysql-test/t/view.test: Auto merged mysys/my_read.c: Auto merged scripts/make_binary_distribution.sh: Auto merged server-tools/instance-manager/Makefile.am: Auto merged sql/event_queue.cc: Auto merged sql/field.h: Auto merged sql/item_sum.h: Auto merged sql/log_event.cc: Auto merged sql/mysqld.cc: Auto merged sql/share/errmsg.txt: Auto merged sql/sql_class.h: Auto merged sql/sql_parse.cc: Auto merged sql/sql_show.cc: Auto merged sql/table.cc: Auto merged storage/federated/ha_federated.cc: Auto merged server-tools/instance-manager/instance.cc: Use remote (global 5.1 version) sql/unireg.cc: Use remote (5.1 global version) mysql-test/t/trigger.test: Manual merge server-tools/instance-manager/guardian.cc: Manual merge
949 lines
25 KiB
C++
949 lines
25 KiB
C++
/* Copyright (C) 2004-2006 MySQL 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 */
|
|
|
|
#include "mysql_priv.h"
|
|
#include "event_queue.h"
|
|
#include "event_data_objects.h"
|
|
#include "event_db_repository.h"
|
|
|
|
|
|
#define EVENT_QUEUE_INITIAL_SIZE 30
|
|
#define EVENT_QUEUE_EXTENT 30
|
|
|
|
#ifdef __GNUC__
|
|
#if __GNUC__ >= 2
|
|
#define SCHED_FUNC __FUNCTION__
|
|
#endif
|
|
#else
|
|
#define SCHED_FUNC "<unknown>"
|
|
#endif
|
|
|
|
#define LOCK_QUEUE_DATA() lock_data(SCHED_FUNC, __LINE__)
|
|
#define UNLOCK_QUEUE_DATA() unlock_data(SCHED_FUNC, __LINE__)
|
|
|
|
struct event_queue_param
|
|
{
|
|
THD *thd;
|
|
Event_queue *queue;
|
|
pthread_mutex_t LOCK_loaded;
|
|
pthread_cond_t COND_loaded;
|
|
bool loading_finished;
|
|
};
|
|
|
|
|
|
/*
|
|
Compares the execute_at members of two Event_queue_element instances.
|
|
Used as callback for the prioritized queue when shifting
|
|
elements inside.
|
|
|
|
SYNOPSIS
|
|
event_queue_element_data_compare_q()
|
|
vptr Not used (set it to NULL)
|
|
a First Event_queue_element object
|
|
b Second Event_queue_element object
|
|
|
|
RETURN VALUE
|
|
-1 a->execute_at < b->execute_at
|
|
0 a->execute_at == b->execute_at
|
|
1 a->execute_at > b->execute_at
|
|
|
|
NOTES
|
|
execute_at.second_part is not considered during comparison
|
|
*/
|
|
|
|
static int
|
|
event_queue_element_compare_q(void *vptr, byte* a, byte *b)
|
|
{
|
|
return my_time_compare(&((Event_queue_element *)a)->execute_at,
|
|
&((Event_queue_element *)b)->execute_at);
|
|
}
|
|
|
|
|
|
/*
|
|
Constructor of class Event_queue.
|
|
|
|
SYNOPSIS
|
|
Event_queue::Event_queue()
|
|
*/
|
|
|
|
Event_queue::Event_queue()
|
|
:mutex_last_unlocked_at_line(0), mutex_last_locked_at_line(0),
|
|
mutex_last_attempted_lock_at_line(0),
|
|
mutex_queue_data_locked(FALSE), mutex_queue_data_attempting_lock(FALSE)
|
|
{
|
|
mutex_last_unlocked_in_func= mutex_last_locked_in_func=
|
|
mutex_last_attempted_lock_in_func= "";
|
|
set_zero_time(&next_activation_at, MYSQL_TIMESTAMP_DATETIME);
|
|
}
|
|
|
|
|
|
/*
|
|
Inits mutexes.
|
|
|
|
SYNOPSIS
|
|
Event_queue::init_mutexes()
|
|
*/
|
|
|
|
void
|
|
Event_queue::init_mutexes()
|
|
{
|
|
pthread_mutex_init(&LOCK_event_queue, MY_MUTEX_INIT_FAST);
|
|
pthread_cond_init(&COND_queue_state, NULL);
|
|
}
|
|
|
|
|
|
/*
|
|
Destroys mutexes.
|
|
|
|
SYNOPSIS
|
|
Event_queue::deinit_mutexes()
|
|
*/
|
|
|
|
void
|
|
Event_queue::deinit_mutexes()
|
|
{
|
|
pthread_mutex_destroy(&LOCK_event_queue);
|
|
pthread_cond_destroy(&COND_queue_state);
|
|
}
|
|
|
|
|
|
/*
|
|
This is a queue's constructor. Until this method is called, the
|
|
queue is unusable. We don't use a C++ constructor instead in
|
|
order to be able to check the return value. The queue is
|
|
initialized once at server startup. Initialization can fail in
|
|
case of a failure reading events from the database or out of
|
|
memory.
|
|
|
|
SYNOPSIS
|
|
Event_queue::init()
|
|
|
|
RETURN VALUE
|
|
FALSE OK
|
|
TRUE Error
|
|
*/
|
|
|
|
bool
|
|
Event_queue::init_queue(THD *thd, Event_db_repository *db_repo)
|
|
{
|
|
bool res;
|
|
struct event_queue_param *event_queue_param_value= NULL;
|
|
|
|
DBUG_ENTER("Event_queue::init_queue");
|
|
DBUG_PRINT("enter", ("this: 0x%lx", (long) this));
|
|
|
|
LOCK_QUEUE_DATA();
|
|
db_repository= db_repo;
|
|
|
|
if (init_queue_ex(&queue, EVENT_QUEUE_INITIAL_SIZE , 0 /*offset*/,
|
|
0 /*max_on_top*/, event_queue_element_compare_q,
|
|
NULL, EVENT_QUEUE_EXTENT))
|
|
{
|
|
sql_print_error("SCHEDULER: Can't initialize the execution queue");
|
|
goto err;
|
|
}
|
|
|
|
if (sizeof(my_time_t) != sizeof(time_t))
|
|
{
|
|
sql_print_error("SCHEDULER: sizeof(my_time_t) != sizeof(time_t) ."
|
|
"The scheduler may not work correctly. Stopping");
|
|
goto err;
|
|
}
|
|
|
|
res= load_events_from_db(thd);
|
|
UNLOCK_QUEUE_DATA();
|
|
if (res)
|
|
deinit_queue();
|
|
|
|
DBUG_RETURN(res);
|
|
|
|
err:
|
|
UNLOCK_QUEUE_DATA();
|
|
DBUG_RETURN(TRUE);
|
|
}
|
|
|
|
|
|
/*
|
|
Deinits the queue. Remove all elements from it and destroys them
|
|
too.
|
|
|
|
SYNOPSIS
|
|
Event_queue::deinit_queue()
|
|
*/
|
|
|
|
void
|
|
Event_queue::deinit_queue()
|
|
{
|
|
DBUG_ENTER("Event_queue::deinit_queue");
|
|
|
|
LOCK_QUEUE_DATA();
|
|
empty_queue();
|
|
delete_queue(&queue);
|
|
UNLOCK_QUEUE_DATA();
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Adds an event to the queue.
|
|
|
|
SYNOPSIS
|
|
Event_queue::create_event()
|
|
dbname The schema of the new event
|
|
name The name of the new event
|
|
|
|
RETURN VALUE
|
|
OP_OK OK or scheduler not working
|
|
OP_LOAD_ERROR Error during loading from disk
|
|
*/
|
|
|
|
int
|
|
Event_queue::create_event(THD *thd, LEX_STRING dbname, LEX_STRING name)
|
|
{
|
|
int res;
|
|
Event_queue_element *new_element;
|
|
DBUG_ENTER("Event_queue::create_event");
|
|
DBUG_PRINT("enter", ("thd: 0x%lx et=%s.%s", (long) thd, dbname.str, name.str));
|
|
|
|
new_element= new Event_queue_element();
|
|
res= db_repository->load_named_event(thd, dbname, name, new_element);
|
|
if (res || new_element->status == Event_queue_element::DISABLED)
|
|
delete new_element;
|
|
else
|
|
{
|
|
new_element->compute_next_execution_time();
|
|
|
|
LOCK_QUEUE_DATA();
|
|
DBUG_PRINT("info", ("new event in the queue: 0x%lx", (long) new_element));
|
|
queue_insert_safe(&queue, (byte *) new_element);
|
|
dbug_dump_queue(thd->query_start());
|
|
pthread_cond_broadcast(&COND_queue_state);
|
|
UNLOCK_QUEUE_DATA();
|
|
}
|
|
|
|
DBUG_RETURN(res);
|
|
}
|
|
|
|
|
|
/*
|
|
Updates an event from the scheduler queue
|
|
|
|
SYNOPSIS
|
|
Event_queue::update_event()
|
|
thd Thread
|
|
dbname Schema of the event
|
|
name Name of the event
|
|
new_schema New schema, in case of RENAME TO, otherwise NULL
|
|
new_name New name, in case of RENAME TO, otherwise NULL
|
|
|
|
RETURN VALUE
|
|
OP_OK OK or scheduler not working
|
|
OP_LOAD_ERROR Error during loading from disk
|
|
*/
|
|
|
|
int
|
|
Event_queue::update_event(THD *thd, LEX_STRING dbname, LEX_STRING name,
|
|
LEX_STRING *new_schema, LEX_STRING *new_name)
|
|
{
|
|
int res;
|
|
Event_queue_element *new_element;
|
|
|
|
DBUG_ENTER("Event_queue::update_event");
|
|
DBUG_PRINT("enter", ("thd: 0x%lx et=[%s.%s]", (long) thd, dbname.str, name.str));
|
|
|
|
new_element= new Event_queue_element();
|
|
|
|
res= db_repository->load_named_event(thd, new_schema ? *new_schema:dbname,
|
|
new_name ? *new_name:name, new_element);
|
|
if (res)
|
|
{
|
|
delete new_element;
|
|
goto end;
|
|
}
|
|
else if (new_element->status == Event_queue_element::DISABLED)
|
|
{
|
|
DBUG_PRINT("info", ("The event is disabled."));
|
|
/*
|
|
Destroy the object but don't skip to end: because we may have to remove
|
|
object from the cache.
|
|
*/
|
|
delete new_element;
|
|
new_element= NULL;
|
|
}
|
|
else
|
|
new_element->compute_next_execution_time();
|
|
|
|
LOCK_QUEUE_DATA();
|
|
find_n_remove_event(dbname, name);
|
|
|
|
/* If not disabled event */
|
|
if (new_element)
|
|
{
|
|
DBUG_PRINT("info", ("new event in the queue: 0x%lx", (long) new_element));
|
|
queue_insert_safe(&queue, (byte *) new_element);
|
|
pthread_cond_broadcast(&COND_queue_state);
|
|
}
|
|
|
|
dbug_dump_queue(thd->query_start());
|
|
UNLOCK_QUEUE_DATA();
|
|
|
|
end:
|
|
DBUG_PRINT("info", ("res=%d", res));
|
|
DBUG_RETURN(res);
|
|
}
|
|
|
|
|
|
/*
|
|
Drops an event from the queue
|
|
|
|
SYNOPSIS
|
|
Event_queue::drop_event()
|
|
thd Thread
|
|
dbname Schema of the event to drop
|
|
name Name of the event to drop
|
|
*/
|
|
|
|
void
|
|
Event_queue::drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name)
|
|
{
|
|
DBUG_ENTER("Event_queue::drop_event");
|
|
DBUG_PRINT("enter", ("thd: 0x%lx db :%s name: %s", (long) thd,
|
|
dbname.str, name.str));
|
|
|
|
LOCK_QUEUE_DATA();
|
|
find_n_remove_event(dbname, name);
|
|
dbug_dump_queue(thd->query_start());
|
|
UNLOCK_QUEUE_DATA();
|
|
|
|
/*
|
|
We don't signal here because the scheduler will catch the change
|
|
next time it wakes up.
|
|
*/
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Drops all events from the in-memory queue and disk that match
|
|
certain pattern evaluated by a comparator function
|
|
|
|
SYNOPSIS
|
|
Event_queue::drop_matching_events()
|
|
thd THD
|
|
pattern A pattern string
|
|
comparator The function to use for comparing
|
|
|
|
RETURN VALUE
|
|
>=0 Number of dropped events
|
|
|
|
NOTE
|
|
Expected is the caller to acquire lock on LOCK_event_queue
|
|
*/
|
|
|
|
void
|
|
Event_queue::drop_matching_events(THD *thd, LEX_STRING pattern,
|
|
bool (*comparator)(LEX_STRING, Event_basic *))
|
|
{
|
|
uint i= 0;
|
|
DBUG_ENTER("Event_queue::drop_matching_events");
|
|
DBUG_PRINT("enter", ("pattern=%s", pattern.str));
|
|
|
|
while (i < queue.elements)
|
|
{
|
|
Event_queue_element *et= (Event_queue_element *) queue_element(&queue, i);
|
|
DBUG_PRINT("info", ("[%s.%s]?", et->dbname.str, et->name.str));
|
|
if (comparator(pattern, et))
|
|
{
|
|
/*
|
|
The queue is ordered. If we remove an element, then all elements
|
|
after it will shift one position to the left, if we imagine it as
|
|
an array from left to the right. In this case we should not
|
|
increment the counter and the (i < queue.elements) condition is ok.
|
|
*/
|
|
queue_remove(&queue, i);
|
|
delete et;
|
|
}
|
|
else
|
|
i++;
|
|
}
|
|
/*
|
|
We don't call pthread_cond_broadcast(&COND_queue_state);
|
|
If we remove the top event:
|
|
1. The queue is empty. The scheduler will wake up at some time and
|
|
realize that the queue is empty. If create_event() comes inbetween
|
|
it will signal the scheduler
|
|
2. The queue is not empty, but the next event after the previous top,
|
|
won't be executed any time sooner than the element we removed. Hence,
|
|
we may not notify the scheduler and it will realize the change when it
|
|
wakes up from timedwait.
|
|
*/
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Drops all events from the in-memory queue and disk that are from
|
|
certain schema.
|
|
|
|
SYNOPSIS
|
|
Event_queue::drop_schema_events()
|
|
thd HD
|
|
schema The schema name
|
|
*/
|
|
|
|
void
|
|
Event_queue::drop_schema_events(THD *thd, LEX_STRING schema)
|
|
{
|
|
DBUG_ENTER("Event_queue::drop_schema_events");
|
|
LOCK_QUEUE_DATA();
|
|
drop_matching_events(thd, schema, event_basic_db_equal);
|
|
UNLOCK_QUEUE_DATA();
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Searches for an event in the queue
|
|
|
|
SYNOPSIS
|
|
Event_queue::find_n_remove_event()
|
|
db The schema of the event to find
|
|
name The event to find
|
|
|
|
NOTE
|
|
The caller should do the locking also the caller is responsible for
|
|
actual signalling in case an event is removed from the queue.
|
|
*/
|
|
|
|
void
|
|
Event_queue::find_n_remove_event(LEX_STRING db, LEX_STRING name)
|
|
{
|
|
uint i;
|
|
DBUG_ENTER("Event_queue::find_n_remove_event");
|
|
|
|
for (i= 0; i < queue.elements; ++i)
|
|
{
|
|
Event_queue_element *et= (Event_queue_element *) queue_element(&queue, i);
|
|
DBUG_PRINT("info", ("[%s.%s]==[%s.%s]?", db.str, name.str,
|
|
et->dbname.str, et->name.str));
|
|
if (event_basic_identifier_equal(db, name, et))
|
|
{
|
|
queue_remove(&queue, i);
|
|
delete et;
|
|
break;
|
|
}
|
|
}
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
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
|
|
Event_queue::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
|
|
Event_queue::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("Event_queue::load_events_from_db");
|
|
DBUG_PRINT("enter", ("thd: 0x%lx", (long) 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;
|
|
}
|
|
|
|
queue_insert_safe(&queue, (byte *) et);
|
|
count++;
|
|
}
|
|
clean_the_queue= FALSE;
|
|
end:
|
|
end_read_record(&read_record_info);
|
|
|
|
if (clean_the_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);
|
|
}
|
|
|
|
|
|
/*
|
|
Recalculates activation times in the queue. There is one reason for
|
|
that. Because the values (execute_at) by which the queue is ordered are
|
|
changed by calls to compute_next_execution_time() on a request from the
|
|
scheduler thread, if it is not running then the values won't be updated.
|
|
Once the scheduler is started again the values has to be recalculated
|
|
so they are right for the current time.
|
|
|
|
SYNOPSIS
|
|
Event_queue::recalculate_activation_times()
|
|
thd Thread
|
|
*/
|
|
|
|
void
|
|
Event_queue::recalculate_activation_times(THD *thd)
|
|
{
|
|
uint i;
|
|
DBUG_ENTER("Event_queue::recalculate_activation_times");
|
|
|
|
LOCK_QUEUE_DATA();
|
|
DBUG_PRINT("info", ("%u loaded events to be recalculated", queue.elements));
|
|
for (i= 0; i < queue.elements; i++)
|
|
{
|
|
((Event_queue_element*)queue_element(&queue, i))->compute_next_execution_time();
|
|
((Event_queue_element*)queue_element(&queue, i))->update_timing_fields(thd);
|
|
}
|
|
queue_fix(&queue);
|
|
UNLOCK_QUEUE_DATA();
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Empties the queue and destroys the Event_queue_element objects in the
|
|
queue.
|
|
|
|
SYNOPSIS
|
|
Event_queue::empty_queue()
|
|
|
|
NOTE
|
|
Should be called with LOCK_event_queue locked
|
|
*/
|
|
|
|
void
|
|
Event_queue::empty_queue()
|
|
{
|
|
uint i;
|
|
DBUG_ENTER("Event_queue::empty_queue");
|
|
DBUG_PRINT("enter", ("Purging the queue. %d element(s)", queue.elements));
|
|
sql_print_information("SCHEDULER: Purging queue. %u events", queue.elements);
|
|
/* empty the queue */
|
|
for (i= 0; i < queue.elements; ++i)
|
|
{
|
|
Event_queue_element *et= (Event_queue_element *) queue_element(&queue, i);
|
|
delete et;
|
|
}
|
|
resize_queue(&queue, 0);
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Dumps the queue to the trace log.
|
|
|
|
SYNOPSIS
|
|
Event_queue::dbug_dump_queue()
|
|
now Current timestamp
|
|
*/
|
|
|
|
void
|
|
Event_queue::dbug_dump_queue(time_t now)
|
|
{
|
|
#ifndef DBUG_OFF
|
|
Event_queue_element *et;
|
|
uint i;
|
|
DBUG_ENTER("Event_queue::dbug_dump_queue");
|
|
DBUG_PRINT("info", ("Dumping queue . Elements=%u", queue.elements));
|
|
for (i = 0; i < queue.elements; i++)
|
|
{
|
|
et= ((Event_queue_element*)queue_element(&queue, i));
|
|
DBUG_PRINT("info", ("et: 0x%lx name: %s.%s", (long) et,
|
|
et->dbname.str, et->name.str));
|
|
DBUG_PRINT("info", ("exec_at: %lu starts: %lu ends: %lu execs_so_far: %u "
|
|
"expr: %ld et.exec_at: %ld now: %ld "
|
|
"(et.exec_at - now): %d if: %d",
|
|
(long) TIME_to_ulonglong_datetime(&et->execute_at),
|
|
(long) TIME_to_ulonglong_datetime(&et->starts),
|
|
(long) TIME_to_ulonglong_datetime(&et->ends),
|
|
et->execution_count,
|
|
(long) et->expression,
|
|
(long) (sec_since_epoch_TIME(&et->execute_at)),
|
|
(long) now,
|
|
(int) (sec_since_epoch_TIME(&et->execute_at) - now),
|
|
sec_since_epoch_TIME(&et->execute_at) <= now));
|
|
}
|
|
DBUG_VOID_RETURN;
|
|
#endif
|
|
}
|
|
|
|
static const char *queue_empty_msg= "Waiting on empty queue";
|
|
static const char *queue_wait_msg= "Waiting for next activation";
|
|
|
|
/*
|
|
Checks whether the top of the queue is elligible for execution and
|
|
returns an Event_job_data instance in case it should be executed.
|
|
`now` is compared against `execute_at` of the top element in the queue.
|
|
|
|
SYNOPSIS
|
|
Event_queue::get_top_for_execution_if_time()
|
|
thd [in] Thread
|
|
job_data [out] The object to execute
|
|
|
|
RETURN VALUE
|
|
FALSE No error. If *job_data==NULL then top not elligible for execution.
|
|
Could be that there is no top.
|
|
TRUE Error
|
|
|
|
*/
|
|
|
|
bool
|
|
Event_queue::get_top_for_execution_if_time(THD *thd, Event_job_data **job_data)
|
|
{
|
|
bool ret= FALSE;
|
|
struct timespec top_time;
|
|
Event_queue_element *top= NULL;
|
|
bool to_free= FALSE;
|
|
bool to_drop= FALSE;
|
|
*job_data= NULL;
|
|
DBUG_ENTER("Event_queue::get_top_for_execution_if_time");
|
|
|
|
LOCK_QUEUE_DATA();
|
|
for (;;)
|
|
{
|
|
int res;
|
|
|
|
/* Break loop if thd has been killed */
|
|
if (thd->killed)
|
|
{
|
|
DBUG_PRINT("info", ("thd->killed=%d", thd->killed));
|
|
goto end;
|
|
}
|
|
|
|
if (!queue.elements)
|
|
{
|
|
/* There are no events in the queue */
|
|
set_zero_time(&next_activation_at, MYSQL_TIMESTAMP_DATETIME);
|
|
|
|
/* Wait on condition until signaled. Release LOCK_queue while waiting. */
|
|
cond_wait(thd, NULL, queue_empty_msg, SCHED_FUNC, __LINE__);
|
|
|
|
continue;
|
|
}
|
|
|
|
top= ((Event_queue_element*) queue_element(&queue, 0));
|
|
|
|
thd->end_time(); /* Get current time */
|
|
|
|
time_t seconds_to_next_event=
|
|
sec_since_epoch_TIME(&top->execute_at) - thd->query_start();
|
|
next_activation_at= top->execute_at;
|
|
if (seconds_to_next_event > 0)
|
|
{
|
|
/*
|
|
Not yet time for top event, wait on condition with
|
|
time or until signaled. Release LOCK_queue while waiting.
|
|
*/
|
|
set_timespec(top_time, seconds_to_next_event);
|
|
cond_wait(thd, &top_time, queue_wait_msg, SCHED_FUNC, __LINE__);
|
|
|
|
continue;
|
|
}
|
|
|
|
DBUG_PRINT("info", ("Ready for execution"));
|
|
if (!(*job_data= new Event_job_data()))
|
|
{
|
|
ret= TRUE;
|
|
break;
|
|
}
|
|
if ((res= db_repository->load_named_event(thd, top->dbname, top->name,
|
|
*job_data)))
|
|
{
|
|
DBUG_PRINT("error", ("Got %d from load_named_event", res));
|
|
delete *job_data;
|
|
*job_data= NULL;
|
|
ret= TRUE;
|
|
break;
|
|
}
|
|
|
|
top->mark_last_executed(thd);
|
|
if (top->compute_next_execution_time())
|
|
top->status= Event_queue_element::DISABLED;
|
|
DBUG_PRINT("info", ("event %s status is %d", top->name.str, top->status));
|
|
|
|
(*job_data)->execution_count= top->execution_count;
|
|
|
|
top->update_timing_fields(thd);
|
|
if (((top->execute_at.year && !top->expression) || top->execute_at_null) ||
|
|
(top->status == Event_queue_element::DISABLED))
|
|
{
|
|
DBUG_PRINT("info", ("removing from the queue"));
|
|
sql_print_information("SCHEDULER: Last execution of %s.%s. %s",
|
|
top->dbname.str, top->name.str,
|
|
top->dropped? "Dropping.":"");
|
|
to_free= TRUE;
|
|
to_drop= top->dropped;
|
|
queue_remove(&queue, 0);
|
|
}
|
|
else
|
|
queue_replaced(&queue);
|
|
|
|
dbug_dump_queue(thd->query_start());
|
|
break;
|
|
}
|
|
end:
|
|
UNLOCK_QUEUE_DATA();
|
|
if (to_drop)
|
|
{
|
|
DBUG_PRINT("info", ("Dropping from disk"));
|
|
top->drop(thd);
|
|
}
|
|
if (to_free)
|
|
delete top;
|
|
|
|
DBUG_PRINT("info", ("returning %d et_new: 0x%lx ", ret, (long) *job_data));
|
|
|
|
if (*job_data)
|
|
DBUG_PRINT("info", ("db: %s name: %s definer=%s", (*job_data)->dbname.str,
|
|
(*job_data)->name.str, (*job_data)->definer.str));
|
|
|
|
DBUG_RETURN(ret);
|
|
}
|
|
|
|
|
|
/*
|
|
Auxiliary function for locking LOCK_event_queue. Used by the
|
|
LOCK_QUEUE_DATA macro
|
|
|
|
SYNOPSIS
|
|
Event_queue::lock_data()
|
|
func Which function is requesting mutex lock
|
|
line On which line mutex lock is requested
|
|
*/
|
|
|
|
void
|
|
Event_queue::lock_data(const char *func, uint line)
|
|
{
|
|
DBUG_ENTER("Event_queue::lock_data");
|
|
DBUG_PRINT("enter", ("func=%s line=%u", func, line));
|
|
mutex_last_attempted_lock_in_func= func;
|
|
mutex_last_attempted_lock_at_line= line;
|
|
mutex_queue_data_attempting_lock= TRUE;
|
|
pthread_mutex_lock(&LOCK_event_queue);
|
|
mutex_last_attempted_lock_in_func= "";
|
|
mutex_last_attempted_lock_at_line= 0;
|
|
mutex_queue_data_attempting_lock= FALSE;
|
|
|
|
mutex_last_locked_in_func= func;
|
|
mutex_last_locked_at_line= line;
|
|
mutex_queue_data_locked= TRUE;
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Auxiliary function for unlocking LOCK_event_queue. Used by the
|
|
UNLOCK_QUEUE_DATA macro
|
|
|
|
SYNOPSIS
|
|
Event_queue::unlock_data()
|
|
func Which function is requesting mutex unlock
|
|
line On which line mutex unlock is requested
|
|
*/
|
|
|
|
void
|
|
Event_queue::unlock_data(const char *func, uint line)
|
|
{
|
|
DBUG_ENTER("Event_queue::unlock_data");
|
|
DBUG_PRINT("enter", ("func=%s line=%u", func, line));
|
|
mutex_last_unlocked_at_line= line;
|
|
mutex_queue_data_locked= FALSE;
|
|
mutex_last_unlocked_in_func= func;
|
|
pthread_mutex_unlock(&LOCK_event_queue);
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Wrapper for pthread_cond_wait/timedwait
|
|
|
|
SYNOPSIS
|
|
Event_queue::cond_wait()
|
|
thd Thread (Could be NULL during shutdown procedure)
|
|
msg Message for thd->proc_info
|
|
abstime If not null then call pthread_cond_timedwait()
|
|
func Which function is requesting cond_wait
|
|
line On which line cond_wait is requested
|
|
*/
|
|
|
|
void
|
|
Event_queue::cond_wait(THD *thd, struct timespec *abstime, const char* msg,
|
|
const char *func, uint line)
|
|
{
|
|
DBUG_ENTER("Event_queue::cond_wait");
|
|
waiting_on_cond= TRUE;
|
|
mutex_last_unlocked_at_line= line;
|
|
mutex_queue_data_locked= FALSE;
|
|
mutex_last_unlocked_in_func= func;
|
|
|
|
thd->enter_cond(&COND_queue_state, &LOCK_event_queue, msg);
|
|
|
|
DBUG_PRINT("info", ("pthread_cond_%swait", abstime? "timed":""));
|
|
if (!abstime)
|
|
pthread_cond_wait(&COND_queue_state, &LOCK_event_queue);
|
|
else
|
|
pthread_cond_timedwait(&COND_queue_state, &LOCK_event_queue, abstime);
|
|
|
|
mutex_last_locked_in_func= func;
|
|
mutex_last_locked_at_line= line;
|
|
mutex_queue_data_locked= TRUE;
|
|
waiting_on_cond= FALSE;
|
|
|
|
/*
|
|
This will free the lock so we need to relock. Not the best thing to
|
|
do but we need to obey cond_wait()
|
|
*/
|
|
thd->exit_cond("");
|
|
lock_data(func, line);
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|
|
|
|
|
|
/*
|
|
Dumps the internal status of the queue
|
|
|
|
SYNOPSIS
|
|
Event_queue::dump_internal_status()
|
|
*/
|
|
|
|
void
|
|
Event_queue::dump_internal_status()
|
|
{
|
|
DBUG_ENTER("Event_queue::dump_internal_status");
|
|
|
|
/* element count */
|
|
puts("");
|
|
puts("Event queue status:");
|
|
printf("Element count : %u\n", queue.elements);
|
|
printf("Data locked : %s\n", mutex_queue_data_locked? "YES":"NO");
|
|
printf("Attempting lock : %s\n", mutex_queue_data_attempting_lock? "YES":"NO");
|
|
printf("LLA : %s:%u\n", mutex_last_locked_in_func,
|
|
mutex_last_locked_at_line);
|
|
printf("LUA : %s:%u\n", mutex_last_unlocked_in_func,
|
|
mutex_last_unlocked_at_line);
|
|
if (mutex_last_attempted_lock_at_line)
|
|
printf("Last lock attempt at: %s:%u\n", mutex_last_attempted_lock_in_func,
|
|
mutex_last_attempted_lock_at_line);
|
|
printf("WOC : %s\n", waiting_on_cond? "YES":"NO");
|
|
printf("Next activation : %04d-%02d-%02d %02d:%02d:%02d\n",
|
|
next_activation_at.year, next_activation_at.month,
|
|
next_activation_at.day, next_activation_at.hour,
|
|
next_activation_at.minute, next_activation_at.second);
|
|
|
|
DBUG_VOID_RETURN;
|
|
}
|