backup.cpp

mysql-5.0.22.tar.gz源码包

  ptr.p->masterData.startBackup.signalNo = 0;
  ptr.p->masterData.startBackup.noOfSignals = 
    (ptr.p->tables.noOfElements() + StartBackupReq::MaxTableTriggers - 1) / 
    StartBackupReq::MaxTableTriggers;
  sendStartBackup(signal, ptr, tabPtr);
}

/*****************************************************************************
 * 
 * Master functionallity - Start backup
 *
 *****************************************************************************/
void
Backup::sendStartBackup(Signal* signal, BackupRecordPtr ptr, TablePtr tabPtr)
{

  ptr.p->masterData.startBackup.tablePtr = tabPtr.i;
  
  StartBackupReq* req = (StartBackupReq*)signal->getDataPtrSend();
  req->backupId = ptr.p->backupId;
  req->backupPtr = ptr.i;
  req->signalNo = ptr.p->masterData.startBackup.signalNo;
  req->noOfSignals = ptr.p->masterData.startBackup.noOfSignals;
  Uint32 i;
  for(i = 0; i<StartBackupReq::MaxTableTriggers; i++) {
    jam();
    req->tableTriggers[i].tableId = tabPtr.p->tableId;
    req->tableTriggers[i].triggerIds[0] = tabPtr.p->triggerIds[0];
    req->tableTriggers[i].triggerIds[1] = tabPtr.p->triggerIds[1];
    req->tableTriggers[i].triggerIds[2] = tabPtr.p->triggerIds[2];
    if(!ptr.p->tables.next(tabPtr)){
      jam();
      i++;
      break;
    }//if
  }//for
  req->noOfTableTriggers = i;

  ptr.p->masterData.gsn = GSN_START_BACKUP_REQ;
  ptr.p->masterData.sendCounter = ptr.p->nodes;
  NodeReceiverGroup rg(BACKUP, ptr.p->nodes);
  sendSignal(rg, GSN_START_BACKUP_REQ, signal, 
	     StartBackupReq::HeaderLength + 
	     (i * StartBackupReq::TableTriggerLength), JBB);
}

void
Backup::execSTART_BACKUP_REF(Signal* signal)
{
  jamEntry();

  StartBackupRef* ref = (StartBackupRef*)signal->getDataPtr();
  const Uint32 ptrI = ref->backupPtr;
  //const Uint32 backupId = ref->backupId;
  const Uint32 signalNo = ref->signalNo;
  const Uint32 nodeId = ref->nodeId;

  BackupRecordPtr ptr;
  c_backupPool.getPtr(ptr, ptrI);

  ptr.p->setErrorCode(ref->errorCode);
  startBackupReply(signal, ptr, nodeId, signalNo);
}

void
Backup::execSTART_BACKUP_CONF(Signal* signal)
{
  jamEntry();
  
  StartBackupConf* conf = (StartBackupConf*)signal->getDataPtr();
  const Uint32 ptrI = conf->backupPtr;
  //const Uint32 backupId = conf->backupId;
  const Uint32 signalNo = conf->signalNo;
  const Uint32 nodeId = refToNode(signal->senderBlockRef());
  
  BackupRecordPtr ptr;
  c_backupPool.getPtr(ptr, ptrI);

  startBackupReply(signal, ptr, nodeId, signalNo);
}

void
Backup::startBackupReply(Signal* signal, BackupRecordPtr ptr, 
			 Uint32 nodeId, Uint32 signalNo)
{

  CRASH_INSERTION((10004));

  ndbrequire(ptr.p->masterData.startBackup.signalNo == signalNo);
  if (!haveAllSignals(ptr, GSN_START_BACKUP_REQ, nodeId)) {
    jam();
    return;
  }

  if (ERROR_INSERTED(10026))
  {
    ptr.p->errorCode = 326;
  }

  if(ptr.p->checkError()){
    jam();
    masterAbort(signal, ptr);
    return;
  }

  TablePtr tabPtr;
  c_tablePool.getPtr(tabPtr, ptr.p->masterData.startBackup.tablePtr);
  for(Uint32 i = 0; i<StartBackupReq::MaxTableTriggers; i++) {
    jam();
    if(!ptr.p->tables.next(tabPtr)) {
      jam();
      break;
    }//if
  }//for
  
  if(tabPtr.i != RNIL) {
    jam();
    ptr.p->masterData.startBackup.signalNo++;
    sendStartBackup(signal, ptr, tabPtr);
    return;
  }

  sendAlterTrig(signal, ptr);
}

