static void createIndex(Ndb &myNdb, bool includePrimary, unsigned int noOfIndexes)
{
  Uint64 before, after;
  NdbDictionary::Dictionary* dict = myNdb.getDictionary();
  char indexName[] = "PNUMINDEX0000";
  int res;

  for(unsigned int indexNum = 0; indexNum < noOfIndexes; indexNum++) {
    sprintf(indexName, "PNUMINDEX%.4u", indexNum);
    NdbDictionary::Index index(indexName);
    index.setTable("PERSON");
    index.setType(NdbDictionary::Index::UniqueHashIndex);
    if (includePrimary) {
      const char* attr_arr[] = {"NAME", "PNUM1", "PNUM3"};
      index.addIndexColumns(3, attr_arr);
    }
    else {
      const char* attr_arr[] = {"PNUM1", "PNUM3"};
      index.addIndexColumns(2, attr_arr);
    }
    before = NdbTick_CurrentMillisecond();
    if ((res = dict->createIndex(index)) == -1) {
      error_handler(dict->getNdbError());
    }    
    after = NdbTick_CurrentMillisecond();
    ndbout << "Created index " << indexName << ", " << after - before << " msec" << endl;
  }
}

void doCopyLap(Uint32 laps, const char * title){

  Uint64 ms1, ms2;
  const Uint32 count = g_count;
  for(Uint32 i = 0; i<count; i++)
    C(i, g_signal);
  laps--;
  for(Uint32 i = 0; i<count; i++)
    C(i, g_signal);
  laps--;
  
  Uint32 div = laps;
  
  ms1 = NdbTick_CurrentMillisecond();
  while(laps > 0){
    for(Uint32 i = 0; i<count; i++){
#if (__GNUC__ == 3 && __GNUC_MINOR >= 1)
      _builtin_prefetch(&g_jobBuffer[i], 1, 0);
#endif
      C(i, g_signal);
    }
    laps--;
  }
  ms2 = NdbTick_CurrentMillisecond();
  
  ms2 -= ms1;
  Uint32 diff = ms2;
  ndbout_c("%s : %d laps in %d millis => %d copies/sec",
	   title, div, diff, (1000*div*g_count+(diff/2))/diff);
}

int InsertInitialRecords(Ndb* pNdb, long nInsert, long nSeed)
{
    int iRes = -1;
    char szMsg[100];
    for(long i=0; i<nInsert; ++i) 
    {
        int iContextID = i+nSeed;
        int nRetry = 0;
        NdbError err;
        memset(&err, 0, sizeof(err));
        NDB_TICKS tStartTrans = NdbTick_CurrentMillisecond();
        iRes = RetryInsertTransaction(pNdb, iContextID, nSeed, iContextID,
            (long)(tStartTrans/1000), (long)((tStartTrans%1000)*1000), 
            STATUS_DATA, err, nRetry);
        NDB_TICKS tEndTrans = NdbTick_CurrentMillisecond();
        long lMillisecForThisTrans = (long)(tEndTrans-tStartTrans);
        if(nRetry>0)
        {
            sprintf(szMsg, "insert retried %d times, time %ld msec.", 
                nRetry, lMillisecForThisTrans);
            ReportNdbError(szMsg, err);
        }
        if(iRes)
        {
            ReportNdbError("Insert initial record failed", err);
            return iRes;
        }
        InterlockedIncrement(&g_nNumberOfInitialInsert);
    }
    return iRes;
}

int VerifyInitialRecords(Ndb* pNdb, long nVerify, long nSeed)
{
    int iRes = -1;
    char* pchContextData = new char[g_nStatusDataSize];
    char szMsg[100];
    long iPrevLockTime = -1;
    long iPrevLockTimeUSec = -1;
    for(long i=0; i<nVerify; ++i) 
    {
        int iContextID = i+nSeed;
        long iVersion = 0;
        long iLockFlag = 0;
        long iLockTime = 0;
        long iLockTimeUSec = 0;
        int nRetry = 0;
        NdbError err;
        memset(&err, 0, sizeof(err));
        NDB_TICKS tStartTrans = NdbTick_CurrentMillisecond();
        iRes = RetryQueryTransaction(pNdb, iContextID, &iVersion, &iLockFlag, 
                    &iLockTime, &iLockTimeUSec, pchContextData, err, nRetry);
        NDB_TICKS tEndTrans = NdbTick_CurrentMillisecond();
        long lMillisecForThisTrans = (long)(tEndTrans-tStartTrans);
        if(nRetry>0)
        {
            sprintf(szMsg, "verify retried %d times, time %ld msec.", 
                nRetry, lMillisecForThisTrans);
            ReportNdbError(szMsg, err);
        }
        if(iRes)
        {
            ReportNdbError("Read initial record failed", err);
            delete[] pchContextData;
            return iRes;
        }
        if(memcmp(pchContextData, STATUS_DATA, g_nStatusDataSize))
        {
            sprintf(szMsg, "wrong context data in tuple %d", iContextID);
            ReportNdbError(szMsg, err);
            delete[] pchContextData;
            return -1;
        }
        if(iVersion!=nSeed 
            || iLockFlag!=iContextID 
            || iLockTime<iPrevLockTime 
            || (iLockTime==iPrevLockTime && iLockTimeUSec<iPrevLockTimeUSec))
        {
            sprintf(szMsg, "wrong call data in tuple %d", iContextID);
            ReportNdbError(szMsg, err);
            delete[] pchContextData;
            return -1;
        }
        iPrevLockTime = iLockTime;
        iPrevLockTimeUSec = iLockTimeUSec;
        InterlockedIncrement(&g_nNumberOfInitialVerify);
    }
    delete[] pchContextData;
    return iRes;
}

