Merge localhost:src/51-new

into  eel.(none):/home/jonas/src/mysql-5.1-new

storage/ndb/include/kernel/signaldata/PgmanContinueB.hpp

@@ -29,7 +29,8 @@ private:
     STATS_LOOP = 0,
     BUSY_LOOP = 1,
     CLEANUP_LOOP = 2,
-    LCP_LOOP = 3
+    LCP_LOOP = 3,
+    LCP_PREPARE = 4
   };
 };
 

storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp

@@ -1009,8 +1009,6 @@ public:
       LCP_SR_STARTED = 10,
       LCP_SR_COMPLETED = 11
     };
-    LcpLocRecord m_acc;
-    LcpLocRecord m_tup;
  
     LcpState lcpState;
     bool firstFragmentFlag;
@@ -1028,6 +1026,7 @@ public:
     bool   reportEmpty;
     NdbNodeBitmask m_EMPTY_LCP_REQ;
 
+    Uint32 m_error;
     Uint32 m_outstanding;
   }; // Size 76 bytes
   typedef Ptr<LcpRecord> LcpRecordPtr;
@@ -2248,10 +2247,6 @@ private:
   bool checkLcpStarted(Signal* signal);
   void checkLcpTupprep(Signal* signal);
   void getNextFragForLcp(Signal* signal);
-  void initLcpLocAcc(Signal* signal, Uint32 fragId);
-  void initLcpLocTup(Signal* signal, Uint32 fragId);
-  void releaseLocalLcps(Signal* signal);
-  void seizeLcpLoc(Signal* signal);
   void sendAccContOp(Signal* signal);
   void sendStartLcp(Signal* signal);
   void setLogTail(Signal* signal, Uint32 keepGci);
@@ -2283,7 +2278,6 @@ private:
   void checkNewMbyte(Signal* signal);
   void checkReadExecSr(Signal* signal);
   void checkScanTcCompleted(Signal* signal);
-  void checkSrCompleted(Signal* signal);
   void closeFile(Signal* signal, LogFileRecordPtr logFilePtr, Uint32 place);
   void completedLogPage(Signal* signal, Uint32 clpType, Uint32 place);
   void deleteFragrec(Uint32 fragId);
@@ -2302,7 +2296,6 @@ private:
   void initialiseFragrec(Signal* signal);
   void initialiseGcprec(Signal* signal);
   void initialiseLcpRec(Signal* signal);
-  void initialiseLcpLocrec(Signal* signal);
   void initialiseLfo(Signal* signal);
   void initialiseLogFile(Signal* signal);
   void initialiseLogPage(Signal* signal);
@@ -2346,7 +2339,6 @@ private:
   void releaseActiveCopy(Signal* signal);
   void releaseAddfragrec(Signal* signal);
   void releaseFragrec();
-  void releaseLcpLoc(Signal* signal);
   void releaseOprec(Signal* signal);
   void releasePageRef(Signal* signal);
   void releaseMmPages(Signal* signal);

storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp

@@ -10924,13 +10924,38 @@ void Dblqh::execLCP_PREPARE_REF(Signal* signal)
   tabptr.i = ref->tableId;
   ptrCheckGuard(tabptr, ctabrecFileSize, tablerec);
   
-  lcpPtr.p->lcpState = LcpRecord::LCP_COMPLETED;
-  lcpPtr.p->m_acc.lcpLocstate = LcpLocRecord::ACC_COMPLETED;
+  ndbrequire(lcpPtr.p->m_outstanding);
+  lcpPtr.p->m_outstanding--;
 