/*****************************************************************************
 * 
 * Master functionallity - Activate triggers
 *
 *****************************************************************************/
void
Backup::sendAlterTrig(Signal* signal, BackupRecordPtr ptr)
{
  AlterTrigReq * req =(AlterTrigReq *)signal->getDataPtrSend();
  
  ptr.p->masterData.gsn = GSN_ALTER_TRIG_REQ;
  ptr.p->masterData.sendCounter = 0;
  
  req->setUserRef(reference());
  req->setConnectionPtr(ptr.i);
  req->setRequestType(AlterTrigReq::RT_USER);
  req->setTriggerInfo(0);       // not used on ALTER via DICT
  req->setOnline(true);
  req->setReceiverRef(reference());

  TablePtr tabPtr;

  if (ptr.p->masterData.alterTrig.tableId == RNIL) {
    jam();
    ptr.p->tables.first(tabPtr);
  } else {
    jam();
    ndbrequire(findTable(ptr, tabPtr, ptr.p->masterData.alterTrig.tableId));
    ptr.p->tables.next(tabPtr);
  }//if
  if (tabPtr.i != RNIL) {
    jam();
    ptr.p->masterData.alterTrig.tableId = tabPtr.p->tableId;
    req->setTableId(tabPtr.p->tableId);

    req->setTriggerId(tabPtr.p->triggerIds[0]);
    sendSignal(DBDICT_REF, GSN_ALTER_TRIG_REQ, 
	       signal, AlterTrigReq::SignalLength, JBB);
    
    req->setTriggerId(tabPtr.p->triggerIds[1]);
    sendSignal(DBDICT_REF, GSN_ALTER_TRIG_REQ, 
	       signal, AlterTrigReq::SignalLength, JBB);

    req->setTriggerId(tabPtr.p->triggerIds[2]);
    sendSignal(DBDICT_REF, GSN_ALTER_TRIG_REQ, 
	       signal, AlterTrigReq::SignalLength, JBB);

    ptr.p->masterData.sendCounter += 3;
    return;
  }//if
  ptr.p->masterData.alterTrig.tableId = RNIL;

  /**
   * Finished with all tables
   */
  ptr.p->masterData.gsn = GSN_WAIT_GCP_REQ;
  ptr.p->masterData.waitGCP.startBackup = true;
  
  WaitGCPReq * waitGCPReq = (WaitGCPReq*)signal->getDataPtrSend();
  waitGCPReq->senderRef = reference();
  waitGCPReq->senderData = ptr.i;
  waitGCPReq->requestType = WaitGCPReq::CompleteForceStart;
  sendSignal(DBDIH_REF, GSN_WAIT_GCP_REQ, signal, 
	     WaitGCPReq::SignalLength,JBB);
}

void
Backup::execALTER_TRIG_CONF(Signal* signal)
{
  jamEntry();

  AlterTrigConf* conf = (AlterTrigConf*)signal->getDataPtr();
  const Uint32 ptrI = conf->getConnectionPtr();
  
  BackupRecordPtr ptr;
  c_backupPool.getPtr(ptr, ptrI);
  
  alterTrigReply(signal, ptr);
}

void
Backup::execALTER_TRIG_REF(Signal* signal)
{
  jamEntry();

  AlterTrigRef* ref = (AlterTrigRef*)signal->getDataPtr();
  const Uint32 ptrI = ref->getConnectionPtr();
  
  BackupRecordPtr ptr;
  c_backupPool.getPtr(ptr, ptrI);

  ptr.p->setErrorCode(ref->getErrorCode());
  
  alterTrigReply(signal, ptr);
}

void
Backup::alterTrigReply(Signal* signal, BackupRecordPtr ptr)
{

  CRASH_INSERTION((10005));

  ndbrequire(ptr.p->masterRef == reference());
  ndbrequire(ptr.p->masterData.gsn == GSN_ALTER_TRIG_REQ);
  ndbrequire(ptr.p->masterData.sendCounter.done() == false);

  ptr.p->masterData.sendCounter--;

  if(ptr.p->masterData.sendCounter.done() == false){
    jam();
    return;
  }//if

  if(ptr.p->checkError()){
    jam();
    masterAbort(signal, ptr);
    return;
  }//if

  sendAlterTrig(signal, ptr);
}