int stopOnError(F_ARGS){

  myRandom48Init((long)NdbTick_CurrentMillisecond());

  int randomId = myRandom48(_restarter.getNumDbNodes());
  int nodeId = _restarter.getDbNodeId(randomId);
  
  do {
    g_info << _restart->m_name << ": node = " << nodeId 
	   << endl;
    
    CHECK(_restarter.waitClusterStarted(300) == 0,
	  "waitClusterStarted failed");
    
    int val = DumpStateOrd::NdbcntrTestStopOnError;
    CHECK(_restarter.dumpStateOneNode(nodeId, &val, 1) == 0,
	  "failed to set NdbcntrTestStopOnError");
    
    NdbSleep_SecSleep(3);
    
    CHECK(_restarter.waitClusterStarted(300) == 0,
	  "waitClusterStarted failed");
  } while (false);
  
  return NDBT_OK;
}

//--------------------------------------------------------------------
// ipControlLoop -- The main loop of ndb.
// Handles the scheduling of signal execution and input/output
// One lap in the loop should take approximately 10 milli seconds
// If the jobbuffer is empty and the laptime is less than 10 milliseconds
// at the end of the loop
// the TransporterRegistry is called in order to sleep on the IO ports
// waiting for another incoming signal to wake us up.
// The timeout value in this call is calculated as (10 ms - laptime)
// This would make ndb use less cpu while improving response time.
//--------------------------------------------------------------------
void ThreadConfig::ipControlLoop()
{

//--------------------------------------------------------------------
// initialise the counter that keeps track of the current millisecond
//--------------------------------------------------------------------
  globalData.internalMillisecCounter = NdbTick_CurrentMillisecond();
  Uint32 i = 0;
  while (globalData.theRestartFlag != perform_stop)  { 

    Uint32 timeOutMillis = 0;
    if (LEVEL_IDLE == globalData.highestAvailablePrio) {
//--------------------------------------------------------------------
// The buffers are empty, we need to wait for a while until we continue.
// We cannot wait forever since we can also have timed events.
//--------------------------------------------------------------------
//--------------------------------------------------------------------
// Set the time we will sleep on the sockets before waking up
// unconditionally to 10 ms. Will never sleep more than 10 milliseconds
// on a socket.
//--------------------------------------------------------------------
      timeOutMillis = 10;
    }//if
//--------------------------------------------------------------------
// Now it is time to check all interfaces. We will send all buffers
// plus checking for any received messages.
//--------------------------------------------------------------------
    if (i++ >= 20) {
      globalTransporterRegistry.update_connections();
      globalData.incrementWatchDogCounter(5);
      i = 0;
    }//if

    globalData.incrementWatchDogCounter(6);
    globalTransporterRegistry.performSend();
    
    globalData.incrementWatchDogCounter(7);
    if (globalTransporterRegistry.pollReceive(timeOutMillis)) {
      globalData.incrementWatchDogCounter(8);
      globalTransporterRegistry.performReceive();
    }

//--------------------------------------------------------------------
// We scan the time queue to see if there are any timed signals that
// is now ready to be executed.
//--------------------------------------------------------------------
    globalData.incrementWatchDogCounter(2);
    scanTimeQueue(); 

//--------------------------------------------------------------------
// This is where the actual execution of signals occur. We execute
// until all buffers are empty or until we have executed 2048 signals.
//--------------------------------------------------------------------
    globalScheduler.doJob();
  }//while

  globalData.incrementWatchDogCounter(6);
  globalTransporterRegistry.performSend();

}//ThreadConfig::ipControlLoop()

