SendBuffer.hpp 5.16 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37
/* 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 */

//****************************************************************************
//
//  NAME
//      SendBuffer
//
//  DESCRIPTION
//      The SendBuffer is a circular buffer storing signals waiting to be sent.
//      The signals can be of variable size and are copied into the buffer 
//      in Protocol 6 format. There will be two SendBuffer instances 
//      (priority level A and B) for each transporter using a buffer for 
//      sending. The buffering will in most cases be done to send as big 
//      packages as possible over TCP/IP.
//
//***************************************************************************/
#ifndef SendBuffer_H
#define SendBuffer_H

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

#ifdef DEBUG_TRANSPORTER
38
#include <ndb_global.h>
39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190
#endif

class SendBuffer {
  friend class TCP_Transporter;
public:
  // Set member variables
  SendBuffer(Uint32 bufSize);

  // Deallocate the buffer memory
  ~SendBuffer();
  
  // Allocate memory for the buffer and initialize the buffer pointers
  bool initBuffer(Uint32 aRemoteNodeId);  

  // Number of bytes remaining in the buffer
  Uint32 bufferSizeRemaining();

  // Number of bytes of data in the buffer 
  int bufferSize(); 

  // Empty the buffer
  void emptyBuffer();
  
  /**
   * The transporter calls updateBuffer after a retrieve followed by 
   * a successful send, to update the cirkular buffer pointers.
   * updateBuffer is called with the number of bytes really sent,
   * it may be that it is less than what was retrived from the buffer.
   * If that is the case there will be an incomplete message (slack)
   * in the SendBuffer. 
   *
   * Returns  0 if buffer empty
   *    else ~0
   */
  Uint32 bytesSent(Uint32 len);
  
#ifdef DEBUG_TRANSPORTER
  // Prints the buffer status on the screen. Can be used for testing purposes.
  void print();
#endif

  Uint32* getInsertPtr(Uint32 bytes);
  void updateInsertPtr(Uint32 bytes);

private:
  
  Uint32   sizeOfBuffer;  // Length, in number of bytes, of the buffer memory
  Uint32   dataSize;      // Number of bytes in buffer
  
  Uint32 * startOfBuffer; // Pointer to the start of the buffer memory
  Uint32 * endOfBuffer;   // Pointer to end of buffer
  
  Uint32 * insertPtr;     // Where to insert next
  
  char *   sendPtr;           // Where data to send starts
  Uint32   sendDataSize;      // Num bytes to send
  
  Uint32   theRemoteNodeId;
};

inline
Uint32
SendBuffer::bytesSent(Uint32 bytes) {

  if(bytes > dataSize){
#ifdef DEBUG_TRANSPORTER
    printf("bytes(%d) > dataSize(%d)\n", bytes, dataSize);
#endif
    abort();
    // reportError(0 ,theRemoteNodeId, TE_INVALID_MESSAGE_LENGTH);
    return 0;
  }//if

  if(bytes > sendDataSize){
#ifdef DEBUG_TRANSPORTER
    printf("bytes(%d) > sendDataSize(%d)\n", bytes, sendDataSize);
#endif
    abort();
    //reportError(0,theRemoteNodeId, TE_INVALID_MESSAGE_LENGTH);
    return 0;
  }//if

  dataSize     -= bytes;
  sendPtr      += bytes;
  sendDataSize -= bytes;
  
  if(sendDataSize == 0){
    if(sendPtr > (char*)insertPtr){
      sendPtr = (char *)startOfBuffer;
      sendDataSize = dataSize;
    } else {
      sendPtr = ((char*)insertPtr) - dataSize;
      sendDataSize = dataSize;
    }
  }
  
  if(dataSize == 0)
    return 0;
  return ~0;
}

inline
Uint32*
SendBuffer::getInsertPtr(Uint32 len){
  if (bufferSizeRemaining() < len){
    return 0;
  }

  const char * const tmpInsertPtr = (char *) insertPtr;

  if(tmpInsertPtr >= sendPtr){
    // Is there enough space at the end of the buffer? 
    if ((tmpInsertPtr + len) < (char*)endOfBuffer){
      sendDataSize += len;
      return insertPtr;
    } else {
      // We have passed the end of the cirkular buffer, 
      // must start from the beginning
      // Is there enough space in the beginning of the buffer?
      if ((Uint32)(sendPtr - (char *)startOfBuffer) <= len){
	// Not enough space available, insert failed
	return 0;
      } else {
	// There is space available at the beginning of the buffer
	// We start from the beginning, set endOfData and insertPtr
	insertPtr = startOfBuffer; 
	if(sendDataSize != 0){
	  return insertPtr;
	}	
	sendPtr      = (char *)startOfBuffer;
	sendDataSize = len;
	return insertPtr;
      }
    }
  } else {
    // sendPtr > insertPtr
    // Is there enought room
    if((tmpInsertPtr + len) < sendPtr){
      return insertPtr;
    }
    return 0;
  }
}

inline
void
SendBuffer::updateInsertPtr(Uint32 lenBytes){
  dataSize  += lenBytes;
  insertPtr += (lenBytes / 4);
}

#endif // Define of SendBuffer_H