-  ndbout_c("execLCP_PREPARE_REF tab: %d frag: %d err: %d", 
-	   fragptr.p->tabRef, fragptr.p->fragId, ref->errorCode);
+  /**
+   * Only BACKUP is allowed to ref LCP_PREPARE
+   */
+  ndbrequire(refToBlock(signal->getSendersBlockRef()) == BACKUP);
+  lcpPtr.p->m_error = ref->errorCode;
+  
+  if (lcpPtr.p->m_outstanding == 0)
+  {
+    jam();
     
+    if(lcpPtr.p->firstFragmentFlag)
+    {
+      jam();
+      LcpFragOrd *ord= (LcpFragOrd*)signal->getDataPtrSend();
+      lcpPtr.p->firstFragmentFlag= false;
+      *ord = lcpPtr.p->currentFragment.lcpFragOrd;
+      EXECUTE_DIRECT(PGMAN, GSN_LCP_FRAG_ORD, signal, signal->length());
+      jamEntry();
+      
+      /**
+       * First fragment mean that last LCP is complete :-)
+       */
+      EXECUTE_DIRECT(TSMAN, GSN_END_LCP_REQ, signal, signal->length());
+      jamEntry();
+    }
+    
+    lcpPtr.p->lcpState = LcpRecord::LCP_COMPLETED;
     contChkpNextFragLab(signal);
+  }
 }
 
 /* --------------------------------------------------------------------------
@@ -10951,14 +10976,45 @@ void Dblqh::execLCP_PREPARE_CONF(Signal* signal)
   fragptr.i = lcpPtr.p->currentFragment.fragPtrI;
   c_fragment_pool.getPtr(fragptr);
 
+  if (refToBlock(signal->getSendersBlockRef()) != PGMAN)
+  {
     ndbrequire(conf->tableId == fragptr.p->tabRef);
     ndbrequire(conf->fragmentId == fragptr.p->fragId);
+  }
   
-  lcpPtr.p->lcpState = LcpRecord::LCP_WAIT_HOLDOPS;
-  lcpPtr.p->m_acc.lcpLocstate = LcpLocRecord::WAIT_LCPHOLDOP;
+  ndbrequire(lcpPtr.p->m_outstanding);
+  lcpPtr.p->m_outstanding--;
+  if (lcpPtr.p->m_outstanding == 0)
+  {
+    jam();
+
+    if(lcpPtr.p->firstFragmentFlag)
+    {
+      jam();
+      LcpFragOrd *ord= (LcpFragOrd*)signal->getDataPtrSend();
+      lcpPtr.p->firstFragmentFlag= false;
+      *ord = lcpPtr.p->currentFragment.lcpFragOrd;
+      EXECUTE_DIRECT(PGMAN, GSN_LCP_FRAG_ORD, signal, signal->length());
+      jamEntry();
+      
+      /**
+       * First fragment mean that last LCP is complete :-)
+       */
+      EXECUTE_DIRECT(TSMAN, GSN_END_LCP_REQ, signal, signal->length());
+      jamEntry();
+    }
+    
+    if (lcpPtr.p->m_error)
+    {
+      jam();
+
+      lcpPtr.p->lcpState = LcpRecord::LCP_COMPLETED;
+      contChkpNextFragLab(signal);
+      return;
+    }
 
+    lcpPtr.p->lcpState = LcpRecord::LCP_WAIT_HOLDOPS;
     lcpPtr.p->lcpState = LcpRecord::LCP_START_CHKP;
-  lcpPtr.p->m_acc.lcpLocstate = LcpLocRecord::HOLDOP_READY;
     
     /* ----------------------------------------------------------------------
      *    UPDATE THE MAX_GCI_IN_LCP AND MAX_GCI_COMPLETED_IN_LCP NOW BEFORE
@@ -10977,21 +11033,6 @@ void Dblqh::execLCP_PREPARE_CONF(Signal* signal)
       *ord = lcpPtr.p->currentFragment.lcpFragOrd;
       EXECUTE_DIRECT(DBTUP, GSN_LCP_FRAG_ORD, signal, signal->length());
       jamEntry();
-
-    if(lcpPtr.p->firstFragmentFlag)
-    {
-      jam();
-      lcpPtr.p->firstFragmentFlag= false;
-      *ord = lcpPtr.p->currentFragment.lcpFragOrd;
-      EXECUTE_DIRECT(PGMAN, GSN_LCP_FRAG_ORD, signal, signal->length());
-      jamEntry();
-
-      /**
-       * First fragment mean that last LCP is complete :-)
-       */
-      EXECUTE_DIRECT(TSMAN, GSN_END_LCP_REQ, signal, signal->length());
-      jamEntry();
-    }
     }
     
     BackupFragmentReq* req= (BackupFragmentReq*)signal->getDataPtr();
@@ -11013,9 +11054,7 @@ void Dblqh::execLCP_PREPARE_CONF(Signal* signal)
       sendSignal(BACKUP_REF, GSN_BACKUP_FRAGMENT_REQ, signal, 
 		 BackupFragmentReq::SignalLength, JBB);
     }
-  
-  lcpPtr.p->m_acc.lcpLocstate = LcpLocRecord::ACC_STARTED;
-  lcpPtr.p->m_tup.lcpLocstate = LcpLocRecord::TUP_COMPLETED;
+  }
 }
 
 void Dblqh::execBACKUP_FRAGMENT_REF(Signal* signal) 
