ndb - bug#21948 & bug#17605

  fix alloc/free extent in undo log
  allow extent to be reused once a lcp is finished (instead of when next lcp starts)


storage/ndb/include/kernel/signaldata/Extent.hpp:
  Add lsn to alloc extent
storage/ndb/src/kernel/blocks/diskpage.hpp:
  Add (unused) undo entries for drop table, and alloc/free extent
storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp:
  Notify TSMAN of both start and stop of LCP
storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp:
  1) Add unused undo entries for drop table, alloc/free extent
  2) handle create_table better (correct?) in undo log
  3) fix some typos/style
storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp:
  1) Add unused undo entries for drop table, alloc/free extent
  2) handle create_table better (correct?) in undo log
  3) fix some typos/style
storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp:
  fix style
storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp:
  fix typo/style
storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp:
  Make sure regFragPtr.p->m_logfile_group_id = RNIL is applicable
storage/ndb/src/kernel/blocks/lgman.cpp:
  Add m_logfile_group_id to log callback
storage/ndb/src/kernel/blocks/print_file.cpp:
  Add (unused) undo entries for drop table, and alloc/free extent
storage/ndb/src/kernel/blocks/tsman.cpp:
  1) change so that LCP limit on resuse of extent is only for duration of lcp
  2) refactor so lookup_extent is put into subroutine
storage/ndb/src/kernel/blocks/tsman.hpp:
  refactor so lookup_extent is put into subroutine

12 files changed, 577 insertions, 163 deletions

diff --git a/storage/ndb/include/kernel/signaldata/Extent.hpp b/storage/ndb/include/kernel/signaldata/Extent.hpp
index 86f45163be5..ea31b0968aa 100644
--- a/storage/ndb/include/kernel/signaldata/Extent.hpp
+++ b/storage/ndb/include/kernel/signaldata/Extent.hpp
@@ -74,6 +74,8 @@ struct FreeExtentReq {
       Local_key key;
       Uint32 table_id;
       Uint32 tablespace_id;
+      Uint32 lsn_hi;
+      Uint32 lsn_lo;
     } request;
     struct 
     {
diff --git a/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp b/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
index eff0910760e..1d0d6558e6d 100644
--- a/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
+++ b/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
@@ -11276,7 +11276,7 @@ void Dblqh::execLCP_PREPARE_REF(Signal* signal)
       /**
        * First fragment mean that last LCP is complete :-)
        */
-      EXECUTE_DIRECT(TSMAN, GSN_END_LCP_REQ, signal, signal->length());
+      EXECUTE_DIRECT(TSMAN, GSN_LCP_FRAG_ORD, signal, signal->length());
       jamEntry();
     }
     
@@ -11327,7 +11327,7 @@ void Dblqh::execLCP_PREPARE_CONF(Signal* signal)
       /**
        * First fragment mean that last LCP is complete :-)
        */
-      EXECUTE_DIRECT(TSMAN, GSN_END_LCP_REQ, signal, signal->length());
+      EXECUTE_DIRECT(TSMAN, GSN_LCP_FRAG_ORD, signal, signal->length());
       jamEntry();
     }
     
@@ -11611,6 +11611,9 @@ void Dblqh::completeLcpRoundLab(Signal* signal, Uint32 lcpId)
   sendSignal(LGMAN_REF, GSN_END_LCP_REQ, signal, 
 	     EndLcpReq::SignalLength, JBB);
   
+  EXECUTE_DIRECT(TSMAN, GSN_END_LCP_REQ, signal, EndLcpReq::SignalLength);
+  jamEntry();
+  
   lcpPtr.i = 0;
   ptrAss(lcpPtr, lcpRecord);
   lcpPtr.p->m_outstanding = 3;
diff --git a/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp b/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
index 871c8b88fee..a27daf0a8ea 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
@@ -625,7 +625,8 @@ struct Fragrecord {
   
   DLList<ScanOp>::Head m_scanList;
 
-  bool m_undo_complete;
+  enum { UC_LCP = 1, UC_CREATE = 2 };
+  Uint32 m_undo_complete;
   Uint32 m_tablespace_id;
   Uint32 m_logfile_group_id;
   Disk_alloc_info m_disk_alloc_info;
@@ -989,6 +990,9 @@ ArrayPool<TupTriggerData> c_triggerPool;
       ,UNDO_UPDATE = File_formats::Undofile::UNDO_TUP_UPDATE
       ,UNDO_FREE = File_formats::Undofile::UNDO_TUP_FREE
       ,UNDO_CREATE = File_formats::Undofile::UNDO_TUP_CREATE
+      ,UNDO_DROP = File_formats::Undofile::UNDO_TUP_DROP
+      ,UNDO_ALLOC_EXTENT = File_formats::Undofile::UNDO_TUP_ALLOC_EXTENT
+      ,UNDO_FREE_EXTENT = File_formats::Undofile::UNDO_TUP_FREE_EXTENT
     };
     
     struct Alloc 
@@ -1021,6 +1025,30 @@ ArrayPool<TupTriggerData> c_triggerPool;
       Uint32 m_table;
       Uint32 m_type_length; // 16 bit type, 16 bit length
     };
+
+    struct Drop
+    {
+      Uint32 m_table;
+      Uint32 m_type_length; // 16 bit type, 16 bit length
+    };
+
+    struct AllocExtent
+    {
+      Uint32 m_table;
+      Uint32 m_fragment;
+      Uint32 m_page_no;
+      Uint32 m_file_no;
+      Uint32 m_type_length;
+    };
+
+    struct FreeExtent
+    {
+      Uint32 m_table;
+      Uint32 m_fragment;
+      Uint32 m_page_no;
+      Uint32 m_file_no;
+      Uint32 m_type_length;
+    };
   };
   
   Extent_info_pool c_extent_pool;
@@ -1420,7 +1448,7 @@ public:
   int nr_delete(Signal*, Uint32, Uint32 fragPtr, const Local_key*, Uint32 gci);
 
   void nr_delete_page_callback(Signal*, Uint32 op, Uint32 page);
-  void nr_delete_logbuffer_callback(Signal*, Uint32 op, Uint32 page);
+  void nr_delete_log_buffer_callback(Signal*, Uint32 op, Uint32 page);
 private:
   BLOCK_DEFINES(Dbtup);
 
@@ -2345,9 +2373,10 @@ private:
                         Uint32 fragId);
 
 
-  void releaseFragment(Signal* signal, Uint32 tableId);
+  void releaseFragment(Signal* signal, Uint32 tableId, Uint32);
   void drop_fragment_free_var_pages(Signal*);
-  void drop_fragment_free_exent(Signal*, TablerecPtr, FragrecordPtr, Uint32);
+  void drop_fragment_free_extent(Signal*, TablerecPtr, FragrecordPtr, Uint32);
+  void drop_fragment_free_extent_log_buffer_callback(Signal*, Uint32, Uint32);
   void drop_fragment_unmap_pages(Signal*, TablerecPtr, FragrecordPtr, Uint32);
   void drop_fragment_unmap_page_callback(Signal* signal, Uint32, Uint32);
   
@@ -2631,6 +2660,9 @@ private:
   void disk_page_commit_callback(Signal*, Uint32 opPtrI, Uint32 page_id);  
   
   void disk_page_log_buffer_callback(Signal*, Uint32 opPtrI, Uint32); 
+
+  void disk_page_alloc_extent_log_buffer_callback(Signal*, Uint32, Uint32);
+  void disk_page_free_extent_log_buffer_callback(Signal*, Uint32, Uint32);
   
   Uint64 disk_page_undo_alloc(Page*, const Local_key*,
 			      Uint32 sz, Uint32 gci, Uint32 logfile_group_id);
@@ -2646,6 +2678,9 @@ private:
   void undo_createtable_callback(Signal* signal, Uint32 opPtrI, Uint32 unused);
   void undo_createtable_logsync_callback(Signal* signal, Uint32, Uint32);
 
