Task ID 499:Add a new settable string variable(init_connect, init_slave)

to mysqld that is executed for all new connections.
(Similar to the client command: mysql_options(... MYSQL_INIT_COMMAND ...).

sql/mysql_priv.h

@@ -499,6 +499,8 @@ bool wait_for_tables(THD *thd);
 bool table_is_used(TABLE *table, bool wait_for_name_lock);
 bool drop_locked_tables(THD *thd,const char *db, const char *table_name);
 void abort_locked_tables(THD *thd,const char *db, const char *table_name);
+void execute_init_command(THD *thd, sys_var_str *init_command_var,
+			  rw_lock_t *var_mutex);
 extern const Field *not_found_field;
 Field *find_field_in_tables(THD *thd, Item_ident *item, TABLE_LIST *tables,
 			    TABLE_LIST **where, bool report_error);

sql/mysqld.cc

@@ -3655,10 +3655,10 @@ Disable with --skip-bdb (will save memory).",
 #endif /* End HAVE_INNOBASE_DB */
   {"help", '?', "Display this help and exit.", 0, 0, 0, GET_NO_ARG, NO_ARG, 0,
    0, 0, 0, 0, 0},
-  {"init-connect", OPT_INIT_CONNECT, "Command what executes for all new connections",
+  {"init-connect", OPT_INIT_CONNECT, "Command(s) that are executed for each new connection",
    (gptr*) &opt_init_connect, (gptr*) &opt_init_connect, 0, GET_STR_ALLOC, REQUIRED_ARG,
    0, 0, 0, 0, 0, 0},
-  {"init-slave", OPT_INIT_SLAVE, "Command what is executed when replication is starting",
+  {"init-slave", OPT_INIT_SLAVE, "Command(s) that are executed when a slave connects to this master",
    (gptr*) &opt_init_slave, (gptr*) &opt_init_slave, 0, GET_STR_ALLOC, REQUIRED_ARG,
    0, 0, 0, 0, 0, 0},
   {"init-file", OPT_INIT_FILE, "Read SQL commands from this file at startup.",

sql/protocol.cc

@@ -92,7 +92,6 @@ void send_error(THD *thd, uint sql_errno, const char *err)
       /* In bootstrap it's ok to print on stderr */
       fprintf(stderr,"ERROR: %d  %s\n",sql_errno,err);
     }
-    thd->init_connect_error= 1;
     DBUG_VOID_RETURN;
   }
 
@@ -211,7 +210,6 @@ net_printf(THD *thd, uint errcode, ...)
       fprintf(stderr,"ERROR: %d  %s\n",errcode,text_pos);
       thd->fatal_error();
     }
-    thd->init_connect_error= 1;
     DBUG_VOID_RETURN;
   }
 

sql/set_var.cc