@@ -11030,8 +11069,7 @@ void Dblqh::execBACKUP_FRAGMENT_CONF(Signal* signal)
 
   lcpPtr.i = 0;
   ptrCheckGuard(lcpPtr, clcpFileSize, lcpRecord);
-  ndbrequire(lcpPtr.p->m_acc.lcpLocstate == LcpLocRecord::ACC_STARTED);
-  lcpPtr.p->m_acc.lcpLocstate = LcpLocRecord::ACC_COMPLETED;
+  ndbrequire(lcpPtr.p->lcpState == LcpRecord::LCP_START_CHKP);
   lcpPtr.p->lcpState = LcpRecord::LCP_COMPLETED;
 
   /* ------------------------------------------------------------------------
@@ -11152,6 +11190,9 @@ void Dblqh::sendLCP_FRAGIDREQ(Signal* signal)
     return;
   }
 
+  lcpPtr.p->m_error = 0;
+  lcpPtr.p->m_outstanding = 1;
+
   ndbrequire(tabPtr.p->tableStatus == Tablerec::TABLE_DEFINED);
   
   lcpPtr.p->lcpState = LcpRecord::LCP_WAIT_FRAGID;
@@ -11166,6 +11207,13 @@ void Dblqh::sendLCP_FRAGIDREQ(Signal* signal)
   req->backupId = lcpPtr.p->currentFragment.lcpFragOrd.lcpId;
   sendSignal(BACKUP_REF, GSN_LCP_PREPARE_REQ, signal, 
 	     LcpPrepareReq::SignalLength, JBB);
+
+  if (lcpPtr.p->firstFragmentFlag)
+  {
+    lcpPtr.p->m_outstanding++;
+    sendSignal(PGMAN_REF, GSN_LCP_PREPARE_REQ, signal, 
+	       LcpPrepareReq::SignalLength, JBB);
+  }
 }//Dblqh::sendLCP_FRAGIDREQ()
 
 void Dblqh::sendEMPTY_LCP_CONF(Signal* signal, bool idle)
@@ -13609,7 +13657,7 @@ void Dblqh::execRESTORE_LCP_CONF(Signal* signal)
     lcpPtr.p->m_outstanding = 1;
     
     signal->theData[0] = c_lcpId;
-    sendSignal(TSMAN_REF, GSN_START_RECREQ, signal, 1, JBB);
+    sendSignal(LGMAN_REF, GSN_START_RECREQ, signal, 1, JBB);
     return;
   }
 }
@@ -13663,7 +13711,7 @@ void Dblqh::execSTART_RECREQ(Signal* signal)
     lcpPtr.p->m_outstanding = 1;
     
     signal->theData[0] = c_lcpId;
-    sendSignal(TSMAN_REF, GSN_START_RECREQ, signal, 1, JBB);
+    sendSignal(LGMAN_REF, GSN_START_RECREQ, signal, 1, JBB);
   }//if
 }//Dblqh::execSTART_RECREQ()
 
@@ -13689,13 +13737,14 @@ void Dblqh::execSTART_RECCONF(Signal* signal)
 
   switch(refToBlock(sender)){
   case TSMAN:
+    jam();
+    break;
+  case LGMAN:
     jam();
     lcpPtr.p->m_outstanding++;
     signal->theData[0] = c_lcpId;
-    sendSignal(LGMAN_REF, GSN_START_RECREQ, signal, 1, JBB);
+    sendSignal(TSMAN_REF, GSN_START_RECREQ, signal, 1, JBB);
     return;
-  case LGMAN:
-    jam();
     break;
   default:
     ndbrequire(false);
@@ -15790,29 +15839,6 @@ void Dblqh::checkScanTcCompleted(Signal* signal)
   }//if
 }//Dblqh::checkScanTcCompleted()
 
-/* ========================================================================== 
- * === CHECK IF ALL PARTS OF A SYSTEM RESTART ON A FRAGMENT ARE COMPLETED === 
- *
- *       SUBROUTINE SHORT NAME = CSC
- * ========================================================================= */
-void Dblqh::checkSrCompleted(Signal* signal) 
-{
-  terrorCode = ZOK;
-  ptrGuard(lcpPtr);
-  if(lcpPtr.p->m_acc.lcpLocstate != LcpLocRecord::SR_ACC_COMPLETED)
-  {
-    ndbrequire(lcpPtr.p->m_acc.lcpLocstate == LcpLocRecord::SR_ACC_STARTED);
-    return;
-  }
-  
-  if(lcpPtr.p->m_tup.lcpLocstate != LcpLocRecord::SR_TUP_COMPLETED) 
-  {
-    ndbrequire(lcpPtr.p->m_tup.lcpLocstate == LcpLocRecord::SR_TUP_STARTED);
-    return;
-  }
-  lcpPtr.p->lcpState = LcpRecord::LCP_SR_COMPLETED;
-}//Dblqh::checkSrCompleted()
-
 /* ------------------------------------------------------------------------- */
 /* ------       CLOSE A FILE DURING EXECUTION OF FRAGMENT LOG        ------- */
 /*                                                                           */