+  void drop_table_log_buffer_callback(Signal*, Uint32, Uint32);
+  void drop_table_logsync_callback(Signal*, Uint32, Uint32);
+
   void disk_page_set_dirty(Ptr<Page>);
   void restart_setup_page(Disk_alloc_info&, Ptr<Page>);
   void update_extent_pos(Disk_alloc_info&, Ptr<Extent_info>);
@@ -2679,7 +2714,7 @@ public:
   
 private:
   void disk_restart_undo_next(Signal*);
-  void disk_restart_undo_lcp(Uint32, Uint32);
+  void disk_restart_undo_lcp(Uint32, Uint32, Uint32 flag);
   void disk_restart_undo_callback(Signal* signal, Uint32, Uint32);
   void disk_restart_undo_alloc(Apply_undo*);
   void disk_restart_undo_update(Apply_undo*);
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
index 56f7fb1dd1e..069114ba287 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
@@ -68,6 +68,26 @@ operator<<(NdbOut& out, const Ptr<Dbtup::Extent_info> & ptr)
   return out;
 }
 
+#if NOT_YET_FREE_EXTENT
+static
+inline
+bool
+check_free(const Dbtup::Extent_info* extP)
+{
+  Uint32 res = 0;
+  for (Uint32 i = 1; i<MAX_FREE_LIST; i++)
+    res += extP->m_free_page_count[i];
+  return res;
+}
+#error "Code for deallocting extents when they get empty"
+#error "This code is not yet complete"
+#endif
+
+#if NOT_YET_UNDO_ALLOC_EXTENT
+#error "This is needed for deallocting extents when they get empty"
+#error "This code is not complete yet"
+#endif
+
 void 
 Dbtup::dump_disk_alloc(Dbtup::Disk_alloc_info & alloc)
 {
@@ -441,23 +461,71 @@ Dbtup::disk_page_prealloc(Signal* signal,
       /**
        * We need to alloc an extent
        */
+#if NOT_YET_UNDO_ALLOC_EXTENT
+      Uint32 logfile_group_id = fragPtr.p->m_logfile_group_id;
+
+      err = c_lgman->alloc_log_space(logfile_group_id,
+				     sizeof(Disk_undo::AllocExtent)>>2);
+      if(unlikely(err))
+      {
+	return -err;
+      }
+#endif
+
       if (!c_extent_pool.seize(ext))
       {
+	jam();
 	//XXX
 	err= 2;
+#if NOT_YET_UNDO_ALLOC_EXTENT
+	c_lgman->free_log_space(logfile_group_id, 
+				sizeof(Disk_undo::AllocExtent)>>2);
+#endif
 	c_page_request_pool.release(req);
 	ndbout_c("no free extent info");
 	return -err;
       }
-
+      
       if ((err= tsman.alloc_extent(&ext.p->m_key)) < 0)
       {
+	jam();
+#if NOT_YET_UNDO_ALLOC_EXTENT
+	c_lgman->free_log_space(logfile_group_id, 
+				sizeof(Disk_undo::AllocExtent)>>2);
+#endif
 	c_extent_pool.release(ext);
 	c_page_request_pool.release(req);
 	return err;
       }
 
       int pages= err;
+#if NOT_YET_UNDO_ALLOC_EXTENT
+      {
+	/**
+	 * Do something here
+	 */
+	{
+	  Callback cb;
+	  cb.m_callbackData= ext.i;
+	  cb.m_callbackFunction = 
+	    safe_cast(&Dbtup::disk_page_alloc_extent_log_buffer_callback);
+	  Uint32 sz= sizeof(Disk_undo::AllocExtent)>>2;
+	  
+	  Logfile_client lgman(this, c_lgman, logfile_group_id);
+	  int res= lgman.get_log_buffer(signal, sz, &cb);
+	  switch(res){
+	  case 0:
+	    break;
+	  case -1:
+	    ndbrequire("NOT YET IMPLEMENTED" == 0);
+	    break;
+	  default:
+	    execute(signal, cb, res);	    
+	  }
+	}
+      }
+#endif
+      
       ndbout << "allocated " << pages << " pages: " << ext.p->m_key << endl;
       ext.p->m_first_page_no = ext.p->m_key.m_page_no;
       bzero(ext.p->m_free_page_count, sizeof(ext.p->m_free_page_count));
@@ -1007,6 +1075,12 @@ Dbtup::disk_page_free(Signal *signal,
   
   extentPtr.p->m_free_space += sz;
   update_extent_pos(alloc, extentPtr);
+#if NOT_YET_FREE_EXTENT
+  if (check_free(extentPtr.p) == 0)
+  {
+    ndbout_c("free: extent is free");
+  }
+#endif
 }
 
 void
@@ -1104,14 +1178,56 @@ Dbtup::disk_page_abort_prealloc_callback_1(Signal* signal,
 
   extentPtr.p->m_free_space += sz;
   update_extent_pos(alloc, extentPtr);
+#if NOT_YET_FREE_EXTENT
+  if (check_free(extentPtr.p) == 0)
+  {
+    ndbout_c("abort: extent is free");
+  }
+#endif
+}
+
+#if NOT_YET_UNDO_ALLOC_EXTENT
+void
+Dbtup::disk_page_alloc_extent_log_buffer_callback(Signal* signal,
+						  Uint32 extentPtrI,
+						  Uint32 unused)
+{
+  Ptr<Extent_info> extentPtr;
+  c_extent_pool.getPtr(extentPtr, extentPtrI);
+
+  Local_key key = extentPtr.p->m_key;
+  Tablespace_client2 tsman(signal, c_tsman, &key);
+
+  Ptr<Tablerec> tabPtr;
+  tabPtr.i= tsman.m_table_id;
+  ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
+  
+  Ptr<Fragrecord> fragPtr;
+  getFragmentrec(fragPtr, tsman.m_fragment_id, tabPtr.p);
+
+  Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id);
+
+  Disk_undo::AllocExtent alloc;
+  alloc.m_table = tabPtr.i;
+  alloc.m_fragment = tsman.m_fragment_id;
+  alloc.m_page_no = key.m_page_no;
+  alloc.m_file_no = key.m_file_no;
+  alloc.m_type_length = (Disk_undo::UNDO_ALLOC_EXTENT<<16)|(sizeof(alloc)>> 2);
+  
+  Logfile_client::Change c[1] = {{ &alloc, sizeof(alloc) >> 2 } };
+  
+  Uint64 lsn= lgman.add_entry(c, 1);
+  
+  tsman.update_lsn(&key, lsn);
 }
+#endif
 
 Uint64
 Dbtup::disk_page_undo_alloc(Page* page, const Local_key* key,
 			    Uint32 sz, Uint32 gci, Uint32 logfile_group_id)
 {
-  Logfile_client lsman(this, c_lgman, logfile_group_id);
-
+  Logfile_client lgman(this, c_lgman, logfile_group_id);
+  
   Disk_undo::Alloc alloc;
   alloc.m_type_length= (Disk_undo::UNDO_ALLOC << 16) | (sizeof(alloc) >> 2);
   alloc.m_page_no = key->m_page_no;
@@ -1119,7 +1235,7 @@ Dbtup::disk_page_undo_alloc(Page* page, const Local_key* key,
   
   Logfile_client::Change c[1] = {{ &alloc, sizeof(alloc) >> 2 } };
   
-  Uint64 lsn= lsman.add_entry(c, 1);
+  Uint64 lsn= lgman.add_entry(c, 1);
   m_pgman.update_lsn(* key, lsn);
 
   return lsn;
@@ -1130,7 +1246,7 @@ Dbtup::disk_page_undo_update(Page* page, const Local_key* key,
 			     const Uint32* src, Uint32 sz,
 			     Uint32 gci, Uint32 logfile_group_id)
 {
-  Logfile_client lsman(this, c_lgman, logfile_group_id);
+  Logfile_client lgman(this, c_lgman, logfile_group_id);
 
   Disk_undo::Update update;
   update.m_page_no = key->m_page_no;
@@ -1148,7 +1264,7 @@ Dbtup::disk_page_undo_update(Page* page, const Local_key* key,
 
   ndbassert(4*(3 + sz + 1) == (sizeof(update) + 4*sz - 4));
     
-  Uint64 lsn= lsman.add_entry(c, 3);
+  Uint64 lsn= lgman.add_entry(c, 3);
   m_pgman.update_lsn(* key, lsn);
 
   return lsn;
@@ -1159,7 +1275,7 @@ Dbtup::disk_page_undo_free(Page* page, const Local_key* key,
 			   const Uint32* src, Uint32 sz,
 			   Uint32 gci, Uint32 logfile_group_id)
 {
-  Logfile_client lsman(this, c_lgman, logfile_group_id);
+  Logfile_client lgman(this, c_lgman, logfile_group_id);
 
   Disk_undo::Free free;
   free.m_page_no = key->m_page_no;
@@ -1177,7 +1293,7 @@ Dbtup::disk_page_undo_free(Page* page, const Local_key* key,
   
   ndbassert(4*(3 + sz + 1) == (sizeof(free) + 4*sz - 4));
   
-  Uint64 lsn= lsman.add_entry(c, 3);
+  Uint64 lsn= lgman.add_entry(c, 3);
   m_pgman.update_lsn(* key, lsn);
 
   return lsn;
@@ -1207,7 +1323,7 @@ Dbtup::disk_restart_undo(Signal* signal, Uint64 lsn,
     ndbrequire(len == 3);
     Uint32 tableId = ptr[1] >> 16;
     Uint32 fragId = ptr[1] & 0xFFFF;
-    disk_restart_undo_lcp(tableId, fragId);
+    disk_restart_undo_lcp(tableId, fragId, Fragrecord::UC_LCP);
     disk_restart_undo_next(signal);
     return;
   }
@@ -1246,10 +1362,20 @@ Dbtup::disk_restart_undo(Signal* signal, Uint64 lsn,
     ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
     for(Uint32 i = 0; i<MAX_FRAG_PER_NODE; i++)
       if (tabPtr.p->fragrec[i] != RNIL)
-	disk_restart_undo_lcp(tabPtr.i, tabPtr.p->fragid[i]);
+	disk_restart_undo_lcp(tabPtr.i, tabPtr.p->fragid[i], 
+			      Fragrecord::UC_CREATE);
     disk_restart_undo_next(signal);
     return;
   }
+  case File_formats::Undofile::UNDO_TUP_DROP:
+    jam();
+  case File_formats::Undofile::UNDO_TUP_ALLOC_EXTENT:
+    jam();
+  case File_formats::Undofile::UNDO_TUP_FREE_EXTENT:
+    jam();
+    disk_restart_undo_next(signal);
+    return;
+
   case File_formats::Undofile::UNDO_END:
     f_undo_done = true;
     return;
@@ -1283,7 +1409,7 @@ Dbtup::disk_restart_undo_next(Signal* signal)
 }
 
 void
-Dbtup::disk_restart_undo_lcp(Uint32 tableId, Uint32 fragId)
+Dbtup::disk_restart_undo_lcp(Uint32 tableId, Uint32 fragId, Uint32 flag)
 {
   Ptr<Tablerec> tabPtr;
   tabPtr.i= tableId;
@@ -1295,7 +1421,7 @@ Dbtup::disk_restart_undo_lcp(Uint32 tableId, Uint32 fragId)
     getFragmentrec(fragPtr, fragId, tabPtr.p);
     if (!fragPtr.isNull())
     {
-      fragPtr.p->m_undo_complete = true;
+      fragPtr.p->m_undo_complete |= flag;
     }
   }
 }
@@ -1502,6 +1628,12 @@ Dbtup::disk_restart_alloc_extent(Uint32 tableId, Uint32 fragId,
   if (tabPtr.p->tableStatus == DEFINED)
   {
     getFragmentrec(fragPtr, fragId, tabPtr.p);
+    if (fragPtr.p->m_undo_complete & Fragrecord::UC_CREATE)
+    {
+      jam();
+      return -1;
+    }
+
     if (!fragPtr.isNull())
     {
       Disk_alloc_info& alloc= fragPtr.p->m_disk_alloc_info;
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
index f36d522005d..5b9e0d2d74f 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
@@ -3129,7 +3129,7 @@ Dbtup::nr_delete(Signal* signal, Uint32 senderData,
     disk_page_set_dirty(disk_page);
 
     preq.m_callback.m_callbackFunction =
-      safe_cast(&Dbtup::nr_delete_logbuffer_callback);      
+      safe_cast(&Dbtup::nr_delete_log_buffer_callback);      
     Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id);
     res= lgman.get_log_buffer(signal, sz, &preq.m_callback);
     switch(res){
@@ -3182,7 +3182,7 @@ Dbtup::nr_delete_page_callback(Signal* signal,
   Callback cb;
   cb.m_callbackData = userpointer;
   cb.m_callbackFunction =
-    safe_cast(&Dbtup::nr_delete_logbuffer_callback);      
+    safe_cast(&Dbtup::nr_delete_log_buffer_callback);      
   Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id);
   int res= lgman.get_log_buffer(signal, sz, &cb);
   switch(res){
@@ -3202,7 +3202,7 @@ Dbtup::nr_delete_page_callback(Signal* signal,
 }
 
 void
-Dbtup::nr_delete_logbuffer_callback(Signal* signal, 
+Dbtup::nr_delete_log_buffer_callback(Signal* signal, 
 				    Uint32 userpointer, 
 				    Uint32 unused)
 {
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp
index fff557583be..e0f0fd84ee3 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp
@@ -164,7 +164,7 @@ void Dbtup::execCONTINUEB(Signal* signal)
     break;
   case ZREL_FRAG:
     ljam();
-    releaseFragment(signal, dataPtr);
+    releaseFragment(signal, dataPtr, signal->theData[2]);
     break;
   case ZREPORT_MEMORY_USAGE:{
     ljam();
@@ -223,7 +223,7 @@ void Dbtup::execCONTINUEB(Signal* signal)
     fragPtr.i= signal->theData[2];
     ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
     ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
-    drop_fragment_free_exent(signal, tabPtr, fragPtr, signal->theData[3]);
+    drop_fragment_free_extent(signal, tabPtr, fragPtr, signal->theData[3]);
     return;
   }
   case ZUNMAP_PAGES:
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
index 8db8f423817..f6f7cef2a5e 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
@@ -309,6 +309,7 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
     
     if(lastAttr)
     {
+      jam();
       /**
        * Init Disk_alloc_info
        */
@@ -320,6 +321,11 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
 	ndbrequire(tsman.get_tablespace_info(&rep) == 0);
 	regFragPtr.p->m_logfile_group_id= rep.tablespace.logfile_group_id;
       }
+      else
+      {
+	jam();
+	regFragPtr.p->m_logfile_group_id = RNIL;
+      }
       new (&regFragPtr.p->m_disk_alloc_info)
 	Disk_alloc_info(regTabPtr.p, rep.tablespace.extent_size);
       releaseFragoperrec(fragOperPtr);      
@@ -564,7 +570,12 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
 			    regFragPtr.p->m_tablespace_id);
     ndbrequire(tsman.get_tablespace_info(&rep) == 0);
     regFragPtr.p->m_logfile_group_id= rep.tablespace.logfile_group_id;
-  }    
+  }
+  else
+  {
+    jam();
+    regFragPtr.p->m_logfile_group_id = RNIL;
+  }
 
   new (&regFragPtr.p->m_disk_alloc_info)
     Disk_alloc_info(regTabPtr.p, rep.tablespace.extent_size); 
@@ -597,7 +608,7 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
 	ndbrequire("NOT YET IMPLEMENTED" == 0);
 	break;
       }
-      execute(signal, cb, 0);
+      execute(signal, cb, regFragPtr.p->m_logfile_group_id);
       return;
     }
   }
@@ -874,7 +885,8 @@ Dbtup::execDROP_TAB_REQ(Signal* signal)
 
   signal->theData[0]= ZREL_FRAG;
   signal->theData[1]= tabPtr.i;
-  sendSignal(cownref, GSN_CONTINUEB, signal, 2, JBB);
+  signal->theData[2]= RNIL;
+  sendSignal(cownref, GSN_CONTINUEB, signal, 3, JBB);
 }
 
 void Dbtup::releaseTabDescr(Tablerec* const regTabPtr) 
@@ -902,7 +914,8 @@ void Dbtup::releaseTabDescr(Tablerec* const regTabPtr)
   }
 }
 
-void Dbtup::releaseFragment(Signal* signal, Uint32 tableId)
+void Dbtup::releaseFragment(Signal* signal, Uint32 tableId, 
+			    Uint32 logfile_group_id)
 {
   TablerecPtr tabPtr;
   tabPtr.i= tableId;
@@ -929,16 +942,35 @@ void Dbtup::releaseFragment(Signal* signal, Uint32 tableId)
     sendSignal(cownref, GSN_CONTINUEB, signal, 4, JBB);  
     return;
   }
+
+#if NOT_YET_UNDO_DROP_TABLE
+#error "This code is complete, but I prefer not to enable it until I need it"
+  if (logfile_group_id != RNIL)
+  {
+    Callback cb;
+    cb.m_callbackData= tabPtr.i;
+    cb.m_callbackFunction = 
+      safe_cast(&Dbtup::drop_table_log_buffer_callback);
+    Uint32 sz= sizeof(Disk_undo::Drop) >> 2;
+    int r0 = c_lgman->alloc_log_space(logfile_group_id, sz);
+    
+    Logfile_client lgman(this, c_lgman, logfile_group_id);
+    int res= lgman.get_log_buffer(signal, sz, &cb);
+    switch(res){
+    case 0:
+      ljam();
+      return;
+    case -1:
+      ndbrequire("NOT YET IMPLEMENTED" == 0);
+      break;
+    default:
+      execute(signal, cb, logfile_group_id);
+      return;
+    }
+  }
+#endif
   
-  DropTabConf * const dropConf= (DropTabConf *)signal->getDataPtrSend();
-  dropConf->senderRef= reference();
-  dropConf->senderData= tabPtr.p->m_dropTable.tabUserPtr;
-  dropConf->tableId= tabPtr.i;
-  sendSignal(tabPtr.p->m_dropTable.tabUserRef, GSN_DROP_TAB_CONF,
-             signal, DropTabConf::SignalLength, JBB);
-  
-  releaseTabDescr(tabPtr.p);
-  initTab(tabPtr.p);
+  drop_table_logsync_callback(signal, tabPtr.i, RNIL);
 }
 
 void
@@ -965,7 +997,7 @@ Dbtup::drop_fragment_unmap_pages(Signal *signal,
 	alloc_info.m_curr_extent_info_ptr_i= RNIL;
       }
       
-      drop_fragment_free_exent(signal, tabPtr, fragPtr, 0);
+      drop_fragment_free_extent(signal, tabPtr, fragPtr, 0);
       return;
     }
     
@@ -998,7 +1030,7 @@ Dbtup::drop_fragment_unmap_pages(Signal *signal,
     }
     return;
   }
