bzr push into mysql-next-mr-wl5569 branch (andrei.elkin:3256 to 3257) WL#5569 WL#5599

[Andrei Elkin]

 3257 Andrei Elkin	2010-12-22
      wl#5569 MTS and wl#5599 MTS recovery
      
      The general recovery implementation is finished by this patch.
      Tested against  ./mtr rpl_parallel_conf_limits.
      
      Warning, ./mtr rpl_parallel_conf_limits rpl_parallel_conf_limits ...
      can fail at the 2nd etc test because of no removal of Worker tables happens at RESET SLAVE.
     @ sql/log_event.cc
        adding a special to mts-recovery branch to the event scheduling routine located
        in Log_event::apply_event().
        todo: think about rli->flush_info() at the end of gap-filling.
     @ sql/rpl_rli.cc
        to be recovered group counter and a running index on the recovery bitmap are
        init-ed, also renaming.
        In recovery phase Coordinator can execute rows-events now.
     @ sql/rpl_rli.h
        to be recovered group counter and a running index on the recovery bitmap is added.
     @ sql/rpl_slave.cc
        engaging to be recovered group counter in mts_recovery_groups() in the end of which
        the recovery bitmap is ready and rli->mts_recovery_group_cnt counted how many bits of 
        interest in there.
        
        No actual recovery case is followed by
        rli->recovery_parallel_workers= rli->slave_parallel_workers
        at Workers startup time.

    modified:
      sql/log_event.cc
      sql/rpl_rli.cc
      sql/rpl_rli.h
      sql/rpl_slave.cc
 3256 Andrei Elkin	2010-12-21
      wl#5569 MTS
      
       slave_worker_info def is updated in the system db.

    modified:
      scripts/mysql_system_tables.sql
