The world's most popular open source database

MySQL 5.1: TCP_Transporter Class Reference

TCP_Transporter Class Reference

#include <TCP_Transporter.hpp>

Inheritance diagram for TCP_Transporter:

[legend]Collaboration diagram for TCP_Transporter:

[legend]List of all members.


Protected Member Functions
virtual bool	connect_server_impl (NDB_SOCKET_TYPE sockfd)
virtual bool	connect_client_impl (NDB_SOCKET_TYPE sockfd)
bool	connect_common (NDB_SOCKET_TYPE sockfd)
virtual void	disconnectImpl ()
Private Member Functions
	TCP_Transporter (TransporterRegistry &, int sendBufferSize, int maxReceiveSize, const char lHostName, const char rHostName, int r_port, bool isMgmConnection, NodeId lHostId, NodeId rHostId, NodeId serverNodeId, bool checksum, bool signalId, Uint32 reportFreq=4096)
virtual	~TCP_Transporter ()
bool	initTransporter ()
Uint32 *	getWritePtr (Uint32 lenBytes, Uint32 prio)
void	updateWritePtr (Uint32 lenBytes, Uint32 prio)
bool	hasDataToSend () const
bool	doSend ()
int	doReceive ()
NDB_SOCKET_TYPE	getSocket () const
virtual Uint32	getReceiveData (Uint32 **ptr)
virtual void	updateReceiveDataPtr (Uint32 bytesRead)
virtual Uint32	get_free_buffer () const
void	setSocketOptions ()
bool	sendIsPossible (struct timeval *timeout)
Static Private Member Functions
static bool	setSocketNonBlocking (NDB_SOCKET_TYPE aSocket)
Private Attributes
SendBuffer	m_sendBuffer
NDB_SOCKET_TYPE	theSocket
Uint32	maxReceiveSize
int	sockOptRcvBufSize
int	sockOptSndBufSize
int	sockOptNodelay
int	sockOptTcpMaxSeg
Uint32	reportFreq
Uint32	receiveCount
Uint64	receiveSize
Uint32	sendCount
Uint64	sendSize
ReceiveBuffer	receiveBuffer
Friends
class	TransporterRegistry

Detailed Description

Definition at line 44 of file TCP_Transporter.hpp.

Constructor & Destructor Documentation

TCP_Transporter::TCP_Transporter	(	TransporterRegistry &	,
		int	sendBufferSize,
		int	maxReceiveSize,
		const char *	lHostName,
		const char *	rHostName,
		int	r_port,
		bool	isMgmConnection,
		NodeId	lHostId,
		NodeId	rHostId,
		NodeId	serverNodeId,
		bool	checksum,
		bool	signalId,
		Uint32	reportFreq = `4096`
	)			`[private]`

Definition at line 66 of file TCP_Transporter.cpp.

References maxReceiveSize, NDB_INVALID_SOCKET, receiveCount, receiveSize, reportFreq, sendCount, sendSize, sockOptNodelay, sockOptRcvBufSize, sockOptSndBufSize, sockOptTcpMaxSeg, and theSocket.

00076                                                      :
00077   Transporter(t_reg, tt_TCP_TRANSPORTER,
00078               lHostName, rHostName, r_port, isMgmConnection,
00079               lNodeId, rNodeId, serverNodeId,
00080               0, false, chksm, signalId),
00081   m_sendBuffer(sendBufSize)
00082 {
00083   maxReceiveSize = maxRecvSize;
00084   
00085   // Initialize member variables
00086   theSocket     = NDB_INVALID_SOCKET;
00087   
00088   sendCount      = receiveCount = 0;
00089   sendSize       = receiveSize  = 0;
00090   reportFreq     = _reportFreq;
00091 
00092   sockOptRcvBufSize = 70080;
00093   sockOptSndBufSize = 71540;
00094   sockOptNodelay    = 1;
00095   sockOptTcpMaxSeg  = 4096;
00096 }

