bug#16787 - ndb dd - fix datafiles with more than 1 extent page

storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp:
  qd fix for inconsistency tsman/tup
storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp:
  Increase extent pool
storage/ndb/src/kernel/blocks/pgman.cpp:
  Increase page entry pool
  Run LCP loop wo/ continueb
storage/ndb/src/kernel/blocks/tsman.cpp:
  Fix creating of datafiles with more than 1 extent page

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

@@ -346,12 +346,19 @@ Dbtup::disk_page_prealloc(Signal* signal,
     if ((pos= alloc.find_extent(sz)) != RNIL)
     {
       jam();
+      Uint32 cnt = 0;
       LocalDLList<Extent_info> list(c_extent_pool, alloc.m_free_extents[pos]);
       list.first(ext);
+      while((pageBits= tsman.alloc_page_from_extent(&ext.p->m_key, bits)) < 0)
+	if(!list.next(ext) || ++cnt < 10)
+	  break;
+      if (cnt == 10 || ext.isNull())
+	goto alloc;
       list.remove(ext);
     }
     else
     {
+  alloc:
       jam();
       /**
        * We need to alloc an extent

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

@@ -307,9 +307,9 @@ void Dbtup::execREAD_CONFIG_REQ(Signal* signal)
   c_buildIndexPool.setSize(c_noOfBuildIndexRec);
   c_triggerPool.setSize(noOfTriggers);
 
-  c_extent_pool.setSize(256);
+  c_extent_pool.setSize(8192);
   c_extent_hash.setSize(1024); // 4k
-  c_page_request_pool.setSize(100);
+  c_page_request_pool.setSize(1000);
   
   Uint32 nScanOp;       // use TUX config for now
   ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_TUX_SCAN_OP, &nScanOp));

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

@@ -125,7 +125,7 @@ Pgman::execREAD_CONFIG_REQ(Signal* signal)
   if (page_buffer > 0)
   {
     page_buffer /= GLOBAL_PAGE_SIZE; // in pages
-    m_page_entry_pool.setSize(2*page_buffer);
+    m_page_entry_pool.setSize(100*page_buffer);
     m_page_request_pool.setSize(10000);
     m_param.m_max_pages = page_buffer;
     m_param.m_max_hot_pages = (page_buffer * 9) / 10;
@@ -145,7 +145,7 @@ Pgman::Param::Param() :
   m_max_io_waits(64),
   m_stats_loop_delay(1000),
   m_cleanup_loop_delay(200),
-  m_lcp_loop_delay(200)
+  m_lcp_loop_delay(0)
 {
 }
 
@@ -411,6 +411,8 @@ Pgman::get_page_entry(Ptr<Page_entry>& ptr, Uint32 file_no, Uint32 page_no)
     return true;
   }
 
+  ndbrequire(false);
+  
   return false;
 }
 
@@ -739,7 +741,10 @@ Pgman::do_lcp_loop(Signal* signal, bool direct)
   {
     Uint32 delay = m_param.m_lcp_loop_delay;
     signal->theData[0] = PgmanContinueB::LCP_LOOP;
-    sendSignalWithDelay(PGMAN_REF, GSN_CONTINUEB, signal, delay, 1);
+    if (delay)
+      sendSignalWithDelay(PGMAN_REF, GSN_CONTINUEB, signal, delay, 1);
+    else
+      sendSignal(PGMAN_REF, GSN_CONTINUEB, signal, 1, JBB);
   }
 #ifdef VM_TRACE
   debugOut << "PGMAN: <do_lcp_loop on=" << m_lcp_loop_on

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

@@ -148,7 +148,15 @@ Tsman::execCONTINUEB(Signal* signal){
     release_extent_pages(signal, ptr);
     return;
   }
+  case TsmanContinueB::LOAD_EXTENT_PAGES:
+  {
+    Ptr<Datafile> ptr;
+    m_file_pool.getPtr(ptr, ptrI);
+    load_extent_pages(signal, ptr);
+    return;
   }
+  }
+  ndbrequire(false);
 }
 
 #ifdef VM_TRACE
@@ -702,8 +710,9 @@ Tsman::open_file(Signal* signal,
    * Update file size
    */
   pages = 1 + extent_pages + data_pages;
-  hi = (pages * File_formats::NDB_PAGE_SIZE) >> 32;
-  lo = (pages * File_formats::NDB_PAGE_SIZE) & 0xFFFFFFFF;
+  Uint64 bytes = pages * File_formats::NDB_PAGE_SIZE;
+  hi = bytes >> 32;
+  lo = bytes & 0xFFFFFFFF;
   req->file_size_hi = hi;
   req->file_size_lo = lo;
 
@@ -1185,7 +1194,7 @@ Tsman::scan_extent_headers(Signal* signal, Ptr<Datafile> ptr)
        * Last extent header page...
        *   set correct no of extent headers
        */
-      extents= datapages / size;
+      extents= (datapages / size) % per_page;
     }
     for(Uint32 j = 0; j<extents; j++)
     {
@@ -1712,9 +1721,12 @@ Tsman::unmap_page(Signal* signal, Local_key *key)
     /**
      * Toggle word
      */
+    Uint32 old = header->get_free_bits(page_no_in_extent);
     unsigned bit = 
       (header->get_free_bits(page_no_in_extent) & ((1 << (SZ - 1)) - 1));
     header->update_free_bits(page_no_in_extent, bit);
+    ndbout_c("toggle page: (%d, %d, %d) from %x to %x", 
+	     key->m_page_no, extent, page_no_in_extent, old, bit);
     return 0;
   }
   
@@ -1815,6 +1827,14 @@ Tsman::execALLOC_PAGE_REQ(Signal* signal)
       shift &= 31;
     }
 
+#if 0
+    printf("req.bits: %d bits: ", req.bits);
+    for(Uint32 i = 0; i<size; i++)
+    {
+      printf("%x", header->get_free_bits(i));
+    }
+    ndbout_c("");
+#endif
     err= AllocPageReq::NoPageFree;
   }
   
@@ -1822,6 +1842,8 @@ Tsman::execALLOC_PAGE_REQ(Signal* signal)
   return;
   
 found:
+  ndbout_c("alloc page: (%d, %d, %d)", 
+	   data_off + extent * size + page_no, per_page + extent, page_no);
   src_bits |= (1 << (SZ - 1)); // high unlogged, allocated bit
   header->update_free_bits(page_no, src_bits);
   rep->bits= src_bits & ((1 << (SZ - 1)) - 1);