bk commit into 5.1 tree (jonas:1.2876) BUG#34596

[jonas]

Below is the list of changes that have just been committed into a local
5.1 repository of jonas.  When jonas does a push these changes
will be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet@stripped, 2008-02-17 11:37:12+01:00, jonas@stripped +1 -0
  ndb - bug#34596
    1) move updating of nextLcp to sendLCP_FRAG_ORD
    2) change impl. of findStartGci to check all LCPs not
       only take first that looks possible

  storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp@stripped, 2008-02-17 11:37:10+01:00,
jonas@stripped +43 -22
    ndb - bug#34596
      1) move updating of nextLcp to sendLCP_FRAG_ORD
      2) change impl. of findStartGci to check all LCPs not
         only take first that looks possible

diff -Nrup a/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
b/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
--- a/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp	2008-01-24 12:30:24 +01:00
+++ b/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp	2008-02-17 11:37:10 +01:00
@@ -11321,6 +11321,15 @@ Dbdih::sendLCP_FRAG_ORD(Signal* signal, 
     return;
   }
   
+  if (replicaPtr.p->nextLcp >= MAX_LCP_USED)
+  {
+    jam();
+    infoEvent("Updating nextLcp from %u to %u tab: %u", 
+              replicaPtr.p->nextLcp, 0,
+              info.tableId);
+    replicaPtr.p->nextLcp = 0;
+  }
+  
   LcpFragOrd * const lcpFragOrd = (LcpFragOrd *)&signal->theData[0];
   lcpFragOrd->tableId    = info.tableId;
   lcpFragOrd->fragmentId = info.fragId;
@@ -12557,23 +12566,42 @@ bool Dbdih::findStartGci(ConstPtr<Replic
                          Uint32& startGci,
                          Uint32& lcpNo) 
 {
-  lcpNo = replicaPtr.p->nextLcp;
-  const Uint32 startLcpNo = lcpNo;
-  do {
-    lcpNo = oldPrevLcpNo(lcpNo);
-    ndbrequire(lcpNo < MAX_LCP_STORED);
-    if (replicaPtr.p->lcpStatus[lcpNo] == ZVALID) {
-      if (replicaPtr.p->maxGciStarted[lcpNo] < stopGci) {
+  Uint32 cnt = 0;
+  Uint32 tmp[MAX_LCP_STORED];
+  for (Uint32 i = 0; i<MAX_LCP_STORED; i++)
+  {
+    jam();
+    if (replicaPtr.p->lcpStatus[i] == ZVALID &&
+        replicaPtr.p->maxGciStarted[i] < stopGci)
+    {
+      jam();
+      tmp[cnt] = i;
+      cnt++;
+    }
+  }
+  
+  if (cnt)
+  {
+    jam();
+    /**
+     * We found atleast one...get the highest
+     */
+    lcpNo = tmp[0];
+    Uint32 lcpId = replicaPtr.p->lcpId[lcpNo];
+    for (Uint32 i = 1; i<cnt; i++)
+    {
+      jam();
+      if (replicaPtr.p->lcpId[tmp[i]] > lcpId)
+      {
         jam();
-	/* ----------------------------------------------------------------- */
-	/*   WE HAVE FOUND A USEFUL LOCAL CHECKPOINT THAT CAN BE USED FOR    */
-	/*   RESTARTING THIS FRAGMENT REPLICA.                               */
-	/* ----------------------------------------------------------------- */
-        startGci = replicaPtr.p->maxGciCompleted[lcpNo] + 1;
-        return true;
-      } 
+        lcpNo = tmp[i];
+        lcpId = replicaPtr.p->lcpId[lcpNo];
+      }
     }
-  } while (lcpNo != startLcpNo);
+    startGci = replicaPtr.p->maxGciCompleted[lcpNo] + 1;
+    return true;
+  }
+
   /* --------------------------------------------------------------------- */
   /*       NO VALID LOCAL CHECKPOINT WAS AVAILABLE. WE WILL ADD THE        */
   /*       FRAGMENT. THUS THE NEXT LCP MUST BE SET TO ZERO.                */
@@ -13588,13 +13616,6 @@ void Dbdih::readReplica(RWFragment* rf, 
     jam();
     readReplicaPtr.p->lcpStatus[trraLcp] = ZINVALID;
   }//if
-
-  if (readReplicaPtr.p->nextLcp >= MAX_LCP_USED)
-  {
-    jam();
-    infoEvent("Updating nextLcp from %u to %u", readReplicaPtr.p->nextLcp, 0);
-    readReplicaPtr.p->nextLcp = 0;
-  }
 
   /* ---------------------------------------------------------------------- */
   /*       WE ALSO HAVE TO INVALIDATE ANY LOCAL CHECKPOINTS THAT HAVE BEEN  */