bzr push into mysql-next-mr-wl5569 branch (andrei.elkin:3251) WL#5569

[Andrei Elkin]

 3251 Andrei Elkin	2010-12-20 [merge]
      wl#5569 MTS
      
      manual merging from the repo and correcting GAQ processing with introducing a volatile byte to indicate whether an item is busy or released.

    added:
      mysql-test/suite/sys_vars/r/mts_checkpoint_group_basic.result
      mysql-test/suite/sys_vars/t/mts_checkpoint_group_basic.test
    modified:
      mysql-test/extra/rpl_tests/rpl_parallel_load.test
      mysql-test/r/mysqld--help-notwin.result
      mysql-test/r/mysqld--help-win.result
      mysql-test/suite/funcs_1/r/is_columns_mysql.result
      mysql-test/suite/rpl/r/rpl_parallel_conf_limits.result
      mysql-test/suite/rpl/t/disabled.def
      mysql-test/suite/sys_vars/r/all_vars.result
      scripts/mysql_system_tables.sql
      sql/log_event.cc
      sql/mysqld.cc
      sql/mysqld.h
      sql/rpl_info_factory.cc
      sql/rpl_info_factory.h
      sql/rpl_rli.cc
      sql/rpl_rli.h
      sql/rpl_rli_pdb.cc
      sql/rpl_rli_pdb.h
      sql/rpl_slave.cc
      sql/rpl_slave.h
      sql/sys_vars.cc
=== modified file 'mysql-test/extra/rpl_tests/rpl_parallel_load.test'
--- a/mysql-test/extra/rpl_tests/rpl_parallel_load.test	2010-12-17 12:46:15 +0000
+++ b/mysql-test/extra/rpl_tests/rpl_parallel_load.test	2010-12-20 15:58:58 +0000
@@ -6,7 +6,7 @@
 # load volume parameter
 #
 
-let $iter = 1000;
+let $iter = 500;
 
 # windows run on PB2 is too slow to time out
 disable_query_log;

=== modified file 'sql/log_event.cc'
--- a/sql/log_event.cc	2010-12-18 00:33:02 +0000
+++ b/sql/log_event.cc	2010-12-20 15:58:58 +0000
@@ -2429,9 +2429,12 @@ Slave_worker *Log_event::get_slave_worke
     g.checkpoint_log_name= NULL;
     g.checkpoint_log_pos= 0;
     g.checkpoint_seqno= (uint) -1;
+    g.done= 0;
 
     // the last occupied GAQ's array index
     gaq_idx= rli->gaq->assigned_group_index= rli->gaq->en_queue((void *) &g);
+    // serves as a mark for Coord to delete events otherwise
+    const_cast<Relay_log_info*>(rli)->curr_group_is_parallel= TRUE;
     
     DBUG_ASSERT(gaq_idx != (ulong) -1 && gaq_idx < rli->gaq->s);
     DBUG_ASSERT(((Slave_job_group *) 
@@ -2450,14 +2453,14 @@ Slave_worker *Log_event::get_slave_worke
 
       // mark the current grup as started with B-event
       const_cast<Relay_log_info*>(rli)->curr_group_seen_begin= TRUE;
+      return NULL;
     } 
     else 
     { 
       DBUG_ASSERT(!rli->curr_group_seen_begin);
+
+      // Parallel single event proceeds to partiting etc ...
     }
-    // serves as a mark for Coord to delete events otherwise
-    const_cast<Relay_log_info*>(rli)->curr_group_is_parallel= TRUE;
-    return NULL;
   }
 
   //else // g
