Below is the list of changes that have just been committed into a local
5.1 repository of jonas. When jonas 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.2230 06/06/28 11:32:29 jonas@stripped +1 -0
Merge perch.ndb.mysql.com:/home/jonas/src/50-work
into perch.ndb.mysql.com:/home/jonas/src/51-work
storage/ndb/src/ndbapi/NdbScanOperation.cpp
1.85 06/06/28 11:32:25 jonas@stripped +0 -0
Auto merged
storage/ndb/src/ndbapi/NdbScanOperation.cpp
1.66.6.2 06/06/28 11:32:25 jonas@stripped +0 -0
Merge rename: ndb/src/ndbapi/NdbScanOperation.cpp -> storage/ndb/src/ndbapi/NdbScanOperation.cpp
# 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: jonas
# Host: perch.ndb.mysql.com
# Root: /home/jonas/src/51-work/RESYNC
--- 1.66.6.1/ndb/src/ndbapi/NdbScanOperation.cpp 2006-06-28 11:27:35 +02:00
+++ 1.85/storage/ndb/src/ndbapi/NdbScanOperation.cpp 2006-06-28 11:32:25 +02:00
@@ -50,6 +50,7 @@ NdbScanOperation::NdbScanOperation(Ndb*
m_receivers = 0;
m_array = new Uint32[1]; // skip if on delete in fix_receivers
theSCAN_TABREQ = 0;
+ m_executed = false;
}
NdbScanOperation::~NdbScanOperation()
@@ -111,6 +112,7 @@ NdbScanOperation::init(const NdbTableImp
theNdbCon->theMagicNumber = 0xFE11DF;
theNoOfTupKeyLeft = tab->m_noOfDistributionKeys;
m_read_range_no = 0;
+ m_executed = false;
return 0;
}
@@ -161,7 +163,17 @@ NdbScanOperation::readTuples(NdbScanOper
return -1;
}
- m_keyInfo = lockExcl ? 1 : 0;
+ m_keyInfo = ((scan_flags & SF_KeyInfo) || lockExcl) ? 1 : 0;
+ bool tupScan = (scan_flags & SF_TupScan);
+
+#if 1 // XXX temp for testing
+ { char* p = getenv("NDB_USE_TUPSCAN");
+ if (p != 0) {
+ unsigned n = atoi(p); // 0-10
+ if ((unsigned int) (::time(0) % 10) < n) tupScan = true;
+ }
+ }
+#endif
bool rangeScan = false;
if (m_accessTable->m_indexType == NdbDictionary::Index::OrderedIndex)
@@ -177,12 +189,9 @@ NdbScanOperation::readTuples(NdbScanOper
theStatus = GetValue;
theOperationType = OpenRangeScanRequest;
rangeScan = true;
- }
-
- bool tupScan = (scan_flags & SF_TupScan);
- if (tupScan && rangeScan)
tupScan = false;
-
+ }
+
if (rangeScan && (scan_flags & SF_OrderBy))
parallel = fragCount;
@@ -202,7 +211,7 @@ NdbScanOperation::readTuples(NdbScanOper
theSCAN_TABREQ->setSignal(GSN_SCAN_TABREQ);
ScanTabReq * req = CAST_PTR(ScanTabReq, theSCAN_TABREQ->getDataPtrSend());
req->apiConnectPtr = theNdbCon->theTCConPtr;
- req->tableId = m_accessTable->m_tableId;
+ req->tableId = m_accessTable->m_id;
req->tableSchemaVersion = m_accessTable->m_version;
req->storedProcId = 0xFFFF;
req->buddyConPtr = theNdbCon->theBuddyConPtr;
@@ -363,7 +372,7 @@ NdbScanOperation::getFirstATTRINFOScan()
int
NdbScanOperation::executeCursor(int nodeId){
NdbTransaction * tCon = theNdbCon;
- TransporterFacade* tp = TransporterFacade::instance();
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
Guard guard(tp->theMutexPtr);
Uint32 magic = tCon->theMagicNumber;
@@ -385,6 +394,7 @@ NdbScanOperation::executeCursor(int node
if (doSendScan(nodeId) == -1)
return -1;
+ m_executed= true; // Mark operation as executed
return 0;
} else {
if (!(tp->get_node_stopping(nodeId) &&
@@ -464,14 +474,25 @@ int NdbScanOperation::nextResultImpl(boo
}
Uint32 nodeId = theNdbCon->theDBnode;
- TransporterFacade* tp = TransporterFacade::instance();
- Guard guard(tp->theMutexPtr);
- if(theError.code)
- return -1;
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
+ /*
+ The PollGuard has an implicit call of unlock_and_signal through the
+ ~PollGuard method. This method is called implicitly by the compiler
+ in all places where the object is out of context due to a return,
+ break, continue or simply end of statement block
+ */
+ PollGuard poll_guard(tp, &theNdb->theImpl->theWaiter,
+ theNdb->theNdbBlockNumber);
- Uint32 seq = theNdbCon->theNodeSequence;
- if(seq == tp->getNodeSequence(nodeId) && send_next_scan(idx, false,
- forceSend) == 0){
+ const Uint32 seq = theNdbCon->theNodeSequence;
+
+ if(theError.code)
+ {
+ goto err4;
+ }
+
+ if(seq == tp->getNodeSequence(nodeId) && send_next_scan(idx, false) == 0)
+ {
idx = m_current_api_receiver;
last = m_api_receivers_count;
@@ -500,10 +521,9 @@ int NdbScanOperation::nextResultImpl(boo
/**
* No completed...
*/
- theNdb->theImpl->theWaiter.m_node = nodeId;
- theNdb->theImpl->theWaiter.m_state = WAIT_SCAN;
- int return_code = theNdb->receiveResponse(WAITFOR_SCAN_TIMEOUT);
- if (return_code == 0 && seq == tp->getNodeSequence(nodeId)) {
+ int ret_code= poll_guard.wait_scan(WAITFOR_SCAN_TIMEOUT, nodeId,
+ forceSend);
+ if (ret_code == 0 && seq == tp->getNodeSequence(nodeId)) {
continue;
} else {
idx = last;
@@ -553,6 +573,10 @@ int NdbScanOperation::nextResultImpl(boo
if(theError.code == 0)
setErrorCode(4028); // seq changed = Node fail
break;
+ case -4:
+err4:
+ setErrorCode(theError.code);
+ break;
}
theNdbCon->theTransactionIsStarted = false;
@@ -562,8 +586,8 @@ int NdbScanOperation::nextResultImpl(boo
}
int
-NdbScanOperation::send_next_scan(Uint32 cnt, bool stopScanFlag,
- bool forceSend){
+NdbScanOperation::send_next_scan(Uint32 cnt, bool stopScanFlag)
+{
if(cnt > 0){
NdbApiSignal tSignal(theNdb->theMyRef);
tSignal.setSignal(GSN_SCAN_NEXTREQ);
@@ -598,7 +622,7 @@ NdbScanOperation::send_next_scan(Uint32
if(sent)
{
Uint32 nodeId = theNdbCon->theDBnode;
- TransporterFacade * tp = TransporterFacade::instance();
+ TransporterFacade * tp = theNdb->theImpl->m_transporter_facade;
if(cnt > 21){
tSignal.setLength(4);
LinearSectionPtr ptr[3];
@@ -610,9 +634,6 @@ NdbScanOperation::send_next_scan(Uint32
ret = tp->sendSignal(&tSignal, nodeId);
}
}
-
- if (!ret) checkForceSend(forceSend);
-
m_sent_receivers_count = last + sent;
m_api_receivers_count -= cnt;
m_current_api_receiver = 0;
@@ -622,15 +643,6 @@ NdbScanOperation::send_next_scan(Uint32
return 0;
}
-void NdbScanOperation::checkForceSend(bool forceSend)
-{
- if (forceSend) {
- TransporterFacade::instance()->forceSend(theNdb->theNdbBlockNumber);
- } else {
- TransporterFacade::instance()->checkForceSend(theNdb->theNdbBlockNumber);
- }//if
-}
-
int
NdbScanOperation::prepareSend(Uint32 TC_ConnectPtr, Uint64 TransactionId)
{
@@ -665,10 +677,16 @@ void NdbScanOperation::close(bool forceS
m_conf_receivers_count,
m_sent_receivers_count);
- TransporterFacade* tp = TransporterFacade::instance();
- Guard guard(tp->theMutexPtr);
- close_impl(tp, forceSend);
-
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
+ /*
+ The PollGuard has an implicit call of unlock_and_signal through the
+ ~PollGuard method. This method is called implicitly by the compiler
+ in all places where the object is out of context due to a return,
+ break, continue or simply end of statement block
+ */
+ PollGuard poll_guard(tp, &theNdb->theImpl->theWaiter,
+ theNdb->theNdbBlockNumber);
+ close_impl(tp, forceSend, &poll_guard);
}
NdbConnection* tCon = theNdbCon;
@@ -678,7 +696,7 @@ void NdbScanOperation::close(bool forceS
if (releaseOp && tTransCon) {
NdbIndexScanOperation* tOp = (NdbIndexScanOperation*)this;
- tTransCon->releaseExecutedScanOperation(tOp);
+ tTransCon->releaseScanOperation(tOp);
}
tCon->theScanningOp = 0;
@@ -773,6 +791,7 @@ int NdbScanOperation::prepareSendScan(Ui
*/
Uint32 reqInfo = req->requestInfo;
ScanTabReq::setKeyinfoFlag(reqInfo, keyInfo);
+ ScanTabReq::setNoDiskFlag(reqInfo, m_no_disk_flag);
req->requestInfo = reqInfo;
for(Uint32 i = 0; i<theParallelism; i++){
@@ -823,7 +842,7 @@ NdbScanOperation::doSendScan(int aProces
req->requestInfo = tmp;
tSignal->setLength(ScanTabReq::StaticLength + theDistrKeyIndicator_);
- TransporterFacade *tp = TransporterFacade::instance();
+ TransporterFacade *tp = theNdb->theImpl->m_transporter_facade;
LinearSectionPtr ptr[3];
ptr[0].p = m_prepared_receivers;
ptr[0].sz = theParallelism;
@@ -907,13 +926,20 @@ NdbScanOperation::doSendScan(int aProces
* the scan process.
****************************************************************************/
int
-NdbScanOperation::getKeyFromKEYINFO20(Uint32* data, unsigned size)
+NdbScanOperation::getKeyFromKEYINFO20(Uint32* data, Uint32 & size)
{
NdbRecAttr * tRecAttr = m_curr_row;
if(tRecAttr)
{
const Uint32 * src = (Uint32*)tRecAttr->aRef();
- memcpy(data, src, 4*size);
+
+ assert(tRecAttr->get_size_in_bytes() > 0);
+ assert(tRecAttr->get_size_in_bytes() < 65536);
+ const Uint32 len = (tRecAttr->get_size_in_bytes() + 3)/4-1;
+
+ assert(size >= len);
+ memcpy(data, src, 4*len);
+ size = len;
return 0;
}
return -1;
@@ -930,18 +956,30 @@ NdbScanOperation::takeOverScanOp(Operati
if (newOp == NULL){
return NULL;
}
+ if (!m_keyInfo)
+ {
+ // Cannot take over lock if no keyinfo was requested
+ setErrorCodeAbort(4604);
+ return NULL;
+ }
pTrans->theSimpleState = 0;
- const Uint32 len = (tRecAttr->attrSize() * tRecAttr->arraySize() + 3)/4-1;
-
+ assert(tRecAttr->get_size_in_bytes() > 0);
+ assert(tRecAttr->get_size_in_bytes() < 65536);
+ const Uint32 len = (tRecAttr->get_size_in_bytes() + 3)/4-1;
+
newOp->theTupKeyLen = len;
newOp->theOperationType = opType;
- if (opType == DeleteRequest) {
- newOp->theStatus = GetValue;
- } else {
- newOp->theStatus = SetValue;
+ switch (opType) {
+ case (ReadRequest):
+ newOp->theLockMode = theLockMode;
+ // Fall through
+ case (DeleteRequest):
+ newOp->theStatus = GetValue;
+ break;
+ default:
+ newOp->theStatus = SetValue;
}
-
const Uint32 * src = (Uint32*)tRecAttr->aRef();
const Uint32 tScanInfo = src[len] & 0x3FFFF;
const Uint32 tTakeOverFragment = src[len] >> 20;
@@ -1027,23 +1065,23 @@ NdbIndexScanOperation::~NdbIndexScanOper
int
NdbIndexScanOperation::setBound(const char* anAttrName, int type,
- const void* aValue, Uint32 len)
+ const void* aValue)
{
- return setBound(m_accessTable->getColumn(anAttrName), type, aValue, len);
+ return setBound(m_accessTable->getColumn(anAttrName), type, aValue);
}
int
NdbIndexScanOperation::setBound(Uint32 anAttrId, int type,
- const void* aValue, Uint32 len)
+ const void* aValue)
{
- return setBound(m_accessTable->getColumn(anAttrId), type, aValue, len);
+ return setBound(m_accessTable->getColumn(anAttrId), type, aValue);
}
int
NdbIndexScanOperation::equal_impl(const NdbColumnImpl* anAttrObject,
- const char* aValue,
- Uint32 len){
- return setBound(anAttrObject, BoundEQ, aValue, len);
+ const char* aValue)
+{
+ return setBound(anAttrObject, BoundEQ, aValue);
}
NdbRecAttr*
@@ -1053,7 +1091,7 @@ NdbIndexScanOperation::getValue_impl(con
return NdbScanOperation::getValue_impl(attrInfo, aValue);
}
- int id = attrInfo->m_attrId; // In "real" table
+ int id = attrInfo->getColumnNo(); // In "real" table
assert(m_accessTable->m_index);
int sz = (int)m_accessTable->m_index->m_key_ids.size();
if(id >= sz || (id = m_accessTable->m_index->m_key_ids[id]) == -1){
@@ -1090,7 +1128,7 @@ NdbIndexScanOperation::getValue_impl(con
*/
int
NdbIndexScanOperation::setBound(const NdbColumnImpl* tAttrInfo,
- int type, const void* aValue, Uint32 len)
+ int type, const void* aValue)
{
if (!tAttrInfo)
{
@@ -1098,24 +1136,23 @@ NdbIndexScanOperation::setBound(const Nd
return -1;
}
if (theOperationType == OpenRangeScanRequest &&
- (0 <= type && type <= 4) &&
- len <= 8000) {
+ (0 <= type && type <= 4)) {
// insert bound type
Uint32 currLen = theTotalNrOfKeyWordInSignal;
Uint32 remaining = KeyInfo::DataLength - currLen;
- Uint32 sizeInBytes = tAttrInfo->m_attrSize * tAttrInfo->m_arraySize;
bool tDistrKey = tAttrInfo->m_distributionKey;
- len = aValue != NULL ? sizeInBytes : 0;
- if (len != sizeInBytes && (len != 0)) {
- setErrorCodeAbort(4209);
- return -1;
- }
+ Uint32 len = 0;
+ if (aValue != NULL)
+ if (! tAttrInfo->get_var_length(aValue, len)) {
+ setErrorCodeAbort(4209);
+ return -1;
+ }
// insert attribute header
Uint32 tIndexAttrId = tAttrInfo->m_attrId;
Uint32 sizeInWords = (len + 3) / 4;
- AttributeHeader ah(tIndexAttrId, sizeInWords);
+ AttributeHeader ah(tIndexAttrId, sizeInWords << 2);
const Uint32 ahValue = ah.m_value;
const Uint32 align = (UintPtr(aValue) & 7);
@@ -1209,6 +1246,31 @@ error:
return -1;
}
+Uint32
+NdbIndexScanOperation::getKeyFromSCANTABREQ(Uint32* data, Uint32 size)
+{
+ DBUG_ENTER("NdbIndexScanOperation::getKeyFromSCANTABREQ");
+ assert(size >= theTotalNrOfKeyWordInSignal);
+ size = theTotalNrOfKeyWordInSignal;
+ NdbApiSignal* tSignal = theSCAN_TABREQ->next();
+ Uint32 pos = 0;
+ while (pos < size) {
+ assert(tSignal != NULL);
+ Uint32* tData = tSignal->getDataPtrSend();
+ Uint32 rem = size - pos;
+ if (rem > KeyInfo::DataLength)
+ rem = KeyInfo::DataLength;
+ Uint32 i = 0;
+ while (i < rem) {
+ data[pos + i] = tData[KeyInfo::HeaderLength + i];
+ i++;
+ }
+ pos += rem;
+ }
+ DBUG_DUMP("key", (char*)data, size << 2);
+ DBUG_RETURN(size);
+}
+
int
NdbIndexScanOperation::readTuples(LockMode lm,
Uint32 scan_flags,
@@ -1308,10 +1370,11 @@ NdbIndexScanOperation::compare(Uint32 sk
return (r1_null ? -1 : 1) * jdir;
}
const NdbColumnImpl & col = NdbColumnImpl::getImpl(* r1->m_column);
- Uint32 len = r1->theAttrSize * r1->theArraySize;
+ Uint32 len1 = r1->get_size_in_bytes();
+ Uint32 len2 = r2->get_size_in_bytes();
if(!r1_null){
const NdbSqlUtil::Type& sqlType = NdbSqlUtil::getType(col.m_type);
- int r = (*sqlType.m_cmp)(col.m_cs, d1, len, d2, len, true);
+ int r = (*sqlType.m_cmp)(col.m_cs, d1, len1, d2, len2, true);
if(r){
assert(r != NdbSqlUtil::CmpUnknown);
return r * jdir;
@@ -1349,21 +1412,27 @@ NdbIndexScanOperation::next_result_order
if(fetchNeeded){
if(fetchAllowed){
if(DEBUG_NEXT_RESULT) ndbout_c("performing fetch...");
- TransporterFacade* tp = TransporterFacade::instance();
- Guard guard(tp->theMutexPtr);
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
+ /*
+ The PollGuard has an implicit call of unlock_and_signal through the
+ ~PollGuard method. This method is called implicitly by the compiler
+ in all places where the object is out of context due to a return,
+ break, continue or simply end of statement block
+ */
+ PollGuard poll_guard(tp, &theNdb->theImpl->theWaiter,
+ theNdb->theNdbBlockNumber);
if(theError.code)
return -1;
Uint32 seq = theNdbCon->theNodeSequence;
Uint32 nodeId = theNdbCon->theDBnode;
if(seq == tp->getNodeSequence(nodeId) &&
- !send_next_scan_ordered(s_idx, forceSend)){
+ !send_next_scan_ordered(s_idx)){
Uint32 tmp = m_sent_receivers_count;
s_idx = m_current_api_receiver;
while(m_sent_receivers_count > 0 && !theError.code){
- theNdb->theImpl->theWaiter.m_node = nodeId;
- theNdb->theImpl->theWaiter.m_state = WAIT_SCAN;
- int return_code = theNdb->receiveResponse(WAITFOR_SCAN_TIMEOUT);
- if (return_code == 0 && seq == tp->getNodeSequence(nodeId)) {
+ int ret_code= poll_guard.wait_scan(WAITFOR_SCAN_TIMEOUT, nodeId,
+ forceSend);
+ if (ret_code == 0 && seq == tp->getNodeSequence(nodeId)) {
continue;
}
if(DEBUG_NEXT_RESULT) ndbout_c("return -1");
@@ -1450,7 +1519,8 @@ NdbIndexScanOperation::next_result_order
}
int
-NdbIndexScanOperation::send_next_scan_ordered(Uint32 idx, bool forceSend){
+NdbIndexScanOperation::send_next_scan_ordered(Uint32 idx)
+{
if(idx == theParallelism)
return 0;
@@ -1485,15 +1555,16 @@ NdbIndexScanOperation::send_next_scan_or
m_sent_receivers_count = last + 1;
Uint32 nodeId = theNdbCon->theDBnode;
- TransporterFacade * tp = TransporterFacade::instance();
+ TransporterFacade * tp = theNdb->theImpl->m_transporter_facade;
tSignal.setLength(4+1);
int ret= tp->sendSignal(&tSignal, nodeId);
- if (!ret) checkForceSend(forceSend);
return ret;
}
int
-NdbScanOperation::close_impl(TransporterFacade* tp, bool forceSend){
+NdbScanOperation::close_impl(TransporterFacade* tp, bool forceSend,
+ PollGuard *poll_guard)
+{
Uint32 seq = theNdbCon->theNodeSequence;
Uint32 nodeId = theNdbCon->theDBnode;
@@ -1568,9 +1639,8 @@ NdbScanOperation::close_impl(Transporter
*/
while(theError.code == 0 && m_sent_receivers_count)
{
- theNdb->theImpl->theWaiter.m_node = nodeId;
- theNdb->theImpl->theWaiter.m_state = WAIT_SCAN;
- int return_code = theNdb->receiveResponse(WAITFOR_SCAN_TIMEOUT);
+ int return_code= poll_guard->wait_scan(WAITFOR_SCAN_TIMEOUT, nodeId,
+ false);
switch(return_code){
case 0:
break;
@@ -1627,7 +1697,7 @@ NdbScanOperation::close_impl(Transporter
}
// Send close scan
- if(send_next_scan(api+conf, true, forceSend) == -1)
+ if(send_next_scan(api+conf, true) == -1)
{
theNdbCon->theReleaseOnClose = true;
return -1;
@@ -1638,9 +1708,8 @@ NdbScanOperation::close_impl(Transporter
*/
while(m_sent_receivers_count+m_api_receivers_count+m_conf_receivers_count)
{
- theNdb->theImpl->theWaiter.m_node = nodeId;
- theNdb->theImpl->theWaiter.m_state = WAIT_SCAN;
- int return_code = theNdb->receiveResponse(WAITFOR_SCAN_TIMEOUT);
+ int return_code= poll_guard->wait_scan(WAITFOR_SCAN_TIMEOUT, nodeId,
+ forceSend);
switch(return_code){
case 0:
break;
@@ -1679,13 +1748,20 @@ int
NdbScanOperation::restart(bool forceSend)
{
- TransporterFacade* tp = TransporterFacade::instance();
- Guard guard(tp->theMutexPtr);
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
+ /*
+ The PollGuard has an implicit call of unlock_and_signal through the
+ ~PollGuard method. This method is called implicitly by the compiler
+ in all places where the object is out of context due to a return,
+ break, continue or simply end of statement block
+ */
+ PollGuard poll_guard(tp, &theNdb->theImpl->theWaiter,
+ theNdb->theNdbBlockNumber);
Uint32 nodeId = theNdbCon->theDBnode;
{
int res;
- if((res= close_impl(tp, forceSend)))
+ if((res= close_impl(tp, forceSend, &poll_guard)))
{
return res;
}
@@ -1699,7 +1775,6 @@ NdbScanOperation::restart(bool forceSend
theError.code = 0;
if (doSendScan(nodeId) == -1)
return -1;
-
return 0;
}
@@ -1708,9 +1783,16 @@ NdbIndexScanOperation::reset_bounds(bool
int res;
{
- TransporterFacade* tp = TransporterFacade::instance();
- Guard guard(tp->theMutexPtr);
- res= close_impl(tp, forceSend);
+ TransporterFacade* tp = theNdb->theImpl->m_transporter_facade;
+ /*
+ The PollGuard has an implicit call of unlock_and_signal through the
+ ~PollGuard method. This method is called implicitly by the compiler
+ in all places where the object is out of context due to a return,
+ break, continue or simply end of statement block
+ */
+ PollGuard poll_guard(tp, &theNdb->theImpl->theWaiter,
+ theNdb->theNdbBlockNumber);
+ res= close_impl(tp, forceSend, &poll_guard);
}
if(!res)
| Thread |
|---|
| • bk commit into 5.1 tree (jonas:1.2230) | jonas | 28 Jun |
