Commit f92086d0 authored by unknown's avatar unknown
Browse files

WL#1034 update 

(cleanups, leaks fixed)


sql/event.cc:
  - update comments
  - remove dead code
  - event_timed is no more copied after allocation on a DYNAMIC_ARRAY
    because there is a problem when the array is reallocated - we get
    dangling pointers from the scheduling queue. anyway it makes little
    sense to keep them there except that cleaning is quite efficient but
    iterating over all events and cleaning them one by one is not that bad
    considering that happens only when the main scheduler thread is killed
    or during server shutdown.
sql/event_executor.cc:
  - DYNAMIC_ARRAY is no more
sql/event_priv.h:
  - remove unneeded code/defines. the scheduler's queue is of 
    type QUEUE and cannot run anymore on top of DYNAMIC_ARRAY
sql/event_timed.cc:
  - after parsing (in ::compile() ) destruct the event_timed object
sql/sql_parse.cc:
  in case of syntax error clean up lex->et because there could be an object
  created. as in the code the same is done for SPs.
parent 112d408f

sql/event.cc

+6 −89
Original line number Diff line number Diff line
@@ -23,8 +23,6 @@ 
 - The default value of created/modified should not be 0000-00-00 because of

   STRICT mode restricions.



 - Use timestamps instead of datetime.



 - CREATE EVENT should not go into binary log! Does it now? The SQL statements

   issued by the EVENT are replicated.

   I have an idea how to solve the problem at failover. So the status field
@@ -43,13 +41,14 @@ 
 - Maybe move all allocations during parsing to evex_mem_root thus saving

    double parsing in evex_create_event!



 - If the server is killed (stopping) try to kill executing events..

 - If the server is killed (stopping) try to kill executing events?

 

 - What happens if one renames an event in the DB while it is in memory?

   Or even deleting it?

  

 - Consider using conditional variable when doing shutdown instead of

   waiting till all worker threads end.

 

 - Make event_timed::get_show_create_event() work



 - Add function documentation whenever needed.
@@ -58,10 +57,6 @@ 


 - Move comparison code to class event_timed



 - Overload event_timed::new to put the event directly in the DYNAMIC_ARRAY.

   This will skip copy operation as well as will simplify the code which is

   now aware of events_array DYNAMIC_ARRAY



Warning:

 - For now parallel execution is not possible because the same sp_head cannot be

   executed few times!!! There is still no lock attached to particular event.
@@ -72,7 +67,6 @@ 




bool mysql_event_table_exists= 1;

DYNAMIC_ARRAY events_array;

QUEUE EVEX_EQ_NAME;

MEM_ROOT evex_mem_root;
@@ -81,54 +75,12 @@ MEM_ROOT evex_mem_root; 
void

evex_queue_init(EVEX_QUEUE_TYPE *queue)

{

#ifndef EVEX_USE_QUEUE

  VOID(my_init_dynamic_array(queue, sizeof(event_timed *), 50, 100));

#else

  if (init_queue_ex(queue, 100 /*num_el*/, 0 /*offset*/, 

                   0 /*smallest_on_top*/, event_timed_compare_q, NULL,

                   100 /*auto_extent*/))

    sql_print_error("Insufficient memory to initialize executing queue.");

#endif

}





int

evex_queue_insert2(EVEX_QUEUE_TYPE *queue, EVEX_PTOQEL element)

{

#ifndef EVEX_USE_QUEUE

  VOID(push_dynamic(queue, element));

  return 0;

#else

  return queue_insert_safe(queue, element);  

#endif

}



void

evex_queue_top_updated(EVEX_QUEUE_TYPE *queue)

{

#ifdef EVEX_USE_QUEUE

  queue_replaced(queue);

#endif

}



void

evex_queue_sort(EVEX_QUEUE_TYPE *queue)

{

#ifndef EVEX_USE_QUEUE

  qsort((gptr) dynamic_element(queue, 0, event_timed**),

                               queue->elements,

                               sizeof(event_timed **),

                               (qsort_cmp) event_timed_compare);

#endif

}



/* NOTE Andrey: Document better

  Compares two TIME structures.



  a > b ->  1

  a = b ->  0

  a < b -> -1

*/



static

