Merge joreland@bk-internal.mysql.com:/home/bk/mysql-5.1-telco-gca

into  perch.ndb.mysql.com:/home/jonas/src/51-telco-gca


storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp:
  Auto merged
storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp:
  Auto merged
storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp:
  Auto merged
storage/ndb/test/run-test/daily-basic-tests.txt:
  Auto merged

13 files changed, 243 insertions, 33 deletions

diff --git a/storage/ndb/src/kernel/blocks/ERROR_codes.txt b/storage/ndb/src/kernel/blocks/ERROR_codes.txt
index b3405679978..67eb89f850f 100644
--- a/storage/ndb/src/kernel/blocks/ERROR_codes.txt
+++ b/storage/ndb/src/kernel/blocks/ERROR_codes.txt
@@ -1,5 +1,5 @@
 Next QMGR 1
-Next NDBCNTR 1001
+Next NDBCNTR 1002
 Next NDBFS 2000
 Next DBACC 3002
 Next DBTUP 4029
@@ -523,3 +523,4 @@ Dbtup:
 NDBCNTR:
 
 1000: Crash insertion on SystemError::CopyFragRef
+1001: Delay sending NODE_FAILREP (to own node), until error is cleared
diff --git a/storage/ndb/src/kernel/blocks/backup/Backup.cpp b/storage/ndb/src/kernel/blocks/backup/Backup.cpp
index a07617f0bfb..57082eaccc8 100644
--- a/storage/ndb/src/kernel/blocks/backup/Backup.cpp
+++ b/storage/ndb/src/kernel/blocks/backup/Backup.cpp
@@ -3543,10 +3543,10 @@ Backup::execBACKUP_FRAGMENT_REQ(Signal* signal)
     ScanFragReq::setHoldLockFlag(req->requestInfo, 0);
     ScanFragReq::setKeyinfoFlag(req->requestInfo, 0);
     ScanFragReq::setAttrLen(req->requestInfo,attrLen); 
+    ScanFragReq::setTupScanFlag(req->requestInfo, 1);
     if (ptr.p->is_lcp())
     {
       ScanFragReq::setScanPrio(req->requestInfo, 1);
-      ScanFragReq::setTupScanFlag(req->requestInfo, 1);
       ScanFragReq::setNoDiskFlag(req->requestInfo, 1);
       ScanFragReq::setLcpScanFlag(req->requestInfo, 1);
     }
diff --git a/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp b/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
index 76ff0dcc41b..a2f8351e308 100644
--- a/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
+++ b/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
@@ -4673,12 +4673,18 @@ void Dbdih::failedNodeLcpHandling(Signal* signal, NodeRecordPtr failedNodePtr)
   jam();
   const Uint32 nodeId = failedNodePtr.i;
 