@@ -18125,10 +18151,8 @@ Dblqh::execDUMP_STATE_ORD(Signal* signal)
     // Print information about the current local checkpoint
     TlcpPtr.i = 0;
     ptrAss(TlcpPtr, lcpRecord);
-    infoEvent(" lcpState=%d", TlcpPtr.p->lcpState);
-    infoEvent(" lcpAccptr=%d lastFragmentFlag=%d",
-	      TlcpPtr.p->m_acc.lcpRef,
-	      TlcpPtr.p->lastFragmentFlag);
+    infoEvent(" lcpState=%d lastFragmentFlag=%d", 
+	      TlcpPtr.p->lcpState, TlcpPtr.p->lastFragmentFlag);
     infoEvent("currentFragment.fragPtrI=%d",
 	      TlcpPtr.p->currentFragment.fragPtrI);
     infoEvent("currentFragment.lcpFragOrd.tableId=%d",

storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp

@@ -2600,7 +2600,7 @@ private:
   void disk_page_prealloc_callback_common(Signal*, 
 					  Ptr<Page_request>, 
 					  Ptr<Fragrecord>,
-					  Ptr<GlobalPage>);
+					  Ptr<Page>);
   
   void disk_page_alloc(Signal*, 
 		       Tablerec*, Fragrecord*, Local_key*, PagePtr, Uint32);
@@ -2631,18 +2631,22 @@ private:
 
   void undo_createtable_callback(Signal* signal, Uint32 opPtrI, Uint32 unused);
 
+  void disk_page_set_dirty(Ptr<Page>);
+  void restart_setup_page(Ptr<Page>);
+  void update_extent_pos(Disk_alloc_info&, Ptr<Extent_info>);
+  
   /**
    * Disk restart code
    */
 public:
   int disk_page_load_hook(Uint32 page_id);
   
-  void disk_page_unmap_callback(Uint32 page_id);
+  void disk_page_unmap_callback(Uint32 page_id, Uint32 dirty_count);
   
   int disk_restart_alloc_extent(Uint32 tableId, Uint32 fragId, 
 				const Local_key* key, Uint32 pages);
   void disk_restart_page_bits(Uint32 tableId, Uint32 fragId,
-			      const Local_key*, Uint32 old_bits, Uint32 bits);
+			      const Local_key*, Uint32 bits);
   void disk_restart_undo(Signal* signal, Uint64 lsn,
 			 Uint32 type, const Uint32 * ptr, Uint32 len);
 
@@ -2654,6 +2658,7 @@ public:
     Ptr<Tablerec> m_table_ptr;
     Ptr<Fragrecord> m_fragment_ptr;
     Ptr<Page> m_page_ptr;
+    Ptr<Extent_info> m_extent_ptr;
     Local_key m_key;
   };
   
@@ -2664,7 +2669,7 @@ private:
   void disk_restart_undo_alloc(Apply_undo*);
   void disk_restart_undo_update(Apply_undo*);
   void disk_restart_undo_free(Apply_undo*);
-  void disk_restart_undo_page_bits(Apply_undo*);
+  void disk_restart_undo_page_bits(Signal*, Apply_undo*);
 
 #ifdef VM_TRACE
   void verify_page_lists(Disk_alloc_info&);

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

