pgman.cpp

/*
   Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
*/

#include "pgman.hpp"
#include <signaldata/FsRef.hpp>
#include <signaldata/FsConf.hpp>
#include <signaldata/FsReadWriteReq.hpp>
#include <signaldata/PgmanContinueB.hpp>
#include <signaldata/LCP.hpp>
#include <signaldata/DataFileOrd.hpp>
#include <signaldata/ReleasePages.hpp>

#include <dbtup/Dbtup.hpp>

#include <DebuggerNames.hpp>
#include <md5_hash.hpp>

#define DIRTY_FLAGS (Page_request::COMMIT_REQ | \
                     Page_request::DIRTY_REQ | \
                     Page_request::ALLOC_REQ)

static bool g_dbg_lcp = false;
#if 1
#define DBG_LCP(x)
#else
#define DBG_LCP(x) if(g_dbg_lcp) ndbout << x
#endif

Pgman::Pgman(Block_context& ctx, Uint32 instanceNumber) :
  SimulatedBlock(PGMAN, ctx, instanceNumber),
  m_file_map(m_data_buffer_pool),
  m_page_hashlist(m_page_entry_pool),
  m_page_stack(m_page_entry_pool),
  m_page_queue(m_page_entry_pool)
#ifdef VM_TRACE
  ,debugFlag(false)
  ,debugSummaryFlag(false)
#endif
{
  BLOCK_CONSTRUCTOR(Pgman);

  // Add received signals
  addRecSignal(GSN_STTOR, &Pgman::execSTTOR);
  addRecSignal(GSN_READ_CONFIG_REQ, &Pgman::execREAD_CONFIG_REQ);
  addRecSignal(GSN_DUMP_STATE_ORD, &Pgman::execDUMP_STATE_ORD);
  addRecSignal(GSN_CONTINUEB, &Pgman::execCONTINUEB);
  addRecSignal(GSN_FSREADREF, &Pgman::execFSREADREF, true);
  addRecSignal(GSN_FSREADCONF, &Pgman::execFSREADCONF);
  addRecSignal(GSN_FSWRITEREF, &Pgman::execFSWRITEREF, true);
  addRecSignal(GSN_FSWRITECONF, &Pgman::execFSWRITECONF);

  addRecSignal(GSN_LCP_FRAG_ORD, &Pgman::execLCP_FRAG_ORD);
  addRecSignal(GSN_END_LCP_REQ, &Pgman::execEND_LCP_REQ);

  addRecSignal(GSN_DATA_FILE_ORD, &Pgman::execDATA_FILE_ORD);
  addRecSignal(GSN_RELEASE_PAGES_REQ, &Pgman::execRELEASE_PAGES_REQ);
  addRecSignal(GSN_DBINFO_SCANREQ, &Pgman::execDBINFO_SCANREQ);
  
  // loop status
  m_stats_loop_on = false;
  m_busy_loop_on = false;
  m_cleanup_loop_on = false;

  // LCP variables
  m_lcp_state = LS_LCP_OFF;
  m_last_lcp = 0;
  m_last_lcp_complete = 0;
  m_lcp_curr_bucket = ~(Uint32)0;
  m_lcp_outstanding = 0;

  // clean-up variables
  m_cleanup_ptr.i = RNIL;

  // should be a factor larger than number of pool pages
  m_data_buffer_pool.setSize(16);
  m_page_hashlist.setSize(512);
  
  for (Uint32 k = 0; k < Page_entry::SUBLIST_COUNT; k++)
    m_page_sublist[k] = new Page_sublist(m_page_entry_pool);

  {
    CallbackEntry& ce = m_callbackEntry[THE_NULL_CALLBACK];
    ce.m_function = TheNULLCallback.m_callbackFunction;
    ce.m_flags = 0;
  }
  {
    CallbackEntry& ce = m_callbackEntry[LOGSYNC_CALLBACK];
    ce.m_function = safe_cast(&Pgman::logsync_callback);
    ce.m_flags = 0;
  }
  {
    CallbackTable& ct = m_callbackTable;
    ct.m_count = COUNT_CALLBACKS;
    ct.m_entry = m_callbackEntry;
    m_callbackTableAddr = &ct;
  }
}

Pgman::~Pgman()
{
  for (Uint32 k = 0; k < Page_entry::SUBLIST_COUNT; k++)
    delete m_page_sublist[k];
}

BLOCK_FUNCTIONS(Pgman)

void 
Pgman::execREAD_CONFIG_REQ(Signal* signal)
{
  jamEntry();

  const ReadConfigReq * req = (ReadConfigReq*)signal->getDataPtr();

  Uint32 ref = req->senderRef;
  Uint32 senderData = req->senderData;

  const ndb_mgm_configuration_iterator * p = 
    m_ctx.m_config.getOwnConfigIterator();
  ndbrequire(p != 0);

  Uint64 page_buffer = 64*1024*1024;
  ndb_mgm_get_int64_parameter(p, CFG_DB_DISK_PAGE_BUFFER_MEMORY, &page_buffer);
  
  if (page_buffer > 0)
  {
    if (isNdbMtLqh())
    {
      // divide between workers - wl4391_todo give extra worker less
      Uint32 workers = getLqhWorkers() + 1;
      page_buffer = page_buffer / workers;
      Uint32 min_buffer = 4*1024*1024;
      if (page_buffer < min_buffer)
        page_buffer = min_buffer;
    }
    // convert to pages
    Uint32 page_cnt = Uint32((page_buffer + GLOBAL_PAGE_SIZE - 1) / GLOBAL_PAGE_SIZE);

    if (ERROR_INSERTED(11009))
    {
      page_cnt = 25;
      ndbout_c("Setting page_cnt = %u", page_cnt);
    }

    m_param.m_max_pages = page_cnt;
    m_page_entry_pool.setSize(m_param.m_lirs_stack_mult * page_cnt);
    m_param.m_max_hot_pages = (page_cnt * 9) / 10;
    ndbrequire(m_param.m_max_hot_pages >= 1);
  }

  Pool_context pc;
  pc.m_block = this;
  m_page_request_pool.wo_pool_init(RT_PGMAN_PAGE_REQUEST, pc);
  
  ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend();
  conf->senderRef = reference();
  conf->senderData = senderData;
  sendSignal(ref, GSN_READ_CONFIG_CONF, signal, 
             ReadConfigConf::SignalLength, JBB);
}

Pgman::Param::Param() :
  m_max_pages(64),      // smallish for testing
  m_lirs_stack_mult(10),
  m_max_hot_pages(56),
  m_max_loop_count(256),
  m_max_io_waits(256),
  m_stats_loop_delay(1000),
  m_cleanup_loop_delay(200),
  m_lcp_loop_delay(0)
{
}

void
Pgman::execSTTOR(Signal* signal)
{
  jamEntry();

  const Uint32 startPhase  = signal->theData[1];

  switch (startPhase) {
  case 1:
    {
      if (!isNdbMtLqh()) {
        c_tup = (Dbtup*)globalData.getBlock(DBTUP);
      } else if (instance() <= getLqhWorkers()) {
        c_tup = (Dbtup*)globalData.getBlock(DBTUP, instance());
        ndbrequire(c_tup != 0);
      } else {
        // extra worker
        c_tup = 0;
      }
      c_lgman = (Lgman*)globalData.getBlock(LGMAN);
      c_tsman = (Tsman*)globalData.getBlock(TSMAN);
    }
    break;
  case 3:
    {
      // start forever loops
      do_stats_loop(signal);
      do_cleanup_loop(signal);
      m_stats_loop_on = true;
      m_cleanup_loop_on = true;
    }
    break;
  case 7:
    break;
  default:
    break;
  }

  sendSTTORRY(signal);
}

void
Pgman::sendSTTORRY(Signal* signal)
{
  signal->theData[0] = 0;
  signal->theData[3] = 1;
  signal->theData[4] = 3;
  signal->theData[5] = 7;
  signal->theData[6] = 255; // No more start phases from missra
  BlockReference cntrRef = !isNdbMtLqh() ? NDBCNTR_REF : PGMAN_REF;
  sendSignal(cntrRef, GSN_STTORRY, signal, 7, JBB);
}

void
Pgman::execCONTINUEB(Signal* signal)
{
  jamEntry();
  Uint32 data1 = signal->theData[1];

  switch (signal->theData[0]) {
  case PgmanContinueB::STATS_LOOP:
    jam();
    do_stats_loop(signal);
    break;
  case PgmanContinueB::BUSY_LOOP:
    jam();
    do_busy_loop(signal);
    break;
  case PgmanContinueB::CLEANUP_LOOP:
    jam();
    do_cleanup_loop(signal);
    break;
  case PgmanContinueB::LCP_LOOP:
    jam();
    do_lcp_loop(signal);
    break;
  case PgmanContinueB::LCP_LOCKED:
  {
    jam();
    Ptr<Page_entry> ptr;
    Page_sublist& pl = *m_page_sublist[Page_entry::SL_LOCKED];
    if (data1 != RNIL)
    {
      jam();
      pl.getPtr(ptr, data1);
      process_lcp_locked(signal, ptr);
    }
    else
    {
      jam();
      if (ERROR_INSERTED(11007))
      {
        ndbout << "No more writes..." << endl;
        SET_ERROR_INSERT_VALUE(11008);
        signal->theData[0] = 9999;
        sendSignalWithDelay(CMVMI_REF, GSN_NDB_TAMPER, signal, 10000, 1);
      }
      EndLcpConf* conf = (EndLcpConf*)signal->getDataPtrSend();
      conf->senderData = m_end_lcp_req.senderData;
      conf->senderRef = reference();
      sendSignal(m_end_lcp_req.senderRef, GSN_END_LCP_CONF,
                 signal, EndLcpConf::SignalLength, JBB);
      m_lcp_state = LS_LCP_OFF;
    }
    return;
  }
  default:
    ndbrequire(false);
    break;
  }
}

// page entry

Pgman::Page_entry::Page_entry(Uint32 file_no, Uint32 page_no) :
  m_file_no(file_no),
  m_state(0),
  m_page_no(page_no),
  m_real_page_i(RNIL),
  m_lsn(0),
  m_last_lcp(0),
  m_dirty_count(0),
  m_copy_page_i(RNIL),
  m_busy_count(0),
  m_requests()
{
}

// page lists

Uint32
Pgman::get_sublist_no(Page_state state)
{
  if (state & Page_entry::REQUEST)
  {
    if (! (state & Page_entry::BOUND))
    {
      return Page_entry::SL_BIND;
    }
    if (! (state & Page_entry::MAPPED))
    {
      if (! (state & Page_entry::PAGEIN))
      {
        return Page_entry::SL_MAP;
      }
      return Page_entry::SL_MAP_IO;
    }
    if (! (state & Page_entry::PAGEOUT))
    {
      return Page_entry::SL_CALLBACK;
    }
    return Page_entry::SL_CALLBACK_IO;
  }
  if (state & Page_entry::BUSY)
  {
    return Page_entry::SL_BUSY;
  }
  if (state & Page_entry::LOCKED)
  {
    return Page_entry::SL_LOCKED;
  }
  if (state == Page_entry::ONSTACK) {
    return Page_entry::SL_IDLE;
  }
  if (state != 0)
  {
    return Page_entry::SL_OTHER;
  }
  return ZNIL;
}