-  drop_fragment_free_exent(signal, tabPtr, fragPtr, 0);  
+  drop_fragment_free_extent(signal, tabPtr, fragPtr, 0);  
 }
 
 void
@@ -1031,10 +1063,10 @@ Dbtup::drop_fragment_unmap_page_callback(Signal* signal,
 }
 
 void
-Dbtup::drop_fragment_free_exent(Signal *signal, 
-				TablerecPtr tabPtr, 
-				FragrecordPtr fragPtr,
-				Uint32 pos)
+Dbtup::drop_fragment_free_extent(Signal *signal, 
+				 TablerecPtr tabPtr, 
+				 FragrecordPtr fragPtr,
+				 Uint32 pos)
 {
   if (tabPtr.p->m_no_of_disk_attributes)
   {
@@ -1044,25 +1076,32 @@ Dbtup::drop_fragment_free_exent(Signal *signal,
       if(!alloc_info.m_free_extents[pos].isEmpty())
       {
 	jam();
-	Local_extent_info_list
-	  list(c_extent_pool, alloc_info.m_free_extents[pos]);
-	Ptr<Extent_info> ext_ptr;
-	list.first(ext_ptr);
+	Callback cb;
+	cb.m_callbackData= fragPtr.i;
+	cb.m_callbackFunction = 
+	  safe_cast(&Dbtup::drop_fragment_free_extent_log_buffer_callback);
+#if NOT_YET_UNDO_FREE_EXTENT
+	Uint32 sz= sizeof(Disk_undo::FreeExtent) >> 2;
+	int r0 = c_lgman->alloc_log_space(fragPtr.p->m_logfile_group_id, sz);
 	
-	Tablespace_client tsman(signal, c_tsman, tabPtr.i, 
-				fragPtr.p->fragmentId,
-				fragPtr.p->m_tablespace_id);
+	Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id);
 	
-	tsman.free_extent(&ext_ptr.p->m_key);
-	c_extent_hash.remove(ext_ptr);
-	list.release(ext_ptr);
-	
-	signal->theData[0] = ZFREE_EXTENT;
-	signal->theData[1] = tabPtr.i;
-	signal->theData[2] = fragPtr.i;
-	signal->theData[3] = pos;
-	sendSignal(cownref, GSN_CONTINUEB, signal, 4, JBB);  
+	int res= lgman.get_log_buffer(signal, sz, &cb);
+	switch(res){
+	case 0:
+	  ljam();
+	  return;
+	case -1:
+	  ndbrequire("NOT YET IMPLEMENTED" == 0);
+	  break;
+	default:
+	  execute(signal, cb, fragPtr.p->m_logfile_group_id);
+	  return;
+	}
+#else
+	execute(signal, cb, fragPtr.p->m_logfile_group_id);	
 	return;
+#endif
       }
     }
     
@@ -1082,6 +1121,123 @@ Dbtup::drop_fragment_free_exent(Signal *signal,
 }
 
 void
+Dbtup::drop_table_log_buffer_callback(Signal* signal, Uint32 tablePtrI,
+				      Uint32 logfile_group_id)
+{
+  TablerecPtr tabPtr;
+  tabPtr.i = tablePtrI;
+  ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
+  
+  ndbrequire(tabPtr.p->m_no_of_disk_attributes);
+
+  Disk_undo::Drop drop;
+  drop.m_table = tabPtr.i;
+  drop.m_type_length = 
+    (Disk_undo::UNDO_DROP << 16) | (sizeof(drop) >> 2);
+  Logfile_client lgman(this, c_lgman, logfile_group_id);
+  
+  Logfile_client::Change c[1] = {{ &drop, sizeof(drop) >> 2 } };
+  Uint64 lsn = lgman.add_entry(c, 1);
+
+  Logfile_client::Request req;
+  req.m_callback.m_callbackData= tablePtrI;
+  req.m_callback.m_callbackFunction = 
+    safe_cast(&Dbtup::drop_table_logsync_callback);
+  
+  int ret = lgman.sync_lsn(signal, lsn, &req, 0);
+  switch(ret){
+  case 0:
+    return;
+  default:
+    ndbout_c("ret: %d", ret);
+    ndbrequire(false);
+  }
+}
+
+void
+Dbtup::drop_table_logsync_callback(Signal* signal, 
+				   Uint32 tabPtrI, 
+				   Uint32 logfile_group_id)
+{
+  TablerecPtr tabPtr;
+  tabPtr.i = tabPtrI;
+  ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
+  
+  DropTabConf * const dropConf= (DropTabConf *)signal->getDataPtrSend();
+  dropConf->senderRef= reference();
+  dropConf->senderData= tabPtr.p->m_dropTable.tabUserPtr;
+  dropConf->tableId= tabPtr.i;
+  sendSignal(tabPtr.p->m_dropTable.tabUserRef, GSN_DROP_TAB_CONF,
+             signal, DropTabConf::SignalLength, JBB);
+  
+  releaseTabDescr(tabPtr.p);
+  initTab(tabPtr.p);
+}
+
+void
+Dbtup::drop_fragment_free_extent_log_buffer_callback(Signal* signal,
+						     Uint32 fragPtrI,
+						     Uint32 unused)
+{
+  FragrecordPtr fragPtr;
+  fragPtr.i = fragPtrI;
+  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
+
+  TablerecPtr tabPtr;
+  tabPtr.i = fragPtr.p->fragTableId;
+  ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
+
+  ndbrequire(tabPtr.p->m_no_of_disk_attributes);
+  Disk_alloc_info& alloc_info= fragPtr.p->m_disk_alloc_info;  
+
+  for(Uint32 pos = 0; pos<EXTENT_SEARCH_MATRIX_SIZE; pos++)
+  {
+    if(!alloc_info.m_free_extents[pos].isEmpty())
+    {
+      jam();
+      Local_extent_info_list
+	list(c_extent_pool, alloc_info.m_free_extents[pos]);
+      Ptr<Extent_info> ext_ptr;
+      list.first(ext_ptr);
+
+#if NOT_YET_UNDO_FREE_EXTENT
+#error "This code is complete"
+#error "but not needed until we do dealloc of empty extents"
+      Disk_undo::FreeExtent free;
+      free.m_table = tabPtr.i;
+      free.m_fragment = fragPtr.p->fragmentId;
+      free.m_file_no = ext_ptr.p->m_key.m_file_no;
+      free.m_page_no = ext_ptr.p->m_key.m_page_no;
+      free.m_type_length = 
+	(Disk_undo::UNDO_FREE_EXTENT << 16) | (sizeof(free) >> 2);
+      Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id);
+      
+      Logfile_client::Change c[1] = {{ &free, sizeof(free) >> 2 } };
+      Uint64 lsn = lgman.add_entry(c, 1);
+#else
+      Uint64 lsn = 0;
+#endif
+      
+      Tablespace_client tsman(signal, c_tsman, tabPtr.i, 
+			      fragPtr.p->fragmentId,
+			      fragPtr.p->m_tablespace_id);
+      
+      tsman.free_extent(&ext_ptr.p->m_key, lsn);
+      c_extent_hash.remove(ext_ptr);
+      list.release(ext_ptr);
+      
+      signal->theData[0] = ZFREE_EXTENT;
+      signal->theData[1] = tabPtr.i;
+      signal->theData[2] = fragPtr.i;
+      signal->theData[3] = pos;
+      sendSignal(cownref, GSN_CONTINUEB, signal, 4, JBB);  
+      return;
+    }
+  }
+  ndbrequire(false);
+}
+
+void
 Dbtup::drop_fragment_free_var_pages(Signal* signal)
 {
   ljam();
@@ -1112,7 +1268,7 @@ Dbtup::drop_fragment_free_var_pages(Signal* signal)
     sendSignal(cownref, GSN_CONTINUEB, signal, 3, JBB);  
     return;
   }
