WL #3337 (Event scheduler new architecture)

Cut Nr. 8.

All tests pass.

Separated Event_scheduler into Event_queue and Event_scheduler.
Added new Event_scheduler_ng which is the new scheduler and is used
system-wide. Will be moved to the event_scheduler.cc in the future.
Using Event_timed in Event_queue as well as cloned during execution.
Next step is to have Event_worker_data which will be used during execution
and will take ::compile()/::execute() out of Event_timed.

mysql-test/r/events.result

@@ -323,7 +323,6 @@ root@localhost	закачка	events_test
 "Should be only 1 process"
 select /*1*/ user, host, db, command, state, info from information_schema.processlist where info is null or info not like '%processlist%' order by info;
 user	host	db	command	state	info
-event_scheduler	localhost	NULL	Connect	Suspended	NULL
 select release_lock("test_lock1");
 release_lock("test_lock1")
 1
@@ -343,7 +342,7 @@ create event закачка on schedule every 10 hour do select get_lock("test_l
 "Should have only 2 processes: the scheduler and the locked event"
 select /*2*/ user, host, db, command, state, info from information_schema.processlist where info is null or info not like '%processlist%' order by info;
 user	host	db	command	state	info
-event_scheduler	localhost	NULL	Connect	Sleeping	NULL
+event_scheduler	localhost	NULL	Connect	Waiting for next activation	NULL
 root	localhost	events_test	Connect	User lock	select get_lock("test_lock2", 20)
 "Release the mutex, the event worker should finish."
 "Release the mutex, the event worker should finish."
@@ -359,13 +358,12 @@ create event закачка21 on schedule every 10 hour do select get_lock("test
 "Should have only 3 processes: the scheduler, our conn and the locked event"
 select /*3*/ user, host, db, command, state, info from information_schema.processlist where info is null or info not like '%processlist%' order by info;
 user	host	db	command	state	info
-event_scheduler	localhost	NULL	Connect	Sleeping	NULL
+event_scheduler	localhost	NULL	Connect	Waiting for next activation	NULL
 root	localhost	events_test	Connect	User lock	select get_lock("test_lock2_1", 20)
 set global event_scheduler=2;
 "Should have only our process now:"
 select /*4*/ user, host, db, command, state, info from information_schema.processlist where info is null or info not like '%processlist%' order by info;
 user	host	db	command	state	info
-event_scheduler	localhost	NULL	Connect	Suspended	NULL
 root	localhost	events_test	Connect	User lock	select get_lock("test_lock2_1", 20)
 drop event закачка21;
 create table t_16 (s1 int);

mysql-test/r/events_bugs.result

@@ -41,7 +41,7 @@ end|
 "Now if everything is fine the event has compiled and is locked
 select /*1*/ user, host, db, command, state, info from information_schema.processlist where info is null or info not like '%processlist%' order by info;
 user	host	db	command	state	info
-event_scheduler	localhost	NULL	Connect	Sleeping	NULL
+event_scheduler	localhost	NULL	Connect	Waiting for next activation	NULL
 root	localhost	events_test	Connect	User lock	select get_lock('test_bug16407', 60)
 select release_lock('test_bug16407');
 release_lock('test_bug16407')
@@ -94,7 +94,7 @@ get_lock('ee_16407_2', 60)
 set global event_scheduler= 1;
 select /*2*/ user, host, db, command, state, info from information_schema.processlist where info is null or info not like '%processlist%' order by info;
 user	host	db	command	state	info
-event_scheduler	localhost	NULL	Connect	Sleeping	NULL
+event_scheduler	localhost	NULL	Connect	Waiting for next activation	NULL
 root	localhost	events_test	Connect	User lock	select get_lock('ee_16407_2', 60) /*ee_16407_2*/
 root	localhost	events_test	Connect	User lock	select get_lock('ee_16407_2', 60) /*ee_16407_3*/
 root	localhost	events_test	Connect	User lock	select get_lock('ee_16407_2', 60) /*ee_16407_4*/
@@ -103,7 +103,7 @@ release_lock('ee_16407_2')
 1
 select /*3*/ user, host, db, command, state, info from information_schema.processlist where info is null or info not like '%processlist%' order by info;
 user	host	db	command	state	info
-event_scheduler	localhost	NULL	Connect	Sleeping	NULL
+event_scheduler	localhost	NULL	Connect	Waiting for next activation	NULL
 set global event_scheduler= 2;
 select * from events_smode_test order by ev_name, a;
 ev_name	a
@@ -142,7 +142,7 @@ set global event_scheduler= 1;
 "Should have 2 locked processes"
 select /*4*/ user, host, db, command, state, info from information_schema.processlist where info is null or info not like '%processlist%' order by info;
 user	host	db	command	state	info
-event_scheduler	localhost	NULL	Connect	Sleeping	NULL
+event_scheduler	localhost	NULL	Connect	Waiting for next activation	NULL
 root	localhost	events_test	Connect	User lock	select get_lock('ee_16407_5', 60) /*ee_16407_5*/
 root	localhost	events_test	Connect	User lock	select get_lock('ee_16407_5', 60) /*ee_16407_6*/
 select release_lock('ee_16407_5');
@@ -151,7 +151,7 @@ release_lock('ee_16407_5')
 "Should have 0 processes locked"
 select /*5*/ user, host, db, command, state, info from information_schema.processlist where info is null or info not like '%processlist%' order by info;
 user	host	db	command	state	info
-event_scheduler	localhost	NULL	Connect	Sleeping	NULL
+event_scheduler	localhost	NULL	Connect	Waiting for next activation	NULL
 select * from events_smode_test order by ev_name, a;
 ev_name	a
 ee_16407_6	2004-02-29

mysql-test/r/ps_1general.result

@@ -299,7 +299,7 @@ t9	MyISAM	10	Dynamic	2	216	432	#	2048	0	NULL	#	#	#	latin1_swedish_ci	NULL
 prepare stmt4 from ' show status like ''Threads_running'' ';
 execute stmt4;
 Variable_name	Value
-Threads_running	2
+Threads_running	1
 prepare stmt4 from ' show variables like ''sql_mode'' ';
 execute stmt4;
 Variable_name	Value

mysql-test/r/skip_name_resolve.result

@@ -10,6 +10,5 @@ user()
 #
 show processlist;
 Id	User	Host	db	Command	Time	State	Info
-<id>	event_scheduler	<host>	NULL	<command>	<time>	<state>	<info>
 <id>	root	<host>	test	<command>	<time>	<state>	<info>
 <id>	root	<host>	test	<command>	<time>	<state>	<info>

mysql-test/r/sp-threads.result

@@ -34,7 +34,6 @@ lock tables t2 write;
  call bug9486();
 show processlist;
 Id	User	Host	db	Command	Time	State	Info
-#	event_scheduler	localhost	NULL	Connect	#	Suspended	NULL
 #	root	localhost	test	Sleep	#		NULL
 #	root	localhost	test	Query	#	Locked	update t1, t2 set val= 1 where id1=id2
 #	root	localhost	test	Query	#	NULL	show processlist

mysql-test/r/sp_notembedded.result

@@ -18,11 +18,9 @@ show processlist;
 end|
 call bug4902_2()|
 Id	User	Host	db	Command	Time	State	Info
-#	event_scheduler	localhost	NULL	Connect	#	Suspended	NULL
 #	root	localhost	test	Query	#	NULL	show processlist
 call bug4902_2()|
 Id	User	Host	db	Command	Time	State	Info
-#	event_scheduler	localhost	NULL	Connect	#	Suspended	NULL
 #	root	localhost	test	Query	#	NULL	show processlist
 drop procedure bug4902_2|
 drop function if exists bug5278|

mysql-test/r/status.result

@@ -52,22 +52,22 @@ drop table t1;
 FLUSH STATUS;
 SHOW STATUS LIKE 'max_used_connections';
 Variable_name	Value
-Max_used_connections	2
+Max_used_connections	1
 SET @save_thread_cache_size=@@thread_cache_size;
 SET GLOBAL thread_cache_size=3;
 SHOW STATUS LIKE 'max_used_connections';
 Variable_name	Value
-Max_used_connections	4
+Max_used_connections	3
 FLUSH STATUS;
 SHOW STATUS LIKE 'max_used_connections';
 Variable_name	Value
-Max_used_connections	3
+Max_used_connections	2
 SHOW STATUS LIKE 'max_used_connections';
 Variable_name	Value
-Max_used_connections	4
+Max_used_connections	3
 SHOW STATUS LIKE 'max_used_connections';
 Variable_name	Value
-Max_used_connections	5
+Max_used_connections	4
 SET GLOBAL thread_cache_size=@save_thread_cache_size;
 show status like 'com_show_status';
 Variable_name	Value

mysql-test/t/events_stress.test

@@ -61,7 +61,7 @@ while ($1)
 --enable_query_log
 SELECT COUNT(*) FROM INFORMATION_SCHEMA.EVENTS WHERE EVENT_SCHEMA='events_conn1_test2';
 SET GLOBAL event_scheduler=1;
---sleep 6
+--sleep 12
 DROP DATABASE events_conn1_test2;
 
 SET GLOBAL event_scheduler=2;
@@ -100,7 +100,7 @@ while ($1)
 }
 --enable_query_log
 SELECT COUNT(*) FROM INFORMATION_SCHEMA.EVENTS WHERE EVENT_SCHEMA='events_conn1_test2';
---sleep 6
+--sleep 12
 connection conn2;
 --send
 DROP DATABASE events_conn2_db;

sql/Makefile.am

@@ -67,7 +67,7 @@ noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			sql_array.h sql_cursor.h events.h \
 			sql_plugin.h authors.h sql_partition.h event_data_objects.h \
                         event_queue.h event_db_repository.h \
-			partition_info.h partition_element.h event_scheduler.h \
+			partition_info.h partition_element.h event_scheduler_ng.h \
 			contributors.h
 mysqld_SOURCES =	sql_lex.cc sql_handler.cc sql_partition.cc \
 			item.cc item_sum.cc item_buff.cc item_func.cc \
@@ -104,8 +104,8 @@ mysqld_SOURCES =	sql_lex.cc sql_handler.cc sql_partition.cc \
 			gstream.cc spatial.cc sql_help.cc sql_cursor.cc \
 			tztime.cc my_time.c my_user.c my_decimal.cc\
 			sp_head.cc sp_pcontext.cc  sp_rcontext.cc sp.cc \
-			sp_cache.cc parse_file.cc sql_trigger.cc \
-                        event_scheduler.cc events.cc event_data_objects.cc \
+			sp_cache.cc parse_file.cc sql_trigger.cc event_scheduler.cc\
+                        event_scheduler_ng.cc events.cc event_data_objects.cc \
                         event_queue.cc event_db_repository.cc \
 			sql_plugin.cc sql_binlog.cc \
 			sql_builtin.cc sql_tablespace.cc partition_info.cc

sql/event_data_objects.cc

@@ -556,6 +556,7 @@ Event_timed::Event_timed():in_spawned_thread(0),locked_by_thread_id(0),
 Event_timed::~Event_timed()
 {    
   deinit_mutexes();
+  free_root(&mem_root, MYF(0));
 
   if (free_sphead_on_delete)
     free_sp();
@@ -622,6 +623,8 @@ Event_timed::init()
   definer_user.length= definer_host.length= 0;
 
   sql_mode= 0;
+  /* init memory root */
+  init_alloc_root(&mem_root, 256, 512);
 
   DBUG_VOID_RETURN;
 }
@@ -644,7 +647,7 @@ Event_timed::init()
 */
 
 int
-Event_timed::load_from_row(MEM_ROOT *mem_root, TABLE *table)
+Event_timed::load_from_row(TABLE *table)
 {
   char *ptr;
   Event_timed *et;
@@ -661,22 +664,22 @@ Event_timed::load_from_row(MEM_ROOT *mem_root, TABLE *table)
   if (table->s->fields != ET_FIELD_COUNT)
     goto error;
 
-  if ((et->dbname.str= get_field(mem_root, table->field[ET_FIELD_DB])) == NULL)
+  if ((et->dbname.str= get_field(&mem_root, table->field[ET_FIELD_DB])) == NULL)
     goto error;
 
   et->dbname.length= strlen(et->dbname.str);
 
-  if ((et->name.str= get_field(mem_root, table->field[ET_FIELD_NAME])) == NULL)
+  if ((et->name.str= get_field(&mem_root, table->field[ET_FIELD_NAME])) == NULL)
     goto error;
 
   et->name.length= strlen(et->name.str);
 
-  if ((et->body.str= get_field(mem_root, table->field[ET_FIELD_BODY])) == NULL)
+  if ((et->body.str= get_field(&mem_root, table->field[ET_FIELD_BODY])) == NULL)
     goto error;
 
   et->body.length= strlen(et->body.str);
 
-  if ((et->definer.str= get_field(mem_root,
+  if ((et->definer.str= get_field(&mem_root,
                                   table->field[ET_FIELD_DEFINER])) == NullS)
     goto error;
   et->definer.length= strlen(et->definer.str);
@@ -688,10 +691,10 @@ Event_timed::load_from_row(MEM_ROOT *mem_root, TABLE *table)
 
   len= ptr - et->definer.str;
 
-  et->definer_user.str= strmake_root(mem_root, et->definer.str, len);
+  et->definer_user.str= strmake_root(&mem_root, et->definer.str, len);
   et->definer_user.length= len;
   len= et->definer.length - len - 1;            //1 is because of @
-  et->definer_host.str= strmake_root(mem_root, ptr + 1, len);/* 1:because of @*/
+  et->definer_host.str= strmake_root(&mem_root, ptr + 1, len);/* 1:because of @*/
   et->definer_host.length= len;
   
   et->starts_null= table->field[ET_FIELD_STARTS]->is_null();
@@ -737,21 +740,21 @@ Event_timed::load_from_row(MEM_ROOT *mem_root, TABLE *table)
   last_executed_changed= false;
 
   /* ToDo : Andrey . Find a way not to allocate ptr on event_mem_root */
-  if ((ptr= get_field(mem_root, table->field[ET_FIELD_STATUS])) == NullS)
+  if ((ptr= get_field(&mem_root, table->field[ET_FIELD_STATUS])) == NullS)
     goto error;
 
   DBUG_PRINT("load_from_row", ("Event [%s] is [%s]", et->name.str, ptr));
   et->status= (ptr[0]=='E'? Event_timed::ENABLED:Event_timed::DISABLED);
 
   /* ToDo : Andrey . Find a way not to allocate ptr on event_mem_root */
-  if ((ptr= get_field(mem_root,
+  if ((ptr= get_field(&mem_root,
                   table->field[ET_FIELD_ON_COMPLETION])) == NullS)
     goto error;
 
   et->on_completion= (ptr[0]=='D'? Event_timed::ON_COMPLETION_DROP:
                                    Event_timed::ON_COMPLETION_PRESERVE);
 
-  et->comment.str= get_field(mem_root, table->field[ET_FIELD_COMMENT]);
+  et->comment.str= get_field(&mem_root, table->field[ET_FIELD_COMMENT]);
   if (et->comment.str != NullS)
     et->comment.length= strlen(et->comment.str);
   else
@@ -953,10 +956,10 @@ Event_timed::compute_next_execution_time()
   int tmp;
 
   DBUG_ENTER("Event_timed::compute_next_execution_time");
-  DBUG_PRINT("enter", ("starts=%llu ends=%llu last_executed=%llu",
+  DBUG_PRINT("enter", ("starts=%llu ends=%llu last_executed=%llu this=%p",
                         TIME_to_ulonglong_datetime(&starts),
                         TIME_to_ulonglong_datetime(&ends),
-                        TIME_to_ulonglong_datetime(&last_executed)));
+                        TIME_to_ulonglong_datetime(&last_executed), this));
 
   if (status == Event_timed::DISABLED)
   {
@@ -1168,7 +1171,8 @@ Event_timed::compute_next_execution_time()
     goto ret;
   }
 ret:
-  DBUG_PRINT("info", ("ret=0"));
+  DBUG_PRINT("info", ("ret=0 execute_at=%llu",
+             TIME_to_ulonglong_datetime(&execute_at)));
   DBUG_RETURN(false);
 err:
   DBUG_PRINT("info", ("ret=1"));
@@ -1392,6 +1396,7 @@ Event_timed::get_create_event(THD *thd, String *buf)
 int
 Event_timed::execute(THD *thd, MEM_ROOT *mem_root)
 {
+  Security_context *save_ctx;
   /* this one is local and not needed after exec */
   Security_context security_ctx;
   int ret= 0;
@@ -1400,14 +1405,8 @@ Event_timed::execute(THD *thd, MEM_ROOT *mem_root)
   DBUG_PRINT("info", ("    EVEX EXECUTING event %s.%s [EXPR:%d]",
                dbname.str, name.str, (int) expression));
 
-  VOID(pthread_mutex_lock(&this->LOCK_running));
-  if (running)
-  {
-    VOID(pthread_mutex_unlock(&this->LOCK_running));
-    DBUG_RETURN(-100);
-  }
-  running= true;
-  VOID(pthread_mutex_unlock(&this->LOCK_running));
+  thd->change_security_context(definer_user, definer_host, dbname,
+                               &security_ctx, &save_ctx);
 
   if (!sphead && (ret= compile(thd, mem_root)))
     goto done;
@@ -1434,14 +1433,11 @@ Event_timed::execute(THD *thd, MEM_ROOT *mem_root)
                definer_host.str, dbname.str));
     ret= -99;
   }
-
-  VOID(pthread_mutex_lock(&this->LOCK_running));
-  running= false;
   /* Will compile every time a new sp_head on different root */
   free_sp();
-  VOID(pthread_mutex_unlock(&this->LOCK_running));
 
 done:
+  thd->restore_security_context(save_ctx);
   /*
     1. Don't cache sphead if allocated on another mem_root
     2. Don't call security_ctx.destroy() because this will free our dbname.str
@@ -1807,3 +1803,4 @@ event_timed_identifier_equal(LEX_STRING db, LEX_STRING name, Event_timed *b)
   return !sortcmp_lex_string(name, b->name, system_charset_info) &&
          !sortcmp_lex_string(db, b->dbname, system_charset_info);
 }
+

sql/event_data_objects.h

@@ -73,7 +73,10 @@ class Event_timed
   bool status_changed;
   bool last_executed_changed;
   
+  MEM_ROOT mem_root;
+
 public:
+  THD *thd;
   enum enum_status
   {
     ENABLED = 1,
@@ -147,7 +150,7 @@ class Event_timed
   deinit_mutexes();
 
   int
-  load_from_row(MEM_ROOT *mem_root, TABLE *table);
+  load_from_row(TABLE *table);
 
   bool
   compute_next_execution_time();
@@ -264,9 +267,33 @@ class Event_parse_data : public Sql_alloc
 };
 
 
-class Event_queue_element : public Event_timed
+class Event_job_data
 {
+public:
+  LEX_STRING dbname;
+  LEX_STRING name;
+  sp_head *sphead;
+  LEX_STRING definer;
+  LEX_STRING body;
+  ulong sql_mode;
 
-};
+  THD *thd;
+  
+  Event_job_data(){}
+  ~Event_job_data(){}
 
+  int
+  execute();
+
+private:
+  int
+  load_from_disk();
+  
+  int
+  compile();
+
+
+  Event_job_data(const Event_job_data &);	/* Prevent use of these */
+  void operator=(Event_job_data &);
+};
 #endif /* _EVENT_DATA_OBJECTS_H_ */

sql/event_db_repository.cc

@@ -130,135 +130,9 @@ TABLE_FIELD_W_TYPE event_table_fields[ET_FIELD_COUNT] = {
   SYNOPSIS
     evex_fill_row()
       thd        THD
-      table  the row to fill out
-      et     Event's data
-
-  RETURN VALUE
-    0 - OK
-    EVEX_GENERAL_ERROR    - bad data
-    EVEX_GET_FIELD_FAILED - field count does not match. table corrupted?
-
-  DESCRIPTION 
-    Used both when an event is created and when it is altered.
-*/
-
-static int
-evex_fill_row(THD *thd, TABLE *table, Event_timed *et, my_bool is_update)
-{
-  CHARSET_INFO *scs= system_charset_info;
-  enum enum_events_table_field field_num;
-
-  DBUG_ENTER("evex_fill_row");
-
-  DBUG_PRINT("info", ("dbname=[%s]", et->dbname.str));
-  DBUG_PRINT("info", ("name  =[%s]", et->name.str));
-  DBUG_PRINT("info", ("body  =[%s]", et->body.str));
-
-  if (table->field[field_num= ET_FIELD_DEFINER]->
-                  store(et->definer.str, et->definer.length, scs))
-    goto err_truncate;
-
-  if (table->field[field_num= ET_FIELD_DB]->
-                  store(et->dbname.str, et->dbname.length, scs))
-    goto err_truncate;
-
-  if (table->field[field_num= ET_FIELD_NAME]->
-                  store(et->name.str, et->name.length, scs))
-    goto err_truncate;
-
-  /* both ON_COMPLETION and STATUS are NOT NULL thus not calling set_notnull()*/
-  table->field[ET_FIELD_ON_COMPLETION]->
-                                       store((longlong)et->on_completion, true);
-
-  table->field[ET_FIELD_STATUS]->store((longlong)et->status, true);
-
-  /*
-    Change the SQL_MODE only if body was present in an ALTER EVENT and of course
-    always during CREATE EVENT.
-  */ 
-  if (et->body.str)
-  {
-    table->field[ET_FIELD_SQL_MODE]->
-                               store((longlong)thd->variables.sql_mode, true);
-
-    if (table->field[field_num= ET_FIELD_BODY]->
-                     store(et->body.str, et->body.length, scs))
-      goto err_truncate;
-  }
-
-  if (et->expression)
-  {
-    table->field[ET_FIELD_INTERVAL_EXPR]->set_notnull();
-    table->field[ET_FIELD_INTERVAL_EXPR]->store((longlong)et->expression, true);
-
-    table->field[ET_FIELD_TRANSIENT_INTERVAL]->set_notnull();
-    /*
-      In the enum (C) intervals start from 0 but in mysql enum valid values start
-      from 1. Thus +1 offset is needed!
-    */
-    table->field[ET_FIELD_TRANSIENT_INTERVAL]->
-                                         store((longlong)et->interval+1, true);
-
-    table->field[ET_FIELD_EXECUTE_AT]->set_null();
-
-    if (!et->starts_null)
-    {
-      table->field[ET_FIELD_STARTS]->set_notnull();
-      table->field[ET_FIELD_STARTS]->
-                            store_time(&et->starts, MYSQL_TIMESTAMP_DATETIME);
-    }
-
-    if (!et->ends_null)
-    {
-      table->field[ET_FIELD_ENDS]->set_notnull();
-      table->field[ET_FIELD_ENDS]->
-                            store_time(&et->ends, MYSQL_TIMESTAMP_DATETIME);
-    }
-  }
-  else if (et->execute_at.year)
-  {
-    table->field[ET_FIELD_INTERVAL_EXPR]->set_null();
-    table->field[ET_FIELD_TRANSIENT_INTERVAL]->set_null();
-    table->field[ET_FIELD_STARTS]->set_null();
-    table->field[ET_FIELD_ENDS]->set_null();
-    
-    table->field[ET_FIELD_EXECUTE_AT]->set_notnull();
-    table->field[ET_FIELD_EXECUTE_AT]->
-                        store_time(&et->execute_at, MYSQL_TIMESTAMP_DATETIME);
-  }
-  else
-  {
-    DBUG_ASSERT(is_update);
-    /*
-      it is normal to be here when the action is update
-      this is an error if the action is create. something is borked
-    */
-  }
-    
-  ((Field_timestamp *)table->field[ET_FIELD_MODIFIED])->set_time();
-
-  if (et->comment.str)
-  {
-    if (table->field[field_num= ET_FIELD_COMMENT]->
-                 store(et->comment.str, et->comment.length, scs))
-      goto err_truncate;
-  }
-
-  DBUG_RETURN(0);
-err_truncate:
-  my_error(ER_EVENT_DATA_TOO_LONG, MYF(0), table->field[field_num]->field_name);
-  DBUG_RETURN(EVEX_GENERAL_ERROR);
-}
-
-
-/*
-  Puts some data common to CREATE and ALTER EVENT into a row.
-
-  SYNOPSIS
-    evex_fill_row()
-      thd    THD
-      table  the row to fill out
+      table      The row to fill out
       et         Event's data
+      is_update  CREATE EVENT or ALTER EVENT
 
   RETURN VALUE
     0 - OK
@@ -596,7 +470,7 @@ Event_db_repository::find_event(THD *thd, LEX_STRING dbname, LEX_STRING name,
   TABLE *table;
   int ret;
   Event_timed *et= NULL;
-  DBUG_ENTER("db_find_event");
+  DBUG_ENTER("Event_db_repository::find_event");
   DBUG_PRINT("enter", ("name: %*s", name.length, name.str));
 
   if (tbl)
@@ -621,7 +495,7 @@ Event_db_repository::find_event(THD *thd, LEX_STRING dbname, LEX_STRING name,
 
     2)::load_from_row() is silent on error therefore we emit error msg here
   */
-  if ((ret= et->load_from_row(root, table)))
+  if ((ret= et->load_from_row(table)))
   {
     my_error(ER_CANNOT_LOAD_FROM_TABLE, MYF(0), "event");
     goto done;
@@ -722,7 +596,7 @@ evex_check_params(THD *thd, Event_parse_data *parse_data)
   const char *pos= NULL;
   Item *bad_item;
 
-  DBUG_ENTER("evex_check_timing_params");
+  DBUG_ENTER("evex_check_params");
   DBUG_PRINT("info", ("execute_at=0x%d expr=0x%d starts=0x%d ends=0x%d",
                       parse_data->item_execute_at,
                       parse_data->item_expression,
@@ -1212,7 +1086,7 @@ Event_db_repository::drop_events_by_field(THD *thd,
   TABLE *table;
   Open_tables_state backup;
   READ_RECORD read_record_info;
-  DBUG_ENTER("drop_events_from_table_by_field");  
+  DBUG_ENTER("Event_db_repository::drop_events_by_field");  
   DBUG_PRINT("enter", ("field=%d field_value=%s", field, field_value.str));
 
   if (open_event_table(thd, TL_WRITE, &table))
@@ -1270,7 +1144,7 @@ Event_db_repository::load_named_event(THD *thd, LEX_STRING dbname, LEX_STRING na
   Event_timed *et_loaded= NULL;
   Open_tables_state backup;
 
-  DBUG_ENTER("Event_scheduler::load_and_compile_event");
+  DBUG_ENTER("Event_db_repository::load_named_event");
   DBUG_PRINT("enter",("thd=%p name:%*s",thd, name.length, name.str));
 
   thd->reset_n_backup_open_tables_state(&backup);
@@ -1297,4 +1171,3 @@ Event_db_repository::load_named_event(THD *thd, LEX_STRING dbname, LEX_STRING na
 
   DBUG_RETURN(OP_OK);
 }
-

sql/event_queue.cc

@@ -16,7 +16,7 @@
 
 #include "mysql_priv.h"
 #include "events.h"
-#include "event_scheduler.h"
+#include "event_scheduler_ng.h"
 #include "event_queue.h"
 #include "event_data_objects.h"
 #include "event_db_repository.h"
@@ -35,10 +35,6 @@
 #define UNLOCK_QUEUE_DATA() unlock_data(SCHED_FUNC, __LINE__)
 
 
-Event_scheduler*
-Event_queue::singleton= NULL;
-
-
 /*
   Compares the execute_at members of 2 Event_timed instances.
   Used as callback for the prioritized queue when shifting
@@ -111,10 +107,10 @@ Event_queue::create_event(THD *thd, Event_parse_data *et, bool check_existence)
     goto end;
   }
 
-  /* We need to load the event on scheduler_root */
   if (!(res= db_repository->
                 load_named_event(thd, et->dbname, et->name, &et_new)))
   {
+    DBUG_PRINT("info", ("new event in the queue %p", et_new));
     queue_insert_safe(&queue, (byte *) et_new);
     on_queue_change();
   }
@@ -130,7 +126,7 @@ Event_queue::create_event(THD *thd, Event_parse_data *et, bool check_existence)
   Updates an event from the scheduler queue
 
   SYNOPSIS
-    Event_scheduler::update_event()
+    Event_queue::update_event()
       thd        Thread
       et         The event to replace(add) into the queue
       new_schema New schema
@@ -172,15 +168,11 @@ Event_queue::update_event(THD *thd, Event_parse_data *et,
     et->dbname= *new_schema;
     et->name= *new_name;
   }
-  /*
-    We need to load the event (it's strings but on the object itself)
-    on scheduler_root. et_new could be NULL :
-    1. Error occured
-    2. If the replace is DISABLED, we don't load it into the queue.
-  */
+
   if (!(res= db_repository->
             load_named_event(thd, et->dbname, et->name, &et_new)))
   {
+    DBUG_PRINT("info", ("new event in the queue %p old %p", et_new, et_old));
     queue_insert_safe(&queue, (byte *) et_new);
     on_queue_change();
   }
@@ -240,7 +232,7 @@ Event_queue::update_event(THD *thd, Event_parse_data *et,
 
 
 /*
-  Drops an event from the scheduler queue
+  Drops an event from the queue
 
   SYNOPSIS
     Event_queue::drop_event()
@@ -303,10 +295,8 @@ Event_queue::drop_event(THD *thd, sp_name *name)
 }
 
 
-
-
 /*
-  Searches for an event in the scheduler queue
+  Searches for an event in the queue
 
   SYNOPSIS
     Event_queue::find_event()
@@ -358,7 +348,6 @@ Event_queue::find_event(LEX_STRING db, LEX_STRING name, bool remove_from_q)
       comparator     The function to use for comparing
 
   RETURN VALUE
-     -1  Scheduler not working
     >=0  Number of dropped events
     
   NOTE
@@ -426,7 +415,6 @@ Event_queue::drop_matching_events(THD *thd, LEX_STRING pattern,
       db         The schema name
 
   RETURN VALUE
-     -1  Scheduler not working
     >=0  Number of dropped events
 */
 
@@ -459,8 +447,7 @@ void
 Event_queue::lock_data(const char *func, uint line)
 {
   DBUG_ENTER("Event_queue::lock_mutex");
-  DBUG_PRINT("enter", ("mutex_lock=%p func=%s line=%u",
-             &LOCK_event_queue, func, line));
+  DBUG_PRINT("enter", ("func=%s line=%u", func, line));
   pthread_mutex_lock(&LOCK_event_queue);
   mutex_last_locked_in_func= func;
   mutex_last_locked_at_line= line;
@@ -481,9 +468,8 @@ Event_queue::lock_data(const char *func, uint line)
 void
 Event_queue::unlock_data(const char *func, uint line)
 {
-  DBUG_ENTER("Event_queue::UNLOCK_mutex");
-  DBUG_PRINT("enter", ("mutex_unlock=%p func=%s line=%u",
-             &LOCK_event_queue, func, line));
+  DBUG_ENTER("Event_queue::unlock_mutex");
+  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;
@@ -510,7 +496,7 @@ Event_queue::events_count()
   LOCK_QUEUE_DATA();
   n= queue.elements;
   UNLOCK_QUEUE_DATA();
-
+  DBUG_PRINT("info", ("n=%u", n));
   DBUG_RETURN(n);
 }
 
@@ -529,7 +515,7 @@ uint
 Event_queue::events_count_no_lock()
 {
   uint n;
-  DBUG_ENTER("Event_scheduler::events_count_no_lock");
+  DBUG_ENTER("Event_queue::events_count_no_lock");
 
   n= queue.elements;
 
@@ -590,7 +576,7 @@ Event_queue::load_events_from_db(THD *thd)
     }
     DBUG_PRINT("info", ("Loading event from row."));
 
-    if ((ret= et->load_from_row(&scheduler_root, table)))
+    if ((ret= et->load_from_row(table)))
     {
       clean_the_queue= TRUE;
       sql_print_error("SCHEDULER: Error while loading from mysql.event. "
@@ -735,7 +721,7 @@ Event_queue::check_system_tables(THD *thd)
 void
 Event_queue::init_mutexes()
 {
-  pthread_mutex_init(&singleton->LOCK_event_queue, MY_MUTEX_INIT_FAST);
+  pthread_mutex_init(&LOCK_event_queue, MY_MUTEX_INIT_FAST);
 }
 
 
@@ -743,13 +729,13 @@ Event_queue::init_mutexes()
   Destroys mutexes.
 
   SYNOPSIS
-    Event_queue::destroy_mutexes()
+    Event_queue::deinit_mutexes()
 */
 
 void
-Event_queue::destroy_mutexes()
+Event_queue::deinit_mutexes()
 {
-  pthread_mutex_destroy(&singleton->LOCK_event_queue);
+  pthread_mutex_destroy(&LOCK_event_queue);
 }
 
 
@@ -765,8 +751,8 @@ void
 Event_queue::on_queue_change()
 {
   DBUG_ENTER("Event_queue::on_queue_change");
-  DBUG_PRINT("info", ("Sending COND_new_work"));
-  singleton->queue_changed();
+  DBUG_PRINT("info", ("Signalling change of the queue"));
+  scheduler->queue_changed();
   DBUG_VOID_RETURN;
 }
 
@@ -787,13 +773,11 @@ Event_queue::init(Event_db_repository *db_repo)
 {
   int i= 0;
   bool ret= FALSE;
-  DBUG_ENTER("Event_scheduler::init");
+  DBUG_ENTER("Event_queue::init");
   DBUG_PRINT("enter", ("this=%p", this));
 
   LOCK_QUEUE_DATA();
   db_repository= db_repo;
-  /* init memory root */
-  init_alloc_root(&scheduler_root, MEM_ROOT_BLOCK_SIZE, MEM_ROOT_PREALLOC);
 
   if (init_queue_ex(&queue, 30 /*num_el*/, 0 /*offset*/, 0 /*smallest_on_top*/,
                     event_timed_compare_q, NULL, 30 /*auto_extent*/))
@@ -824,8 +808,8 @@ Event_queue::deinit()
   DBUG_ENTER("Event_queue::deinit");
 
   LOCK_QUEUE_DATA();
+  empty_queue();
   delete_queue(&queue);
-  free_root(&scheduler_root, MYF(0));
   UNLOCK_QUEUE_DATA();
 
   DBUG_VOID_RETURN;
@@ -835,7 +819,7 @@ Event_queue::deinit()
 void
 Event_queue::recalculate_queue(THD *thd)
 {
-  int i;
+  uint i;
   for (i= 0; i < queue.elements; i++)
   {
     ((Event_timed*)queue_element(&queue, i))->compute_next_execution_time();
@@ -848,13 +832,118 @@ Event_queue::recalculate_queue(THD *thd)
 void
 Event_queue::empty_queue()
 {
-  int i;
+  uint i;
   /* empty the queue */
   for (i= 0; i < events_count_no_lock(); ++i)
   {
     Event_timed *et= (Event_timed *) queue_element(&queue, i);
-    et->free_sp();
     delete et;
   }
   resize_queue(&queue, 0);
 }
+
+
+Event_timed*
+Event_queue::get_top()
+{
+  return (Event_timed *)queue_top(&queue); 
+}
+
+
+void
+Event_queue::remove_top()
+{
+  queue_remove(&queue, 0);// 0 is top, internally 1
+}
+
+
+void
+Event_queue::top_changed()
+{
+  queue_replaced(&queue);
+}
+
+
+Event_timed *
+Event_queue::get_top_for_execution_if_time(THD *thd, time_t now,
+                                           struct timespec *abstime)
+{
+  struct timespec top_time;
+  Event_timed *et_new= NULL;
+  DBUG_ENTER("Event_queue::get_top_for_execution_if_time");
+  DBUG_PRINT("enter", ("thd=%p now=%d", thd, now));
+  abstime->tv_nsec= 0;
+  LOCK_QUEUE_DATA();
+  do {
+    int res;
+    Event_timed *et= NULL;
+    if (!queue.elements)
+    {
+      abstime->tv_sec= 0;
+      break;
+    }
+    int i;
+    DBUG_PRINT("info", ("Dumping queue . Elements=%u", queue.elements));
+    for (i = 0; i < queue.elements; i++)
+    {
+      et= ((Event_timed*)queue_element(&queue, i));
+      DBUG_PRINT("info",("et=%p db=%s name=%s",et, et->dbname.str, et->name.str));
+      DBUG_PRINT("info", ("exec_at=%llu starts=%llu ends=%llu "
+               " expr=%lld et.exec_at=%d now=%d (et.exec_at - now)=%d if=%d",
+               TIME_to_ulonglong_datetime(&et->execute_at),
+               TIME_to_ulonglong_datetime(&et->starts),
+               TIME_to_ulonglong_datetime(&et->ends),
+               et->expression, sec_since_epoch_TIME(&et->execute_at), now,
+               (int)(sec_since_epoch_TIME(&et->execute_at) - now),
+               sec_since_epoch_TIME(&et->execute_at) <= now));
+    }
+    et= ((Event_timed*)queue_element(&queue, 0));
+    top_time.tv_sec= sec_since_epoch_TIME(&et->execute_at);
+
+    if (top_time.tv_sec <= now)
+    {
+      DBUG_PRINT("info", ("Ready for execution"));
+      abstime->tv_sec= 0;
+      if ((res= db_repository->load_named_event(thd, et->dbname, et->name,
+                                                &et_new)))
+      {
+        DBUG_ASSERT(0);
+        break;
+      }
+
+      et->mark_last_executed(thd);
+      if (et->compute_next_execution_time())
+        et->status= Event_timed::DISABLED;
+      DBUG_PRINT("info", ("event's status is %d", et->status));
+
+      et->update_fields(thd);
+      if (((et->execute_at.year && !et->expression) || et->execute_at_null) ||
+          (et->status == Event_timed::DISABLED))
+      {
+        DBUG_PRINT("info", ("removing from the queue"));
+        if (et->dropped)
+          et->drop(thd);
+        delete et;
+        queue_remove(&queue, 0);
+      }
+      else
+        queue_replaced(&queue);
+    }
+    else
+    {
+      abstime->tv_sec= top_time.tv_sec;
+      DBUG_PRINT("info", ("Have to wait %d till %d", abstime->tv_sec - now,
+                 abstime->tv_sec));
+    }
+  } while (0);
+  UNLOCK_QUEUE_DATA();
+  
+  DBUG_PRINT("info", ("returning. et_new=%p abstime.tv_sec=%d ", et_new,
+             abstime->tv_sec));
+  if (et_new)
+    DBUG_PRINT("info", ("db=%s  name=%s definer=%s "
+               "et_new.execute_at=%lld", et_new->dbname.str, et_new->name.str,
+               et_new->definer.str,
+               TIME_to_ulonglong_datetime(&et_new->execute_at)));
+  DBUG_RETURN(et_new);
+}

sql/event_queue.h

@@ -19,22 +19,23 @@
 class sp_name;
 class Event_timed;
 class Event_db_repository;
+class Event_job_data;
 
 class THD;
 typedef bool * (*event_timed_identifier_comparator)(Event_timed*, Event_timed*);
 
-class Event_scheduler;
+class Event_scheduler_ng;
 
 class Event_queue
 {
 public:
   Event_queue();
 
-  static void
+  void
   init_mutexes();
   
-  static void
-  destroy_mutexes();
+  void
+  deinit_mutexes();
   
   bool
   init(Event_db_repository *db_repo);
@@ -76,6 +77,18 @@ class Event_queue
   void
   empty_queue();
 
+  Event_timed *
+  get_top_for_execution_if_time(THD *thd, time_t now, struct timespec *abstime);
+ 
+  Event_timed*
+  get_top();
+  
+  void
+  remove_top();
+  
+  void
+  top_changed();
+
 ///////////////protected
   Event_timed *
   find_event(LEX_STRING db, LEX_STRING name, bool remove_from_q);
@@ -92,9 +105,6 @@ class Event_queue
 
   Event_db_repository *db_repository;
 
-  /* The MEM_ROOT of the object */
-  MEM_ROOT scheduler_root;
-
   /* The sorted queue with the Event_timed objects */
   QUEUE queue;
   
@@ -111,11 +121,11 @@ class Event_queue
   void
   unlock_data(const char *func, uint line);
 
-  static void
+  void
   on_queue_change();
+  
+  Event_scheduler_ng *scheduler;
 protected:
-  /* Singleton instance */
-  static Event_scheduler *singleton;
 
 };
 

sql/event_scheduler.h

@@ -19,6 +19,7 @@
 class sp_name;
 class Event_timed;
 class Event_db_repository;
+class Event_queue;
 
 class THD;
 
@@ -31,7 +32,7 @@ events_shutdown();
 #include "event_queue.h"
 #include "event_scheduler.h"
 
-class Event_scheduler : public Event_queue
+class Event_scheduler
 {
 public:
   enum enum_state
@@ -56,7 +57,13 @@ class Event_scheduler : public Event_queue
 
 
   static void
-  create_instance();
+  create_instance(Event_queue *queue);
+
+  static void
+  init_mutexes();
+  
+  static void
+  destroy_mutexes();
 
   /* Singleton access */
   static Event_scheduler*
@@ -122,6 +129,8 @@ class Event_scheduler : public Event_queue
   void
   queue_changed();
 
+  Event_queue *event_queue;
+
 protected:
 
   uint
@@ -147,9 +156,11 @@ class Event_scheduler : public Event_queue
   /* Singleton DP is used */
   Event_scheduler();
   
-
+  pthread_mutex_t LOCK_data;
   pthread_mutex_t *LOCK_scheduler_data;
 
+  /* The MEM_ROOT of the object */
+  MEM_ROOT scheduler_root;
 
   /* Set to start the scheduler in suspended state */
   bool start_scheduler_suspended;
@@ -172,18 +183,20 @@ class Event_scheduler : public Event_queue
     COND_LAST
   };
 
-  uint mutex_last_locked_at_line_nr;
-  uint mutex_last_unlocked_at_line_nr;
-  const char* mutex_last_locked_in_func_name;
-  const char* mutex_last_unlocked_in_func_name;
+  uint mutex_last_locked_at_line;
+  uint mutex_last_unlocked_at_line;
+  const char* mutex_last_locked_in_func;
+  const char* mutex_last_unlocked_in_func;
   int cond_waiting_on;
   bool mutex_scheduler_data_locked;
 
-
   static const char * const cond_vars_names[COND_LAST];
 
   pthread_cond_t cond_vars[COND_LAST];
 
+  /* Singleton instance */
+  static Event_scheduler *singleton;
+
 private:
   /* Prevent use of these */
   Event_scheduler(const Event_scheduler &);

sql/event_scheduler_ng.h

+#ifndef _EVENT_SCHEDULER_NG_H_
+#define _EVENT_SCHEDULER_NG_H_
+/* 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; either version 2 of the License, or
+   (at your option) any later version.
+
+   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 */
+
+class Event_timed;
+class Event_queue;
+
+class Event_scheduler_ng
+{
+public:
+  Event_scheduler_ng(){}
+  ~Event_scheduler_ng(){}
+
+  enum enum_state
+  {
+    INITIALIZED = 0,
+    RUNNING,
+    STOPPING
+  };
+
+  /* State changing methods follow */
+
+  bool
+  start();
+
+  bool
+  stop();
+
+  /*
+    Need to be public because has to be called from the function 
+    passed to pthread_create.
+  */
+  bool
+  run(THD *thd);
+  
+  bool 
+  init(Event_queue *queue);
+
+  void
+  deinit();
+
+  void
+  init_mutexes();
+  
+  void
+  deinit_mutexes();
+
+  /* Information retrieving methods follow */
+
+  enum enum_state
+  get_state();
+
+  void
+  queue_changed();
+
+  static int
+  dump_internal_status(THD *thd);
+
+private:
+  uint
+  workers_count();
+
+  /* helper functions */
+  bool
+  execute_top(THD *thd, Event_timed *job_data);
+
+  void
+  stop_all_running_events(THD *thd);
+
+  /* helper functions for working with mutexes & conditionals */
+  void
+  lock_data(const char *func, uint line);
+
+  void
+  unlock_data(const char *func, uint line);
+  
+  pthread_mutex_t LOCK_scheduler_state;
+
+  /* This is the current status of the life-cycle of the scheduler. */
+  enum enum_state state;
+
+  /*
+    Holds the thread id of the executor thread or 0 if the scheduler is not
+    running. It is used by ::shutdown() to know which thread to kill with
+    kill_one_thread(). The latter wake ups a thread if it is waiting on a
+    conditional variable and sets thd->killed to non-zero.
+  */
+  ulong thread_id;
+
+  pthread_cond_t COND_state;
+  
+  Event_queue *queue;
+  Event_db_repository *db_repository;
+
+  uint mutex_last_locked_at_line;
+  uint mutex_last_unlocked_at_line;
+  const char* mutex_last_locked_in_func;
+  const char* mutex_last_unlocked_in_func;
+  bool mutex_scheduler_data_locked;
+
+private:
+  /* Prevent use of these */
+  Event_scheduler_ng(const Event_scheduler_ng &);
+  void operator=(Event_scheduler_ng &);
+};
+
+#endif /* _EVENT_SCHEDULER_NG_H_ */

sql/events.cc

@@ -20,6 +20,7 @@
 #include "event_scheduler.h"
 #include "event_db_repository.h"
 #include "sp_head.h"
+#include "event_scheduler_ng.h"
 
 /*
  TODO list :
@@ -293,9 +294,7 @@ Events::create_event(THD *thd, Event_parse_data *parse_data, uint create_options
                               create_options & HA_LEX_CREATE_IF_NOT_EXISTS,
                               rows_affected)))
   {
-    Event_scheduler *scheduler= Event_scheduler::get_instance();
-    if (scheduler->initialized() &&
-        (ret= scheduler->create_event(thd, parse_data, true)))
+    if ((ret= event_queue->create_event(thd, parse_data, true)))
       my_error(ER_EVENT_MODIFY_QUEUE_ERROR, MYF(0), ret);
   }
   /* No need to close the table, it will be closed in sql_parse::do_command */
@@ -336,9 +335,7 @@ Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *new_name,
   */
   if (!(ret= db_repository->update_event(thd, parse_data, new_name)))
   {
-    Event_scheduler *scheduler= Event_scheduler::get_instance();
-    if (scheduler->initialized() &&
-        (ret= scheduler->update_event(thd, parse_data,
+    if ((ret= event_queue->update_event(thd, parse_data,
                                         new_name? &new_name->m_db: NULL,
                                         new_name? &new_name->m_name: NULL)))
       my_error(ER_EVENT_MODIFY_QUEUE_ERROR, MYF(0), ret);
@@ -373,8 +370,7 @@ Events::drop_event(THD *thd, sp_name *name, bool drop_if_exists,
   if (!(ret= db_repository->drop_event(thd, name->m_db, name->m_name,
                                       drop_if_exists, rows_affected)))
   {
-    Event_scheduler *scheduler= Event_scheduler::get_instance();
-    if (scheduler->initialized() && (ret= scheduler->drop_event(thd, name)))
+    if ((ret= event_queue->drop_event(thd, name)))
       my_error(ER_EVENT_MODIFY_QUEUE_ERROR, MYF(0), ret);
   }
   DBUG_RETURN(ret);
@@ -476,8 +472,7 @@ Events::drop_schema_events(THD *thd, char *db)
   DBUG_ENTER("evex_drop_db_events");  
   DBUG_PRINT("enter", ("dropping events from %s", db));
 
-  Event_scheduler *scheduler= Event_scheduler::get_instance();
-  ret= scheduler->drop_schema_events(thd, db_lex);
+  ret= event_queue->drop_schema_events(thd, db_lex);
   ret= db_repository->drop_schema_events(thd, db_lex);
 
   DBUG_RETURN(ret);
@@ -505,16 +500,18 @@ Events::init()
   Event_db_repository *db_repo;
   DBUG_ENTER("Events::init");
   db_repository->init_repository();
+  event_queue->init(db_repository);
+  event_queue->scheduler= scheduler_ng;
+  scheduler_ng->init(event_queue);
 
   /* it should be an assignment! */
   if (opt_event_scheduler)
   {
-    Event_scheduler *scheduler= Event_scheduler::get_instance();
     DBUG_ASSERT(opt_event_scheduler == 1 || opt_event_scheduler == 2);
-    DBUG_RETURN(scheduler->init(db_repository) || 
-                (opt_event_scheduler == 1? scheduler->start():
-                                           scheduler->start_suspended()));
+    if (opt_event_scheduler == 1)
+      DBUG_RETURN(scheduler_ng->start());
   }
+
   DBUG_RETURN(0);
 }
 
@@ -534,13 +531,9 @@ Events::deinit()
 {
   DBUG_ENTER("Events::deinit");
 
-  Event_scheduler *scheduler= Event_scheduler::get_instance();
-  if (scheduler->initialized())
-  {
-    scheduler->stop();
-    scheduler->destroy();
-  }
-
+  scheduler_ng->stop();
+  scheduler_ng->deinit();
+  event_queue->deinit();
   db_repository->deinit_repository();
 
   DBUG_VOID_RETURN;
@@ -559,8 +552,12 @@ void
 Events::init_mutexes()
 {
   db_repository= new Event_db_repository;
-  Event_scheduler::create_instance();
-  Event_scheduler::init_mutexes();
+
+  event_queue= new Event_queue;
+  event_queue->init_mutexes();
+
+  scheduler_ng= new Event_scheduler_ng();
+  scheduler_ng->init_mutexes();
 }
 
 
@@ -574,9 +571,11 @@ Events::init_mutexes()
 void
 Events::destroy_mutexes()
 {
-  Event_scheduler::destroy_mutexes();
+  event_queue->deinit_mutexes();
+  scheduler_ng->deinit_mutexes();
+  
+  delete scheduler_ng;
   delete db_repository;
-  db_repository= NULL;
 }
 
 
@@ -595,7 +594,7 @@ Events::destroy_mutexes()
 int
 Events::dump_internal_status(THD *thd)
 {
-  return Event_scheduler::dump_internal_status(thd);
+  return Event_scheduler_ng::dump_internal_status(thd);
 }
 
 
@@ -633,3 +632,26 @@ Events::fill_schema_events(THD *thd, TABLE_LIST *tables, COND * /* cond */)
   }
   DBUG_RETURN(get_instance()->db_repository->fill_schema_events(thd, tables, db));
 }
+
+
+bool
+Events::start_execution_of_events()
+{
+  DBUG_ENTER("Events::start_execution_of_events");
+  DBUG_RETURN(scheduler_ng->start());
+}
+
+
+bool
+Events::stop_execution_of_events()
+{
+  DBUG_ENTER("Events::stop_execution_of_events");
+  DBUG_RETURN(scheduler_ng->stop());
+}
+
+bool
+Events::is_started()
+{
+  DBUG_ENTER("Events::is_started");
+  DBUG_RETURN(scheduler_ng->get_state() == Event_scheduler_ng::RUNNING);
+}

sql/events.h

@@ -19,6 +19,8 @@
 class sp_name;
 class Event_parse_data;
 class Event_db_repository;
+class Event_queue;
+class Event_scheduler_ng;
 
 /* Return codes */
 enum enum_events_error_code
@@ -60,6 +62,15 @@ class Events
   void
   destroy_mutexes();
 
+  bool
+  start_execution_of_events();
+  
+  bool
+  stop_execution_of_events();
+  
+  bool
+  is_started();
+
   static Events*
   get_instance();
 
@@ -95,6 +106,8 @@ class Events
   dump_internal_status(THD *thd);
 
   Event_db_repository *db_repository;
+  Event_queue *event_queue;
+  Event_scheduler_ng *scheduler_ng;
 
 private:
   /* Singleton DP is used */

sql/mysqld.cc

@@ -864,7 +864,7 @@ static void close_connections(void)
     DBUG_PRINT("quit",("Informing thread %ld that it's time to die",
 		       tmp->thread_id));
     /* We skip slave threads & scheduler on this first loop through. */
-    if (tmp->slave_thread || tmp->system_thread == SYSTEM_THREAD_EVENT_SCHEDULER)
+    if (tmp->slave_thread)
       continue;
 
     tmp->killed= THD::KILL_CONNECTION;

sql/set_var.cc

@@ -58,6 +58,7 @@
 #include <my_dir.h>
 
 #include "event_scheduler.h"
+#include "events.h"
 
 /* WITH_BERKELEY_STORAGE_ENGINE */
 extern bool berkeley_shared_data;
@@ -3896,26 +3897,29 @@ sys_var_event_scheduler::update(THD *thd, set_var *var)
   Event_scheduler *scheduler= Event_scheduler::get_instance();
   /* here start the thread if not running. */
   DBUG_ENTER("sys_var_event_scheduler::update");
-
-  DBUG_PRINT("new_value", ("%lu", (bool)var->save_result.ulong_value));
-  if (!scheduler->initialized())
+  if (Events::opt_event_scheduler == 0)
   {
     my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--event-scheduler=0");
-    DBUG_RETURN(true);
+    DBUG_RETURN(TRUE);
   }
 
+  DBUG_PRINT("new_value", ("%lu", (bool)var->save_result.ulong_value));
+
   if (var->save_result.ulonglong_value < 1 ||
       var->save_result.ulonglong_value > 2)
   {
     char buf[64];
     my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), "event_scheduler",
              llstr(var->save_result.ulonglong_value, buf));
-    DBUG_RETURN(true);
+    DBUG_RETURN(TRUE);
   }
-  if ((res= scheduler->suspend_or_resume(var->save_result.ulonglong_value == 1?
-                                         Event_scheduler::RESUME :
-                                         Event_scheduler::SUSPEND)))
-    my_error(ER_EVENT_SET_VAR_ERROR, MYF(0), (uint) res);
+  if (var->save_result.ulonglong_value == 1)
+    res= Events::get_instance()->start_execution_of_events();
+  else
+    res= Events::get_instance()->stop_execution_of_events();
+
+  if (res)
+    my_error(ER_EVENT_SET_VAR_ERROR, MYF(0));
   DBUG_RETURN((bool) res);
 }
 
@@ -3925,9 +3929,9 @@ byte *sys_var_event_scheduler::value_ptr(THD *thd, enum_var_type type,
 {
   Event_scheduler *scheduler= Event_scheduler::get_instance();
 
-  if (!scheduler->initialized())
+  if (Events::opt_event_scheduler == 0)
     thd->sys_var_tmp.long_value= 0;
-  else if (scheduler->get_state() == Event_scheduler::RUNNING)
+  else if (Events::get_instance()->is_started())
     thd->sys_var_tmp.long_value= 1;
   else
     thd->sys_var_tmp.long_value= 2;

sql/share/errmsg.txt

@@ -5831,7 +5831,7 @@ ER_DUP_ENTRY_AUTOINCREMENT_CASE
 ER_EVENT_MODIFY_QUEUE_ERROR
         eng "Internal scheduler error %d"
 ER_EVENT_SET_VAR_ERROR
-        eng "Error during starting/stopping of the scheduler. Error code %u"
+        eng "Error during starting/stopping of the scheduler."
 ER_PARTITION_MERGE_ERROR
         eng "%s handler cannot be used in partitioned tables"
         swe "%s kan inte anvndas i en partitionerad tabell"

sql/sql_show.cc

@@ -4177,7 +4177,7 @@ copy_event_to_schema_table(THD *thd, TABLE *sch_table, TABLE *event_table)
 
   restore_record(sch_table, s->default_values);
 
-  if (et.load_from_row(thd->mem_root, event_table))
+  if (et.load_from_row(event_table))
   {
     my_error(ER_CANNOT_LOAD_FROM_TABLE, MYF(0));
     DBUG_RETURN(1);