void
Pgman::set_page_state(Ptr<Page_entry> ptr, Page_state new_state)
{
  D(">set_page_state: state=" << hex << new_state);
  D(ptr << ": before");

  Page_state old_state = ptr.p->m_state;
  if (old_state != new_state)
  {
    Uint32 old_list_no = get_sublist_no(old_state);
    Uint32 new_list_no = get_sublist_no(new_state);
    if (old_state != 0)
    {
      ndbrequire(old_list_no != ZNIL);
      if (old_list_no != new_list_no)
      {
        Page_sublist& old_list = *m_page_sublist[old_list_no];
        old_list.remove(ptr);
      }
    }
    if (new_state != 0)
    {
      ndbrequire(new_list_no != ZNIL);
      if (old_list_no != new_list_no)
      {
        Page_sublist& new_list = *m_page_sublist[new_list_no];
        new_list.add(ptr);
      }
    }
    ptr.p->m_state = new_state;

    bool old_hot = (old_state & Page_entry::HOT);
    bool new_hot = (new_state & Page_entry::HOT);
    if (! old_hot && new_hot)
    {
      jam();
      m_stats.m_num_hot_pages++;
    }
    if (old_hot && ! new_hot)
    {
      jam();
      ndbrequire(m_stats.m_num_hot_pages != 0);
      m_stats.m_num_hot_pages--;
    }
  }

  D(ptr << ": after");
#ifdef VM_TRACE
  verify_page_entry(ptr);
#endif
  D("<set_page_state");
}

// seize/release pages and entries

bool
Pgman::seize_cache_page(Ptr<GlobalPage>& gptr)
{
  // page cache has no own pool yet
  bool ok = m_global_page_pool.seize(gptr);

  // zero is reserved as return value for queued request
  if (ok && gptr.i == 0)
    ok = m_global_page_pool.seize(gptr);

  if (ok)
  {
    ndbrequire(m_stats.m_num_pages < m_param.m_max_pages);
    m_stats.m_num_pages++;
  }
  return ok;
}

void
Pgman::release_cache_page(Uint32 i)
{
  m_global_page_pool.release(i);

  ndbrequire(m_stats.m_num_pages != 0);
  m_stats.m_num_pages--;
}

bool
Pgman::find_page_entry(Ptr<Page_entry>& ptr, Uint32 file_no, Uint32 page_no)
{
  Page_entry key;
  key.m_file_no = file_no;
  key.m_page_no = page_no;
  
  if (m_page_hashlist.find(ptr, key))
  {
    D("find_page_entry");
    D(ptr);
    return true;
  }
  return false;
}

Uint32
Pgman::seize_page_entry(Ptr<Page_entry>& ptr, Uint32 file_no, Uint32 page_no)
{
  if (m_page_entry_pool.seize(ptr))
  {
    new (ptr.p) Page_entry(file_no, page_no);
    m_page_hashlist.add(ptr);
#ifdef VM_TRACE
    ptr.p->m_this = this;
#endif
    D("seize_page_entry");
    D(ptr);

    return true;
  }
  return false;
}

bool
Pgman::get_page_entry(Ptr<Page_entry>& ptr, Uint32 file_no, Uint32 page_no)
{
  if (find_page_entry(ptr, file_no, page_no))
  {
    jam();
    ndbrequire(ptr.p->m_state != 0);
    m_stats.m_page_hits++;

    D("get_page_entry: found");
    D(ptr);
    return true;
  }

  if (m_page_entry_pool.getNoOfFree() == 0)
  {
    jam();
    Page_sublist& pl_idle = *m_page_sublist[Page_entry::SL_IDLE];
    Ptr<Page_entry> idle_ptr;
    if (pl_idle.first(idle_ptr))
    {
      jam();

      D("get_page_entry: re-use idle entry");
      D(idle_ptr);

      Page_state state = idle_ptr.p->m_state;
      ndbrequire(state == Page_entry::ONSTACK);

      Page_stack& pl_stack = m_page_stack;
      ndbrequire(pl_stack.hasPrev(idle_ptr));
      pl_stack.remove(idle_ptr);
      state &= ~ Page_entry::ONSTACK;
      set_page_state(idle_ptr, state);
      ndbrequire(idle_ptr.p->m_state == 0);

      release_page_entry(idle_ptr);
    }
  }

  if (seize_page_entry(ptr, file_no, page_no))
  {
    jam();
    ndbrequire(ptr.p->m_state == 0);
    m_stats.m_page_faults++;

    D("get_page_entry: seize");
    D(ptr);
    return true;
  }

  ndbrequire(false);
  
  return false;
}

void
Pgman::release_page_entry(Ptr<Page_entry>& ptr)
{
  D("release_page_entry");
  D(ptr);
  Page_state state = ptr.p->m_state;

  ndbrequire(ptr.p->m_requests.isEmpty());

  ndbrequire(! (state & Page_entry::ONSTACK));
  ndbrequire(! (state & Page_entry::ONQUEUE));
  ndbrequire(ptr.p->m_real_page_i == RNIL);

  if (! (state & Page_entry::LOCKED))
  {
    ndbrequire(! (state & Page_entry::REQUEST));
  }

  if (ptr.p->m_copy_page_i != RNIL)
  {
    m_global_page_pool.release(ptr.p->m_copy_page_i);
  }
  
  set_page_state(ptr, 0);
  m_page_hashlist.remove(ptr);
  m_page_entry_pool.release(ptr);
}

// LIRS

/*
 * After the hot entry at stack bottom is removed, additional entries
 * are removed until next hot entry is found.  There are 3 cases for the
 * removed entry:  1) a bound entry is already on queue 2) an unbound
 * entry with open requests enters queue at bind time 3) an unbound
 * entry without requests is returned to entry pool.
 */
void
Pgman::lirs_stack_prune()
{
  D(">lirs_stack_prune");
  Page_stack& pl_stack = m_page_stack;
  Ptr<Page_entry> ptr;

  while (pl_stack.first(ptr))      // first is stack bottom
  {
    Page_state state = ptr.p->m_state;
    if (state & Page_entry::HOT)
    {
      jam();
      break;
    }

    D(ptr << ": prune from stack");

    pl_stack.remove(ptr);
    state &= ~ Page_entry::ONSTACK;
    set_page_state(ptr, state);

    if (state & Page_entry::BOUND)
    {
      jam();
      ndbrequire(state & Page_entry::ONQUEUE);
    }
    else if (state & Page_entry::REQUEST)
    {
      // enters queue at bind
      jam();
      ndbrequire(! (state & Page_entry::ONQUEUE));
    }
    else
    {
      jam();
      release_page_entry(ptr);
    }
  }
  D("<lirs_stack_prune");
}

/*
 * Remove the hot entry at stack bottom and make it cold and do stack
 * pruning.  There are 2 cases for the removed entry:  1) a bound entry
 * is moved to queue 2) an unbound entry must have requests and enters
 * queue at bind time.
 */
void
Pgman::lirs_stack_pop()
{
  D("lirs_stack_pop");
  Page_stack& pl_stack = m_page_stack;
  Page_queue& pl_queue = m_page_queue;

  Ptr<Page_entry> ptr;
  bool ok = pl_stack.first(ptr);
  ndbrequire(ok);
  Page_state state = ptr.p->m_state;

  D(ptr << ": pop from stack");

  ndbrequire(state & Page_entry::HOT);
  ndbrequire(state & Page_entry::ONSTACK);
  pl_stack.remove(ptr);
  state &= ~ Page_entry::HOT;
  state &= ~ Page_entry::ONSTACK;
  ndbrequire(! (state & Page_entry::ONQUEUE));

  if (state & Page_entry::BOUND)
  {
    jam();
    pl_queue.add(ptr);
    state |= Page_entry::ONQUEUE;
  }
  else
  {
    // enters queue at bind
    jam();
    ndbrequire(state & Page_entry::REQUEST);
  }

  set_page_state(ptr, state);
  lirs_stack_prune();
}

/*
 * Update LIRS lists when page is referenced.
 */
void
Pgman::lirs_reference(Ptr<Page_entry> ptr)
{
  D(">lirs_reference");
  D(ptr);
  Page_stack& pl_stack = m_page_stack;
  Page_queue& pl_queue = m_page_queue;

  Page_state state = ptr.p->m_state;
  ndbrequire(! (state & Page_entry::LOCKED));

  ndbrequire(m_stats.m_num_hot_pages <= m_param.m_max_hot_pages);

  // LIRS kicks in when we have max hot pages
  if (m_stats.m_num_hot_pages == m_param.m_max_hot_pages)
  {
    if (state & Page_entry::HOT)
    {
      // case 1
      jam();
      ndbrequire(state & Page_entry::ONSTACK);
      bool at_bottom = ! pl_stack.hasPrev(ptr);
      pl_stack.remove(ptr);
      pl_stack.add(ptr);
      if (at_bottom)
      {
        jam();
        lirs_stack_prune();
      }
    }
    else if (state & Page_entry::ONSTACK)
    {
      // case 2a 3a
      jam();
      pl_stack.remove(ptr);
      if (! pl_stack.isEmpty())
      {
        jam();
        lirs_stack_pop();
      }
      pl_stack.add(ptr);
      state |= Page_entry::HOT;
      if (state & Page_entry::ONQUEUE)
      {
        jam();
        move_cleanup_ptr(ptr);
        pl_queue.remove(ptr);
        state &= ~ Page_entry::ONQUEUE;
      }
    }
    else
    {
      // case 2b 3b
      jam();
      pl_stack.add(ptr);
      state |= Page_entry::ONSTACK;
      /*
       * bug#48910.  Using hot page count (not total page count)
       * guarantees that stack is not empty here.  Therefore the new
       * entry (added to top) is not at bottom and need not be hot.
       */
      ndbrequire(pl_stack.hasPrev(ptr));
      if (state & Page_entry::ONQUEUE)
      {
        jam();
        move_cleanup_ptr(ptr);
        pl_queue.remove(ptr);
        state &= ~ Page_entry::ONQUEUE;
      }
      if (state & Page_entry::BOUND)
      {
        jam();
        pl_queue.add(ptr);
        state |= Page_entry::ONQUEUE;
      }
      else
      {
        // enters queue at bind
        jam();
      }
    }
  }
  else
  {
    D("filling up hot pages: " << m_stats.m_num_hot_pages << "/"
                               << m_param.m_max_hot_pages);
    jam();
    if (state & Page_entry::ONSTACK)
    {
      jam();
      bool at_bottom = ! pl_stack.hasPrev(ptr);
      pl_stack.remove(ptr);
      if (at_bottom)
      {
        jam();
        ndbassert(state & Page_entry::HOT);
        lirs_stack_prune();
      }
    }
    pl_stack.add(ptr);
    state |= Page_entry::ONSTACK;
    state |= Page_entry::HOT;
    // it could be on queue already
    if (state & Page_entry::ONQUEUE) {
      jam();
      pl_queue.remove(ptr);
      state &= ~Page_entry::ONQUEUE;
    }
  }

  set_page_state(ptr, state);
  D("<lirs_reference");
}