/**
 * For each millisecond that has passed since this function was last called:
 *   Scan the job buffer and increment the internalMillisecCounter 
 *      with 1 to keep track of where we are
 */
inline
void 
ThreadConfig::scanTimeQueue()
{
  unsigned int maxCounter;
  Uint64 currMilliSecond;
  maxCounter = 0;
  currMilliSecond = NdbTick_CurrentMillisecond();
  if (currMilliSecond < globalData.internalMillisecCounter) {
//--------------------------------------------------------------------
// This could occur around 2036 or if the operator decides to change
// time backwards. We cannot know how long time has past since last
// time and we make a best try with 0 milliseconds.
//--------------------------------------------------------------------
    g_eventLogger->warning("Time moved backwards with %llu ms",
                           globalData.internalMillisecCounter-currMilliSecond);
    globalData.internalMillisecCounter = currMilliSecond;
  }//if
  if (currMilliSecond > (globalData.internalMillisecCounter + 1500)) {
//--------------------------------------------------------------------
// Time has moved forward more than a second. Either it could happen
// if operator changed the time or if the OS has misbehaved badly.
// We set the new time to one second from the past.
//--------------------------------------------------------------------
    g_eventLogger->warning("Time moved forward with %llu ms",
                           currMilliSecond-globalData.internalMillisecCounter);
    globalData.internalMillisecCounter = currMilliSecond - 1000;
  }//if
  while (((currMilliSecond - globalData.internalMillisecCounter) > 0) &&
         (maxCounter < 20)){
    globalData.internalMillisecCounter++;
    maxCounter++;
    globalTimeQueue.scanTable();
  }//while
}//ThreadConfig::scanTimeQueue()

int getRandomNodeId(NdbRestarter& _restarter) {
  myRandom48Init((long)NdbTick_CurrentMillisecond());
  int randomId = myRandom48(_restarter.getNumDbNodes());
  int nodeId = _restarter.getDbNodeId(randomId);

  return nodeId;
}

bool
NDBT_Context::closeToTimeout(int safety)
{
  if (safety == 0)
    return false;

  if (m_env_timeout == 0)
  {
    char buf[1024];
    const char * p = NdbEnv_GetEnv("ATRT_TIMEOUT", buf, sizeof(buf));
    if (p)
    {
      m_env_timeout = atoi(p);
      ndbout_c("FOUND ATRT_TIMEOUT: %d", m_env_timeout);
    }
    else
    {
      m_env_timeout = -1;
    }
  }

  if (m_env_timeout < 0)
    return false;

  Uint64 to = (1000 * m_env_timeout * (100 - safety)) / 100;
  Uint64 now = NdbTick_CurrentMillisecond();
  if (now >= m_test_start_time + to)
  {
    ndbout_c("closeToTimeout(%d) => true env(timeout): %d",
             safety, m_env_timeout);
    return true;
  }

  return false;
}

int
write_socket(NDB_SOCKET_TYPE socket, int timeout_millis, int *time,
	     const char buf[], int len){
  fd_set writeset;
  FD_ZERO(&writeset);
  FD_SET(socket, &writeset);
  struct timeval timeout;
  timeout.tv_sec  = (timeout_millis / 1000);
  timeout.tv_usec = (timeout_millis % 1000) * 1000;


  Uint64 tick= NdbTick_CurrentMillisecond();
  const int selectRes = select(socket + 1, 0, &writeset, 0, &timeout);
  *time= NdbTick_CurrentMillisecond() - tick;

  if(selectRes != 1){
    return -1;
  }

  const char * tmp = &buf[0];
  while(len > 0){
    const int w = send(socket, tmp, len, 0);
    if(w == -1){
      return -1;
    }
    len -= w;
    tmp += w;
    
    if(len == 0)
      break;
    
    FD_ZERO(&writeset);
    FD_SET(socket, &writeset);
    timeout.tv_sec  = ((timeout_millis - *time) / 1000);
    timeout.tv_usec = ((timeout_millis - *time) % 1000) * 1000;

    Uint64 tick= NdbTick_CurrentMillisecond();
    const int selectRes2 = select(socket + 1, 0, &writeset, 0, &timeout);
    *time= NdbTick_CurrentMillisecond() - tick;

    if(selectRes2 != 1){
      return -1;
    }
  }
  
  return 0;
}