-  if (c_lcpState.m_participatingLQH.get(failedNodePtr.i)){
+  if (isMaster() && c_lcpState.m_participatingLQH.get(failedNodePtr.i))
+  {
     /*----------------------------------------------------*/
     /*  THE NODE WAS INVOLVED IN A LOCAL CHECKPOINT. WE   */
     /* MUST UPDATE THE ACTIVE STATUS TO INDICATE THAT     */
     /* THE NODE HAVE MISSED A LOCAL CHECKPOINT.           */
     /*----------------------------------------------------*/
+
+    /**
+     * Bug#28717, Only master should do this, as this status is copied
+     *   to other nodes
+     */
     switch (failedNodePtr.p->activeStatus) {
     case Sysfile::NS_Active:
       jam();
diff --git a/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp b/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
index 3fc8891c082..0f88933f617 100644
--- a/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
+++ b/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
@@ -2678,7 +2678,8 @@ private:
   UintR cfirstfreeLogFile;
   UintR clogFileFileSize;
 
-#define ZLFO_FILE_SIZE 256            /* MAX 256 OUTSTANDING FILE OPERATIONS */
+#define ZLFO_MIN_FILE_SIZE 256
+// RedoBuffer/32K minimum ZLFO_MIN_FILE_SIZE
   LogFileOperationRecord *logFileOperationRecord;
   LogFileOperationRecordPtr lfoPtr;
   UintR cfirstfreeLfo;
@@ -2695,7 +2696,7 @@ private:
   UintR cfirstfreePageRef;
   UintR cpageRefFileSize;
 
-#define ZSCANREC_FILE_SIZE 100
+// Configurable
   ArrayPool<ScanRecord> c_scanRecordPool;
   ScanRecordPtr scanptr;
   UintR cscanNoFreeRec;
diff --git a/storage/ndb/src/kernel/blocks/dblqh/DblqhInit.cpp b/storage/ndb/src/kernel/blocks/dblqh/DblqhInit.cpp
index 8ddb96f9111..8aaf86de73a 100644
--- a/storage/ndb/src/kernel/blocks/dblqh/DblqhInit.cpp
+++ b/storage/ndb/src/kernel/blocks/dblqh/DblqhInit.cpp
@@ -30,11 +30,11 @@ void Dblqh::initData()
   cgcprecFileSize = ZGCPREC_FILE_SIZE;
   chostFileSize = MAX_NDB_NODES;
   clcpFileSize = ZNO_CONCURRENT_LCP;
-  clfoFileSize = ZLFO_FILE_SIZE;
+  clfoFileSize = 0;
   clogFileFileSize = 0;
   clogPartFileSize = ZLOG_PART_FILE_SIZE;
   cpageRefFileSize = ZPAGE_REF_FILE_SIZE;
-  cscanrecFileSize = ZSCANREC_FILE_SIZE;
+  cscanrecFileSize = 0;
   ctabrecFileSize = 0;
   ctcConnectrecFileSize = 0;
   ctcNodeFailrecFileSize = MAX_NDB_NODES;
diff --git a/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp b/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
index a4b1293ba55..33696ebba27 100644
--- a/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
+++ b/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
@@ -1002,6 +1002,11 @@ void Dblqh::execREAD_CONFIG_REQ(Signal* signal)
     clogPageFileSize+= (16 - mega_byte_part);
   }
 
+  /* maximum number of log file operations */
+  clfoFileSize = clogPageFileSize;
+  if (clfoFileSize < ZLFO_MIN_FILE_SIZE)
+    clfoFileSize = ZLFO_MIN_FILE_SIZE;
+
   ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_LQH_TABLE, &ctabrecFileSize));
   ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_LQH_TC_CONNECT, 
 					&ctcConnectrecFileSize));
@@ -8478,9 +8483,32 @@ void Dblqh::continueAfterReceivingAllAiLab(Signal* signal)
   AccScanReq::setLockMode(req->requestInfo, scanptr.p->scanLockMode);
   AccScanReq::setReadCommittedFlag(req->requestInfo, scanptr.p->readCommitted);
   AccScanReq::setDescendingFlag(req->requestInfo, scanptr.p->descending);
-  AccScanReq::setNoDiskScanFlag(req->requestInfo, 
-				!tcConnectptr.p->m_disk_table);
-  AccScanReq::setLcpScanFlag(req->requestInfo, scanptr.p->lcpScan);
+
+  if (refToBlock(tcConnectptr.p->clientBlockref) == BACKUP)
+  {
+    if (scanptr.p->lcpScan)
+    {
+      AccScanReq::setNoDiskScanFlag(req->requestInfo, 1);
+      AccScanReq::setLcpScanFlag(req->requestInfo, 1);
+    }
+    else
+    {
+      /* If backup scan disktables in disk order */
+      AccScanReq::setNoDiskScanFlag(req->requestInfo,
+                                    !tcConnectptr.p->m_disk_table);
+      AccScanReq::setLcpScanFlag(req->requestInfo, 0);
+    }
+  }
+  else
+  {
+#if BUG_27776_FIXED
+    AccScanReq::setNoDiskScanFlag(req->requestInfo,
+                                  !tcConnectptr.p->m_disk_table);
+#else
+    AccScanReq::setNoDiskScanFlag(req->requestInfo, 1);
+#endif
+    AccScanReq::setLcpScanFlag(req->requestInfo, 0);
+  }
   
   req->transId1 = tcConnectptr.p->transid[0];
   req->transId2 = tcConnectptr.p->transid[1];
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
index 54abbf18664..db336df6652 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
@@ -318,6 +318,7 @@ Dbtup::restart_setup_page(Disk_alloc_info& alloc, PagePtr pagePtr)
     unsigned uncommitted, committed;
     uncommitted = committed = ~(unsigned)0;
     int ret = tsman.get_page_free_bits(&page, &uncommitted, &committed);
