priotransportertest.cpp

来自「MySQL数据库开发源码 值得一看哦」· C++ 代码 · 共 769 行 · 第 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 "prioTransporterTest.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 = {
  100000, // shmSize
  0,      // shmKey
  1,           // local ndb node id (server)
  2,           // remote ndb node id (client)
  0,              // byteOrder;
  false,          // compression;
  true,           // checksum;
  true            // signalId;
};

TCP_TransporterConfiguration tcpTemplate = {
  17000,          // port; 
  "",             // remoteHostName;
  "",             // localhostname
  2,              // remoteNodeId;
  1,              // localNodeId;
  2000000,          // sendBufferSize - Size of SendBuffer of priority B 
  2000,           // 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 << " localNodeId localHostName" 
	 << " remoteHostName"
	 << " [<loop count>] [<send buf size>] [<recv buf size>]" << 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 * _conf,
		     NodeId localNodeId,
		     NodeId remoteNodeId,
		     const char * localHostName,
		     const char * remoteHostName,
		     int sendBuf,
		     int recvBuf){
  
  ndbout << "Creating OSE transporter from node " 
	 << localNodeId << "(" << localHostName << ") to "
	 << remoteNodeId << "(" << remoteHostName << ")..." << endl;;
  
  OSE_TransporterConfiguration * conf = (OSE_TransporterConfiguration*)_conf;
  
  if(sendBuf != -1){
    conf->prioBSignalSize = sendBuf;
  }
  if(recvBuf != -1){
    conf->receiveBufferSize = recvBuf;
  }

  ndbout << "\tSendBufferSize:    " << conf->prioBSignalSize << endl;
  ndbout << "\tReceiveBufferSize: " << conf->receiveBufferSize << endl;

  conf->localNodeId    = localNodeId;
  conf->localHostName  = localHostName;
  conf->remoteNodeId   = remoteNodeId;
  conf->remoteHostName = remoteHostName;
  bool res = tReg->createTransporter(conf);
  if(res)
    ndbout << "... -- Success " << endl;
  else
    ndbout << "... -- Failure " << endl;
}


void
createSCITransporter(void * _conf,
		     NodeId localNodeId,
		     NodeId remoteNodeId,
		     const char * localHostName,
		     const char * remoteHostName,
		     int sendbuf,
		     int recvbuf) {


  ndbout << "Creating SCI transporter from node " 
	 << localNodeId << "(" << localHostName << ") to "
	 << remoteNodeId << "(" << remoteHostName << ")..." << endl;;
  
  
  SCI_TransporterConfiguration * conf = (SCI_TransporterConfiguration*)_conf;

  conf->remoteSciNodeId0= (Uint16)atoi(localHostName);
  conf->remoteSciNodeId1= (Uint16)atoi(remoteHostName);


  conf->localNodeId    = localNodeId;
  conf->remoteNodeId   = remoteNodeId;

  bool res = tReg->createTransporter(conf);
  if(res)
    ndbout << "... -- Success " << endl;
  else
    ndbout << "... -- Failure " << endl;
}

void
createSHMTransporter(void * _conf,
		     NodeId localNodeId,
		     NodeId remoteNodeId,
		     const char * localHostName,
		     const char * remoteHostName,
		     int sendbuf,
		     int recvbuf) {


  ndbout << "Creating SHM transporter from node " 
	 << localNodeId << "(" << localHostName << ") to "
	 << remoteNodeId << "(" << remoteHostName << ")..." << endl;;
  
  
  SHM_TransporterConfiguration * conf = (SHM_TransporterConfiguration*)_conf;


  conf->localNodeId    = localNodeId;
  conf->remoteNodeId   = remoteNodeId;

  bool res = tReg->createTransporter(conf);
  if(res)
    ndbout << "... -- Success " << endl;
  else
    ndbout << "... -- Failure " << endl;
}