-  
+  Uint32 logfile_group_id = fragPtr.p->m_logfile_group_id ;
   releaseFragPages(fragPtr.p);
   Uint32 i;
   for(i= 0; i<MAX_FRAG_PER_NODE; i++)
@@ -1126,7 +1282,8 @@ Dbtup::drop_fragment_free_var_pages(Signal* signal)
   
   signal->theData[0]= ZREL_FRAG;
   signal->theData[1]= tabPtr.i;
-  sendSignal(cownref, GSN_CONTINUEB, signal, 2, JBB);  
+  signal->theData[2]= logfile_group_id;
+  sendSignal(cownref, GSN_CONTINUEB, signal, 3, JBB);  
   return;
 }
 
diff --git a/storage/ndb/src/kernel/blocks/diskpage.hpp b/storage/ndb/src/kernel/blocks/diskpage.hpp
index 044597be1cf..c75f2a0c070 100644
--- a/storage/ndb/src/kernel/blocks/diskpage.hpp
+++ b/storage/ndb/src/kernel/blocks/diskpage.hpp
@@ -154,7 +154,10 @@ struct File_formats
       ,UNDO_TUP_UPDATE = 4
       ,UNDO_TUP_FREE   = 5
       ,UNDO_TUP_CREATE = 6
-
+      ,UNDO_TUP_DROP   = 7
+      ,UNDO_TUP_ALLOC_EXTENT = 8
+      ,UNDO_TUP_FREE_EXTENT  = 9
+      
       ,UNDO_END        = 0x7FFF 
       ,UNDO_NEXT_LSN   = 0x8000
     };