void
Backup::execWAIT_GCP_REF(Signal* signal)
{
  jamEntry();
  
  CRASH_INSERTION((10006));

  WaitGCPRef * ref = (WaitGCPRef*)signal->getDataPtr();
  const Uint32 ptrI = ref->senderData;
  
  BackupRecordPtr ptr;
  c_backupPool.getPtr(ptr, ptrI);

  ndbrequire(ptr.p->masterRef == reference());
  ndbrequire(ptr.p->masterData.gsn == GSN_WAIT_GCP_REQ);

  WaitGCPReq * req = (WaitGCPReq*)signal->getDataPtrSend();
  req->senderRef = reference();
  req->senderData = ptr.i;
  req->requestType = WaitGCPReq::CompleteForceStart;
  sendSignal(DBDIH_REF, GSN_WAIT_GCP_REQ, signal, 
	     WaitGCPReq::SignalLength,JBB);
}

void
Backup::execWAIT_GCP_CONF(Signal* signal){
  jamEntry();

  CRASH_INSERTION((10007));

  WaitGCPConf * conf = (WaitGCPConf*)signal->getDataPtr();
  const Uint32 ptrI = conf->senderData;
  const Uint32 gcp = conf->gcp;
  
  BackupRecordPtr ptr;
  c_backupPool.getPtr(ptr, ptrI);
  
  ndbrequire(ptr.p->masterRef == reference());
  ndbrequire(ptr.p->masterData.gsn == GSN_WAIT_GCP_REQ);
  
  if(ptr.p->checkError()) {
    jam();
    masterAbort(signal, ptr);
    return;
  }//if
  
  if(ptr.p->masterData.waitGCP.startBackup) {
    jam();
    CRASH_INSERTION((10008));
    ptr.p->startGCP = gcp;
    ptr.p->masterData.sendCounter= 0;
    ptr.p->masterData.gsn = GSN_BACKUP_FRAGMENT_REQ;
    nextFragment(signal, ptr);
    return;
  } else {
    jam();
    if(gcp >= ptr.p->startGCP + 3)
    {
      CRASH_INSERTION((10009));
      ptr.p->stopGCP = gcp;
      sendDropTrig(signal, ptr); // regular dropping of triggers
      return;
    }//if
    
    /**
     * Make sure that we got entire stopGCP 
     */
    WaitGCPReq * req = (WaitGCPReq*)signal->getDataPtrSend();
    req->senderRef = reference();
    req->senderData = ptr.i;
    req->requestType = WaitGCPReq::CompleteForceStart;
    sendSignal(DBDIH_REF, GSN_WAIT_GCP_REQ, signal, 
	       WaitGCPReq::SignalLength,JBB);
    return;
  }
}

/*****************************************************************************
 * 
 * Master functionallity - Backup fragment
 *
 *****************************************************************************/
void
Backup::nextFragment(Signal* signal, BackupRecordPtr ptr)
{
  jam();

  BackupFragmentReq* req = (BackupFragmentReq*)signal->getDataPtrSend();
  req->backupPtr = ptr.i;
  req->backupId = ptr.p->backupId;

  NodeBitmask nodes = ptr.p->nodes;
  Uint32 idleNodes = nodes.count();
  Uint32 saveIdleNodes = idleNodes;
  ndbrequire(idleNodes > 0);

  TablePtr tabPtr;
  ptr.p->tables.first(tabPtr);
  for(; tabPtr.i != RNIL && idleNodes > 0; ptr.p->tables.next(tabPtr)) {
    jam();
    FragmentPtr fragPtr;
    Array<Fragment> & frags = tabPtr.p->fragments;
    const Uint32 fragCount = frags.getSize();
    
    for(Uint32 i = 0; i<fragCount && idleNodes > 0; i++) {
      jam();
      tabPtr.p->fragments.getPtr(fragPtr, i);
      const Uint32 nodeId = fragPtr.p->node;
      if(fragPtr.p->scanning != 0) {
        jam();
	ndbrequire(nodes.get(nodeId));
	nodes.clear(nodeId);
	idleNodes--;
      } else if(fragPtr.p->scanned == 0 && nodes.get(nodeId)){
	jam();
	fragPtr.p->scanning = 1;
	nodes.clear(nodeId);
	idleNodes--;
	
	req->tableId = tabPtr.p->tableId;
	req->fragmentNo = i;
	req->count = 0;

	ptr.p->masterData.sendCounter++;
	const BlockReference ref = numberToRef(BACKUP, nodeId);
	sendSignal(ref, GSN_BACKUP_FRAGMENT_REQ, signal,
		   BackupFragmentReq::SignalLength, JBB);
      }//if
    }//for
  }//for
  
  if(idleNodes != saveIdleNodes){
    jam();
    return;
  }//if

  /**
   * Finished with all tables
   */
  {
    ptr.p->masterData.gsn = GSN_WAIT_GCP_REQ;
    ptr.p->masterData.waitGCP.startBackup = false;
    
    WaitGCPReq * req = (WaitGCPReq*)signal->getDataPtrSend();
    req->senderRef = reference();
    req->senderData = ptr.i;
    req->requestType = WaitGCPReq::CompleteForceStart;
    sendSignal(DBDIH_REF, GSN_WAIT_GCP_REQ, signal, 
	       WaitGCPReq::SignalLength, JBB);
  }
}

