ndb - Use BitmaskImpl in TUP (in BitField operations)

ndb/include/util/Bitmask.hpp:
  New methods for working with bit fields of
    arbitrary size
ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp:
  Use BitmaskImpl at all time (in BitField operations)

2 files changed, 88 insertions, 68 deletions

diff --git a/ndb/include/util/Bitmask.hpp b/ndb/include/util/Bitmask.hpp
index 19aa604e4a1..6d932cac7c5 100644
--- a/ndb/include/util/Bitmask.hpp
+++ b/ndb/include/util/Bitmask.hpp
@@ -131,6 +131,19 @@ public:
   static void setField(unsigned size, Uint32 data[],
       unsigned pos, unsigned len, Uint32 val);
 
+
+  /**
+   * getField - Get bitfield at given position and length
+   */
+  static void getField(unsigned size, const Uint32 data[],
+		       unsigned pos, unsigned len, Uint32 dst[]);
+  
+  /**
+   * setField - Set bitfield at given position and length
+   */
+  static void setField(unsigned size, Uint32 data[],
+		       unsigned pos, unsigned len, const Uint32 src[]);
+  
   /**
    * getText - Return as hex-digits (only for debug routines).
    */
@@ -794,4 +807,38 @@ public:
   Bitmask() { this->clear();}
 };
 
+inline void
+BitmaskImpl::getField(unsigned size, const Uint32 data[],
+		      unsigned pos, unsigned len, Uint32 dst[])
+{
+  assert(len > 0);
+  assert(pos + len < (size << 5));
+  Uint32 word = pos >> 5;
+  Uint32 offset = pos & 31;
+  if(offset + len <= 32)
+  {
+    dst[0] = (data[word] >> offset) & ((1 << len) - 1);
+    return;
+  }
+  abort();
+}
+
+inline void
+BitmaskImpl::setField(unsigned size, Uint32 data[],
+		      unsigned pos, unsigned len, const Uint32 src[])
+{
+  assert(len > 0);
+  assert(pos + len < (size << 5));
+  Uint32 word = pos >> 5;
+  Uint32 offset = pos & 31;
+  if(offset + len <= 32)
+  {
+    Uint32 mask = ((1 << len) - 1) << offset;
+    data[word] = (data[word] & ~mask) | ((src[0] << offset) & mask);
+    return;
+  }
+  abort();
+}
+
+
 #endif