diff --git a/storage/ndb/src/kernel/blocks/lgman.cpp b/storage/ndb/src/kernel/blocks/lgman.cpp
index d4cc0cb89e8..a71b817944f 100644
--- a/storage/ndb/src/kernel/blocks/lgman.cpp
+++ b/storage/ndb/src/kernel/blocks/lgman.cpp
@@ -1160,14 +1160,15 @@ Lgman::process_log_sync_waiters(Signal* signal, Ptr<Logfile_group> ptr)
   bool removed= false;
   Ptr<Log_waiter> waiter;
   list.first(waiter);
+  Uint32 logfile_group_id = ptr.p->m_logfile_group_id;
 
   if(waiter.p->m_sync_lsn <= ptr.p->m_last_synced_lsn)
   {
     removed= true;
     Uint32 block = waiter.p->m_block;
     SimulatedBlock* b = globalData.getBlock(block);
-    b->execute(signal, waiter.p->m_callback, 0);
-
+    b->execute(signal, waiter.p->m_callback, logfile_group_id);
+    
     list.releaseFirst(waiter);
   }
   
@@ -1522,12 +1523,13 @@ Lgman::process_log_buffer_waiters(Signal* signal, Ptr<Logfile_group> ptr)
   bool removed= false;
   Ptr<Log_waiter> waiter;
   list.first(waiter);
+  Uint32 logfile_group_id = ptr.p->m_logfile_group_id;
   if(waiter.p->m_size + 2*File_formats::UNDO_PAGE_WORDS < free_buffer)
   {
     removed= true;
     Uint32 block = waiter.p->m_block;
     SimulatedBlock* b = globalData.getBlock(block);
-    b->execute(signal, waiter.p->m_callback, 0);
+    b->execute(signal, waiter.p->m_callback, logfile_group_id);
 
     list.releaseFirst(waiter);
   }
@@ -2699,6 +2701,9 @@ Lgman::execute_undo_record(Signal* signal)
     case File_formats::Undofile::UNDO_TUP_UPDATE:
     case File_formats::Undofile::UNDO_TUP_FREE:
     case File_formats::Undofile::UNDO_TUP_CREATE:
+    case File_formats::Undofile::UNDO_TUP_DROP:
+    case File_formats::Undofile::UNDO_TUP_ALLOC_EXTENT:
+    case File_formats::Undofile::UNDO_TUP_FREE_EXTENT:
       tup->disk_restart_undo(signal, lsn, mask, ptr - len + 1, len);
       return;
     default:
diff --git a/storage/ndb/src/kernel/blocks/print_file.cpp b/storage/ndb/src/kernel/blocks/print_file.cpp
index 14b83cccaee..696658ef8de 100644
--- a/storage/ndb/src/kernel/blocks/print_file.cpp
+++ b/storage/ndb/src/kernel/blocks/print_file.cpp
@@ -304,6 +304,8 @@ print_undo_page(int count, void* ptr, Uint32 sz){
 	case File_formats::Undofile::UNDO_LCP:
 	  printf("[ %lld LCP %d tab: %d frag: %d ]", lsn, 
 		 src[0], src[1] >> 16, src[1] & 0xFFFF);
+	  if(g_verbosity <= 3)
+	    printf("\n");
 	  break;
 	case File_formats::Undofile::UNDO_TUP_ALLOC:
 	  if(g_verbosity > 3)
@@ -340,6 +342,48 @@ print_undo_page(int count, void* ptr, Uint32 sz){
 		   req->m_gci);
 	  }
 	  break;