@@ -327,6 +327,8 @@ Dbtup::disk_page_commit_callback(Signal* signal,
   regOperPtr.p->m_commit_disk_callback_page= page_id;
   m_global_page_pool.getPtr(m_pgman.m_ptr, page_id);
   
+  disk_page_set_dirty(*(Ptr<Page>*)&m_pgman.m_ptr);
+  
   execTUP_COMMITREQ(signal);
   if(signal->theData[0] == 0)
     c_lqh->tupcommit_conf_callback(signal, regOperPtr.p->userpointer);
@@ -477,8 +479,7 @@ void Dbtup::execTUP_COMMITREQ(Signal* signal)
      * the page hot.   XXX move to TUP which knows better.
      */
     int flags= regOperPtr.p->op_struct.op_type |
-      Page_cache_client::COMMIT_REQ | Page_cache_client::STRICT_ORDER |
-      Page_cache_client::CORR_REQ;
+      Page_cache_client::COMMIT_REQ | Page_cache_client::CORR_REQ;
     int res= m_pgman.get_page(signal, req, flags);
     switch(res){
     case 0:
@@ -491,6 +492,7 @@ void Dbtup::execTUP_COMMITREQ(Signal* signal)
       ndbrequire("NOT YET IMPLEMENTED" == 0);
       break;
     }
+    disk_page_set_dirty(*(Ptr<Page>*)&m_pgman.m_ptr);
     regOperPtr.p->m_commit_disk_callback_page= res;
     regOperPtr.p->op_struct.m_load_diskpage_on_commit= 0;
   } 

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