// continueB loops

void
Pgman::do_stats_loop(Signal* signal)
{
  D("do_stats_loop");
#ifdef VM_TRACE
  verify_all();
#endif
  Uint32 delay = m_param.m_stats_loop_delay;
  signal->theData[0] = PgmanContinueB::STATS_LOOP;
  sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, delay, 1);
}

void
Pgman::do_busy_loop(Signal* signal, bool direct)
{
  D(">do_busy_loop on=" << m_busy_loop_on << " direct=" << direct);
  Uint32 restart = false;
  if (direct)
  {
    // may not cover the calling entry
    (void)process_bind(signal);
    (void)process_map(signal);
    // callback must be queued
    if (! m_busy_loop_on)
    {
      restart = true;
      m_busy_loop_on = true;
    }
  }
  else
  {
    ndbrequire(m_busy_loop_on);
    restart += process_bind(signal);
    restart += process_map(signal);
    restart += process_callback(signal);
    if (! restart)
    {
      m_busy_loop_on = false;
    }
  }
  if (restart)
  {
    signal->theData[0] = PgmanContinueB::BUSY_LOOP;
    sendSignal(reference(), GSN_CONTINUEB, signal, 1, JBB);
  }
  D("<do_busy_loop on=" << m_busy_loop_on << " restart=" << restart);
}

void
Pgman::do_cleanup_loop(Signal* signal)
{
  D("do_cleanup_loop");
  process_cleanup(signal);

  Uint32 delay = m_param.m_cleanup_loop_delay;
  signal->theData[0] = PgmanContinueB::CLEANUP_LOOP;
  sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, delay, 1);
}

void
Pgman::do_lcp_loop(Signal* signal)
{
  D(">do_lcp_loop m_lcp_state=" << Uint32(m_lcp_state));
  ndbrequire(m_lcp_state != LS_LCP_OFF);
  LCP_STATE newstate = process_lcp(signal);

  switch(newstate) {
  case LS_LCP_OFF:
    jam();
    break;
  case LS_LCP_ON:
    jam();
    signal->theData[0] = PgmanContinueB::LCP_LOOP;
    sendSignal(reference(), GSN_CONTINUEB, signal, 1, JBB);
    break;
  case LS_LCP_MAX_LCP_OUTSTANDING: // wait until io is completed
    jam();
    break;
  case LS_LCP_LOCKED:
    jam();
    break;
  }
  m_lcp_state = newstate;
  D("<do_lcp_loop m_lcp_state=" << Uint32(m_lcp_state));
}

// busy loop

bool
Pgman::process_bind(Signal* signal)
{
  D(">process_bind");
  int max_count = 32;
  Page_sublist& pl_bind = *m_page_sublist[Page_entry::SL_BIND];

  while (! pl_bind.isEmpty() && --max_count >= 0)
  {
    jam();
    Ptr<Page_entry> ptr;
    pl_bind.first(ptr);
    if (! process_bind(signal, ptr))
    {
      jam();
      break;
    }
  }
  D("<process_bind");
  return ! pl_bind.isEmpty();
}

bool
Pgman::process_bind(Signal* signal, Ptr<Page_entry> ptr)
{
  D(ptr << " : process_bind");
  Page_queue& pl_queue = m_page_queue;
  Ptr<GlobalPage> gptr;

  if (m_stats.m_num_pages < m_param.m_max_pages)
  {
    jam();
    bool ok = seize_cache_page(gptr);
    // to handle failure requires some changes in LIRS
    ndbrequire(ok);
  }
  else
  {
    jam();
    Ptr<Page_entry> clean_ptr;
    if (! pl_queue.first(clean_ptr))
    {
      jam();
      D("bind failed: queue empty");
      // XXX busy loop
      return false;
    }
    Page_state clean_state = clean_ptr.p->m_state;
    // under unusual circumstances it could still be paging in
    if (! (clean_state & Page_entry::MAPPED) ||
        clean_state & Page_entry::DIRTY ||
        clean_state & Page_entry::REQUEST)
    {
      jam();
      D("bind failed: queue front not evictable");
      D(clean_ptr);
      // XXX busy loop
      return false;
    }

    D(clean_ptr << " : evict");

    ndbassert(clean_ptr.p->m_dirty_count == 0);
    ndbrequire(clean_state & Page_entry::ONQUEUE);
    ndbrequire(clean_state & Page_entry::BOUND);
    ndbrequire(clean_state & Page_entry::MAPPED);

    move_cleanup_ptr(clean_ptr);
    pl_queue.remove(clean_ptr);
    clean_state &= ~ Page_entry::ONQUEUE;

    gptr.i = clean_ptr.p->m_real_page_i;

    clean_ptr.p->m_real_page_i = RNIL;
    clean_state &= ~ Page_entry::BOUND;
    clean_state &= ~ Page_entry::MAPPED;

    set_page_state(clean_ptr, clean_state);

    if (! (clean_state & Page_entry::ONSTACK))
      release_page_entry(clean_ptr);

    m_global_page_pool.getPtr(gptr);
  }

  Page_state state = ptr.p->m_state;

  ptr.p->m_real_page_i = gptr.i;
  state |= Page_entry::BOUND;
  if (state & Page_entry::EMPTY)
  {
    jam();
    state |= Page_entry::MAPPED;
  }

  if (! (state & Page_entry::LOCKED) &&
      ! (state & Page_entry::ONQUEUE) &&
      ! (state & Page_entry::HOT))
  {
    jam();

    D(ptr << " : add to queue at bind");
    pl_queue.add(ptr);
    state |= Page_entry::ONQUEUE;
  }

  set_page_state(ptr, state);
  return true;
}

bool
Pgman::process_map(Signal* signal)
{
  D(">process_map");
  int max_count = 0;
  if (m_param.m_max_io_waits > m_stats.m_current_io_waits) {
    max_count = m_param.m_max_io_waits - m_stats.m_current_io_waits;
    max_count = max_count / 2 + 1;
  }
  Page_sublist& pl_map = *m_page_sublist[Page_entry::SL_MAP];

  while (! pl_map.isEmpty() && --max_count >= 0)
  {
    jam();
    Ptr<Page_entry> ptr;
    pl_map.first(ptr);
    if (! process_map(signal, ptr))
    {
      jam();
      break;
    }
  }
  D("<process_map");
  return ! pl_map.isEmpty();
}

bool
Pgman::process_map(Signal* signal, Ptr<Page_entry> ptr)
{
  D(ptr << " : process_map");
  pagein(signal, ptr);
  return true;
}

bool
Pgman::process_callback(Signal* signal)
{
  D(">process_callback");
  int max_count = 1;
  Page_sublist& pl_callback = *m_page_sublist[Page_entry::SL_CALLBACK];

  Ptr<Page_entry> ptr;
  pl_callback.first(ptr);

  while (! ptr.isNull() && --max_count >= 0)
  {
    jam();
    Ptr<Page_entry> curr = ptr;
    pl_callback.next(ptr);
    
    if (! process_callback(signal, curr))
    {
      jam();
      break;
    }
  }
  D("<process_callback");
  return ! pl_callback.isEmpty();
}

bool
Pgman::process_callback(Signal* signal, Ptr<Page_entry> ptr)
{
  D(ptr << " : process_callback");
  int max_count = 1;

  while (! ptr.p->m_requests.isEmpty() && --max_count >= 0)
  {
    jam();
    Page_state state = ptr.p->m_state;
    SimulatedBlock* b;
    Callback callback;
    {
      Local_page_request_list req_list(m_page_request_pool, ptr.p->m_requests);
      Ptr<Page_request> req_ptr;

      req_list.first(req_ptr);
      D(req_ptr << " : process_callback");

#ifdef ERROR_INSERT
      if (req_ptr.p->m_flags & Page_request::DELAY_REQ)
      {
        Uint64 now = NdbTick_CurrentMillisecond();
        if (now < req_ptr.p->m_delay_until_time)
        {
          break;
        }
      }
#endif
      
      Uint32 blockNo = blockToMain(req_ptr.p->m_block);
      Uint32 instanceNo = blockToInstance(req_ptr.p->m_block);
      b = globalData.getBlock(blockNo, instanceNo);
      callback = req_ptr.p->m_callback;
      
      if (req_ptr.p->m_flags & DIRTY_FLAGS)
      {
        jam();
        state |= Page_entry::DIRTY;
        ndbassert(ptr.p->m_dirty_count);
        ptr.p->m_dirty_count --;
      }

      req_list.releaseFirst(req_ptr);
    }
    ndbrequire(state & Page_entry::BOUND);
    ndbrequire(state & Page_entry::MAPPED);

    // make REQUEST state consistent before set_page_state()
    if (ptr.p->m_requests.isEmpty())
    {
      jam();
      state &= ~ Page_entry::REQUEST;
    }
    
    // callback may re-enter PGMAN and change page state
    set_page_state(ptr, state);
    b->execute(signal, callback, ptr.p->m_real_page_i);
  }
  return true;
}

// cleanup loop

bool
Pgman::process_cleanup(Signal* signal)
{
  D(">process_cleanup");
  Page_queue& pl_queue = m_page_queue;

  // XXX for now start always from beginning
  m_cleanup_ptr.i = RNIL;

  if (m_cleanup_ptr.i == RNIL && ! pl_queue.first(m_cleanup_ptr))
  {
    jam();
    D("<process_cleanup: empty queue");
    return false;
  }

  int max_loop_count = m_param.m_max_loop_count;
  int max_count = 0;
  if (m_param.m_max_io_waits > m_stats.m_current_io_waits) {
    max_count = m_param.m_max_io_waits - m_stats.m_current_io_waits;
    max_count = max_count / 2 + 1;
  }

  Ptr<Page_entry> ptr = m_cleanup_ptr;
  while (max_loop_count != 0 && max_count != 0)
  {
    Page_state state = ptr.p->m_state;
    ndbrequire(! (state & Page_entry::LOCKED));
    if (state & Page_entry::BUSY)
    {
      D("process_cleanup: break on busy page");
      D(ptr);
      break;
    }
    if (state & Page_entry::DIRTY &&
        ! (state & Page_entry::PAGEIN) &&
        ! (state & Page_entry::PAGEOUT))
    {
      D(ptr << " : process_cleanup");
      if (c_tup != 0)
        c_tup->disk_page_unmap_callback(0, 
                                        ptr.p->m_real_page_i, 
                                        ptr.p->m_dirty_count);
      pageout(signal, ptr);
      max_count--;
    }
    if (! pl_queue.hasNext(ptr))
      break;
    pl_queue.next(ptr);
    max_loop_count--;
  }
  m_cleanup_ptr = ptr;
  D("<process_cleanup");
  return true;
}