+	case File_formats::Undofile::UNDO_TUP_CREATE:
+	{
+	  Dbtup::Disk_undo::Create *req = (Dbtup::Disk_undo::Create*)src;
+	  printf("[ %lld Create %d ]", lsn, req->m_table);
+	  if(g_verbosity <= 3)
+	    printf("\n");
+	  break;
+	}
+	case File_formats::Undofile::UNDO_TUP_DROP:
+	{
+	  Dbtup::Disk_undo::Drop *req = (Dbtup::Disk_undo::Drop*)src;
+	  printf("[ %lld Drop %d ]", lsn, req->m_table);
+	  if(g_verbosity <= 3)
+	    printf("\n");
+	  break;
+	}
+	case File_formats::Undofile::UNDO_TUP_ALLOC_EXTENT:
+	{
+	  Dbtup::Disk_undo::AllocExtent *req = (Dbtup::Disk_undo::AllocExtent*)src;
+	  printf("[ %lld AllocExtent tab: %d frag: %d file: %d page: %d ]", 
+		 lsn, 
+		 req->m_table,
+		 req->m_fragment,
+		 req->m_file_no,
+		 req->m_page_no);
+	  if(g_verbosity <= 3)
+	    printf("\n");
+	  break;
+	}
+	case File_formats::Undofile::UNDO_TUP_FREE_EXTENT:
+	{
+	  Dbtup::Disk_undo::FreeExtent *req = (Dbtup::Disk_undo::FreeExtent*)src;
+	  printf("[ %lld FreeExtent tab: %d frag: %d file: %d page: %d ]", 
+		 lsn, 
+		 req->m_table,
+		 req->m_fragment,
+		 req->m_file_no,
+		 req->m_page_no);
+	  if(g_verbosity <= 3)
+	    printf("\n");
+	  break;
+	}
 	default:
 	  ndbout_c("[ Unknown type %d len: %d, pos: %d ]", type, len, pos);
 	  if(!(len && type))
@@ -361,11 +405,11 @@ print_undo_page(int count, void* ptr, Uint32 sz){
       }
     }
   }
-
+  
   if(count == g_uf_zero.m_undo_pages + 1)
   {
   }
-
+  
   return 0;
 }
 
diff --git a/storage/ndb/src/kernel/blocks/tsman.cpp b/storage/ndb/src/kernel/blocks/tsman.cpp
index 89725ca309e..3c1ddac714c 100644
--- a/storage/ndb/src/kernel/blocks/tsman.cpp
+++ b/storage/ndb/src/kernel/blocks/tsman.cpp
@@ -79,12 +79,14 @@ Tsman::Tsman(Block_context& ctx,
   
   addRecSignal(GSN_START_RECREQ, &Tsman::execSTART_RECREQ);
 
+  addRecSignal(GSN_LCP_FRAG_ORD, &Tsman::execLCP_FRAG_ORD);
   addRecSignal(GSN_END_LCP_REQ, &Tsman::execEND_LCP_REQ);
 
   addRecSignal(GSN_GET_TABINFOREQ, &Tsman::execGET_TABINFOREQ);
 
   m_tablespace_hash.setSize(10);
   m_file_hash.setSize(10);
+  m_lcp_ongoing = false;
 }
   
 Tsman::~Tsman()