@@ -674,14 +674,23 @@ struct show_var_st init_vars[]= {
 */
 
 
-static bool sys_update_init_connect(THD *thd, set_var *var)
+/*
+  Update variables 'init_connect, init_slave'.
+
+  In case of 'DEFAULT' value
+  (for example: 'set GLOBAL init_connect=DEFAULT')
+  'var' parameter is NULL pointer.
+*/
+
+bool update_sys_var_str(sys_var_str *var_str, rw_lock_t *var_mutex,
+			set_var *var)
 {
   char *res= 0, *old_value;
-  uint new_length;
+  uint new_length= 0;
   /* If the string is "", delete old init command */
-  if ((new_length= var->value->str_value.length()))
+  if (var && (new_length= var->value->str_value.length()))
   {
-    if (!(res= my_strdup_with_length(var->value->str_value.c_ptr(),
+    if (!(res= my_strdup_with_length(var->value->str_value.ptr(),
 		     new_length,
 		     MYF(0))))
       return 1;
@@ -690,63 +699,37 @@ static bool sys_update_init_connect(THD *thd, set_var *var)
     Replace the old value in such a way that the any thread using
     the value will work.
   */
-  rw_wrlock(&LOCK_sys_init_connect);
-  old_value= sys_init_connect.value;
-  sys_init_connect.value= res;
-  sys_init_connect.value_length= new_length;
-  rw_unlock(&LOCK_sys_init_connect);
+  rw_wrlock(var_mutex);
+  old_value= var_str->value;
+  var_str->value= res;
+  var_str->value_length= new_length;
+  rw_unlock(var_mutex);
   my_free(old_value, MYF(MY_ALLOW_ZERO_PTR));
   return 0;
 }
 
 
+static bool sys_update_init_connect(THD *thd, set_var *var)
+{
+  return update_sys_var_str(&sys_init_connect, &LOCK_sys_init_connect, var);
+}
+
+
 static void sys_default_init_connect(THD* thd, enum_var_type type)
 {
-  char *old_value;
-  rw_wrlock(&LOCK_sys_init_connect);
-  old_value= sys_init_connect.value;
-  sys_init_connect.value= 0;
-  sys_init_connect.value_length= 0;
-  rw_unlock(&LOCK_sys_init_connect);
-  my_free(old_value, MYF(MY_ALLOW_ZERO_PTR));
+  update_sys_var_str(&sys_init_connect, &LOCK_sys_init_connect, 0);
 }
 
 
 static bool sys_update_init_slave(THD *thd, set_var *var)
 {
-  char *res= 0, *old_value;
-  uint new_length;
-  /* If the string is "", delete old init command */
-  if ((new_length= var->value->str_value.length()))
-  {
-    if (!(res= my_strdup_with_length(var->value->str_value.c_ptr(),
-		     new_length,
-		     MYF(0))))
-      return 1;
-  }
-  /*
-    Replace the old value in such a way that the any thread using
-    the value will work.
-  */
-  rw_wrlock(&LOCK_sys_init_slave);
-  old_value= sys_init_slave.value;
-  sys_init_slave.value= res;
-  sys_init_slave.value_length= new_length;
-  rw_unlock(&LOCK_sys_init_slave);
-  my_free(old_value, MYF(MY_ALLOW_ZERO_PTR));
-  return 0;
+  return update_sys_var_str(&sys_init_slave, &LOCK_sys_init_slave, var);
 }
 
 
 static void sys_default_init_slave(THD* thd, enum_var_type type)
 {
-  char *old_value;
-  rw_wrlock(&LOCK_sys_init_slave);
-  old_value= sys_init_slave.value;
-  sys_init_slave.value= 0;
-  sys_init_slave.value_length= 0;
-  rw_unlock(&LOCK_sys_init_slave);
-  my_free(old_value, MYF(MY_ALLOW_ZERO_PTR));
+  update_sys_var_str(&sys_init_slave, &LOCK_sys_init_slave, 0);
 }
 
 

sql/slave.cc

@@ -2448,24 +2448,6 @@ err:
 }
 
 
-void init_slave_execute(THD *thd, sys_var_str *init_slave_var)
-{
-  Vio* save_vio;
-  ulong save_client_capabilities;
-
-  thd->proc_info= "Execution of init_slave";
-  thd->query= init_slave_var->value;
-  thd->query_length= init_slave_var->value_length;
-  save_client_capabilities= thd->client_capabilities;
-  thd->client_capabilities|= CLIENT_MULTI_QUERIES;
-  save_vio= thd->net.vio;
-  thd->net.vio= 0;
-  dispatch_command(COM_QUERY, thd, thd->query, thd->query_length+1);
-  thd->client_capabilities= save_client_capabilities;
-  thd->net.vio= save_vio;
-}
-
-
 /* Slave SQL Thread entry point */
 
 extern "C" pthread_handler_decl(handle_slave_sql,arg)
@@ -2551,9 +2533,7 @@ log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
   /* execute init_slave variable */
   if (sys_init_slave.value)
   {
-    rw_wrlock(&LOCK_sys_init_slave);
-    init_slave_execute(thd, &sys_init_slave);
-    rw_unlock(&LOCK_sys_init_slave);
+    execute_init_command(thd, &sys_init_slave, &LOCK_sys_init_slave);
     if (thd->query_error)
     {
       sql_print_error("\

sql/sql_class.cc

@@ -121,7 +121,6 @@ THD::THD():user_time(0), is_fatal_error(0),
   system_thread=cleanup_done=0;
   peer_port= 0;					// For SHOW PROCESSLIST
   transaction.changed_tables = 0;
-  init_connect_error= 0;
 #ifdef	__WIN__
   real_id = 0;
 #endif

sql/sql_class.h

@@ -560,7 +560,6 @@ public:
   bool       prepare_command;
   bool	     tmp_table_used;
 
-  bool       init_connect_error;
   /*
     If we do a purge of binary logs, log index info of the threads
     that are currently reading it needs to be adjusted. To do that

sql/sql_parse.cc

@@ -794,19 +794,31 @@ check_connections(THD *thd)
 }
 
 
-void init_connect_execute(THD *thd, sys_var_str *init_connect_var)
+void execute_init_command(THD *thd, sys_var_str *init_command_var,
+			  rw_lock_t *var_mutex)
 {
   Vio* save_vio;
   ulong save_client_capabilities;
 
-  thd->proc_info= "Execution of init_connect";
-  thd->query= init_connect_var->value;
-  thd->query_length= init_connect_var->value_length;
+  thd->proc_info= "Execution of init_command";
+  /*
+    We need to lock init_command_var because
+    during execution of init_command_var query
+    values of init_command_var can't be changed
+  */
+  rw_rdlock(var_mutex);
+  thd->query= init_command_var->value;
+  thd->query_length= init_command_var->value_length;
   save_client_capabilities= thd->client_capabilities;
   thd->client_capabilities|= CLIENT_MULTI_QUERIES;
+  /*
+    We don't need return result of execution to client side.
+    To forbid this we should set thd->net.vio to 0.
+  */
   save_vio= thd->net.vio;
   thd->net.vio= 0;
   dispatch_command(COM_QUERY, thd, thd->query, thd->query_length+1);
+  rw_unlock(var_mutex);
   thd->client_capabilities= save_client_capabilities;
   thd->net.vio= save_vio;
 }
@@ -887,29 +899,9 @@ pthread_handler_decl(handle_one_connection,arg)
     thd->version= refresh_version;
     if (sys_init_connect.value && !(thd->master_access & SUPER_ACL))
     {
-      rw_wrlock(&LOCK_sys_init_connect);
-      init_connect_execute(thd, &sys_init_connect);
-      rw_unlock(&LOCK_sys_init_connect);
-      if (thd->init_connect_error)
-      {
-        if (thd->user_connect)
-          decrease_user_connections(thd->user_connect);
-        free_root(&thd->mem_root,MYF(0));
-        if (!thd->killed && thd->variables.log_warnings)
-        {
-          sql_print_error(ER(ER_NEW_ABORTING_CONNECTION),
-	          thd->thread_id,(thd->db ? thd->db : "unconnected"),
-	          thd->user ? thd->user : "unauthenticated",
-	          thd->host_or_ip,
-	          "Can't execute init_connect query");
-          statistic_increment(aborted_threads,&LOCK_status);
-        }
-        else if (thd->killed)
-        {
-          statistic_increment(aborted_threads,&LOCK_status);
-        }
-        goto end_thread;
-      }
+      execute_init_command(thd, &sys_init_connect, &LOCK_sys_init_connect);
+      if (thd->query_error)
+	thd->killed= 1;
     }
 
     thd->proc_info=0;