updateLiveTransactionsAndStatus();
return true;
} catch (Exception *e) {
- SetIterator<int64_t> *kit = getKeyIterator(localCommunicationTable);
+ SetIterator<int64_t, Pair<IoTString *, int32_t> *> *kit = getKeyIterator(localCommunicationTable);
while (kit->hasNext()) {
int64_t m = kit->next();
updateFromLocal(m);
}
}
- SetIterator<Transaction *> *trit = getKeyIterator(lastTransactionPartsSent);
+ SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
while (trit->hasNext()) {
Transaction *transaction = trit->next();
transaction->resetServerFailure();
}
}
- SetIterator<Transaction *> *trit = getKeyIterator(lastTransactionPartsSent);
+ SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
while (trit->hasNext()) {
Transaction *transaction = trit->next();
transaction->resetServerFailure();
}
}
- SetIterator<Transaction *> *trit = getKeyIterator(lastTransactionPartsSent);
+ SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
while (trit->hasNext()) {
Transaction *transaction = trit->next();
transaction->resetServerFailure();
}
}
- SetIterator<Transaction *> *trit = getKeyIterator(lastTransactionPartsSent);
+ SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
while (trit->hasNext()) {
Transaction *transaction = trit->next();
transaction->resetServerFailure();
}
delete trit;
} else {
- SetIterator<Transaction *> *trit = getKeyIterator(lastTransactionPartsSent);
+ SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
while (trit->hasNext()) {
Transaction *transaction = trit->next();
transaction->resetServerFailure();
if (needsResize) {
// Reset which transaction to send
- SetIterator<Transaction *> *trit = getKeyIterator(transactionPartsSent);
+ SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(transactionPartsSent);
while (trit->hasNext()) {
Transaction *transaction = trit->next();
transaction->resetNextPartToSend();
}
pendingSendArbitrationRounds->setSize(oldcount);
- SetIterator<Transaction *> *trit = getKeyIterator(transactionPartsSent);
+ SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(transactionPartsSent);
while (trit->hasNext()) {
Transaction *transaction = trit->next();
transaction->resetServerFailure();
delete trit;
} else {
// Reset which transaction to send
- SetIterator<Transaction *> *trit = getKeyIterator(transactionPartsSent);
+ SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(transactionPartsSent);
while (trit->hasNext()) {
Transaction *transaction = trit->next();
transaction->resetNextPartToSend();
}
} catch (ServerException *e) {
- if (e->getType() != ServerException->TypeInputTimeout) {
+ if (e->getType() != ServerException_TypeInputTimeout) {
// Nothing was able to be sent to the server so just clear these data structures
- SetIterator<Transaction *> *trit = getKeyIterator(transactionPartsSent);
+ SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(transactionPartsSent);
while (trit->hasNext()) {
Transaction *transaction = trit->next();
transaction->resetNextPartToSend();
hadPartialSendToServer = true;
// Nothing was able to be sent to the server so just clear these data structures
- SetIterator<Transaction *> *trit = getKeyIterator(transactionPartsSent);
+ SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(transactionPartsSent);
while (trit->hasNext()) {
Transaction *transaction = trit->next();
transaction->resetNextPartToSend();
int sendDataSize = sizeof(int32_t) + sizeof(int64_t);
int64_t lastArbitrationDataLocalSequenceNumber = (int64_t) -1;
- if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->get(machineId) != NULL) {
+ if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->contains(machineId)) {
lastArbitrationDataLocalSequenceNumber = lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->get(machineId);
}
bbEncode->putInt(0);
// Send by local
- Array<char> *returnData = cloud->sendLocalData(sendData, localSequenceNumber, localCommunicationInformation.getFirst(), localCommunicationInformation.getSecond());
+ Array<char> *returnData = cloud->sendLocalData(sendData, localSequenceNumber, localCommunicationInformation->getFirst(), localCommunicationInformation->getSecond());
localSequenceNumber++;
if (returnData == NULL) {
Pair<bool, bool> Table::sendTransactionToLocal(Transaction *transaction) {
// Get the devices local communications
- if (!localCommunicationTable->contains(machineId))
+ if (!localCommunicationTable->contains(transaction->getArbitrator()))
return Pair<bool, bool>(true, false);
- Pair<IoTString *, int32_t> localCommunicationInformation = localCommunicationTable->get(transaction->getArbitrator());
+ Pair<IoTString *, int32_t> * localCommunicationInformation = localCommunicationTable->get(transaction->getArbitrator());
// Get the size of the send data
int sendDataSize = sizeof(int32_t) + sizeof(int64_t);
- for (TransactionPart *part : transaction->getParts()->values()) {
- sendDataSize += part->getSize();
+ {
+ Vector<TransactionPart *> * tParts = transaction->getParts();
+ uint tPartsSize = tParts->size();
+ for (uint i = 0; i < tPartsSize; i++) {
+ TransactionPart * part = tParts->get(i);
+ sendDataSize += part->getSize();
+ }
}
int64_t lastArbitrationDataLocalSequenceNumber = (int64_t) -1;
- if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->get(transaction->getArbitrator()) != NULL) {
+ if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->contains(transaction->getArbitrator())) {
lastArbitrationDataLocalSequenceNumber = lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->get(transaction->getArbitrator());
}
// Encode the data
bbEncode->putLong(lastArbitrationDataLocalSequenceNumber);
bbEncode->putInt(transaction->getParts()->size());
- for (TransactionPart *part : transaction->getParts()->values()) {
- part->encode(bbEncode);
+ {
+ Vector<TransactionPart *> * tParts = transaction->getParts();
+ uint tPartsSize = tParts->size();
+ for (uint i = 0; i < tPartsSize; i++) {
+ TransactionPart * part = tParts->get(i);
+ part->encode(bbEncode);
+ }
}
-
// Send by local
- Array<char> *returnData = cloud->sendLocalData(sendData, localSequenceNumber, localCommunicationInformation.getFirst(), localCommunicationInformation.getSecond());
+ Array<char> *returnData = cloud->sendLocalData(sendData, localSequenceNumber, localCommunicationInformation->getFirst(), localCommunicationInformation->getSecond());
localSequenceNumber++;
if (returnData == NULL) {
// Transaction was sent to the server so keep track of it to prevent double commit
if (transaction->getSequenceNumber() != -1) {
- offlineTransactionsCommittedAndAtServer->add(transaction->getId());
+ offlineTransactionsCommittedAndAtServer->add(new Pair<int64_t, int64_t>(transaction->getId()));
}
}
* there is already a sufficient entry in the queue (e->g->,
* equalsto value of true and same sequence number)-> */
- int64_t old_seqn = rejectedSlotVector->firstElement();
+ int64_t old_seqn = rejectedSlotVector->get(0);
if (rejectedSlotVector->size() > Table_REJECTED_THRESHOLD) {
int64_t new_seqn = rejectedSlotVector->lastElement();
RejectedMessage *rm = new RejectedMessage(s, s->getSequenceNumber(), localMachineId, old_seqn, new_seqn, false);
// Guard evaluated as true
// Update the local changes so we can make the commit
- SetIterator<KeyValue *> *kvit = getKeyIterator(transaction->getKeyValueUpdateSet());
+ SetIterator<KeyValue *, KeyValue *> *kvit = transaction->getKeyValueUpdateSet()->iterator();
while (kvit->hasNext()) {
KeyValue *kv = kvit->next();
speculativeTableTmp->put(kv->getKey(), kv);
localArbitrationSequenceNumber++;
// Update the local changes so we can make the commit
- SetIterator<KeyValue *> *kvit = getKeyIterator(transaction->getKeyValueUpdateSet());
+ SetIterator<KeyValue *, KeyValue *> *kvit = transaction->getKeyValueUpdateSet()->iterator();
while (kvit->hasNext()) {
KeyValue *kv = kvit->next();
newCommit->addKV(kv);
// Get what commits should be edited, these are the commits that
// have live values for their keys
Hashset<Commit *> *commitsToEdit = new Hashset<Commit *>();
- SetIterator<KeyValue *> *kvit = getKeyIterator(commit->getKeyValueUpdateSet());
- while (kvit->hasNext()) {
- KeyValue *kv = kvit->next();
- commitsToEdit->add(liveCommitsByKeyTable->get(kv->getKey()));
+ {
+ SetIterator<KeyValue *, KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue> *kvit = commit->getKeyValueUpdateSet()->iterator();
+ while (kvit->hasNext()) {
+ KeyValue *kv = kvit->next();
+ commitsToEdit->add(liveCommitsByKeyTable->get(kv->getKey()));
+ }
+ delete kvit;
}
- delete kvit;
commitsToEdit->remove(NULL); // remove NULL since it could be in this set
// Update each previous commit that needs to be updated
if (previousCommit->isLive()) {
// Update which keys in the old commits are still live
- SetIterator<KeyValue *> *kvit = getKeyIterator(commit->getKeyValueUpdateSet());
- while (kvit->hasNext()) {
- KeyValue *kv = kvit->next();
- previousCommit->invalidateKey(kv->getKey());
+ {
+ SetIterator<KeyValue *, KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue> *kvit = commit->getKeyValueUpdateSet()->iterator();
+ while (kvit->hasNext()) {
+ KeyValue *kv = kvit->next();
+ previousCommit->invalidateKey(kv->getKey());
+ }
+ delete kvit;
}
- delete kvit;
// if the commit is now dead then remove it
if (!previousCommit->isLive()) {
didProcessANewCommit = true;
// Update the committed table of keys and which commit is using which key
- SetIterator<KeyValue *> *kvit = getKeyIterator(commit->getKeyValueUpdateSet());
- while (kvit->hasNext()) {
- KeyValue *kv = kvit->next();
- committedKeyValueTable->put(kv->getKey(), kv);
- liveCommitsByKeyTable->put(kv->getKey(), commit);
+ {
+ SetIterator<KeyValue *, KeyValue *, uintptr_t, 0, hashKeyValue, equalsKeyValue> *kvit = commit->getKeyValueUpdateSet()->iterator();
+ while (kvit->hasNext()) {
+ KeyValue *kv = kvit->next();
+ committedKeyValueTable->put(kv->getKey(), kv);
+ liveCommitsByKeyTable->put(kv->getKey(), commit);
+ }
+ delete kvit;
}
- delete kvit;
}
}
if (transaction->evaluateGuard(committedKeyValueTable, speculatedKeyValueTable, NULL)) {
// Guard evaluated to true so update the speculative table
- SetIterator<KeyValue *> *kvit = getKeyIterator(commit->getKeyValueUpdateSet());
- while (kvit->hasNext()) {
- KeyValue *kv = kvit->next();
- speculatedKeyValueTable->put(kv->getKey(), kv);
+ {
+ SetIterator<KeyValue *, KeyValue *> *kvit = transaction->getKeyValueUpdateSet()->iterator();
+ while (kvit->hasNext()) {
+ KeyValue *kv = kvit->next();
+ speculatedKeyValueTable->put(kv->getKey(), kv);
+ }
+ delete kvit;
}
- delete kvit;
}
}
if (transaction->evaluateGuard(committedKeyValueTable, speculatedKeyValueTable, pendingTransactionSpeculatedKeyValueTable)) {
// Guard evaluated to true so update the speculative table
- SetIterator<KeyValue *> *kvit = getKeyIterator(commit->getKeyValueUpdateSet());
+ SetIterator<KeyValue *, KeyValue *> *kvit = transaction->getKeyValueUpdateSet()->iterator();
while (kvit->hasNext()) {
KeyValue *kv = kvit->next();
pendingTransactionSpeculatedKeyValueTable->put(kv->getKey(), kv);
if ((entry->getSequenceNumber() != -1) && (lastMessageTable->get(entry->getTransactionMachineId()).getFirst() >= entry->getSequenceNumber())) {
// The machine already saw this so it is dead
entry->setDead();
- liveAbortTable->remove(entry->getAbortId());
+ liveAbortTable->remove(&entry->getAbortId());
if (entry->getTransactionArbitrator() == localMachineId) {
liveAbortsGeneratedByLocal->remove(entry->getArbitratorLocalSequenceNumber());
}
// Update the last arbitration data that we have seen so far
- if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->get(entry->getTransactionArbitrator()) != NULL) {
+ if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->contains(entry->getTransactionArbitrator())) {
int64_t lastArbitrationSequenceNumber = lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->get(entry->getTransactionArbitrator());
if (entry->getSequenceNumber() > lastArbitrationSequenceNumber) {
// Is larger
* our own last message or that other clients have not had a rollback
* on the last message->
*/
-void Table::updateLastMessage(int64_t machineId, int64_t seqNum, Liveness *liveness, bool acceptUpdatesToLocal, Hashset<intr64_t> *machineSet) {
+void Table::updateLastMessage(int64_t machineId, int64_t seqNum, Liveness *liveness, bool acceptUpdatesToLocal, Hashset<int64_t> *machineSet) {
// We have seen this machine ID
machineSet->remove(machineId);
if (watchset != NULL) {
// Go through each rejected message that this machine Id has not
// seen yet
- for (Iterator<RejectedMessage *> *rmit = watchset->iterator(); rmit->hasNext(); ) {
+
+ SetIterator<RejectedMessage *, RejectedMessage *> *rmit = watchset->iterator();
+ while(rmit->hasNext()) {
RejectedMessage *rm = rmit->next();
// If this machine Id has seen this rejected message->->->
if (rm->getSequenceNumber() <= seqNum) {
rm->removeWatcher(machineId);
}
}
+ delete rmit;
}
// Set dead the abort
}
if (machineId == localMachineId) {
// Our own messages are immediately dead->
- if (liveness instanceof LastMessage) {
+ char livenessType = liveness->getType();
+ if (livenessType==TypeLastMessage) {
((LastMessage *)liveness)->setDead();
- } else if (liveness instanceof Slot) {
+ } else if (livenessType == TypeSlot) {
((Slot *)liveness)->setDead();
} else {
throw new Error("Unrecognized type");
}
}
// Get the old last message for this device
- Pair<int64_t, Liveness *> lastMessageEntry = lastMessageTable->put(machineId, Pair<int64_t, Liveness *>(seqNum, liveness));
+ Pair<int64_t, Liveness *> * lastMessageEntry = lastMessageTable->put(machineId, new Pair<int64_t, Liveness *>(seqNum, liveness));
if (lastMessageEntry == NULL) {
// If no last message then there is nothing else to process
return;
}
- int64_t lastMessageSeqNum = lastMessageEntry.getFirst();
- Liveness *lastEntry = lastMessageEntry.getSecond();
-
+ int64_t lastMessageSeqNum = lastMessageEntry->getFirst();
+ Liveness *lastEntry = lastMessageEntry->getSecond();
+ delete lastMessageEntry;
+
// If it is not our machine Id since we already set ours to dead
if (machineId != localMachineId) {
- if (lastEntry instanceof LastMessage) {
+ char lastEntryType = lastEntry->getType();
+
+ if (lastEntryType == TypeLastMessage) {
((LastMessage *)lastEntry)->setDead();
- } else if (lastEntry instanceof Slot) {
+ } else if (lastEntryType == TypeSlot) {
((Slot *)lastEntry)->setDead();
} else {
throw new Error("Unrecognized type");