开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 数据库系统 > MySQL数据库 > 查看源码
FastScheduler.cpp
资源名称：mysql-4.1.16-win-src.zip　[点击查看]
上传用户：romrleung
上传日期：2022-05-23
资源大小：18897k
文件大小：14k
源码类别：

MySQL数据库

开发平台：

Visual C++

FastScheduler.cpp：源码内容
  1. /* Copyright (C) 2003 MySQL AB
  3.    This program is free software; you can redistribute it and/or modify
  4.    it under the terms of the GNU General Public License as published by
  5.    the Free Software Foundation; either version 2 of the License, or
  6.    (at your option) any later version.
  8.    This program is distributed in the hope that it will be useful,
  9.    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10.    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11.    GNU General Public License for more details.
  13.    You should have received a copy of the GNU General Public License
  14.    along with this program; if not, write to the Free Software
  15.    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
  17. #include "FastScheduler.hpp"
  18. #include "RefConvert.hpp"
  20. #include "Emulator.hpp"
  21. #include "VMSignal.hpp"
  22. #include <Error.hpp>
  24. #include <SignalLoggerManager.hpp>
  25. #include <BlockNumbers.h>
  26. #include <GlobalSignalNumbers.h>
  27. #include <signaldata/EventReport.hpp>
  28. #include "LongSignal.hpp"
  29. #include <NdbTick.h>
  31. #define MIN_NUMBER_OF_SIG_PER_DO_JOB 64
  32. #define MAX_NUMBER_OF_SIG_PER_DO_JOB 2048
  33. #define EXTRA_SIGNALS_PER_DO_JOB 32
  35. FastScheduler::FastScheduler()
  36. {
  37.    // These constants work for sun only, but they should be initated from
  38.    // Emulator.C as soon as VMTime has been initiated.
  39.    theJobBuffers[0].newBuffer(JBASIZE);
  40.    theJobBuffers[1].newBuffer(JBBSIZE);
  41.    theJobBuffers[2].newBuffer(JBCSIZE);
  42.    theJobBuffers[3].newBuffer(JBDSIZE);
  43.    clear();
  44. }
  46. FastScheduler::~FastScheduler()
  47. {
  48. }
  50. void 
  51. FastScheduler::clear()
  52. {
  53.   int i;
  54.   // Make sure the restart signals are not sent too early
  55.   // the prio is set back in 'main' using the 'ready' method.
  56.   globalData.highestAvailablePrio = LEVEL_IDLE;
  57.   globalData.sendPackedActivated = 0;
  58.   globalData.activateSendPacked = 0;
  59.   for (i = 0; i < JB_LEVELS; i++){
  60.     theJobBuffers[i].clear();
  61.   }
  62.   globalData.JobCounter = 0;
  63.   globalData.JobLap = 0;
  64.   globalData.loopMax = 32;
  65.   globalData.VMSignals[0].header.theSignalId = 0;
  66.   
  67.   theDoJobTotalCounter = 0;
  68.   theDoJobCallCounter = 0;
  69. }
  71. void
  72. FastScheduler::activateSendPacked()
  73. {
  74.   globalData.sendPackedActivated = 1;
  75.   globalData.activateSendPacked = 0;
  76.   globalData.loopMax = 2048;
  77. }//FastScheduler::activateSendPacked()
  79. //------------------------------------------------------------------------
  80. // sendPacked is executed at the end of the loop.
  81. // To ensure that we don't send any messages before executing all local
  82. // packed signals we do another turn in the loop (unless we have already
  83. // executed too many signals in the loop).
  84. //------------------------------------------------------------------------
  85. void 
  86. FastScheduler::doJob()
  87. {
  88.   Uint32 loopCount = 0;
  89.   Uint32 TminLoops = getBOccupancy() + EXTRA_SIGNALS_PER_DO_JOB;
  90.   Uint32 TloopMax = (Uint32)globalData.loopMax;
  91.   if (TminLoops < TloopMax) {
  92.     TloopMax = TminLoops;
  93.   }//if
  94.   if (TloopMax < MIN_NUMBER_OF_SIG_PER_DO_JOB) {
  95.     TloopMax = MIN_NUMBER_OF_SIG_PER_DO_JOB;
  96.   }//if
  97.   register Signal* signal = getVMSignals();
  98.   register Uint32 tHighPrio= globalData.highestAvailablePrio;
  99.   do{
  100.     while ((tHighPrio < LEVEL_IDLE) && (loopCount < TloopMax)) {
  101.       // signal->garbage_register(); 
  102.       // To ensure we find bugs quickly
  103.       register Uint32 gsnbnr = theJobBuffers[tHighPrio].retrieve(signal);
  104.       register BlockNumber reg_bnr = gsnbnr & 0xFFF;
  105.       register GlobalSignalNumber reg_gsn = gsnbnr >> 16;
  106.       globalData.incrementWatchDogCounter(1);
  107.       if (reg_bnr > 0) {
  108.         Uint32 tJobCounter = globalData.JobCounter;
  109.         Uint32 tJobLap = globalData.JobLap;
  110.         SimulatedBlock* b = globalData.getBlock(reg_bnr);
  111.         theJobPriority[tJobCounter] = (Uint8)tHighPrio;
  112.         globalData.JobCounter = (tJobCounter + 1) & 4095;
  113.         globalData.JobLap = tJobLap + 1;
  115. #ifdef VM_TRACE_TIME
  116. Uint32 us1, us2;
  117. Uint64 ms1, ms2;
  118. NdbTick_CurrentMicrosecond(&ms1, &us1);
  119. b->m_currentGsn = reg_gsn;
  120. #endif
  122. getSections(signal->header.m_noOfSections, signal->m_sectionPtr);
  123. #ifdef VM_TRACE
  124.         {
  125.           if (globalData.testOn) {
  126.     signal->header.theVerId_signalNumber = reg_gsn;
  127.     signal->header.theReceiversBlockNumber = reg_bnr;
  128.     
  129.             globalSignalLoggers.executeSignal(signal->header,
  130.       tHighPrio,
  131.       &signal->theData[0], 
  132.       globalData.ownId,
  133.                                               signal->m_sectionPtr,
  134.                                               signal->header.m_noOfSections);
  135.           }//if
  136.         }
  137. #endif
  138.         b->executeFunction(reg_gsn, signal);
  139. releaseSections(signal->header.m_noOfSections, signal->m_sectionPtr);
  140. signal->header.m_noOfSections = 0;
  141. #ifdef VM_TRACE_TIME
  142. NdbTick_CurrentMicrosecond(&ms2, &us2);
  143. Uint64 diff = ms2;
  144. diff -= ms1;
  145. diff *= 1000000;
  146. diff += us2;
  147. diff -= us1;
  148. b->addTime(reg_gsn, diff);
  149. #endif
  150.         tHighPrio = globalData.highestAvailablePrio;
  151.       } else {
  152.         tHighPrio++;
  153.         globalData.highestAvailablePrio = tHighPrio;
  154.       }//if
  155.       loopCount++;
  156.     }//while
  157.     sendPacked();
  158.     tHighPrio = globalData.highestAvailablePrio;
  159.     if(getBOccupancy() > MAX_OCCUPANCY)
  160.     {
  161.       if(loopCount != TloopMax)
  162. abort();
  163.       assert( loopCount == TloopMax );
  164.       TloopMax += 512;
  165.     }
  166.   } while ((getBOccupancy() > MAX_OCCUPANCY) ||
  167.            ((loopCount < TloopMax) &&
  168.             (tHighPrio < LEVEL_IDLE)));
  170.   theDoJobCallCounter ++;
  171.   theDoJobTotalCounter += loopCount;
  172.   if (theDoJobCallCounter == 8192) {
  173.     reportDoJobStatistics(theDoJobTotalCounter >> 13);
  174.     theDoJobCallCounter = 0;
  175.     theDoJobTotalCounter = 0;
  176.   }//if
  178. }//FastScheduler::doJob()
  180. void FastScheduler::sendPacked()
  181. {
  182.   if (globalData.sendPackedActivated == 1) {
  183.     SimulatedBlock* b_lqh = globalData.getBlock(DBLQH);
  184.     SimulatedBlock* b_tc = globalData.getBlock(DBTC);
  185.     SimulatedBlock* b_tup = globalData.getBlock(DBTUP);
  186.     Signal* signal = getVMSignals();
  187.     b_lqh->executeFunction(GSN_SEND_PACKED, signal);
  188.     b_tc->executeFunction(GSN_SEND_PACKED, signal);
  189.     b_tup->executeFunction(GSN_SEND_PACKED, signal);
  190.     return;
  191.   } else if (globalData.activateSendPacked == 0) {
  192.     return;
  193.   } else {
  194.     activateSendPacked();
  195.   }//if
  196.   return;
  197. }//FastScheduler::sendPacked()
  199. Uint32
  200. APZJobBuffer::retrieve(Signal* signal)
  201. {              
  202.   Uint32 tOccupancy = theOccupancy;
  203.   Uint32 myRPtr = rPtr;
  204.   BufferEntry& buf = buffer[myRPtr];
  205.   Uint32 gsnbnr;
  206.   Uint32 cond =  (++myRPtr == bufSize) - 1;
  207.   Uint32 tRecBlockNo = buf.header.theReceiversBlockNumber;
  208.   
  209.   if (tOccupancy != 0) {
  210.     if (tRecBlockNo != 0) {
  211.       // Transform protocol to signal. 
  212.       rPtr = myRPtr & cond;
  213.       theOccupancy = tOccupancy - 1;
  214.       gsnbnr = buf.header.theVerId_signalNumber << 16 | tRecBlockNo;
  215.       
  216.       Uint32 tSignalId = globalData.theSignalId;
  217.       Uint32 tLength = buf.header.theLength;
  218.       Uint32 tFirstData = buf.theDataRegister[0];
  219.       signal->header = buf.header;
  220.       
  221.       // Recall our signal Id for restart purposes
  222.       buf.header.theSignalId = tSignalId;  
  223.       globalData.theSignalId = tSignalId + 1;
  224.       
  225.       Uint32* tDataRegPtr = &buf.theDataRegister[0];
  226.       Uint32* tSigDataPtr = signal->getDataPtrSend();
  227.       *tSigDataPtr = tFirstData;
  228.       tDataRegPtr++;
  229.       tSigDataPtr++;
  230.       Uint32  tLengthCopied = 1;
  231.       while (tLengthCopied < tLength) {
  232.         Uint32 tData0 = tDataRegPtr[0];
  233.         Uint32 tData1 = tDataRegPtr[1];
  234.         Uint32 tData2 = tDataRegPtr[2];
  235.         Uint32 tData3 = tDataRegPtr[3];
  237.         tDataRegPtr += 4;
  238.         tLengthCopied += 4;
  240.         tSigDataPtr[0] = tData0;
  241.         tSigDataPtr[1] = tData1;
  242.         tSigDataPtr[2] = tData2;
  243.         tSigDataPtr[3] = tData3;
  244.         tSigDataPtr += 4;
  245.       }//while
  247.       /**
  248.        * Copy sections references (copy all without if-statements)
  249.        */
  250.       tDataRegPtr = &buf.theDataRegister[tLength];
  251.       SegmentedSectionPtr * tSecPtr = &signal->m_sectionPtr[0];
  252.       Uint32 tData0 = tDataRegPtr[0];
  253.       Uint32 tData1 = tDataRegPtr[1];
  254.       Uint32 tData2 = tDataRegPtr[2];
  255.       
  256.       tSecPtr[0].i = tData0;
  257.       tSecPtr[1].i = tData1;
  258.       tSecPtr[2].i = tData2;
  259.       
  260.       //---------------------------------------------------------
  261.       // Prefetch of buffer[rPtr] is done here. We prefetch for
  262.       // read both the first cache line and the next 64 byte
  263.       // entry
  264.       //---------------------------------------------------------
  265.       PREFETCH((void*)&buffer[rPtr]);
  266.       PREFETCH((void*)(((char*)&buffer[rPtr]) + 64));
  267.       return gsnbnr;
  268.     } else {
  269.       bnr_error();
  270.       return 0; // Will never come here, simply to keep GCC happy.
  271.     }//if
  272.   } else {
  273.     //------------------------------------------------------------
  274.     // The Job Buffer was empty, signal this by return zero.
  275.     //------------------------------------------------------------
  276.     return 0;
  277.   }//if
  278. }//APZJobBuffer::retrieve()
  280. void 
  281. APZJobBuffer::signal2buffer(Signal* signal,
  282.     BlockNumber bnr, GlobalSignalNumber gsn,
  283.     BufferEntry& buf)
  284. {
  285.   Uint32 tSignalId = globalData.theSignalId;
  286.   Uint32 tFirstData = signal->theData[0];
  287.   Uint32 tLength = signal->header.theLength;
  288.   Uint32 tSigId  = buf.header.theSignalId;
  289.   
  290.   buf.header = signal->header;
  291.   buf.header.theVerId_signalNumber = gsn;
  292.   buf.header.theReceiversBlockNumber = bnr;
  293.   buf.header.theSendersSignalId = tSignalId - 1;
  294.   buf.header.theSignalId = tSigId;
  295.   buf.theDataRegister[0] = tFirstData;
  296.   
  297.   Uint32 tLengthCopied = 1;
  298.   Uint32* tSigDataPtr = &signal->theData[1];
  299.   Uint32* tDataRegPtr = &buf.theDataRegister[1];
  300.   while (tLengthCopied < tLength) {
  301.     Uint32 tData0 = tSigDataPtr[0];
  302.     Uint32 tData1 = tSigDataPtr[1];
  303.     Uint32 tData2 = tSigDataPtr[2];
  304.     Uint32 tData3 = tSigDataPtr[3];
  305.     
  306.     tLengthCopied += 4;
  307.     tSigDataPtr += 4;
  309.     tDataRegPtr[0] = tData0;
  310.     tDataRegPtr[1] = tData1;
  311.     tDataRegPtr[2] = tData2;
  312.     tDataRegPtr[3] = tData3;
  313.     tDataRegPtr += 4;
  314.   }//while
  316.   /**
  317.    * Copy sections references (copy all without if-statements)
  318.    */
  319.   tDataRegPtr = &buf.theDataRegister[tLength];
  320.   SegmentedSectionPtr * tSecPtr = &signal->m_sectionPtr[0];
  321.   Uint32 tData0 = tSecPtr[0].i;
  322.   Uint32 tData1 = tSecPtr[1].i;
  323.   Uint32 tData2 = tSecPtr[2].i;
  324.   tDataRegPtr[0] = tData0;
  325.   tDataRegPtr[1] = tData1;
  326.   tDataRegPtr[2] = tData2;
  327. }//APZJobBuffer::signal2buffer()
  329. void
  330. APZJobBuffer::insert(const SignalHeader * const sh,
  331.      const Uint32 * const theData, const Uint32 secPtrI[3]){
  332.   Uint32 tOccupancy = theOccupancy + 1;
  333.   Uint32 myWPtr = wPtr;
  334.   register BufferEntry& buf = buffer[myWPtr];
  335.   
  336.   if (tOccupancy < bufSize) {
  337.     Uint32 cond =  (++myWPtr == bufSize) - 1;
  338.     wPtr = myWPtr & cond;
  339.     theOccupancy = tOccupancy;
  340.     
  341.     buf.header = * sh;
  342.     const Uint32 len = buf.header.theLength;
  343.     memcpy(buf.theDataRegister, theData, 4 * len);
  344.     memcpy(&buf.theDataRegister[len], &secPtrI[0], 4 * 3);
  345.     //---------------------------------------------------------
  346.     // Prefetch of buffer[wPtr] is done here. We prefetch for
  347.     // write both the first cache line and the next 64 byte
  348.     // entry
  349.     //---------------------------------------------------------
  350.     WRITEHINT((void*)&buffer[wPtr]);
  351.     WRITEHINT((void*)(((char*)&buffer[wPtr]) + 64));
  352.     
  353.   } else {
  354.     jbuf_error();
  355.   }//if
  356. }
  357. APZJobBuffer::APZJobBuffer()
  358.   : bufSize(0), buffer(NULL), memRef(NULL)
  359. {
  360.   clear();
  361. }
  363. APZJobBuffer::~APZJobBuffer()
  364. {
  365.   delete [] buffer;
  366. }
  368. void
  369. APZJobBuffer::newBuffer(int size)
  370. {
  371.   buffer = new BufferEntry[size + 1]; // +1 to support "overrrun"
  372.   if(buffer){
  373. #ifndef NDB_PURIFY
  374.     ::memset(buffer, 0, (size * sizeof(BufferEntry)));
  375. #endif
  376.     bufSize = size;
  377.   } else
  378.     bufSize = 0;
  379. }
  381. void
  382. APZJobBuffer::clear()
  383. {
  384.   rPtr = 0;
  385.   wPtr = 0;
  386.   theOccupancy = 0;
  387. }
  389. /**
  390.  * Function prototype for print_restart
  391.  *
  392.  *   Defined later in this file
  393.  */
  394. void print_restart(FILE * output, Signal* signal, Uint32 aLevel);
  396. void FastScheduler::dumpSignalMemory(FILE * output)
  397. {
  398.   Signal signal;
  399.   Uint32 ReadPtr[5];
  400.   Uint32 tJob;
  401.   Uint32 tLastJob;
  403.   fprintf(output, "n");
  404.  
  405.   if (globalData.JobLap > 4095) {
  406.     if (globalData.JobCounter != 0)
  407.       tJob = globalData.JobCounter - 1;
  408.     else
  409.       tJob = 4095;
  410.     tLastJob = globalData.JobCounter;
  411.   } else {
  412.     if (globalData.JobCounter == 0)
  413.       return; // No signals sent
  414.     else {
  415.       tJob = globalData.JobCounter - 1;
  416.       tLastJob = 4095;
  417.     }
  418.   }
  419.   ReadPtr[0] = theJobBuffers[0].getReadPtr();
  420.   ReadPtr[1] = theJobBuffers[1].getReadPtr();
  421.   ReadPtr[2] = theJobBuffers[2].getReadPtr();
  422.   ReadPtr[3] = theJobBuffers[3].getReadPtr();
  423.   
  424.   do {
  425.     unsigned char tLevel = theJobPriority[tJob];
  426.     globalData.incrementWatchDogCounter(4);
  427.     if (ReadPtr[tLevel] == 0)
  428.       ReadPtr[tLevel] = theJobBuffers[tLevel].getBufSize() - 1;
  429.     else
  430.       ReadPtr[tLevel]--;
  431.     
  432.     theJobBuffers[tLevel].retrieveDump(&signal, ReadPtr[tLevel]);
  433.     print_restart(output, &signal, tLevel);
  434.     
  435.     if (tJob == 0)
  436.       tJob = 4095;
  437.     else
  438.       tJob--;
  439.     
  440.   } while (tJob != tLastJob);
  441.   fflush(output);
  442. }
  444. void
  445. FastScheduler::prio_level_error()
  446. {
  447.   ERROR_SET(ecError, ERROR_WRONG_PRIO_LEVEL, 
  448.     "Wrong Priority Level", "FastScheduler.C");
  449. }
  451. void 
  452. jbuf_error()
  453. {
  454.   ERROR_SET(ecError, BLOCK_ERROR_JBUFCONGESTION, 
  455.     "Job Buffer Full", "APZJobBuffer.C");
  456. }
  458. void 
  459. bnr_error()
  460. {
  461.   ERROR_SET(ecError, BLOCK_ERROR_BNR_ZERO, 
  462.     "Block Number Zero", "FastScheduler.C");
  463. }
  465. void
  466. print_restart(FILE * output, Signal* signal, Uint32 aLevel)
  467. {
  468.   fprintf(output, "--------------- Signal ----------------n");
  469.   SignalLoggerManager::printSignalHeader(output, 
  470.  signal->header,
  471.  aLevel,
  472.  globalData.ownId, 
  473.  true);
  474.   SignalLoggerManager::printSignalData  (output, 
  475.  signal->header,
  476.  &signal->theData[0]);
  477. }
  479. /**
  480.  * This method used to be a Cmvmi member function
  481.  * but is now a "ordinary" function"
  482.  *
  483.  * See TransporterCallback.cpp for explanation
  484.  */
  485. void 
  486. FastScheduler::reportDoJobStatistics(Uint32 tMeanLoopCount) {
  487.   Signal signal; 
  488.   memset(&signal.header, 0, sizeof(signal.header));
  490.   signal.theData[0] = EventReport::JobStatistic;
  491.   signal.theData[1] = tMeanLoopCount;
  492.   
  493.   memset(&signal.header, 0, sizeof(SignalHeader));
  494.   signal.header.theLength = 2;
  495.   signal.header.theSendersSignalId = 0;
  496.   signal.header.theSendersBlockRef = numberToRef(0, 0);
  497.   
  498.   execute(&signal, JBA, CMVMI, GSN_EVENT_REP);
  499. }