diff --git a/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp b/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp
index 37224bbca5f..4c872e62cef 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp
+++ b/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp
@@ -1028,22 +1028,22 @@ Dbtup::readBitsNotNULL(Uint32* outBuffer,
   offsetInTuple += regTabPtr->tupNullIndex;
   ndbrequire(offsetInTuple < tCheckOffset);
 
+  Uint32 pos = offsetInTuple << 5 + offsetInWord;
   Uint32 indexBuf = tOutBufIndex;
   Uint32 newIndexBuf = indexBuf + ((bitCount + 31) >> 5);
   Uint32 maxRead = tMaxRead;
-
-  outBuffer[indexBuf] = 
-    (tTupleHeader[offsetInTuple] >> offsetInWord) & ((1 << bitCount) - 1);
-
-  if(offsetInWord + bitCount > 32)
-  {
-    ndbrequire(false);
-  }
-
+  
   if (newIndexBuf <= maxRead) {
     ljam();
     ahOut->setDataSize((bitCount + 31) >> 5);
     tOutBufIndex = newIndexBuf;
+    
+    BitmaskImpl::getField(regTabPtr->tupNullWords,
+			  tTupleHeader+regTabPtr->tupNullIndex,
+			  pos, 
+			  bitCount,
+			  outBuffer+indexBuf);
+    
     return true;
   } else {
     ljam();
@@ -1069,34 +1069,27 @@ Dbtup::readBitsNULLable(Uint32* outBuffer,
   Uint32 indexBuf = tOutBufIndex;
   Uint32 newIndexBuf = indexBuf + (bitCount + 31) >> 5;
   Uint32 maxRead = tMaxRead;
-
-  if((tTupleHeader[offsetInTuple] >> offsetInWord) & 1)
+  Uint32 pos = offsetInWord << 5 + offsetInTuple;
+  
+  if(BitmaskImpl::get(regTabPtr->tupNullWords,
+		      tTupleHeader+regTabPtr->tupNullIndex,
+		      pos))
   {
     ljam();
     ahOut->setNULL();
     return true;
   }
 
-  offsetInWord ++;
-  if(offsetInWord == 32)
-  {
-    ljam();
-    offsetInWord = 0;
-    offsetInTuple++;
-  }
-
-  outBuffer[indexBuf] = 
-    (tTupleHeader[offsetInTuple] >> offsetInWord) & ((1 << bitCount) - 1);
-  
-  if(offsetInWord + bitCount > 32)
-  {
-    ndbrequire(false);
-  }
 
   if (newIndexBuf <= maxRead) {
     ljam();
     ahOut->setDataSize((bitCount + 31) >> 5);
     tOutBufIndex = newIndexBuf;
+    BitmaskImpl::getField(regTabPtr->tupNullWords,
+			  tTupleHeader+regTabPtr->tupNullIndex,
+			  pos+1, 
+			  bitCount,
+			  outBuffer+indexBuf);
     return true;
   } else {
     ljam();
@@ -1123,24 +1116,16 @@ Dbtup::updateBitsNotNULL(Uint32* inBuffer,
   ndbrequire((nullFlagOffset < regTabPtr->tupNullWords) &&
              (nullWordOffset < tCheckOffset));
   Uint32 nullBits = tTupleHeader[nullWordOffset];
-
+  Uint32 pos = (nullFlagOffset << 5) + nullFlagBitOffset;
+  
   if (newIndex <= inBufLen) {
-    Uint32 updateWord = inBuffer[indexBuf + 1];
     if (!nullIndicator) {
+      BitmaskImpl::setField(regTabPtr->tupNullWords,
+			    tTupleHeader+regTabPtr->tupNullIndex,
+			    pos,
+			    bitCount,
+			    inBuffer+indexBuf+1);
       tInBufIndex = newIndex;
-      Uint32 mask = ((1 << bitCount) - 1) << nullFlagBitOffset;
-      if(nullFlagBitOffset + bitCount <= 32)
-      {
-	ljam();
-	tTupleHeader[nullWordOffset] = 
-	  (nullBits & ~mask) | 
-	  ((updateWord << nullFlagBitOffset) & mask);
-	return true;
-      }
-      else
-      {
-	abort();
-      }
       return true;
     } else {
       ljam();
@@ -1171,41 +1156,29 @@ Dbtup::updateBitsNULLable(Uint32* inBuffer,
              (nullWordOffset < tCheckOffset));
   Uint32 nullBits = tTupleHeader[nullWordOffset];
   Uint32 bitCount = AttributeDescriptor::getArraySize(attrDescriptor);  
+  Uint32 pos = (nullFlagOffset << 5) + nullFlagBitOffset;
 
   if (!nullIndicator) {
-    Uint32 newIndex = indexBuf + 1 + ((bitCount + 31) >> 5);
-    nullBits &= (~(1 << nullFlagBitOffset));
-    ljam();
-    tTupleHeader[nullWordOffset] = nullBits;
+    BitmaskImpl::clear(regTabPtr->tupNullWords,
+		       tTupleHeader+regTabPtr->tupNullIndex,
+		       pos);
+    BitmaskImpl::setField(regTabPtr->tupNullWords,
+			  tTupleHeader+regTabPtr->tupNullIndex,
+			  pos+1,
+			  bitCount,
+			  inBuffer+indexBuf+1);
 
-    nullFlagBitOffset++;
-    if(nullFlagBitOffset == 32)
-    {
-      ljam();
-      nullFlagBitOffset = 0;
-      nullWordOffset++;
-      nullBits = tTupleHeader[nullWordOffset];
-    }
-
-    Uint32 updateWord = inBuffer[indexBuf + 1];
-    Uint32 mask = ((1 << bitCount) - 1) << nullFlagBitOffset;
+    Uint32 newIndex = indexBuf + 1 + ((bitCount + 31) >> 5);
     tInBufLen = newIndex;
-    if(nullFlagBitOffset + bitCount <= 32)
-      {
-	ljam();
-      tTupleHeader[nullWordOffset] = 
-	(nullBits & ~mask) | 
-	((updateWord << nullFlagBitOffset) & mask);
-      return true;
-    }  
-    
-    abort();
+    return true;
   } else {
     Uint32 newIndex = tInBufIndex + 1;
     if (newIndex <= tInBufLen) {
-      nullBits |= (1 << nullFlagBitOffset);
       ljam();
-      tTupleHeader[nullWordOffset] = nullBits;
+      BitmaskImpl::set(regTabPtr->tupNullWords,
+		       tTupleHeader+regTabPtr->tupNullIndex,
+		       pos);
+      
       tInBufIndex = newIndex;
       return true;
     } else {