+    jamEntry();
     
     idx = alloc.calc_page_free_bits(real_free);
     ddassert(idx == committed);
@@ -428,12 +429,12 @@ Dbtup::disk_page_prealloc(Signal* signal,
     c_extent_pool.getPtr(ext);
     if ((pageBits= tsman.alloc_page_from_extent(&ext.p->m_key, bits)) >= 0) 
     {
-      jam();
+      jamEntry();
       found= true;
     }
     else
     {
-      jam();
+      jamEntry();
       /**
        * The current extent is not in a free list
        *   and since it couldn't accomadate the request
@@ -490,7 +491,7 @@ Dbtup::disk_page_prealloc(Signal* signal,
       
       if ((err= tsman.alloc_extent(&ext.p->m_key)) < 0)
       {
-	jam();
+	jamEntry();
 #if NOT_YET_UNDO_ALLOC_EXTENT
 	c_lgman->free_log_space(logfile_group_id, 
 				sizeof(Disk_undo::AllocExtent)>>2);
@@ -542,6 +543,7 @@ Dbtup::disk_page_prealloc(Signal* signal,
     alloc.m_curr_extent_info_ptr_i= ext.i;
     ext.p->m_free_matrix_pos= RNIL;
     pageBits= tsman.alloc_page_from_extent(&ext.p->m_key, bits);
+    jamEntry();
     ddassert(pageBits >= 0);
   }
   
@@ -601,6 +603,7 @@ Dbtup::disk_page_prealloc(Signal* signal,
   }
   
   int res= m_pgman.get_page(signal, preq, flags);
+  jamEntry();
   switch(res)
   {
   case 0:
@@ -900,6 +903,7 @@ Dbtup::disk_page_set_dirty(PagePtr pagePtr)
   
   // Make sure no one will allocate it...
   tsman.unmap_page(&key, MAX_FREE_LIST - 1);
+  jamEntry();
 }
 
 void
@@ -951,6 +955,7 @@ Dbtup::disk_page_unmap_callback(Uint32 page_id, Uint32 dirty_count)
 			    fragPtr.p->m_tablespace_id);
     
     tsman.unmap_page(&key, idx);
+    jamEntry();
     pagePtr.p->list_index = idx | 0x8000;
   }
     
@@ -999,6 +1004,7 @@ Dbtup::disk_page_alloc(Signal* signal,
 			    fragPtrP->m_tablespace_id);
     
     tsman.update_page_free_bits(key, new_bits, lsn);
+    jamEntry();
   }
 }
 
@@ -1051,6 +1057,7 @@ Dbtup::disk_page_free(Signal *signal,
 			    fragPtrP->m_tablespace_id);
     
     tsman.update_page_free_bits(key, new_bits, lsn);
+    jamEntry();
   }
 
   Uint32 ext = pagePtr.p->m_extent_info_ptr;
@@ -1104,6 +1111,7 @@ Dbtup::disk_page_abort_prealloc(Signal *signal, Fragrecord* fragPtrP,
   memcpy(&req.m_page, key, sizeof(Local_key));
 
   int res= m_pgman.get_page(signal, req, flags);
+  jamEntry();
   switch(res)
   {
   case 0:
@@ -1232,6 +1240,7 @@ Dbtup::disk_page_alloc_extent_log_buffer_callback(Signal* signal,
   Uint64 lsn= lgman.add_entry(c, 1);
   
   tsman.update_lsn(&key, lsn);
+  jamEntry();
 }
 #endif
 
@@ -1250,6 +1259,7 @@ Dbtup::disk_page_undo_alloc(Page* page, const Local_key* key,
   
   Uint64 lsn= lgman.add_entry(c, 1);
   m_pgman.update_lsn(* key, lsn);
+  jamEntry();
 
   return lsn;
 }
@@ -1279,6 +1289,7 @@ Dbtup::disk_page_undo_update(Page* page, const Local_key* key,
     
   Uint64 lsn= lgman.add_entry(c, 3);
   m_pgman.update_lsn(* key, lsn);
+  jamEntry();
 
   return lsn;
 }
@@ -1308,6 +1319,7 @@ Dbtup::disk_page_undo_free(Page* page, const Local_key* key,
   
   Uint64 lsn= lgman.add_entry(c, 3);
   m_pgman.update_lsn(* key, lsn);
+  jamEntry();
 
   return lsn;
 }
@@ -1402,6 +1414,7 @@ Dbtup::disk_restart_undo(Signal* signal, Uint64 lsn,
   
   int flags = 0;
   int res= m_pgman.get_page(signal, preq, flags);
+  jamEntry();
   switch(res)
   {
   case 0:
@@ -1545,6 +1558,7 @@ Dbtup::disk_restart_undo_callback(Signal* signal,
     lsn = undo->m_lsn - 1; // make sure undo isn't run again...
     
     m_pgman.update_lsn(undo->m_key, lsn);
+    jamEntry();
   }
   else if (DBG_UNDO)
   {
@@ -1637,6 +1651,7 @@ Dbtup::disk_restart_undo_page_bits(Signal* signal, Apply_undo* undo)
 			  fragPtrP->m_tablespace_id);
   
   tsman.restart_undo_page_free_bits(&undo->m_key, new_bits, undo->m_lsn, lsn);
+  jamEntry();
 }
 
 int
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp
index 58d7fa8e35c..d7ee08f29dd 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp
@@ -62,13 +62,11 @@ Dbtup::execACC_SCANREQ(Signal* signal)
 	break;
       }
 
-#if BUG_27776_FIXED
       if (!AccScanReq::getNoDiskScanFlag(req->requestInfo)
 	  && tablePtr.p->m_no_of_disk_attributes)
       {
 	bits |= ScanOp::SCAN_DD;
       }
-#endif
       
       bool mm = (bits & ScanOp::SCAN_DD);
       if (tablePtr.p->m_attributes[mm].m_no_of_varsize > 0) {
diff --git a/storage/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp b/storage/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp
index 5810aea1932..69673796fee 100644
--- a/storage/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp
+++ b/storage/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp
@@ -1411,6 +1411,13 @@ void Ndbcntr::execNODE_FAILREP(Signal* signal)
 {
   jamEntry();
 
+  if (ERROR_INSERTED(1001))
+  {
+    sendSignalWithDelay(reference(), GSN_NODE_FAILREP, signal, 100, 
+                        signal->getLength());
+    return;
+  }
+  
   const NodeFailRep * nodeFail = (NodeFailRep *)&signal->theData[0];
   NdbNodeBitmask allFailed; 
   allFailed.assign(NdbNodeBitmask::Size, nodeFail->theNodes);
diff --git a/storage/ndb/src/kernel/blocks/pgman.cpp b/storage/ndb/src/kernel/blocks/pgman.cpp
index af648c71253..72333856cf1 100644
--- a/storage/ndb/src/kernel/blocks/pgman.cpp
+++ b/storage/ndb/src/kernel/blocks/pgman.cpp
@@ -123,8 +123,8 @@ Pgman::execREAD_CONFIG_REQ(Signal* signal)
   if (page_buffer > 0)
   {
     page_buffer /= GLOBAL_PAGE_SIZE; // in pages
-    m_page_entry_pool.setSize(100*page_buffer);
     m_param.m_max_pages = page_buffer;
+    m_page_entry_pool.setSize(m_param.m_lirs_stack_mult * page_buffer);
     m_param.m_max_hot_pages = (page_buffer * 9) / 10;
   }
 
@@ -141,6 +141,7 @@ Pgman::execREAD_CONFIG_REQ(Signal* signal)
 
 Pgman::Param::Param() :
   m_max_pages(64),      // smallish for testing
+  m_lirs_stack_mult(10),
   m_max_hot_pages(56),
   m_max_loop_count(256),
   m_max_io_waits(64),
@@ -301,6 +302,9 @@ Pgman::get_sublist_no(Page_state state)
   {
     return Page_entry::SL_LOCKED;
   }
+  if (state == Page_entry::ONSTACK) {
+    return Page_entry::SL_IDLE;
+  }
   return Page_entry::SL_OTHER;
 }
 
@@ -415,15 +419,55 @@ Pgman::get_page_entry(Ptr<Page_entry>& ptr, Uint32 file_no, Uint32 page_no)
 {
   if (find_page_entry(ptr, file_no, page_no))
   {
+    jam();
     ndbrequire(ptr.p->m_state != 0);
     m_stats.m_page_hits++;
+
+#ifdef VM_TRACE
+  debugOut << "PGMAN: get_page_entry: found" << endl;
+  debugOut << "PGMAN: " << ptr << endl;
+#endif
     return true;
   }
 
+  if (m_page_entry_pool.getNoOfFree() == 0)
+  {
+    jam();
+    Page_sublist& pl_idle = *m_page_sublist[Page_entry::SL_IDLE];
+    Ptr<Page_entry> idle_ptr;
+    if (pl_idle.first(idle_ptr))
+    {
+      jam();
+
+#ifdef VM_TRACE
+    debugOut << "PGMAN: get_page_entry: re-use idle entry" << endl;
+    debugOut << "PGMAN: " << idle_ptr << endl;
+#endif
+
+      Page_state state = idle_ptr.p->m_state;
+      ndbrequire(state == Page_entry::ONSTACK);
+
+      Page_stack& pl_stack = m_page_stack;
+      ndbrequire(pl_stack.hasPrev(idle_ptr));
+      pl_stack.remove(idle_ptr);
+      state &= ~ Page_entry::ONSTACK;
+      set_page_state(idle_ptr, state);
+      ndbrequire(idle_ptr.p->m_state == 0);
+
+      release_page_entry(idle_ptr);
+    }
+  }
+
   if (seize_page_entry(ptr, file_no, page_no))
   {
+    jam();
     ndbrequire(ptr.p->m_state == 0);
     m_stats.m_page_faults++;
+
+#ifdef VM_TRACE
+  debugOut << "PGMAN: get_page_entry: seize" << endl;
+  debugOut << "PGMAN: " << ptr << endl;
+#endif
     return true;
   }
 
@@ -904,9 +948,11 @@ Pgman::process_map(Signal* signal)
 #ifdef VM_TRACE
   debugOut << "PGMAN: >process_map" << endl;
 #endif
-  int max_count = m_param.m_max_io_waits - m_stats.m_current_io_waits;
-  if (max_count > 0)
+  int max_count = 0;
+  if (m_param.m_max_io_waits > m_stats.m_current_io_waits) {
+    max_count = m_param.m_max_io_waits - m_stats.m_current_io_waits;
     max_count = max_count / 2 + 1;
+  }
   Page_sublist& pl_map = *m_page_sublist[Page_entry::SL_MAP];
 
   while (! pl_map.isEmpty() && --max_count >= 0)
@@ -1058,15 +1104,10 @@ Pgman::process_cleanup(Signal* signal)
   }
 
   int max_loop_count = m_param.m_max_loop_count;
-  int max_count = m_param.m_max_io_waits - m_stats.m_current_io_waits;
-
-  if (max_count > 0)
-  {
+  int max_count = 0;
+  if (m_param.m_max_io_waits > m_stats.m_current_io_waits) {
+    max_count = m_param.m_max_io_waits - m_stats.m_current_io_waits;
     max_count = max_count / 2 + 1;
-    /*
-     * Possibly add code here to avoid writing too rapidly.  May be
-     * unnecessary since only cold pages are cleaned.
-     */
   }
 
   Ptr<Page_entry> ptr = m_cleanup_ptr;
@@ -1168,9 +1209,12 @@ bool
 Pgman::process_lcp(Signal* signal)
 {
   Page_hashlist& pl_hash = m_page_hashlist;
-  int max_count = m_param.m_max_io_waits - m_stats.m_current_io_waits;
-  if (max_count > 0)
+
+  int max_count = 0;
+  if (m_param.m_max_io_waits > m_stats.m_current_io_waits) {
+    max_count = m_param.m_max_io_waits - m_stats.m_current_io_waits;
     max_count = max_count / 2 + 1;
+  }
 
 #ifdef VM_TRACE
   debugOut
@@ -1561,6 +1605,7 @@ Pgman::execFSWRITEREF(Signal* signal)
 int
 Pgman::get_page(Signal* signal, Ptr<Page_entry> ptr, Page_request page_req)
 {
+  jamEntry();
 #ifdef VM_TRACE
   Ptr<Page_request> tmp = { &page_req, RNIL};
   debugOut << "PGMAN: >get_page" << endl;
@@ -1708,6 +1753,7 @@ Pgman::get_page(Signal* signal, Ptr<Page_entry> ptr, Page_request page_req)
 void
 Pgman::update_lsn(Ptr<Page_entry> ptr, Uint32 block, Uint64 lsn)
 {
+  jamEntry();
 #ifdef VM_TRACE
   const char* bname = getBlockName(block, "?");
   debugOut << "PGMAN: >update_lsn: block=" << bname << " lsn=" << lsn << endl;
@@ -1927,6 +1973,8 @@ Pgman::verify_page_entry(Ptr<Page_entry> ptr)
     break;
   case Page_entry::SL_LOCKED:
     break;
+  case Page_entry::SL_IDLE:
+    break;
   case Page_entry::SL_OTHER:
     break;
   default:
@@ -1973,8 +2021,11 @@ Pgman::verify_page_lists()
   ndbrequire(stack_count == pl_stack.count() || dump_page_lists());
   ndbrequire(queue_count == pl_queue.count() || dump_page_lists());
 
+  Uint32 hot_count = 0;
   Uint32 hot_bound_count = 0;
   Uint32 cold_bound_count = 0;
+  Uint32 stack_request_count = 0;
+  Uint32 queue_request_count = 0;
 
   Uint32 i1 = RNIL;
   for (pl_stack.first(ptr); ptr.i != RNIL; pl_stack.next(ptr))
@@ -1985,9 +2036,13 @@ Pgman::verify_page_lists()
     ndbrequire(state & Page_entry::ONSTACK || dump_page_lists());
     if (! pl_stack.hasPrev(ptr))
       ndbrequire(state & Page_entry::HOT || dump_page_lists());
-    if (state & Page_entry::HOT &&
-        state & Page_entry::BOUND)
-      hot_bound_count++;
+    if (state & Page_entry::HOT) {
+      hot_count++;
+      if (state & Page_entry::BOUND)
+        hot_bound_count++;
+    }
+    if (state & Page_entry::REQUEST)
+      stack_request_count++;
   }
 
   Uint32 i2 = RNIL;
@@ -1999,6 +2054,8 @@ Pgman::verify_page_lists()
     ndbrequire(state & Page_entry::ONQUEUE || dump_page_lists());
     ndbrequire(state & Page_entry::BOUND || dump_page_lists());
     cold_bound_count++;
+    if (state & Page_entry::REQUEST)
+      queue_request_count++;
   }
 
   Uint32 tot_bound_count =
@@ -2031,7 +2088,11 @@ Pgman::verify_page_lists()
            << " cache:" << m_stats.m_num_pages
            << "(" << locked_bound_count << "L)"
            << " stack:" << pl_stack.count()
+           << " hot:" << hot_count
+           << " hot_bound:" << hot_bound_count
+           << " stack_request:" << stack_request_count
            << " queue:" << pl_queue.count()
+           << " queue_request:" << queue_request_count
            << " queuewait:" << queuewait_count << endl;
 
   debugOut << "PGMAN:";
@@ -2139,6 +2200,8 @@ Pgman::get_sublist_name(Uint32 list_no)
     return "busy";
   case Page_entry::SL_LOCKED:
     return "locked";
+  case Page_entry::SL_IDLE:
+    return "idle";
   case Page_entry::SL_OTHER:
     return "other";
   }
diff --git a/storage/ndb/src/kernel/blocks/pgman.hpp b/storage/ndb/src/kernel/blocks/pgman.hpp
index 07029d1c3e5..e3bf0fa5780 100644
--- a/storage/ndb/src/kernel/blocks/pgman.hpp
+++ b/storage/ndb/src/kernel/blocks/pgman.hpp
@@ -325,8 +325,9 @@ private:
       ,SL_CALLBACK_IO = 4
       ,SL_BUSY = 5
       ,SL_LOCKED = 6
-      ,SL_OTHER = 7
-      ,SUBLIST_COUNT = 8
+      ,SL_IDLE = 7
+      ,SL_OTHER = 8
+      ,SUBLIST_COUNT = 9
     };
 
     Uint16 m_file_no;       // disk page address set at seize
