The world's most popular open source database

MySQL 5.1: OSE_Transporter Class Reference

OSE_Transporter Class Reference

#include <OSE_Transporter.hpp>

Inheritance diagram for OSE_Transporter:

[legend]Collaboration diagram for OSE_Transporter:

[legend]List of all members.


Public Member Functions
	OSE_Transporter (int prioASignalSize, int prioBSignalSize, NodeId localNodeId, const char lHostName, NodeId remoteNodeId, NodeId serverNodeId, const char rHostName, int byteorder, bool compression, bool checksum, bool signalId, Uint32 reportFreq=4096)
	~OSE_Transporter ()
bool	initTransporter ()
virtual void	doConnect ()
virtual void	doDisconnect ()
Uint32 *	getWritePtr (Uint32 lenBytes, Uint32 prio)
void	updateWritePtr (Uint32 lenBytes, Uint32 prio)
void	doSend ()
bool	hasDataToSend () const
Protected Member Functions
virtual bool	connectImpl (Uint32 timeOut)
virtual void	disconnectImpl ()
Private Types
	DISCONNECTED = 0
	WAITING_FOR_HUNT = 1
	WAITING_FOR_CONNECT_REQ = 2
	WAITING_FOR_CONNECT_CONF = 3
	CONNECTED = 4
enum	OSETransporterState { DISCONNECTED = 0, WAITING_FOR_HUNT = 1, WAITING_FOR_CONNECT_REQ = 2, WAITING_FOR_CONNECT_CONF = 3, CONNECTED = 4 }
Private Member Functions
NdbTransporterData *	allocPrioASignal (Uint32 lenBytes) const
void	initSignals ()
void	huntReceived (struct NdbTransporterHunt *sig)
bool	connectReq (struct NdbTransporterConnectReq *sig)
bool	connectRef (struct NdbTransporterConnectRef *sig)
bool	connectConf (struct NdbTransporterConnectConf *sig)
bool	disconnectOrd (struct NdbTransporterDisconnectOrd *sig)
Private Attributes
const bool	isServer
int	maxPrioBDataSize
PROCESS	remoteNodePid
OSATTREF	remoteNodeRef
char	remoteNodeName [256]
Uint32	signalIdCounter
int	prioBSignalSize
Uint32 *	prioBInsertPtr
SIGNAL *	prioBSignal
Uint32	reportFreq
Uint32	receiveCount
Uint64	receiveSize
Uint32	sendCount
Uint64	sendSize
enum OSE_Transporter::OSETransporterState	state
Friends
class	OSE_Receiver
class	TransporterRegistry

Detailed Description

Definition at line 40 of file OSE_Transporter.hpp.

Member Enumeration Documentation

enum OSE_Transporter::OSETransporterState [private]

Enumerator:

DISCONNECTED
WAITING_FOR_HUNT
WAITING_FOR_CONNECT_REQ
WAITING_FOR_CONNECT_CONF
CONNECTED

Definition at line 149 of file OSE_Transporter.hpp.

00149                            {
00150     DISCONNECTED              = 0,
00151     WAITING_FOR_HUNT          = 1,
00152     WAITING_FOR_CONNECT_REQ   = 2,
00153     WAITING_FOR_CONNECT_CONF  = 3,
00154     CONNECTED                 = 4
00155   } state;

Constructor & Destructor Documentation

OSE_Transporter::OSE_Transporter	(	int	prioASignalSize,
		int	prioBSignalSize,
		NodeId	localNodeId,
		const char *	lHostName,
		NodeId	remoteNodeId,
		NodeId	serverNodeId,
		const char *	rHostName,
		int	byteorder,
		bool	compression,
		bool	checksum,
		bool	signalId,
		Uint32	reportFreq = `4096`
	)

Definition at line 30 of file OSE_Transporter.cpp.

References NIL, prioBSignal, prioBSignalSize, remoteNodeName, signalIdCounter, BaseString::snprintf(), and strcmp().

