speculatedKeyValueTable = new Hashtable<IoTString *, KeyValue *>();
pendingTransactionSpeculatedKeyValueTable = new Hashtable<IoTString *, KeyValue *>();
liveNewKeyTable = new Hashtable<IoTString *, NewKey *>();
- lastMessageTable = new Hashtable<int64_t, Pair<int64_t, Liveness *> *>();
+ lastMessageTable = new Hashtable<int64_t, Pair<int64_t, Liveness *> >();
rejectedMessageWatchVectorTable = new Hashtable<int64_t, Hashset<RejectedMessage *> * >();
arbitratorTable = new Hashtable<IoTString *, int64_t>();
- liveAbortTable = new Hashtable<Pair<int64_t, int64_t> *, Abort *>();
- newTransactionParts = new Hashtable<int64_t, Hashtable<Pair<int64_t, int32_t> *, TransactionPart *> *>();
- newCommitParts = new Hashtable<int64_t, Hashtable<Pair<int64_t, int32_t> *, CommitPart *> *>();
+ liveAbortTable = new Hashtable<Pair<int64_t, int64_t>, Abort *>();
+ newTransactionParts = new Hashtable<int64_t, Hashtable<Pair<int64_t, int32_t>, TransactionPart *> *>();
+ newCommitParts = new Hashtable<int64_t, Hashtable<Pair<int64_t, int32_t>, CommitPart *> *>();
lastArbitratedTransactionNumberByArbitratorTable = new Hashtable<int64_t, int64_t>();
liveTransactionBySequenceNumberTable = new Hashtable<int64_t, Transaction *>();
- liveTransactionByTransactionIdTable = new Hashtable<Pair<int64_t, int64_t> *, Transaction *>();
+ liveTransactionByTransactionIdTable = new Hashtable<Pair<int64_t, int64_t>, Transaction *>();
liveCommitsTable = new Hashtable<int64_t, Hashtable<int64_t, Commit *> >();
liveCommitsByKeyTable = new Hashtable<IoTString *, Commit *>();
lastCommitSeenSequenceNumberByArbitratorTable = new Hashtable<int64_t, int64_t>();
transactionPartsSent = new Hashtable<Transaction *, Vector<int32_t> *>();
outstandingTransactionStatus = new Hashtable<int64_t, TransactionStatus *>();
liveAbortsGeneratedByLocal = new Hashtable<int64_t, Abort *>();
- offlineTransactionsCommittedAndAtServer = new Hashset<Pair<int64_t, int64_t> *>();
+ offlineTransactionsCommittedAndAtServer = new Hashset<Pair<int64_t, int64_t> >();
localCommunicationTable = new Hashtable<int64_t, Pair<IoTString *, int32_t> >();
lastTransactionSeenFromMachineFromServer = new Hashtable<int64_t, int64_t>();
pendingSendArbitrationRounds = new Vector<ArbitrationRound *>();
continue;
}
- Pair<bool, bool> *sendReturn = sendTransactionToLocal(transaction);
+ Pair<bool, bool> sendReturn = sendTransactionToLocal(transaction);
if (sendReturn->getFirst()) {
// Failed to contact over local
Array<Slot *> *newSlots = cloud->getSlots(sequenceNumber + 1);
if (newSlots->length() == 0) {
fromRetry = true;
- ThreeTuple<bool, bool, Array<Slot *> *> *sendSlotsReturn = sendSlotsToServer(lastSlotAttemptedToSend, lastNewSize, lastIsNewKey);
+ ThreeTuple<bool, bool, Array<Slot *> *> sendSlotsReturn = sendSlotsToServer(lastSlotAttemptedToSend, lastNewSize, lastIsNewKey);
if (sendSlotsReturn->getFirst()) {
if (newKey != NULL) {
if (localCommunicationInformation == NULL) {
// Cant talk to that device locally so do nothing
- return new Pair<bool, bool>(true, false);
+ return Pair<bool, bool>(true, false);
}
// Get the size of the send data
if (returnData == NULL) {
// Could not contact server
- return new Pair<bool, bool>(true, false);
+ return Pair<bool, bool>(true, false);
}
// Decode the data
}
}
- return new Pair<bool, bool>(false, true);
+ return Pair<bool, bool>(false, true);
}
Array<char> *Table::acceptDataFromLocal(Array<char> *data) {
return returnData;
}
-ThreeTuple<bool, bool, Array<Slot *> *> *Table::sendSlotsToServer(Slot *slot, int newSize, bool isNewKey) {
+ThreeTuple<bool, bool, Array<Slot *> *> Table::sendSlotsToServer(Slot *slot, int newSize, bool isNewKey) {
bool attemptedToSendToServerTmp = attemptedToSendToServer;
attemptedToSendToServer = true;
}
}
- return new ThreeTuple<bool, bool, Array<Slot *> *>(inserted, lastTryInserted, array);
+ return ThreeTuple<bool, bool, Array<Slot *> *>(inserted, lastTryInserted, array);
}
/**
* Returns false if a resize was needed
*/
-ThreeTuple<bool, int32_t, bool> *Table::fillSlot(Slot *slot, bool resize, NewKey *newKeyEntry) {
+ThreeTuple<bool, int32_t, bool> Table::fillSlot(Slot *slot, bool resize, NewKey *newKeyEntry) {
int newSize = 0;
if (needsResize && !resize) {
// We need to resize but we are not resizing so return false
- return new ThreeTuple<bool, int32_t, bool>(true, NULL, NULL);
+ return ThreeTuple<bool, int32_t, bool>(true, NULL, NULL);
}
bool inserted = false;
// Fill the remainder of the slot with rescue data
doOptionalRescue(slot, seenLiveSlot, currentRescueSequenceNumber, resize);
- return new ThreeTuple<bool, int32_t, bool>(false, newSize, inserted);
+ return ThreeTuple<bool, int32_t, bool>(false, newSize, inserted);
}
void Table::doRejectedMessages(Slot *s) {
} else if (currentSequenceNumber == firstIfFull) {
//if there's no space but the entry is about to fall off the queue
System->out->println("B"); //?
- return new ThreeTuple<bool, bool, int64_t>(true, seenLiveSlot, currentSequenceNumber);
+ return ThreeTuple<bool, bool, int64_t>(true, seenLiveSlot, currentSequenceNumber);
}
}
}
// Did not resize
- return new ThreeTuple<bool, bool, int64_t>(false, seenLiveSlot, currentSequenceNumber);
+ return ThreeTuple<bool, bool, int64_t>(false, seenLiveSlot, currentSequenceNumber);
}
void Table::doOptionalRescue(Slot *s, bool seenliveslot, int64_t seqn, bool resize) {
// Iterate through all the machine Ids that we received new parts for
for (int64_t machineId : newTransactionParts->keySet()) {
- Hashtable<Pair<int64_t int32_t> *, TransactionPart *> *parts = newTransactionParts->get(machineId);
+ Hashtable<Pair<int64_t int32_t>, TransactionPart *> *parts = newTransactionParts->get(machineId);
// Iterate through all the parts for that machine Id
- for (Pair<int64_t, int32_t> *partId : parts->keySet()) {
+ for (Pair<int64_t, int32_t> partId : parts->keySet()) {
TransactionPart *part = parts->get(partId);
int64_t lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable->get(part->getArbitratorId());
// Check if this machine arbitrates for this transaction if not then we cant arbitrate this transaction
if (transaction->getArbitrator() != localMachineId) {
- return new Pair<bool, bool>(false, false);
+ return Pair<bool, bool>(false, false);
}
if (!transaction->isComplete()) {
// Will arbitrate in incorrect order if we continue so just break
// Most likely this
- return new Pair<bool, bool>(false, false);
+ return Pair<bool, bool>(false, false);
}
if (transaction->getMachineId() != localMachineId) {
if (lastTransactionSeenFromMachineFromServer->get(transaction->getMachineId()) != NULL) {
if (lastTransactionSeenFromMachineFromServer->get(transaction->getMachineId()) > transaction->getClientLocalSequenceNumber()) {
// We've have already seen this from the server
- return new Pair<bool, bool>(false, false);
+ return Pair<bool, bool>(false, false);
}
}
}
}
updateLiveStateFromLocal();
- return new Pair<bool, bool>(true, true);
+ return Pair<bool, bool>(true, true);
} else {
if (transaction->getMachineId() == localMachineId) {
}
updateLiveStateFromLocal();
- return new Pair<bool, bool>(true, false);
+ return Pair<bool, bool>(true, false);
}
}
// Iterate through all the machine Ids that we received new parts for
for (int64_t machineId : newCommitParts->keySet()) {
- Hashtable<Pair<int64_t, int32_t> *, CommitPart *> *parts = newCommitParts->get(machineId);
+ Hashtable<Pair<int64_t, int32_t>, CommitPart *> *parts = newCommitParts->get(machineId);
// Iterate through all the parts for that machine Id
- for (Pair<int64_t, int32_t> *partId : parts->keySet()) {
+ for (Pair<int64_t, int32_t> partId : parts->keySet()) {
CommitPart *part = parts->get(partId);
// Get the transaction object for that sequence number
// Create a list of clients to watch until they see this rejected message entry->
Hashset<int64_t> *deviceWatchSet = new Hashset<int64_t>();
- for (Map->Entry<int64_t, Pair<int64_t, Liveness *> *> *lastMessageEntry : lastMessageTable->entrySet()) {
+ for (Map->Entry<int64_t, Pair<int64_t, Liveness *> > *lastMessageEntry : lastMessageTable->entrySet()) {
// Machine ID for the last message entry
int64_t lastMessageEntryMachineId = lastMessageEntry->getKey();
continue;
}
- Pair<int64_t, Liveness *> *lastMessageValue = lastMessageEntry->getValue();
+ Pair<int64_t, Liveness *> lastMessageValue = lastMessageEntry->getValue();
int64_t entrySequenceNumber = lastMessageValue->getFirst();
if (entrySequenceNumber < seq) {
// Set dead a transaction if we can
- Transaction *transactionToSetDead = liveTransactionByTransactionIdTable->remove(new Pair<int64_t, int64_t>(entry->getTransactionMachineId(), entry->getTransactionClientLocalSequenceNumber()));
+ Transaction *transactionToSetDead = liveTransactionByTransactionIdTable->remove(Pair<int64_t, int64_t>(entry->getTransactionMachineId(), entry->getTransactionClientLocalSequenceNumber()));
if (transactionToSetDead != NULL) {
liveTransactionBySequenceNumberTable->remove(transactionToSetDead->getSequenceNumber());
}
}
// This part is still alive
- Hashtable<Pair<int64_t, int32_t> *, TransactionPart *> *transactionPart = newTransactionParts->get(entry->getMachineId());
+ Hashtable<Pair<int64_t, int32_t>, TransactionPart *> *transactionPart = newTransactionParts->get(entry->getMachineId());
if (transactionPart == NULL) {
// Dont have a table for this machine Id yet so make one
- transactionPart = new Hashtable<Pair<int64_t, int32_t> *, TransactionPart *>();
+ transactionPart = new Hashtable<Pair<int64_t, int32_t>, TransactionPart *>();
newTransactionParts->put(entry->getMachineId(), transactionPart);
}
- Hashtable<Pair<int64_t, int32_t> *, CommitPart *> *commitPart = newCommitParts->get(entry->getMachineId());
+ Hashtable<Pair<int64_t, int32_t>, CommitPart *> *commitPart = newCommitParts->get(entry->getMachineId());
if (commitPart == NULL) {
// Don't have a table for this machine Id yet so make one
- commitPart = new Hashtable<Pair<int64_t, int32_t> *, CommitPart *>();
+ commitPart = new Hashtable<Pair<int64_t, int32_t>, CommitPart *>();
newCommitParts->put(entry->getMachineId(), commitPart);
}
}
// Set dead the abort
- for (Iterator<Map->Entry<Pair<int64_t, int64_t> *, Abort *> > i = liveAbortTable->entrySet()->iterator(); i->hasNext();) {
+ for (Iterator<Map->Entry<Pair<int64_t, int64_t>, Abort *> > i = liveAbortTable->entrySet()->iterator(); i->hasNext();) {
Abort *abort = i->next()->getValue();
if ((abort->getTransactionMachineId() == machineId) && (abort->getSequenceNumber() <= seqNum)) {
}
// Get the old last message for this device
- Pair<int64_t, Liveness *> *lastMessageEntry = lastMessageTable->put(machineId, new Pair<int64_t, Liveness *>(seqNum, liveness));
+ Pair<int64_t, Liveness *> lastMessageEntry = lastMessageTable->put(machineId, Pair<int64_t, Liveness *>(seqNum, liveness));
if (lastMessageEntry == NULL) {
// If no last message then there is nothing else to process
return;
Hashtable<IoTString *, KeyValue *> *speculatedKeyValueTable; // Table of speculated key value pairs, if there is a speculative value
Hashtable<IoTString *, KeyValue *> *pendingTransactionSpeculatedKeyValueTable; // Table of speculated key value pairs, if there is a speculative value from the pending transactions
Hashtable<IoTString *, NewKey *> *liveNewKeyTable; // Table of live new keys
- Hashtable<int64_t, Pair<int64_t, Liveness *> *> *lastMessageTable; // Last message sent by a client machine id -> (Seq Num, Slot or LastMessage);
+ Hashtable<int64_t, Pair<int64_t, Liveness *> > *lastMessageTable; // Last message sent by a client machine id -> (Seq Num, Slot or LastMessage);
Hashtable<int64_t, Hashset<RejectedMessage *> *> *rejectedMessageWatchVectorTable; // Table of machine Ids and the set of rejected messages they have not seen yet
Hashtable<IoTString *, int64_t> *arbitratorTable;// Table of keys and their arbitrators
- Hashtable<Pair<int64_t, int64_t> *, Abort *> *liveAbortTable;// Table live abort messages
- Hashtable<int64_t, Hashtable<Pair<int64_t, int32_t> *, TransactionPart *> *> *newTransactionParts; // transaction parts that are seen in this latest round of slots from the server
- Hashtable<int64_t, Hashtable<Pair<int64_t, int32_t> *, CommitPart *> *> *newCommitParts; // commit parts that are seen in this latest round of slots from the server
+ Hashtable<Pair<int64_t, int64_t>, Abort *> *liveAbortTable;// Table live abort messages
+ Hashtable<int64_t, Hashtable<Pair<int64_t, int32_t>, TransactionPart *> *> *newTransactionParts; // transaction parts that are seen in this latest round of slots from the server
+ Hashtable<int64_t, Hashtable<Pair<int64_t, int32_t>, CommitPart *> *> *newCommitParts; // commit parts that are seen in this latest round of slots from the server
Hashtable<int64_t, int64_t> *lastArbitratedTransactionNumberByArbitratorTable; // Last transaction sequence number that an arbitrator arbitrated on
Hashtable<int64_t, Transaction *> *liveTransactionBySequenceNumberTable; // live transaction grouped by the sequence number
- Hashtable<Pair<int64_t, int64_t> *, Transaction *> *liveTransactionByTransactionIdTable; // live transaction grouped by the transaction ID
+ Hashtable<Pair<int64_t, int64_t>, Transaction *> *liveTransactionByTransactionIdTable; // live transaction grouped by the transaction ID
Hashtable<int64_t, Hashtable<int64_t, Commit *> > *liveCommitsTable;
Hashtable<IoTString *, Commit *> *liveCommitsByKeyTable;
Hashtable<int64_t, int64_t> *lastCommitSeenSequenceNumberByArbitratorTable;
Hashtable<Transaction *, Vector<int32_t> *> *transactionPartsSent;
Hashtable<int64_t, TransactionStatus *> *outstandingTransactionStatus;
Hashtable<int64_t, Abort *> *liveAbortsGeneratedByLocal;
- Hashset<Pair<int64_t, int64_t> *> *offlineTransactionsCommittedAndAtServer;
+ Hashset<Pair<int64_t, int64_t> > *offlineTransactionsCommittedAndAtServer;
Hashtable<int64_t, Pair<IoTString *, int32_t> > *localCommunicationTable;
Hashtable<int64_t, int64_t> *lastTransactionSeenFromMachineFromServer;
Hashtable<int64_t, int64_t> *lastArbitrationDataLocalSequenceNumberSeenFromArbitrator;
bool sendToServer(NewKey *newKey);
bool updateFromLocal(int64_t machineId);
Pair<bool, bool> sendTransactionToLocal(Transaction *transaction);
- ThreeTuple<bool, bool, Array<Slot *> *> *sendSlotsToServer(Slot *slot, int newSize, bool isNewKey);
+ ThreeTuple<bool, bool, Array<Slot *> *> sendSlotsToServer(Slot *slot, int newSize, bool isNewKey);
/**
* Returns false if a resize was needed
*/
- ThreeTuple<bool, int32_t, bool> *fillSlot(Slot *slot, bool resize, NewKey *newKeyEntry);
+ ThreeTuple<bool, int32_t, bool> fillSlot(Slot *slot, bool resize, NewKey *newKeyEntry);
void doRejectedMessages(Slot *s);
ThreeTuple<bool, bool, int64_t> doMandatoryResuce(Slot *slot, bool resize);