bk commit into 5.1-ndb tree (joreland:1.1794)

[jonas.oreland]

Below is the list of changes that have just been committed into a local
5.1-ndb repository of jonas. When jonas does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet
  1.1794 05/03/21 12:20:59 joreland@stripped +3 -0
  wl1870 - ndb diskdata
    functional flushing of UNDO log

  ndb/src/kernel/blocks/ndbfs/Ndbfs.cpp
    1.19 05/03/21 12:20:55 joreland@stripped +1 -1
    allow multi page operations for global pages

  ndb/src/kernel/blocks/lgman.hpp
    1.17 05/03/21 12:20:55 joreland@stripped +22 -8
    functional flush log

  ndb/src/kernel/blocks/lgman.cpp
    1.20 05/03/21 12:20:55 joreland@stripped +131 -24
    functional flush log

# This is a BitKeeper patch.  What follows are the unified diffs for the
# set of deltas contained in the patch.  The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User:	joreland
# Host:	eel.ndb.mysql.com
# Root:	/home/jonas/src/mysql-5.1-ndb-dd

--- 1.19/ndb/src/kernel/blocks/lgman.cpp	Sun Mar 20 21:09:43 2005
+++ 1.20/ndb/src/kernel/blocks/lgman.cpp	Mon Mar 21 12:20:55 2005
@@ -59,6 +59,8 @@
   addRecSignal(GSN_DROP_FILEGROUP_REQ, &Lgman::execDROP_FILEGROUP_REQ);
 
   addRecSignal(GSN_FSWRITEREQ, &Lgman::execFSWRITEREQ);
+  addRecSignal(GSN_FSWRITEREF, &Lgman::execFSWRITEREF);
+  addRecSignal(GSN_FSWRITECONF, &Lgman::execFSWRITECONF);
 
   addRecSignal(GSN_FSOPENREF, &Lgman::execFSOPENREF);
   addRecSignal(GSN_FSOPENCONF, &Lgman::execFSOPENCONF);
@@ -219,7 +221,7 @@
     Ptr<Undofile> file_ptr;
     Undofile key;
     key.m_file_no = req->file_id;
-    key.m_logfile_group_id = req->filegroup_id;
+    key.m_logfile_group_ptr_i = ptr.i;
     if (m_file_hash.find(file_ptr, key))
     {
       jam();
@@ -234,7 +236,7 @@
       break;
     }
     
-    new (file_ptr.p) Undofile(req);
+    new (file_ptr.p) Undofile(req, ptr.i);
     m_file_hash.add(file_ptr);
     LocalDLList<Undofile> tmp(m_file_pool, ptr.p->m_meta_files);
     tmp.add(file_ptr);
@@ -309,7 +311,7 @@
 			     ndbGetOwnVersion(),
 			     time(0));
     page->m_file_no = ptr.p->m_file_no;
-    page->m_logfile_group_id = ptr.p->m_logfile_group_id;
+    page->m_logfile_group_id = ptr.p->m_create.m_logfile_group_id;
     page->m_logfile_group_version = ptr.p->m_create.m_logfile_group_version;
     page->m_undo_pages = ptr.p->m_file_size - 1; // minus zero page
   }
@@ -337,7 +339,7 @@
   Uint32 osErrCode = ref->osErrorCode;
 
   m_file_pool.getPtr(ptr, ref->userPointer);
-  ndbrequire(m_logfile_group_hash.find(lg_ptr, ptr.p->m_logfile_group_id));
+  m_logfile_group_pool.getPtr(lg_ptr, ptr.p->m_logfile_group_ptr_i);
 
   {
     CreateFileImplRef* ref= (CreateFileImplRef*)signal->getDataPtr();
@@ -365,9 +367,9 @@
   
   Uint32 fd = conf->filePointer;
   m_file_pool.getPtr(ptr, conf->userPointer);
-  ndbrequire(m_logfile_group_hash.find(lg_ptr, ptr.p->m_logfile_group_id));
+  m_logfile_group_pool.getPtr(lg_ptr, ptr.p->m_logfile_group_ptr_i);
 
-  LocalDLFifoList<Undofile> free(m_file_pool, lg_ptr.p->m_free_files);
+  LocalDLFifoList<Undofile> free(m_file_pool, lg_ptr.p->m_files);
   LocalDLList<Undofile> meta(m_file_pool, lg_ptr.p->m_meta_files);
   meta.remove(ptr);
   free.add(ptr);
@@ -376,8 +378,10 @@
   Uint32 senderData = ptr.p->m_create.m_senderData;
 
   ptr.p->m_online.m_fd = fd;
-  ptr.p->m_online.m_tail_offset = 0;
-  ptr.p->m_online.m_head_offset = 0;
+  ptr.p->m_online.m_tail_offset = 0; // skip zero page
+  ptr.p->m_online.m_head_offset = 0; // skip zero page
+  ptr.p->m_online.m_outstanding = 0;
+  ptr.p->m_state = Undofile::FS_ONLINE;
   
   {
     CreateFileImplConf* conf= (CreateFileImplConf*)signal->getDataPtr();
@@ -387,7 +391,11 @@
 	       CreateFileImplConf::SignalLength, JBB);
   }
 
