From: Date: September 8 2005 11:35am Subject: bk commit into 4.1 tree (tomas:1.2410) BUG#13054 List-Archive: http://lists.mysql.com/internals/29479 X-Bug: 13054 Message-Id: <20050908093503.36CC3807E8@poseidon.mysql.com> Below is the list of changes that have just been committed into a local 4.1 repository of tomas. When tomas does a push these changes will be propagated to the main repository and, within 24 hours after the push, to the public repository. For information on how to access the public repository see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html ChangeSet 1.2410 05/09/08 11:34:55 tomas@stripped +11 -0 Bug #13054 start backup command return may be interrpted ndb/src/ndbapi/SignalSender.hpp 1.1 05/09/08 11:34:51 tomas@stripped +83 -0 ndb/src/ndbapi/SignalSender.cpp 1.1 05/09/08 11:34:51 tomas@stripped +270 -0 ndb/src/ndbapi/SignalSender.hpp 1.0 05/09/08 11:34:51 tomas@stripped +0 -0 BitKeeper file /home/tomas/mysql-4.1/ndb/src/ndbapi/SignalSender.hpp ndb/src/ndbapi/SignalSender.cpp 1.0 05/09/08 11:34:51 tomas@stripped +0 -0 BitKeeper file /home/tomas/mysql-4.1/ndb/src/ndbapi/SignalSender.cpp ndb/src/ndbapi/Makefile.am 1.11 05/09/08 11:34:51 tomas@stripped +2 -1 Bug #13054 start backup command return may be interrpted ndb/src/mgmsrv/MgmtSrvr.hpp 1.28 05/09/08 11:34:51 tomas@stripped +0 -3 Bug #13054 start backup command return may be interrpted ndb/src/mgmsrv/MgmtSrvr.cpp 1.64 05/09/08 11:34:51 tomas@stripped +126 -139 Bug #13054 start backup command return may be interrpted ndb/src/kernel/blocks/backup/Backup.hpp 1.8 05/09/08 11:34:51 tomas@stripped +1 -0 Bug #13054 start backup command return may be interrpted ndb/src/kernel/blocks/backup/Backup.cpp 1.19 05/09/08 11:34:51 tomas@stripped +31 -11 Bug #13054 start backup command return may be interrpted ndb/include/kernel/signaldata/NodeFailRep.hpp 1.2 05/09/08 11:34:51 tomas@stripped +1 -27 Bug #13054 start backup command return may be interrpted ndb/include/kernel/signaldata/NFCompleteRep.hpp 1.2 05/09/08 11:34:51 tomas@stripped +1 -17 Bug #13054 start backup command return may be interrpted ndb/include/kernel/signaldata/BackupSignalData.hpp 1.4 05/09/08 11:34:51 tomas@stripped +4 -1 Bug #13054 start backup command return may be interrpted ndb/include/kernel/signaldata/BackupImpl.hpp 1.4 05/09/08 11:34:51 tomas@stripped +8 -1 Bug #13054 start backup command return may be interrpted # This is a BitKeeper patch. What follows are the unified diffs for the # set of deltas contained in the patch. The rest of the patch, the part # that BitKeeper cares about, is below these diffs. # User: tomas # Host: poseidon.ndb.mysql.com # Root: /home/tomas/mysql-4.1 --- 1.3/ndb/include/kernel/signaldata/BackupImpl.hpp 2005-04-22 09:07:22 +02:00 +++ 1.4/ndb/include/kernel/signaldata/BackupImpl.hpp 2005-09-08 11:34:51 +02:00 @@ -33,7 +33,7 @@ friend bool printDEFINE_BACKUP_REQ(FILE *, const Uint32 *, Uint32, Uint16); public: - STATIC_CONST( SignalLength = 8 + NdbNodeBitmask::Size); + STATIC_CONST( SignalLength = 9 + NdbNodeBitmask::Size); private: /** @@ -60,6 +60,13 @@ * Length of backup data */ Uint32 backupDataLen; + + /** + * Backup flags + */ + /* & 0x3 - waitCompleted + */ + Uint32 flags; }; class DefineBackupRef { --- 1.3/ndb/include/kernel/signaldata/BackupSignalData.hpp 2005-04-22 09:07:22 +02:00 +++ 1.4/ndb/include/kernel/signaldata/BackupSignalData.hpp 2005-09-08 11:34:51 +02:00 @@ -36,11 +36,14 @@ friend bool printBACKUP_REQ(FILE *, const Uint32 *, Uint32, Uint16); public: - STATIC_CONST( SignalLength = 2 ); + STATIC_CONST( SignalLength = 3 ); private: Uint32 senderData; Uint32 backupDataLen; + /* & 0x3 - waitCompleted + */ + Uint32 flags; }; class BackupData { --- 1.1/ndb/include/kernel/signaldata/NFCompleteRep.hpp 2004-04-14 10:23:55 +02:00 +++ 1.2/ndb/include/kernel/signaldata/NFCompleteRep.hpp 2005-09-08 11:34:51 +02:00 @@ -30,28 +30,12 @@ * from the failed NDB node * */ -class NFCompleteRep { - /** - * Sender(s) - */ - friend class Dbdict; - friend class Dblqh; - friend class Dbtc; - friend class Qmgr; - - /** - * Sender/Reciver - */ - friend class Dbdih; - friend class ClusterMgr; +struct NFCompleteRep { friend bool printNF_COMPLETE_REP(FILE *, const Uint32 *, Uint32, Uint16); -public: STATIC_CONST( SignalLength = 5 ); -private: - /** * Which block has completed... * --- 1.1/ndb/include/kernel/signaldata/NodeFailRep.hpp 2004-04-14 10:23:55 +02:00 +++ 1.2/ndb/include/kernel/signaldata/NodeFailRep.hpp 2005-09-08 11:34:51 +02:00 @@ -24,34 +24,8 @@ * This signals is sent by Qmgr to NdbCntr * and then from NdbCntr sent to: dih, dict, lqh, tc & API */ -class NodeFailRep { - /** - * Sender(s) - */ - friend class Qmgr; - - /** - * Sender(s) / Reciver(s) - */ - friend class Ndbcntr; - friend class Dbdict; - - /** - * Reciver(s) - */ - friend class Dbdih; - friend class Dblqh; - friend class Dbtc; - friend class ClusterMgr; - friend class Trix; - friend class Backup; - friend class Suma; - friend class Grep; - friend class SafeCounterManager; - -public: +struct NodeFailRep { STATIC_CONST( SignalLength = 3 + NodeBitmask::Size ); -private: Uint32 failNo; --- 1.18/ndb/src/kernel/blocks/backup/Backup.cpp 2005-07-20 11:48:45 +02:00 +++ 1.19/ndb/src/kernel/blocks/backup/Backup.cpp 2005-09-08 11:34:51 +02:00 @@ -69,6 +69,9 @@ static Uint32 g_TypeOfStart = NodeState::ST_ILLEGAL_TYPE; +#define SEND_BACKUP_STARTED_FLAG(A) (((A) & 0x3) > 0) +#define SEND_BACKUP_COMPLETED_FLAG(A) (((A) & 0x3) > 1) + void Backup::execSTTOR(Signal* signal) { @@ -852,7 +855,8 @@ const Uint32 senderData = req->senderData; const BlockReference senderRef = signal->senderBlockRef(); const Uint32 dataLen32 = req->backupDataLen; // In 32 bit words - + const Uint32 flags = signal->getLength() > 2 ? req->flags : 2; + if(getOwnNodeId() != getMasterNodeId()) { jam(); sendBackupRef(senderRef, signal, senderData, BackupRef::IAmNotMaster); @@ -894,6 +898,7 @@ ptr.p->errorCode = 0; ptr.p->clientRef = senderRef; ptr.p->clientData = senderData; + ptr.p->flags = flags; ptr.p->masterRef = reference(); ptr.p->nodes = c_aliveNodes; ptr.p->backupId = 0; @@ -931,7 +936,10 @@ Backup::sendBackupRef(Signal* signal, BackupRecordPtr ptr, Uint32 errorCode) { jam(); - sendBackupRef(ptr.p->clientRef, signal, ptr.p->clientData, errorCode); + if (SEND_BACKUP_STARTED_FLAG(ptr.p->flags)) + { + sendBackupRef(ptr.p->clientRef, signal, ptr.p->clientData, errorCode); + } cleanup(signal, ptr); } @@ -1098,6 +1106,7 @@ req->backupKey[1] = ptr.p->backupKey[1]; req->nodes = ptr.p->nodes; req->backupDataLen = ptr.p->backupDataLen; + req->flags = ptr.p->flags; ptr.p->masterData.gsn = GSN_DEFINE_BACKUP_REQ; ptr.p->masterData.sendCounter = ptr.p->nodes; @@ -1193,13 +1202,18 @@ /** * Reply to client */ + CRASH_INSERTION((10034)); + BackupConf * conf = (BackupConf*)signal->getDataPtrSend(); conf->backupId = ptr.p->backupId; conf->senderData = ptr.p->clientData; conf->nodes = ptr.p->nodes; - sendSignal(ptr.p->clientRef, GSN_BACKUP_CONF, signal, - BackupConf::SignalLength, JBB); - + if (SEND_BACKUP_STARTED_FLAG(ptr.p->flags)) + { + sendSignal(ptr.p->clientRef, GSN_BACKUP_CONF, signal, + BackupConf::SignalLength, JBB); + } + signal->theData[0] = EventReport::BackupStarted; signal->theData[1] = ptr.p->clientRef; signal->theData[2] = ptr.p->backupId; @@ -2090,9 +2104,12 @@ rep->noOfLogBytes = ptr.p->noOfLogBytes; rep->noOfLogRecords = ptr.p->noOfLogRecords; rep->nodes = ptr.p->nodes; - sendSignal(ptr.p->clientRef, GSN_BACKUP_COMPLETE_REP, signal, - BackupCompleteRep::SignalLength, JBB); - + if (SEND_BACKUP_COMPLETED_FLAG(ptr.p->flags)) + { + sendSignal(ptr.p->clientRef, GSN_BACKUP_COMPLETE_REP, signal, + BackupCompleteRep::SignalLength, JBB); + } + signal->theData[0] = EventReport::BackupCompleted; signal->theData[1] = ptr.p->clientRef; signal->theData[2] = ptr.p->backupId; @@ -2133,9 +2150,11 @@ rep->backupId = ptr.p->backupId; rep->senderData = ptr.p->clientData; rep->reason = ptr.p->errorCode; - sendSignal(ptr.p->clientRef, GSN_BACKUP_ABORT_REP, signal, - BackupAbortRep::SignalLength, JBB); - + if (SEND_BACKUP_COMPLETED_FLAG(ptr.p->flags)) + { + sendSignal(ptr.p->clientRef, GSN_BACKUP_ABORT_REP, signal, + BackupAbortRep::SignalLength, JBB); + } signal->theData[0] = EventReport::BackupAborted; signal->theData[1] = ptr.p->clientRef; signal->theData[2] = ptr.p->backupId; @@ -2267,6 +2286,7 @@ ptr.p->errorCode = 0; ptr.p->clientRef = req->clientRef; ptr.p->clientData = req->clientData; + ptr.p->flags = req->flags; ptr.p->masterRef = senderRef; ptr.p->nodes = req->nodes; ptr.p->backupId = backupId; --- 1.7/ndb/src/kernel/blocks/backup/Backup.hpp 2005-04-22 09:07:22 +02:00 +++ 1.8/ndb/src/kernel/blocks/backup/Backup.hpp 2005-09-08 11:34:51 +02:00 @@ -413,6 +413,7 @@ Uint32 clientRef; Uint32 clientData; + Uint32 flags; Uint32 backupId; Uint32 backupKey[2]; Uint32 masterRef; --- 1.63/ndb/src/mgmsrv/MgmtSrvr.cpp 2005-09-05 16:33:37 +02:00 +++ 1.64/ndb/src/mgmsrv/MgmtSrvr.cpp 2005-09-08 11:34:51 +02:00 @@ -39,6 +39,8 @@ #include #include #include +#include +#include #include #include #include @@ -56,6 +58,8 @@ #include #include +#include + //#define MGM_SRV_DEBUG #ifdef MGM_SRV_DEBUG #define DEBUG(x) do ndbout << x << endl; while(0) @@ -709,6 +713,15 @@ } } +void report_unknown_signal(SimpleSignal *signal) +{ + g_eventLogger.error("Unknown signal received. SignalNumber: " + "%i from (%d, %x)", + signal->readSignalNumber(), + refToNode(signal->header.theSendersBlockRef), + refToBlock(signal->header.theSendersBlockRef)); +} + /***************************************************************************** * Starting and stopping database nodes ****************************************************************************/ @@ -1911,81 +1924,6 @@ } break; - case GSN_BACKUP_CONF:{ - const BackupConf * const conf = - CAST_CONSTPTR(BackupConf, signal->getDataPtr()); - BackupEvent event; - event.Event = BackupEvent::BackupStarted; - event.Started.BackupId = conf->backupId; - event.Nodes = conf->nodes; -#ifdef VM_TRACE - ndbout_c("Backup master is %d", refToNode(signal->theSendersBlockRef)); -#endif - backupCallback(event); - } - break; - - case GSN_BACKUP_REF:{ - const BackupRef * const ref = - CAST_CONSTPTR(BackupRef, signal->getDataPtr()); - Uint32 errCode = ref->errorCode; - if(ref->errorCode == BackupRef::IAmNotMaster){ - const Uint32 aNodeId = refToNode(ref->masterRef); -#ifdef VM_TRACE - ndbout_c("I'm not master resending to %d", aNodeId); -#endif - theWaitNode= aNodeId; - NdbApiSignal aSignal(_ownReference); - BackupReq* req = CAST_PTR(BackupReq, aSignal.getDataPtrSend()); - aSignal.set(TestOrd::TraceAPI, BACKUP, GSN_BACKUP_REQ, - BackupReq::SignalLength); - req->senderData = 19; - req->backupDataLen = 0; - - int i = theFacade->sendSignalUnCond(&aSignal, aNodeId); - if(i == 0){ - return; - } - errCode = 5030; - } - BackupEvent event; - event.Event = BackupEvent::BackupFailedToStart; - event.FailedToStart.ErrorCode = errCode; - backupCallback(event); - break; - } - - case GSN_BACKUP_ABORT_REP:{ - const BackupAbortRep * const rep = - CAST_CONSTPTR(BackupAbortRep, signal->getDataPtr()); - BackupEvent event; - event.Event = BackupEvent::BackupAborted; - event.Aborted.Reason = rep->reason; - event.Aborted.BackupId = rep->backupId; - event.Aborted.ErrorCode = rep->reason; - backupCallback(event); - } - break; - - case GSN_BACKUP_COMPLETE_REP:{ - const BackupCompleteRep * const rep = - CAST_CONSTPTR(BackupCompleteRep, signal->getDataPtr()); - BackupEvent event; - event.Event = BackupEvent::BackupCompleted; - event.Completed.BackupId = rep->backupId; - - event.Completed.NoOfBytes = rep->noOfBytes; - event.Completed.NoOfLogBytes = rep->noOfLogBytes; - event.Completed.NoOfRecords = rep->noOfRecords; - event.Completed.NoOfLogRecords = rep->noOfLogRecords; - event.Completed.stopGCP = rep->stopGCP; - event.Completed.startGCP = rep->startGCP; - event.Nodes = rep->nodes; - - backupCallback(event); - } - break; - case GSN_MGM_LOCK_CONFIG_REP: case GSN_MGM_LOCK_CONFIG_REQ: case GSN_MGM_UNLOCK_CONFIG_REP: @@ -2446,6 +2384,9 @@ int MgmtSrvr::startBackup(Uint32& backupId, int waitCompleted) { + SignalSender ss(theFacade); + ss.lock(); // lock will be released on exit + bool next; NodeId nodeId = 0; while((next = getNextNodeId(&nodeId, NDB_MGM_NODE_TYPE_NDB)) == true && @@ -2453,49 +2394,125 @@ if(!next) return NO_CONTACT_WITH_DB_NODES; - NdbApiSignal* signal = getSignal(); - if (signal == NULL) { - return COULD_NOT_ALLOCATE_MEMORY; - } + SimpleSignal ssig; - BackupReq* req = CAST_PTR(BackupReq, signal->getDataPtrSend()); - signal->set(TestOrd::TraceAPI, BACKUP, GSN_BACKUP_REQ, - BackupReq::SignalLength); + BackupReq* req = CAST_PTR(BackupReq, ssig.getDataPtrSend()); + ssig.set(ss, TestOrd::TraceAPI, BACKUP, GSN_BACKUP_REQ, + BackupReq::SignalLength); req->senderData = 19; req->backupDataLen = 0; + assert(waitCompleted < 3); + req->flags = waitCompleted & 0x3; - int result; - if (waitCompleted == 2) { - result = sendRecSignal(nodeId, WAIT_BACKUP_COMPLETED, - signal, true, 48*60*60*1000 /* 48 hours */); - } - else if (waitCompleted == 1) { - result = sendRecSignal(nodeId, WAIT_BACKUP_STARTED, - signal, true, 5*60*1000 /*5 mins*/); - } - else { - result = sendRecSignal(nodeId, NO_WAIT, signal, true); - } - if (result == -1) { - return SEND_OR_RECEIVE_FAILED; - } + BackupEvent event; + int do_send = 1; + while (1) { + if (do_send) + { + SendStatus result = ss.sendSignal(nodeId, &ssig); + if (result != SEND_OK) { + return SEND_OR_RECEIVE_FAILED; + } + if (waitCompleted == 0) + return 0; + do_send = 0; + } + SimpleSignal *signal = ss.waitFor(); + + int gsn = signal->readSignalNumber(); + switch (gsn) { + case GSN_BACKUP_CONF:{ + const BackupConf * const conf = + CAST_CONSTPTR(BackupConf, signal->getDataPtr()); + event.Event = BackupEvent::BackupStarted; + event.Started.BackupId = conf->backupId; + event.Nodes = conf->nodes; +#ifdef VM_TRACE + ndbout_c("Backup(%d) master is %d", conf->backupId, + refToNode(signal->header.theSendersBlockRef)); +#endif + backupId = conf->backupId; + if (waitCompleted == 1) + return 0; + // wait for next signal + break; + } + case GSN_BACKUP_COMPLETE_REP:{ + const BackupCompleteRep * const rep = + CAST_CONSTPTR(BackupCompleteRep, signal->getDataPtr()); +#ifdef VM_TRACE + ndbout_c("Backup(%d) completed %d", rep->backupId); +#endif + event.Event = BackupEvent::BackupCompleted; + event.Completed.BackupId = rep->backupId; + + event.Completed.NoOfBytes = rep->noOfBytes; + event.Completed.NoOfLogBytes = rep->noOfLogBytes; + event.Completed.NoOfRecords = rep->noOfRecords; + event.Completed.NoOfLogRecords = rep->noOfLogRecords; + event.Completed.stopGCP = rep->stopGCP; + event.Completed.startGCP = rep->startGCP; + event.Nodes = rep->nodes; - if (waitCompleted){ - switch(m_lastBackupEvent.Event){ - case BackupEvent::BackupCompleted: - backupId = m_lastBackupEvent.Completed.BackupId; + backupId = rep->backupId; + return 0; + } + case GSN_BACKUP_REF:{ + const BackupRef * const ref = + CAST_CONSTPTR(BackupRef, signal->getDataPtr()); + if(ref->errorCode == BackupRef::IAmNotMaster){ + nodeId = refToNode(ref->masterRef); +#ifdef VM_TRACE + ndbout_c("I'm not master resending to %d", nodeId); +#endif + do_send = 1; // try again + continue; + } + event.Event = BackupEvent::BackupFailedToStart; + event.FailedToStart.ErrorCode = ref->errorCode; + return ref->errorCode; + } + case GSN_BACKUP_ABORT_REP:{ + const BackupAbortRep * const rep = + CAST_CONSTPTR(BackupAbortRep, signal->getDataPtr()); + event.Event = BackupEvent::BackupAborted; + event.Aborted.Reason = rep->reason; + event.Aborted.BackupId = rep->backupId; + event.Aborted.ErrorCode = rep->reason; +#ifdef VM_TRACE + ndbout_c("Backup %d aborted", rep->backupId); +#endif + return rep->reason; + } + case GSN_NF_COMPLETEREP:{ + const NFCompleteRep * const rep = + CAST_CONSTPTR(NFCompleteRep, signal->getDataPtr()); +#ifdef VM_TRACE + ndbout_c("Node %d fail completed", rep->failedNodeId); +#endif + if (rep->failedNodeId == nodeId || + waitCompleted == 1) + return 1326; + // wait for next signal + // master node will report aborted backup break; - case BackupEvent::BackupStarted: - backupId = m_lastBackupEvent.Started.BackupId; + } + case GSN_NODE_FAILREP:{ + const NodeFailRep * const rep = + CAST_CONSTPTR(NodeFailRep, signal->getDataPtr()); +#ifdef VM_TRACE + ndbout_c("Node %d failed", rep->failNo); +#endif + if (rep->failNo == nodeId || + waitCompleted == 1) + // wait for next signal + // master node will report aborted backup break; - case BackupEvent::BackupFailedToStart: - return m_lastBackupEvent.FailedToStart.ErrorCode; - case BackupEvent::BackupAborted: - return m_lastBackupEvent.Aborted.ErrorCode; + } default: - return -1; - break; + report_unknown_signal(signal); + return SEND_OR_RECEIVE_FAILED; } } @@ -2533,36 +2550,6 @@ } return 0; -} - -void -MgmtSrvr::backupCallback(BackupEvent & event) -{ - DBUG_ENTER("MgmtSrvr::backupCallback"); - m_lastBackupEvent = event; - switch(event.Event){ - case BackupEvent::BackupFailedToStart: - DBUG_PRINT("info",("BackupEvent::BackupFailedToStart")); - theWaitState = NO_WAIT; - break; - case BackupEvent::BackupAborted: - DBUG_PRINT("info",("BackupEvent::BackupAborted")); - theWaitState = NO_WAIT; - break; - case BackupEvent::BackupCompleted: - DBUG_PRINT("info",("BackupEvent::BackupCompleted")); - theWaitState = NO_WAIT; - break; - case BackupEvent::BackupStarted: - if(theWaitState == WAIT_BACKUP_STARTED) - { - DBUG_PRINT("info",("BackupEvent::BackupStarted NO_WAIT")); - theWaitState = NO_WAIT; - } else { - DBUG_PRINT("info",("BackupEvent::BackupStarted")); - } - } - DBUG_VOID_RETURN; } --- 1.27/ndb/src/mgmsrv/MgmtSrvr.hpp 2005-07-15 13:32:05 +02:00 +++ 1.28/ndb/src/mgmsrv/MgmtSrvr.hpp 2005-09-08 11:34:51 +02:00 @@ -757,9 +757,6 @@ static void *signalRecvThread_C(void *); void signalRecvThreadRun(); - void backupCallback(BackupEvent &); - BackupEvent m_lastBackupEvent; - Config *_props; int send(class NdbApiSignal* signal, Uint32 node, Uint32 node_type); --- 1.10/ndb/src/ndbapi/Makefile.am 2004-11-11 21:25:32 +01:00 +++ 1.11/ndb/src/ndbapi/Makefile.am 2005-09-08 11:34:51 +02:00 @@ -35,7 +35,8 @@ NdbDictionaryImpl.cpp \ DictCache.cpp \ ndb_cluster_connection.cpp \ - NdbBlob.cpp + NdbBlob.cpp \ + SignalSender.cpp INCLUDES_LOC = -I$(top_srcdir)/ndb/src/mgmapi --- New file --- +++ ndb/src/ndbapi/SignalSender.cpp 05/09/08 11:34:51 /* 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 */ #include "SignalSender.hpp" #include #include #include #include SimpleSignal::SimpleSignal(bool dealloc){ memset(this, 0, sizeof(* this)); deallocSections = dealloc; } SimpleSignal::~SimpleSignal(){ if(!deallocSections) return; if(ptr[0].p != 0) delete []ptr[0].p; if(ptr[1].p != 0) delete []ptr[1].p; if(ptr[2].p != 0) delete []ptr[2].p; } void SimpleSignal::set(class SignalSender& ss, Uint8 trace, Uint16 recBlock, Uint16 gsn, Uint32 len){ header.theTrace = trace; header.theReceiversBlockNumber = recBlock; header.theVerId_signalNumber = gsn; header.theLength = len; header.theSendersBlockRef = refToBlock(ss.getOwnRef()); } void SimpleSignal::print(FILE * out){ fprintf(out, "---- Signal ----------------\n"); SignalLoggerManager::printSignalHeader(out, header, 0, 0, false); SignalLoggerManager::printSignalData(out, header, theData); for(Uint32 i = 0; i= 7){ fprintf(out, " H\'%.8x H\'%.8x H\'%.8x H\'%.8x H\'%.8x H\'%.8x H\'%.8x\n", signalData[0], signalData[1], signalData[2], signalData[3], signalData[4], signalData[5], signalData[6]); len -= 7; signalData += 7; } if(len > 0){ fprintf(out, " H\'%.8x", signalData[0]); for(Uint32 i = 1; iopen(this, execSignal, execNodeStatus); assert(m_blockNo > 0); } SignalSender::~SignalSender(){ int i; if (m_lock) unlock(); theFacade->close(m_blockNo,0); // free these _after_ closing theFacade to ensure that // we delete all signals for (i= m_jobBuffer.size()-1; i>= 0; i--) delete m_jobBuffer[i]; for (i= m_usedBuffer.size()-1; i>= 0; i--) delete m_usedBuffer[i]; NdbCondition_Destroy(m_cond); } int SignalSender::lock() { if (NdbMutex_Lock(theFacade->theMutexPtr)) return -1; m_lock= 1; return 0; } int SignalSender::unlock() { if (NdbMutex_Unlock(theFacade->theMutexPtr)) return -1; m_lock= 0; return 0; } Uint32 SignalSender::getOwnRef() const { return numberToRef(m_blockNo, theFacade->ownId()); } Uint32 SignalSender::getAliveNode() const{ return theFacade->get_an_alive_node(); } const ClusterMgr::Node & SignalSender::getNodeInfo(Uint16 nodeId) const { return theFacade->theClusterMgr->getNodeInfo(nodeId); } Uint32 SignalSender::getNoOfConnectedNodes() const { return theFacade->theClusterMgr->getNoOfConnectedNodes(); } SendStatus SignalSender::sendSignal(Uint16 nodeId, const SimpleSignal * s){ return theFacade->theTransporterRegistry->prepareSend(&s->header, 1, // JBB &s->theData[0], nodeId, &s->ptr[0]); } template SimpleSignal * SignalSender::waitFor(Uint32 timeOutMillis, T & t) { SimpleSignal * s = t.check(m_jobBuffer); if(s != 0){ return s; } NDB_TICKS now = NdbTick_CurrentMillisecond(); NDB_TICKS stop = now + timeOutMillis; Uint32 wait = (timeOutMillis == 0 ? 10 : timeOutMillis); do { NdbCondition_WaitTimeout(m_cond, theFacade->theMutexPtr, wait); SimpleSignal * s = t.check(m_jobBuffer); if(s != 0){ m_usedBuffer.push_back(s); return s; } now = NdbTick_CurrentMillisecond(); wait = (timeOutMillis == 0 ? 10 : stop - now); } while(stop > now || timeOutMillis == 0); return 0; } class WaitForAny { public: SimpleSignal * check(Vector & m_jobBuffer){ if(m_jobBuffer.size() > 0){ SimpleSignal * s = m_jobBuffer[0]; m_jobBuffer.erase(0); return s; } return 0; } }; SimpleSignal * SignalSender::waitFor(Uint32 timeOutMillis){ WaitForAny w; return waitFor(timeOutMillis, w); } class WaitForNode { public: Uint32 m_nodeId; SimpleSignal * check(Vector & m_jobBuffer){ Uint32 len = m_jobBuffer.size(); for(Uint32 i = 0; iheader.theSendersBlockRef) == m_nodeId){ SimpleSignal * s = m_jobBuffer[i]; m_jobBuffer.erase(i); return s; } } return 0; } }; SimpleSignal * SignalSender::waitFor(Uint16 nodeId, Uint32 timeOutMillis){ WaitForNode w; w.m_nodeId = nodeId; return waitFor(timeOutMillis, w); } #include void SignalSender::execSignal(void* signalSender, NdbApiSignal* signal, class LinearSectionPtr ptr[3]){ SimpleSignal * s = new SimpleSignal(true); s->header = * signal; memcpy(&s->theData[0], signal->getDataPtr(), 4 * s->header.theLength); for(Uint32 i = 0; iheader.m_noOfSections; i++){ s->ptr[i].p = new Uint32[ptr[i].sz]; s->ptr[i].sz = ptr[i].sz; memcpy(s->ptr[i].p, ptr[i].p, 4 * ptr[i].sz); } SignalSender * ss = (SignalSender*)signalSender; ss->m_jobBuffer.push_back(s); NdbCondition_Signal(ss->m_cond); } void SignalSender::execNodeStatus(void* signalSender, Uint32 nodeId, bool alive, bool nfCompleted){ if (alive) { // node connected return; } SimpleSignal * s = new SimpleSignal(true); SignalSender * ss = (SignalSender*)signalSender; // node disconnected if(nfCompleted) { // node shutdown complete s->header.theVerId_signalNumber = GSN_NF_COMPLETEREP; NFCompleteRep *rep = (NFCompleteRep *)s->getDataPtrSend(); rep->failedNodeId = nodeId; } else { // node failure s->header.theVerId_signalNumber = GSN_NODE_FAILREP; NodeFailRep *rep = (NodeFailRep *)s->getDataPtrSend(); rep->failNo = nodeId; } ss->m_jobBuffer.push_back(s); NdbCondition_Signal(ss->m_cond); } template SimpleSignal* SignalSender::waitFor(unsigned, WaitForNode&); template SimpleSignal* SignalSender::waitFor(unsigned, WaitForAny&); template Vector; --- New file --- +++ ndb/src/ndbapi/SignalSender.hpp 05/09/08 11:34:51 /* 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 */ #ifndef SIGNAL_SENDER_HPP #define SIGNAL_SENDER_HPP #include #include "TransporterFacade.hpp" #include struct SimpleSignal { public: SimpleSignal(bool dealloc = false); ~SimpleSignal(); void set(class SignalSender&, Uint8 trace, Uint16 recBlock, Uint16 gsn, Uint32 len); struct SignalHeader header; Uint32 theData[25]; LinearSectionPtr ptr[3]; int readSignalNumber() {return header.theVerId_signalNumber; } Uint32 *getDataPtrSend() { return theData; } const Uint32 *getDataPtr() const { return theData; } void print(FILE * out = stdout); private: bool deallocSections; }; class SignalSender { public: SignalSender(TransporterFacade *facade); virtual ~SignalSender(); int lock(); int unlock(); Uint32 getOwnRef() const; Uint32 getAliveNode() const; const ClusterMgr::Node &getNodeInfo(Uint16 nodeId) const; Uint32 getNoOfConnectedNodes() const; SendStatus sendSignal(Uint16 nodeId, const SimpleSignal *); SimpleSignal * waitFor(Uint32 timeOutMillis = 0); SimpleSignal * waitFor(Uint16 nodeId, Uint32 timeOutMillis = 0); SimpleSignal * waitFor(Uint16 nodeId, Uint16 gsn, Uint32 timeOutMillis = 0); private: int m_blockNo; TransporterFacade * theFacade; static void execSignal(void* signalSender, NdbApiSignal* signal, class LinearSectionPtr ptr[3]); static void execNodeStatus(void* signalSender, Uint32 nodeId, bool alive, bool nfCompleted); int m_lock; struct NdbCondition * m_cond; Vector m_jobBuffer; Vector m_usedBuffer; template SimpleSignal * waitFor(Uint32 timeOutMillis, T & t); }; #endif