/*
 * Call this before queue.remove(ptr).  If the removed entry is the
 * clean-up pointer, move it towards front.
 */
void
Pgman::move_cleanup_ptr(Ptr<Page_entry> ptr)
{
  Page_queue& pl_queue = m_page_queue;
  if (ptr.i == m_cleanup_ptr.i)
  {
    jam();
    pl_queue.prev(m_cleanup_ptr);
  }
}

// LCP


void
Pgman::execLCP_FRAG_ORD(Signal* signal)
{
  if (ERROR_INSERTED(11008))
  {
    ndbout_c("Ignore LCP_FRAG_ORD");
    return;
  }
  LcpFragOrd* ord = (LcpFragOrd*)signal->getDataPtr();
  ndbrequire(ord->lcpId >= m_last_lcp_complete + 1 || m_last_lcp_complete == 0);
  m_last_lcp = ord->lcpId;
  DBG_LCP("Pgman::execLCP_FRAG_ORD lcp: " << m_last_lcp << endl);
  
  D("execLCP_FRAG_ORD"
    << " this=" << m_last_lcp
    << " last_complete=" << m_last_lcp_complete
    << " bucket=" << m_lcp_curr_bucket);
}

void
Pgman::execEND_LCP_REQ(Signal* signal)
{
  if (ERROR_INSERTED(11008))
  {
    ndbout_c("Ignore END_LCP");
    return;
  }

  EndLcpReq* req = (EndLcpReq*)signal->getDataPtr();
  m_end_lcp_req = *req;

  DBG_LCP("execEND_LCP_REQ" << endl);

  ndbrequire(!m_lcp_outstanding);
  m_lcp_curr_bucket = 0;
  
  D("execEND_LCP_REQ"
    << " this=" << m_last_lcp
    << " last_complete=" << m_last_lcp_complete
    << " bucket=" << m_lcp_curr_bucket
    << " outstanding=" << m_lcp_outstanding);

  m_last_lcp_complete = m_last_lcp;
  ndbrequire(m_lcp_state == LS_LCP_OFF);
  m_lcp_state = LS_LCP_ON;
  do_lcp_loop(signal);
}

Pgman::LCP_STATE
Pgman::process_lcp(Signal* signal)
{
  Page_hashlist& pl_hash = m_page_hashlist;

  int max_count = 0;
  if (m_param.m_max_io_waits > m_stats.m_current_io_waits)
  {
    jam();
    max_count = m_param.m_max_io_waits - m_stats.m_current_io_waits;
    max_count = max_count / 2 + 1;
  }

  D("process_lcp"
    << " this=" << m_last_lcp
    << " last_complete=" << m_last_lcp_complete
    << " bucket=" << m_lcp_curr_bucket
    << " outstanding=" << m_lcp_outstanding);

  // start or re-start from beginning of current hash bucket
  if (m_lcp_curr_bucket != ~(Uint32)0)
  {
    jam();
    Page_hashlist::Iterator iter;
    pl_hash.next(m_lcp_curr_bucket, iter);
    Uint32 loop = 0;
    while (iter.curr.i != RNIL && 
           m_lcp_outstanding < (Uint32) max_count &&
           (loop ++ < 32 || iter.bucket == m_lcp_curr_bucket))
    {
      jam();
      Ptr<Page_entry>& ptr = iter.curr;
      Page_state state = ptr.p->m_state;
      
      DBG_LCP("LCP " << ptr << " - ");
      
      if (ptr.p->m_last_lcp < m_last_lcp &&
          (state & Page_entry::DIRTY) &&
          (! (state & Page_entry::LOCKED)))
      {
        jam();
        if(! (state & Page_entry::BOUND))
        {
          ndbout << ptr << endl;
          ndbrequire(false);
        }
        if (state & Page_entry::BUSY)
        {
          jam();
          DBG_LCP(" BUSY" << endl);
          break;  // wait for it
        } 
        else if (state & Page_entry::PAGEOUT)
        {
          jam();
          DBG_LCP(" PAGEOUT -> state |= LCP" << endl);
          set_page_state(ptr, state | Page_entry::LCP);
        }
        else
        {
          jam();
          DBG_LCP(" pageout()" << endl);
          ptr.p->m_state |= Page_entry::LCP;
          if (c_tup != 0)
            c_tup->disk_page_unmap_callback(0,
                                            ptr.p->m_real_page_i, 
                                            ptr.p->m_dirty_count);
          pageout(signal, ptr);
        }
        ptr.p->m_last_lcp = m_last_lcp;
        m_lcp_outstanding++;
      }
      else
      {
        jam();
        DBG_LCP(" NOT DIRTY" << endl);
      } 
      pl_hash.next(iter);
    }
    
    m_lcp_curr_bucket = (iter.curr.i != RNIL ? iter.bucket : ~(Uint32)0);
  }

  if (m_lcp_curr_bucket == ~(Uint32)0  && !m_lcp_outstanding)
  {
    jam();
    Ptr<Page_entry> ptr;
    Page_sublist& pl = *m_page_sublist[Page_entry::SL_LOCKED];
    if (pl.first(ptr))
    {
      jam();
      process_lcp_locked(signal, ptr);
      return LS_LCP_LOCKED;
    }
    else
    {
      jam();
      if (ERROR_INSERTED(11007))
      {
        ndbout << "No more writes..." << endl;
        signal->theData[0] = 9999;
        sendSignalWithDelay(CMVMI_REF, GSN_NDB_TAMPER, signal, 10000, 1);
        SET_ERROR_INSERT_VALUE(11008);
      }
      EndLcpConf* conf = (EndLcpConf*)signal->getDataPtrSend();
      conf->senderData = m_end_lcp_req.senderData;
      conf->senderRef = reference();
      sendSignal(m_end_lcp_req.senderRef, GSN_END_LCP_CONF,
                 signal, EndLcpConf::SignalLength, JBB);
      return LS_LCP_OFF;
    }
  }

  if (m_lcp_outstanding >= (Uint32) max_count)
  {
    jam();
    return LS_LCP_MAX_LCP_OUTSTANDING;
  }
  
  return LS_LCP_ON;
}

void
Pgman::process_lcp_locked(Signal* signal, Ptr<Page_entry> ptr)
{
  CRASH_INSERTION(11006);

  // protect from tsman parallel access
  Tablespace_client tsman(signal, this, c_tsman, 0, 0, 0);
  ptr.p->m_last_lcp = m_last_lcp;
  if (ptr.p->m_state & Page_entry::DIRTY)
  {
    Ptr<GlobalPage> org, copy;
    ndbrequire(m_global_page_pool.seize(copy));
    m_global_page_pool.getPtr(org, ptr.p->m_real_page_i);
    memcpy(copy.p, org.p, sizeof(GlobalPage));
    ptr.p->m_copy_page_i = copy.i;

    m_lcp_outstanding++;
    ptr.p->m_state |= Page_entry::LCP;
    pageout(signal, ptr);
    return;
  }
  
  Page_sublist& pl = *m_page_sublist[Page_entry::SL_LOCKED];
  pl.next(ptr);
  
  signal->theData[0] = PgmanContinueB::LCP_LOCKED;
  signal->theData[1] = ptr.i;
  sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
}

void
Pgman::process_lcp_locked_fswriteconf(Signal* signal, Ptr<Page_entry> ptr)
{
  Ptr<GlobalPage> org, copy;
  m_global_page_pool.getPtr(copy, ptr.p->m_copy_page_i);
  m_global_page_pool.getPtr(org, ptr.p->m_real_page_i);
  memcpy(org.p, copy.p, sizeof(GlobalPage));
  m_global_page_pool.release(copy);
  ptr.p->m_copy_page_i = RNIL;

  Page_sublist& pl = *m_page_sublist[Page_entry::SL_LOCKED];
  pl.next(ptr);
  
  signal->theData[0] = PgmanContinueB::LCP_LOCKED;
  signal->theData[1] = ptr.i;
  sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
}

// page read and write

void
Pgman::pagein(Signal* signal, Ptr<Page_entry> ptr)
{
  D("pagein");
  D(ptr);

  ndbrequire(! (ptr.p->m_state & Page_entry::PAGEIN));
  set_page_state(ptr, ptr.p->m_state | Page_entry::PAGEIN);

  fsreadreq(signal, ptr);
  m_stats.m_current_io_waits++;
}

void
Pgman::fsreadconf(Signal* signal, Ptr<Page_entry> ptr)
{
  D("fsreadconf");
  D(ptr);

  ndbrequire(ptr.p->m_state & Page_entry::PAGEIN);
  Page_state state = ptr.p->m_state;

  state &= ~ Page_entry::PAGEIN;
  state &= ~ Page_entry::EMPTY;
  state |= Page_entry::MAPPED;
  set_page_state(ptr, state);

  {
    Ptr<GlobalPage> pagePtr;
    m_global_page_pool.getPtr(pagePtr, ptr.p->m_real_page_i);
    File_formats::Datafile::Data_page* page =
      (File_formats::Datafile::Data_page*)pagePtr.p;
    
    Uint64 lsn = 0;
    lsn += page->m_page_header.m_page_lsn_hi; lsn <<= 32;
    lsn += page->m_page_header.m_page_lsn_lo;
    ptr.p->m_lsn = lsn;
  }
  
  ndbrequire(m_stats.m_current_io_waits > 0);
  m_stats.m_current_io_waits--;
  m_stats.m_pages_read++;

  ptr.p->m_last_lcp = m_last_lcp_complete;
  do_busy_loop(signal, true);
}