TCP_Transporter::~TCP_Transporter ( ) [private, virtual]

Definition at line 98 of file TCP_Transporter.cpp.

References ReceiveBuffer::destroy(), Transporter::doDisconnect(), NDB_INVALID_SOCKET, receiveBuffer, and theSocket.

00098                                   {
00099   
00100   // Disconnect
00101   if (theSocket != NDB_INVALID_SOCKET)
00102     doDisconnect();
00103   
00104   // Delete send buffers
00105   
00106   // Delete receive buffer!!
00107   receiveBuffer.destroy();
00108 }

Here is the call graph for this function:

Member Function Documentation

bool TCP_Transporter::connect_client_impl ( NDB_SOCKET_TYPE sockfd ) [protected, virtual]

Implements Transporter.

Definition at line 116 of file TCP_Transporter.cpp.

References connect_common(), DBUG_ENTER, and DBUG_RETURN.

00117 {
00118   DBUG_ENTER("TCP_Transpporter::connect_client_impl");
00119   DBUG_RETURN(connect_common(sockfd));
00120 }

Here is the call graph for this function:

bool TCP_Transporter::connect_common ( NDB_SOCKET_TYPE sockfd ) [protected]

Definition at line 122 of file TCP_Transporter.cpp.

References DBUG_PRINT, Transporter::remoteNodeId, setSocketNonBlocking(), setSocketOptions(), and theSocket.

Referenced by connect_client_impl(), and connect_server_impl().

