version2/src/C/Commit.cc

   1 #include "Commit.h"
   2 #include "CommitPart.h"
   3 #include "ByteBuffer.h"
   4 #include "IoTString.h"
   5
   6 Commit::Commit() :
   7         parts(new Vector<CommitPart *>()),
   8         partCount(0),
   9         missingParts(NULL),
  10         fldisComplete(false),
  11         hasLastPart(false),
  12         keyValueUpdateSet(new Hashset<KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue>()),
  13         isDead(false),
  14         sequenceNumber(-1),
  15         machineId(-1),
  16         transactionSequenceNumber(-1),
  17         liveKeys(new Hashset<IoTString *>) {
  18 }
  19
  20 Commit::Commit(int64_t _sequenceNumber, int64_t _machineId, int64_t _transactionSequenceNumber) :
  21         parts(new Vector<CommitPart *>()),
  22         partCount(0),
  23         missingParts(NULL),
  24         fldisComplete(true),
  25         hasLastPart(false),
  26         keyValueUpdateSet(new Hashset<KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue>()),
  27         isDead(false),
  28         sequenceNumber(_sequenceNumber),
  29         machineId(_machineId),
  30         transactionSequenceNumber(_transactionSequenceNumber),
  31         liveKeys(new Hashset<IoTString *>) {
  32 }
  33
  34 void Commit::addPartDecode(CommitPart *newPart) {
  35         if (isDead) {
  36                 // If dead then just kill this part and move on
  37                 newPart->setDead();
  38                 return;
  39         }
  40
  41         CommitPart *previouslySeenPart = parts->setExpand(newPart->getPartNumber(), newPart);
  42         if (previouslySeenPart == NULL)
  43                 partCount++;
  44
  45         if (previouslySeenPart != NULL) {
  46                 // Set dead the old one since the new one is a rescued version of this part
  47                 previouslySeenPart->setDead();
  48         } else if (newPart->isLastPart()) {
  49                 missingParts = new Hashset<int32_t>();
  50                 hasLastPart = true;
  51
  52                 for (int i = 0; i < newPart->getPartNumber(); i++) {
  53                         if (parts->get(i) == NULL) {
  54                                 missingParts->add(i);
  55                         }
  56                 }
  57         }
  58
  59         if (!fldisComplete && hasLastPart) {
  60
  61                 // We have seen this part so remove it from the set of missing parts
  62                 missingParts->remove(newPart->getPartNumber());
  63
  64                 // Check if all the parts have been seen
  65                 if (missingParts->size() == 0) {
  66
  67                         // We have all the parts
  68                         fldisComplete = true;
  69
  70                         // Decode all the parts and create the key value guard and update sets
  71                         decodeCommitData();
  72
  73                         // Get the sequence number and arbitrator of this transaction
  74                         sequenceNumber = parts->get(0)->getSequenceNumber();
  75                         machineId = parts->get(0)->getMachineId();
  76                         transactionSequenceNumber = parts->get(0)->getTransactionSequenceNumber();
  77                 }
  78         }
  79 }
  80
  81 int64_t Commit::getSequenceNumber() {
  82         return sequenceNumber;
  83 }
  84
  85 int64_t Commit::getTransactionSequenceNumber() {
  86         return transactionSequenceNumber;
  87 }
  88
  89 Vector<CommitPart *> *Commit::getParts() {
  90         return parts;
  91 }
  92
  93 void Commit::addKV(KeyValue *kv) {
  94         keyValueUpdateSet->add(kv);
  95         liveKeys->add(kv->getKey());
  96 }
  97
  98 void Commit::invalidateKey(IoTString *key) {
  99         liveKeys->remove(key);
 100
 101         if (liveKeys->size() == 0) {
 102                 setDead();
 103         }
 104 }
 105
 106 Hashset<KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue> *Commit::getKeyValueUpdateSet() {
 107         return keyValueUpdateSet;
 108 }
 109
 110 int32_t Commit::getNumberOfParts() {
 111         return partCount;
 112 }
 113
 114 void Commit::setDead() {
 115         if (!isDead) {
 116                 isDead = true;
 117                 // Make all the parts of this transaction dead
 118                 for (int32_t partNumber = 0; partNumber < parts->size(); partNumber++) {
 119                         CommitPart *part = parts->get(partNumber);
 120                         if (parts != NULL)
 121                                 part->setDead();
 122                 }
 123         }
 124 }
 125
 126 CommitPart *Commit::getPart(int index) {
 127         return parts->get(index);
 128 }
 129
 130 void Commit::createCommitParts() {
 131         parts->clear();
 132         partCount = 0;
 133         // Convert to chars
 134         Array<char> *charData = convertDataToBytes();
 135
 136         int commitPartCount = 0;
 137         int currentPosition = 0;
 138         int remaining = charData->length();
 139
 140         while (remaining > 0) {
 141                 bool isLastPart = false;
 142                 // determine how much to copy
 143                 int copySize = CommitPart_MAX_NON_HEADER_SIZE;
 144                 if (remaining <= CommitPart_MAX_NON_HEADER_SIZE) {
 145                         copySize = remaining;
 146                         isLastPart = true;// last bit of data so last part
 147                 }
 148
 149                 // Copy to a smaller version
 150                 Array<char> *partData = new Array<char>(copySize);
 151                 System_arraycopy(charData, currentPosition, partData, 0, copySize);
 152
 153                 CommitPart *part = new CommitPart(NULL, machineId, sequenceNumber, transactionSequenceNumber, commitPartCount, partData, isLastPart);
 154                 parts->setExpand(part->getPartNumber(), part);
 155
 156                 // Update position, count and remaining
 157                 currentPosition += copySize;
 158                 commitPartCount++;
 159                 remaining -= copySize;
 160         }
 161 }
 162
 163 void Commit::decodeCommitData() {
 164         // Calculate the size of the data section
 165         int dataSize = 0;
 166         for (int i = 0; i < parts->size(); i++) {
 167                 CommitPart *tp = parts->get(i);
 168                 if (tp != NULL)
 169                         dataSize += tp->getDataSize();
 170         }
 171
 172         Array<char> *combinedData = new Array<char>(dataSize);
 173         int currentPosition = 0;
 174
 175         // Stitch all the data sections together
 176         for (int i = 0; i < parts->size(); i++) {
 177                 CommitPart *tp = parts->get(i);
 178                 if (tp != NULL) {
 179                         System_arraycopy(tp->getData(), 0, combinedData, currentPosition, tp->getDataSize());
 180                         currentPosition += tp->getDataSize();
 181                 }
 182         }
 183
 184         // Decoder Object
 185         ByteBuffer *bbDecode = ByteBuffer_wrap(combinedData);
 186
 187         // Decode how many key value pairs need to be decoded
 188         int numberOfKVUpdates = bbDecode->getInt();
 189
 190         // Decode all the updates key values
 191         for (int i = 0; i < numberOfKVUpdates; i++) {
 192                 KeyValue *kv = (KeyValue *)KeyValue_decode(bbDecode);
 193                 keyValueUpdateSet->add(kv);
 194                 liveKeys->add(kv->getKey());
 195         }
 196 }
 197
 198 Array<char> *Commit::convertDataToBytes() {
 199         // Calculate the size of the data
 200         int sizeOfData = sizeof(int32_t);       // Number of Update KV's
 201         SetIterator<KeyValue *, KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue> *kvit = keyValueUpdateSet->iterator();
 202         while (kvit->hasNext()) {
 203                 KeyValue *kv = kvit->next();
 204                 sizeOfData += kv->getSize();
 205         }
 206         delete kvit;
 207
 208         // Data handlers and storage
 209         Array<char> *dataArray = new Array<char>(sizeOfData);
 210         ByteBuffer *bbEncode = ByteBuffer_wrap(dataArray);
 211
 212         // Encode the size of the updates and guard sets
 213         bbEncode->putInt(keyValueUpdateSet->size());
 214
 215         // Encode all the updates
 216         kvit = keyValueUpdateSet->iterator();
 217         while (kvit->hasNext()) {
 218                 KeyValue *kv = kvit->next();
 219                 kv->encode(bbEncode);
 220         }
 221         delete kvit;
 222
 223         return bbEncode->array();
 224 }
 225
 226 void Commit::setKVsMap(Hashset<KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue> *newKVs) {
 227         keyValueUpdateSet->clear();
 228         keyValueUpdateSet->addAll(newKVs);
 229         liveKeys->clear();
 230         SetIterator<KeyValue *, KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue> *kvit = newKVs->iterator();
 231         while (kvit->hasNext()) {
 232                 liveKeys->add(kvit->next()->getKey());
 233         }
 234         delete kvit;
 235 }
 236
 237 Commit *Commit_merge(Commit *newer, Commit *older, int64_t newSequenceNumber) {
 238         if (older == NULL) {
 239                 return newer;
 240         } else if (newer == NULL) {
 241                 return older;
 242         }
 243         Hashset<KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue> *kvSet = new Hashset<KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue>();
 244         SetIterator<KeyValue *, KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue> *kvit = older->getKeyValueUpdateSet()->iterator();
 245         while (kvit->hasNext()) {
 246                 KeyValue *kv = kvit->next();
 247                 kvSet->add(kv);
 248         }
 249         delete kvit;
 250         kvit = newer->getKeyValueUpdateSet()->iterator();
 251         while (kvit->hasNext()) {
 252                 KeyValue *kv = kvit->next();
 253                 kvSet->add(kv);
 254         }
 255         delete kvit;
 256
 257         int64_t transactionSequenceNumber = newer->getTransactionSequenceNumber();
 258         if (transactionSequenceNumber == -1) {
 259                 transactionSequenceNumber = older->getTransactionSequenceNumber();
 260         }
 261
 262         Commit *newCommit = new Commit(newSequenceNumber, newer->getMachineId(), transactionSequenceNumber);
 263         newCommit->setKVsMap(kvSet);
 264
 265         delete kvSet;
 266         return newCommit;
 267 }