void
Pgman::pageout(Signal* signal, Ptr<Page_entry> ptr)
{
  D("pageout");
  D(ptr);
  
  Page_state state = ptr.p->m_state;
  ndbrequire(state & Page_entry::BOUND);
  ndbrequire(state & Page_entry::MAPPED);
  ndbrequire(! (state & Page_entry::BUSY));
  ndbrequire(! (state & Page_entry::PAGEOUT));

  state |= Page_entry::PAGEOUT;

  // update lsn on page prior to write
  Ptr<GlobalPage> pagePtr;
  m_global_page_pool.getPtr(pagePtr, ptr.p->m_real_page_i);
  File_formats::Datafile::Data_page* page =
    (File_formats::Datafile::Data_page*)pagePtr.p;
  page->m_page_header.m_page_lsn_hi = (Uint32)(ptr.p->m_lsn >> 32);
  page->m_page_header.m_page_lsn_lo = (Uint32)(ptr.p->m_lsn & 0xFFFFFFFF);

  // undo WAL
  Logfile_client::Request req;
  req.m_callback.m_callbackData = ptr.i;
  req.m_callback.m_callbackIndex = LOGSYNC_CALLBACK;
  D("Logfile_client - pageout");
  Logfile_client lgman(this, c_lgman, RNIL);
  int ret = lgman.sync_lsn(signal, ptr.p->m_lsn, &req, 0);
  if (ret > 0)
  {
    fswritereq(signal, ptr);
    m_stats.m_current_io_waits++;
  }
  else
  {
    ndbrequire(ret == 0);
    m_stats.m_log_waits++;
    state |= Page_entry::LOGSYNC;
  }
  set_page_state(ptr, state);
}

void
Pgman::logsync_callback(Signal* signal, Uint32 ptrI, Uint32 res)
{
  Ptr<Page_entry> ptr;
  m_page_entry_pool.getPtr(ptr, ptrI);

  D("logsync_callback");
  D(ptr);

  // it is OK to be "busy" at this point (the commit is queued)
  Page_state state = ptr.p->m_state;
  ndbrequire(state & Page_entry::PAGEOUT);
  ndbrequire(state & Page_entry::LOGSYNC);
  state &= ~ Page_entry::LOGSYNC;
  set_page_state(ptr, state);

  fswritereq(signal, ptr);
  m_stats.m_current_io_waits++;
}

void
Pgman::fswriteconf(Signal* signal, Ptr<Page_entry> ptr)
{
  D("fswriteconf");
  D(ptr);

  Page_state state = ptr.p->m_state;
  ndbrequire(state & Page_entry::PAGEOUT);

  if (c_tup != 0)
    c_tup->disk_page_unmap_callback(1, 
                                    ptr.p->m_real_page_i, 
                                    ptr.p->m_dirty_count);
  
  state &= ~ Page_entry::PAGEOUT;
  state &= ~ Page_entry::EMPTY;
  state &= ~ Page_entry::DIRTY;

  ndbrequire(m_stats.m_current_io_waits > 0);
  m_stats.m_current_io_waits--;

  if (state & Page_entry::LCP)
  {
    jam();
    state &= ~ Page_entry::LCP;
    ndbrequire(m_lcp_outstanding);
    m_lcp_outstanding--;
    m_stats.m_pages_written_lcp++;
    if (ptr.p->m_copy_page_i != RNIL)
    {
      jam();
      Tablespace_client tsman(signal, this, c_tsman, 0, 0, 0);
      process_lcp_locked_fswriteconf(signal, ptr);
      set_page_state(ptr, state);
      do_busy_loop(signal, true);
      return;
    }
  }
  else
  {
    m_stats.m_pages_written++;
  }
  
  set_page_state(ptr, state);
  do_busy_loop(signal, true);

  if (m_lcp_state == LS_LCP_MAX_LCP_OUTSTANDING)
  {
    jam();
    do_lcp_loop(signal);
  }
}

// file system interface

void
Pgman::fsreadreq(Signal* signal, Ptr<Page_entry> ptr)
{
  File_map::ConstDataBufferIterator it;
  bool ret = m_file_map.first(it) && m_file_map.next(it, ptr.p->m_file_no);
  ndbrequire(ret);
  Uint32 fd = * it.data;

  ndbrequire(ptr.p->m_page_no > 0);

  FsReadWriteReq* req = (FsReadWriteReq*)signal->getDataPtrSend();
  req->filePointer = fd;
  req->userReference = reference();
  req->userPointer = ptr.i;
  req->varIndex = ptr.p->m_page_no;
  req->numberOfPages = 1;
  req->operationFlag = 0;
  FsReadWriteReq::setFormatFlag(req->operationFlag,
                                FsReadWriteReq::fsFormatGlobalPage);
  req->data.pageData[0] = ptr.p->m_real_page_i;
  sendSignal(NDBFS_REF, GSN_FSREADREQ, signal,
             FsReadWriteReq::FixedLength + 1, JBA);
}

void
Pgman::execFSREADCONF(Signal* signal)
{
  jamEntry();
  FsConf* conf = (FsConf*)signal->getDataPtr();
  Ptr<Page_entry> ptr;
  m_page_entry_pool.getPtr(ptr, conf->userPointer);

  fsreadconf(signal, ptr);
}

void
Pgman::execFSREADREF(Signal* signal)
{
  jamEntry();
  SimulatedBlock::execFSREADREF(signal);
  ndbrequire(false);
}

void
Pgman::fswritereq(Signal* signal, Ptr<Page_entry> ptr)
{
  File_map::ConstDataBufferIterator it;
  m_file_map.first(it);
  m_file_map.next(it, ptr.p->m_file_no);
  Uint32 fd = * it.data;

  ndbrequire(ptr.p->m_page_no > 0);

  FsReadWriteReq* req = (FsReadWriteReq*)signal->getDataPtrSend();
  req->filePointer = fd;
  req->userReference = reference();
  req->userPointer = ptr.i;
  req->varIndex = ptr.p->m_page_no;
  req->numberOfPages = 1;
  req->operationFlag = 0;
  FsReadWriteReq::setFormatFlag(req->operationFlag,
                                FsReadWriteReq::fsFormatGlobalPage);
  req->data.pageData[0] = ptr.p->m_real_page_i;

#if ERROR_INSERT_CODE
  if (ptr.p->m_state & Page_entry::LOCKED)
  {
    sendSignalWithDelay(NDBFS_REF, GSN_FSWRITEREQ, signal,
                        3000, FsReadWriteReq::FixedLength + 1);
    ndbout_c("pageout locked (3s)");
    return;
  }
#endif
  
  if (!ERROR_INSERTED(11008))
  {
    sendSignal(NDBFS_REF, GSN_FSWRITEREQ, signal,
               FsReadWriteReq::FixedLength + 1, JBA);
  }
}

void
Pgman::execFSWRITECONF(Signal* signal)
{
  jamEntry();
  FsConf* conf = (FsConf*)signal->getDataPtr();
  Ptr<Page_entry> ptr;
  m_page_entry_pool.getPtr(ptr, conf->userPointer);

  fswriteconf(signal, ptr);
}


void
Pgman::execFSWRITEREF(Signal* signal)
{
  jamEntry();
  SimulatedBlock::execFSWRITEREF(signal);
  ndbrequire(false);
}

// client methods

int
Pgman::get_page_no_lirs(Signal* signal, Ptr<Page_entry> ptr, Page_request page_req)
{
  jamEntry();

#ifdef VM_TRACE
  Ptr<Page_request> tmp = { &page_req, RNIL};

  D(">get_page");
  D(ptr);
  D(tmp);
#endif

  Uint32 req_flags = page_req.m_flags;

  if (req_flags & Page_request::EMPTY_PAGE)
  {
    // Only one can "init" a page at a time
    //ndbrequire(ptr.p->m_requests.isEmpty());
  }

  Page_state state = ptr.p->m_state;
  bool is_new = (state == 0);
  bool busy_count = false;

  if (req_flags & Page_request::LOCK_PAGE)
  {
    jam();
    state |= Page_entry::LOCKED;
  }
  
  if (req_flags & Page_request::ALLOC_REQ)
  {
    jam();
  }
  else if (req_flags & Page_request::COMMIT_REQ)
  {
    busy_count = true;
    state |= Page_entry::BUSY;
  }
  else if ((req_flags & Page_request::OP_MASK) != ZREAD)
  {
    jam();
  }

  const Page_state LOCKED = Page_entry::LOCKED | Page_entry::MAPPED;
  if ((state & LOCKED) == LOCKED && 
      ! (req_flags & Page_request::UNLOCK_PAGE))
  {
    ptr.p->m_state |= (req_flags & DIRTY_FLAGS ? Page_entry::DIRTY : 0);
    m_stats.m_page_requests_direct_return++;
    if (ptr.p->m_copy_page_i != RNIL)
    {
      D("<get_page: immediate copy_page");
      return ptr.p->m_copy_page_i;
    }
    
    D("<get_page: immediate locked");
    return ptr.p->m_real_page_i;
  }
  
  bool only_request = ptr.p->m_requests.isEmpty();
#ifdef ERROR_INSERT
  if (req_flags & Page_request::DELAY_REQ)
  {
    jam();
    only_request = false;
  }
#endif  
  if (only_request &&
      state & Page_entry::MAPPED)
  {
    if (! (state & Page_entry::PAGEOUT))
    {
      if (req_flags & DIRTY_FLAGS)
        state |= Page_entry::DIRTY;
      
      ptr.p->m_busy_count += busy_count;
      set_page_state(ptr, state);
      
      D("<get_page: immediate");

      ndbrequire(ptr.p->m_real_page_i != RNIL);
      m_stats.m_page_requests_direct_return++;
      return ptr.p->m_real_page_i;
    }
  }

  if (! (req_flags & (Page_request::LOCK_PAGE | Page_request::UNLOCK_PAGE)))
  {
    ndbrequire(! (state & Page_entry::LOCKED));
  }

  // queue the request

  if ((state & Page_entry::MAPPED) && ! (state & Page_entry::PAGEOUT))
    m_stats.m_page_requests_wait_q++;
  else
    m_stats.m_page_requests_wait_io++;

  Ptr<Pgman::Page_request> req_ptr;
  {
    Local_page_request_list req_list(m_page_request_pool, ptr.p->m_requests);
    if (! (req_flags & Page_request::ALLOC_REQ))
      req_list.seizeLast(req_ptr);
    else
      req_list.seizeFirst(req_ptr);
  }
  
  if (req_ptr.i == RNIL)
  {
    if (is_new)
    {
      release_page_entry(ptr);
    }
    D("<get_page: error out of requests");
    return -1;
  }

  req_ptr.p->m_block = page_req.m_block;
  req_ptr.p->m_flags = page_req.m_flags;
  req_ptr.p->m_callback = page_req.m_callback;
#ifdef ERROR_INSERT
  req_ptr.p->m_delay_until_time = page_req.m_delay_until_time;
#endif
  
  state |= Page_entry::REQUEST;
  if (only_request && (req_flags & Page_request::EMPTY_PAGE))
  {
    state |= Page_entry::EMPTY;
  }

  if (req_flags & Page_request::UNLOCK_PAGE)
  {
    // keep it locked
  }
  
  ptr.p->m_busy_count += busy_count;
  ptr.p->m_dirty_count += !!(req_flags & DIRTY_FLAGS);
  set_page_state(ptr, state);

  D(req_ptr);
  D("<get_page: queued");
  return 0;
}

