*/
void Table::validateAndUpdate(Array<Slot *> *newSlots, bool acceptUpdatesToLocal) {
- // The cloud communication layer has checked slot HMACs already before decoding
+ // The cloud communication layer has checked slot HMACs already
+ // before decoding
if (newSlots->length() == 0) {
return;
}
- // Make sure all slots are newer than the last largest slot this client has seen
- int64_t firstSeqNum = newSlots[0]->getSequenceNumber();
+ // Make sure all slots are newer than the last largest slot this
+ // client has seen
+ int64_t firstSeqNum = newSlots->get(0)->getSequenceNumber();
if (firstSeqNum <= sequenceNumber) {
throw new Error("Server Error: Sent older slots!");
}
- // Create an object that can access both new slots and slots in our local chain
- // without committing slots to our local chain
+ // Create an object that can access both new slots and slots in our
+ // local chain without committing slots to our local chain
SlotIndexer *indexer = new SlotIndexer(newSlots, buffer);
// Check that the HMAC chain is not broken
updateExpectedSize();
}
- // If there is a gap, check to see if the server sent us everything->
+ // If there is a gap, check to see if the server sent us
+ // everything->
if (firstSeqNum != (sequenceNumber + 1)) {
// Check the size of the slots that were sent down by the server->
// Can only check the size if there was a gap
checkNumSlots(newSlots->length);
- // Since there was a gap every machine must have pushed a slot or must have
- // a last message message-> If not then the server is hiding slots
+ // Since there was a gap every machine must have pushed a slot or
+ // must have a last message message-> If not then the server is
+ // hiding slots
if (!machineSet->isEmpty()) {
throw new Error("Missing record for machines: " + machineSet);
}
// Insert this slot into our local block chain copy->
buffer->putSlot(slot);
- // Keep track of how many slots are currently live (have live data in them)->
+ // Keep track of how many slots are currently live (have live data
+ // in them)->
liveSlotCount++;
}
// Get the sequence number of the latest slot in the system
- sequenceNumber = newSlots[newSlots->length() - 1]->getSequenceNumber();
-
+ sequenceNumber = newSlots->get(newSlots->length() - 1)->getSequenceNumber();
updateLiveStateFromServer();
// No Need to remember after we pulled from the server
}
void Table::initExpectedSize(int64_t firstSequenceNumber, int64_t numberOfSlots) {
- // if (didFindTableStatus) {
- // return;
- // }
int64_t prevslots = firstSequenceNumber;
-
if (didFindTableStatus) {
- // expectedsize = (prevslots < ((int64_t) numberOfSlots)) ? (int) prevslots : expectedsize;
- // System->out->println("Here2: " + expectedsize + " " + numberOfSlots + " " + prevslots);
-
} else {
expectedsize = (prevslots < ((int64_t) numberOfSlots)) ? (int) prevslots : numberOfSlots;
- // System->out->println("Here: " + expectedsize);
}
- // System->out->println(numberOfSlots);
-
didFindTableStatus = true;
currMaxSize = numberOfSlots;
}
/**
- * Check the size of the block chain to make sure there are enough slots sent back by the server->
- * This is only called when we have a gap between the slots that we have locally and the slots
- * sent by the server therefore in the slots sent by the server there will be at least 1 Table
- * status message
+ * Check the size of the block chain to make sure there are enough
+ * slots sent back by the server-> This is only called when we have a
+ * gap between the slots that we have locally and the slots sent by
+ * the server therefore in the slots sent by the server there will be
+ * at least 1 Table status message
*/
void Table::checkNumSlots(int numberOfSlots) {
if (numberOfSlots != expectedsize) {
// Change the number of local slots to the new size
numberOfSlots = (int32_t)currMaxSize;
-
- // Recalculate the resize threshold since the size of the local buffer has changed
+ // Recalculate the resize threshold since the size of the local
+ // buffer has changed
setResizeThreshold();
}
/**
- * Process the new transaction parts from this latest round of slots received from the server
+ * Process the new transaction parts from this latest round of slots
+ * received from the server
*/
void Table::processNewTransactionParts() {
return;
}
- // Iterate through all the machine Ids that we received new parts for
+ // 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);
}
}
- // Clear all the new transaction parts in preparation for the next time the server sends slots
+ // Clear all the new transaction parts in preparation for the next
+ // time the server sends slots
newTransactionParts->clear();
}
- // Check if this machine arbitrates for this transaction if not then we cant arbitrate this transaction
+ // Check if this machine arbitrates for this transaction if not
+ // then we cant arbitrate this transaction
if (transaction->getArbitrator() != localMachineId) {
continue;
}
lastTransactionCommitted = transactionSequenceNumber;
} else {
// Guard evaluated was false so create abort
-
// Create the abort
Abort *newAbort = new Abort(NULL,
transaction->getClientLocalSequenceNumber(),
transaction->getArbitrator(),
localArbitrationSequenceNumber);
localArbitrationSequenceNumber++;
-
generatedAborts->add(newAbort);
// Insert the abort so we can process
}
lastSeqNumArbOn = transactionSequenceNumber;
-
- // liveTransactionBySequenceNumberTable->remove(transactionSequenceNumber);
}
Commit *newCommit = NULL;
// If there is something to commit
if (speculativeTableTmp->size() != 0) {
-
// Create the commit and increment the commit sequence number
newCommit = new Commit(localArbitrationSequenceNumber, localMachineId, lastTransactionCommitted);
localArbitrationSequenceNumber++;
// create the commit parts
newCommit->createCommitParts();
- // Append all the commit parts to the end of the pending queue waiting for sending to the server
-
+ // Append all the commit parts to the end of the pending queue
+ // waiting for sending to the server
// Insert the commit so we can process it
for (CommitPart *commitPart : newCommit->getParts()->values()) {
processEntry(commitPart);
Pair<bool, bool> Table::arbitrateOnLocalTransaction(Transaction *transaction) {
- // Check if this machine arbitrates for this transaction if not then we cant arbitrate this transaction
+ // Check if this machine arbitrates for this transaction if not then
+ // we cant arbitrate this transaction
if (transaction->getArbitrator() != localMachineId) {
return Pair<bool, bool>(false, false);
}
}
if (transaction->evaluateGuard(committedKeyValueTable, NULL, NULL)) {
- // Guard evaluated as true
-
- // Create the commit and increment the commit sequence number
+ // Guard evaluated as true Create the commit and increment the
+ // commit sequence number
Commit *newCommit = new Commit(localArbitrationSequenceNumber, localMachineId, -1);
localArbitrationSequenceNumber++;
// create the commit parts
newCommit->createCommitParts();
- // Append all the commit parts to the end of the pending queue waiting for sending to the server
+ // Append all the commit parts to the end of the pending queue
+ // waiting for sending to the server
ArbitrationRound *arbitrationRound = new ArbitrationRound(newCommit, new Hashset<Abort *>());
pendingSendArbitrationRounds->add(arbitrationRound);
updateLiveStateFromLocal();
return Pair<bool, bool>(true, true);
} else {
-
if (transaction->getMachineId() == localMachineId) {
// For locally created messages update the status
-
// Guard evaluated was false so create abort
TransactionStatus status = transaction->getTransactionStatus();
if (status != NULL) {
} else {
Hashset<Abort *> *addAbortSet = new Hashset<Abort * >();
-
// Create the abort
Abort *newAbort = new Abort(NULL,
transaction->getClientLocalSequenceNumber(),
transaction->getArbitrator(),
localArbitrationSequenceNumber);
localArbitrationSequenceNumber++;
-
addAbortSet->add(newAbort);
-
- // Append all the commit parts to the end of the pending queue waiting for sending to the server
+ // Append all the commit parts to the end of the pending queue
+ // waiting for sending to the server
ArbitrationRound *arbitrationRound = new ArbitrationRound(NULL, addAbortSet);
pendingSendArbitrationRounds->add(arbitrationRound);
}
/**
- * Compacts the arbitration data my merging commits and aggregating aborts so that a single large push of commits can be done instead of many small updates
+ * Compacts the arbitration data my merging commits and aggregating
+ * aborts so that a single large push of commits can be done instead
+ * of many small updates
*/
bool Table::compactArbitrationData() {
-
if (pendingSendArbitrationRounds->size() < 2) {
// Nothing to compact so do nothing
return false;
ArbitrationRound *xs round = pendingSendArbitrationRounds->get(pendingSendArbitrationRounds->size() - numberToDelete - 1);
if (round->isFull() || round->didSendPart()) {
- // Stop since there is a part that cannot be compacted and we need to compact in order
+ // Stop since there is a part that cannot be compacted and we
+ // need to compact in order
break;
}
if (round->getCommit() == NULL) {
-
// Try compacting aborts only
int newSize = round->getCurrentSize() + lastRound->getAbortsCount();
if (newSize > ArbitrationRound->MAX_PARTS) {
}
lastRound->addAborts(round->getAborts());
} else {
-
// Create a new larger commit
Commit newCommit = Commit->merge(lastRound->getCommit(), round->getCommit(), localArbitrationSequenceNumber);
localArbitrationSequenceNumber++;
if (numberToDelete != 1) {
// If there is a compaction
-
// Delete the previous pieces that are now in the new compacted piece
if (numberToDelete == pendingSendArbitrationRounds->size()) {
pendingSendArbitrationRounds->clear();
return false;
}
-// bool compactArbitrationData() {
-// return false;
-// }
/**
- * Update all the commits and the committed tables, sets dead the dead transactions
+ * Update all the commits and the committed tables, sets dead the dead
+ * transactions
*/
bool Table::updateCommittedTable() {
}
}
- // Clear all the new commits parts in preparation for the next time the server sends slots
+ // Clear all the new commits parts in preparation for the next time
+ // the server sends slots
newCommitParts->clear();
// If we process a new commit keep track of it for future use
// Special processing if a commit is not complete
if (!commit->isComplete()) {
if (i == (commitSequenceNumbers->size() - 1)) {
- // If there is an incomplete commit and this commit is the latest one seen then this commit cannot be processed and there are no other commits
+ // If there is an incomplete commit and this commit is the
+ // latest one seen then this commit cannot be processed and
+ // there are no other commits
break;
} else {
- // This is a commit that was already dead but parts of it are still in the block chain (not flushed out yet)->
+ // This is a commit that was already dead but parts of it
+ // are still in the block chain (not flushed out yet)->
// Delete it and move on
commit->setDead();
commitForClientTable->remove(commit->getSequenceNumber());
lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->put(commit->getMachineId(), commit->getSequenceNumber());
}
- // We have already seen this commit before so need to do the full processing on this commit
+ // We have already seen this commit before so need to do the
+ // full processing on this commit
if (commit->getSequenceNumber() <= lastCommitSeenSequenceNumber) {
// Update the last transaction that was updated if we can
continue;
}
- // If we got here then this is a brand new commit and needs full processing
-
- // Get what commits should be edited, these are the commits that have live values for their keys
+ // If we got here then this is a brand new commit and needs full
+ // processing
+ // Get what commits should be edited, these are the commits that
+ // have live values for their keys
Hashset<Commit *> *commitsToEdit = new Hashset<Commit *>();
for (KeyValue *kv : commit->getKeyValueUpdateSet()) {
commitsToEdit->add(liveCommitsByKeyTable->get(kv->getKey()));
}
/**
- * Create the speculative table from transactions that are still live and have come from the cloud
+ * Create the speculative table from transactions that are still live
+ * and have come from the cloud
*/
bool Table::updateSpeculativeTable(bool didProcessNewCommits) {
if (liveTransactionBySequenceNumberTable->keySet()->size() == 0) {
return false;
}
- // Create a list of the transaction sequence numbers and sort them from oldest to newest
+ // Create a list of the transaction sequence numbers and sort them
+ // from oldest to newest
Vector<int64_t> *transactionSequenceNumbersSorted = new Vector<int64_t>(liveTransactionBySequenceNumberTable->keySet());
Collections->sort(transactionSequenceNumbersSorted);
if (hasGapInTransactionSequenceNumbers || didProcessNewCommits) {
- // If there is a gap in the transaction sequence numbers then there was a commit or an abort of a transaction
- // OR there was a new commit (Could be from offline commit) so a redo the speculation from scratch
+ // If there is a gap in the transaction sequence numbers then
+ // there was a commit or an abort of a transaction OR there was a
+ // new commit (Could be from offline commit) so a redo the
+ // speculation from scratch
// Start from scratch
speculatedKeyValueTable->clear();
Transaction *transaction = liveTransactionBySequenceNumberTable->get(transactionSequenceNumber);
if (!transaction->isComplete()) {
- // If there is an incomplete transaction then there is nothing we can do
- // add this transactions arbitrator to the list of arbitrators we should ignore
+ // If there is an incomplete transaction then there is nothing
+ // we can do add this transactions arbitrator to the list of
+ // arbitrators we should ignore
incompleteTransactionArbitrator->add(transaction->getArbitrator());
didSkip = true;
continue;
}
/**
- * Create the pending transaction speculative table from transactions that are still in the pending transaction buffer
+ * Create the pending transaction speculative table from transactions
+ * that are still in the pending transaction buffer
*/
void Table::updatePendingTransactionSpeculativeTable(bool didProcessNewCommitsOrSpeculate) {
if (pendingTransactionQueue->size() == 0) {
}
/**
- * Set dead and remove from the live transaction tables the transactions that are dead
+ * Set dead and remove from the live transaction tables the
+ * transactions that are dead
*/
void Table::updateLiveTransactionsAndStatus() {
// Process each entry in the slot
for (Entry *entry : slot->getEntries()) {
switch (entry->getType()) {
-
case TypeCommitPart:
processEntry((CommitPart *)entry);
break;
-
case TypeAbort:
processEntry((Abort *)entry);
break;
-
case TypeTransactionPart:
processEntry((TransactionPart *)entry);
break;
-
case TypeNewKey:
processEntry((NewKey *)entry);
break;
-
case TypeLastMessage:
processEntry((LastMessage *)entry, machineSet);
break;
-
case TypeRejectedMessage:
processEntry((RejectedMessage *)entry, indexer);
break;
-
case TypeTableStatus:
processEntry((TableStatus *)entry, slot->getSequenceNumber());
break;
-
default:
throw new Error("Unrecognized type: " + entry->getType());
}
}
/**
- * Add the new key to the arbitrators table and update the set of live new keys (in case of a rescued new key message)
+ * Add the new key to the arbitrators table and update the set of live
+ * new keys (in case of a rescued new key message)
*/
void Table::processEntry(NewKey *entry) {
-
// Update the arbitrator table with the new key information
arbitratorTable->put(entry->getKey(), entry->getMachineID());
}
/**
- * Process new table status entries and set dead the old ones as new ones come in->
- * keeps track of the largest and smallest table status seen in this current round
- * of updating the local copy of the block chain
+ * Process new table status entries and set dead the old ones as new
+ * ones come in-> keeps track of the largest and smallest table status
+ * seen in this current round of updating the local copy of the block
+ * chain
*/
void Table::processEntry(TableStatus entry, int64_t seq) {
int newNumSlots = entry->getMaxSlots();
updateCurrMaxSize(newNumSlots);
-
initExpectedSize(seq, newNumSlots);
if (liveTableStatus != NULL) {
- // We have a larger table status so the old table status is no int64_ter alive
+ // We have a larger table status so the old table status is no
+ // int64_ter alive
liveTableStatus->setDead();
}
}
/**
- * Check old messages to see if there is a block chain violation-> Also
+ * Check old messages to see if there is a block chain violation->
+ * Also
*/
void Table::processEntry(RejectedMessage *entry, SlotIndexer *indexer) {
int64_t oldSeqNum = entry->getOldSeqNum();
int64_t machineId = entry->getMachineID();
int64_t seq = entry->getSequenceNumber();
-
- // Check if we have messages that were supposed to be rejected in our local block chain
+ // Check if we have messages that were supposed to be rejected in
+ // our local block chain
for (int64_t seqNum = oldSeqNum; seqNum <= newSeqNum; seqNum++) {
-
// Get the slot
Slot *slot = indexer->getSlot(seqNum);
if (slot != NULL) {
- // If we have this slot make sure that it was not supposed to be a rejected slot
-
+ // If we have this slot make sure that it was not supposed to be
+ // a rejected slot
int64_t slotMachineId = slot->getMachineID();
if (isequal != (slotMachineId == machineId)) {
throw new Error("Server Error: Trying to insert rejected message for slot " + seqNum);
}
}
-
- // Create a list of clients to watch until they see this rejected message entry->
+ // 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()) {
-
// Machine ID for the last message entry
int64_t lastMessageEntryMachineId = lastMessageEntry->getKey();
int64_t entrySequenceNumber = lastMessageValue->getFirst();
if (entrySequenceNumber < seq) {
-
- // Add this rejected message to the set of messages that this machine ID did not see yet
+ // Add this rejected message to the set of messages that this
+ // machine ID did not see yet
addWatchVector(lastMessageEntryMachineId, entry);
-
- // This client did not see this rejected message yet so add it to the watch set to monitor
+ // This client did not see this rejected message yet so add it
+ // to the watch set to monitor
deviceWatchSet->add(lastMessageEntryMachineId);
}
}
-
if (deviceWatchSet->isEmpty()) {
// This rejected message has been seen by all the clients so
entry->setDead();
}
/**
- * Check if this abort is live, if not then save it so we can kill it later->
- * update the last transaction number that was arbitrated on->
+ * Check if this abort is live, if not then save it so we can kill it
+ * later-> update the last transaction number that was arbitrated on->
*/
void Table::processEntry(Abort *entry) {
-
-
if (entry->getTransactionSequenceNumber() != -1) {
// update the transaction status if it was sent to the server
TransactionStatus *status = outstandingTransactionStatus->remove(entry->getTransactionSequenceNumber());
}
}
- // Abort has not been seen by the client it is for yet so we need to keep track of it
+ // Abort has not been seen by the client it is for yet so we need to
+ // keep track of it
Abort *previouslySeenAbort = liveAbortTable->put(entry->getAbortId(), entry);
if (previouslySeenAbort != NULL) {
previouslySeenAbort->setDead(); // Delete old version of the abort since we got a rescued newer version
}
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());
if (entry->getTransactionArbitrator() == localMachineId) {
liveAbortsGeneratedByLocal->remove(entry->getArbitratorLocalSequenceNumber());
}
-
return;
}
// Update the last arbitration data that we have seen so far
if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->get(entry->getTransactionArbitrator()) != NULL) {
-
int64_t lastArbitrationSequenceNumber = lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->get(entry->getTransactionArbitrator());
if (entry->getSequenceNumber() > lastArbitrationSequenceNumber) {
// Is larger
lastArbitrationDataLocalSequenceNumberSeenFromArbitrator->put(entry->getTransactionArbitrator(), entry->getSequenceNumber());
}
-
// Set dead a transaction if we can
Transaction *transactionToSetDead = liveTransactionByTransactionIdTable->remove(Pair<int64_t, int64_t>(entry->getTransactionMachineId(), entry->getTransactionClientLocalSequenceNumber()));
if (transactionToSetDead != NULL) {
liveTransactionBySequenceNumberTable->remove(transactionToSetDead->getSequenceNumber());
}
- // Update the last transaction sequence number that the arbitrator arbitrated on
+ // Update the last transaction sequence number that the arbitrator
+ // arbitrated on
int64_t lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable->get(entry->getTransactionArbitrator());
if ((lastTransactionNumber == NULL) || (lastTransactionNumber < entry->getTransactionSequenceNumber())) {
-
// Is a valid one
if (entry->getTransactionSequenceNumber() != -1) {
lastArbitratedTransactionNumberByArbitratorTable->put(entry->getTransactionArbitrator(), entry->getTransactionSequenceNumber());
}
/**
- * Set dead the transaction part if that transaction is dead and keep track of all new parts
+ * Set dead the transaction part if that transaction is dead and keep
+ * track of all new parts
*/
void Table::processEntry(TransactionPart *entry) {
- // Check if we have already seen this transaction and set it dead OR if it is not alive
+ // Check if we have already seen this transaction and set it dead OR
+ // if it is not alive
int64_t lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable->get(entry->getArbitratorId());
if ((lastTransactionNumber != NULL) && (lastTransactionNumber >= entry->getSequenceNumber())) {
// This transaction is dead, it was already committed or aborted
newTransactionParts->put(entry->getMachineId(), transactionPart);
}
- // Update the part and set dead ones we have already seen (got a rescued version)
+ // Update the part and set dead ones we have already seen (got a
+ // rescued version)
TransactionPart *previouslySeenPart = transactionPart->put(entry->getPartId(), entry);
if (previouslySeenPart != NULL) {
previouslySeenPart->setDead();
// Update the last transaction that was updated if we can
if (entry->getTransactionSequenceNumber() != -1) {
int64_t lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable->get(entry->getMachineId());
-
- // Update the last transaction sequence number that the arbitrator arbitrated on
+ // Update the last transaction sequence number that the arbitrator
+ // arbitrated on
if ((lastTransactionNumber == NULL) || (lastTransactionNumber < entry->getTransactionSequenceNumber())) {
lastArbitratedTransactionNumberByArbitratorTable->put(entry->getMachineId(), entry->getTransactionSequenceNumber());
}
}
-
-
-
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 *>();
newCommitParts->put(entry->getMachineId(), commitPart);
}
-
- // Update the part and set dead ones we have already seen (got a rescued version)
+ // Update the part and set dead ones we have already seen (got a
+ // rescued version)
CommitPart *previouslySeenPart = commitPart->put(entry->getPartId(), entry);
if (previouslySeenPart != NULL) {
previouslySeenPart->setDead();
}
/**
- * Update the last message seen table-> Update and set dead the appropriate RejectedMessages as clients see them->
- * Updates the live aborts, removes those that are dead and sets them dead->
- * Check that the last message seen is correct and that there is no mismatch of our own last message or that
- * other clients have not had a rollback on the last message->
+ * Update the last message seen table-> Update and set dead the
+ * appropriate RejectedMessages as clients see them-> Updates the live
+ * aborts, removes those that are dead and sets them dead-> Check that
+ * the last message seen is correct and that there is no mismatch of
+ * 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) {
-
// We have seen this machine ID
machineSet->remove(machineId);
// Get the set of rejected messages that this machine Id is has not seen yet
Hashset<RejectedMessage *> *watchset = rejectedMessageWatchVectorTable->get(machineId);
-
// If there is a rejected message that this machine Id has not seen yet
if (watchset != NULL) {
-
- // Go through each rejected message that this machine Id has not seen yet
+ // Go through each rejected message that this machine Id has not
+ // seen yet
for (Iterator<RejectedMessage *> *rmit = watchset->iterator(); rmit->hasNext(); ) {
-
RejectedMessage *rm = rmit->next();
-
// If this machine Id has seen this rejected message->->->
if (rm->getSequenceNumber() <= seqNum) {
-
// Remove it from our watchlist
rmit->remove();
-
// Decrement machines that need to see this notification
rm->removeWatcher(machineId);
}
// Set dead the abort
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)) {
abort->setDead();
i->remove();
-
if (abort->getTransactionArbitrator() == localMachineId) {
liveAbortsGeneratedByLocal->remove(abort->getArbitratorLocalSequenceNumber());
}
}
}
-
-
-
if (machineId == localMachineId) {
// Our own messages are immediately dead->
if (liveness instanceof LastMessage) {
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));
if (lastMessageEntry == NULL) {
throw new Error("Unrecognized type");
}
}
-
// Make sure the server is not playing any games
if (machineId == localMachineId) {
-
if (hadPartialSendToServer) {
// We were not making any updates and we had a machine mismatch
if (lastMessageSeqNum > seqNum && !acceptUpdatesToLocal) {
throw new Error("Server Error: Mismatch on local machine sequence number, needed at least: " + lastMessageSeqNum + " got: " + seqNum);
}
-
} else {
// We were not making any updates and we had a machine mismatch
if (lastMessageSeqNum != seqNum && !acceptUpdatesToLocal) {
}
/**
- * Add a rejected message entry to the watch set to keep track of which clients have seen that
- * rejected message entry and which have not->
+ * Add a rejected message entry to the watch set to keep track of
+ * which clients have seen that rejected message entry and which have
+ * not.
*/
void Table::addWatchVector(int64_t machineId, RejectedMessage *entry) {
Hashset<RejectedMessage *> *entries = rejectedMessageWatchVectorTable->get(machineId);
* Check if the HMAC chain is not violated
*/
void Table::checkHMACChain(SlotIndexer *indexer, Array<Slot *> *newSlots) {
- for (int i = 0; i < newSlots->length; i++) {
- Slot *currSlot = newSlots[i];
+ for (int i = 0; i < newSlots->length(); i++) {
+ Slot *currSlot = newSlots->get(i);
Slot *prevSlot = indexer->getSlot(currSlot->getSequenceNumber() - 1);
if (prevSlot != NULL &&
!prevSlot->getHMAC()->equals(currSlot->getPrevHMAC()))
throw new Error("Server Error: Invalid HMAC Chain");
}
}
-
projects / iotcloud.git / commitdiff