int sortcmp_lex_string(LEX_STRING s, LEX_STRING t, CHARSET_INFO *cs)
@@ -714,7 +666,7 @@ evex_load_and_compile_event(THD * thd, sp_name *spn, bool use_lock) 
{

  int ret= 0;

  MEM_ROOT *tmp_mem_root;

  event_timed *ett, *ett_copy;

  event_timed *ett;



  DBUG_ENTER("db_load_and_compile_event");

  DBUG_PRINT("enter", ("name: %*s", spn->m_name.length, spn->m_name.str));
@@ -737,18 +689,12 @@ evex_load_and_compile_event(THD * thd, sp_name *spn, bool use_lock) 
  if (use_lock)

    VOID(pthread_mutex_lock(&LOCK_event_arrays));



  VOID(push_dynamic(&events_array,(gptr) ett));

  ett_copy= dynamic_element(&events_array, events_array.elements - 1,

                            event_timed*);



  evex_queue_insert(&EVEX_EQ_NAME, (EVEX_PTOQEL) ett_copy);

  evex_queue_insert(&EVEX_EQ_NAME, (EVEX_PTOQEL) ett);



  /*

    There is a copy in the array which we don't need. sphead won't be

    destroyed.

  */

  ett->free_sphead_on_delete= false;

  delete ett;



  if (use_lock)

    VOID(pthread_mutex_unlock(&LOCK_event_arrays));
@@ -783,43 +729,14 @@ evex_remove_from_cache(LEX_STRING *db, LEX_STRING *name, bool use_lock) 
    if (!sortcmp_lex_string(*name, et->name, system_charset_info) &&

        !sortcmp_lex_string(*db, et->dbname, system_charset_info))

    {

      int idx= get_index_dynamic(&events_array, (gptr) et);

      //we are lucky the event is in the executing queue, no need of second pass

      //destruct first and then remove. the destructor will delete sp_head

      et->free_sp();

      delete_dynamic_element(&events_array, idx);

      delete et;

      evex_queue_delete_element(&EVEX_EQ_NAME, i);

      // ok, we have cleaned

      goto done;

    }

  }



  /*

    ToDo Andrey : Think about whether second pass is needed. All events

                  that are in memory are enabled. If an event is being

                  disabled (by a SQL stmt) it will be uncached. Hmm...

                  However is this true for events that has been 

                  disabled because of another reason like - no need

                  to be executed because ENDS is in the past?

                  For instance, second_pass is needed when an event

                  was created as DISABLED but then altered as ENABLED.

  */

  /*

    we haven't found the event in the executing queue. This is nice! :)

    Look for it in the events_array.

  */

  for (i= 0; i < events_array.elements; ++i)

  {

    event_timed *et= dynamic_element(&events_array, i, event_timed*);

      

    if (!sortcmp_lex_string(*name, et->name, system_charset_info) &&

        !sortcmp_lex_string(*db, et->dbname, system_charset_info))

    {

      delete_dynamic_element(&events_array, i);

      break;

    }

  } 



done:

  if (use_lock)

    VOID(pthread_mutex_unlock(&LOCK_event_arrays));

sql/event_executor.cc

+25 −47
Original line number Diff line number Diff line
@@ -160,6 +160,7 @@ event_executor_main(void *arg) 
  THD *thd;			/* needs to be first for thread_stack */

  ulonglong iter_num= 0;

  uint i=0, j=0;

  my_ulonglong cnt= 0;



  DBUG_ENTER("event_executor_main");

  DBUG_PRINT("event_executor_main", ("EVEX thread started"));
@@ -194,24 +195,16 @@ event_executor_main(void *arg) 
  thread_running++;

  VOID(pthread_mutex_unlock(&LOCK_thread_count));



  DBUG_PRINT("EVEX main thread", ("Initing events_array"));



  VOID(pthread_mutex_lock(&LOCK_event_arrays));

  /*

    my_malloc is used as underlying allocator which does not use a mem_root

    thus data should be freed at later stage.

  */

  VOID(my_init_dynamic_array(&events_array, sizeof(event_timed), 50, 100));



  evex_queue_init(&EVEX_EQ_NAME);



  VOID(pthread_mutex_unlock(&LOCK_event_arrays));  

  DBUG_PRINT("EVEX main thread", ("Initing events_queuey"));



  /*

    eventually manifest that we are running, not to crashe because of

    usage of non-initialized memory structures.

  */

  VOID(pthread_mutex_lock(&LOCK_evex_running));

  VOID(pthread_mutex_lock(&LOCK_event_arrays));

  evex_queue_init(&EVEX_EQ_NAME);

  VOID(pthread_mutex_unlock(&LOCK_event_arrays));  

  evex_is_running= true;  

  event_executor_running_global_var= opt_event_executor;

  VOID(pthread_mutex_unlock(&LOCK_evex_running));
