bzr commit into mysql-5.1-telco-6.3 branch (jonas:3130)

[Jonas Oreland]

#At file:///home/jonas/src/telco-6.3/ based on revid:jonas@stripped

 3130 Jonas Oreland	2009-10-23
      ndb - fix bug with multi-page caching, where pages could be linked incorrectly, with ndbrequire as a consequence

    modified:
      storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
      storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
=== modified file 'storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp'
--- a/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp	2009-10-20 20:13:47 +0000
+++ b/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp	2009-10-23 11:37:58 +0000
@@ -2995,6 +2995,7 @@ private:
   } m_redo_page_cache;
 
   void evict(RedoPageCache&, Uint32 cnt);
+  void do_evict(RedoPageCache&, Ptr<RedoCacheLogPageRecord>);
   void addCachePages(RedoPageCache&,
                      Uint32 partNo,
                      Uint32 startPageNo,

=== modified file 'storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp'
--- a/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp	2009-10-20 20:13:47 +0000
+++ b/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp	2009-10-23 11:37:58 +0000
@@ -17502,6 +17502,7 @@ void Dblqh::findPageRef(Signal* signal, 
     for (Uint32 i = 0; i<extra; i++)
     {
       jam();
+      Uint32 prevPtrI = loopPtr.i;
       loopPtr.i = loopPtr.p->logPageWord[ZNEXT_PAGE];
       if (loopPtr.i == RNIL)
       {
@@ -17520,7 +17521,7 @@ void Dblqh::findPageRef(Signal* signal, 
           Ptr<LogPageRecord> tmp;
           tmp.i = pagePtr.i;
           tmp.p = reinterpret_cast<LogPageRecord*>(pagePtr.p);
-          tmp.p->logPageWord[ZPREV_PAGE] = loopPtr.i;
+          tmp.p->logPageWord[ZPREV_PAGE] = prevPtrI;
           loopPtr.p->logPageWord[ZNEXT_PAGE] = tmp.i;
           loopPtr.i = tmp.i;
         }
@@ -20713,58 +20714,63 @@ Dblqh::get_node_status(Uint32 nodeId) co
 #define MIN_REDO_PAGES_FREE 64
 
 void
-Dblqh::evict(RedoPageCache& cache, Uint32 cnt)
+Dblqh::do_evict(RedoPageCache& cache, Ptr<RedoCacheLogPageRecord> pagePtr)
 {
   LogPageRecordPtr save = logPagePtr;
-  while (cnoOfLogPages < (cnt + MIN_REDO_PAGES_FREE) && !cache.m_lru.isEmpty())
-  {
-    jam();
-    Ptr<RedoCacheLogPageRecord> pagePtr;
-    cache.m_lru.last(pagePtr);
-    cache.m_lru.remove(pagePtr);
-    cache.m_hash.remove(pagePtr);
-    if (0)
-    ndbout_c("evict part: %u file: %u page: %u cnoOfLogPages: %u",
-             pagePtr.p->m_part_no,
-             pagePtr.p->m_file_no,
-             pagePtr.p->m_page_no,
-             cnoOfLogPages);
+  cache.m_lru.remove(pagePtr);
+  cache.m_hash.remove(pagePtr);
+  if (0)
+  ndbout_c("evict part: %u file: %u page: %u cnoOfLogPages: %u",
+           pagePtr.p->m_part_no,
+           pagePtr.p->m_file_no,
+           pagePtr.p->m_page_no,
+           cnoOfLogPages);
 
-    logPagePtr.i = pagePtr.i;
-    ptrCheckGuard(logPagePtr, clogPageFileSize, logPageRecord);
+  logPagePtr.i = pagePtr.i;
+  ptrCheckGuard(logPagePtr, clogPageFileSize, logPageRecord);
 
-    Ptr<LogPageRecord> prevPagePtr, nextPagePtr;
-    prevPagePtr.i = logPagePtr.p->logPageWord[ZPREV_PAGE];
-    nextPagePtr.i = logPagePtr.p->logPageWord[ZNEXT_PAGE];
-    if (prevPagePtr.i != RNIL)
-    {
-      jam();
-      /**
-       * Remove ZNEXT pointer from prevPagePtr
-       *   so we don't try to "serve" multi-page request
-       *   if next-page has been evicted
-       */
-      ptrCheckGuard(prevPagePtr, clogPageFileSize, logPageRecord);
-      ndbrequire(prevPagePtr.p->logPageWord[ZNEXT_PAGE] == logPagePtr.i);
-      prevPagePtr.p->logPageWord[ZNEXT_PAGE] = RNIL;
+  Ptr<LogPageRecord> prevPagePtr, nextPagePtr;
+  prevPagePtr.i = logPagePtr.p->logPageWord[ZPREV_PAGE];
+  nextPagePtr.i = logPagePtr.p->logPageWord[ZNEXT_PAGE];
+  if (prevPagePtr.i != RNIL)
+  {
+    jam();
+    /**
+     * Remove ZNEXT pointer from prevPagePtr
+     *   so we don't try to "serve" multi-page request
+     *   if next-page has been evicted
+     */
+    ptrCheckGuard(prevPagePtr, clogPageFileSize, logPageRecord);
+    ndbrequire(prevPagePtr.p->logPageWord[ZNEXT_PAGE] == logPagePtr.i);
+    prevPagePtr.p->logPageWord[ZNEXT_PAGE] = RNIL;
+  }
 
-    }
+  if (nextPagePtr.i != RNIL)
+  {
+    jam();
+    /**
+     * Remove ZPREV pointer from nextPagePtr
+     *   so don't try to do above if prev has been evicted
+     */
+    ptrCheckGuard(nextPagePtr, clogPageFileSize, logPageRecord);
+    ndbrequire(nextPagePtr.p->logPageWord[ZPREV_PAGE] == logPagePtr.i);
+    nextPagePtr.p->logPageWord[ZPREV_PAGE] = RNIL;
+  }
 
-    if (nextPagePtr.i != RNIL)
-    {
-      jam();
-      /**
-       * Remove ZPREV pointer from nextPagePtr
-       *   so don't try to do above if prev has been evicted
-       */
-      ptrCheckGuard(nextPagePtr, clogPageFileSize, logPageRecord);
-      ndbrequire(nextPagePtr.p->logPageWord[ZPREV_PAGE] == logPagePtr.i);
-      nextPagePtr.p->logPageWord[ZPREV_PAGE] = RNIL;
-    }
+  releaseLogpage(0);
+  logPagePtr = save;
+}
 
-    releaseLogpage(0);
+void
+Dblqh::evict(RedoPageCache& cache, Uint32 cnt)
+{
+  while (cnoOfLogPages < (cnt + MIN_REDO_PAGES_FREE) && !cache.m_lru.isEmpty())
+  {
+    jam();
+    Ptr<RedoCacheLogPageRecord> pagePtr;
+    cache.m_lru.last(pagePtr);
+    do_evict(cache, pagePtr);
   }
-  logPagePtr = save;
 }
 
 void
@@ -20804,12 +20810,11 @@ Dblqh::addCachePages(RedoPageCache& cach
       key.m_page_no = startPageNo + i;
       key.m_file_no = filePtr.p->fileNo;
       Ptr<RedoCacheLogPageRecord> tmp;
-      cache.m_hash.remove(tmp, key);
-      if (!tmp.isNull())
+      if (cache.m_hash.find(tmp, key))
       {
         jam();
-        cache.m_lru.remove(tmp);
         found = true;
+        do_evict(cache, tmp);
       }
     }
 


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