perftransportertest.cpp

来自「MySQL数据库开发源码 值得一看哦」· C++ 代码 · 共 774 行 · 第 1/2 页

/* Copyright (C) 2003 MySQL AB

   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; either version 2 of the License, or
   (at your option) any later version.

   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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

#include <ndb_global.h>

#include "TransporterRegistry.hpp"
#include "TransporterDefinitions.hpp"
#include "TransporterCallback.hpp"
#include <RefConvert.hpp>

#include <NdbTick.h>
#include <NdbMain.h>
#include <NdbOut.hpp>
#include <NdbSleep.h>

int basePortTCP = 17000;

SCI_TransporterConfiguration sciTemplate = {
  2000, 
       // Packet size
  2000000,      // Buffer size
  2,           // number of adapters
  1,           // remote node id SCI 
  2,           // Remote node Id SCI
  0,           // local ndb node id (server)
  0,           // remote ndb node id (client)
  0,              // byteOrder;
  false,          // compression;
  true,          // checksum;
  true            // signalId;
};


SHM_TransporterConfiguration shmTemplate = {
  0,      //remoteNodeId
  0,      //localNodeId;
  false,  //compression
  true,   //checksum;
  true,   //signalId;
  0,      //byteOrder;
  123,    //shmKey;
  25000000 //shmSize;
};


TCP_TransporterConfiguration tcpTemplate = {
  17000,          // port; 
  "",             // remoteHostName;
  "",             // localhostname
  2,              // remoteNodeId;
  1,              // localNodeId;
  25000000,        // sendBufferSize - Size of SendBuffer of priority B 
  5000000,         // maxReceiveSize - Maximum no of bytes to receive
  0,              // byteOrder;
  false,          // compression;
  true,           // checksum;
  true            // signalId;
};

OSE_TransporterConfiguration oseTemplate = {
  "",    // remoteHostName;
  "",    // localHostName;
  0,     // remoteNodeId;
  0,     // localNodeId;
  false, // compression;
  true,  // checksum;
  true,  // signalId;
  0,     // byteOrder;
  
  2000,  // prioASignalSize;
  2000,  // prioBSignalSize;
  10     // Recv buf size
};

TransporterRegistry *tReg = 0;

#ifndef OSE_DELTA
#include <signal.h>
#endif

extern "C"
void
signalHandler(int signo){
#ifndef OSE_DELTA
  ::signal(13, signalHandler);
#endif
  char buf[255];
  sprintf(buf,"Signal: %d\n", signo);
  ndbout << buf << endl;
}

void 
usage(const char * progName){
  ndbout << "Usage: " << progName << " <type> localNodeId localHostName" 
	 << " remoteHostName"
	 << " [<loop count>] [<send buf size>] [<recv buf size>]" << endl;
  ndbout << "  type = shm tcp ose sci" << endl;
  ndbout << "  localNodeId - {1,2}" << endl;
}

typedef void (* CreateTransporterFunc)(void * conf, 
				       NodeId localNodeId,
				       NodeId remoteNodeId,
				       const char * localHostName,
				       const char * remoteHostName,
				       int sendBuf,
				       int recvBuf);

void 
createOSETransporter(void*, NodeId, NodeId, const char*, const char*, int, int);
void
createTCPTransporter(void*, NodeId, NodeId, const char*, const char*, int, int);
void
createSHMTransporter(void*, NodeId, NodeId, const char*, const char*, int, int);
void
createSCITransporter(void*, NodeId, NodeId, const char*, const char*, int, int);

struct TestPhase {
  int signalSize;
  int noOfSignals;
  int noOfSignalSent;
  int noOfSignalReceived;
  NDB_TICKS startTime;
  NDB_TICKS stopTime;
  NDB_TICKS accTime;
  int loopCount;
  Uint64 sendLenBytes, sendCount;
  Uint64 recvLenBytes, recvCount;
};

TestPhase testSpec[] = {
   {  1,    10, 0,0, 0,0,0,0,0,0,0 } //    10 signals of size 1  word
  ,{  1,   100, 0,0, 0,0,0,0,0,0,0 } //   100 signals of size 1  word
  ,{  1,  1000, 0,0, 0,0,0,0,0,0,0 } //  1000 signals of size 1  word
  ,{  1, 10000, 0,0, 0,0,0,0,0,0,0 } // 10000 signals of size 1  word

  ,{  8,    10, 0,0, 0,0,0,0,0,0,0 } //    10 signals of size 1  word
  ,{  8,   100, 0,0, 0,0,0,0,0,0,0 } //   100 signals of size 1  word
  ,{  8,  1000, 0,0, 0,0,0,0,0,0,0 } //  1000 signals of size 1  word
  ,{  8, 10000, 0,0, 0,0,0,0,0,0,0 } // 10000 signals of size 1  word

  ,{ 16,    10, 0,0, 0,0,0,0,0,0,0 } //    10 signals of size 1  word
  ,{ 16,   100, 0,0, 0,0,0,0,0,0,0 } //   100 signals of size 1  word
  ,{ 16,  1000, 0,0, 0,0,0,0,0,0,0 } //  1000 signals of size 1  word
  ,{ 16, 10000, 0,0, 0,0,0,0,0,0,0 } // 10000 signals of size 1  word

  ,{ 24,    10, 0,0, 0,0,0,0,0,0,0 } //    10 signals of size 1  word
  ,{ 24,   100, 0,0, 0,0,0,0,0,0,0 } //   100 signals of size 1  word
  ,{ 24,  1000, 0,0, 0,0,0,0,0,0,0 } //  1000 signals of size 1  word
  ,{ 24, 10000, 0,0, 0,0,0,0,0,0,0 } // 10000 signals of size 1  word

  ,{  0,    10, 0,0, 0,0,0,0,0,0,0 } //    10 signals of random size
  ,{  0,   100, 0,0, 0,0,0,0,0,0,0 } //   100 signals of random size
  ,{  0,  1000, 0,0, 0,0,0,0,0,0,0 } //  1000 signals of random size
  ,{  0, 10000, 0,0, 0,0,0,0,0,0,0 } // 10000 signals of random size

  ,{ 100,    10, 0,0, 0,0,0,0,0,0,0 } //    10 signals 
  ,{ 100,   100, 0,0, 0,0,0,0,0,0,0 } //   100 signals 
  ,{ 100,  1000, 0,0, 0,0,0,0,0,0,0 } //  1000 signals 
  ,{ 100, 10000, 0,0, 0,0,0,0,0,0,0 } // 10000 signals 

  ,{ 500,    10, 0,0, 0,0,0,0,0,0,0 } //    10 signals 
  ,{ 500,   100, 0,0, 0,0,0,0,0,0,0 } //   100 signals 
  ,{ 500,  1000, 0,0, 0,0,0,0,0,0,0 } //  1000 signals 
  ,{ 500, 10000, 0,0, 0,0,0,0,0,0,0 } // 10000 signals 

  ,{ 1000,    10, 0,0, 0,0,0,0,0,0,0 } //    10 signals 
  ,{ 1000,   100, 0,0, 0,0,0,0,0,0,0 } //   100 signals 
  ,{ 1000,  1000, 0,0, 0,0,0,0,0,0,0 } //  1000 signals 
  ,{ 1000, 10000, 0,0, 0,0,0,0,0,0,0 } // 10000 signals 
};

const int noOfTests = sizeof(testSpec)/sizeof(TestPhase);

Uint32 StaticBuffer[1000];

SendStatus
sendSignalTo(NodeId nodeId, int signalSize, Uint32 count){
  if(signalSize == 0)
    signalSize = (rand() % 25) + 1;

  SignalHeader sh;
  sh.theLength               = (signalSize > 25 ? 25 : signalSize);
  sh.theVerId_signalNumber   = count;
  sh.theReceiversBlockNumber = rand();   
  sh.theSendersBlockRef      = rand(); 
  sh.theSendersSignalId      = rand(); 
  sh.theSignalId             = rand(); 
  sh.theTrace                = rand(); 
  
  Uint32 theData[25];
  for(int i = 0; i<25; i++)
    theData[i] = (i+1) * (Uint32)(&theData[i]);
  
  theData[0] = count;
  LinearSectionPtr ptr[3];
  
  if(signalSize <= 25){
    sh.m_noOfSections = 0;
  } else {
    sh.m_noOfSections = 1;
    ptr[0].sz = signalSize - 25;
    ptr[0].p = &StaticBuffer[0];
  }

  return tReg->prepareSend(&sh, 1, theData, nodeId, ptr);
}

void
reportHeader(){
  ndbout << "#Sigs\tSz\tTime\tSig/sec\tBps\tBps-tot\t"
	 << "s len\tr len" << endl;
}

void
print(char * dst, int i){
  if(i > 1000000){
    const int d = i / 1000000;
    const int r = (i - (d * 1000000)) / 100000;
    if(d < 100)
      sprintf(dst, "%d.%dM", d, r);
    else
      sprintf(dst, "%dM", d);
  } else if(i > 1000){
    const int d = i / 1000;
    const int r = (i - (d * 1000)) / 100;
    if(d < 100)
      sprintf(dst, "%d.%dk", d, r);
    else
      sprintf(dst, "%dk", d);
  } else {
    sprintf(dst, "%d", i);
  }
}

void
printReport(TestPhase & p){
  if(p.accTime > 0) {
    Uint32 secs = (p.accTime/p.loopCount)/1000;
    Uint32 mill = (p.accTime/p.loopCount)%1000;
    char st[255];
    if(secs > 0){
      sprintf(st, "%d.%.2ds", secs, (mill/10));
    } else {
      sprintf(st, "%dms", mill);
    }
  
    Uint32 sps = (1000*p.noOfSignals*p.loopCount)/p.accTime;
    Uint32 dps = ((4000*p.noOfSignals)/p.accTime)*(p.loopCount*p.signalSize);
    Uint32 bps = ((4000*p.noOfSignals)/p.accTime)*(p.loopCount*(p.signalSize+3));
    if(p.signalSize == 0){
      dps = ((4000*p.noOfSignals)/p.accTime)*(p.loopCount*(13));
      bps = ((4000*p.noOfSignals)/p.accTime)*(p.loopCount*(13+3));
    }
    char ssps[255];
    char sbps[255];
    char sdps[255];

    print(ssps, sps);
    print(sbps, bps);
    print(sdps, dps);
    
    
    char buf[255];
    if(p.signalSize != 0){
      BaseString::snprintf(buf, 255,
	       "%d\t%d\t%s\t%s\t%s\t%s\t%d\t%d",
	       p.noOfSignals,
	       4*p.signalSize,
	       st,
	       ssps,
	       sdps,
	       sbps,
	       (int)(p.sendLenBytes / (p.sendCount == 0 ? 1 : p.sendCount)),
	       (int)(p.recvLenBytes / (p.recvCount == 0 ? 1 : p.recvCount)));
    } else {
      BaseString::snprintf(buf, 255,
	       "%d\trand\t%s\t%s\t%s\t%s\t%d\t%d",
	       p.noOfSignals,
	       st,
	       ssps,
	       sdps,
	       sbps,
	       (int)(p.sendLenBytes / (p.sendCount == 0 ? 1 : p.sendCount)),
	       (int)(p.recvLenBytes / (p.recvCount == 0 ? 1 : p.recvCount)));
      
    }
    ndbout << buf << endl;
  }
}

int loopCount = 1;
int sendBufSz = -1;
int recvBufSz = -1;

bool      isClient     = false;
bool      isConnected  = false;
bool      isStarted    = false;
int       currentPhase = 0;
TestPhase allPhases[noOfTests];
Uint32    signalToEcho;
Uint32    signalsEchoed;
NDB_TICKS startTime, stopTime;

void
client(NodeId remoteNodeId){
  isClient = true;

  currentPhase = 0;
  memcpy(allPhases, testSpec, sizeof(testSpec));

  int counter = 0;
  int sigCounter = 0;

  while(true){
    TestPhase * current = &allPhases[currentPhase];
    if(current->noOfSignals == current->noOfSignalSent &&
       current->noOfSignals == current->noOfSignalReceived){

      /**
       * Test phase done
       */
      current->stopTime  = NdbTick_CurrentMillisecond();
      current->accTime  += (current->stopTime - current->startTime);

      NdbSleep_MilliSleep(500 / loopCount);
      
      current->startTime = NdbTick_CurrentMillisecond();
      
      current->noOfSignalSent     = 0;
      current->noOfSignalReceived = 0;

      current->loopCount ++;
      if(current->loopCount == loopCount){

	printReport(allPhases[currentPhase]);

	currentPhase ++;
	if(currentPhase == noOfTests){
	  /**
	   * Now we are done
	   */
	  break;
	}
	NdbSleep_MilliSleep(500);
	current = &allPhases[currentPhase];
	current->startTime = NdbTick_CurrentMillisecond();
      }
    } 
    
    int signalsLeft = current->noOfSignals - current->noOfSignalSent;
    if(signalsLeft > 0){
      for(; signalsLeft > 0; signalsLeft--){
	if(sendSignalTo(remoteNodeId,current->signalSize,sigCounter)== SEND_OK){
	  current->noOfSignalSent++;
	  sigCounter++;
	} else {
	  ndbout << "Failed to send: " << sigCounter << endl;
	  tReg->external_IO(10);
	  break;
	}
      }
    }
    if(counter % 10 == 0)
      tReg->checkConnections();
    tReg->external_IO(0);
    counter++;
  }
}

void 
server(){
  isClient = false;