release/src/router/mysql/storage/ndb/src/kernel/vm/FastScheduler.hpp

   1 /* Copyright (c) 2003-2005 MySQL AB
   2
   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; version 2 of the License.
   6
   7    This program is distributed in the hope that it will be useful,
   8    but WITHOUT ANY WARRANTY; without even the implied warranty of
   9    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  10    GNU General Public License for more details.
  11
  12    You should have received a copy of the GNU General Public License
  13    along with this program; if not, write to the Free Software
  14    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA */
  15
  16 #ifndef FastScheduler_H
  17 #define FastScheduler_H
  18
  19 #include <VMSignal.hpp>
  20 #include <kernel_types.h>
  21 #include <Prio.hpp>
  22 #include <SignalLoggerManager.hpp>
  23 #include <SimulatedBlock.hpp>
  24 #include <ErrorHandlingMacros.hpp>
  25 #include <GlobalData.hpp>
  26 #include <TransporterDefinitions.hpp>
  27 #include <prefetch.h>
  28
  29 #define MAX_OCCUPANCY 1024
  30
  31 #define JBASIZE   1280 // Jobs which have dead lines to meet use this level
  32 #define JBBSIZE   4096 // Most jobs use this level
  33 #define JBCSIZE   64   // Only used by STTOR and STTORRY currently
  34 #define JBDSIZE   4096 // Time Queue uses this level for storage, not supported
  35                        // as priority level
  36 void bnr_error();
  37 void jbuf_error();
  38 class Signal;
  39 class Block;
  40
  41 class BufferEntry
  42 {
  43 public:
  44   SignalHeader header;
  45   Uint32 theDataRegister[25];
  46 };
  47
  48 class APZJobBuffer
  49 {
  50 public:
  51   APZJobBuffer();
  52   ~APZJobBuffer();
  53
  54   void newBuffer(int size);
  55
  56   void insert(Signal* signal, BlockNumber bnr, GlobalSignalNumber gsn);
  57   void insert(const SignalHeader * const sh, const Uint32 * const theData, const Uint32 secPtrI[3]);
  58   void insert(Signal* signal, BlockNumber bnr, GlobalSignalNumber gsn,
  59               Uint32 myWPtr);
  60
  61   Uint32 retrieve(Signal *signal);
  62   void retrieve(Signal *signal, Uint32 myRptr);
  63
  64   /**
  65    * Used when dumping to trace file
  66    */
  67   void retrieveDump(Signal *signal, Uint32 myRptr);
  68
  69   void clear();
  70   Uint32 getOccupancy() const;
  71
  72   Uint32 getReadPtr() const;
  73   Uint32 getWritePtr() const;
  74   Uint32 getBufSize() const;
  75
  76 private:
  77   void signal2buffer(Signal* signal, BlockNumber bnr,
  78                      GlobalSignalNumber gsn, BufferEntry& buf);
  79   Uint32 rPtr;
  80   Uint32 wPtr;
  81   Uint32 theOccupancy;
  82   Uint32 bufSize;
  83   BufferEntry* buffer;
  84   BufferEntry* memRef;
  85 };
  86
  87
  88 class FastScheduler
  89 {
  90 public:
  91    FastScheduler();
  92    ~FastScheduler();
  93
  94   void doJob();
  95   int checkDoJob();
  96
  97   void activateSendPacked();
  98
  99   void execute(Signal* signal,
 100                Priority prio,
 101                BlockNumber bnr,
 102                GlobalSignalNumber gsn);
 103
 104   void execute(const SignalHeader * const sh,
 105                Uint8 prio, const Uint32 * const theData, const Uint32 secPtr[3]);
 106
 107   void clear();
 108   Signal* getVMSignals();
 109
 110   void dumpSignalMemory(FILE * output);
 111   Priority highestAvailablePrio() const;
 112   Uint32 getBOccupancy() const;
 113   void sendPacked();
 114
 115   void insertTimeQueue(Signal* aSignal, BlockNumber bnr,
 116                        GlobalSignalNumber gsn, Uint32 aIndex);
 117   void scheduleTimeQueue(Uint32 aIndex);
 118
 119 private:
 120   void highestAvailablePrio(Priority prio);
 121   void reportJob(Priority aPriority);
 122   void prio_level_error();
 123
 124   Uint32 theDoJobTotalCounter;
 125   Uint32 theDoJobCallCounter;
 126   Uint8 theJobPriority[4096];
 127   APZJobBuffer theJobBuffers[JB_LEVELS];
 128
 129   void reportDoJobStatistics(Uint32 meanLoopCount);
 130 };
 131
 132 inline
 133 Uint32
 134 FastScheduler::getBOccupancy() const {
 135   return theJobBuffers[JBB].getOccupancy();
 136 }//FastScheduler::getBOccupancy()
 137
 138 inline
 139 int
 140 FastScheduler::checkDoJob()
 141 {
 142   /*
 143    * Job buffer overload protetction
 144    * If the job buffer B is filled over a certain limit start
 145    * to execute the signals in the job buffer's
 146    */
 147   if (getBOccupancy() < MAX_OCCUPANCY) {
 148     return 0;
 149   } else {
 150     doJob();
 151     return 1;
 152   }//if
 153 }//FastScheduler::checkDoJob()
 154
 155 inline
 156 void
 157 FastScheduler::reportJob(Priority aPriority)
 158 {
 159   Uint32 tJobCounter = globalData.JobCounter;
 160   Uint32 tJobLap = globalData.JobLap;
 161   theJobPriority[tJobCounter] = (Uint8)aPriority;
 162   globalData.JobCounter = (tJobCounter + 1) & 4095;
 163   globalData.JobLap = tJobLap + 1;
 164 }
 165
 166 inline
 167 Priority
 168 FastScheduler::highestAvailablePrio() const
 169 {
 170    return (Priority)globalData.highestAvailablePrio;
 171 }
 172
 173 inline
 174 void
 175 FastScheduler::highestAvailablePrio(Priority prio)
 176 {
 177    globalData.highestAvailablePrio = (Uint32)prio;
 178 }
 179
 180 inline
 181 Signal*
 182 FastScheduler::getVMSignals()
 183 {
 184   return &globalData.VMSignals[0];
 185 }
 186
 187
 188 // Inserts of a protocol object into the Job Buffer.
 189 inline
 190 void
 191 FastScheduler::execute(const SignalHeader * const sh, Uint8 prio,
 192                        const Uint32 * const theData, const Uint32 secPtrI[3]){
 193 #ifdef VM_TRACE
 194   if (prio >= LEVEL_IDLE)
 195     prio_level_error();
 196 #endif
 197
 198   theJobBuffers[prio].insert(sh, theData, secPtrI);
 199   if (prio < (Uint8)highestAvailablePrio())
 200     highestAvailablePrio((Priority)prio);
 201 }
 202
 203 inline
 204 void
 205 FastScheduler::execute(Signal* signal, Priority prio,
 206                        BlockNumber bnr, GlobalSignalNumber gsn)
 207 {
 208 #ifdef VM_TRACE
 209   if (prio >= LEVEL_IDLE)
 210     prio_level_error();
 211 #endif
 212   theJobBuffers[prio].insert(signal, bnr, gsn);
 213   if (prio < highestAvailablePrio())
 214     highestAvailablePrio(prio);
 215 }
 216
 217 inline
 218 void
 219 FastScheduler::insertTimeQueue(Signal* signal, BlockNumber bnr,
 220                                GlobalSignalNumber gsn, Uint32 aIndex)
 221 {
 222   theJobBuffers[3].insert(signal, bnr, gsn, aIndex);
 223 }
 224
 225 inline
 226 void
 227 FastScheduler::scheduleTimeQueue(Uint32 aIndex)
 228 {
 229   Signal* signal = getVMSignals();
 230   theJobBuffers[3].retrieve(signal, aIndex);
 231   theJobBuffers[0].insert
 232     (signal,
 233      (BlockNumber)signal->header.theReceiversBlockNumber,
 234      (GlobalSignalNumber)signal->header.theVerId_signalNumber);
 235   if (highestAvailablePrio() > JBA)
 236     highestAvailablePrio(JBA);
 237 }
 238
 239 inline
 240 Uint32
 241 APZJobBuffer::getWritePtr() const
 242 {
 243   return wPtr;
 244 }
 245
 246 inline
 247 Uint32
 248 APZJobBuffer::getReadPtr() const
 249 {
 250   return rPtr;
 251 }
 252
 253 inline
 254 Uint32
 255 APZJobBuffer::getOccupancy() const
 256 {
 257   return theOccupancy;
 258 }
 259
 260 inline
 261 Uint32
 262 APZJobBuffer::getBufSize() const
 263 {
 264   return bufSize;
 265 }
 266
 267 inline
 268 void
 269 APZJobBuffer::retrieve(Signal* signal, Uint32 myRptr)
 270 {
 271   register BufferEntry& buf = buffer[myRptr];
 272
 273   buf.header.theSignalId = globalData.theSignalId++;
 274
 275   signal->header = buf.header;
 276
 277   Uint32 *from = (Uint32*) &buf.theDataRegister[0];
 278   Uint32 *to   = (Uint32*) &signal->theData[0];
 279   Uint32 noOfWords = buf.header.theLength;
 280   for(; noOfWords; noOfWords--)
 281     *to++ = *from++;
 282   // Copy sections references (copy all without if-statements)
 283   SegmentedSectionPtr * tSecPtr = &signal->m_sectionPtr[0];
 284   tSecPtr[0].i = from[0];
 285   tSecPtr[1].i = from[1];
 286   tSecPtr[2].i = from[2];
 287   return;
 288 }
 289
 290 inline
 291 void
 292 APZJobBuffer::retrieveDump(Signal* signal, Uint32 myRptr)
 293 {
 294   /**
 295    * Note that signal id is not taken from global data
 296    */
 297
 298   register BufferEntry& buf = buffer[myRptr];
 299   signal->header = buf.header;
 300
 301   Uint32 *from = (Uint32*) &buf.theDataRegister[0];
 302   Uint32 *to   = (Uint32*) &signal->theData[0];
 303   Uint32 noOfWords = buf.header.theLength;
 304   for(; noOfWords; noOfWords--)
 305     *to++ = *from++;
 306   return;
 307 }
 308
 309 inline
 310 void
 311 APZJobBuffer::insert(Signal* signal,
 312                      BlockNumber bnr, GlobalSignalNumber gsn)
 313 {
 314   Uint32 tOccupancy = theOccupancy + 1;
 315   Uint32 myWPtr = wPtr;
 316   if (tOccupancy < bufSize) {
 317     register BufferEntry& buf = buffer[myWPtr];
 318     Uint32 cond =  (++myWPtr == bufSize) - 1;
 319     wPtr = myWPtr & cond;
 320     theOccupancy = tOccupancy;
 321     signal2buffer(signal, bnr, gsn, buf);
 322     //---------------------------------------------------------
 323     // Prefetch of buffer[wPtr] is done here. We prefetch for
 324     // write both the first cache line and the next 64 byte
 325     // entry
 326     //---------------------------------------------------------
 327     WRITEHINT((void*)&buffer[wPtr]);
 328     WRITEHINT((void*)(((char*)&buffer[wPtr]) + 64));
 329   } else {
 330     jbuf_error();
 331   }//if
 332 }
 333
 334
 335 inline
 336 void
 337 APZJobBuffer::insert(Signal* signal, BlockNumber bnr,
 338                      GlobalSignalNumber gsn, Uint32 myWPtr)
 339 {
 340   register BufferEntry& buf = buffer[myWPtr];
 341   signal2buffer(signal, bnr, gsn, buf);
 342 }
 343
 344 #endif