@@ -401,6 +402,7 @@ private:
   struct Param {
     Param();
     Uint32 m_max_pages;         // max number of cache pages
+    Uint32 m_lirs_stack_mult;   // in m_max_pages (around 3-10)
     Uint32 m_max_hot_pages;     // max hot cache pages (up to 99%)
     Uint32 m_max_loop_count;    // limit purely local loops
     Uint32 m_max_io_waits;
diff --git a/storage/ndb/test/ndbapi/testNodeRestart.cpp b/storage/ndb/test/ndbapi/testNodeRestart.cpp
index fb05681bab4..b11d1942303 100644
--- a/storage/ndb/test/ndbapi/testNodeRestart.cpp
+++ b/storage/ndb/test/ndbapi/testNodeRestart.cpp
@@ -1535,6 +1535,85 @@ runBug28023(NDBT_Context* ctx, NDBT_Step* step)
       return NDBT_FAILED;
     }
   }
+
+  return NDBT_OK;
+}
+
+
+int
+runBug28717(NDBT_Context* ctx, NDBT_Step* step)
+{
+  int result = NDBT_OK;
+  int loops = ctx->getNumLoops();
+  int records = ctx->getNumRecords();
+  Ndb* pNdb = GETNDB(step);
+  NdbRestarter res;
+
+  if (res.getNumDbNodes() < 4)
+  {
+    return NDBT_OK;
+  }
+
+  int master = res.getMasterNodeId();
+  int node0 = res.getRandomNodeOtherNodeGroup(master, rand());
+  int node1 = res.getRandomNodeSameNodeGroup(node0, rand());
+  
+  ndbout_c("master: %d node0: %d node1: %d", master, node0, node1);
+  
+  if (res.restartOneDbNode(node0, false, true, true))
+  {
+    return NDBT_FAILED;
+  }
+
+  {
+    int filter[] = { 15, NDB_MGM_EVENT_CATEGORY_CHECKPOINT, 0 };
+    NdbLogEventHandle handle = 
+      ndb_mgm_create_logevent_handle(res.handle, filter);
+    
+
+    int dump[] = { DumpStateOrd::DihStartLcpImmediately };
+    struct ndb_logevent event;
+    
+    for (Uint32 i = 0; i<3; i++)
+    {
+      res.dumpStateOneNode(master, dump, 1);
+      while(ndb_logevent_get_next(handle, &event, 0) >= 0 &&
+            event.type != NDB_LE_LocalCheckpointStarted);
+      while(ndb_logevent_get_next(handle, &event, 0) >= 0 &&
+            event.type != NDB_LE_LocalCheckpointCompleted);
+    } 
+  }
+  
+  if (res.waitNodesNoStart(&node0, 1))
+    return NDBT_FAILED;
+  
+  int val2[] = { DumpStateOrd::CmvmiSetRestartOnErrorInsert, 1 };
+  
+  if (res.dumpStateOneNode(node0, val2, 2))
+    return NDBT_FAILED;
+  
+  if (res.insertErrorInNode(node0, 5010))
+    return NDBT_FAILED;
+  
+  if (res.insertErrorInNode(node1, 1001))
+    return NDBT_FAILED;
+  
+  if (res.startNodes(&node0, 1))
+    return NDBT_FAILED;
+  
+  NdbSleep_SecSleep(3);
+
+  if (res.insertErrorInNode(node1, 0))
+    return NDBT_FAILED;
+
+  if (res.waitNodesNoStart(&node0, 1))
+    return NDBT_FAILED;
+
+  if (res.startNodes(&node0, 1))
+    return NDBT_FAILED;
+
+  if (res.waitClusterStarted())
+    return NDBT_FAILED;
   
   return NDBT_OK;
 }
@@ -1896,6 +1975,12 @@ TESTCASE("Bug27466", ""){
 TESTCASE("Bug28023", ""){
   INITIALIZER(runBug28023);
 }
+TESTCASE("Bug25554", ""){
+  INITIALIZER(runBug25554);
+}
+TESTCASE("Bug28717", ""){
+  INITIALIZER(runBug28717);
+}
 NDBT_TESTSUITE_END(testNodeRestart);
 
 int main(int argc, const char** argv){
diff --git a/storage/ndb/test/run-test/daily-basic-tests.txt b/storage/ndb/test/run-test/daily-basic-tests.txt
index b6177be02da..d133ed0b400 100644
--- a/storage/ndb/test/run-test/daily-basic-tests.txt
+++ b/storage/ndb/test/run-test/daily-basic-tests.txt
@@ -559,6 +559,10 @@ max-time: 1500
 cmd: testDict
 args: -n CreateAndDrop 
 
+max-time: 1000
+cmd: testNodeRestart
+args: -n Bug28717 T1
+
 max-time: 1500
 cmd: testDict
 args: -n CreateAndDropAtRandom -l 200 T1