@@ -222,15 +215,16 @@ event_executor_main(void *arg) 
  THD_CHECK_SENTRY(thd);

  /* Read queries from the IO/THREAD until this thread is killed */

  evex_main_thread_id= thd->thread_id;



  sql_print_information("Scheduler thread started");

  while (!thd->killed)

  {

    TIME time_now;

    my_time_t now;

    my_ulonglong cnt;

    event_timed *et;

    

    DBUG_PRINT("info", ("EVEX External Loop %d", ++cnt));

    cnt++;

    DBUG_PRINT("info", ("EVEX External Loop %d", cnt));

    if (cnt > 1000) continue;

    thd->proc_info = "Sleeping";

    if (!evex_queue_num_elements(EVEX_EQ_NAME) ||

        !event_executor_running_global_var)
@@ -326,16 +320,10 @@ event_executor_main(void *arg) 
    }

    if ((et->flags & EVENT_EXEC_NO_MORE) || et->status == MYSQL_EVENT_DISABLED)

    {

      evex_queue_delete_element(&EVEX_EQ_NAME, 1);// 1 is top

      if (et->dropped)

      {

        // we have to drop the event

        int idx;

        et->drop(thd);

        idx= get_index_dynamic(&events_array, (gptr) et);

        DBUG_ASSERT(idx != -1);

        delete_dynamic_element(&events_array, idx);

      }

      delete et;

      evex_queue_delete_element(&EVEX_EQ_NAME, 1);// 1 is top

    } else

      evex_queue_first_updated(&EVEX_EQ_NAME);
@@ -365,15 +353,15 @@ event_executor_main(void *arg) 
    LEX_STRINGs reside in the memory root and will be destroyed with it.

    Hence no need of delete but only freeing of SP

  */

  for (i= 0; i < events_array.elements; ++i)

    dynamic_element(&events_array, i, event_timed*)->free_sp();



  VOID(pthread_mutex_lock(&LOCK_event_arrays));

  // No need to use lock here if EVEX is not running but anyway

  delete_queue(&executing_queue);

  // First we free all objects ...

  for (i= 0; i < evex_queue_num_elements(EVEX_EQ_NAME); ++i)

  {

    event_timed *et= evex_queue_element(&EVEX_EQ_NAME, i, event_timed*);

    et->free_sp();

    delete et;

  }

  // ... then we can thras the whole queue at once

  evex_queue_destroy(&EVEX_EQ_NAME);

  delete_dynamic(&events_array);

  VOID(pthread_mutex_unlock(&LOCK_event_arrays));

  

  thd->proc_info = "Clearing";

  DBUG_ASSERT(thd->net.buff != 0);