@@ -484,8 +484,6 @@ Dbtup::load_diskpage(Signal* signal,
     req.m_callback.m_callbackFunction= 
       safe_cast(&Dbtup::disk_page_load_callback);
     
-    // Make sure we maintain order
-    flags |= Page_cache_client::STRICT_ORDER;
     if((res= m_pgman.get_page(signal, req, flags)) > 0)
     {
       //ndbout_c("in cache");
@@ -563,8 +561,6 @@ Dbtup::load_diskpage_scan(Signal* signal,
     req.m_callback.m_callbackFunction= 
       safe_cast(&Dbtup::disk_page_load_scan_callback);
     
-    // Make sure we maintain order
-    flags |= Page_cache_client::STRICT_ORDER;
     if((res= m_pgman.get_page(signal, req, flags)) > 0)
     {
       // ndbout_c("in cache");
@@ -3111,8 +3107,7 @@ Dbtup::nr_delete(Signal* signal, Uint32 senderData,
     preq.m_callback.m_callbackData = senderData;
     preq.m_callback.m_callbackFunction =
       safe_cast(&Dbtup::nr_delete_page_callback);
-    int flags = Page_cache_client::COMMIT_REQ | 
-      Page_cache_client::STRICT_ORDER;
+    int flags = Page_cache_client::COMMIT_REQ;
     res = m_pgman.get_page(signal, preq, flags);
     if (res == 0)
     {

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

@@ -915,13 +915,17 @@ Dbtup::drop_fragment_unmap_pages(Signal *signal,
       return;
     }
     
-    Uint32 page_id = alloc_info.m_dirty_pages[pos].firstItem;
-    Ptr<GlobalPage> page;
-    m_global_page_pool.getPtr(page, page_id);
+    Ptr<Page> pagePtr;
+    ArrayPool<Page> *pool= (ArrayPool<Page>*)&m_global_page_pool;
+    {
+      LocalDLList<Page> list(*pool, alloc_info.m_dirty_pages[pos]);
+      list.first(pagePtr);
+      list.remove(pagePtr);
+    }
     
     Page_cache_client::Request req;
-    req.m_page.m_page_no = ((Page*)page.p)->m_page_no;
-    req.m_page.m_file_no = ((Page*)page.p)->m_file_no;
+    req.m_page.m_page_no = pagePtr.p->m_page_no;
+    req.m_page.m_file_no = pagePtr.p->m_file_no;
     
     req.m_callback.m_callbackData= pos;
     req.m_callback.m_callbackFunction = 
@@ -935,7 +939,7 @@ Dbtup::drop_fragment_unmap_pages(Signal *signal,
     case -1:
       break;
     default:
-      ndbrequire(res == page_id);
+      ndbrequire(res == pagePtr.i);
       drop_fragment_unmap_page_callback(signal, pos, res);
     }
     return;

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

@@ -694,10 +694,11 @@ Dbtup::scanNext(Signal* signal, ScanOpPtr scanPtr)
 				  frag.fragTableId, 
 				  frag.fragmentId, 
 				  frag.m_tablespace_id);
-	  unsigned bits = ~(unsigned)0;
-	  int ret = tsman.get_page_free_bits(&key, &bits);
+	  unsigned uncommitted, committed;
+	  uncommitted = committed = ~(unsigned)0;
+	  int ret = tsman.get_page_free_bits(&key, &uncommitted, &committed);
 	  ndbrequire(ret == 0);
-	  if (bits == 0) {
+	  if (committed == 0) {
 	    // skip empty page
 	    jam();
 	    pos.m_get = ScanPos::Get_next_page_dd;
@@ -710,7 +711,7 @@ Dbtup::scanNext(Signal* signal, ScanOpPtr scanPtr)
         preq.m_callback.m_callbackData = scanPtr.i;
         preq.m_callback.m_callbackFunction =
           safe_cast(&Dbtup::disk_page_tup_scan_callback);
-        int flags = Page_cache_client::STRICT_ORDER;
+        int flags = 0;
         int res = m_pgman.get_page(signal, preq, flags);
         if (res == 0) {
           jam();

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

@@ -222,7 +222,7 @@ File_formats::Datafile::Extent_header::check_free(Uint32 extent_size) const
   for(; words; words--)
     sum |= m_page_bitmask[words-1];
 
-  if(sum & 0x7777)
+  if(sum & 0x3333)
     return false;
   
   return true;

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

@@ -2887,6 +2887,9 @@ Lgman::stop_run_undo_log(Signal* signal)
 void
 Lgman::execEND_LCP_CONF(Signal* signal)
 {
+  Dbtup* tup= (Dbtup*)globalData.getBlock(DBTUP);
+  tup->disk_restart_undo(signal, 0, File_formats::Undofile::UNDO_END, 0, 0);
+  
   /**
    * pgman has completed flushing all pages
    *

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

@@ -249,15 +249,13 @@ private:
   struct Page_request {
     enum Flags {
       OP_MASK       = 0x000F // 4 bits for TUP operation
-      ,STRICT_ORDER = 0x0010 // maintain request order
       ,LOCK_PAGE    = 0x0020 // lock page in memory
       ,EMPTY_PAGE   = 0x0040 // empty (new) page
       ,ALLOC_REQ    = 0x0080 // part of alloc
       ,COMMIT_REQ   = 0x0100 // part of commit
       ,DIRTY_REQ    = 0x0200 // make page dirty wo/ update_lsn
-      ,NO_HOOK      = 0x0400 // dont run load hook
-      ,UNLOCK_PAGE  = 0x0800
-      ,CORR_REQ     = 0x1000 // correlated request (no LIRS update)
+      ,UNLOCK_PAGE  = 0x0400
+      ,CORR_REQ     = 0x0800 // correlated request (no LIRS update)
     };
 
     Uint16 m_block;
@@ -286,6 +284,8 @@ private:
     Uint32 prevList;
   };
 
+  typedef Uint16 Page_state;
+  
   struct Page_entry : Page_entry_stack_ptr,
                       Page_entry_queue_ptr,
                       Page_entry_sublist_ptr {
@@ -305,7 +305,7 @@ private:
       ,PAGEIN  = 0x0100 // paging in
       ,PAGEOUT = 0x0200 // paging out
       ,LOGSYNC = 0x0400 // undo WAL as part of pageout
-      ,NO_HOOK = 0x0800 // don't run load hook
+      ,COPY    = 0x0800 // Copy page for LCP
       ,LCP     = 0x1000 // page is LCP flushed
       ,HOT     = 0x2000 // page is hot
       ,ONSTACK = 0x4000 // page is on LIRS stack
@@ -324,16 +324,16 @@ private:
       ,SUBLIST_COUNT = 8
     };
 
-    Uint16 m_state;             // flags (0 for new entry)
-
     Uint16 m_file_no;       // disk page address set at seize
+    Page_state m_state;         // flags (0 for new entry)
+    
     Uint32 m_page_no;
     Uint32 m_real_page_i;
-    Uint32 m_copy_real_page_i;  // used for flushing LOCKED pages
-    
     Uint64 m_lsn;
-    Uint32 m_last_lcp;
     
+    Uint32 m_last_lcp;
+    Uint32 m_dirty_count;
+    Uint32 m_copy_page_i;
     union {
       Uint32 m_busy_count;        // non-zero means BUSY
       Uint32 nextPool;
@@ -374,8 +374,6 @@ private:
   Uint32 m_last_lcp_complete;
   Uint32 m_lcp_curr_bucket;
   Uint32 m_lcp_outstanding;     // remaining i/o waits
-  Uint32 m_lcp_copy_page;
-  bool m_lcp_copy_page_free;
   EndLcpReq m_end_lcp_req;
   
   // clean-up variables
@@ -421,6 +419,7 @@ protected:
   void execREAD_CONFIG_REQ(Signal* signal);
   void execCONTINUEB(Signal* signal);
 
+  void execLCP_PREPARE_REQ(Signal* signal);
   void execLCP_FRAG_ORD(Signal*);
   void execEND_LCP_REQ(Signal*);
   
@@ -432,8 +431,8 @@ protected:
   void execDUMP_STATE_ORD(Signal* signal);
 
 private:
-  static Uint32 get_sublist_no(Uint16 state);
-  void set_page_state(Ptr<Page_entry> ptr, Uint16 new_state);
+  static Uint32 get_sublist_no(Page_state state);
+  void set_page_state(Ptr<Page_entry> ptr, Page_state new_state);
 
   bool seize_cache_page(Ptr<GlobalPage>& gptr);
   void release_cache_page(Uint32 i);
@@ -463,6 +462,9 @@ private:
   void move_cleanup_ptr(Ptr<Page_entry> ptr);
 
   bool process_lcp(Signal*);
+  void process_lcp_prepare(Signal* signal, Ptr<Page_entry> ptr);
+  int create_copy_page(Ptr<Page_entry>, Uint32 req_flags);
+  void restore_copy_page(Ptr<Page_entry>);
 
   void pagein(Signal*, Ptr<Page_entry>);
   void fsreadreq(Signal*, Ptr<Page_entry>);
@@ -510,13 +512,11 @@ public:
   Ptr<GlobalPage> m_ptr;        // TODO remove
 
   enum RequestFlags {
-    STRICT_ORDER = Pgman::Page_request::STRICT_ORDER
-    ,LOCK_PAGE = Pgman::Page_request::LOCK_PAGE
+    LOCK_PAGE = Pgman::Page_request::LOCK_PAGE
     ,EMPTY_PAGE = Pgman::Page_request::EMPTY_PAGE
     ,ALLOC_REQ = Pgman::Page_request::ALLOC_REQ
     ,COMMIT_REQ = Pgman::Page_request::COMMIT_REQ
     ,DIRTY_REQ = Pgman::Page_request::DIRTY_REQ
-    ,NO_HOOK = Pgman::Page_request::NO_HOOK
     ,UNLOCK_PAGE = Pgman::Page_request::UNLOCK_PAGE
     ,CORR_REQ = Pgman::Page_request::CORR_REQ
   };

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

@@ -195,10 +195,12 @@ private:
   void load_extent_page_callback(Signal*, Uint32, Uint32);
   void create_file_ref(Signal*, Ptr<Tablespace>, Ptr<Datafile>, 
 		       Uint32,Uint32,Uint32);
-  int update_page_free_bits(Signal*, Local_key*, unsigned bits, Uint64 lsn);
-  int get_page_free_bits(Signal*, Local_key*, unsigned* bits);
-  int unmap_page(Signal*, Local_key*);
-  int restart_undo_page_free_bits(Signal*, Local_key*, unsigned, Uint64);
+  int update_page_free_bits(Signal*, Local_key*, unsigned committed_bits, 
+			    Uint64 lsn);
+  int get_page_free_bits(Signal*, Local_key*, unsigned*, unsigned*);
+  int unmap_page(Signal*, Local_key*, unsigned uncommitted_bits);
+  int restart_undo_page_free_bits(Signal*, Uint32, Uint32, Local_key*, 
+				  unsigned committed_bits, Uint64 lsn);
   
   int alloc_extent(Signal* signal, Uint32 tablespace, Local_key* key);
   int alloc_page_from_extent(Signal*, Uint32, Local_key*, Uint32 bits);
@@ -270,12 +272,13 @@ public:
   /**
    * Get page free bits
    */
-  int get_page_free_bits(Local_key*, unsigned* bits);
+  int get_page_free_bits(Local_key*, 
+			 unsigned* uncommitted, unsigned* committed);
   
   /**
    * Update unlogged page free bit
    */
-  int unmap_page(Local_key*);
+  int unmap_page(Local_key*, Uint32 bits);
   
   /**
    * Undo handling of page bits
@@ -353,32 +356,40 @@ Tablespace_client::free_extent(Local_key* key)
 inline
 int
 Tablespace_client::update_page_free_bits(Local_key *key, 
-					 unsigned bits, Uint64 lsn)
+					 unsigned committed_bits,
+					 Uint64 lsn)
 {
-  return m_tsman->update_page_free_bits(m_signal, key, bits, lsn);
+  return m_tsman->update_page_free_bits(m_signal, key, committed_bits, lsn);
 }
 
 inline
 int
-Tablespace_client::get_page_free_bits(Local_key *key, unsigned* bits)
+Tablespace_client::get_page_free_bits(Local_key *key, 
+				      unsigned* uncommited, 
+				      unsigned* commited)
 {
-  return m_tsman->get_page_free_bits(m_signal, key, bits);
+  return m_tsman->get_page_free_bits(m_signal, key, uncommited, commited);
 }
 
 inline
 int
-Tablespace_client::unmap_page(Local_key *key)
+Tablespace_client::unmap_page(Local_key *key, unsigned uncommitted_bits)
 {
-  return m_tsman->unmap_page(m_signal, key);
+  return m_tsman->unmap_page(m_signal, key, uncommitted_bits);
 }
 
 inline
 int 
 Tablespace_client::restart_undo_page_free_bits(Local_key* key, 
-					       unsigned bits, Uint64 lsn)
+					       unsigned committed_bits,
+					       Uint64 lsn)
 {
   return m_tsman->restart_undo_page_free_bits(m_signal,
-					      key, bits, lsn);
+					      m_table_id,
+					      m_fragment_id,
+					      key, 
+					      committed_bits,
+					      lsn);
 }
 
 

storage/ndb/src/kernel/vm/DLFifoList.hpp

@@ -51,6 +51,13 @@ public:
    */
   bool seize(Ptr<T> &);
 
+  /**
+   * Allocate an object from pool - update Ptr - put in front of list
+   *
+   * Return i
+   */
+  bool seizeFront(Ptr<T> &);
+
   /**
    * Allocate object <b>i</b> from pool - update Ptr
    *
@@ -248,6 +255,32 @@ DLFifoList<T,U>::seize(Ptr<T> & p){
   return false;
 }
 
+template <class T, class U>
+inline
+bool
+DLFifoList<T,U>::seizeFront(Ptr<T> & p){
+  Uint32 ff = head.firstItem;
+  thePool.seize(p);
+  if (p.i != RNIL) 
+  {
+    p.p->U::prevList = RNIL;
+    p.p->U::nextList = ff;
+    head.firstItem = p.i;
+    if (ff == RNIL)
+    {
+      head.lastItem = p.i;
+    }
+    else
+    {
+      T * t2 = thePool.getPtr(ff);
+      t2->U::prevList = p.i;
+    }
+    return true;
+  }
+  p.p = NULL;
+  return false;
+}
+
 /**
  * Allocate an object from pool - update Ptr
  *

storage/ndb/test/tools/hugoLoad.cpp

@@ -30,6 +30,7 @@ int main(int argc, const char** argv){
   int _help = 0;
   int _batch = 512;
   int _loops = -1;
+  int _rand = 0;
   const char* db = 0;
 
   struct getargs args[] = {
@@ -37,7 +38,8 @@ int main(int argc, const char** argv){
     { "batch", 'b', arg_integer, &_batch, "Number of operations in each transaction", "batch" },
     { "loops", 'l', arg_integer, &_loops, "Number of loops", "" },
     { "database", 'd', arg_string, &db, "Database", "" },
-    { "usage", '?', arg_flag, &_help, "Print help", "" }
+    { "usage", '?', arg_flag, &_help, "Print help", "" },
+    { "rnd-rows", 0, arg_flag, &_rand, "Rand number of records", "recs" }
   };
   int num_args = sizeof(args) / sizeof(args[0]);
   int optind = 0;
@@ -89,8 +91,9 @@ int main(int argc, const char** argv){
     
     HugoTransactions hugoTrans(*pTab);
 loop:    
+    int rows = (_rand ? rand() % _records : _records);
     if (hugoTrans.loadTable(&MyNdb, 
-			    _records,
+			    rows,
 			    _batch,
 			    true, 0, false, _loops) != 0){
       return NDBT_ProgramExit(NDBT_FAILED);
@@ -98,6 +101,7 @@ loop:
     
     if(_loops > 0)
     {
+      ndbout << "clearing..." << endl;
       hugoTrans.clearTable(&MyNdb);
       //hugoTrans.pkDelRecords(&MyNdb, _records);
       _loops--;

storage/ndb/tools/delete_all.cpp

@@ -130,7 +130,8 @@ int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab, int parallelism)
       goto failed;
     }
     
-    if( pOp->readTuples(NdbOperation::LM_Exclusive,par) ) {
+    if( pOp->readTuples(NdbOperation::LM_Exclusive, 
+			NdbScanOperation::SF_TupScan, par) ) {
       goto failed;
     }