ndb - bug#29044

  Improve buddy high order allocation
  Make removeCommonArea O(1) instead of O(N)
  Add limit to left/right search


storage/ndb/src/kernel/blocks/dbtup/DbtupDebug.cpp:
  Add info to buddy module test about 
  1) loops being made in buddy
  2) how much was allocated
storage/ndb/src/kernel/blocks/dbtup/DbtupPagMan.cpp:
  1) make removeCommonArea o(1) - as list is (after fix) double linked anyway
  2) set page_state = ZFREE_COMMON insertCommonArea and ~ZFREE_COMMON in removeCommonArea
  3) add max loops in search left/right
  4) add more debug info

storage/ndb/src/kernel/blocks/dbtup/DbtupDebug.cpp

@@ -76,6 +76,10 @@ Dbtup::reportMemoryUsage(Signal* signal, int incDec){
   sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 6, JBB);
 }
 
+#ifdef VM_TRACE
+extern Uint32 fc_left, fc_right, fc_remove;
+#endif
+
 void
 Dbtup::execDUMP_STATE_ORD(Signal* signal)
 {
@@ -157,12 +161,20 @@ Dbtup::execDUMP_STATE_ORD(Signal* signal)
     return;
   }//if
 #endif
-#if defined VM_TRACE && 0
-  if (type == 1211){
-    ndbout_c("Startar modul test av Page Manager");
+#if defined VM_TRACE
+  if (type == 1211 || type == 1212 || type == 1213){
+    Uint32 seed = time(0);
+    if (signal->getLength() > 1)
+      seed = signal->theData[1];
+    ndbout_c("Startar modul test av Page Manager (seed: 0x%x)", seed);
+    srand(seed);
 
     Vector<Chunk> chunks;
     const Uint32 LOOPS = 1000;
+    Uint32 sum_req = 0;
+    Uint32 sum_conf = 0;
+    Uint32 sum_loop = 0;
+    Uint32 max_loop = 0;
     for(Uint32 i = 0; i<LOOPS; i++){
 
       // Case
@@ -180,7 +192,14 @@ Dbtup::execDUMP_STATE_ORD(Signal* signal)
 	c = 1 + rand() % 2;
       }
       
+      if (type == 1211)
         ndbout_c("loop=%d case=%d free=%d alloc=%d", i, c, free, alloc);
+
+      if (type == 1213)
+      {
+        c = 1;
+        alloc = 2 + (sum_conf >> 3) + (sum_conf >> 4);
+      }
       switch(c){ 
       case 0:{ // Release
 	const int ch = rand() % chunks.size();
@@ -192,15 +211,19 @@ Dbtup::execDUMP_STATE_ORD(Signal* signal)
       case 2: { // Seize(n) - fail
 	alloc += free;
 	// Fall through
+        sum_req += free;
+        goto doalloc;
       }
       case 1: { // Seize(n) (success)
-
+        sum_req += alloc;
+    doalloc:
 	Chunk chunk;
 	allocConsPages(alloc, chunk.pageCount, chunk.pageId);
 	ndbrequire(chunk.pageCount <= alloc);
 	if(chunk.pageCount != 0){
 	  chunks.push_back(chunk);
 	  if(chunk.pageCount != alloc) {
+	    if (type == 1211)
               ndbout_c("  Tried to allocate %d - only allocated %d - free: %d",
                        alloc, chunk.pageCount, free);
 	  }
@@ -209,6 +232,12 @@ Dbtup::execDUMP_STATE_ORD(Signal* signal)
 		   alloc, free);
 	}
 	
+        sum_conf += chunk.pageCount;
+        Uint32 tot = fc_left + fc_right + fc_remove;
+        sum_loop += tot;
+        if (tot > max_loop)
+          max_loop = tot;
+
 	for(Uint32 i = 0; i<chunk.pageCount; i++){
 	  PagePtr pagePtr;
 	  pagePtr.i = chunk.pageId + i;
@@ -227,6 +256,10 @@ Dbtup::execDUMP_STATE_ORD(Signal* signal)
       returnCommonArea(chunk.pageId, chunk.pageCount);      
       chunks.erase(chunks.size() - 1);
     }
+
+    ndbout_c("Got %u%% of requested allocs, loops : %u 100*avg: %u max: %u",
+             (100 * sum_conf) / sum_req, sum_loop, 100*sum_loop / LOOPS,
+             max_loop);
   }
 #endif
 }//Dbtup::execDUMP_STATE_ORD()

storage/ndb/src/kernel/blocks/dbtup/DbtupPagMan.cpp

@@ -148,10 +148,17 @@ void Dbtup::initializePage()
   cnoOfAllocatedPages = tmp; // Is updated by returnCommonArea
 }//Dbtup::initializePage()
 
+#ifdef VM_TRACE
+Uint32 fc_left, fc_right, fc_remove;
+#endif
+
 void Dbtup::allocConsPages(Uint32 noOfPagesToAllocate,
                            Uint32& noOfPagesAllocated,
                            Uint32& allocPageRef)
 {
+#ifdef VM_TRACE
+  fc_left = fc_right = fc_remove = 0;
+#endif
   if (noOfPagesToAllocate == 0){ 
     ljam();
     noOfPagesAllocated = 0;
@@ -230,7 +237,10 @@ void Dbtup::findFreeLeftNeighbours(Uint32& allocPageRef,
 {
   PagePtr pageFirstPtr, pageLastPtr;
   Uint32 remainAllocate = noOfPagesToAllocate - noPagesAllocated;
-  while (allocPageRef > 0) {
+  Uint32 loop = 0;
+  while (allocPageRef > 0 && 
+         ++loop < 16) 
+  {
     ljam();
     pageLastPtr.i = allocPageRef - 1;
     c_page_pool.getPtr(pageLastPtr);
@@ -258,6 +268,9 @@ void Dbtup::findFreeLeftNeighbours(Uint32& allocPageRef,
         remainAllocate -= listSize;
       }//if
     }//if
+#ifdef VM_TRACE
+    fc_left++;
+#endif
   }//while
 }//Dbtup::findFreeLeftNeighbours()
 
@@ -271,7 +284,10 @@ void Dbtup::findFreeRightNeighbours(Uint32& allocPageRef,
     ljam();
     return;
   }//if
-  while ((allocPageRef + noPagesAllocated) < c_page_pool.getSize()) {
+  Uint32 loop = 0;
+  while ((allocPageRef + noPagesAllocated) < c_page_pool.getSize() &&
+         ++loop < 16) 
+  {
     ljam();
     pageFirstPtr.i = allocPageRef + noPagesAllocated;
     c_page_pool.getPtr(pageFirstPtr);
@@ -298,24 +314,37 @@ void Dbtup::findFreeRightNeighbours(Uint32& allocPageRef,
         remainAllocate -= listSize;
       }//if
     }//if
+#ifdef VM_TRACE
+    fc_right++;
+#endif
   }//while
 }//Dbtup::findFreeRightNeighbours()
 
 void Dbtup::insertCommonArea(Uint32 insPageRef, Uint32 insList) 
 {
   cnoOfAllocatedPages -= (1 << insList);
-  PagePtr pageLastPtr, pageInsPtr;
+  PagePtr pageLastPtr, pageInsPtr, pageHeadPtr;
 
+  pageHeadPtr.i = cfreepageList[insList];
   c_page_pool.getPtr(pageInsPtr, insPageRef);
   ndbrequire(insList < 16);
   pageLastPtr.i = (pageInsPtr.i + (1 << insList)) - 1;
 
-  pageInsPtr.p->next_cluster_page = cfreepageList[insList];
+  pageInsPtr.p->page_state = ZFREE_COMMON;
+  pageInsPtr.p->next_cluster_page = pageHeadPtr.i;
   pageInsPtr.p->prev_cluster_page = RNIL;
   pageInsPtr.p->last_cluster_page = pageLastPtr.i;
   cfreepageList[insList] = pageInsPtr.i;
 
+  if (pageHeadPtr.i != RNIL)
+  {
+    jam();
+    c_page_pool.getPtr(pageHeadPtr);
+    pageHeadPtr.p->prev_cluster_page = pageInsPtr.i;
+  }
+  
   c_page_pool.getPtr(pageLastPtr);
+  pageLastPtr.p->page_state = ZFREE_COMMON;
   pageLastPtr.p->first_cluster_page = pageInsPtr.i;
   pageLastPtr.p->next_page = RNIL;
 }//Dbtup::insertCommonArea()
@@ -323,12 +352,13 @@ void Dbtup::insertCommonArea(Uint32 insPageRef, Uint32 insList)
 void Dbtup::removeCommonArea(Uint32 remPageRef, Uint32 list) 
 {
   cnoOfAllocatedPages += (1 << list);  
-  PagePtr pagePrevPtr, pageNextPtr, pageLastPtr, pageSearchPtr, remPagePtr;
+  PagePtr pagePrevPtr, pageNextPtr, pageLastPtr, remPagePtr;
 
   c_page_pool.getPtr(remPagePtr, remPageRef);
   ndbrequire(list < 16);
   if (cfreepageList[list] == remPagePtr.i) {
     ljam();
+    ndbassert(remPagePtr.p->prev_cluster_page == RNIL);
     cfreepageList[list] = remPagePtr.p->next_cluster_page;
     pageNextPtr.i = cfreepageList[list];
     if (pageNextPtr.i != RNIL) {
@@ -337,30 +367,25 @@ void Dbtup::removeCommonArea(Uint32 remPageRef, Uint32 list)
       pageNextPtr.p->prev_cluster_page = RNIL;
     }//if
   } else {
-    pageSearchPtr.i = cfreepageList[list];
-    while (true) {
-      ljam();
-      c_page_pool.getPtr(pageSearchPtr);
-      pagePrevPtr = pageSearchPtr;
-      pageSearchPtr.i = pageSearchPtr.p->next_cluster_page;
-      if (pageSearchPtr.i == remPagePtr.i) {
-        ljam();
-        break;
-      }//if
-    }//while
+    pagePrevPtr.i = remPagePtr.p->prev_cluster_page;
     pageNextPtr.i = remPagePtr.p->next_cluster_page;
+    c_page_pool.getPtr(pagePrevPtr);
     pagePrevPtr.p->next_cluster_page = pageNextPtr.i;
-    if (pageNextPtr.i != RNIL) {
+    if (pageNextPtr.i != RNIL)
+    {
       ljam();
       c_page_pool.getPtr(pageNextPtr);
       pageNextPtr.p->prev_cluster_page = pagePrevPtr.i;
-    }//if
+    }
   }//if
   remPagePtr.p->next_cluster_page= RNIL;
   remPagePtr.p->last_cluster_page= RNIL;
   remPagePtr.p->prev_cluster_page= RNIL;
+  remPagePtr.p->page_state = ~ZFREE_COMMON;
 
   pageLastPtr.i = (remPagePtr.i + (1 << list)) - 1;
   c_page_pool.getPtr(pageLastPtr);
   pageLastPtr.p->first_cluster_page= RNIL;
+  pageLastPtr.p->page_state = ~ZFREE_COMMON;
+
 }//Dbtup::removeCommonArea()