--- /dev/null
+#include "Transaction.h"
+#include "TransactionPart.h"
+#include "KeyValue.h"
+#include "ByteBuffer.h"
+#include "IoTString.h"
+#include "TransactionStatus.h"
+
+Transaction::Transaction() :
+ parts(new Vector<TransactionPart *>()),
+ partCount(0),
+ missingParts(NULL),
+ partsPendingSend(new Vector<int32_t>()),
+ fldisComplete(false),
+ hasLastPart(false),
+ keyValueGuardSet(new Hashset<KeyValue *>()),
+ keyValueUpdateSet(new Hashset<KeyValue *>()),
+ isDead(false),
+ sequenceNumber(-1),
+ clientLocalSequenceNumber(-1),
+ arbitratorId(-1),
+ machineId(-1),
+ transactionId(Pair<int64_t, int64_t>(0,0)),
+ nextPartToSend(0),
+ flddidSendAPartToServer(false),
+ transactionStatus(NULL),
+ hadServerFailure(false) {
+}
+
+Transaction::~Transaction() {
+ if (missingParts)
+ delete missingParts;
+ {
+ uint Size = parts->size();
+ for(uint i=0; i<Size; i++)
+ parts->get(i)->releaseRef();
+ delete parts;
+ }
+ {
+ SetIterator<KeyValue *, KeyValue *> *kvit = keyValueGuardSet->iterator();
+ while (kvit->hasNext()) {
+ KeyValue *kvGuard = kvit->next();
+ delete kvGuard;
+ }
+ delete kvit;
+ delete keyValueGuardSet;
+ }
+ {
+ SetIterator<KeyValue *, KeyValue *> *kvit = keyValueUpdateSet->iterator();
+ while (kvit->hasNext()) {
+ KeyValue *kvUpdate = kvit->next();
+ delete kvUpdate;
+ }
+ delete kvit;
+ delete keyValueUpdateSet;
+ }
+ delete partsPendingSend;
+}
+
+void Transaction::addPartEncode(TransactionPart *newPart) {
+ newPart->acquireRef();
+ printf("Add part %d\n", newPart->getPartNumber());
+ TransactionPart *old = parts->setExpand(newPart->getPartNumber(), newPart);
+ if (old == NULL) {
+ partCount++;
+ } else {
+ old->releaseRef();
+ }
+ partsPendingSend->add(newPart->getPartNumber());
+
+ sequenceNumber = newPart->getSequenceNumber();
+ arbitratorId = newPart->getArbitratorId();
+ transactionId = newPart->getTransactionId();
+ clientLocalSequenceNumber = newPart->getClientLocalSequenceNumber();
+ machineId = newPart->getMachineId();
+ fldisComplete = true;
+}
+
+void Transaction::addPartDecode(TransactionPart *newPart) {
+ if (isDead) {
+ // If dead then just kill this part and move on
+ newPart->setDead();
+ return;
+ }
+ newPart->acquireRef();
+ sequenceNumber = newPart->getSequenceNumber();
+ arbitratorId = newPart->getArbitratorId();
+ transactionId = newPart->getTransactionId();
+ clientLocalSequenceNumber = newPart->getClientLocalSequenceNumber();
+ machineId = newPart->getMachineId();
+
+ TransactionPart *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->releaseRef();
+ previouslySeenPart->setDead();
+ } 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
+ decodeTransactionData();
+ }
+ }
+}
+
+void Transaction::addUpdateKV(KeyValue *kv) {
+ keyValueUpdateSet->add(kv);
+}
+
+void Transaction::addGuardKV(KeyValue *kv) {
+ keyValueGuardSet->add(kv);
+}
+
+
+int64_t Transaction::getSequenceNumber() {
+ return sequenceNumber;
+}
+
+void Transaction::setSequenceNumber(int64_t _sequenceNumber) {
+ sequenceNumber = _sequenceNumber;
+
+ for (uint32_t i = 0; i < parts->size(); i++) {
+ TransactionPart *tp = parts->get(i);
+ if (tp != NULL)
+ tp->setSequenceNumber(sequenceNumber);
+ }
+}
+
+int64_t Transaction::getClientLocalSequenceNumber() {
+ return clientLocalSequenceNumber;
+}
+
+Vector<TransactionPart *> *Transaction::getParts() {
+ return parts;
+}
+
+bool Transaction::didSendAPartToServer() {
+ return flddidSendAPartToServer;
+}
+
+void Transaction::resetNextPartToSend() {
+ nextPartToSend = 0;
+}
+
+TransactionPart *Transaction::getNextPartToSend() {
+ if ((partsPendingSend->size() == 0) || (partsPendingSend->size() == nextPartToSend)) {
+ return NULL;
+ }
+ TransactionPart *part = parts->get(partsPendingSend->get(nextPartToSend));
+ nextPartToSend++;
+ return part;
+}
+
+
+void Transaction::setServerFailure() {
+ hadServerFailure = true;
+}
+
+bool Transaction::getServerFailure() {
+ return hadServerFailure;
+}
+
+
+void Transaction::resetServerFailure() {
+ hadServerFailure = false;
+}
+
+
+void Transaction::setTransactionStatus(TransactionStatus *_transactionStatus) {
+ transactionStatus = _transactionStatus;
+}
+
+TransactionStatus *Transaction::getTransactionStatus() {
+ return transactionStatus;
+}
+
+void Transaction::removeSentParts(Vector<int32_t> *sentParts) {
+ nextPartToSend = 0;
+ bool changed = false;
+ uint lastusedindex = 0;
+ for (uint i = 0; i < partsPendingSend->size(); i++) {
+ int32_t parti = partsPendingSend->get(i);
+ for (uint j = 0; j < sentParts->size(); j++) {
+ int32_t partj = sentParts->get(j);
+ if (parti == partj) {
+ changed = true;
+ goto NextElement;
+ }
+ }
+ partsPendingSend->set(lastusedindex++, parti);
+NextElement:
+ ;
+ }
+ if (changed) {
+ partsPendingSend->setSize(lastusedindex);
+ flddidSendAPartToServer = true;
+ transactionStatus->setTransactionSequenceNumber(sequenceNumber);
+ }
+}
+
+bool Transaction::didSendAllParts() {
+ return partsPendingSend->isEmpty();
+}
+
+Hashset<KeyValue *> *Transaction::getKeyValueUpdateSet() {
+ return keyValueUpdateSet;
+}
+
+int Transaction::getNumberOfParts() {
+ return partCount;
+}
+
+int64_t Transaction::getMachineId() {
+ return machineId;
+}
+
+int64_t Transaction::getArbitrator() {
+ return arbitratorId;
+}
+
+bool Transaction::isComplete() {
+ return fldisComplete;
+}
+
+Pair<int64_t, int64_t> *Transaction::getId() {
+ return &transactionId;
+}
+
+void Transaction::setDead() {
+ if (!isDead) {
+ // Set dead
+ isDead = true;
+ // Make all the parts of this transaction dead
+ for (uint32_t partNumber = 0; partNumber < parts->size(); partNumber++) {
+ TransactionPart *part = parts->get(partNumber);
+ if (part != NULL)
+ part->setDead();
+ }
+ }
+}
+
+void Transaction::decodeTransactionData() {
+ // Calculate the size of the data section
+ int dataSize = 0;
+ for (uint i = 0; i < parts->size(); i++) {
+ TransactionPart *tp = parts->get(i);
+ 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++) {
+ TransactionPart *tp = parts->get(i);
+ 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 numberOfKVGuards = bbDecode->getInt();
+ int numberOfKVUpdates = bbDecode->getInt();
+
+ // Decode all the guard key values
+ for (int i = 0; i < numberOfKVGuards; i++) {
+ KeyValue *kv = (KeyValue *)KeyValue_decode(bbDecode);
+ keyValueGuardSet->add(kv);
+ }
+
+ // Decode all the updates key values
+ for (int i = 0; i < numberOfKVUpdates; i++) {
+ KeyValue *kv = (KeyValue *)KeyValue_decode(bbDecode);
+ keyValueUpdateSet->add(kv);
+ }
+ delete bbDecode;
+}
+
+bool Transaction::evaluateGuard(Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals> *committedKeyValueTable, Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals> *speculatedKeyValueTable, Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals> *pendingTransactionSpeculatedKeyValueTable) {
+ SetIterator<KeyValue *, KeyValue *> *kvit = keyValueGuardSet->iterator();
+ while (kvit->hasNext()) {
+ KeyValue *kvGuard = kvit->next();
+ // First check if the key is in the speculative table, this is the value of the latest assumption
+ KeyValue *kv = NULL;
+
+ // If we have a speculation table then use it first
+ if (pendingTransactionSpeculatedKeyValueTable != NULL) {
+ kv = pendingTransactionSpeculatedKeyValueTable->get(kvGuard->getKey());
+ }
+
+ // If we have a speculation table then use it first
+ if ((kv == NULL) && (speculatedKeyValueTable != NULL)) {
+ kv = speculatedKeyValueTable->get(kvGuard->getKey());
+ }
+
+ if (kv == NULL) {
+ // if it is not in the speculative table then check the committed table and use that
+ // value as our latest assumption
+ kv = committedKeyValueTable->get(kvGuard->getKey());
+ }
+
+ if (kvGuard->getValue() != NULL) {
+ if ((kv == NULL) || (!kvGuard->getValue()->equals(kv->getValue()))) {
+
+
+ if (kv != NULL) {
+ printf("%s %s\n", kvGuard->getKey()->internalBytes()->internalArray(), kv->getValue()->internalBytes()->internalArray());
+ } else {
+ printf("%s null\n", kvGuard->getValue()->internalBytes()->internalArray());
+ }
+ delete kvit;
+ return false;
+ }
+ } else {
+ if (kv != NULL) {
+ delete kvit;
+ return false;
+ }
+ }
+ }
+ delete kvit;
+ return true;
+}
+
projects / iotcloud.git / blobdiff