void
Backup::execBACKUP_FRAGMENT_CONF(Signal* signal)
{
  jamEntry();

  CRASH_INSERTION((10010));
  
  BackupFragmentConf * conf = (BackupFragmentConf*)signal->getDataPtr();
  const Uint32 ptrI = conf->backupPtr;
  //const Uint32 backupId = conf->backupId;
  const Uint32 tableId = conf->tableId;
  const Uint32 fragmentNo = conf->fragmentNo;
  const Uint32 nodeId = refToNode(signal->senderBlockRef());
  const Uint32 noOfBytes = conf->noOfBytes;
  const Uint32 noOfRecords = conf->noOfRecords;

  BackupRecordPtr ptr;
  c_backupPool.getPtr(ptr, ptrI);

  ptr.p->noOfBytes += noOfBytes;
  ptr.p->noOfRecords += noOfRecords;
  ptr.p->masterData.sendCounter--;

  TablePtr tabPtr;
  ndbrequire(findTable(ptr, tabPtr, tableId));

  FragmentPtr fragPtr;
  tabPtr.p->fragments.getPtr(fragPtr, fragmentNo);

  ndbrequire(fragPtr.p->scanned == 0);
  ndbrequire(fragPtr.p->scanning == 1);
  ndbrequire(fragPtr.p->node == nodeId);

  fragPtr.p->scanned = 1;
  fragPtr.p->scanning = 0;

  if (ERROR_INSERTED(10028)) 
  {
    ptr.p->errorCode = 328;
  }

  if(ptr.p->checkError()) 
  {
    if(ptr.p->masterData.sendCounter.done())
    {
      jam();
      masterAbort(signal, ptr);
      return;
    }//if
  }
  else
  {
    nextFragment(signal, ptr);
  }
}

void
Backup::execBACKUP_FRAGMENT_REF(Signal* signal)
{
  jamEntry();

  CRASH_INSERTION((10011));

  BackupFragmentRef * ref = (BackupFragmentRef*)signal->getDataPtr();
  const Uint32 ptrI = ref->backupPtr;
  //const Uint32 backupId = ref->backupId;
  const Uint32 nodeId = ref->nodeId;
  
  BackupRecordPtr ptr;
  c_backupPool.getPtr(ptr, ptrI);

  TablePtr tabPtr;
  ptr.p->tables.first(tabPtr);
  for(; tabPtr.i != RNIL; ptr.p->tables.next(tabPtr)) {
    jam();
    FragmentPtr fragPtr;
    Array<Fragment> & frags = tabPtr.p->fragments;
    const Uint32 fragCount = frags.getSize();
    
    for(Uint32 i = 0; i<fragCount; i++) {
      jam();
      tabPtr.p->fragments.getPtr(fragPtr, i);
        if(fragPtr.p->scanning != 0 && nodeId == fragPtr.p->node) 
      {
        jam();
	ndbrequire(fragPtr.p->scanned == 0);
	fragPtr.p->scanned = 1;
	fragPtr.p->scanning = 0;
	goto done;
      }
    }
  }
  goto err;

done:
  ptr.p->masterData.sendCounter--;
  ptr.p->setErrorCode(ref->errorCode);
  
  if(ptr.p->masterData.sendCounter.done())