void
createTCPTransporter(void * _conf,
		     NodeId localNodeId,
		     NodeId remoteNodeId,
		     const char * localHostName,
		     const char * remoteHostName,
		     int sendBuf,
		     int recvBuf){
  ndbout << "Creating TCP transporter from node " 
	 << localNodeId << "(" << localHostName << ") to "
	 << remoteNodeId << "(" << remoteHostName << ")..." << endl;;
  
  TCP_TransporterConfiguration * conf = (TCP_TransporterConfiguration*)_conf;
  
  int port;
  if(localNodeId == 1 && remoteNodeId == 2) port = basePortTCP + 0;
  if(localNodeId == 1 && remoteNodeId == 3) port = basePortTCP + 1;
  if(localNodeId == 2 && remoteNodeId == 1) port = basePortTCP + 0;
  if(localNodeId == 2 && remoteNodeId == 3) port = basePortTCP + 2;
  if(localNodeId == 3 && remoteNodeId == 1) port = basePortTCP + 1;
  if(localNodeId == 3 && remoteNodeId == 2) port = basePortTCP + 2;

  if(sendBuf != -1){
    conf->sendBufferSize = sendBuf;
  }
  if(recvBuf != -1){
    conf->maxReceiveSize = recvBuf;
  }

  ndbout << "\tSendBufferSize:    " << conf->sendBufferSize << endl;
  ndbout << "\tReceiveBufferSize: " << conf->maxReceiveSize << endl;

  conf->localNodeId    = localNodeId;
  conf->localHostName  = localHostName;
  conf->remoteNodeId   = remoteNodeId;
  conf->remoteHostName = remoteHostName;
  conf->port           = port;
  bool res = tReg->createTransporter(conf);
  if(res)
    ndbout << "... -- Success " << endl;
  else
    ndbout << "... -- Failure " << endl;
}

struct TestPhase {
  int signalSize;
  int noOfSignals;
  int noOfSignalSent;
  int noOfSignalReceived;
  NDB_TICKS startTime;
  NDB_TICKS stopTime;

  NDB_TICKS startTimePrioA;
  NDB_TICKS stopTimePrioA;
  NDB_TICKS totTimePrioA;
  int bytesSentBeforePrioA;
  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,   10000, 0,0, 0,0,0,0,0,0,0 } //   100 signals of size 1  word
  ,{  1,  10000, 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,   10000, 0,0, 0,0,0,0,0,0,0 } //   100 signals of size 1  word
  ,{  8,  10000, 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
};

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

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

  SignalHeader sh;
  sh.theLength               = signalSize;
  sh.theVerId_signalNumber   = rand();
  sh.theReceiversBlockNumber = rand();   
  sh.theSendersBlockRef      = rand(); 
  sh.theSendersSignalId      = rand(); 
  sh.theSignalId             = rand(); 
  sh.theTrace                = rand(); 
  
  Uint32 theData[25];
  for(int i = 0; i<signalSize; i++)
    theData[i] = (i+1) * (Uint32)(&theData[i]);
  
  return tReg->prepareSend(&sh, prio, theData, nodeId);
}

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

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 bps = ((4000*p.noOfSignals)/p.accTime)*(p.loopCount*(p.signalSize+3));
    if(p.signalSize == 0)
      ((4000*p.noOfSignals)/p.accTime)*(p.loopCount*(13+3));
    
    char ssps[255];
    if(sps > 1000000){
      sps /= 1000000;
      sprintf(ssps, "%dM", (int)sps);
    } else if(sps > 1000){
    sps /= 1000;
    sprintf(ssps, "%dk", (int)sps);
    } else {
      sprintf(ssps, "%d", (int)sps);
    }
    
    char sbps[255];
    if(bps > 1000000){
      bps /= 1000000;
      sprintf(sbps, "%dM", bps);
    } else if(bps>1000){
      bps /= 1000;
      sprintf(sbps, "%dk", bps);
    } else {
      sprintf(sbps, "%d", bps);
  }
    
    char buf[255];
    if(p.signalSize != 0){
      BaseString::snprintf(buf, 255,
	       "%d\t%d\t%d\t%s\t%s\t%s\t%d\t%d\t%d\t%d",
	       p.noOfSignals,
	       p.signalSize,
	       (4*p.signalSize),
	       st,
	       ssps,
	       sbps,