@@ -2765,32 +2768,6 @@ int Log_event::apply_event(Relay_log_inf
       // This `only-sequential' case relates to a DDL Query case
 
       DBUG_ASSERT(rli->curr_group_da.elements == 0);
-#if 0      
-      if (rli->curr_group_da.elements > 0)
-      {
-        int res;
-        Log_event *ev_begin= * (Log_event**) pop_dynamic(&c_rli->curr_group_da);
-
-        DBUG_ASSERT(rli->curr_group_da.elements == 0);
-        DBUG_ASSERT(rli->curr_group_seen_begin);
-
-        // While Query-log-event is not supported GAQ needs rollback
-        if (rli->curr_group_seen_begin)
-        {
-          Slave_job_group g;
-          ulong ind= rli->gaq->de_tail((uchar *) &g);
-          const_cast<Relay_log_info*>(rli)->mts_total_groups--;
-
-          DBUG_ASSERT(rli->last_assigned_worker == NULL);
-          DBUG_ASSERT(c_rli->gaq->assigned_group_index == ind);
-        }
-        res= ev_begin->do_apply_event(rli);
-        delete ev_begin;
-        c_rli->curr_group_is_parallel= FALSE;   // Coord will destruct events
-        if (res)
-          DBUG_RETURN(res);
-      }
-#endif
       DBUG_ASSERT(!rli->curr_group_seen_begin);
       
       c_rli->curr_group_is_parallel= FALSE;   // Coord will destruct events

=== modified file 'sql/rpl_rli_pdb.cc'
--- a/sql/rpl_rli_pdb.cc	2010-12-18 00:33:02 +0000
+++ b/sql/rpl_rli_pdb.cc	2010-12-20 15:58:58 +0000
@@ -526,6 +526,7 @@ void Slave_worker::slave_worker_ends_gro
     // first ever group must have relay log name
     DBUG_ASSERT(last_group_done_index != c_rli->gaq->s ||
                 ptr_g->group_relay_log_name != NULL);
+    DBUG_ASSERT(ptr_g->worker_id == id);
 
     if (ptr_g->group_relay_log_name != NULL)
     {
@@ -544,8 +545,9 @@ void Slave_worker::slave_worker_ends_gro
 
     ptr_g->group_master_log_pos= group_master_log_pos;
     ptr_g->group_relay_log_pos= group_relay_log_pos;
+    ptr_g->done= 1;    // GAQ index is available to C now
 
-    last_group_done_index = gaq_idx;   // GAQ index is available to C now
+    last_group_done_index= gaq_idx;
   }
 
   // cleanup relating to the last executed group regardless of error
@@ -618,6 +620,7 @@ ulong circular_buffer_queue::de_queue(uc
   DBUG_ASSERT(e == s ||
               (len == (a >= e)? (a - e) :
                (s + a - e)));
+  DBUG_ASSERT(a != e);
 
   return ret;
 }
@@ -644,6 +647,7 @@ ulong circular_buffer_queue::de_tail(uch
   DBUG_ASSERT(e == s ||
               (len == (a >= e)? (a - e) :
                (s + a - e)));
+  DBUG_ASSERT(a != e);
 
   return a;
 }
@@ -675,9 +679,12 @@ ulong circular_buffer_queue::en_queue(vo
   // post-boundary cond
   if (a == e)
     a= s;
+
   DBUG_ASSERT(a == e || 
               len == (a >= e) ?
               (a - e) : (s + a - e));
+  DBUG_ASSERT(a != e);
+
   return ret;
 }
 
@@ -703,12 +710,13 @@ void* circular_buffer_queue::head_queue(
            range, incl cases the queue is empty or full.
 
    @return TRUE  if the first arg identifies a queue entity ordered
-                 before one defined by the 2nd arg,
+                 after one defined by the 2nd arg,
            FALSE otherwise.
 */
 bool circular_buffer_queue::gt(ulong i, ulong k)
 {
   DBUG_ASSERT(i < s && k < s);
+  DBUG_ASSERT(a != e);
 
   if (i >= e)
     if (k >= e)
@@ -759,8 +767,12 @@ ulong Slave_committed_queue::move_queue_
     if (ptr_g->worker_id == (ulong) -1)
       break; /* the head is not even assigned */
     get_dynamic(ws, (uchar *) &w_i, ptr_g->worker_id);
-    
-    if (w_i->last_group_done_index == s || gt(i, w_i->last_group_done_index))
+
+    // no stale last_group_done_index value
+    // DBUG_ASSERT(w_i->checkpoint_notified);
+
+    //if (in(w_i->last_group_done_index) && gt(i, w_i->last_group_done_index))
+    if (!ptr_g->done)
       break; /* gap at i'th */
 
     // memorize the last met group_relay_log_name

=== modified file 'sql/rpl_rli_pdb.h'
--- a/sql/rpl_rli_pdb.h	2010-12-18 00:33:02 +0000
+++ b/sql/rpl_rli_pdb.h	2010-12-20 15:58:58 +0000
@@ -98,6 +98,9 @@ public:
   */
   void* head_queue();
   bool   gt(ulong i, ulong k); // comparision of ordering of two entities
+  /* index is within the valid range */
+  bool in(ulong k) { return !empty() && 
+      (e > a ? (k >= e || k < a) : (k >= e && k < a)); }
   bool empty() { return e == s; }
   bool full() { return a == s; }
 };
@@ -127,6 +130,7 @@ typedef struct st_slave_job_group
   uint  checkpoint_seqno;
   my_off_t checkpoint_log_pos; // T-event lop_pos filled by W for CheckPoint
   char*    checkpoint_log_name;
+  volatile uchar done;  // Flag raised by W,  read and reset by C
 } Slave_job_group;
 
 #define get_job(from, to) \


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