+++ /dev/null
-#include "Commit.h"
-#include "CommitPart.h"
-#include "ByteBuffer.h"
-#include "IoTString.h"
-
-Commit::Commit() :
- parts(new Vector<CommitPart *>()),
- partCount(0),
- missingParts(NULL),
- fldisComplete(false),
- hasLastPart(false),
- keyValueUpdateSet(new Hashset<KeyValue *, uintptr_t, 0>()),
- isDead(false),
- sequenceNumber(-1),
- machineId(-1),
- transactionSequenceNumber(-1),
- dataBytes(NULL),
- liveKeys(new Hashset<IoTString *>()) {
-}
-
-Commit::Commit(int64_t _sequenceNumber, int64_t _machineId, int64_t _transactionSequenceNumber) :
- parts(new Vector<CommitPart *>()),
- partCount(0),
- missingParts(NULL),
- fldisComplete(true),
- hasLastPart(false),
- keyValueUpdateSet(new Hashset<KeyValue *, uintptr_t, 0>()),
- isDead(false),
- sequenceNumber(_sequenceNumber),
- machineId(_machineId),
- transactionSequenceNumber(_transactionSequenceNumber),
- dataBytes(NULL),
- liveKeys(new Hashset<IoTString *>()) {
-}
-
-Commit::~Commit() {
- {
- uint Size = parts->size();
- for(uint i=0;i<Size; i++)
- parts->get(i)->releaseRef();
- delete parts;
- }
- {
- SetIterator<KeyValue *, KeyValue *, uintptr_t, 0> * keyit = keyValueUpdateSet->iterator();
- while(keyit->hasNext()) {
- delete keyit->next();
- }
- delete keyit;
- delete keyValueUpdateSet;
- }
- delete liveKeys;
- if (missingParts != NULL)
- delete missingParts;
- if (dataBytes != NULL)
- delete dataBytes;
-}
-
-void Commit::addPartDecode(CommitPart *newPart) {
- if (isDead) {
- // If dead then just kill this part and move on
- newPart->setDead();
- return;
- }
-
- newPart->acquireRef();
- CommitPart *previouslySeenPart = parts->setExpand(newPart->getPartNumber(), newPart);
- if (previouslySeenPart == NULL)
- partCount++;
-
- if (previouslySeenPart != NULL) {
- // Set dead the old one since the new one is a rescued version of this part
- previouslySeenPart->setDead();
- previouslySeenPart->releaseRef();
- } else if (newPart->isLastPart()) {
- missingParts = new Hashset<int32_t>();
- hasLastPart = true;
-
- for (int i = 0; i < newPart->getPartNumber(); i++) {
- if (parts->get(i) == NULL) {
- missingParts->add(i);
- }
- }
- }
-
- if (!fldisComplete && hasLastPart) {
-
- // We have seen this part so remove it from the set of missing parts
- missingParts->remove(newPart->getPartNumber());
-
- // Check if all the parts have been seen
- if (missingParts->size() == 0) {
-
- // We have all the parts
- fldisComplete = true;
-
- // Decode all the parts and create the key value guard and update sets
- decodeCommitData();
-
- // Get the sequence number and arbitrator of this transaction
- sequenceNumber = parts->get(0)->getSequenceNumber();
- machineId = parts->get(0)->getMachineId();
- transactionSequenceNumber = parts->get(0)->getTransactionSequenceNumber();
- }
- }
-}
-
-int64_t Commit::getSequenceNumber() {
- return sequenceNumber;
-}
-
-int64_t Commit::getTransactionSequenceNumber() {
- return transactionSequenceNumber;
-}
-
-Vector<CommitPart *> *Commit::getParts() {
- return parts;
-}
-
-void Commit::addKV(KeyValue *kv) {
- KeyValue * kvcopy = kv->getCopy();
- keyValueUpdateSet->add(kvcopy);
- liveKeys->add(kvcopy->getKey());
-}
-
-void Commit::invalidateKey(IoTString *key) {
- liveKeys->remove(key);
-
- if (liveKeys->size() == 0) {
- setDead();
- }
-}
-
-Hashset<KeyValue *, uintptr_t, 0> *Commit::getKeyValueUpdateSet() {
- return keyValueUpdateSet;
-}
-
-int32_t Commit::getNumberOfParts() {
- return partCount;
-}
-
-void Commit::setDead() {
- if (!isDead) {
- isDead = true;
- // Make all the parts of this transaction dead
- for (uint32_t partNumber = 0; partNumber < parts->size(); partNumber++) {
- CommitPart *part = parts->get(partNumber);
- part->setDead();
- }
- }
-}
-
-void Commit::createCommitParts() {
- uint Size = parts->size();
- for(uint i=0;i < Size; i++) {
- Entry * e=parts->get(i);
- e->releaseRef();
- }
- parts->clear();
- partCount = 0;
- // Convert to chars
- Array<char> *charData = convertDataToBytes();
-
- int commitPartCount = 0;
- int currentPosition = 0;
- int remaining = charData->length();
-
- while (remaining > 0) {
- bool isLastPart = false;
- // determine how much to copy
- int copySize = CommitPart_MAX_NON_HEADER_SIZE;
- if (remaining <= CommitPart_MAX_NON_HEADER_SIZE) {
- copySize = remaining;
- isLastPart = true;// last bit of data so last part
- }
-
- // Copy to a smaller version
- Array<char> *partData = new Array<char>(copySize);
- System_arraycopy(charData, currentPosition, partData, 0, copySize);
-
- CommitPart *part = new CommitPart(NULL, machineId, sequenceNumber, transactionSequenceNumber, commitPartCount, partData, isLastPart);
- parts->setExpand(part->getPartNumber(), part);
-
- // Update position, count and remaining
- currentPosition += copySize;
- commitPartCount++;
- remaining -= copySize;
- }
- delete charData;
-}
-
-void Commit::decodeCommitData() {
- // Calculate the size of the data section
- int dataSize = 0;
- for (uint i = 0; i < parts->size(); i++) {
- CommitPart *tp = parts->get(i);
- if (tp != NULL)
- dataSize += tp->getDataSize();
- }
-
- Array<char> *combinedData = new Array<char>(dataSize);
- int currentPosition = 0;
-
- // Stitch all the data sections together
- for (uint i = 0; i < parts->size(); i++) {
- CommitPart *tp = parts->get(i);
- if (tp != NULL) {
- System_arraycopy(tp->getData(), 0, combinedData, currentPosition, tp->getDataSize());
- currentPosition += tp->getDataSize();
- }
- }
-
- // Decoder Object
- ByteBuffer *bbDecode = ByteBuffer_wrap(combinedData);
-
- // Decode how many key value pairs need to be decoded
- int numberOfKVUpdates = bbDecode->getInt();
-
- // Decode all the updates key values
- for (int i = 0; i < numberOfKVUpdates; i++) {
- KeyValue *kv = (KeyValue *)KeyValue_decode(bbDecode);
- keyValueUpdateSet->add(kv);
- liveKeys->add(kv->getKey());
- }
- delete bbDecode;
-}
-
-Array<char> *Commit::convertDataToBytes() {
- // Calculate the size of the data
- int sizeOfData = sizeof(int32_t); // Number of Update KV's
- SetIterator<KeyValue *, KeyValue *, uintptr_t, 0> *kvit = keyValueUpdateSet->iterator();
- while (kvit->hasNext()) {
- KeyValue *kv = kvit->next();
- sizeOfData += kv->getSize();
- }
- delete kvit;
-
- // Data handlers and storage
- Array<char> *dataArray = new Array<char>(sizeOfData);
- ByteBuffer *bbEncode = ByteBuffer_wrap(dataArray);
-
- // Encode the size of the updates and guard sets
- bbEncode->putInt(keyValueUpdateSet->size());
-
- // Encode all the updates
- kvit = keyValueUpdateSet->iterator();
- while (kvit->hasNext()) {
- KeyValue *kv = kvit->next();
- kv->encode(bbEncode);
- }
- delete kvit;
- Array<char> * array = bbEncode->array();
- bbEncode->releaseArray();
- delete bbEncode;
- return array;
-}
-
-void Commit::setKVsMap(Hashset<KeyValue *, uintptr_t, 0, hashKeyValue, KeyValueEquals> *newKVs) {
- keyValueUpdateSet->clear();
- liveKeys->clear();
- SetIterator<KeyValue *, KeyValue *, uintptr_t, 0, hashKeyValue, KeyValueEquals> *kvit = newKVs->iterator();
- while (kvit->hasNext()) {
- KeyValue *kv = kvit->next();
- KeyValue *kvcopy = kv->getCopy();
- liveKeys->add(kvcopy->getKey());
- keyValueUpdateSet->add(kvcopy);
- }
- delete kvit;
-}
-
-Commit *Commit_merge(Commit *newer, Commit *older, int64_t newSequenceNumber) {
- if (older == NULL) {
- return newer;
- } else if (newer == NULL) {
- return older;
- }
- Hashset<KeyValue *, uintptr_t, 0, hashKeyValue, KeyValueEquals> *kvSet = new Hashset<KeyValue *, uintptr_t, 0, hashKeyValue, KeyValueEquals>();
- SetIterator<KeyValue *, KeyValue *, uintptr_t, 0> *kvit = older->getKeyValueUpdateSet()->iterator();
- while (kvit->hasNext()) {
- KeyValue *kv = kvit->next();
- kvSet->add(kv);
- }
- delete kvit;
- kvit = newer->getKeyValueUpdateSet()->iterator();
- while (kvit->hasNext()) {
- KeyValue *kv = kvit->next();
- kvSet->add(kv);
- }
- delete kvit;
-
- int64_t transactionSequenceNumber = newer->getTransactionSequenceNumber();
- if (transactionSequenceNumber == -1) {
- transactionSequenceNumber = older->getTransactionSequenceNumber();
- }
-
- Commit *newCommit = new Commit(newSequenceNumber, newer->getMachineId(), transactionSequenceNumber);
- newCommit->setKVsMap(kvSet);
-
- delete kvSet;
- return newCommit;
-}
projects / iotcloud.git / blobdiff