00123 {
00124   theSocket = sockfd;
00125   setSocketOptions();
00126   setSocketNonBlocking(theSocket);
00127   DBUG_PRINT("info", ("Successfully set-up TCP transporter to node %d",
00128               remoteNodeId));
00129   return true;
00130 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool TCP_Transporter::connect_server_impl ( NDB_SOCKET_TYPE sockfd ) [protected, virtual]

Setup client/server and perform connect/accept Is used both by clients and servers A client connects to the remote server A server accepts any new connections

Implements Transporter.

Definition at line 110 of file TCP_Transporter.cpp.

References connect_common(), DBUG_ENTER, and DBUG_RETURN.

00111 {
00112   DBUG_ENTER("TCP_Transpporter::connect_server_impl");
00113   DBUG_RETURN(connect_common(sockfd));
00114 }

Here is the call graph for this function:

void TCP_Transporter::disconnectImpl ( ) [protected, virtual]

Disconnects a TCP/IP node. Empty send and receivebuffer.

Implements Transporter.

Definition at line 427 of file TCP_Transporter.cpp.

References ReceiveBuffer::clear(), SendBuffer::emptyBuffer(), m_sendBuffer, NDB_CLOSE_SOCKET(), NDB_INVALID_SOCKET, receiveBuffer, Transporter::report_error(), TE_ERROR_CLOSING_SOCKET, and theSocket.

00427                                 {
00428   if(theSocket != NDB_INVALID_SOCKET){
00429     if(NDB_CLOSE_SOCKET(theSocket) < 0){
00430       report_error(TE_ERROR_CLOSING_SOCKET);
00431     }
00432   }
00433   
00434   // Empty send och receive buffers 
00435   receiveBuffer.clear();
00436   m_sendBuffer.emptyBuffer();
00437 
00438   theSocket = NDB_INVALID_SOCKET;
00439 }

Here is the call graph for this function:

int TCP_Transporter::doReceive ( ) [private]

It reads the external TCP/IP interface once and puts the data in the receiveBuffer

Definition at line 370 of file TCP_Transporter.cpp.

References DISCONNECT_ERRNO, Transporter::doDisconnect(), Transporter::get_callback_obj(), InetErrno, ReceiveBuffer::insertPtr, maxReceiveSize, ndbout_c(), ndbstrerror, receiveBuffer, receiveCount, receiveSize, Transporter::remoteNodeId, Transporter::report_disconnect(), Transporter::report_error(), reportFreq, reportReceiveLen(), ReceiveBuffer::sizeOfBuffer, ReceiveBuffer::sizeOfData, TE_INVALID_MESSAGE_LENGTH, and theSocket.

Referenced by TransporterRegistry::performReceive().

00370                            {
00371   // Select-function must return the socket for read
00372   // before this method is called
00373   // It reads the external TCP/IP interface once
00374   Uint32 size = receiveBuffer.sizeOfBuffer - receiveBuffer.sizeOfData;
00375   if(size > 0){
00376     const int nBytesRead = recv(theSocket, 
00377                                 receiveBuffer.insertPtr, 
00378                                 size < maxReceiveSize ? size : maxReceiveSize, 
00379                                 0);
00380     
00381     if (nBytesRead > 0) {
00382       receiveBuffer.sizeOfData += nBytesRead;
00383       receiveBuffer.insertPtr  += nBytesRead;
00384       
00385       if(receiveBuffer.sizeOfData > receiveBuffer.sizeOfBuffer){
00386 #ifdef DEBUG_TRANSPORTER
00387         ndbout_c("receiveBuffer.sizeOfData(%d) > receiveBuffer.sizeOfBuffer(%d)",
00388                  receiveBuffer.sizeOfData, receiveBuffer.sizeOfBuffer);
00389         ndbout_c("nBytesRead = %d", nBytesRead);
00390 #endif
00391         ndbout_c("receiveBuffer.sizeOfData(%d) > receiveBuffer.sizeOfBuffer(%d)",
00392                  receiveBuffer.sizeOfData, receiveBuffer.sizeOfBuffer);
00393         report_error(TE_INVALID_MESSAGE_LENGTH);
00394         return 0;
00395       }
00396       
00397       receiveCount ++;
00398       receiveSize  += nBytesRead;
00399       
00400       if(receiveCount == reportFreq){
00401         reportReceiveLen(get_callback_obj(), remoteNodeId, receiveCount, receiveSize);
00402         receiveCount = 0;
00403         receiveSize  = 0;
00404       }
00405       return nBytesRead;
00406     } else {
00407 #if defined DEBUG_TRANSPORTER
00408       ndbout_c("Receive Failure(disconnect==%d) to node = %d nBytesSent = %d "
00409                "errno = %d strerror = %s",
00410                DISCONNECT_ERRNO(InetErrno, nBytesRead),
00411                remoteNodeId, nBytesRead, InetErrno, 
00412                (char*)ndbstrerror(InetErrno));
00413 #endif   
00414       if(DISCONNECT_ERRNO(InetErrno, nBytesRead)){
00415         // The remote node has closed down
00416         doDisconnect();
00417         report_disconnect(InetErrno);
00418       } 
00419     }
00420     return nBytesRead;
00421   } else {
00422     return 0;
00423   }
00424 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool TCP_Transporter::doSend ( ) [private]

Retrieves the contents of the send buffers and writes it on the external TCP/IP interface until the send buffers are empty and as long as write is possible.

Definition at line 323 of file TCP_Transporter.cpp.

References SendBuffer::bytesSent(), DISCONNECT_ERRNO, Transporter::doDisconnect(), Transporter::get_callback_obj(), inet_send, InetErrno, m_sendBuffer, ndbout_c(), ndbstrerror, Transporter::remoteNodeId, Transporter::report_disconnect(), reportFreq, reportSendLen(), sendCount, SendBuffer::sendDataSize, SendBuffer::sendPtr, sendSize, and theSocket.

Referenced by getWritePtr(), TransporterRegistry::performSend(), and updateWritePtr().

00323                         {
00324   // If no sendbuffers are used nothing is done
00325   // Sends the contents of the SendBuffers until they are empty
00326   // or until select does not select the socket for write.
00327   // Before calling send, the socket must be selected for write
00328   // using "select"
00329   // It writes on the external TCP/IP interface until the send buffer is empty
00330   // and as long as write is possible (test it using select)
00331 
00332   // Empty the SendBuffers
00333   
00334   const char * const sendPtr = m_sendBuffer.sendPtr;
00335   const Uint32 sizeToSend    = m_sendBuffer.sendDataSize;
00336   if (sizeToSend > 0){
00337     const int nBytesSent = inet_send(theSocket, sendPtr, sizeToSend, 0);
00338     
00339     if (nBytesSent > 0) {
00340       m_sendBuffer.bytesSent(nBytesSent);
00341       
00342       sendCount ++;
00343       sendSize  += nBytesSent;
00344       if(sendCount == reportFreq){
00345         reportSendLen(get_callback_obj(), remoteNodeId, sendCount, sendSize);
00346         sendCount = 0;
00347         sendSize  = 0;
00348       }
00349     } else {
00350       // Send failed
00351 #if defined DEBUG_TRANSPORTER
00352       ndbout_c("Send Failure(disconnect==%d) to node = %d nBytesSent = %d "
00353                "errno = %d strerror = %s",
00354                DISCONNECT_ERRNO(InetErrno, nBytesSent),
00355                remoteNodeId, nBytesSent, InetErrno, 
00356                (char*)ndbstrerror(InetErrno));
00357 #endif   
00358       if(DISCONNECT_ERRNO(InetErrno, nBytesSent)){
00359         doDisconnect();
00360         report_disconnect(InetErrno);
00361       }
00362       
00363       return false;
00364     }
00365   }
00366   return true;
00367 }

Here is the call graph for this function:

Here is the caller graph for this function:

Uint32 TCP_Transporter::get_free_buffer ( ) const [private, virtual]

Implements Transporter.

Definition at line 257 of file TCP_Transporter.cpp.

References SendBuffer::bufferSizeRemaining(), and m_sendBuffer.

00258 {
00259   return m_sendBuffer.bufferSizeRemaining();
00260 }

Here is the call graph for this function:

Uint32 TCP_Transporter::getReceiveData ( Uint32 ** ptr ) [inline, private, virtual]

Get Receive Data

Returns - no of bytes to read and set ptr

Definition at line 170 of file TCP_Transporter.hpp.

References ReceiveBuffer::readPtr, receiveBuffer, and ReceiveBuffer::sizeOfData.

Referenced by TransporterRegistry::performReceive().

00170                                             {
00171   (* ptr) = receiveBuffer.readPtr;
00172   return receiveBuffer.sizeOfData;
00173 }

Here is the caller graph for this function:

NDB_SOCKET_TYPE TCP_Transporter::getSocket ( ) const [inline, private]

Returns socket (used for select)

Definition at line 164 of file TCP_Transporter.hpp.

References theSocket.

Referenced by TransporterRegistry::performReceive(), and TransporterRegistry::performSend().

00164                                  {
00165   return theSocket;
00166 }

Here is the caller graph for this function:

Uint32 * TCP_Transporter::getWritePtr	(	Uint32	lenBytes,
		Uint32	prio
	)			`[private, virtual]`

Implements Transporter.

Definition at line 263 of file TCP_Transporter.cpp.

References doSend(), SendBuffer::getInsertPtr(), m_sendBuffer, and sendIsPossible().

00263                                                         {
00264   
00265   Uint32 * insertPtr = m_sendBuffer.getInsertPtr(lenBytes);
00266   
00267   struct timeval timeout = {0, 10000};
00268 
00269   if (insertPtr == 0) {
00270     //-------------------------------------------------
00271     // Buffer was completely full. We have severe problems.
00272     // We will attempt to wait for a small time
00273     //-------------------------------------------------
00274     if(sendIsPossible(&timeout)) {
00275       //-------------------------------------------------
00276       // Send is possible after the small timeout.
00277       //-------------------------------------------------
00278       if(!doSend()){
00279         return 0;
00280       } else {
00281         //-------------------------------------------------
00282         // Since send was successful we will make a renewed
00283         // attempt at inserting the signal into the buffer.
00284         //-------------------------------------------------
00285         insertPtr = m_sendBuffer.getInsertPtr(lenBytes);
00286       }//if
00287     } else {
00288       return 0;
00289     }//if
00290   }
00291   return insertPtr;
00292 }

Here is the call graph for this function:

bool TCP_Transporter::hasDataToSend ( ) const [inline, private]

Definition at line 187 of file TCP_Transporter.hpp.

References SendBuffer::dataSize, and m_sendBuffer.

00187                                      {
00188   return m_sendBuffer.dataSize > 0;
00189 }

bool TCP_Transporter::initTransporter ( ) [private, virtual]

Allocate buffers for sending and receiving

Implements Transporter.

Definition at line 133 of file TCP_Transporter.cpp.

References ReceiveBuffer::init(), SendBuffer::initBuffer(), m_sendBuffer, MAX_MESSAGE_SIZE, maxReceiveSize, receiveBuffer, and Transporter::remoteNodeId.

Referenced by TransporterRegistry::createTCPTransporter().

00133                                  {
00134   
00135   // Allocate buffer for receiving
00136   // Let it be the maximum size we receive plus 8 kB for any earlier received
00137   // incomplete messages (slack)
00138   Uint32 recBufSize = maxReceiveSize;
00139   if(recBufSize < MAX_MESSAGE_SIZE){
00140     recBufSize = MAX_MESSAGE_SIZE;
00141   }
00142   
00143   if(!receiveBuffer.init(recBufSize+MAX_MESSAGE_SIZE)){
00144     return false;
00145   }
00146   
00147   // Allocate buffers for sending
00148   if (!m_sendBuffer.initBuffer(remoteNodeId)) {
00149     // XXX What shall be done here? 
00150     // The same is valid for the other init-methods 
00151     return false;
00152   }
00153   
00154   return true;
00155 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool TCP_Transporter::sendIsPossible ( struct timeval * timeout ) [private]

Definition at line 223 of file TCP_Transporter.cpp.

References NDB_INVALID_SOCKET, NdbSleep_MilliSleep(), NULL, and theSocket.

Referenced by getWritePtr(), and updateWritePtr().

00223                                                         {
00224 #ifdef NDB_OSE
00225 
00233   if(theSocket >= 0){
00234     Uint32 timeOutMillis = timeout->tv_sec * 1000 + timeout->tv_usec / 1000;
00235     NdbSleep_MilliSleep(timeOutMillis);
00236     return true;
00237   }
00238   return false;
00239 #else
00240   if(theSocket != NDB_INVALID_SOCKET){
00241     fd_set   writeset;
00242     FD_ZERO(&writeset);
00243     FD_SET(theSocket, &writeset);
00244     
00245     int selectReply = select(theSocket + 1, NULL, &writeset, NULL, timeout);
00246 
00247     if ((selectReply > 0) && FD_ISSET(theSocket, &writeset)) 
00248       return true;
00249     else
00250       return false;
00251   }
00252   return false;
00253 #endif
00254 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool TCP_Transporter::setSocketNonBlocking ( NDB_SOCKET_TYPE aSocket ) [static, private]

Definition at line 203 of file TCP_Transporter.cpp.

References flags, InetErrno, NDB_NONBLOCK, and ndbout_c().

Referenced by connect_common().

00203                                                            {
00204   int flags;
00205   flags = fcntl(socket, F_GETFL, 0);
00206   if (flags < 0) {
00207 #ifdef DEBUG_TRANSPORTER
00208     ndbout_c("Set non-blocking server error1: %s", strerror(InetErrno));
00209 #endif
00210   }//if
00211   flags |= NDB_NONBLOCK;
00212   if (fcntl(socket, F_SETFL, flags) == -1) {
00213 #ifdef DEBUG_TRANSPORTER
00214     ndbout_c("Set non-blocking server error2: %s", strerror(InetErrno));
00215 #endif
00216   }//if
00217   return true;
00218 }

Here is the call graph for this function:

Here is the caller graph for this function:

void TCP_Transporter::setSocketOptions ( ) [private]

Definition at line 158 of file TCP_Transporter.cpp.

References InetErrno, ndbout_c(), sockOptNodelay, sockOptRcvBufSize, sockOptSndBufSize, and theSocket.

Referenced by connect_common().

00158                                  {
00159   if (setsockopt(theSocket, SOL_SOCKET, SO_RCVBUF,
00160                  (char*)&sockOptRcvBufSize, sizeof(sockOptRcvBufSize)) < 0) {
00161 #ifdef DEBUG_TRANSPORTER
00162     ndbout_c("The setsockopt SO_RCVBUF error code = %d", InetErrno);
00163 #endif
00164   }//if
00165   
00166   if (setsockopt(theSocket, SOL_SOCKET, SO_SNDBUF,
00167                  (char*)&sockOptSndBufSize, sizeof(sockOptSndBufSize)) < 0) {
00168 #ifdef DEBUG_TRANSPORTER
00169     ndbout_c("The setsockopt SO_SNDBUF error code = %d", InetErrno);
00170 #endif
00171   }//if
00172   
00173   //-----------------------------------------------
00174   // Set the TCP_NODELAY option so also small packets are sent
00175   // as soon as possible
00176   //-----------------------------------------------
00177   if (setsockopt(theSocket, IPPROTO_TCP, TCP_NODELAY, 
00178                  (char*)&sockOptNodelay, sizeof(sockOptNodelay)) < 0) {
00179 #ifdef DEBUG_TRANSPORTER
00180     ndbout_c("The setsockopt TCP_NODELAY error code = %d", InetErrno);
00181 #endif
00182   }//if
00183 }

Here is the call graph for this function:

Here is the caller graph for this function:

void TCP_Transporter::updateReceiveDataPtr ( Uint32 bytesRead ) [inline, private, virtual]

Update receive data ptr

Definition at line 177 of file TCP_Transporter.hpp.

References ReceiveBuffer::incompleteMessage(), ReceiveBuffer::readPtr, receiveBuffer, and ReceiveBuffer::sizeOfData.

Referenced by TransporterRegistry::performReceive().

00177                                                      {
00178   char * ptr = (char *)receiveBuffer.readPtr;
00179   ptr += bytesRead;
00180   receiveBuffer.readPtr = (Uint32*)ptr;
00181   receiveBuffer.sizeOfData -= bytesRead;
00182   receiveBuffer.incompleteMessage();
00183 }

Here is the call graph for this function:

Here is the caller graph for this function:

void TCP_Transporter::updateWritePtr	(	Uint32	lenBytes,
		Uint32	prio
	)			`[private, virtual]`

Implements Transporter.

Definition at line 295 of file TCP_Transporter.cpp.

References SendBuffer::bufferSize(), doSend(), m_sendBuffer, sendIsPossible(), TCP_SEND_LIMIT, and SendBuffer::updateInsertPtr().

00295                                                            {
00296   m_sendBuffer.updateInsertPtr(lenBytes);
00297   
00298   const int bufsize = m_sendBuffer.bufferSize();
00299   if(bufsize > TCP_SEND_LIMIT) {
00300     //-------------------------------------------------
00301     // Buffer is full and we are ready to send. We will
00302     // not wait since the signal is already in the buffer.
00303     // Force flag set has the same indication that we
00304     // should always send. If it is not possible to send
00305     // we will not worry since we will soon be back for
00306     // a renewed trial.
00307     //-------------------------------------------------
00308     struct timeval no_timeout = {0,0};
00309     if(sendIsPossible(&no_timeout)) {
00310       //-------------------------------------------------
00311       // Send was possible, attempt at a send.
00312       //-------------------------------------------------
00313       doSend();
00314     }//if
00315   }
00316 }