00041                                                     :
00042   Transporter(localNodeId,
00043               remoteNodeId,
00044               serverNodeId,
00045               byteorder,
00046               compression,
00047               checksum,
00048               signalId),
00049   isServer(localNodeId < remoteNodeId)
00050 {
00051 
00052   signalIdCounter = 0;
00053   prioBSignalSize = _prioBSignalSize;
00054   
00055   if (strcmp(lHostName, rHostName) == 0){    
00056     BaseString::snprintf(remoteNodeName, sizeof(remoteNodeName), 
00057              "ndb_node%d", remoteNodeId);
00058   } else {
00059     BaseString::snprintf(remoteNodeName, sizeof(remoteNodeName), 
00060              "%s/ndb_node%d", rHostName, remoteNodeId); 
00061   }
00062   
00063   prioBSignal      = NIL;
00064 }

Here is the call graph for this function:

OSE_Transporter::~OSE_Transporter ( )

Definition at line 66 of file OSE_Transporter.cpp.

References NIL, and prioBSignal.

00066                                   {
00067 
00068 #if 0 
00069 
00073   if(prioBSignal != NIL)
00074     free_buf(&prioBSignal);
00075 #endif
00076 }

Member Function Documentation

NdbTransporterData * OSE_Transporter::allocPrioASignal ( Uint32 lenBytes ) const [private]

Definition at line 123 of file OSE_Transporter.cpp.

References Transporter::localNodeId, NDB_TRANSPORTER_PRIO_A, and sig().

Referenced by getWritePtr().

