ndb dd

  Fix a copule of dd bugs
    typo in lgman, potential run-time crash
    inconsistency in tup, potentially causing crash after SR


storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp:
  Fix so that extentPtr.p->m_free_space is updated after page is written first time
    plus a number of new asserts
storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp:
  Fix so that extentPtr.p->m_free_space is updated after page is written first time
    plus a number of new asserts
storage/ndb/src/kernel/blocks/lgman.cpp:
  Fix a couple of asserts
  Fix a typo causing unpredictable crash "&= ~x" was "&= x"
storage/ndb/test/ndbapi/test_event.cpp:
  Fix test_event

4 files changed, 82 insertions, 50 deletions

diff --git a/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp b/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
index bb179ac52fb..9f8a1ad2295 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
@@ -2632,7 +2632,7 @@ 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 restart_setup_page(Disk_alloc_info&, Ptr<Page>);
   void update_extent_pos(Disk_alloc_info&, Ptr<Extent_info>);
   
   /**
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
index 611105cc39c..aeb42946b89 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
@@ -231,6 +231,16 @@ void
 Dbtup::update_extent_pos(Disk_alloc_info& alloc, 
 			 Ptr<Extent_info> extentPtr)
 {
+#ifdef VM_TRACE
+  Uint32 min_free = 0;
+  for(Uint32 i = 0; i<MAX_FREE_LIST; i++)
+  {
+    Uint32 sum = alloc.calc_page_free_space(i);
+    min_free += sum * extentPtr.p->m_free_page_count[i];
+  }
+  ddassert(extentPtr.p->m_free_space >= min_free);
+#endif
+  
   Uint32 old = extentPtr.p->m_free_matrix_pos;
   if (old != RNIL)
   {
@@ -252,7 +262,7 @@ Dbtup::update_extent_pos(Disk_alloc_info& alloc,
 }
 
 void
-Dbtup::restart_setup_page(Ptr<Page> pagePtr)
+Dbtup::restart_setup_page(Disk_alloc_info& alloc, Ptr<Page> pagePtr)
 {
   /**
    * Link to extent, clear uncommitted_used_space
@@ -266,7 +276,18 @@ Dbtup::restart_setup_page(Ptr<Page> pagePtr)
   Ptr<Extent_info> extentPtr;
   ndbrequire(c_extent_hash.find(extentPtr, key));
   pagePtr.p->m_extent_info_ptr = extentPtr.i;
-}  
+
+  Uint32 idx = pagePtr.p->list_index & ~0x8000;
+  Uint32 estimated = alloc.calc_page_free_space(idx);
+  Uint32 real_free = pagePtr.p->free_space;
+
+  ddassert(real_free >= estimated);
+  if (real_free != estimated)
+  {
+    extentPtr.p->m_free_space += (real_free - estimated);
+    update_extent_pos(alloc, extentPtr);
+  }
+}
 
 
 /**
@@ -608,7 +629,7 @@ Dbtup::disk_page_prealloc_callback(Signal* signal,
 
   if (unlikely(pagePtr.p->m_restart_seq != globalData.m_restart_seq))
   {
-    restart_setup_page(pagePtr);
+    restart_setup_page(fragPtr.p->m_disk_alloc_info, pagePtr);
   }
 
   disk_page_prealloc_callback_common(signal, req, fragPtr, pagePtr);
@@ -746,11 +767,6 @@ Dbtup::disk_page_set_dirty(Ptr<Page> pagePtr)
     return ;
   }
   
-  if (unlikely(pagePtr.p->m_restart_seq != globalData.m_restart_seq))
-  {
-    restart_setup_page(pagePtr);
-  }
-  
   Local_key key;
   key.m_page_no = pagePtr.p->m_page_no;
   key.m_file_no = pagePtr.p->m_file_no;
@@ -772,6 +788,12 @@ Dbtup::disk_page_set_dirty(Ptr<Page> pagePtr)
   getFragmentrec(fragPtr, pagePtr.p->m_fragment_id, tabPtr.p);
   
   Disk_alloc_info& alloc= fragPtr.p->m_disk_alloc_info;
+
+  if (unlikely(pagePtr.p->m_restart_seq != globalData.m_restart_seq))
+  {
+    restart_setup_page(alloc, pagePtr);
+  }
+  
   Tablespace_client tsman(0, c_tsman,
 			  fragPtr.p->fragTableId,
 			  fragPtr.p->fragmentId,
diff --git a/storage/ndb/src/kernel/blocks/lgman.cpp b/storage/ndb/src/kernel/blocks/lgman.cpp
index e38cd44fd72..884742f8788 100644
--- a/storage/ndb/src/kernel/blocks/lgman.cpp
+++ b/storage/ndb/src/kernel/blocks/lgman.cpp
@@ -1087,7 +1087,7 @@ Lgman::force_log_sync(Signal* signal,
        * Update free space with extra NOOP
        */
       ndbrequire(ptr.p->m_free_file_words >= free);