@@ -1101,6 +1103,7 @@ Tsman::load_extent_page_callback(Signal* signal,
   ptr.p->m_online.m_lcp_free_extent_tail = RNIL;  
   ptr.p->m_online.m_data_pages = data_pages;
   ptr.p->m_online.m_used_extent_cnt = 0;
+  ptr.p->m_online.m_extent_headers_per_extent_page = per_page;
 
   Ptr<Tablespace> ts_ptr;
   m_tablespace_pool.getPtr(ts_ptr, ptr.p->m_tablespace_ptr_i);
@@ -1182,9 +1185,8 @@ Tsman::scan_extent_headers(Signal* signal, Ptr<Datafile> ptr)
   m_tablespace_pool.getPtr(lg_ptr, ptr.p->m_tablespace_ptr_i);
 
   Uint32 firstFree= RNIL;
-  Uint32 size = lg_ptr.p->m_extent_size;
-  Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
-  Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
+  Uint32 size = ptr.p->m_extent_size;
+  Uint32 per_page = ptr.p->m_online.m_extent_headers_per_extent_page;
   Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;  
   Uint32 pages= ptr.p->m_online.m_offset_data_pages - 1;
   Uint32 datapages= ptr.p->m_online.m_data_pages;
@@ -1410,23 +1412,21 @@ Tsman::execALLOC_EXTENT_REQ(Signal* signal)
   AllocExtentReq::ErrorCode err;
   
   ndbrequire(m_tablespace_hash.find(ts_ptr, req.request.tablespace_id));
-  Uint32 size = ts_ptr.p->m_extent_size;
   Local_datafile_list tmp(m_file_pool, ts_ptr.p->m_free_files);
   
   if (tmp.first(file_ptr))
   {
-    Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
-    Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
-
+    Uint32 size = file_ptr.p->m_extent_size;
     Uint32 extent = file_ptr.p->m_online.m_first_free_extent;
     Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
+    Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
+    Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
     Uint32 page_no = extent / per_page;
     Uint32 extent_no = extent % per_page;
 
     Page_cache_client::Request preq;
     preq.m_page.m_page_no = page_no;
     preq.m_page.m_file_no = file_ptr.p->m_file_no;
-    preq.m_page.m_page_idx = extent;
 
     /**
      * Handling of unmapped extent header pages is not implemented
@@ -1501,28 +1501,21 @@ void
 Tsman::execFREE_EXTENT_REQ(Signal* signal)
 {
   jamEntry();
-  Ptr<Tablespace> ts_ptr;
   Ptr<Datafile> file_ptr;
   FreeExtentReq req = *(FreeExtentReq*)signal->getDataPtr();
   FreeExtentReq::ErrorCode err = (FreeExtentReq::ErrorCode)0;
   
-  ndbrequire(m_tablespace_hash.find(ts_ptr, req.request.tablespace_id));
   Datafile file_key;
   file_key.m_file_no = req.request.key.m_file_no;
   ndbrequire(m_file_hash.find(file_ptr, file_key));
-  Uint32 size = ts_ptr.p->m_extent_size;
-
-  Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
-  Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
-  Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
-  Uint32 extent = (req.request.key.m_page_no - data_off) / size + per_page;
 
-
-  Uint32 page_no = extent / per_page;
-  Uint32 extent_no = extent % per_page;
+  struct req val = lookup_extent(req.request.key.m_page_no, file_ptr.p);
+  Uint32 extent = 
+    (req.request.key.m_page_no - val.m_extent_pages) / val.m_extent_size + 
+    file_ptr.p->m_online.m_extent_headers_per_extent_page;
   
   Page_cache_client::Request preq;
-  preq.m_page.m_page_no = page_no;
+  preq.m_page.m_page_no = val.m_extent_page_no;
   preq.m_page.m_file_no = req.request.key.m_file_no;
 
   ndbout << "Free extent: " << req.request.key << endl;
@@ -1539,16 +1532,38 @@ Tsman::execFREE_EXTENT_REQ(Signal* signal)
     File_formats::Datafile::Extent_page* page = 
       (File_formats::Datafile::Extent_page*)ptr_p;
     File_formats::Datafile::Extent_header* header = 
-      page->get_header(extent_no, size);
+      page->get_header(val.m_extent_no, val.m_extent_size);
     
     ndbrequire(header->m_table == req.request.table_id);
     header->m_table = RNIL;
-    header->m_next_free_extent= file_ptr.p->m_online.m_lcp_free_extent_head;
         
-    if(file_ptr.p->m_online.m_lcp_free_extent_head == RNIL)
-      file_ptr.p->m_online.m_lcp_free_extent_tail= extent;
-    file_ptr.p->m_online.m_lcp_free_extent_head= extent;
     file_ptr.p->m_online.m_used_extent_cnt--;
+    if (m_lcp_ongoing)
+    {
+      jam();
+      header->m_next_free_extent= file_ptr.p->m_online.m_lcp_free_extent_head;
+      if(file_ptr.p->m_online.m_lcp_free_extent_head == RNIL)
+	file_ptr.p->m_online.m_lcp_free_extent_tail= extent;
+      file_ptr.p->m_online.m_lcp_free_extent_head= extent;
+    }
+    else
+    {
+      jam();
+      header->m_next_free_extent = file_ptr.p->m_online.m_first_free_extent;
+      if (file_ptr.p->m_online.m_first_free_extent == RNIL)
+      {
+	/**
+	 * Move from full to free
+	 */
+	Ptr<Tablespace> ptr;
+	m_tablespace_pool.getPtr(ptr, file_ptr.p->m_tablespace_ptr_i);
+	Local_datafile_list free(m_file_pool, ptr.p->m_free_files);
+	Local_datafile_list full(m_file_pool, ptr.p->m_full_files);
+	full.remove(file_ptr);
+	free.add(file_ptr);
+      }
+      file_ptr.p->m_online.m_first_free_extent = extent;
+    }
   }
   else
   {
@@ -1583,18 +1598,10 @@ Tsman::update_page_free_bits(Signal* signal,
   file_key.m_file_no = key->m_file_no;
   ndbrequire(m_file_hash.find(file_ptr, file_key));
 
-  Uint32 size = file_ptr.p->m_extent_size;
-  Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
-  Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
-  Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
-  Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
-  
-  Uint32 extent = (key->m_page_no - data_off) / size + per_page;
-  Uint32 page_no = extent / per_page;
-  Uint32 extent_no = extent % per_page;
+  struct req val = lookup_extent(key->m_page_no, file_ptr.p);
   
   Page_cache_client::Request preq;
-  preq.m_page.m_page_no = page_no;
+  preq.m_page.m_page_no = val.m_extent_page_no;
   preq.m_page.m_file_no = key->m_file_no;
   
   /**
@@ -1609,12 +1616,12 @@ Tsman::update_page_free_bits(Signal* signal,
     File_formats::Datafile::Extent_page* page = 
       (File_formats::Datafile::Extent_page*)ptr_p;
     File_formats::Datafile::Extent_header* header = 
-      page->get_header(extent_no, size);
+      page->get_header(val.m_extent_no, val.m_extent_size);
     
     ndbrequire(header->m_table != RNIL);
 
-    Uint32 page_no_in_extent = (key->m_page_no - data_off) % size;
-
+    Uint32 page_no_in_extent = calc_page_no_in_extent(key->m_page_no, &val);
+    
     /**
      * Toggle word
      */
@@ -1637,26 +1644,15 @@ Tsman::get_page_free_bits(Signal* signal, Local_key *key,
 {
   jamEntry();
 
-  /**
-   * XXX make into subroutine
-   */   
   Ptr<Datafile> file_ptr;
   Datafile file_key;
   file_key.m_file_no = key->m_file_no;
   ndbrequire(m_file_hash.find(file_ptr, file_key));
-
-  Uint32 size = file_ptr.p->m_extent_size;
-  Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
-  Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
-  Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
-  Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
   
-  Uint32 extent = (key->m_page_no - data_off) / size + per_page;
-  Uint32 page_no = extent / per_page;
-  Uint32 extent_no = extent % per_page;
+  struct req val = lookup_extent(key->m_page_no, file_ptr.p);
   
   Page_cache_client::Request preq;
-  preq.m_page.m_page_no = page_no;
+  preq.m_page.m_page_no = val.m_extent_page_no;
   preq.m_page.m_file_no = key->m_file_no;
   
   /**
@@ -1671,11 +1667,11 @@ Tsman::get_page_free_bits(Signal* signal, Local_key *key,
     File_formats::Datafile::Extent_page* page = 
       (File_formats::Datafile::Extent_page*)ptr_p;
     File_formats::Datafile::Extent_header* header = 
-      page->get_header(extent_no, size);
+      page->get_header(val.m_extent_no, val.m_extent_size);
     
     ndbrequire(header->m_table != RNIL);
 
-    Uint32 page_no_in_extent = (key->m_page_no - data_off) % size;
+    Uint32 page_no_in_extent = calc_page_no_in_extent(key->m_page_no, &val);
     Uint32 bits = header->get_free_bits(page_no_in_extent);
     *uncommitted = (bits & UNCOMMITTED_MASK) >> UNCOMMITTED_SHIFT;
     *committed = (bits & COMMITTED_MASK);
@@ -1700,19 +1696,11 @@ Tsman::unmap_page(Signal* signal, Local_key *key, Uint32 uncommitted_bits)
   Datafile file_key;
   file_key.m_file_no = key->m_file_no;
   ndbrequire(m_file_hash.find(file_ptr, file_key));
-  
-  Uint32 size = file_ptr.p->m_extent_size;
-  Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
-  Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
-  Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
-  Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
-  
-  Uint32 extent = (key->m_page_no - data_off) / size + per_page;
-  Uint32 page_no = extent / per_page;
-  Uint32 extent_no = extent % per_page;
+
+  struct req val = lookup_extent(key->m_page_no, file_ptr.p);
   
   Page_cache_client::Request preq;
-  preq.m_page.m_page_no = page_no;
+  preq.m_page.m_page_no = val.m_extent_page_no;
   preq.m_page.m_file_no = key->m_file_no;
   
   /**
@@ -1727,12 +1715,12 @@ Tsman::unmap_page(Signal* signal, Local_key *key, Uint32 uncommitted_bits)
     File_formats::Datafile::Extent_page* page = 
       (File_formats::Datafile::Extent_page*)ptr_p;
     File_formats::Datafile::Extent_header* header = 
-      page->get_header(extent_no, size);
+      page->get_header(val.m_extent_no, val.m_extent_size);
     
     ndbrequire(header->m_table != RNIL);
 
-    Uint32 page_no_in_extent = (key->m_page_no - data_off) % size;
-
+    Uint32 page_no_in_extent = calc_page_no_in_extent(key->m_page_no, &val);
+    
     /**
      * Toggle word
      */
@@ -1767,18 +1755,10 @@ Tsman::restart_undo_page_free_bits(Signal* signal,
   file_key.m_file_no = key->m_file_no;
   ndbrequire(m_file_hash.find(file_ptr, file_key));
 
-  Uint32 size = file_ptr.p->m_extent_size;
-  Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
-  Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
-  Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
-  Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
-  
-  Uint32 extent = (key->m_page_no - data_off) / size + per_page;
-  Uint32 page_no = extent / per_page;
-  Uint32 extent_no = extent % per_page;
-  
+  struct req val = lookup_extent(key->m_page_no, file_ptr.p);
+
   Page_cache_client::Request preq;
-  preq.m_page.m_page_no = page_no;
+  preq.m_page.m_page_no = val.m_extent_page_no;
   preq.m_page.m_file_no = key->m_file_no;
   
   /**
@@ -1793,7 +1773,7 @@ Tsman::restart_undo_page_free_bits(Signal* signal,
     File_formats::Datafile::Extent_page* page = 
       (File_formats::Datafile::Extent_page*)ptr_p;
     File_formats::Datafile::Extent_header* header = 
-      page->get_header(extent_no, size);
+      page->get_header(val.m_extent_no, val.m_extent_size);
     
     Uint64 lsn = 0;
     lsn += page->m_page_header.m_page_lsn_hi; lsn <<= 32;
@@ -1816,7 +1796,7 @@ Tsman::restart_undo_page_free_bits(Signal* signal,
       return 0;
     }
 
-    Uint32 page_no_in_extent = (key->m_page_no - data_off) % size;
+    Uint32 page_no_in_extent = calc_page_no_in_extent(key->m_page_no, &val);
     Uint32 src = header->get_free_bits(page_no_in_extent);
         
     ndbrequire(header->m_table == tableId);
@@ -1862,17 +1842,11 @@ Tsman::execALLOC_PAGE_REQ(Signal* signal)
   Datafile file_key;
   file_key.m_file_no = req.key.m_file_no;
   ndbrequire(m_file_hash.find(file_ptr, file_key));
-  
-  Uint32 size = file_ptr.p->m_extent_size;
-  Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
-  Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
-  Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
-  
-  Uint32 extent = (req.key.m_page_no - data_off) / size;
-  Uint32 extent_no = extent % per_page;
+
+  struct req val = lookup_extent(req.key.m_page_no, file_ptr.p);
   
   Page_cache_client::Request preq;
-  preq.m_page.m_page_no = 1 /* zero */ + extent / per_page;
+  preq.m_page.m_page_no = val.m_extent_page_no;
   preq.m_page.m_file_no = req.key.m_file_no;
   
   Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
@@ -1891,11 +1865,11 @@ Tsman::execALLOC_PAGE_REQ(Signal* signal)
     
     File_formats::Datafile::Extent_page* page = 
       (File_formats::Datafile::Extent_page*)ptr_p;
-    header= page->get_header(extent_no, size);
+    header= page->get_header(val.m_extent_no, val.m_extent_size);
     
     ndbrequire(header->m_table == req.request.table_id);
     
-    Uint32 page_no_in_extent = (req.key.m_page_no - data_off) % size;
+    Uint32 page_no_in_extent = calc_page_no_in_extent(req.key.m_page_no, &val);
     Uint32 word = header->get_free_word_offset(page_no_in_extent);
     Uint32 shift = SZ * (page_no_in_extent & 7);
     
@@ -1912,7 +1886,7 @@ Tsman::execALLOC_PAGE_REQ(Signal* signal)
      * Search
      */
     Uint32 *src= header->m_page_bitmask + word;
-    for(page_no= page_no_in_extent; page_no<size; page_no++)
+    for(page_no= page_no_in_extent; page_no<val.m_extent_size; page_no++)
     {
       src_bits= (* src >> shift) & ((1 << SZ) - 1);
       if((src_bits & UNCOMMITTED_MASK) <= reqbits)
@@ -1955,15 +1929,26 @@ Tsman::execALLOC_PAGE_REQ(Signal* signal)
 found:
   header->update_free_bits(page_no, src_bits | UNCOMMITTED_MASK);
   rep->bits= (src_bits & UNCOMMITTED_MASK) >> UNCOMMITTED_SHIFT;
-  rep->key.m_page_no= data_off + extent * size + page_no;
+  rep->key.m_page_no= 
+    val.m_extent_pages + val.m_extent_no * val.m_extent_size + page_no;
   rep->reply.errorCode= 0;
   return;
 }
 
 void
+Tsman::execLCP_FRAG_ORD(Signal* signal)
+{
+  jamEntry();
+  ndbrequire(!m_lcp_ongoing);
+  m_lcp_ongoing = true;
+}
+
+void
 Tsman::execEND_LCP_REQ(Signal* signal)
 {
   jamEntry();
+  ndbrequire(m_lcp_ongoing);
+  m_lcp_ongoing = false;
 
   /**
    * Move extents from "lcp" free list to real free list
diff --git a/storage/ndb/src/kernel/blocks/tsman.hpp b/storage/ndb/src/kernel/blocks/tsman.hpp
index 7c92dfcb068..018d648de07 100644
--- a/storage/ndb/src/kernel/blocks/tsman.hpp
+++ b/storage/ndb/src/kernel/blocks/tsman.hpp
@@ -63,6 +63,7 @@ protected:
 
   void execALLOC_PAGE_REQ(Signal* signal);
 
+  void execLCP_FRAG_ORD(Signal*);
   void execEND_LCP_REQ(Signal*);
   void end_lcp(Signal*, Uint32 tablespace, Uint32 list, Uint32 file);
 
@@ -108,6 +109,7 @@ public:
 	Uint32 m_offset_data_pages;    // 1(zero) + extent header pages
 	Uint32 m_data_pages;
 	Uint32 m_used_extent_cnt;
+	Uint32 m_extent_headers_per_extent_page;
       } m_online;
       struct {
 	Uint32 m_senderData;
@@ -196,6 +198,7 @@ private:
   Datafile_pool m_file_pool;
   Tablespace_pool m_tablespace_pool;
   
+  bool m_lcp_ongoing;
   Datafile_hash m_file_hash;
   Tablespace_list m_tablespace_list;
   Tablespace_hash m_tablespace_hash;
@@ -226,15 +229,52 @@ private:
 
   void release_extent_pages(Signal* signal, Ptr<Datafile> ptr);
   void release_extent_pages_callback(Signal*, Uint32, Uint32);
+
+  struct req
+  {
+    Uint32 m_extent_pages;
+    Uint32 m_extent_size;
+    Uint32 m_extent_no;      // on extent page
+    Uint32 m_extent_page_no;
+  };
+  
+  struct req lookup_extent(Uint32 page_no, const Datafile*) const;
+  Uint32 calc_page_no_in_extent(Uint32 page_no, const struct req* val) const;
 };
 
+inline
+Tsman::req
+Tsman::lookup_extent(Uint32 page_no, const Datafile * filePtrP) const
+{
+  struct req val;
+  val.m_extent_size = filePtrP->m_extent_size;
+  val.m_extent_pages = filePtrP->m_online.m_offset_data_pages;
+  Uint32 per_page = filePtrP->m_online.m_extent_headers_per_extent_page;
+  
+  Uint32 extent = 
+    (page_no - val.m_extent_pages) / val.m_extent_size + per_page;
+  
+  val.m_extent_page_no = extent / per_page;
+  val.m_extent_no = extent % per_page;
+  return val;
+}
+
+inline
+Uint32 
+Tsman::calc_page_no_in_extent(Uint32 page_no, const Tsman::req* val) const
+{
+  return (page_no - val->m_extent_pages) % val->m_extent_size;
+}
+
 class Tablespace_client
 {
+public:
   Tsman * m_tsman;
   Signal* m_signal;
   Uint32 m_table_id;
   Uint32 m_fragment_id;
   Uint32 m_tablespace_id;
+
 public:
   Tablespace_client(Signal* signal, Tsman* tsman, 
 		    Uint32 table, Uint32 fragment, Uint32 tablespaceId) {
@@ -244,6 +284,8 @@ public:
     m_fragment_id= fragment;
     m_tablespace_id= tablespaceId;
   }
+
+  Tablespace_client(Signal* signal, Tsman* tsman, Local_key* key);
   
   /**
    * Return >0 if success, no of pages in extent, sets key
@@ -274,7 +316,7 @@ public:
   /**
    * Free extent
    */
-  int free_extent(Local_key* key);
+  int free_extent(Local_key* key, Uint64 lsn);
   
   /**
    * Update page free bits
@@ -307,6 +349,11 @@ public:
    *        <0 - on error
    */
   int get_tablespace_info(CreateFilegroupImplReq* rep);
+
+  /**
+   * Update lsn of page corresponing to key
+   */
+  int update_lsn(Local_key* key, Uint64 lsn);
 };
 
 #include <signaldata/Extent.hpp>
@@ -351,12 +398,14 @@ Tablespace_client::alloc_page_from_extent(Local_key* key, Uint32 bits)
 
 inline
 int
-Tablespace_client::free_extent(Local_key* key)
+Tablespace_client::free_extent(Local_key* key, Uint64 lsn)
 {
   FreeExtentReq* req = (FreeExtentReq*)m_signal->theData;
   req->request.key = *key;
   req->request.table_id = m_table_id;
   req->request.tablespace_id = m_tablespace_id;
+  req->request.lsn_hi = (Uint32)(lsn >> 32);
+  req->request.lsn_lo = (Uint32)(lsn & 0xFFFFFFFF);
   m_tsman->execFREE_EXTENT_REQ(m_signal);
   
   if(req->reply.errorCode == 0){
@@ -407,5 +456,4 @@ Tablespace_client::restart_undo_page_free_bits(Local_key* key,
 					      page_lsn);
 }
 
-
 #endif