00124 {
00125   
00126   const Uint32 lenBytes = messageLenBytes + sizeof(NdbTransporterData) - 4;
00127   
00128   NdbTransporterData * sig = 
00129     (NdbTransporterData*)alloc(lenBytes, NDB_TRANSPORTER_PRIO_A);
00130   
00131   sig->length = 0;
00132   sig->senderNodeId = localNodeId;
00133   
00134   return sig;
00135 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool OSE_Transporter::connectConf ( struct NdbTransporterConnectConf * sig ) [private]

Definition at line 421 of file OSE_Transporter.cpp.

References CONNECTED, initSignals(), isServer, NDB_TRANSPORTER_DISCONNECT_ORD, SIGNAL::ndbDisconnect, NdbMutex_Lock(), NdbMutex_Unlock(), NIL, prioBSignal, NdbTransporterDisconnectOrd::PROCESS_DIED, NdbTransporterDisconnectOrd::reason, Transporter::remoteNodeId, remoteNodePid, remoteNodeRef, NdbTransporterDisconnectOrd::senderNodeId, sig(), state, WAITING_FOR_CONNECT_CONF, and WARNING.

Referenced by OSE_Receiver::doReceive().

00421                                                                   {
00422   if(isServer){
00423     WARNING("OSE Connect Conf received for server: senderNodeId: " <<
00424             sig->senderNodeId);
00425     return false;
00426   }
00427   if(state != WAITING_FOR_CONNECT_CONF){
00428     WARNING("OSE Connect Conf received while in state: " <<
00429             state);
00430     return false;
00431   }
00432   NdbMutex_Lock(theMutexPtr);
00433 
00434   // Free the buffers to get rid of any "junk" that they might contain
00435   if(prioBSignal != NIL){
00436     restore(prioBSignal);
00437     free_buf(&prioBSignal);
00438   }
00439   initSignals();
00440 
00441   union SIGNAL * discon = alloc(sizeof(NdbTransporterDisconnectOrd),
00442                                 NDB_TRANSPORTER_DISCONNECT_ORD);
00443   discon->ndbDisconnect.senderNodeId = remoteNodeId;
00444   discon->ndbDisconnect.reason= NdbTransporterDisconnectOrd::PROCESS_DIED;
00445   
00446   remoteNodeRef = attach(&discon, remoteNodePid);
00447   
00448   state = CONNECTED;
00449   _connected     = true;
00450   _connecting    = false;
00451   _disconnecting = false;
00452 
00453   // Free the buffers to get rid of any "junk" that they might contain
00454   if(prioBSignal != NIL){
00455     restore(prioBSignal);
00456     free_buf(&prioBSignal);
00457   }
00458   initSignals();
00459   
00460   NdbMutex_Unlock(theMutexPtr);
00461   return true;
00462 }

Here is the call graph for this function:

Here is the caller graph for this function:

virtual bool OSE_Transporter::connectImpl ( Uint32 timeOut ) [inline, protected, virtual]

Not implemented OSE uses async connect/disconnect

Definition at line 95 of file OSE_Transporter.hpp.

00095                                           {
00096     return false;
00097   }

bool OSE_Transporter::connectRef ( struct NdbTransporterConnectRef * sig ) [private]

Definition at line 397 of file OSE_Transporter.cpp.

References doConnect(), doDisconnect(), isServer, sig(), state, WAITING_FOR_CONNECT_CONF, and WARNING.

Referenced by OSE_Receiver::doReceive().

00397                                                                 {
00398   if(isServer){
00399     WARNING("OSE Connect Ref received for server: senderNodeId: " <<
00400             sig->senderNodeId);
00401     return false;
00402   }
00403   if(state != WAITING_FOR_CONNECT_CONF){
00404     WARNING("OSE Connect Ref received for client while in state: " <<
00405             state << " senderNodeId: " << sig->senderNodeId);
00406     return false;
00407   }
00408   doDisconnect();
00409 #if 0
00410 
00414   doConnect();
00415 #endif
00416   return true;
00417 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool OSE_Transporter::connectReq ( struct NdbTransporterConnectReq * sig ) [private]

Definition at line 339 of file OSE_Transporter.cpp.

References CONNECTED, DEBUG, initSignals(), NdbTransporterConnectRef::INVALID_STATE, isServer, Transporter::localNodeId, NDB_TRANSPORTER_CONNECT_CONF, NDB_TRANSPORTER_CONNECT_REF, NDB_TRANSPORTER_DISCONNECT_ORD, SIGNAL::ndbConnectConf, SIGNAL::ndbConnectRef, SIGNAL::ndbDisconnect, NdbMutex_Lock(), NdbMutex_Unlock(), NIL, prioBSignal, PROCESS, NdbTransporterDisconnectOrd::PROCESS_DIED, NdbTransporterDisconnectOrd::reason, NdbTransporterConnectRef::reason, Transporter::remoteNodeId, NdbTransporterConnectConf::remoteNodeId, remoteNodePid, remoteNodeRef, NdbTransporterDisconnectOrd::senderNodeId, NdbTransporterConnectConf::senderNodeId, NdbTransporterConnectRef::senderNodeId, sig(), state, WAITING_FOR_CONNECT_REQ, and WARNING.

Referenced by OSE_Receiver::doReceive().

00339                                                                 {
00340   if(!isServer){
00341     WARNING("OSE Connect Req received for client: senderNodeId: " <<
00342             sig->senderNodeId);
00343     return false;
00344   }
00345   
00346   if(state != WAITING_FOR_CONNECT_REQ){
00347     PROCESS pid = sender((union SIGNAL**)&sig);
00348     union SIGNAL * signal = alloc(sizeof(NdbTransporterConnectRef),
00349                                   NDB_TRANSPORTER_CONNECT_REF);
00350     signal->ndbConnectRef.senderNodeId = localNodeId;
00351     signal->ndbConnectRef.reason = NdbTransporterConnectRef::INVALID_STATE;
00352 
00353     DEBUG("Sending connect ref to pid: " << hex << pid);
00354 
00355     ::send(&signal, pid);
00356     return false;
00357   }
00358   
00359   NdbMutex_Lock(theMutexPtr);
00360 
00361   if(prioBSignal != NIL){
00362     restore(prioBSignal);
00363     free_buf(&prioBSignal);
00364   }
00365   initSignals();
00366 
00367   remoteNodePid = sender((union SIGNAL**)&sig);
00368   union SIGNAL * signal = alloc(sizeof(NdbTransporterConnectRef),
00369                                 NDB_TRANSPORTER_CONNECT_CONF);
00370   signal->ndbConnectConf.senderNodeId = localNodeId;
00371   signal->ndbConnectConf.remoteNodeId = remoteNodeId;
00372 
00373   union SIGNAL * discon = alloc(sizeof(NdbTransporterDisconnectOrd),
00374                                 NDB_TRANSPORTER_DISCONNECT_ORD);
00375   discon->ndbDisconnect.senderNodeId = remoteNodeId;
00376   discon->ndbDisconnect.reason = NdbTransporterDisconnectOrd::PROCESS_DIED;
00377   
00378   DEBUG("Attaching to pid: " << hex << remoteNodePid);
00379 
00380   remoteNodeRef = attach(&discon, remoteNodePid);
00381   
00382   DEBUG("Sending connect conf to pid: " << hex << remoteNodePid);
00383 
00384   ::send(&signal, remoteNodePid);
00385   state = CONNECTED;
00386   
00387   _connected     = true;
00388   _connecting    = false;
00389   _disconnecting = false;
00390 
00391   NdbMutex_Unlock(theMutexPtr);
00392   
00393   return true;
00394 }

Here is the call graph for this function:

Here is the caller graph for this function:

virtual void OSE_Transporter::disconnectImpl ( ) [inline, protected, virtual]

Not implemented OSE uses async connect/disconnect

Implements Transporter.

Definition at line 103 of file OSE_Transporter.hpp.

00103                                {
00104   }

bool OSE_Transporter::disconnectOrd ( struct NdbTransporterDisconnectOrd * sig ) [private]

Definition at line 466 of file OSE_Transporter.cpp.

References CONNECTED, DISCONNECTED, doDisconnect(), NdbTransporterDisconnectOrd::PROCESS_DIED, Transporter::remoteNodeId, reportDisconnect(), sig(), state, and WARNING.

Referenced by OSE_Receiver::doReceive().

00466                                                                       {
00467   if(state != CONNECTED){
00468     WARNING("OSE Disconnect Ord received while in state: " << state <<
00469             " reason: " << sig->reason);
00470     return false;
00471   }
00472 
00473   if(sig->reason == NdbTransporterDisconnectOrd::PROCESS_DIED){
00474     state = DISCONNECTED;
00475   }
00476   
00477   doDisconnect();
00478   reportDisconnect(callbackObj, remoteNodeId,0);
00479   return true;
00480 }

Here is the call graph for this function:

Here is the caller graph for this function:

void OSE_Transporter::doConnect ( ) [virtual]

Connect

Definition at line 248 of file OSE_Transporter.cpp.

References DEBUG, isServer, NDB_TRANSPORTER_HUNT, SIGNAL::ndbHunt, NdbMutex_Lock(), NdbMutex_Unlock(), NULL, Transporter::remoteNodeId, NdbTransporterHunt::remoteNodeId, remoteNodeName, signalIdCounter, state, WAITING_FOR_CONNECT_REQ, and WAITING_FOR_HUNT.

Referenced by connectRef().

00248                            {
00249   
00250   NdbMutex_Lock(theMutexPtr);
00251   if(_connecting || _disconnecting || _connected){
00252     NdbMutex_Unlock(theMutexPtr);
00253     return;
00254   }
00255   
00256   _connecting = true;
00257   signalIdCounter = 0;
00258 
00259   if(isServer){
00260     DEBUG("Waiting for connect req: ");
00261     state = WAITING_FOR_CONNECT_REQ;
00262   } else {
00263     state = WAITING_FOR_HUNT;
00264     
00265     DEBUG("Hunting for: " << remoteNodeName);
00266     
00267     union SIGNAL* huntsig;
00268     huntsig = alloc(sizeof(NdbTransporterHunt), NDB_TRANSPORTER_HUNT);
00269     huntsig->ndbHunt.remoteNodeId = remoteNodeId;
00270     hunt(remoteNodeName, 0, NULL, &huntsig);
00271   }   
00272   NdbMutex_Unlock(theMutexPtr);
00273 }

Here is the call graph for this function:

Here is the caller graph for this function:

void OSE_Transporter::doDisconnect ( ) [virtual]

Disconnect

Reimplemented from Transporter.

Definition at line 276 of file OSE_Transporter.cpp.

References CONNECTED, DEBUG, DISCONNECTED, doSend(), Transporter::localNodeId, NdbTransporterDisconnectOrd::NDB_DISCONNECT, NDB_TRANSPORTER_DISCONNECT_ORD, NdbMutex_Lock(), NdbMutex_Unlock(), remoteNodePid, remoteNodeRef, sig(), state, WAITING_FOR_CONNECT_CONF, WAITING_FOR_CONNECT_REQ, and WAITING_FOR_HUNT.

Referenced by connectRef(), and disconnectOrd().

00276                               {  
00277   NdbMutex_Lock(theMutexPtr);
00278 
00279   switch(state){
00280   case DISCONNECTED:
00281   case WAITING_FOR_HUNT:
00282   case WAITING_FOR_CONNECT_REQ:
00283   case WAITING_FOR_CONNECT_CONF:
00284     break;
00285   case CONNECTED:
00286     {
00287 #if 0      
00288 
00291       DEBUG("Doing send before disconnect");
00292       doSend();
00293 #endif
00294       union SIGNAL * sig = alloc(sizeof(NdbTransporterDisconnectOrd),
00295                                  NDB_TRANSPORTER_DISCONNECT_ORD);
00296       sig->ndbDisconnect.senderNodeId = localNodeId;
00297       sig->ndbDisconnect.reason = NdbTransporterDisconnectOrd::NDB_DISCONNECT;
00298       ::send(&sig, remoteNodePid);
00299       detach(&remoteNodeRef);
00300 
00301     }
00302     break;
00303   }
00304   state = DISCONNECTED;
00305   
00306   _connected = false;
00307   _connecting = false;
00308   _disconnecting = false;
00309 
00310   NdbMutex_Unlock(theMutexPtr);
00311 }

Here is the call graph for this function:

Here is the caller graph for this function:

void OSE_Transporter::doSend ( )

Retrieves the contents of the send buffers, copies it into an OSE signal and sends it. Until the send buffers are empty

restore is always called to make sure the signal buffer is taken over by a process that is alive, this will otherwise lead to that these buffers are removed when the process that allocated them dies

Definition at line 229 of file OSE_Transporter.cpp.

References SIGNAL::dataSignal, initSignals(), NdbTransporterData::length, prioBSignal, remoteNodePid, NdbTransporterData::sigId, and signalIdCounter.

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

00229                         {
00235   restore(prioBSignal);
00236   if(prioBSignal->dataSignal.length > 0){
00237 
00238     prioBSignal->dataSignal.sigId = signalIdCounter;
00239     signalIdCounter++;
00240 
00241     ::send(&prioBSignal, remoteNodePid);
00242   }
00243   
00244   initSignals();
00245 }

Here is the call graph for this function:

Here is the caller graph for this function:

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

Implements Transporter.

Definition at line 138 of file OSE_Transporter.cpp.

References allocPrioASignal(), prioBInsertPtr, and prioBSignal.

00138                                                         {
00139   if(prio >= 1){
00140     prio = 1;
00141     insertPtr  = prioBInsertPtr;
00142     signal     = (NdbTransporterData*)prioBSignal;
00143   } else {
00144     signal    = allocPrioASignal(lenBytes);
00145     insertPtr = &signal->data[0];
00146   }
00147   return insertPtr;
00148 }

Here is the call graph for this function:

bool OSE_Transporter::hasDataToSend ( ) const [inline]

Definition at line 86 of file OSE_Transporter.hpp.

References SIGNAL::dataSignal, NdbTransporterData::length, and prioBSignal.

00086                              {
00087     return prioBSignal->dataSignal.length > 0;
00088   }

void OSE_Transporter::huntReceived ( struct NdbTransporterHunt * sig ) [private]

OSE Receiver callbacks

Definition at line 314 of file OSE_Transporter.cpp.

References DEBUG, isServer, Transporter::localNodeId, NDB_TRANSPORTER_CONNECT_REQ, SIGNAL::ndbConnectReq, Transporter::remoteNodeId, NdbTransporterConnectReq::remoteNodeId, remoteNodePid, NdbTransporterConnectReq::senderNodeId, sig(), state, WAITING_FOR_CONNECT_CONF, WAITING_FOR_HUNT, and WARNING.

Referenced by OSE_Receiver::doReceive().

00314                                                             {
00315   if(isServer){
00316     WARNING("Hunt received for server: remoteNodeId: " <<
00317             sig->remoteNodeId);
00318     return;
00319   }
00320   
00321   if(state != WAITING_FOR_HUNT){
00322     WARNING("Hunt received while in state: " << state);
00323     return;
00324   }
00325   remoteNodePid = sender((union SIGNAL**)&sig);
00326   union SIGNAL * signal = alloc(sizeof(NdbTransporterConnectReq),
00327                                 NDB_TRANSPORTER_CONNECT_REQ);
00328   signal->ndbConnectReq.remoteNodeId = remoteNodeId;
00329   signal->ndbConnectReq.senderNodeId = localNodeId;
00330 
00331   DEBUG("Sending connect req to pid: " << hex << remoteNodePid);
00332   
00333   ::send(&signal, remoteNodePid);
00334   state = WAITING_FOR_CONNECT_CONF;
00335   return;
00336 }

Here is the call graph for this function:

Here is the caller graph for this function:

void OSE_Transporter::initSignals ( ) [private]

Definition at line 111 of file OSE_Transporter.cpp.

References NdbTransporterData::data, SIGNAL::dataSignal, NdbTransporterData::length, Transporter::localNodeId, NDB_TRANSPORTER_DATA, NIL, prioBInsertPtr, prioBSignal, prioBSignalSize, and NdbTransporterData::senderNodeId.

Referenced by connectConf(), connectReq(), doSend(), and initTransporter().

00111                             {
00112   if(prioBSignal == NIL){
00113     prioBSignal = alloc(prioBSignalSize, NDB_TRANSPORTER_DATA);
00114     prioBInsertPtr = &prioBSignal->dataSignal.data[0];
00115   
00116     prioBSignal->dataSignal.length = 0;
00117     prioBSignal->dataSignal.senderNodeId = localNodeId;
00118   }
00119   dataToSend = 0;
00120 }

Here is the caller graph for this function:

bool OSE_Transporter::initTransporter ( ) [virtual]

Allocate buffers for sending and receiving

Implements Transporter.

Definition at line 79 of file OSE_Transporter.cpp.

References DEBUG, initSignals(), MAX_MESSAGE_SIZE, maxPrioBDataSize, NULL, and prioBSignalSize.

Referenced by TransporterRegistry::createOSETransporter().

00079                                  {
00080 
00081   struct OS_pcb * pcb = get_pcb(current_process());
00082   if(pcb != NULL){
00083     if(pcb->type != OS_ILLEGAL){
00084       if(prioBSignalSize > pcb->max_sigsize){
00085         DEBUG("prioBSignalSize(" << prioBSignalSize << ") > max_sigsize("
00086               << pcb->max_sigsize << ") using max_sigsize");
00087         prioBSignalSize = pcb->max_sigsize;
00088       }
00089     }
00090     free_buf((union SIGNAL **)&pcb);
00091   }
00092 
00093   maxPrioBDataSize = prioBSignalSize;
00094   maxPrioBDataSize -= (sizeof(NdbTransporterData) + MAX_MESSAGE_SIZE - 4);
00095   
00096   if(maxPrioBDataSize < 0){
00097     
00098 #ifdef DEBUG_TRANSPORTER
00099     printf("maxPrioBDataSize < 0 %d\n",
00100            maxPrioBDataSize);
00101 #endif
00102     return false;
00103   }
00104 
00105   initSignals();
00106   
00107   return true;
00108 }

Here is the call graph for this function:

Here is the caller graph for this function:

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

Implements Transporter.

Definition at line 151 of file OSE_Transporter.cpp.

References doSend(), maxPrioBDataSize, prioBInsertPtr, and remoteNodePid.

00151                                                            {
00152 
00153   Uint32 bufferSize = signal->length;
00154   bufferSize += lenBytes;
00155   signal->length = bufferSize;
00156   if(prio >= 1){
00157     prioBInsertPtr += (lenBytes / 4);
00158     if(bufferSize >= maxPrioBDataSize)
00159       doSend();
00160   } else {
00164     signal->sigId = 0;
00165     
00166     ::send((union SIGNAL**)&signal, remoteNodePid);
00167   }
00168 }