-      ndbrequire(ptr.p->m_free_buffer_words >= free);
+      ndbrequire(ptr.p->m_free_buffer_words > free);
       ptr.p->m_free_file_words -= free;
       ptr.p->m_free_buffer_words -= free;
       
@@ -1171,7 +1171,7 @@ Lgman::get_log_buffer(Ptr<Logfile_group> ptr, Uint32 sz)
   {
 next:
     // fits this page wo/ problem
-    ndbrequire(total_free >= sz);
+    ndbrequire(total_free > sz);
     ptr.p->m_free_buffer_words = total_free - sz;
     ptr.p->m_pos[PRODUCER].m_current_pos.m_idx = pos + sz;
     return ((File_formats::Undofile::Undo_page*)page)->m_data + pos;
@@ -1340,7 +1340,7 @@ Lgman::flush_log(Signal* signal, Ptr<Logfile_group> ptr, Uint32 force)
 	
 	Uint32 free= File_formats::UNDO_PAGE_WORDS - pos.m_idx;
 
-	ndbout_c("force flush %d", free);
+	ndbout_c("force flush %d %d", pos.m_idx, ptr.p->m_free_buffer_words);
 	
 	ndbrequire(pos.m_idx); // don't flush empty page...
 	Uint64 lsn= ptr.p->m_last_lsn - 1;
@@ -1355,7 +1355,7 @@ Lgman::flush_log(Signal* signal, Ptr<Logfile_group> ptr, Uint32 force)
 	 * Update free space with extra NOOP
 	 */
 	ndbrequire(ptr.p->m_free_file_words >= free);
-	ndbrequire(ptr.p->m_free_buffer_words >= free);
+	ndbrequire(ptr.p->m_free_buffer_words > free);
 	ptr.p->m_free_file_words -= free;
 	ptr.p->m_free_buffer_words -= free;
 	
@@ -1486,7 +1486,7 @@ Lgman::process_log_buffer_waiters(Signal* signal, Ptr<Logfile_group> ptr)
 
   if(list.isEmpty())
   {
-    ptr.p->m_state &= (Uint32)Logfile_group::LG_WAITERS_THREAD;
+    ptr.p->m_state &= ~(Uint32)Logfile_group::LG_WAITERS_THREAD;
     return;
   }
   
@@ -1661,23 +1661,26 @@ Lgman::execFSWRITECONF(Signal* signal)
   {
     Uint32 tot= 0;
     Uint64 lsn = 0;
-    LocalDLFifoList<Undofile> files(m_file_pool, lg_ptr.p->m_files);
-    while(cnt && ! (ptr.p->m_state & Undofile::FS_OUTSTANDING))
     {
-      Uint32 state= ptr.p->m_state;
-      Uint32 pages= ptr.p->m_online.m_outstanding;
-      ndbrequire(pages);
-      ptr.p->m_online.m_outstanding= 0;
-      ptr.p->m_state &= ~(Uint32)Undofile::FS_MOVE_NEXT;
-      tot += pages;
-      cnt--;
-      
-      lsn = ptr.p->m_online.m_lsn;
-
-      if((state & Undofile::FS_MOVE_NEXT) && !files.next(ptr))
-	files.first(ptr);
+      LocalDLFifoList<Undofile> files(m_file_pool, lg_ptr.p->m_files);
+      while(cnt && ! (ptr.p->m_state & Undofile::FS_OUTSTANDING))
+      {
+	Uint32 state= ptr.p->m_state;
+	Uint32 pages= ptr.p->m_online.m_outstanding;
+	ndbrequire(pages);
+	ptr.p->m_online.m_outstanding= 0;
+	ptr.p->m_state &= ~(Uint32)Undofile::FS_MOVE_NEXT;
+	tot += pages;
+	cnt--;
+	
+	lsn = ptr.p->m_online.m_lsn;
+	
+	if((state & Undofile::FS_MOVE_NEXT) && !files.next(ptr))
+	  files.first(ptr);
+      }
     }
     
+    ndbassert(tot);
     lg_ptr.p->m_outstanding_fs = cnt;
     lg_ptr.p->m_free_buffer_words += (tot * File_formats::UNDO_PAGE_WORDS);
     lg_ptr.p->m_next_reply_ptr_i = ptr.i;
@@ -1693,6 +1696,10 @@ Lgman::execFSWRITECONF(Signal* signal)
       process_log_buffer_waiters(signal, lg_ptr);
     }
   }