int
Pgman::get_page(Signal* signal, Ptr<Page_entry> ptr, Page_request page_req)
{
  int i = get_page_no_lirs(signal, ptr, page_req);
  if (unlikely(i == -1))
  {
    jam();
    return -1;
  }

  Uint32 req_flags = page_req.m_flags;
  Page_state state = ptr.p->m_state;

  // update LIRS
  if (! (state & Page_entry::LOCKED) &&
      ! (req_flags & Page_request::CORR_REQ))
  {
    jam();
    lirs_reference(ptr);
  }

  // start processing if request was queued
  if (i == 0)
  {
    jam();
    do_busy_loop(signal, true);
  }

  return i;
}

void
Pgman::update_lsn(Ptr<Page_entry> ptr, Uint32 block, Uint64 lsn)
{
  jamEntry();
  D(">update_lsn: block=" << hex << block << dec << " lsn=" << lsn);
  D(ptr);

  Page_state state = ptr.p->m_state;
  ptr.p->m_lsn = lsn;
  
  if (state & Page_entry::BUSY)
  {
    ndbrequire(ptr.p->m_busy_count != 0);
    if (--ptr.p->m_busy_count == 0)
    {
      state &= ~ Page_entry::BUSY;
    }
  }
  
  state |= Page_entry::DIRTY;
  set_page_state(ptr, state);
  
  D(ptr);
  D("<update_lsn");
}

Uint32
Pgman::create_data_file()
{
  File_map::DataBufferIterator it;
  if(m_file_map.first(it))
  {
    do 
    {
      if(*it.data == RNIL)
      {
        *it.data = (1u << 31) | it.pos;
        D("create_data_file:" << V(it.pos));
        return it.pos;
      }
    } while(m_file_map.next(it));
  }

  Uint32 file_no = m_file_map.getSize();
  Uint32 fd = (1u << 31) | file_no;

  if (m_file_map.append(&fd, 1))
  {
    D("create_data_file:" << V(file_no));
    return file_no;
  }
  D("create_data_file: RNIL");
  return RNIL;
}

Uint32
Pgman::alloc_data_file(Uint32 file_no)
{
  Uint32 sz = m_file_map.getSize();
  if (file_no >= sz)
  {
    Uint32 len = file_no - sz + 1;
    Uint32 fd = RNIL;
    while (len--)
    {
      if (! m_file_map.append(&fd, 1))
      {
        D("alloc_data_file: RNIL");
        return RNIL;
      }
    }
  }

  File_map::DataBufferIterator it;
  m_file_map.first(it);
  m_file_map.next(it, file_no);
  if (* it.data != RNIL)
  {
    D("alloc_data_file: RNIL");
    return RNIL;
  }

  *it.data = (1u << 31) | file_no;
  D("alloc_data_file:" << V(file_no));
  return file_no;
}

void
Pgman::map_file_no(Uint32 file_no, Uint32 fd)
{
  File_map::DataBufferIterator it;
  m_file_map.first(it);
  m_file_map.next(it, file_no);

  assert(*it.data == ((1u << 31) | file_no));
  *it.data = fd;
  D("map_file_no:" << V(file_no) << V(fd));
}

void
Pgman::free_data_file(Uint32 file_no, Uint32 fd)
{
  File_map::DataBufferIterator it;
  m_file_map.first(it);
  m_file_map.next(it, file_no);
  
  if (fd == RNIL)
  {
    ndbrequire(*it.data == ((1u << 31) | file_no));
  }
  else
  {
    ndbrequire(*it.data == fd);
  }
  *it.data = RNIL;
  D("free_data_file:" << V(file_no) << V(fd));
}

void
Pgman::execDATA_FILE_ORD(Signal* signal)
{
  const DataFileOrd* ord = (const DataFileOrd*)signal->getDataPtr();
  Uint32 ret;
  switch (ord->cmd) {
  case DataFileOrd::CreateDataFile:
    ret = create_data_file();
    ndbrequire(ret == ord->ret);
    break;
  case DataFileOrd::AllocDataFile:
    ret = alloc_data_file(ord->file_no);
    ndbrequire(ret == ord->ret);
    break;
  case DataFileOrd::MapFileNo:
    map_file_no(ord->file_no, ord->fd);
    break;
  case DataFileOrd::FreeDataFile:
    free_data_file(ord->file_no, ord->fd);
    break;
  default:
    ndbrequire(false);
    break;
  }
}

int
Pgman::drop_page(Ptr<Page_entry> ptr)
{
  D("drop_page");
  D(ptr);

  Page_stack& pl_stack = m_page_stack;
  Page_queue& pl_queue = m_page_queue;

  Page_state state = ptr.p->m_state;
  if (! (state & (Page_entry::PAGEIN | Page_entry::PAGEOUT)))
  {
    if (state & Page_entry::ONSTACK)
    {
      jam();
      bool at_bottom = ! pl_stack.hasPrev(ptr);
      pl_stack.remove(ptr);
      state &= ~ Page_entry::ONSTACK;
      if (at_bottom)
      {
        jam();
        lirs_stack_prune();
      }
      if (state & Page_entry::HOT)
      {
        jam();
        state &= ~ Page_entry::HOT;
      }
    }

    if (state & Page_entry::ONQUEUE)
    {
      jam();
      pl_queue.remove(ptr);
      state &= ~ Page_entry::ONQUEUE;
    }

    if (state & Page_entry::BUSY)
    {
      jam();
      state &= ~ Page_entry::BUSY;
    }

    if (state & Page_entry::DIRTY)
    {
      jam();
      state &= ~ Page_entry::DIRTY;
    }

    if (state & Page_entry::EMPTY)
    {
      jam();
      state &= ~ Page_entry::EMPTY;
    }

    if (state & Page_entry::MAPPED)
    {
      jam();
      state &= ~ Page_entry::MAPPED;
    }

    if (state & Page_entry::BOUND)
    {
      jam();
      ndbrequire(ptr.p->m_real_page_i != RNIL);
      release_cache_page(ptr.p->m_real_page_i);
      ptr.p->m_real_page_i = RNIL;
      state &= ~ Page_entry::BOUND;
    }

    set_page_state(ptr, state);
    release_page_entry(ptr);
    return 1;
  }
  
  ndbrequire(false);
  return -1;
}

void
Pgman::execRELEASE_PAGES_REQ(Signal* signal)
{
  const ReleasePagesReq* req = (const ReleasePagesReq*)signal->getDataPtr();
  const Uint32 senderData = req->senderData;
  const Uint32 senderRef = req->senderRef;
  const Uint32 requestType = req->requestType;
  const Uint32 bucket = req->requestData;
  ndbrequire(req->requestType == ReleasePagesReq::RT_RELEASE_UNLOCKED);

  Page_hashlist& pl_hash = m_page_hashlist;
  Page_hashlist::Iterator iter;
  pl_hash.next(bucket, iter);

  Uint32 loop = 0;
  while (iter.curr.i != RNIL && (loop++ < 8 || iter.bucket == bucket))
  {
    jam();
    Ptr<Page_entry> ptr = iter.curr;
    if (!(ptr.p->m_state & Page_entry::LOCKED) &&
        (ptr.p->m_state & Page_entry::BOUND) &&
        (ptr.p->m_state & Page_entry::MAPPED)) // should be
    {
      jam();
      D(ptr << ": release");
      ndbrequire(!(ptr.p->m_state & Page_entry::REQUEST));
      ndbrequire(!(ptr.p->m_state & Page_entry::EMPTY));
      ndbrequire(!(ptr.p->m_state & Page_entry::DIRTY));
      ndbrequire(!(ptr.p->m_state & Page_entry::BUSY));
      ndbrequire(!(ptr.p->m_state & Page_entry::PAGEIN));
      ndbrequire(!(ptr.p->m_state & Page_entry::PAGEOUT));
      ndbrequire(!(ptr.p->m_state & Page_entry::LOGSYNC));
      drop_page(ptr);
    }
    pl_hash.next(iter);
  }

  if (iter.curr.i != RNIL) {
    jam();
    ndbassert(iter.bucket > bucket);
    ReleasePagesReq* req = (ReleasePagesReq*)signal->getDataPtrSend();
    req->senderData = senderData;
    req->senderRef = senderRef;
    req->requestType = requestType;
    req->requestData = iter.bucket;
    sendSignal(reference(), GSN_RELEASE_PAGES_REQ,
               signal, ReleasePagesReq::SignalLength, JBB);
    return;
  }

  ReleasePagesConf* conf = (ReleasePagesConf*)signal->getDataPtrSend();
  conf->senderData = senderData;
  conf->senderRef = reference();
  sendSignal(senderRef, GSN_RELEASE_PAGES_CONF,
             signal, ReleasePagesConf::SignalLength, JBB);
}

// page cache client

#include <PgmanProxy.hpp>

Page_cache_client::Page_cache_client(SimulatedBlock* block,
                                     SimulatedBlock* pgman)
{
  m_block = numberToBlock(block->number(), block->instance());

  if (pgman->isNdbMtLqh() && pgman->instance() == 0) {
    m_pgman_proxy = (PgmanProxy*)pgman;
    m_pgman = 0;
  } else {
    m_pgman_proxy = 0;
    m_pgman = (Pgman*)pgman;
  }
}

int
Page_cache_client::get_page(Signal* signal, Request& req, Uint32 flags)
{
  if (m_pgman_proxy != 0) {
    return m_pgman_proxy->get_page(*this, signal, req, flags);
  }

  Ptr<Pgman::Page_entry> entry_ptr;
  Uint32 file_no = req.m_page.m_file_no;
  Uint32 page_no = req.m_page.m_page_no;

  D("get_page" << V(file_no) << V(page_no) << hex << V(flags));

  // make sure TUP does not peek at obsolete data
  m_ptr.i = RNIL;
  m_ptr.p = 0;

  // find or seize
  bool ok = m_pgman->get_page_entry(entry_ptr, file_no, page_no);
  if (! ok)
  {
    return -1;
  }

  Pgman::Page_request page_req;
  page_req.m_block = m_block;
  page_req.m_flags = flags;
  page_req.m_callback = req.m_callback;
#ifdef ERROR_INSERT
  page_req.m_delay_until_time = req.m_delay_until_time;
#endif
  
  int i = m_pgman->get_page(signal, entry_ptr, page_req);
  if (i > 0)
  {
    // TODO remove
    m_pgman->m_global_page_pool.getPtr(m_ptr, (Uint32)i);
  }
  return i;
}

void
Page_cache_client::update_lsn(Local_key key, Uint64 lsn)
{
  if (m_pgman_proxy != 0) {
    m_pgman_proxy->update_lsn(*this, key, lsn);
    return;
  }

  Ptr<Pgman::Page_entry> entry_ptr;
  Uint32 file_no = key.m_file_no;
  Uint32 page_no = key.m_page_no;

  D("update_lsn" << V(file_no) << V(page_no) << V(lsn));

  bool found = m_pgman->find_page_entry(entry_ptr, file_no, page_no);
  assert(found);

  m_pgman->update_lsn(entry_ptr, m_block, lsn);
}