@@ -529,7 +517,7 @@ evex_load_events_from_db(THD *thd) 
  init_read_record(&read_record_info, thd, table ,NULL,1,0);

  while (!(read_record_info.read_record(&read_record_info)))

  {

    event_timed *et, *et_copy;

    event_timed *et;

    if (!(et= new event_timed))

    {

      DBUG_PRINT("evex_load_events_from_db", ("Out of memory"));
@@ -564,20 +552,10 @@ evex_load_events_from_db(THD *thd) 
    et->compute_next_execution_time();

    

    DBUG_PRINT("evex_load_events_from_db", ("Adding to the exec list."));

    VOID(push_dynamic(&events_array,(gptr) et));

    /*

      We always add at the end so the number of elements - 1 is the place

      in the buffer.

      DYNAMIC_ARRAY copies the object bit by bit so we have a hollow copy

      in event_array. We don't need the original therefore we delete it.

    */

    et_copy= dynamic_element(&events_array, events_array.elements - 1,

                             event_timed*);



    evex_queue_insert(&EVEX_EQ_NAME, (EVEX_PTOQEL) et_copy);

    printf("%p %s\n", et_copy, et_copy->name.str);

    et->free_sphead_on_delete= false;

    delete et; 

    evex_queue_insert(&EVEX_EQ_NAME, (EVEX_PTOQEL) et);

    DBUG_PRINT("evex_load_events_from_db", ("%p %*s",

                et, et->name.length,et->name.str));

  }



  ret= 0;

sql/event_priv.h

+12 −35
Original line number Diff line number Diff line
@@ -67,52 +67,29 @@ evex_time_diff(TIME *a, TIME *b); 
#define EXEC_QUEUE_QUEUE_NAME executing_queue

#define EXEC_QUEUE_DARR_NAME evex_executing_queue



#ifdef EVEX_USE_QUEUE



#define EVEX_QUEUE_TYPE QUEUE

#define EVEX_PTOQEL byte *

 #define EVEX_EQ_NAME executing_queue



#define EVEX_EQ_NAME executing_queue

#define evex_queue_first_element(queue, __cast) ((__cast)queue_top(queue))

 #define evex_queue_element(queue, idx, __cast) ((__cast)queue_top(queue))

#define evex_queue_element(queue, idx, __cast) ((__cast)queue_element(queue, idx))

#define evex_queue_delete_element(queue, idx)  queue_remove(queue, idx)

#define evex_queue_destroy(queue)              delete_queue(queue)

#define evex_queue_first_updated(queue)        queue_replaced(queue)

#define evex_queue_insert(queue, element)      queue_insert_safe(queue, element);



#else

 #define EVEX_QUEUE_TYPE DYNAMIC_ARRAY

 #define EVEX_PTOQEL gptr

 #define EVEX_EQ_NAME evex_executing_queue



 #define evex_queue_element(queue, idx, __cast) dynamic_element(queue,idx, __cast)

 #define evex_queue_delete_element(queue, idx)  delete_dynamic_element(queue, idx);

 #define evex_queue_destroy(queue)              delete_dynamic(queue)

/*

  push_dynamic() expects ptr to the memory to put in, to make things fast

  so when a pointer has to be put inside a ptr-to-ptr is being passed

*/

 #define evex_queue_first_updated(queue)

 #define evex_queue_insert(queue, element)      VOID(push_dynamic(queue, &element))

 

 

#endif





void

evex_queue_init(EVEX_QUEUE_TYPE *queue);



int

evex_queue_insert2(EVEX_QUEUE_TYPE *queue, EVEX_PTOQEL element);



void

evex_queue_sort(EVEX_QUEUE_TYPE *queue);



#define evex_queue_num_elements(queue) queue.elements





extern bool evex_is_running;

extern bool mysql_event_table_exists;

extern DYNAMIC_ARRAY events_array;

//extern DYNAMIC_ARRAY events_array;

extern MEM_ROOT evex_mem_root;

extern pthread_mutex_t LOCK_event_arrays,

                       LOCK_workers_count,

sql/event_timed.cc

+1 −0
Original line number Diff line number Diff line
@@ -939,6 +939,7 @@ event_timed::compile(THD *thd, MEM_ROOT *mem_root) 
  sphead->optimize();

  ret= 0;

done:

  delete lex.et;

  lex_end(&lex);

  thd->lex= old_lex;

  thd->query= old_query;

sql/sql_parse.cc

+12 −0
Original line number Diff line number Diff line
@@ -3685,9 +3685,11 @@ mysql_execute_command(THD *thd) 
        res= true;

        break;

      }



      if (check_access(thd, EVENT_ACL, lex->et->dbname.str, 0, 0, 0,

                       is_schema_db(lex->et->dbname.str)))

        break;



      switch (lex->sql_command) {

      case SQLCOM_CREATE_EVENT:

        res= evex_create_event(thd, lex->et, (uint) lex->create_info.options);
@@ -5652,6 +5654,11 @@ void mysql_parse(THD *thd, char *inBuf, uint length) 
	    delete thd->lex->sphead;

	    thd->lex->sphead= NULL;

	  }

          if (thd->lex->et)

          {

            delete thd->lex->et;

            thd->lex->et= NULL;

          }

	}

	else

	{
@@ -5687,6 +5694,11 @@ void mysql_parse(THD *thd, char *inBuf, uint length) 
	delete thd->lex->sphead;

	thd->lex->sphead= NULL;

      }

      if (thd->lex->et)

      {

        delete thd->lex->et;

        thd->lex->et= NULL;

      }

    }

    thd->proc_info="freeing items";

    thd->end_statement();