+  else
+  {
+    ndbout_c("miss matched writes");
+  }
   
   return;
 }
@@ -3004,29 +3011,32 @@ Lgman::execEND_LCP_CONF(Signal* signal)
 void 
 Lgman::validate_logfile_group(Ptr<Logfile_group> ptr, const char * heading)
 {
-  if (ptr.p->m_file_pos[HEAD].m_ptr_i == RNIL)
-    return;
-
-  Uint32 pages = compute_free_file_pages(ptr);
-
-  Uint32 group_pages = 
-    ((ptr.p->m_free_file_words + File_formats::UNDO_PAGE_WORDS - 1)/ File_formats::UNDO_PAGE_WORDS) ;
-  Uint32 last = ptr.p->m_free_file_words % File_formats::UNDO_PAGE_WORDS;
-  
-  if(! (pages >= group_pages))
+  do 
   {
-    ndbout << heading << " Tail: " << ptr.p->m_file_pos[TAIL] 
-	   << " Head: " << ptr.p->m_file_pos[HEAD] 
-	   << " free: " << group_pages << "(" << last << ")" 
-	   << " found: " << pages;
-    for(Uint32 i = 0; i<3; i++)
+    if (ptr.p->m_file_pos[HEAD].m_ptr_i == RNIL)
+      break;
+    
+    Uint32 pages = compute_free_file_pages(ptr);
+    
+    Uint32 group_pages = 
+      ((ptr.p->m_free_file_words + File_formats::UNDO_PAGE_WORDS - 1)/ File_formats::UNDO_PAGE_WORDS) ;
+    Uint32 last = ptr.p->m_free_file_words % File_formats::UNDO_PAGE_WORDS;
+    
+    if(! (pages >= group_pages))
     {
-      ndbout << " - " << ptr.p->m_tail_pos[i];
+      ndbout << heading << " Tail: " << ptr.p->m_file_pos[TAIL] 
+	     << " Head: " << ptr.p->m_file_pos[HEAD] 
+	     << " free: " << group_pages << "(" << last << ")" 
+	     << " found: " << pages;
+      for(Uint32 i = 0; i<3; i++)
+      {
+	ndbout << " - " << ptr.p->m_tail_pos[i];
+      }
+      ndbout << endl;
+      
+      ndbrequire(pages >= group_pages);
     }
-    ndbout << endl;
-
-    ndbrequire(pages >= group_pages);
-  }
+  } while(0);
 }
 #endif
 
diff --git a/storage/ndb/test/ndbapi/test_event.cpp b/storage/ndb/test/ndbapi/test_event.cpp
index a09f6d7c9c8..1e7db14ff40 100644
--- a/storage/ndb/test/ndbapi/test_event.cpp
+++ b/storage/ndb/test/ndbapi/test_event.cpp
@@ -990,9 +990,9 @@ static int copy_events(Ndb *ndb)
     while ((pOp= ndb->nextEvent()))
     {
       char buf[1024];
-      sprintf(buf, "%s_SHADOW", pOp->getTable()->getName());
+      sprintf(buf, "%s_SHADOW", pOp->getEvent()->getTable()->getName());
       const NdbDictionary::Table *table= dict->getTable(buf);
-
+      
       if (table == 0)
       {
 	g_err << "unable to find table " << buf << endl;