int
Page_cache_client::drop_page(Local_key key, Uint32 page_id)
{
  if (m_pgman_proxy != 0) {
    return m_pgman_proxy->drop_page(*this, key, page_id);
  }

  Ptr<Pgman::Page_entry> entry_ptr;
  Uint32 file_no = key.m_file_no;
  Uint32 page_no = key.m_page_no;

  D("drop_page" << V(file_no) << V(page_no));

  bool found = m_pgman->find_page_entry(entry_ptr, file_no, page_no);
  assert(found);
  assert(entry_ptr.p->m_real_page_i == page_id);

  return m_pgman->drop_page(entry_ptr);
}

Uint32
Page_cache_client::create_data_file(Signal* signal)
{
  if (m_pgman_proxy != 0) {
    return m_pgman_proxy->create_data_file(signal);
  }
  return m_pgman->create_data_file();
}

Uint32
Page_cache_client::alloc_data_file(Signal* signal, Uint32 file_no)
{
  if (m_pgman_proxy != 0) {
    return m_pgman_proxy->alloc_data_file(signal, file_no);
  }
  return m_pgman->alloc_data_file(file_no);
}

void
Page_cache_client::map_file_no(Signal* signal, Uint32 file_no, Uint32 fd)
{
  if (m_pgman_proxy != 0) {
    m_pgman_proxy->map_file_no(signal, file_no, fd);
    return;
  }
  m_pgman->map_file_no(file_no, fd);
}

void
Page_cache_client::free_data_file(Signal* signal, Uint32 file_no, Uint32 fd)
{
  if (m_pgman_proxy != 0) {
    m_pgman_proxy->free_data_file(signal, file_no, fd);
    return;
  }
  m_pgman->free_data_file(file_no, fd);
}

// debug

#ifdef VM_TRACE

void
Pgman::verify_page_entry(Ptr<Page_entry> ptr)
{
  Page_stack& pl_stack = m_page_stack;

  Uint32 ptrI = ptr.i;
  Page_state state = ptr.p->m_state;

  bool has_req = state & Page_entry::REQUEST;
  bool has_req2 = ! ptr.p->m_requests.isEmpty();
  ndbrequire(has_req == has_req2 || dump_page_lists(ptrI));

  bool is_bound = state & Page_entry::BOUND;
  bool is_bound2 = ptr.p->m_real_page_i != RNIL;
  ndbrequire(is_bound == is_bound2 || dump_page_lists(ptrI));

  bool is_mapped = state & Page_entry::MAPPED;
  // mapped implies bound
  ndbrequire(! is_mapped || is_bound || dump_page_lists(ptrI));
  // bound is mapped or has open requests
  ndbrequire(! is_bound || is_mapped || has_req || dump_page_lists(ptrI));

  bool on_stack = state & Page_entry::ONSTACK;
  bool is_hot = state & Page_entry::HOT;
  // hot entry must be on stack
  ndbrequire(! is_hot || on_stack || dump_page_lists(ptrI));

  // stack bottom is hot
  bool at_bottom = on_stack && ! pl_stack.hasPrev(ptr);
  ndbrequire(! at_bottom || is_hot || dump_page_lists(ptrI));

  bool on_queue = state & Page_entry::ONQUEUE;
  // hot entry is not on queue
  ndbrequire(! is_hot || ! on_queue || dump_page_lists(ptrI));

  bool is_locked = state & Page_entry::LOCKED;
  bool on_queue2 = ! is_locked && ! is_hot && is_bound;
  ndbrequire(on_queue == on_queue2 || dump_page_lists(ptrI));

  // entries waiting to enter queue
  bool to_queue = ! is_locked && ! is_hot && ! is_bound && has_req;

  // page is about to be released
  bool to_release = (state == 0);

  // page is either LOCKED or under LIRS or about to be released
  bool is_lirs = on_stack || to_queue || on_queue;
  ndbrequire(to_release || is_locked == ! is_lirs || dump_page_lists(ptrI));

  bool pagein = state & Page_entry::PAGEIN;
  bool pageout = state & Page_entry::PAGEOUT;
  // cannot read and write at same time
  ndbrequire(! pagein || ! pageout || dump_page_lists(ptrI));

  Uint32 no = get_sublist_no(state);
  switch (no) {
  case Page_entry::SL_BIND:
    ndbrequire((! pagein && ! pageout) || dump_page_lists(ptrI));
    break;
  case Page_entry::SL_MAP:
    ndbrequire((! pagein && ! pageout) || dump_page_lists(ptrI));
    break;
  case Page_entry::SL_MAP_IO:
    ndbrequire((pagein && ! pageout) || dump_page_lists(ptrI));
    break;
  case Page_entry::SL_CALLBACK:
    ndbrequire((! pagein && ! pageout) || dump_page_lists(ptrI));
    break;
  case Page_entry::SL_CALLBACK_IO:
    ndbrequire((! pagein && pageout) || dump_page_lists(ptrI));
    break;
  case Page_entry::SL_BUSY:
    break;
  case Page_entry::SL_LOCKED:
    break;
  case Page_entry::SL_IDLE:
    break;
  case Page_entry::SL_OTHER:
    break;
  case ZNIL:
    ndbrequire(to_release || dump_page_lists(ptrI));
    break;
  default:
    ndbrequire(false || dump_page_lists(ptrI));
    break;
  }
}

void
Pgman::verify_page_lists()
{
  const Stats& stats = m_stats;
  const Param& param = m_param;
  Page_hashlist& pl_hash = m_page_hashlist;
  Page_stack& pl_stack = m_page_stack;
  Page_queue& pl_queue = m_page_queue;
  Ptr<Page_entry> ptr;

  Uint32 is_locked = 0;
  Uint32 is_bound = 0;
  Uint32 is_mapped = 0;
  Uint32 is_hot = 0;
  Uint32 on_stack = 0;
  Uint32 on_queue = 0;
  Uint32 to_queue = 0;

  Page_hashlist::Iterator iter;
  pl_hash.next(0, iter);
  while (iter.curr.i != RNIL)
  {
    ptr = iter.curr;
    Page_state state = ptr.p->m_state;
    // (state == 0) occurs only within a time-slice
    ndbrequire(state != 0);
    verify_page_entry(ptr);

    if (state & Page_entry::LOCKED)
      is_locked++;
    if (state & Page_entry::BOUND)
      is_bound++;
    if (state & Page_entry::MAPPED)
      is_mapped++;
    if (state & Page_entry::HOT)
      is_hot++;
    if (state & Page_entry::ONSTACK)
      on_stack++;
    if (state & Page_entry::ONQUEUE)
      on_queue++;
    if (! (state & Page_entry::LOCKED) &&
        ! (state & Page_entry::HOT) &&
        (state & Page_entry::REQUEST) &&
        ! (state & Page_entry::BOUND))
      to_queue++;
    pl_hash.next(iter);
  }

  for (pl_stack.first(ptr); ptr.i != RNIL; pl_stack.next(ptr))
  {
    Page_state state = ptr.p->m_state;
    ndbrequire(state & Page_entry::ONSTACK || dump_page_lists(ptr.i));
    if (! pl_stack.hasPrev(ptr))
    {
      ndbrequire(state & Page_entry::HOT || dump_page_lists(ptr.i));
    }
  }

  for (pl_queue.first(ptr); ptr.i != RNIL; pl_queue.next(ptr))
  {
    Page_state state = ptr.p->m_state;
    ndbrequire(state & Page_entry::ONQUEUE || dump_page_lists(ptr.i));
    ndbrequire(state & Page_entry::BOUND || dump_page_lists(ptr.i));
    ndbrequire(! (state & Page_entry::HOT) || dump_page_lists(ptr.i));
  }

  ndbrequire(is_bound == stats.m_num_pages || dump_page_lists());
  ndbrequire(is_hot == stats.m_num_hot_pages || dump_page_lists());
  ndbrequire(on_stack == pl_stack.count() || dump_page_lists());
  ndbrequire(on_queue == pl_queue.count() || dump_page_lists());

  Uint32 k;
  Uint32 entry_count = 0;
  char sublist_info[200] = "";
  for (k = 0; k < Page_entry::SUBLIST_COUNT; k++)
  {
    const Page_sublist& pl = *m_page_sublist[k];
    for (pl.first(ptr); ptr.i != RNIL; pl.next(ptr))
      ndbrequire(get_sublist_no(ptr.p->m_state) == k || dump_page_lists(ptr.i));
    entry_count += pl.count();
    sprintf(sublist_info + strlen(sublist_info),
            " %s:%u", get_sublist_name(k), pl.count());
  }
  ndbrequire(entry_count == pl_hash.count() || dump_page_lists());

  Uint32 hit_pct = 0;
  char hit_pct_str[20];
  if (stats.m_page_hits + stats.m_page_faults != 0)
    hit_pct = 10000 * stats.m_page_hits /
              (stats.m_page_hits + stats.m_page_faults);
  sprintf(hit_pct_str, "%u.%02u", hit_pct / 100, hit_pct % 100);

  D("loop"
    << " stats:" << m_stats_loop_on
    << " busy:" << m_busy_loop_on
    << " cleanup:" << m_cleanup_loop_on
    << " lcp:" << Uint32(m_lcp_state));

  D("page"
    << " entries:" << pl_hash.count()
    << " pages:" << stats.m_num_pages << "/" << param.m_max_pages
    << " mapped:" << is_mapped
    << " hot:" << is_hot
    << " io:" << stats.m_current_io_waits << "/" << param.m_max_io_waits
    << " hit pct:" << hit_pct_str);

  D("list"
    << " locked:" << is_locked
    << " stack:" << pl_stack.count()
    << " queue:" << pl_queue.count()
    << " to queue:" << to_queue);

  D(sublist_info);
}

void
Pgman::verify_all()
{
  Page_sublist& pl_bind = *m_page_sublist[Page_entry::SL_BIND];
  Page_sublist& pl_map = *m_page_sublist[Page_entry::SL_MAP];
  Page_sublist& pl_callback = *m_page_sublist[Page_entry::SL_CALLBACK];

  if (! pl_bind.isEmpty() || ! pl_map.isEmpty() || ! pl_callback.isEmpty())
  {
    ndbrequire(m_busy_loop_on || dump_page_lists());
  }
  verify_page_lists();
}

