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

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

MySQL数据库

开发平台：

Visual C++

TransporterRegistry.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 <ndb_global.h>
  18. #include <my_pthread.h>
  20. #include <TransporterRegistry.hpp>
  21. #include "TransporterInternalDefinitions.hpp"
  23. #include "Transporter.hpp"
  24. #include <SocketAuthenticator.hpp>
  26. #ifdef NDB_TCP_TRANSPORTER
  27. #include "TCP_Transporter.hpp"
  28. #endif
  30. #ifdef NDB_OSE_TRANSPORTER
  31. #include "OSE_Receiver.hpp"
  32. #include "OSE_Transporter.hpp"
  33. #endif
  35. #ifdef NDB_SCI_TRANSPORTER
  36. #include "SCI_Transporter.hpp"
  37. #endif
  39. #ifdef NDB_SHM_TRANSPORTER
  40. #include "SHM_Transporter.hpp"
  41. extern int g_ndb_shm_signum;
  42. #endif
  44. #include "TransporterCallback.hpp"
  45. #include "NdbOut.hpp"
  46. #include <NdbSleep.h>
  47. #include <NdbTick.h>
  48. #include <InputStream.hpp>
  49. #include <OutputStream.hpp>
  51. #include <EventLogger.hpp>
  52. extern EventLogger g_eventLogger;
  54. struct in_addr
  55. TransporterRegistry::get_connect_address(NodeId node_id) const
  56. {
  57.   return theTransporters[node_id]->m_connect_address;
  58. }
  60. SocketServer::Session * TransporterService::newSession(NDB_SOCKET_TYPE sockfd)
  61. {
  62.   DBUG_ENTER("SocketServer::Session * TransporterService::newSession");
  63.   if (m_auth && !m_auth->server_authenticate(sockfd)){
  64.     NDB_CLOSE_SOCKET(sockfd);
  65.     DBUG_RETURN(0);
  66.   }
  68.   if (!m_transporter_registry->connect_server(sockfd))
  69.   {
  70.     NDB_CLOSE_SOCKET(sockfd);
  71.     DBUG_RETURN(0);
  72.   }
  74.   DBUG_RETURN(0);
  75. }
  77. TransporterRegistry::TransporterRegistry(void * callback,
  78.  unsigned _maxTransporters,
  79.  unsigned sizeOfLongSignalMemory)
  80. {
  81.   DBUG_ENTER("TransporterRegistry::TransporterRegistry");
  83.   nodeIdSpecified = false;
  84.   maxTransporters = _maxTransporters;
  85.   sendCounter = 1;
  86.   
  87.   callbackObj=callback;
  89.   theTCPTransporters  = new TCP_Transporter * [maxTransporters];
  90.   theSCITransporters  = new SCI_Transporter * [maxTransporters];
  91.   theSHMTransporters  = new SHM_Transporter * [maxTransporters];
  92.   theOSETransporters  = new OSE_Transporter * [maxTransporters];
  93.   theTransporterTypes = new TransporterType   [maxTransporters];
  94.   theTransporters     = new Transporter     * [maxTransporters];
  95.   performStates       = new PerformState      [maxTransporters];
  96.   ioStates            = new IOState           [maxTransporters]; 
  97.   
  98.   // Initialize member variables
  99.   nTransporters    = 0;
  100.   nTCPTransporters = 0;
  101.   nSCITransporters = 0;
  102.   nSHMTransporters = 0;
  103.   nOSETransporters = 0;
  104.   
  105.   // Initialize the transporter arrays
  106.   for (unsigned i=0; i<maxTransporters; i++) {
  107.     theTCPTransporters[i] = NULL;
  108.     theSCITransporters[i] = NULL;
  109.     theSHMTransporters[i] = NULL;
  110.     theOSETransporters[i] = NULL;
  111.     theTransporters[i]    = NULL;
  112.     performStates[i]      = DISCONNECTED;
  113.     ioStates[i]           = NoHalt;
  114.   }
  115.   theOSEReceiver = 0;
  116.   theOSEJunkSocketSend = 0;
  117.   theOSEJunkSocketRecv = 0;
  119.   DBUG_VOID_RETURN;
  120. }
  122. TransporterRegistry::~TransporterRegistry()
  123. {
  124.   DBUG_ENTER("TransporterRegistry::~TransporterRegistry");
  125.   
  126.   removeAll();
  127.   
  128.   delete[] theTCPTransporters;
  129.   delete[] theSCITransporters;
  130.   delete[] theSHMTransporters;
  131.   delete[] theOSETransporters;
  132.   delete[] theTransporterTypes;
  133.   delete[] theTransporters;
  134.   delete[] performStates;
  135.   delete[] ioStates;
  137. #ifdef NDB_OSE_TRANSPORTER
  138.   if(theOSEReceiver != NULL){
  139.     theOSEReceiver->destroyPhantom();
  140.     delete theOSEReceiver;
  141.     theOSEReceiver = 0;
  142.   }
  143. #endif
  145.   DBUG_VOID_RETURN;
  146. }
  148. void
  149. TransporterRegistry::removeAll(){
  150.   for(unsigned i = 0; i<maxTransporters; i++){
  151.     if(theTransporters[i] != NULL)
  152.       removeTransporter(theTransporters[i]->getRemoteNodeId());
  153.   }
  154. }
  156. void
  157. TransporterRegistry::disconnectAll(){
  158.   for(unsigned i = 0; i<maxTransporters; i++){
  159.     if(theTransporters[i] != NULL)
  160.       theTransporters[i]->doDisconnect();
  161.   }
  162. }
  164. bool
  165. TransporterRegistry::init(NodeId nodeId) {
  166.   DBUG_ENTER("TransporterRegistry::init");
  167.   nodeIdSpecified = true;
  168.   localNodeId = nodeId;
  169.   
  170.   DEBUG("TransporterRegistry started node: " << localNodeId);
  171.   
  172.   DBUG_RETURN(true);
  173. }
  175. bool
  176. TransporterRegistry::connect_server(NDB_SOCKET_TYPE sockfd)
  177. {
  178.   DBUG_ENTER("TransporterRegistry::connect_server");
  180.   // read node id from client
  181.   // read transporter type
  182.   int nodeId, remote_transporter_type= -1;
  183.   SocketInputStream s_input(sockfd);
  184.   char buf[256];
  185.   if (s_input.gets(buf, 256) == 0) {
  186.     DBUG_PRINT("error", ("Could not get node id from client"));
  187.     DBUG_RETURN(false);
  188.   }
  189.   int r= sscanf(buf, "%d %d", &nodeId, &remote_transporter_type);
  190.   switch (r) {
  191.   case 2:
  192.     break;
  193.   case 1:
  194.     // we're running version prior to 4.1.9
  195.     // ok, but with no checks on transporter configuration compatability
  196.     break;
  197.   default:
  198.     DBUG_PRINT("error", ("Error in node id from client"));
  199.     DBUG_RETURN(false);
  200.   }
  202.   DBUG_PRINT("info", ("nodeId=%d remote_transporter_type=%d",
  203.       nodeId,remote_transporter_type));
  205.   //check that nodeid is valid and that there is an allocated transporter
  206.   if ( nodeId < 0 || nodeId >= (int)maxTransporters) {
  207.     DBUG_PRINT("error", ("Node id out of range from client"));
  208.     DBUG_RETURN(false);
  209.   }
  210.   if (theTransporters[nodeId] == 0) {
  211.       DBUG_PRINT("error", ("No transporter for this node id from client"));
  212.       DBUG_RETURN(false);
  213.   }
  215.   //check that the transporter should be connected
  216.   if (performStates[nodeId] != TransporterRegistry::CONNECTING) {
  217.     DBUG_PRINT("error", ("Transporter in wrong state for this node id from client"));
  218.     DBUG_RETURN(false);
  219.   }
  221.   Transporter *t= theTransporters[nodeId];
  223.   // send info about own id (just as response to acknowledge connection)
  224.   // send info on own transporter type
  225.   SocketOutputStream s_output(sockfd);
  226.   s_output.println("%d %d", t->getLocalNodeId(), t->m_type);
  228.   if (remote_transporter_type != -1)
  229.   {
  230.     if (remote_transporter_type != t->m_type)
  231.     {
  232.       DBUG_PRINT("error", ("Transporter types mismatch this=%d remote=%d",
  233.    t->m_type, remote_transporter_type));
  234.       g_eventLogger.error("Incompatible configuration: Transporter type "
  235.   "mismatch with node %d", nodeId);
  237.       // wait for socket close for 1 second to let message arrive at client
  238.       {
  239. fd_set a_set;
  240. FD_ZERO(&a_set);
  241. FD_SET(sockfd, &a_set);
  242. struct timeval timeout;
  243. timeout.tv_sec  = 1; timeout.tv_usec = 0;
  244. select(sockfd+1, &a_set, 0, 0, &timeout);
  245.       }
  246.       DBUG_RETURN(false);
  247.     }
  248.   }
  249.   else if (t->m_type == tt_SHM_TRANSPORTER)
  250.   {
  251.     g_eventLogger.warning("Unable to verify transporter compatability with node %d", nodeId);
  252.   }
  254.   // setup transporter (transporter responsible for closing sockfd)
  255.   t->connect_server(sockfd);
  257.   DBUG_RETURN(true);
  258. }
  260. bool
  261. TransporterRegistry::createTransporter(TCP_TransporterConfiguration *config) {
  262. #ifdef NDB_TCP_TRANSPORTER
  264.   if(!nodeIdSpecified){
  265.     init(config->localNodeId);
  266.   }
  267.   
  268.   if(config->localNodeId != localNodeId) 
  269.     return false;
  270.   
  271.   if(theTransporters[config->remoteNodeId] != NULL)
  272.     return false;
  273.    
  274.   TCP_Transporter * t = new TCP_Transporter(*this,
  275.     config->sendBufferSize,
  276.     config->maxReceiveSize,
  277.     config->localHostName,
  278.     config->remoteHostName,
  279.     config->port,
  280.     localNodeId,
  281.     config->remoteNodeId,
  282.     config->checksum,
  283.     config->signalId);
  284.   if (t == NULL) 
  285.     return false;
  286.   else if (!t->initTransporter()) {
  287.     delete t;
  288.     return false;
  289.   }
  291.   // Put the transporter in the transporter arrays
  292.   theTCPTransporters[nTCPTransporters]      = t;
  293.   theTransporters[t->getRemoteNodeId()]     = t;
  294.   theTransporterTypes[t->getRemoteNodeId()] = tt_TCP_TRANSPORTER;
  295.   performStates[t->getRemoteNodeId()]       = DISCONNECTED;
  296.   nTransporters++;
  297.   nTCPTransporters++;
  299. #if defined NDB_OSE || defined NDB_SOFTOSE
  300.   t->theReceiverPid = theReceiverPid;
  301. #endif
  302.   
  303.   return true;
  304. #else
  305.   return false;
  306. #endif
  307. }
  309. bool
  310. TransporterRegistry::createTransporter(OSE_TransporterConfiguration *conf) {
  311. #ifdef NDB_OSE_TRANSPORTER
  313.   if(!nodeIdSpecified){
  314.     init(conf->localNodeId);
  315.   }
  316.   
  317.   if(conf->localNodeId != localNodeId)
  318.     return false;
  319.   
  320.   if(theTransporters[conf->remoteNodeId] != NULL)
  321.     return false;
  323.   if(theOSEReceiver == NULL){
  324.     theOSEReceiver = new OSE_Receiver(this,
  325.       10,
  326.       localNodeId);
  327.   }
  328.   
  329.   OSE_Transporter * t = new OSE_Transporter(conf->prioASignalSize,
  330.     conf->prioBSignalSize,
  331.     localNodeId,
  332.     conf->localHostName,
  333.     conf->remoteNodeId,
  334.     conf->remoteHostName,
  335.     conf->checksum,
  336.     conf->signalId);
  337.   if (t == NULL)
  338.     return false;
  339.   else if (!t->initTransporter()) {
  340.     delete t;
  341.     return false;
  342.   }
  343.   // Put the transporter in the transporter arrays
  344.   theOSETransporters[nOSETransporters]      = t;
  345.   theTransporters[t->getRemoteNodeId()]     = t;
  346.   theTransporterTypes[t->getRemoteNodeId()] = tt_OSE_TRANSPORTER;
  347.   performStates[t->getRemoteNodeId()]       = DISCONNECTED;
  348.   
  349.   nTransporters++;
  350.   nOSETransporters++;
  352.   return true;
  353. #else
  354.   return false;
  355. #endif
  356. }
  358. bool
  359. TransporterRegistry::createTransporter(SCI_TransporterConfiguration *config) {
  360. #ifdef NDB_SCI_TRANSPORTER
  362.   if(!SCI_Transporter::initSCI())
  363.     abort();
  364.   
  365.   if(!nodeIdSpecified){
  366.     init(config->localNodeId);
  367.   }
  368.   
  369.   if(config->localNodeId != localNodeId)
  370.     return false;
  371.  
  372.   if(theTransporters[config->remoteNodeId] != NULL)
  373.     return false;
  374.  
  375.   SCI_Transporter * t = new SCI_Transporter(*this,
  376.                                             config->localHostName,
  377.                                             config->remoteHostName,
  378.                                             config->port,
  379.                                             config->sendLimit, 
  380.     config->bufferSize,
  381.     config->nLocalAdapters,
  382.     config->remoteSciNodeId0,
  383.     config->remoteSciNodeId1,
  384.     localNodeId,
  385.     config->remoteNodeId,
  386.     config->checksum,
  387.     config->signalId);
  388.   
  389.   if (t == NULL) 
  390.     return false;
  391.   else if (!t->initTransporter()) {
  392.     delete t;
  393.     return false;
  394.   }
  395.   // Put the transporter in the transporter arrays
  396.   theSCITransporters[nSCITransporters]      = t;
  397.   theTransporters[t->getRemoteNodeId()]     = t;
  398.   theTransporterTypes[t->getRemoteNodeId()] = tt_SCI_TRANSPORTER;
  399.   performStates[t->getRemoteNodeId()]       = DISCONNECTED;
  400.   nTransporters++;
  401.   nSCITransporters++;
  402.   
  403.   return true;
  404. #else
  405.   return false;
  406. #endif
  407. }
  409. bool
  410. TransporterRegistry::createTransporter(SHM_TransporterConfiguration *config) {
  411.   DBUG_ENTER("TransporterRegistry::createTransporter SHM");
  412. #ifdef NDB_SHM_TRANSPORTER
  413.   if(!nodeIdSpecified){
  414.     init(config->localNodeId);
  415.   }
  416.   
  417.   if(config->localNodeId != localNodeId)
  418.     return false;
  419.   
  420.   if (!g_ndb_shm_signum) {
  421.     g_ndb_shm_signum= config->signum;
  422.     DBUG_PRINT("info",("Block signum %d",g_ndb_shm_signum));
  423.     /**
  424.      * Make sure to block g_ndb_shm_signum
  425.      *   TransporterRegistry::init is run from "main" thread
  426.      */
  427.     sigset_t mask;
  428.     sigemptyset(&mask);
  429.     sigaddset(&mask, g_ndb_shm_signum);
  430.     pthread_sigmask(SIG_BLOCK, &mask, 0);
  431.   }
  433.   if(config->signum != g_ndb_shm_signum)
  434.     return false;
  435.   
  436.   if(theTransporters[config->remoteNodeId] != NULL)
  437.     return false;
  439.   SHM_Transporter * t = new SHM_Transporter(*this,
  440.     config->localHostName,
  441.     config->remoteHostName,
  442.     config->port,
  443.     localNodeId,
  444.     config->remoteNodeId,
  445.     config->checksum,
  446.     config->signalId,
  447.     config->shmKey,
  448.     config->shmSize
  449.     );
  450.   if (t == NULL)
  451.     return false;
  452.   else if (!t->initTransporter()) {
  453.     delete t;
  454.     return false;
  455.   }
  456.   // Put the transporter in the transporter arrays
  457.   theSHMTransporters[nSHMTransporters]      = t;
  458.   theTransporters[t->getRemoteNodeId()]     = t;
  459.   theTransporterTypes[t->getRemoteNodeId()] = tt_SHM_TRANSPORTER;
  460.   performStates[t->getRemoteNodeId()]       = DISCONNECTED;
  461.   
  462.   nTransporters++;
  463.   nSHMTransporters++;
  465.   DBUG_RETURN(true);
  466. #else
  467.   DBUG_RETURN(false);
  468. #endif
  469. }
  472. void
  473. TransporterRegistry::removeTransporter(NodeId nodeId) {
  475.   DEBUG("Removing transporter from " << localNodeId
  476. << " to " << nodeId);
  477.   
  478.   if(theTransporters[nodeId] == NULL)
  479.     return;
  480.   
  481.   theTransporters[nodeId]->doDisconnect();
  482.   
  483.   const TransporterType type = theTransporterTypes[nodeId];
  485.   int ind = 0;
  486.   switch(type){
  487.   case tt_TCP_TRANSPORTER:
  488. #ifdef NDB_TCP_TRANSPORTER
  489.     for(; ind < nTCPTransporters; ind++)
  490.       if(theTCPTransporters[ind]->getRemoteNodeId() == nodeId)
  491. break;
  492.     ind++;
  493.     for(; ind<nTCPTransporters; ind++)
  494.       theTCPTransporters[ind-1] = theTCPTransporters[ind];
  495.     nTCPTransporters --;
  496. #endif
  497.     break;
  498.   case tt_SCI_TRANSPORTER:
  499. #ifdef NDB_SCI_TRANSPORTER
  500.     for(; ind < nSCITransporters; ind++)
  501.       if(theSCITransporters[ind]->getRemoteNodeId() == nodeId)
  502. break;
  503.     ind++;
  504.     for(; ind<nSCITransporters; ind++)
  505.       theSCITransporters[ind-1] = theSCITransporters[ind];
  506.     nSCITransporters --;
  507. #endif
  508.     break;
  509.   case tt_SHM_TRANSPORTER:
  510. #ifdef NDB_SHM_TRANSPORTER
  511.     for(; ind < nSHMTransporters; ind++)
  512.       if(theSHMTransporters[ind]->getRemoteNodeId() == nodeId)
  513. break;
  514.     ind++;
  515.     for(; ind<nSHMTransporters; ind++)
  516.       theSHMTransporters[ind-1] = theSHMTransporters[ind];
  517.     nSHMTransporters --;
  518. #endif
  519.     break;
  520.   case tt_OSE_TRANSPORTER:
  521. #ifdef NDB_OSE_TRANSPORTER
  522.     for(; ind < nOSETransporters; ind++)
  523.       if(theOSETransporters[ind]->getRemoteNodeId() == nodeId)
  524. break;
  525.     ind++;
  526.     for(; ind<nOSETransporters; ind++)
  527.       theOSETransporters[ind-1] = theOSETransporters[ind];
  528.     nOSETransporters --;
  529. #endif
  530.     break;
  531.   }
  532.   
  533.   nTransporters--;
  535.   // Delete the transporter and remove it from theTransporters array
  536.   delete theTransporters[nodeId];
  537.   theTransporters[nodeId] = NULL;        
  538. }
  540. Uint32
  541. TransporterRegistry::get_free_buffer(Uint32 node) const
  542. {
  543.   Transporter *t;
  544.   if(likely((t = theTransporters[node]) != 0))
  545.   {
  546.     return t->get_free_buffer();
  547.   }
  548.   return 0;
  549. }
  552. SendStatus
  553. TransporterRegistry::prepareSend(const SignalHeader * const signalHeader, 
  554.  Uint8 prio,
  555.  const Uint32 * const signalData,
  556.  NodeId nodeId, 
  557.  const LinearSectionPtr ptr[3]){
  560.   Transporter *t = theTransporters[nodeId];
  561.   if(t != NULL && 
  562.      (((ioStates[nodeId] != HaltOutput) && (ioStates[nodeId] != HaltIO)) || 
  563.       ((signalHeader->theReceiversBlockNumber == 252) ||
  564.        (signalHeader->theReceiversBlockNumber == 4002)))) {
  565.  
  566.     if(t->isConnected()){
  567.       Uint32 lenBytes = t->m_packer.getMessageLength(signalHeader, ptr);
  568.       if(lenBytes <= MAX_MESSAGE_SIZE){
  569. Uint32 * insertPtr = t->getWritePtr(lenBytes, prio);
  570. if(insertPtr != 0){
  571.   t->m_packer.pack(insertPtr, prio, signalHeader, signalData, ptr);
  572.   t->updateWritePtr(lenBytes, prio);
  573.   return SEND_OK;
  574. }
  576. int sleepTime = 2;
  578. /**
  579.  * @note: on linux/i386 the granularity is 10ms
  580.  *        so sleepTime = 2 generates a 10 ms sleep.
  581.  */
  582. for(int i = 0; i<50; i++){
  583.   if((nSHMTransporters+nSCITransporters) == 0)
  584.     NdbSleep_MilliSleep(sleepTime); 
  585.   insertPtr = t->getWritePtr(lenBytes, prio);
  586.   if(insertPtr != 0){
  587.     t->m_packer.pack(insertPtr, prio, signalHeader, signalData, ptr);
  588.     t->updateWritePtr(lenBytes, prio);
  589.     break;
  590.   }
  591. }
  593. if(insertPtr != 0){
  594.   /**
  595.    * Send buffer full, but resend works
  596.    */
  597.   reportError(callbackObj, nodeId, TE_SEND_BUFFER_FULL);
  598.   return SEND_OK;
  599. }
  601. WARNING("Signal to " << nodeId << " lost(buffer)");
  602. reportError(callbackObj, nodeId, TE_SIGNAL_LOST_SEND_BUFFER_FULL);
  603. return SEND_BUFFER_FULL;
  604.       } else {
  605. return SEND_MESSAGE_TOO_BIG;
  606.       }
  607.     } else {
  608.       DEBUG("Signal to " << nodeId << " lost(disconnect) ");
  609.       return SEND_DISCONNECTED;
  610.     }
  611.   } else {
  612.     DEBUG("Discarding message to block: " 
  613.   << signalHeader->theReceiversBlockNumber 
  614.   << " node: " << nodeId);
  615.     
  616.     if(t == NULL)
  617.       return SEND_UNKNOWN_NODE;
  618.     
  619.     return SEND_BLOCKED;
  620.   }
  621. }
  623. SendStatus
  624. TransporterRegistry::prepareSend(const SignalHeader * const signalHeader, 
  625.  Uint8 prio,
  626.  const Uint32 * const signalData,
  627.  NodeId nodeId, 
  628.  class SectionSegmentPool & thePool,
  629.  const SegmentedSectionPtr ptr[3]){
  630.   
  632.   Transporter *t = theTransporters[nodeId];
  633.   if(t != NULL && 
  634.      (((ioStates[nodeId] != HaltOutput) && (ioStates[nodeId] != HaltIO)) || 
  635.       ((signalHeader->theReceiversBlockNumber == 252)|| 
  636.        (signalHeader->theReceiversBlockNumber == 4002)))) {
  637.     
  638.     if(t->isConnected()){
  639.       Uint32 lenBytes = t->m_packer.getMessageLength(signalHeader, ptr);
  640.       if(lenBytes <= MAX_MESSAGE_SIZE){
  641. Uint32 * insertPtr = t->getWritePtr(lenBytes, prio);
  642. if(insertPtr != 0){
  643.   t->m_packer.pack(insertPtr, prio, signalHeader, signalData, thePool, ptr);
  644.   t->updateWritePtr(lenBytes, prio);
  645.   return SEND_OK;
  646. }
  649. /**
  650.  * @note: on linux/i386 the granularity is 10ms
  651.  *        so sleepTime = 2 generates a 10 ms sleep.
  652.  */
  653. int sleepTime = 2;
  654. for(int i = 0; i<50; i++){
  655.   if((nSHMTransporters+nSCITransporters) == 0)
  656.     NdbSleep_MilliSleep(sleepTime); 
  657.   insertPtr = t->getWritePtr(lenBytes, prio);
  658.   if(insertPtr != 0){
  659.     t->m_packer.pack(insertPtr, prio, signalHeader, signalData, thePool, ptr);
  660.     t->updateWritePtr(lenBytes, prio);
  661.     break;
  662.   }
  663. }
  665. if(insertPtr != 0){
  666.   /**
  667.    * Send buffer full, but resend works
  668.    */
  669.   reportError(callbackObj, nodeId, TE_SEND_BUFFER_FULL);
  670.   return SEND_OK;
  671. }
  673. WARNING("Signal to " << nodeId << " lost(buffer)");
  674. reportError(callbackObj, nodeId, TE_SIGNAL_LOST_SEND_BUFFER_FULL);
  675. return SEND_BUFFER_FULL;
  676.       } else {
  677. return SEND_MESSAGE_TOO_BIG;
  678.       }
  679.     } else {
  680.       DEBUG("Signal to " << nodeId << " lost(disconnect) ");
  681.       return SEND_DISCONNECTED;
  682.     }
  683.   } else {
  684.     DEBUG("Discarding message to block: " 
  685.   << signalHeader->theReceiversBlockNumber 
  686.   << " node: " << nodeId);
  687.     
  688.     if(t == NULL)
  689.       return SEND_UNKNOWN_NODE;
  690.     
  691.     return SEND_BLOCKED;
  692.   }
  693. }
  695. void
  696. TransporterRegistry::external_IO(Uint32 timeOutMillis) {
  697.   //-----------------------------------------------------------
  698.   // Most of the time we will send the buffers here and then wait
  699.   // for new signals. Thus we start by sending without timeout
  700.   // followed by the receive part where we expect to sleep for
  701.   // a while.
  702.   //-----------------------------------------------------------
  703.   if(pollReceive(timeOutMillis)){
  704.     performReceive();
  705.   }
  706.   performSend();
  707. }
  709. Uint32
  710. TransporterRegistry::pollReceive(Uint32 timeOutMillis){
  711.   Uint32 retVal = 0;
  712. #ifdef NDB_OSE_TRANSPORTER
  713.   retVal |= poll_OSE(timeOutMillis);
  714.   retVal |= poll_TCP(0);
  715.   return retVal;
  716. #endif
  717.   
  718.   if((nSCITransporters) > 0)
  719.   {
  720.     timeOutMillis=0;
  721.   }
  723. #ifdef NDB_SHM_TRANSPORTER
  724.   if(nSHMTransporters > 0)
  725.   {
  726.     Uint32 res = poll_SHM(0);
  727.     if(res)
  728.     {
  729.       retVal |= res;
  730.       timeOutMillis = 0;
  731.     }
  732.   }
  733. #endif
  735. #ifdef NDB_TCP_TRANSPORTER
  736.   if(nTCPTransporters > 0 || retVal == 0)
  737.   {
  738.     retVal |= poll_TCP(timeOutMillis);
  739.   }
  740.   else
  741.     tcpReadSelectReply = 0;
  742. #endif
  743. #ifdef NDB_SCI_TRANSPORTER
  744.   if(nSCITransporters > 0)
  745.     retVal |= poll_SCI(timeOutMillis);
  746. #endif
  747. #ifdef NDB_SHM_TRANSPORTER
  748.   if(nSHMTransporters > 0 && retVal == 0)
  749.   {
  750.     int res = poll_SHM(0);
  751.     retVal |= res;
  752.   }
  753. #endif
  754.   return retVal;
  755. }
  758. #ifdef NDB_SCI_TRANSPORTER
  759. Uint32
  760. TransporterRegistry::poll_SCI(Uint32 timeOutMillis)
  761. {
  762.   for (int i=0; i<nSCITransporters; i++) {
  763.     SCI_Transporter * t = theSCITransporters[i];
  764.     if (t->isConnected()) {
  765.       if(t->hasDataToRead())
  766. return 1;
  767.     }
  768.   }
  769.   return 0;
  770. }
  771. #endif
  774. #ifdef NDB_SHM_TRANSPORTER
  775. static int g_shm_counter = 0;
  776. Uint32
  777. TransporterRegistry::poll_SHM(Uint32 timeOutMillis)
  778. {  
  779.   for(int j=0; j < 100; j++)
  780.   {
  781.     for (int i=0; i<nSHMTransporters; i++) {
  782.       SHM_Transporter * t = theSHMTransporters[i];
  783.       if (t->isConnected()) {
  784. if(t->hasDataToRead()) {
  785.   return 1;
  786. }
  787.       }
  788.     }
  789.   }
  790.   return 0;
  791. }
  792. #endif
  794. #ifdef NDB_OSE_TRANSPORTER
  795. Uint32
  796. TransporterRegistry::poll_OSE(Uint32 timeOutMillis)
  797. {
  798.   if(theOSEReceiver != NULL){
  799.     return theOSEReceiver->doReceive(timeOutMillis);
  800.   }
  801.   NdbSleep_MilliSleep(timeOutMillis);
  802.   return 0;
  803. }
  804. #endif
  806. #ifdef NDB_TCP_TRANSPORTER
  807. Uint32 
  808. TransporterRegistry::poll_TCP(Uint32 timeOutMillis)
  809. {
  810.   if (false && nTCPTransporters == 0)
  811.   {
  812.     tcpReadSelectReply = 0;
  813.     return 0;
  814.   }
  815.   
  816.   struct timeval timeout;
  817. #ifdef NDB_OSE
  818.   // Return directly if there are no TCP transporters configured
  819.   
  820.   if(timeOutMillis <= 1){
  821.     timeout.tv_sec  = 0;
  822.     timeout.tv_usec = 1025;
  823.   } else {
  824.     timeout.tv_sec  = timeOutMillis / 1000;
  825.     timeout.tv_usec = (timeOutMillis % 1000) * 1000;
  826.   }
  827. #else  
  828.   timeout.tv_sec  = timeOutMillis / 1000;
  829.   timeout.tv_usec = (timeOutMillis % 1000) * 1000;
  830. #endif
  832.   NDB_SOCKET_TYPE maxSocketValue = -1;
  833.   
  834.   // Needed for TCP/IP connections
  835.   // The read- and writeset are used by select
  836.   
  837.   FD_ZERO(&tcpReadset);
  839.   // Prepare for sending and receiving
  840.   for (int i = 0; i < nTCPTransporters; i++) {
  841.     TCP_Transporter * t = theTCPTransporters[i];
  842.     
  843.     // If the transporter is connected
  844.     if (t->isConnected()) {
  845.       
  846.       const NDB_SOCKET_TYPE socket = t->getSocket();
  847.       // Find the highest socket value. It will be used by select
  848.       if (socket > maxSocketValue)
  849. maxSocketValue = socket;
  850.       
  851.       // Put the connected transporters in the socket read-set 
  852.       FD_SET(socket, &tcpReadset);
  853.     }
  854.   }
  855.   
  856.   // The highest socket value plus one
  857.   maxSocketValue++; 
  858.   
  859.   tcpReadSelectReply = select(maxSocketValue, &tcpReadset, 0, 0, &timeout);  
  860.   if(false && tcpReadSelectReply == -1 && errno == EINTR)
  861.     ndbout_c("woke-up by signal");
  863. #ifdef NDB_WIN32
  864.   if(tcpReadSelectReply == SOCKET_ERROR)
  865.   {
  866.     NdbSleep_MilliSleep(timeOutMillis);
  867.   }
  868. #endif
  869.   
  870.   return tcpReadSelectReply;
  871. }
  872. #endif
  875. void
  876. TransporterRegistry::performReceive()
  877. {
  878. #ifdef NDB_OSE_TRANSPORTER
  879.   if(theOSEReceiver != 0)
  880.   {
  881.     while(theOSEReceiver->hasData())
  882.     {
  883.       NodeId remoteNodeId;
  884.       Uint32 * readPtr;
  885.       Uint32 sz = theOSEReceiver->getReceiveData(&remoteNodeId, &readPtr);
  886.       Uint32 szUsed = unpack(readPtr,
  887.      sz,
  888.      remoteNodeId,
  889.      ioStates[remoteNodeId]);
  890. #ifdef DEBUG_TRANSPORTER
  891.       /**
  892.        * OSE transporter can handle executions of
  893.        *   half signals
  894.        */
  895.       assert(sz == szUsed);
  896. #endif
  897.       theOSEReceiver->updateReceiveDataPtr(szUsed);
  898.       theOSEReceiver->doReceive(0);
  899.       //      checkJobBuffer();
  900.     }
  901.   }
  902. #endif
  904. #ifdef NDB_TCP_TRANSPORTER
  905.   if(tcpReadSelectReply > 0)
  906.   {
  907.     for (int i=0; i<nTCPTransporters; i++) 
  908.     {
  909.       checkJobBuffer();
  910.       TCP_Transporter *t = theTCPTransporters[i];
  911.       const NodeId nodeId = t->getRemoteNodeId();
  912.       const NDB_SOCKET_TYPE socket    = t->getSocket();
  913.       if(is_connected(nodeId)){
  914. if(t->isConnected() && FD_ISSET(socket, &tcpReadset)) 
  915. {
  916.   const int receiveSize = t->doReceive();
  917.   if(receiveSize > 0)
  918.   {
  919.     Uint32 * ptr;
  920.     Uint32 sz = t->getReceiveData(&ptr);
  921.     Uint32 szUsed = unpack(ptr, sz, nodeId, ioStates[nodeId]);
  922.     t->updateReceiveDataPtr(szUsed);
  923.           }
  924. }
  925.       } 
  926.     }
  927.   }
  928. #endif
  929.   
  930. #ifdef NDB_SCI_TRANSPORTER
  931.   //performReceive
  932.   //do prepareReceive on the SCI transporters  (prepareReceive(t,,,,))
  933.   for (int i=0; i<nSCITransporters; i++) 
  934.   {
  935.     checkJobBuffer();
  936.     SCI_Transporter  *t = theSCITransporters[i];
  937.     const NodeId nodeId = t->getRemoteNodeId();
  938.     if(is_connected(nodeId))
  939.     {
  940.       if(t->isConnected() && t->checkConnected())
  941.       {
  942. Uint32 * readPtr, * eodPtr;
  943. t->getReceivePtr(&readPtr, &eodPtr);
  944. Uint32 *newPtr = unpack(readPtr, eodPtr, nodeId, ioStates[nodeId]);
  945. t->updateReceivePtr(newPtr);
  946.       }
  947.     } 
  948.   }
  949. #endif
  950. #ifdef NDB_SHM_TRANSPORTER
  951.   for (int i=0; i<nSHMTransporters; i++) 
  952.   {
  953.     checkJobBuffer();
  954.     SHM_Transporter *t = theSHMTransporters[i];
  955.     const NodeId nodeId = t->getRemoteNodeId();
  956.     if(is_connected(nodeId)){
  957.       if(t->isConnected() && t->checkConnected())
  958.       {
  959. Uint32 * readPtr, * eodPtr;
  960. t->getReceivePtr(&readPtr, &eodPtr);
  961. Uint32 *newPtr = unpack(readPtr, eodPtr, nodeId, ioStates[nodeId]);
  962. t->updateReceivePtr(newPtr);
  963.       }
  964.     } 
  965.   }
  966. #endif
  967. }
  969. static int x = 0;
  970. void
  971. TransporterRegistry::performSend()
  972. {
  973.   int i; 
  974.   sendCounter = 1;
  975.   
  976. #ifdef NDB_OSE_TRANSPORTER
  977.   for (int i = 0; i < nOSETransporters; i++)
  978.   {
  979.     OSE_Transporter *t = theOSETransporters[i]; 
  980.     if(is_connected(t->getRemoteNodeId()) &&& (t->isConnected()))
  981.     {
  982.       t->doSend();
  983.     }//if
  984.   }//for
  985. #endif
  986.   
  987. #ifdef NDB_TCP_TRANSPORTER
  988. #ifdef NDB_OSE
  989.   {
  990.     int maxSocketValue = 0;
  991.     
  992.     // Needed for TCP/IP connections
  993.     // The writeset are used by select
  994.     fd_set writeset;
  995.     FD_ZERO(&writeset);
  996.     
  997.     // Prepare for sending and receiving
  998.     for (i = 0; i < nTCPTransporters; i++) {
  999.       TCP_Transporter * t = theTCPTransporters[i];
  1000.       
  1001.       // If the transporter is connected
  1002.       if ((t->hasDataToSend()) && (t->isConnected())) {
  1003. const int socket = t->getSocket();
  1004. // Find the highest socket value. It will be used by select
  1005. if (socket > maxSocketValue) {
  1006.   maxSocketValue = socket;
  1007. }//if
  1008. FD_SET(socket, &writeset);
  1009.       }//if
  1010.     }//for
  1011.     
  1012.     // The highest socket value plus one
  1013.     if(maxSocketValue == 0)
  1014.       return;
  1015.     
  1016.     maxSocketValue++; 
  1017.     struct timeval timeout = { 0, 1025 };
  1018.     Uint32 tmp = select(maxSocketValue, 0, &writeset, 0, &timeout);
  1019.     
  1020.     if (tmp == 0) 
  1021.     {
  1022.       return;
  1023.     }//if
  1024.     for (i = 0; i < nTCPTransporters; i++) {
  1025.       TCP_Transporter *t = theTCPTransporters[i];
  1026.       const NodeId nodeId = t->getRemoteNodeId();
  1027.       const int socket    = t->getSocket();
  1028.       if(is_connected(nodeId)){
  1029.   if(t->isConnected() && FD_ISSET(socket, &writeset)) {
  1030.     t->doSend();
  1031.   }//if
  1032. }//if
  1033.       }//for
  1034.     }
  1035. #endif
  1036. #ifdef NDB_TCP_TRANSPORTER
  1037.   for (i = x; i < nTCPTransporters; i++) 
  1038.   {
  1039.     TCP_Transporter *t = theTCPTransporters[i];
  1040.     if (t && t->hasDataToSend() && t->isConnected() &&
  1041. is_connected(t->getRemoteNodeId())) 
  1042.     {
  1043.       t->doSend();
  1044.     }
  1045.   }
  1046.   for (i = 0; i < x && i < nTCPTransporters; i++) 
  1047.   {
  1048.     TCP_Transporter *t = theTCPTransporters[i];
  1049.     if (t && t->hasDataToSend() && t->isConnected() &&
  1050. is_connected(t->getRemoteNodeId())) 
  1051.     {
  1052.       t->doSend();
  1053.     }
  1054.   }
  1055.   x++;
  1056.   if (x == nTCPTransporters) x = 0;
  1057. #endif
  1058. #endif
  1059. #ifdef NDB_SCI_TRANSPORTER
  1060.   //scroll through the SCI transporters, 
  1061.   // get each transporter, check if connected, send data
  1062.   for (i=0; i<nSCITransporters; i++) {
  1063.     SCI_Transporter  *t = theSCITransporters[i];
  1064.     const NodeId nodeId = t->getRemoteNodeId();
  1065.     
  1066.     if(is_connected(nodeId))
  1067.     {
  1068.       if(t->isConnected() && t->hasDataToSend()) {
  1069. t->doSend();
  1070.       } //if
  1071.     } //if
  1072.   }
  1073. #endif
  1074.   
  1075. #ifdef NDB_SHM_TRANSPORTER
  1076.   for (i=0; i<nSHMTransporters; i++) 
  1077.   {
  1078.     SHM_Transporter  *t = theSHMTransporters[i];
  1079.     const NodeId nodeId = t->getRemoteNodeId();
  1080.     if(is_connected(nodeId))
  1081.     {
  1082.       if(t->isConnected())
  1083.       {
  1084. t->doSend();
  1085.       }
  1086.     }
  1087.   }
  1088. #endif
  1089. }
  1091. int
  1092. TransporterRegistry::forceSendCheck(int sendLimit){
  1093.   int tSendCounter = sendCounter;
  1094.   sendCounter = tSendCounter + 1;
  1095.   if (tSendCounter >= sendLimit) {
  1096.     performSend();
  1097.     sendCounter = 1;
  1098.     return 1;
  1099.   }//if
  1100.   return 0;
  1101. }//TransporterRegistry::forceSendCheck()
  1103. #ifdef DEBUG_TRANSPORTER
  1104. void
  1105. TransporterRegistry::printState(){
  1106.   ndbout << "-- TransporterRegistry -- " << endl << endl
  1107.  << "Transporters = " << nTransporters << endl;
  1108.   for(int i = 0; i<maxTransporters; i++)
  1109.     if(theTransporters[i] != NULL){
  1110.       const NodeId remoteNodeId = theTransporters[i]->getRemoteNodeId();
  1111.       ndbout << "Transporter: " << remoteNodeId 
  1112.      << " PerformState: " << performStates[remoteNodeId]
  1113.      << " IOState: " << ioStates[remoteNodeId] << endl;
  1114.     }
  1115. }
  1116. #endif
  1118. IOState
  1119. TransporterRegistry::ioState(NodeId nodeId) { 
  1120.   return ioStates[nodeId]; 
  1121. }
  1123. void
  1124. TransporterRegistry::setIOState(NodeId nodeId, IOState state) {
  1125.   DEBUG("TransporterRegistry::setIOState("
  1126. << nodeId << ", " << state << ")");
  1127.   ioStates[nodeId] = state;
  1128. }
  1130. static void * 
  1131. run_start_clients_C(void * me)
  1132. {
  1133.   ((TransporterRegistry*) me)->start_clients_thread();
  1134.   return 0;
  1135. }
  1137. // Run by kernel thread
  1138. void
  1139. TransporterRegistry::do_connect(NodeId node_id)
  1140. {
  1141.   PerformState &curr_state = performStates[node_id];
  1142.   switch(curr_state){
  1143.   case DISCONNECTED:
  1144.     break;
  1145.   case CONNECTED:
  1146.     return;
  1147.   case CONNECTING:
  1148.     return;
  1149.   case DISCONNECTING:
  1150.     break;
  1151.   }
  1152.   curr_state= CONNECTING;
  1153. }
  1154. void
  1155. TransporterRegistry::do_disconnect(NodeId node_id)
  1156. {
  1157.   PerformState &curr_state = performStates[node_id];
  1158.   switch(curr_state){
  1159.   case DISCONNECTED:
  1160.     return;
  1161.   case CONNECTED:
  1162.     break;
  1163.   case CONNECTING:
  1164.     break;
  1165.   case DISCONNECTING:
  1166.     return;
  1167.   }
  1168.   curr_state= DISCONNECTING;
  1169. }
  1171. void
  1172. TransporterRegistry::report_connect(NodeId node_id)
  1173. {
  1174.   performStates[node_id] = CONNECTED;
  1175.   reportConnect(callbackObj, node_id);
  1176. }
  1178. void
  1179. TransporterRegistry::report_disconnect(NodeId node_id, int errnum)
  1180. {
  1181.   performStates[node_id] = DISCONNECTED;
  1182.   reportDisconnect(callbackObj, node_id, errnum);
  1183. }
  1185. void
  1186. TransporterRegistry::update_connections()
  1187. {
  1188.   for (int i= 0, n= 0; n < nTransporters; i++){
  1189.     Transporter * t = theTransporters[i];
  1190.     if (!t)
  1191.       continue;
  1192.     n++;
  1194.     const NodeId nodeId = t->getRemoteNodeId();
  1195.     switch(performStates[nodeId]){
  1196.     case CONNECTED:
  1197.     case DISCONNECTED:
  1198.       break;
  1199.     case CONNECTING:
  1200.       if(t->isConnected())
  1201. report_connect(nodeId);
  1202.       break;
  1203.     case DISCONNECTING:
  1204.       if(!t->isConnected())
  1205. report_disconnect(nodeId, 0);
  1206.       break;
  1207.     }
  1208.   }
  1209. }
  1211. // run as own thread
  1212. void
  1213. TransporterRegistry::start_clients_thread()
  1214. {
  1215.   DBUG_ENTER("TransporterRegistry::start_clients_thread");
  1216.   while (m_run_start_clients_thread) {
  1217.     NdbSleep_MilliSleep(100);
  1218.     for (int i= 0, n= 0; n < nTransporters && m_run_start_clients_thread; i++){
  1219.       Transporter * t = theTransporters[i];
  1220.       if (!t)
  1221. continue;
  1222.       n++;
  1224.       const NodeId nodeId = t->getRemoteNodeId();
  1225.       switch(performStates[nodeId]){
  1226.       case CONNECTING:
  1227. if(!t->isConnected() && !t->isServer)
  1228.     t->connect_client();
  1229. break;
  1230.       case DISCONNECTING:
  1231. if(t->isConnected())
  1232.   t->doDisconnect();
  1233. break;
  1234.       default:
  1235. break;
  1236.       }
  1237.     }
  1238.   }
  1239.   DBUG_VOID_RETURN;
  1240. }
  1242. bool
  1243. TransporterRegistry::start_clients()
  1244. {
  1245.   m_run_start_clients_thread= true;
  1246.   m_start_clients_thread= NdbThread_Create(run_start_clients_C,
  1247.    (void**)this,
  1248.    32768,
  1249.    "ndb_start_clients",
  1250.    NDB_THREAD_PRIO_LOW);
  1251.   if (m_start_clients_thread == 0) {
  1252.     m_run_start_clients_thread= false;
  1253.     return false;
  1254.   }
  1255.   return true;
  1256. }
  1258. bool
  1259. TransporterRegistry::stop_clients()
  1260. {
  1261.   if (m_start_clients_thread) {
  1262.     m_run_start_clients_thread= false;
  1263.     void* status;
  1264.     int r= NdbThread_WaitFor(m_start_clients_thread, &status);
  1265.     NdbThread_Destroy(&m_start_clients_thread);
  1266.   }
  1267.   return true;
  1268. }
  1270. void
  1271. TransporterRegistry::add_transporter_interface(const char *interf, unsigned short port)
  1272. {
  1273.   DBUG_ENTER("TransporterRegistry::add_transporter_interface");
  1274.   DBUG_PRINT("enter",("interface=%s, port= %d", interf, port));
  1275.   if (interf && strlen(interf) == 0)
  1276.     interf= 0;
  1278.   for (unsigned i= 0; i < m_transporter_interface.size(); i++)
  1279.   {
  1280.     Transporter_interface &tmp= m_transporter_interface[i];
  1281.     if (port != tmp.m_service_port)
  1282.       continue;
  1283.     if (interf != 0 && tmp.m_interface != 0 &&
  1284. strcmp(interf, tmp.m_interface) == 0)
  1285.     {
  1286.       DBUG_VOID_RETURN; // found match, no need to insert
  1287.     }
  1288.     if (interf == 0 && tmp.m_interface == 0)
  1289.     {
  1290.       DBUG_VOID_RETURN; // found match, no need to insert
  1291.     }
  1292.   }
  1293.   Transporter_interface t;
  1294.   t.m_service_port= port;
  1295.   t.m_interface= interf;
  1296.   m_transporter_interface.push_back(t);
  1297.   DBUG_PRINT("exit",("interface and port added"));
  1298.   DBUG_VOID_RETURN;
  1299. }
  1301. bool
  1302. TransporterRegistry::start_service(SocketServer& socket_server)
  1303. {
  1304.   if (m_transporter_interface.size() > 0 && !nodeIdSpecified)
  1305.   {
  1306.     ndbout_c("TransporterRegistry::startReceiving: localNodeId not specified");
  1307.     return false;
  1308.   }
  1310.   for (unsigned i= 0; i < m_transporter_interface.size(); i++)
  1311.   {
  1312.     Transporter_interface &t= m_transporter_interface[i];
  1313.     if (t.m_service_port == 0)
  1314.     {
  1315.       continue;
  1316.     }
  1317.     TransporterService *transporter_service =
  1318.       new TransporterService(new SocketAuthSimple("ndbd", "ndbd passwd"));
  1319.     if(!socket_server.setup(transporter_service,
  1320.     t.m_service_port, t.m_interface))
  1321.     {
  1322.       ndbout_c("Unable to setup transporter service port: %s:%d!n"
  1323.        "Please check if the port is already used,n"
  1324.        "(perhaps the node is already running)",
  1325.        t.m_interface ? t.m_interface : "*", t.m_service_port);
  1326.       delete transporter_service;
  1327.       return false;
  1328.     }
  1329.     transporter_service->setTransporterRegistry(this);
  1330.   }
  1331.   return true;
  1332. }
  1334. #ifdef NDB_SHM_TRANSPORTER
  1335. static
  1336. RETSIGTYPE 
  1337. shm_sig_handler(int signo)
  1338. {
  1339.   g_shm_counter++;
  1340. }
  1341. #endif
  1343. void
  1344. TransporterRegistry::startReceiving()
  1345. {
  1346.   DBUG_ENTER("TransporterRegistry::startReceiving");
  1347. #ifdef NDB_OSE_TRANSPORTER
  1348.   if(theOSEReceiver != NULL){
  1349.     theOSEReceiver->createPhantom();
  1350.   }
  1351. #endif
  1353. #ifdef NDB_OSE
  1354.   theOSEJunkSocketRecv = socket(AF_INET, SOCK_STREAM, 0);
  1355. #endif
  1357. #if defined NDB_OSE || defined NDB_SOFTOSE
  1358.   theReceiverPid = current_process();
  1359.   for(int i = 0; i<nTCPTransporters; i++)
  1360.     theTCPTransporters[i]->theReceiverPid = theReceiverPid;
  1361. #endif
  1363. #ifdef NDB_SHM_TRANSPORTER
  1364.   m_shm_own_pid = getpid();
  1365.   if (g_ndb_shm_signum)
  1366.   {
  1367.     DBUG_PRINT("info",("Install signal handler for signum %d",
  1368.        g_ndb_shm_signum));
  1369.     struct sigaction sa;
  1370.     sigemptyset(&sa.sa_mask);
  1371.     sigaddset(&sa.sa_mask, g_ndb_shm_signum);
  1372.     pthread_sigmask(SIG_UNBLOCK, &sa.sa_mask, 0);
  1373.     sa.sa_handler = shm_sig_handler;
  1374.     sigemptyset(&sa.sa_mask);
  1375.     sa.sa_flags = 0;
  1376.     int ret;
  1377.     while((ret = sigaction(g_ndb_shm_signum, &sa, 0)) == -1 && errno == EINTR);
  1378.     if(ret != 0)
  1379.     {
  1380.       DBUG_PRINT("error",("Install failed"));
  1381.       g_eventLogger.error("Failed to install signal handler for"
  1382.   " SHM transporter errno: %d (%s)", errno, 
  1383. #ifdef HAVE_STRERROR
  1384.   strerror(errno)
  1385. #else
  1386.                           ""
  1387. #endif
  1388.   );
  1389.     }
  1390.   }
  1391. #endif // NDB_SHM_TRANSPORTER
  1392.   DBUG_VOID_RETURN;
  1393. }
  1395. void
  1396. TransporterRegistry::stopReceiving(){
  1397. #ifdef NDB_OSE_TRANSPORTER
  1398.   if(theOSEReceiver != NULL){
  1399.     theOSEReceiver->destroyPhantom();
  1400.   }
  1401. #endif
  1403.   /**
  1404.    * Disconnect all transporters, this includes detach from remote node
  1405.    * and since that must be done from the same process that called attach
  1406.    * it's done here in the receive thread
  1407.    */
  1408.   disconnectAll();
  1410. #if defined NDB_OSE || defined NDB_SOFTOSE
  1411.   if(theOSEJunkSocketRecv > 0)
  1412.     close(theOSEJunkSocketRecv);
  1413.   theOSEJunkSocketRecv = -1;
  1414. #endif
  1416. }
  1418. void
  1419. TransporterRegistry::startSending(){
  1420. #if defined NDB_OSE || defined NDB_SOFTOSE
  1421.   theOSEJunkSocketSend = socket(AF_INET, SOCK_STREAM, 0);
  1422. #endif
  1423. }
  1425. void
  1426. TransporterRegistry::stopSending(){
  1427. #if defined NDB_OSE || defined NDB_SOFTOSE
  1428.   if(theOSEJunkSocketSend > 0)
  1429.     close(theOSEJunkSocketSend);
  1430.   theOSEJunkSocketSend = -1;
  1431. #endif
  1432. }
  1434. NdbOut & operator <<(NdbOut & out, SignalHeader & sh){
  1435.   out << "-- Signal Header --" << endl;
  1436.   out << "theLength:    " << sh.theLength << endl;
  1437.   out << "gsn:          " << sh.theVerId_signalNumber << endl;
  1438.   out << "recBlockNo:   " << sh.theReceiversBlockNumber << endl;
  1439.   out << "sendBlockRef: " << sh.theSendersBlockRef << endl;
  1440.   out << "sendersSig:   " << sh.theSendersSignalId << endl;
  1441.   out << "theSignalId:  " << sh.theSignalId << endl;
  1442.   out << "trace:        " << (int)sh.theTrace << endl;
  1443.   return out;
  1444. } 
  1446. template class Vector<TransporterRegistry::Transporter_interface>;