-  
+  if(lg_ptr.p->m_current_file.m_ptr_i == RNIL)
+  {
+    jam();
+    lg_ptr.p->m_current_file.m_ptr_i= ptr.i;
+  }
 }
 
 void
@@ -514,20 +522,19 @@
   map.release();
 }
 
-Lgman::Undofile::Undofile(const struct CreateFileImplReq* req)
+Lgman::Undofile::Undofile(const struct CreateFileImplReq* req, Uint32 ptrI)
 {
   m_file_no = req->file_id;
-  m_logfile_group_id = req->filegroup_id;
+  m_logfile_group_ptr_i= ptrI;
   
   Uint64 pages = req->file_size_hi;
   pages = (pages << 32) | req->file_size_lo;
   pages /= GLOBAL_PAGE_SIZE;
   m_file_size = pages;
   
-  ndbout_c("pages: %d", pages);
-
   m_create.m_senderRef = req->senderRef; // During META
   m_create.m_senderData = req->senderData; // During META
+  m_create.m_logfile_group_id = req->filegroup_id;
 }
 
 Logfile_client::Logfile_client(SimulatedBlock* block, 
@@ -724,12 +731,11 @@
   Logfile_group::Position reader= ptr.p->m_pos[READER];
   Logfile_group::Position writer= ptr.p->m_pos[WRITER];
  
-  Ptr<Undofile> filePtr;
-  //m_file_pool.getPtr(filePtr, ptr.p->m_current_file.m_ptr_i);
+  jamEntry();
 
-  if(reader.m_current_page == writer.m_current_page)
+  if(ptr.p->m_current_file.m_ptr_i == RNIL ||
+     reader.m_current_page == writer.m_current_page)
   {
-    jamEntry();
     signal->theData[0] = LgmanContinueB::FLUSH_LOG;
     signal->theData[1] = ptr.i;
     sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 100, 2);
@@ -746,8 +752,9 @@
     {
       if(reader.m_current_page.m_idx > writer.m_current_page.m_idx)
       {
+	jam();
 	Uint32 tmp= reader.m_current_page.m_idx - writer.m_current_page.m_idx;
-	cnt= write_log_pages(signal, ptr, filePtr, page, tmp);
+	cnt= write_log_pages(signal, ptr, page, tmp);
 	assert(cnt <= tmp);
 	
 	reader.m_current_pos.m_ptr_i += cnt;
@@ -759,11 +766,12 @@
 	// Only 1 chunk
 	ndbassert(ptr.p->m_buffer_pages.getSize() == 2); 
 	Uint32 tmp= reader.m_current_page.m_idx + 1;
-	cnt= write_log_pages(signal, ptr, filePtr, page, tmp);
+	cnt= write_log_pages(signal, ptr, page, tmp);
 	assert(cnt <= tmp);
 
 	if(cnt == tmp)
 	{
+	  jam();
 	  /**
 	   * Entire chunk is written
 	   *   move to next
@@ -774,6 +782,7 @@
  	}
 	else
 	{
+	  jam();
 	  /**
 	   * Failed to write entire chunk...
 	   */
@@ -785,13 +794,13 @@
     }
     else
     {
-      ndbout_c("b");
       Uint32 tmp= reader.m_current_page.m_idx + 1;
-      cnt= write_log_pages(signal, ptr, filePtr, page, tmp);
+      cnt= write_log_pages(signal, ptr, page, tmp);
       assert(cnt <= tmp);
 
       if(cnt == tmp)
       {
+	jam();
 	/**
 	 * Entire chunk is written
 	 *   move to next
@@ -802,6 +811,7 @@
       }
       else
       {
+	jam();
 	/**
 	 * Failed to write entire chunk...
 	 */
@@ -811,7 +821,6 @@
       }
     }
     
-    ndbout_c("writing (%d %d)", page, cnt);
     tot += cnt;
     ptr.p->m_free_buffer_words += (cnt * File_formats::UNDO_PAGE_WORDS);
   }
@@ -824,10 +833,108 @@
 }
 
 Uint32
-Lgman::write_log_pages(Signal*, Ptr<Logfile_group>, Ptr<Undofile>, 
+Lgman::write_log_pages(Signal* signal, Ptr<Logfile_group> ptr,
 		       Uint32 pageId, Uint32 pages)
 {
-  return pages;
+  assert(pages);
+  Ptr<Undofile> filePtr;
+  m_file_pool.getPtr(filePtr, ptr.p->m_current_file.m_ptr_i);
+  
+  if(filePtr.p->m_online.m_outstanding > 0 || 
+     filePtr.p->m_state & Undofile::FS_FULL)
+  {
+    jam();
+    return 0;
+  }
+
+  filePtr.p->m_online.m_outstanding++;
+  
+  Uint32 sz= filePtr.p->m_file_size - 1; // skip zero
+  Uint32 head= filePtr.p->m_online.m_head_offset;
+  Uint32 tail= filePtr.p->m_online.m_tail_offset;
+  Uint32 free= head >= tail ? sz - head : tail - head;
+  
+  FsReadWriteReq* req= (FsReadWriteReq*)signal->getDataPtrSend();
+  req->filePointer = filePtr.p->m_online.m_fd;
+  req->userReference = reference();
+  req->userPointer = filePtr.i;
+  req->varIndex = 1+head; // skip zero page
+  req->numberOfPages = pages;
+  req->data.pageData[0] = pageId;
+  req->operationFlag = 0;
+  FsReadWriteReq::setFormatFlag(req->operationFlag, 
+				FsReadWriteReq::fsFormatGlobalPage);
+
+  Ptr<GlobalPage> pagePtr;
+  Uint32 cnt;
+  if(head + pages < free)
+  {
+    jam();
+    cnt= pages;
+    filePtr.p->m_online.m_head_offset = head + pages;
+    sendSignal(NDBFS_REF, GSN_FSWRITEREQ, signal, 
+	       FsReadWriteReq::FixedLength + 1, JBA);
+
+  }
+  else
+  {
+    jam();
+    cnt= free > pages ? pages : free;
+    
+    req->numberOfPages = cnt;
+    filePtr.p->m_online.m_head_offset = head + cnt;
+    FsReadWriteReq::setSyncFlag(req->operationFlag, 1);
+    
+    sendSignal(NDBFS_REF, GSN_FSWRITEREQ, signal, 
+	       FsReadWriteReq::FixedLength + 1, JBA);
+
+    if(((head + cnt) % sz) == tail)
+    {
+      jam();
+      filePtr.p->m_state |= Undofile::FS_FULL;
+    }
+  }
+  
+  if(head + cnt >= sz)
+  {
+    jam();
+    
+    Ptr<Undofile> next = filePtr;
+    LocalDLFifoList<Undofile> files(m_file_pool, ptr.p->m_files);
+    if(!files.next(next))
+    {
+      jam();
+      files.first(next);
+    }
+    ptr.p->m_current_file.m_ptr_i = next.i;
+
+    if(cnt != pages)
+    {
+      jam();
+      return cnt + write_log_pages(signal, ptr, pageId + cnt, pages - cnt);
+    }
+  }
+  
+  return cnt;
+}
+
+void
+Lgman::execFSWRITEREF(Signal* signal)
+{
+  jamEntry();
+  ndbrequire(false);
+}
+
+void
+Lgman::execFSWRITECONF(Signal* signal)
+{
+  jamEntry();
+  FsConf * conf = (FsConf*)signal->getDataPtr();
+  Ptr<Undofile> filePtr;
+  m_file_pool.getPtr(filePtr, conf->userPointer);
+
+  ndbassert(filePtr.p->m_online.m_outstanding);
+  filePtr.p->m_online.m_outstanding--;
 }
 
 void

--- 1.16/ndb/src/kernel/blocks/lgman.hpp	Sun Mar 20 21:09:43 2005
+++ 1.17/ndb/src/kernel/blocks/lgman.hpp	Mon Mar 21 12:20:55 2005
@@ -47,6 +47,9 @@
   void execDROP_FILEGROUP_REQ(Signal* signal);
   
   void execFSWRITEREQ(Signal*);
+  void execFSWRITEREF(Signal*);
+  void execFSWRITECONF(Signal*);
+
   void execFSOPENREF(Signal*);
   void execFSOPENCONF(Signal*);
 
@@ -64,21 +67,33 @@
   struct Undofile
   {
     Undofile(){}
-    Undofile(const struct CreateFileImplReq*);
-
+    Undofile(const struct CreateFileImplReq*, Uint32 lg_ptr_i);
+    
     Uint32 m_file_no; // Unique within logfile_group
-    Uint32 m_logfile_group_id;
+    Uint32 m_logfile_group_ptr_i;
 
     Uint32 m_file_size;
+    Uint32 m_state;
+
+    enum FileState 
+    {
+      FS_CREATING = 0x1  // File is being created
+      ,FS_DROPPING = 0x2 // File is being dropped
+      ,FS_ONLINE   = 0x4 // File is online
+      ,FS_FULL     = 0x8 // File is full (head == tail)
+    };
+    
     union {
       struct {
 	Uint32 m_fd; // When speaking to NDBFS
 	Uint32 m_tail_offset; // in pages
 	Uint32 m_head_offset; // in pages
+	Uint32 m_outstanding; // requests
       } m_online;
       struct {
 	Uint32 m_senderData;
 	Uint32 m_senderRef;
+	Uint32 m_logfile_group_id;
 	Uint32 m_logfile_group_version;
       } m_create;
     };
@@ -92,11 +107,11 @@
     };
 
     Uint32 hashValue() const {
-      return m_file_no ^ m_logfile_group_id;
+      return m_file_no ^ m_logfile_group_ptr_i;
     }
     bool equal(const Undofile& rec) const {
       return m_file_no == rec.m_file_no && 
-	m_logfile_group_id  == rec.m_logfile_group_id;
+	m_logfile_group_ptr_i  == rec.m_logfile_group_ptr_i;
     }
   };
   
@@ -130,8 +145,7 @@
     Lcp_record m_lcp[2];
 
     Buffer_idx m_current_file; // { file_ptr.i, page no }
-    DLFifoList<Undofile>::Head m_free_files; // Files w/ free space
-    DLFifoList<Undofile>::Head m_full_files; // Files wo/ free space
+    DLFifoList<Undofile>::Head m_files; // Files w/ free space
     DLList<Undofile>::Head m_meta_files;     // Files being created or dropped
     
     Uint32 m_free_buffer_words;    // Free buffer page words
@@ -176,7 +190,7 @@
   void open_file(Signal*, Ptr<Undofile>, Uint32 requestInfo);
 
   void flush_log(Signal*, Ptr<Logfile_group>);
-  Uint32 write_log_pages(Signal*, Ptr<Logfile_group>, Ptr<Undofile>, 
+  Uint32 write_log_pages(Signal*, Ptr<Logfile_group>, 
 			 Uint32 pageId, Uint32 pages);
 };
 

--- 1.18/ndb/src/kernel/blocks/ndbfs/Ndbfs.cpp	Thu Mar 10 08:31:03 2005
+++ 1.19/ndb/src/kernel/blocks/ndbfs/Ndbfs.cpp	Mon Mar 21 12:20:55 2005
@@ -410,7 +410,7 @@
     Ptr<GlobalPage> ptr;
     m_global_page_pool.getPtr(ptr, fsRWReq->data.pageData[0]);
     request->par.readWrite.pages[0].buf = (char*)ptr.p;
-    request->par.readWrite.pages[0].size = GLOBAL_PAGE_SIZE;
+    request->par.readWrite.pages[0].size = GLOBAL_PAGE_SIZE*fsRWReq->numberOfPages;
     request->par.readWrite.pages[0].offset= GLOBAL_PAGE_SIZE*fsRWReq->varIndex;
     request->par.readWrite.numberOfPages = 1;
   }