ndb dd - Impl. support for using delayed page request, to simulate "full" page buffer cache

storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp:
  Impl. support for using delayed page request, to simulate "full" page buffer cache
storage/ndb/src/kernel/blocks/pgman.cpp:
  Impl. support for using delayed page request, to simulate "full" page buffer cache
storage/ndb/src/kernel/blocks/pgman.hpp:
  Impl. support for using delayed page request, to simulate "full" page buffer cache

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

@@ -484,6 +484,14 @@ Dbtup::load_diskpage(Signal* signal,
     req.m_callback.m_callbackFunction= 
       safe_cast(&Dbtup::disk_page_load_callback);
 
+#ifdef ERROR_INSERTED
+    if (ERROR_INSERTED(4022))
+    {
+      flags |= Page_cache_client::DELAY_REQ;
+      req.m_delay_until_time = NdbTick_CurrentMillisecond()+(Uint64)3000;
+    }
+#endif
+    
     if((res= m_pgman.get_page(signal, req, flags)) > 0)
     {
       //ndbout_c("in cache");
@@ -3119,6 +3127,35 @@ Dbtup::nr_delete(Signal* signal, Uint32 senderData,
     preq.m_callback.m_callbackFunction =
       safe_cast(&Dbtup::nr_delete_page_callback);
     int flags = Page_cache_client::COMMIT_REQ;
+    
+#ifdef ERROR_INSERT
+    if (ERROR_INSERTED(4023) || ERROR_INSERTED(4024))
+    {
+      int rnd = rand() % 100;
+      int slp = 0;
+      if (ERROR_INSERTED(4024))
+      {
+	slp = 3000;
+      }
+      else if (rnd > 90)
+      {
+	slp = 3000;
+      }
+      else if (rnd > 70)
+      {
+	slp = 100;
+      }
+      
+      ndbout_c("rnd: %d slp: %d", rnd, slp);
+      
+      if (slp)
+      {
+	flags |= Page_cache_client::DELAY_REQ;
+	preq.m_delay_until_time = NdbTick_CurrentMillisecond()+(Uint64)slp;
+      }
+    }
+#endif
+    
     res = m_pgman.get_page(signal, preq, flags);
     if (res == 0)
     {

storage/ndb/src/kernel/blocks/pgman.cpp

@@ -944,12 +944,16 @@ Pgman::process_callback(Signal* signal)
   int max_count = 1;
   Page_sublist& pl_callback = *m_page_sublist[Page_entry::SL_CALLBACK];
 
-  while (! pl_callback.isEmpty() && --max_count >= 0)
-  {
-    jam();
   Ptr<Page_entry> ptr;
   pl_callback.first(ptr);
-    if (! process_callback(signal, ptr))
+
+  while (! ptr.isNull() && --max_count >= 0)
+  {
+    jam();
+    Ptr<Page_entry> curr = ptr;
+    pl_callback.next(ptr);
+    
+    if (! process_callback(signal, curr))
     {
       jam();
       break;
@@ -987,6 +991,18 @@ Pgman::process_callback(Signal* signal, Ptr<Page_entry> ptr)
 #ifdef VM_TRACE
       debugOut << "PGMAN: " << req_ptr << " : process_callback" << endl;
 #endif
+
+#ifdef ERROR_INSERT
+      if (req_ptr.p->m_flags & Page_request::DELAY_REQ)
+      {
+	Uint64 now = NdbTick_CurrentMillisecond();
+	if (now < req_ptr.p->m_delay_until_time)
+	{
+	  break;
+	}
+      }
+#endif
+      
       b = globalData.getBlock(req_ptr.p->m_block);
       callback = req_ptr.p->m_callback;
       
@@ -1576,6 +1592,12 @@ Pgman::get_page(Signal* signal, Ptr<Page_entry> ptr, Page_request page_req)
   
   bool only_request = ptr.p->m_requests.isEmpty();
   
+  if (req_flags & Page_request::DELAY_REQ)
+  {
+    jam();
+    only_request = false;
+  }
+  
   if (only_request &&
       state & Page_entry::MAPPED)
   {
@@ -1623,6 +1645,9 @@ Pgman::get_page(Signal* signal, Ptr<Page_entry> ptr, Page_request page_req)
   req_ptr.p->m_block = page_req.m_block;
   req_ptr.p->m_flags = page_req.m_flags;
   req_ptr.p->m_callback = page_req.m_callback;
+#ifdef ERROR_INSERT
+  req_ptr.p->m_delay_until_time = page_req.m_delay_until_time;
+#endif
   
   state |= Page_entry::REQUEST;
   if (only_request && req_flags & Page_request::EMPTY_PAGE)

storage/ndb/src/kernel/blocks/pgman.hpp

@@ -256,12 +256,18 @@ private:
       ,DIRTY_REQ    = 0x0200 // make page dirty wo/ update_lsn
       ,UNLOCK_PAGE  = 0x0400
       ,CORR_REQ     = 0x0800 // correlated request (no LIRS update)
+#ifdef ERROR_INSERT
+      ,DELAY_REQ    = 0x1000 // Force request to be delayed
+#endif
     };
     
     Uint16 m_block;
     Uint16 m_flags;
     SimulatedBlock::Callback m_callback;
 
+#ifdef ERROR_INSERT
+    Uint64 m_delay_until_time;
+#endif
     Uint32 nextList;
     Uint32 m_magic;
   };
@@ -508,6 +514,10 @@ public:
   struct Request {
     Local_key m_page;
     SimulatedBlock::Callback m_callback;
+    
+#ifdef ERROR_INSERT
+    Uint64 m_delay_until_time;
+#endif
   };
 
   Ptr<GlobalPage> m_ptr;        // TODO remove
@@ -520,6 +530,9 @@ public:
     ,DIRTY_REQ = Pgman::Page_request::DIRTY_REQ
     ,UNLOCK_PAGE = Pgman::Page_request::UNLOCK_PAGE
     ,CORR_REQ = Pgman::Page_request::CORR_REQ
+#ifdef ERROR_INSERT
+    ,DELAY_REQ = Pgman::Page_request::DELAY_REQ
+#endif
   };
   
   /**
@@ -588,6 +601,9 @@ Page_cache_client::get_page(Signal* signal, Request& req, Uint32 flags)
   page_req.m_block = m_block;
   page_req.m_flags = flags;
   page_req.m_callback = req.m_callback;
+#ifdef ERROR_INSERT
+  page_req.m_delay_until_time = req.m_delay_until_time;
+#endif
   
   int i = m_pgman->get_page(signal, entry_ptr, page_req);
   if (i > 0)