Merge joreland@bk-internal.mysql.com:/home/bk/mysql-5.1-new-ndb

into  perch.ndb.mysql.com:/home/jonas/src/mysql-5.1-new-ndb

storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp

@@ -5016,12 +5016,13 @@ void Dblqh::packLqhkeyreqLab(Signal* signal)
 
   Uint32 nextNodeId = regTcPtr->nextReplica;
   Uint32 nextVersion = getNodeInfo(nextNodeId).m_version;
+  UintR TAiLen = regTcPtr->reclenAiLqhkey;
 
   UintR TapplAddressIndicator = (regTcPtr->nextSeqNoReplica == 0 ? 0 : 1);
   LqhKeyReq::setApplicationAddressFlag(Treqinfo, TapplAddressIndicator);
   LqhKeyReq::setInterpretedFlag(Treqinfo, regTcPtr->opExec);
   LqhKeyReq::setSeqNoReplica(Treqinfo, regTcPtr->nextSeqNoReplica);
-  LqhKeyReq::setAIInLqhKeyReq(Treqinfo, regTcPtr->reclenAiLqhkey);
+  LqhKeyReq::setAIInLqhKeyReq(Treqinfo, TAiLen);
   
   if (unlikely(nextVersion < NDBD_ROWID_VERSION))
   {
@@ -5124,13 +5125,16 @@ void Dblqh::packLqhkeyreqLab(Signal* signal)
   lqhKeyReq->variableData[nextPos + 0] = sig0;
   nextPos += LqhKeyReq::getGCIFlag(Treqinfo);
 
+  BlockReference lqhRef = calcLqhBlockRef(regTcPtr->nextReplica);
+  
+  if (likely(nextPos + TAiLen + LqhKeyReq::FixedSignalLength <= 25))
+  {
+    jam();
     sig0 = regTcPtr->firstAttrinfo[0];
     sig1 = regTcPtr->firstAttrinfo[1];
     sig2 = regTcPtr->firstAttrinfo[2];
     sig3 = regTcPtr->firstAttrinfo[3];
     sig4 = regTcPtr->firstAttrinfo[4];
-  UintR TAiLen = regTcPtr->reclenAiLqhkey;
-  BlockReference lqhRef = calcLqhBlockRef(regTcPtr->nextReplica);
 
     lqhKeyReq->variableData[nextPos] = sig0;
     lqhKeyReq->variableData[nextPos + 1] = sig1;
@@ -5139,6 +5143,13 @@ void Dblqh::packLqhkeyreqLab(Signal* signal)
     lqhKeyReq->variableData[nextPos + 4] = sig4;
     
     nextPos += TAiLen;
+    TAiLen = 0;
+  }
+  else
+  {
+    Treqinfo &= ~(Uint32)(RI_AI_IN_THIS_MASK << RI_AI_IN_THIS_SHIFT);
+    lqhKeyReq->requestInfo = Treqinfo;
+  }
   
   sendSignal(lqhRef, GSN_LQHKEYREQ, signal, 
              nextPos + LqhKeyReq::FixedSignalLength, JBB);
@@ -5165,6 +5176,17 @@ void Dblqh::packLqhkeyreqLab(Signal* signal)
   signal->theData[0] = sig0;
   signal->theData[1] = sig1;
   signal->theData[2] = sig2;
+  
+  if (unlikely(nextPos + TAiLen + LqhKeyReq::FixedSignalLength > 25))
+  {
+    jam();
+    /**
+     * 4 replicas...
+     */
+    memcpy(signal->theData+3, regTcPtr->firstAttrinfo, TAiLen << 2);
+    sendSignal(lqhRef, GSN_ATTRINFO, signal, 3 + TAiLen, JBB);    
+  }
+
   AttrbufPtr regAttrinbufptr;
   regAttrinbufptr.i = regTcPtr->firstAttrinbuf;
   while (regAttrinbufptr.i != RNIL) {

storage/ndb/src/kernel/blocks/ndbfs/AsyncFile.cpp

@@ -312,11 +312,12 @@ void AsyncFile::openReq(Request* request)
   Uint32 new_flags = 0;
 
   // Convert file open flags from Solaris to Liux
-  if(flags & FsOpenReq::OM_CREATE){
+  if (flags & FsOpenReq::OM_CREATE)
+  {
     new_flags |= O_CREAT;
   }
   
-  if(flags & FsOpenReq::OM_TRUNCATE){
+  if (flags & FsOpenReq::OM_TRUNCATE){
 #if 0
     if(Global_unlinkO_CREAT){
       unlink(theFileName.c_str());
@@ -330,25 +331,25 @@ void AsyncFile::openReq(Request* request)
     m_syncFrequency = 1024*1024; // Hard coded to 1M
   }
 
-  if(flags & FsOpenReq::OM_APPEND){
+  if (flags & FsOpenReq::OM_APPEND){
     new_flags |= O_APPEND;
   }
 
-  if((flags & FsOpenReq::OM_SYNC) && ! (flags & FsOpenReq::OM_INIT))
+  if ((flags & FsOpenReq::OM_SYNC) && ! (flags & FsOpenReq::OM_INIT))
   {
 #ifdef O_SYNC
     new_flags |= O_SYNC;
 #endif
   }
   
-#ifndef NDB_NO_O_DIRECT  /* to allow tmpfs */
+//#ifndef NDB_NO_O_DIRECT  /* to allow tmpfs */
 #ifdef O_DIRECT
   if (flags & FsOpenReq::OM_DIRECT) 
   {
     new_flags |= O_DIRECT;
   }
 #endif
-#endif
+//#endif
   
   switch(flags & 0x3){
   case FsOpenReq::OM_READONLY:
@@ -370,8 +371,14 @@ void AsyncFile::openReq(Request* request)
   const int mode = S_IRUSR | S_IWUSR |
 	           S_IRGRP | S_IWGRP |
 		   S_IROTH | S_IWOTH;
-  if(flags & FsOpenReq::OM_CREATE_IF_NONE){
-    if((theFd = ::open(theFileName.c_str(), new_flags, mode)) != -1) {
+  if (flags & FsOpenReq::OM_CREATE_IF_NONE)
+  {
+    Uint32 tmp_flags = new_flags;
+#ifdef O_DIRECT
+    tmp_flags &= ~O_DIRECT;
+#endif
+    if ((theFd = ::open(theFileName.c_str(), tmp_flags, mode)) != -1) 
+    {
       close(theFd);
       request->error = FsRef::fsErrFileExists;      
       return;
@@ -379,35 +386,58 @@ void AsyncFile::openReq(Request* request)
     new_flags |= O_CREAT;
   }
 
-  if (-1 == (theFd = ::open(theFileName.c_str(), new_flags, mode))) {
+no_odirect:
+  if (-1 == (theFd = ::open(theFileName.c_str(), new_flags, mode))) 
+  {
     PRINT_ERRORANDFLAGS(new_flags);
-    if( (errno == ENOENT ) && (new_flags & O_CREAT ) ) {
+    if ((errno == ENOENT) && (new_flags & O_CREAT)) 
+    {
       createDirectories();
-      if (-1 == (theFd = ::open(theFileName.c_str(), new_flags, mode))) {
+      if (-1 == (theFd = ::open(theFileName.c_str(), new_flags, mode))) 
+      {
+#ifdef O_DIRECT
+	if (new_flags & O_DIRECT)
+	{
+	  new_flags &= ~O_DIRECT;
+	  goto no_odirect;
+	}
+#endif
         PRINT_ERRORANDFLAGS(new_flags);
         request->error = errno;
 	return;
       }
-    } else {
+    }
+#ifdef O_DIRECT
+    else if (new_flags & O_DIRECT)
+    {
+      new_flags &= ~O_DIRECT;
+      goto no_odirect;
+    }
+#endif
+    else
+    {
       request->error = errno;
       return;
     }
   }
   
-  if(flags & FsOpenReq::OM_CHECK_SIZE)
+  if (flags & FsOpenReq::OM_CHECK_SIZE)
   {
     struct stat buf;
-    if((fstat(theFd, &buf) == -1))
+    if ((fstat(theFd, &buf) == -1))
     {
       request->error = errno;
-    } else if(buf.st_size != request->par.open.file_size){
+    } 
+    else if(buf.st_size != request->par.open.file_size)
+    {
       request->error = FsRef::fsErrInvalidFileSize;
     }
-    if(request->error)
+    if (request->error)
       return;
   }
   
-  if(flags & FsOpenReq::OM_INIT){
+  if (flags & FsOpenReq::OM_INIT)
+  {
     off_t off = 0;
     const off_t sz = request->par.open.file_size;
     Uint32 tmp[sizeof(SignalHeader)+25];

storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp

@@ -1279,6 +1279,23 @@ find_bucket(Vector<Gci_container> * active, Uint64 gci)
   return find_bucket_chained(active,gci);
 }
 
+static
+void
+crash_on_invalid_SUB_GCP_COMPLETE_REP(const Gci_container* bucket,
+				      const SubGcpCompleteRep * const rep,
+				      Uint32 nodes)
+{
+  Uint32 old_cnt = bucket->m_gcp_complete_rep_count;
+  
+  ndbout_c("INVALID SUB_GCP_COMPLETE_REP");
+  ndbout_c("gci: %d", rep->gci);
+  ndbout_c("sender: %x", rep->senderRef);
+  ndbout_c("count: %d", rep->gcp_complete_rep_count);
+  ndbout_c("bucket count: %u", old_cnt);
+  ndbout_c("nodes: %u", nodes);
+  abort();
+}
+
 void
 NdbEventBuffer::execSUB_GCP_COMPLETE_REP(const SubGcpCompleteRep * const rep)
 {
@@ -1317,7 +1334,11 @@ NdbEventBuffer::execSUB_GCP_COMPLETE_REP(const SubGcpCompleteRep * const rep)
     old_cnt = m_system_nodes;
   }
   
-  assert(old_cnt >= cnt);
+  //assert(old_cnt >= cnt);
+  if (unlikely(! (old_cnt >= cnt)))
+  {
+    crash_on_invalid_SUB_GCP_COMPLETE_REP(bucket, rep, m_system_nodes);
+  }
   bucket->m_gcp_complete_rep_count = old_cnt - cnt;
   
   if(old_cnt == cnt)

storage/ndb/src/ndbapi/Ndbif.cpp

@@ -1435,8 +1435,7 @@ NdbTransaction::sendTC_COMMIT_ACK(TransporterFacade *tp,
   Uint32 * dataPtr = aSignal->getDataPtrSend();
   dataPtr[0] = transId1;
   dataPtr[1] = transId2;
-
-  tp->sendSignal(aSignal, refToNode(aTCRef));
+  tp->sendSignalUnCond(aSignal, refToNode(aTCRef));
 }
 
 int

storage/ndb/src/ndbapi/TransporterFacade.cpp

@@ -343,7 +343,7 @@ execute(void * callbackObj, SignalHeader * const header,
 	 Uint32 aNodeId= refToNode(ref);
 	 tSignal.theReceiversBlockNumber= refToBlock(ref);
 	 tSignal.theVerId_signalNumber= GSN_SUB_GCP_COMPLETE_ACK;
-	 theFacade->sendSignal(&tSignal, aNodeId);
+	 theFacade->sendSignalUnCond(&tSignal, aNodeId);
        }
        break;
      }
@@ -987,7 +987,7 @@ TransporterFacade::sendSignal(NdbApiSignal * aSignal, NodeId aNode){
       LinearSectionPtr ptr[3];
       signalLogger.sendSignal(* aSignal,
 			      1,
-			      aSignal->getDataPtr(),
+			      tDataPtr,
 			      aNode, ptr, 0);
       signalLogger.flushSignalLog();
       aSignal->theSendersBlockRef = tmp;
@@ -1014,6 +1014,7 @@ TransporterFacade::sendSignal(NdbApiSignal * aSignal, NodeId aNode){
 
 int
 TransporterFacade::sendSignalUnCond(NdbApiSignal * aSignal, NodeId aNode){
+  Uint32* tDataPtr = aSignal->getDataPtrSend();
 #ifdef API_TRACE
   if(setSignalLog() && TRACE_GSN(aSignal->theVerId_signalNumber)){
     Uint32 tmp = aSignal->theSendersBlockRef;
@@ -1021,7 +1022,7 @@ TransporterFacade::sendSignalUnCond(NdbApiSignal * aSignal, NodeId aNode){
     LinearSectionPtr ptr[3];
     signalLogger.sendSignal(* aSignal,
 			    0,
-			    aSignal->getDataPtr(),
+			    tDataPtr,
 			    aNode, ptr, 0);
     signalLogger.flushSignalLog();
     aSignal->theSendersBlockRef = tmp;
@@ -1032,7 +1033,7 @@ TransporterFacade::sendSignalUnCond(NdbApiSignal * aSignal, NodeId aNode){
          (aSignal->theReceiversBlockNumber != 0));
   SendStatus ss = theTransporterRegistry->prepareSend(aSignal, 
 						      0, 
-						      aSignal->getDataPtr(), 
+						      tDataPtr,
 						      aNode, 
 						      0);
   

storage/ndb/src/ndbapi/TransporterFacade.hpp

@@ -175,6 +175,7 @@ private:
   friend class GrepSS;
   friend class Ndb;
   friend class Ndb_cluster_connection_impl;
+  friend class NdbTransaction;
   
   int sendSignalUnCond(NdbApiSignal *, NodeId nodeId);