=== modified file 'sql/log_event.cc'
--- a/sql/log_event.cc	2010-12-20 22:18:33 +0000
+++ b/sql/log_event.cc	2010-12-22 11:57:18 +0000
@@ -2767,6 +2767,7 @@ int Log_event::apply_event(Relay_log_inf
   bool term_event;
 
   if (!(parallel= rli->is_parallel_exec()) ||
+      rli->mts_recovery_group_cnt != 0 ||
       ((seq_event=
         only_sequential_exec(rli->run_query_in_parallel,
                              rli->curr_group_seen_begin /* todo: obs 2nd arg */))
@@ -2784,6 +2785,23 @@ int Log_event::apply_event(Relay_log_inf
       if (!parallel_exec_by_coordinator(::server_id))
         (void) wait_for_workers_to_finish(rli);
     }
+    else if (rli->is_mts_recovery())
+    {
+      // recovery
+      bool skip= _bitmap_is_set(&c_rli->recovery_groups, c_rli->mts_recovery_index);
+
+      if (ends_group()) // todo: || rli->run_query_in_parallel && ! seen_begin
+      {
+        c_rli->mts_recovery_index++;
+        if (--c_rli->mts_recovery_group_cnt == 0)
+        {
+          c_rli->recovery_parallel_workers= c_rli->slave_parallel_workers;
+          c_rli->mts_recovery_index= 0;
+        }
+      }
+      if (skip)
+        DBUG_RETURN(0);
+    }
     DBUG_RETURN(do_apply_event(rli));
   }
 

=== modified file 'sql/rpl_rli.cc'
--- a/sql/rpl_rli.cc	2010-12-21 14:11:05 +0000
+++ b/sql/rpl_rli.cc	2010-12-22 11:57:18 +0000
@@ -71,7 +71,8 @@ Relay_log_info::Relay_log_info(bool is_s
    rows_query_ev(NULL), last_event_start_time(0),
    this_worker(NULL), slave_parallel_workers(0),
    recovery_parallel_workers(0),
-   checkpoint_group(mts_checkpoint_group),
+   checkpoint_group(mts_checkpoint_group), mts_recovery_group_cnt(0),
+   mts_recovery_index(0),
    sql_delay(0), sql_delay_end(0), m_flags(0)
 {
   DBUG_ENTER("Relay_log_info::Relay_log_info");
@@ -82,7 +83,7 @@ Relay_log_info::Relay_log_info(bool is_s
 
   set_timespec_nsec(last_clock, 0);
 
-  bitmap_init(&groups, NULL, checkpoint_group, FALSE);
+  bitmap_init(&recovery_groups, NULL, checkpoint_group, FALSE);
   bzero((char*) &cache_buf, sizeof(cache_buf));
   cached_charset_invalidate();
   mysql_mutex_init(key_relay_log_info_log_space_lock,
@@ -148,7 +149,7 @@ Relay_log_info::~Relay_log_info()
 {
   DBUG_ENTER("Relay_log_info::~Relay_log_info");
 
-  bitmap_free(&groups);
+  bitmap_free(&recovery_groups);
   mysql_mutex_destroy(&log_space_lock);
   mysql_cond_destroy(&log_space_cond);
   relay_log.cleanup();
@@ -1088,7 +1089,7 @@ void Relay_log_info::cleanup_context(THD
 {
   DBUG_ENTER("Relay_log_info::cleanup_context");
 
-  DBUG_ASSERT((info_thd == thd) || is_parallel_exec());
+  DBUG_ASSERT((info_thd == thd) || is_parallel_exec() || is_mts_recovery());
   /*
     1) Instances of Table_map_log_event, if ::do_apply_event() was called on them,
     may have opened tables, which we cannot be sure have been closed (because
@@ -1261,8 +1262,8 @@ int Relay_log_info::init_info()
     if (hot_log)
       mysql_mutex_unlock(log_lock);
 
-    bitmap_clear_all(&groups);
-    DBUG_RETURN(recovery_parallel_workers ? mts_recovery_groups(this, &groups) : 0);
+    DBUG_RETURN(recovery_parallel_workers ?
+                mts_recovery_groups(this, &recovery_groups) : 0);
   }
 
   cur_log_fd = -1;

=== modified file 'sql/rpl_rli.h'
--- a/sql/rpl_rli.h	2010-12-21 14:11:05 +0000
+++ b/sql/rpl_rli.h	2010-12-22 11:57:18 +0000
@@ -471,8 +471,9 @@ public:
   DYNAMIC_ARRAY least_occupied_workers;
   uint checkpoint_seqno;  // counter of groups executed after the most recent CP
   uint checkpoint_group;  // counter of groups after which a checkpoint is called.
-  MY_BITMAP groups;  // bitmap used during recovery.
-
+  MY_BITMAP recovery_groups;  // bitmap used during recovery.
+  ulong mts_recovery_group_cnt; // number of groups to execute at recovery
+  ulong mts_recovery_index;     // running index of recoverable groups
   /* most of allocation in the coordinator rli is there */
   void init_workers(ulong);
 
@@ -480,11 +481,20 @@ public:
   void deinit_workers();
 
   /**
+     returns true if there is any gap-group of events to execute
+                  at slave starting phase.
+  */
+  inline bool is_mts_recovery() const
+  {
+    return mts_recovery_group_cnt != 0;
+  }
+
+  /**
      returns true if events are to be executed in parallel
   */
   inline bool is_parallel_exec() const
   {
-    bool ret= slave_parallel_workers > 0;
+    bool ret= (slave_parallel_workers > 0) && !is_mts_recovery();
 
     DBUG_ASSERT(!ret || workers.elements > 0);
 

=== modified file 'sql/rpl_slave.cc'
--- a/sql/rpl_slave.cc	2010-12-21 19:34:58 +0000
+++ b/sql/rpl_slave.cc	2010-12-22 11:57:18 +0000
@@ -2810,9 +2810,6 @@ int apply_event_and_update_pos(Log_event
   else
     mysql_mutex_unlock(&rli->data_lock);
 
-  // TODO -- ALFRANIO
-  rli->recovery_parallel_workers= rli->slave_parallel_workers;
-
   DBUG_PRINT("info", ("apply_event error = %d", exec_res));
   if (exec_res == 0)
   {
@@ -3898,13 +3895,13 @@ bool mts_recovery_groups(Relay_log_info 
     sql_print_error("%s", errmsg);
     goto err;
   }
+             
   DBUG_ASSERT(my_stat(log_name, &s, MYF(0))); // TODO: Alfranio, why my_stat?
-
   my_b_seek(&log, (my_off_t) rli->get_group_relay_log_pos());
 
   bitmap_clear_all(groups);
-
-  for (uint it_job= 0, group_worker_counter= 0; it_job < above_lwm_jobs.elements; it_job++)
+  rli->mts_recovery_group_cnt= 0;
+  for (uint it_job= 0; it_job < above_lwm_jobs.elements; it_job++)
   {
     Slave_worker *w= ((Slave_job_group *)
                       dynamic_array_ptr(&above_lwm_jobs, it_job))->worker;
@@ -3925,7 +3922,7 @@ bool mts_recovery_groups(Relay_log_info 
                                           opt_master_verify_checksum)))
     {
         DBUG_ASSERT(ev->is_valid());
-        DBUG_ASSERT(group_worker_counter < rli->checkpoint_group);
+        DBUG_ASSERT(rli->mts_recovery_group_cnt < rli->checkpoint_group);
 
         // TODO: relax condition to allow --mts_exp_run_query_in_parallel= 1
         if (ev->starts_group())
@@ -3941,14 +3938,14 @@ bool mts_recovery_groups(Relay_log_info 
               // hit it
               // w.B << group_cnt++;
               // RB |= w.B;
-              for (uint i= w->checkpoint_seqno - group_worker_counter, j= 0;
+              for (uint i= w->checkpoint_seqno - rli->mts_recovery_group_cnt, j= 0;
                    i <= w->checkpoint_seqno; i++, j++)
               {
                 //bitmap_intersect(&rli->groups, &w->group_execed);
                 if (_bitmap_is_set(&w->group_execed, i))
                   bitmap_fast_test_and_set(groups, j);
               }
-              group_worker_counter++;
+              rli->mts_recovery_group_cnt++;
               delete ev;
               ev= NULL;
               break;
@@ -3956,7 +3953,7 @@ bool mts_recovery_groups(Relay_log_info 
             else
             {
               DBUG_ASSERT(ret < 0);
-              group_worker_counter++;
+              rli->mts_recovery_group_cnt++;
             }
           }
         delete ev;
@@ -3964,6 +3961,8 @@ bool mts_recovery_groups(Relay_log_info 
     }
   }
 
+  DBUG_ASSERT(rli->mts_recovery_group_cnt < groups->n_bits);
+
 #if 0
   for (uint it_job= 0; it_job < above_lwm_jobs.elements; it_job++)
   {
@@ -4329,6 +4328,12 @@ int slave_start_workers(Relay_log_info *
   rli->run_query_in_parallel= opt_mts_slave_run_query_in_parallel;
   rli->checkpoint_seqno= 0;
   //rli->worker_bitmap_buf= my_malloc(n/8 + 1,MYF(MY_WME));
+
+  // end recovery right now if mts_recovery_groups() did not find any gaps
+  if (rli->mts_recovery_group_cnt == 0)
+    rli->recovery_parallel_workers= rli->slave_parallel_workers;
+
+
   for (i= 0; i < n; i++)
   {
     if ((error= slave_start_single_worker(rli, i)))


Attachment: [text/bzr-bundle] bzr/[e-mail]