bool
Pgman::dump_page_lists(Uint32 ptrI)
{
  // use debugOut directly
  debugOut << "PGMAN: page list dump" << endl;
  if (ptrI != RNIL)
    debugOut << "PGMAN: error on PE [" << ptrI << "]" << "\n";

  Page_stack& pl_stack = m_page_stack;
  Page_queue& pl_queue = m_page_queue;
  Ptr<Page_entry> ptr;
  Uint32 n;

  debugOut << "stack:" << "\n";
  n = 0;
  for (pl_stack.first(ptr); ptr.i != RNIL; pl_stack.next(ptr))
    debugOut << n++ << " " << ptr << "\n";

  debugOut << "queue:" << "\n";
  n = 0;
  for (pl_queue.first(ptr); ptr.i != RNIL; pl_queue.next(ptr))
    debugOut << n++ << " " << ptr << "\n";

  Uint32 k;
  for (k = 0; k < Page_entry::SUBLIST_COUNT; k++)
  {
    debugOut << get_sublist_name(k) << ":" << "\n";
    const Page_sublist& pl = *m_page_sublist[k];
    n = 0;
    for (pl.first(ptr); ptr.i != RNIL; pl.next(ptr))
      debugOut << n++ << " " << ptr << "\n";
  }

  debugOut.flushline();
  return false;
}

#endif

const char*
Pgman::get_sublist_name(Uint32 list_no)
{
  switch (list_no) {
  case Page_entry::SL_BIND:
    return "bind";
  case Page_entry::SL_MAP:
    return "map";
  case Page_entry::SL_MAP_IO:
    return "map_io";
  case Page_entry::SL_CALLBACK:
    return "cb";
  case Page_entry::SL_CALLBACK_IO:
    return "cb_io";
  case Page_entry::SL_BUSY:
    return "busy";
  case Page_entry::SL_LOCKED:
    return "locked";
  case Page_entry::SL_IDLE:
    return "idle";
  case Page_entry::SL_OTHER:
    return "other";
  }
  return "?";
}

NdbOut&
operator<<(NdbOut& out, Ptr<Pgman::Page_request> ptr)
{
  const Pgman::Page_request& pr = *ptr.p;
  out << "PR";
  if (ptr.i != RNIL)
    out << " [" << dec << ptr.i << "]";
  out << " block=" << hex << pr.m_block;
  out << " flags=" << hex << pr.m_flags;
  out << "," << dec << (pr.m_flags & Pgman::Page_request::OP_MASK);
  {
    if (pr.m_flags & Pgman::Page_request::LOCK_PAGE)
      out << ",lock_page";
    if (pr.m_flags & Pgman::Page_request::EMPTY_PAGE)
      out << ",empty_page";
    if (pr.m_flags & Pgman::Page_request::ALLOC_REQ)
      out << ",alloc_req";
    if (pr.m_flags & Pgman::Page_request::COMMIT_REQ)
      out << ",commit_req";
    if (pr.m_flags & Pgman::Page_request::DIRTY_REQ)
      out << ",dirty_req";
    if (pr.m_flags & Pgman::Page_request::CORR_REQ)
      out << ",corr_req";
  }
  return out;
}

NdbOut&
operator<<(NdbOut& out, Ptr<Pgman::Page_entry> ptr)
{
  const Pgman::Page_entry pe = *ptr.p;
  Uint32 list_no = Pgman::get_sublist_no(pe.m_state);
  out << "PE [" << dec << ptr.i << "]";
  out << " state=" << hex << pe.m_state;
  {
    if (pe.m_state & Pgman::Page_entry::REQUEST)
      out << ",request";
    if (pe.m_state & Pgman::Page_entry::EMPTY)
      out << ",empty";
    if (pe.m_state & Pgman::Page_entry::BOUND)
      out << ",bound";
    if (pe.m_state & Pgman::Page_entry::MAPPED)
      out << ",mapped";
    if (pe.m_state & Pgman::Page_entry::DIRTY)
      out << ",dirty";
    if (pe.m_state & Pgman::Page_entry::USED)
      out << ",used";
    if (pe.m_state & Pgman::Page_entry::BUSY)
      out << ",busy";
    if (pe.m_state & Pgman::Page_entry::LOCKED)
      out << ",locked";
    if (pe.m_state & Pgman::Page_entry::PAGEIN)
      out << ",pagein";
    if (pe.m_state & Pgman::Page_entry::PAGEOUT)
      out << ",pageout";
    if (pe.m_state & Pgman::Page_entry::LOGSYNC)
      out << ",logsync";
    if (pe.m_state & Pgman::Page_entry::LCP)
      out << ",lcp";
    if (pe.m_state & Pgman::Page_entry::HOT)
      out << ",hot";
    if (pe.m_state & Pgman::Page_entry::ONSTACK)
      out << ",onstack";
    if (pe.m_state & Pgman::Page_entry::ONQUEUE)
      out << ",onqueue";
  }
  out << " list=";
  if (list_no == ZNIL)
    out << "NONE";
  else
  {
    out << dec << list_no;
    out << "," << Pgman::get_sublist_name(list_no);
  }
  out << " diskpage=" << dec << pe.m_file_no << "," << pe.m_page_no;
  if (pe.m_real_page_i == RNIL)
    out << " realpage=RNIL";
  else {
    out << " realpage=" << dec << pe.m_real_page_i;
#ifdef VM_TRACE
    if (pe.m_state & Pgman::Page_entry::MAPPED) {
      Ptr<GlobalPage> gptr;
      pe.m_this->m_global_page_pool.getPtr(gptr, pe.m_real_page_i);
      Uint32 hash_result[4];      
      /* NOTE: Assuming "data" is 64 bit aligned as required by 'md5_hash' */
      md5_hash(hash_result,
               (Uint64*)gptr.p->data, sizeof(gptr.p->data)/sizeof(Uint32));
      out.print(" md5=%08x%08x%08x%08x",
                hash_result[0], hash_result[1],
                hash_result[2], hash_result[3]);
    }
#endif
  }
  out << " lsn=" << dec << pe.m_lsn;
  out << " busy_count=" << dec << pe.m_busy_count;
#ifdef VM_TRACE
  {
    Pgman::Page_stack& pl_stack = pe.m_this->m_page_stack;
    if (! pl_stack.hasNext(ptr))
      out << " top";
    if (! pl_stack.hasPrev(ptr))
      out << " bottom";
  }
  {
    Pgman::Local_page_request_list 
      req_list(ptr.p->m_this->m_page_request_pool, ptr.p->m_requests);
    if (! req_list.isEmpty())
    {
      Ptr<Pgman::Page_request> req_ptr;
      out << " req:";
      for (req_list.first(req_ptr); req_ptr.i != RNIL; req_list.next(req_ptr))
      {
        out << " " << req_ptr;
      }
    }
  }
#endif
  return out;
}

void
Pgman::execDUMP_STATE_ORD(Signal* signal)
{
  jamEntry();
  Page_hashlist& pl_hash = m_page_hashlist;
#ifdef VM_TRACE
  if (signal->theData[0] == 11000 && signal->getLength() == 2)
  {
    // has no effect currently
    Uint32 flag = signal->theData[1];
    debugFlag = flag & 1;
    debugSummaryFlag = flag & 2;
  }
#endif

  if (signal->theData[0] == 11001)
  {
    // XXX print hash list if no sublist
    Uint32 list = 0;
    if (signal->getLength() > 1)
      list = signal->theData[1];

    Page_sublist& pl = *m_page_sublist[list];
    Ptr<Page_entry> ptr;
    
    for (pl.first(ptr); ptr.i != RNIL; pl.next(ptr))
    {
      ndbout << ptr << endl;
      infoEvent(" PE [ file: %d page: %d ] state: %x lsn: %lld lcp: %d busy: %d req-list: %d",
                ptr.p->m_file_no, ptr.p->m_page_no,
                ptr.p->m_state, ptr.p->m_lsn, ptr.p->m_last_lcp,
                ptr.p->m_busy_count,
                !ptr.p->m_requests.isEmpty());
    }
  }

  if (signal->theData[0] == 11002 && signal->getLength() == 3)
  {
    Page_entry key;
    key.m_file_no = signal->theData[1];
    key.m_page_no = signal->theData[2];
    
    Ptr<Page_entry> ptr;
    if (pl_hash.find(ptr, key))
    {
      ndbout << "pageout " << ptr << endl;
      if (c_tup != 0)
        c_tup->disk_page_unmap_callback(0,
                                        ptr.p->m_real_page_i, 
                                        ptr.p->m_dirty_count);
      pageout(signal, ptr);
    }
  }
  

  if (signal->theData[0] == 11003)
  {
#ifdef VM_TRACE
    verify_page_lists();
    dump_page_lists();
#else
    ndbout << "Only in VM_TRACE builds" << endl;
#endif
  }

  if (signal->theData[0] == 11004)
  {
    ndbout << "Dump LCP bucket m_lcp_outstanding: " << m_lcp_outstanding;
    if (m_lcp_curr_bucket != ~(Uint32)0)
    {
      Page_hashlist::Iterator iter;
      pl_hash.next(m_lcp_curr_bucket, iter);
      
      ndbout_c(" %d", m_lcp_curr_bucket);

      while (iter.curr.i != RNIL && iter.bucket == m_lcp_curr_bucket)
      {
        Ptr<Page_entry>& ptr = iter.curr;
        ndbout << ptr << endl;
        pl_hash.next(iter);
      }

      ndbout_c("-- done");
    }
    else
    {
      ndbout_c(" == ~0");
    }
  }

  if (signal->theData[0] == 11005)
  {
    g_dbg_lcp = !g_dbg_lcp;
  }

  if (signal->theData[0] == 11006)
  {
    SET_ERROR_INSERT_VALUE(11006);
  }

  if (signal->theData[0] == 11007)
  {
    SET_ERROR_INSERT_VALUE(11007);
  }

  if (signal->theData[0] == 11008)
  {
    SET_ERROR_INSERT_VALUE(11008);
  }

  if (signal->theData[0] == 11009)
  {
    SET_ERROR_INSERT_VALUE(11009);
  }
}

void
Pgman::execDBINFO_SCANREQ(Signal *signal)
{
  DbinfoScanReq req= *(DbinfoScanReq*)signal->theData;
  Ndbinfo::Ratelimit rl;

  jamEntry();
  switch(req.tableId) {
  case Ndbinfo::DISKPAGEBUFFER_TABLEID:
  {
    jam();
    Ndbinfo::Row row(signal, req);
    row.write_uint32(getOwnNodeId());
    row.write_uint32(instance());   // block instance
    row.write_uint64(m_stats.m_pages_written);
    row.write_uint64(m_stats.m_pages_written_lcp);
    row.write_uint64(m_stats.m_pages_read);
    row.write_uint64(m_stats.m_log_waits);
    row.write_uint64(m_stats.m_page_requests_direct_return);
    row.write_uint64(m_stats.m_page_requests_wait_q);
    row.write_uint64(m_stats.m_page_requests_wait_io);

    ndbinfo_send_row(signal, req, row, rl);
  }
  default:
    break;
  }
  ndbinfo_send_scan_conf(signal, req, rl);
}