static void dropIndex(Ndb &myNdb, unsigned int noOfIndexes)
{
  for(unsigned int indexNum = 0; indexNum < noOfIndexes; indexNum++) {
    char indexName[255];
    sprintf(indexName, "PNUMINDEX%.4u", indexNum);
    const Uint64 before = NdbTick_CurrentMillisecond();
    const int retVal = myNdb.getDictionary()->dropIndex(indexName, "PERSON");
    const Uint64 after = NdbTick_CurrentMillisecond();
    
    if(retVal == 0){
      ndbout << "Dropped index " << indexName << ", " 
	     << after - before << " msec" << endl;
    } else {
      ndbout << "Failed to drop index " << indexName << endl;
      ndbout << myNdb.getDictionary()->getNdbError() << endl;
    }
  }
}

int
main(void){
  
  srand(NdbTick_CurrentMillisecond());

  //test( 1, 1000, 1000);
  test(54, 1000, 1000);
  test(59, 1000, 1000);
  test(60, 1000, 1000);
  test(61, 1000, 1000);
  return 0;
}

SimpleSignal *
SignalSender::waitFor(Uint32 timeOutMillis, T & t)
{
  SimpleSignal * s = t.check(m_jobBuffer);
  if(s != 0){
    if (m_usedBuffer.push_back(s))
    {
      return 0;
    }
    assert(s->header.theLength > 0);
    return s;
  }

  /* Remove old signals from usedBuffer */
  for (unsigned i= 0; i < m_usedBuffer.size(); i++)
    delete m_usedBuffer[i];
  m_usedBuffer.clear();

  NDB_TICKS now = NdbTick_CurrentMillisecond();
  NDB_TICKS stop = now + timeOutMillis;
  Uint32 wait = (timeOutMillis == 0 ? 10 : timeOutMillis);
  do {
    do_poll(wait);
    
    SimpleSignal * s = t.check(m_jobBuffer);
    if(s != 0){
      if (m_usedBuffer.push_back(s))
      {
        return 0;
      }
      assert(s->header.theLength > 0);
      return s;
    }
    
    now = NdbTick_CurrentMillisecond();
    wait = (Uint32)(timeOutMillis == 0 ? 10 : stop - now);
  } while(stop > now || timeOutMillis == 0);
  
  return 0;
} 

extern "C" void* NdbThreadFuncUpdate(void* pArg)
{
    myRandom48Init((long int)NdbTick_CurrentMillisecond());
    unsigned nSucc = 0;
    unsigned nFail = 0;
    Ndb* pNdb = NULL ;
    pNdb = new Ndb("TEST_DB");
    VerifyMethodInt(pNdb, init());
    VerifyMethodInt(pNdb, waitUntilReady());

    while(NdbMutex_Trylock(g_pNdbMutex)) {
        Uint32 nWarehouse = myRandom48(g_nWarehouseCount);
        NdbConnection* pNdbConnection = NULL ;
        VerifyMethodPtr(pNdbConnection, pNdb, startTransaction());
        CHK_TR(pNdbConnection) ; // epaulsa
        NdbOperation* pNdbOperationW = NULL ;
        VerifyMethodPtr(pNdbOperationW, pNdbConnection, getNdbOperation(c_szWarehouse));
        VerifyMethodInt(pNdbOperationW, interpretedUpdateTuple());
        VerifyMethodInt(pNdbOperationW, equal(c_szWarehouseNumber, nWarehouse));
        VerifyMethodInt(pNdbOperationW, incValue(c_szWarehouseCount, Uint32(1)));
        Uint32 nWarehouseSum = 0;
        for(Uint32 nDistrict=0; nDistrict<g_nDistrictPerWarehouse; ++nDistrict) {
            NdbOperation* pNdbOperationD = NULL ;
            VerifyMethodPtr(pNdbOperationD, pNdbConnection, getNdbOperation(c_szDistrict));
            VerifyMethodInt(pNdbOperationD, interpretedUpdateTuple());
            VerifyMethodInt(pNdbOperationD, equal(c_szDistrictWarehouseNumber, nWarehouse));
            VerifyMethodInt(pNdbOperationD, equal(c_szDistrictNumber, nDistrict));
            VerifyMethodInt(pNdbOperationD, incValue(c_szDistrictCount, Uint32(1)));
            Uint32 nDistrictSum = myRandom48(100);
            nWarehouseSum += nDistrictSum;
            VerifyMethodInt(pNdbOperationD, setValue(c_szDistrictSum, nDistrictSum));
        }
        VerifyMethodInt(pNdbOperationW, setValue(c_szWarehouseSum, nWarehouseSum));
        int iExec = pNdbConnection->execute(Commit);
        int iError = pNdbConnection->getNdbError().code;

        if(iExec<0 && iError!=0 && iError!=266 && iError!=626) {
            ReportMethodInt(iExec, pNdbConnection, "pNdbConnection", "execute(Commit)", __FILE__, __LINE__);
        }
        if(iExec==0) {
            ++nSucc;
        } else {
            ++nFail;
        }
        VerifyMethodVoid(pNdb, closeTransaction(pNdbConnection));
    }
    ndbout << "update: " << nSucc << " succeeded, " << nFail << " failed " << endl;
    NdbMutex_Unlock(g_pNdbMutex);
    delete pNdb;
    pNdb = NULL ;
    return NULL;
}

int restartRandomNodeInitial(F_ARGS){

  myRandom48Init((long)NdbTick_CurrentMillisecond());
  int randomId = myRandom48(_restarter.getNumDbNodes());
  int nodeId = _restarter.getDbNodeId(randomId);
  
  g_info << _restart->m_name << ": node = "<<nodeId << endl;

  CHECK(_restarter.restartOneDbNode(nodeId, true) == 0,
	"Could not restart node "<<nodeId);

  return NDBT_OK;
}

int restartRandomNodeError(F_ARGS){

  myRandom48Init((long)NdbTick_CurrentMillisecond());
  int randomId = myRandom48(_restarter.getNumDbNodes());
  int nodeId = _restarter.getDbNodeId(randomId);
  
  ndbout << _restart->m_name << ": node = "<<nodeId << endl;

  CHECK(_restarter.insertErrorInNode(nodeId, 9999) == 0, 
	"Could not restart node "<<nodeId);

  return NDBT_OK;
}

void InterlockedIncrementAndReport(void)
{
    NdbMutex_Lock(g_pNdbMutexIncrement);
    ++g_nNumCallsProcessed;
    if((g_nNumCallsProcessed%1000)==0) 
    {
        g_tEndTime = NdbTick_CurrentMillisecond();
        if(g_tStartTime) 
            ReportCallsPerSecond(1000, g_tStartTime, g_tEndTime);

        g_tStartTime = g_tEndTime;
    }
    NdbMutex_Unlock(g_pNdbMutexIncrement);
}

NDBT_Context::NDBT_Context(Ndb_cluster_connection& con)
  : m_cluster_connection(con)
{
  suite = NULL;
  testcase = NULL;
  ndb = NULL;
  records = 1;
  loops = 1;
  stopped = false;
  propertyMutexPtr = NdbMutex_Create();
  propertyCondPtr = NdbCondition_Create();
  m_env_timeout = 0;
  m_test_start_time = NdbTick_CurrentMillisecond();
}

int 
doTime(Uint32 ms){
  
  Uint64 ms1, ms2;
  const Uint32 count = g_count;
  for(Uint32 i = 0; i<count; i++)
    C(i, g_signal);
  for(Uint32 i = 0; i<count; i++)
    C(i, g_signal);
  
  Uint32 laps = 0;
  
  ms1 = NdbTick_CurrentMillisecond();
  do {
    for(int j = 100; j>= 0; j--)
      for(Uint32 i = 0; i<count; i++){
	C(i, g_signal);
      }
    ms2 = NdbTick_CurrentMillisecond();
    laps += 100;
  } while((ms2 - ms1) < ms);
  
  return laps;
}

void
Restore::release_file(FilePtr file_ptr)
{
  LocalDataBuffer<15> pages(m_databuffer_pool, file_ptr.p->m_pages);
  LocalDataBuffer<15> columns(m_databuffer_pool, file_ptr.p->m_columns);

  List::Iterator it;
  for (pages.first(it); !it.isNull(); pages.next(it))
  {
    if (* it.data == RNIL)
      continue;
    m_global_page_pool.release(* it.data);
  }

  {
    Uint64 millis = NdbTick_CurrentMillisecond() -
                   file_ptr.p->m_restore_start_time;
    if (millis == 0)
      millis = 1;
    Uint64 bps = file_ptr.p->m_bytes_restored * 1000 / millis;

    g_eventLogger->info("LDM instance %u: Restored T%dF%u LCP %llu rows, "
                        "%llu bytes, %llu millis, %llu bytes/s)", 
                        instance(),
                        file_ptr.p->m_table_id,
                        file_ptr.p->m_fragment_id,
                        file_ptr.p->m_rows_restored,
                        file_ptr.p->m_bytes_restored,
                        millis,
                        bps);

    m_rows_restored+= file_ptr.p->m_rows_restored;
    m_bytes_restored+= file_ptr.p->m_bytes_restored;
    m_millis_spent+= millis;
    m_frags_restored++;
  }
  
  columns.release();
  pages.release();
  m_file_list.release(file_ptr);
}