projects
/
iotcloud.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
tabbing
[iotcloud.git]
/
version2
/
src
/
C
/
Table.cc
diff --git
a/version2/src/C/Table.cc
b/version2/src/C/Table.cc
index 562e164be72e6b371afc62ee7ce9050e0cecd264..76a7f16adfc7b6a7431859fa3742ccf1f31db7dd 100644
(file)
--- a/
version2/src/C/Table.cc
+++ b/
version2/src/C/Table.cc
@@
-32,7
+32,7
@@
Table::Table(String baseurl, String password, int64_t _localMachineId, int liste
committedKeyValueTable(NULL),
speculatedKeyValueTable(NULL),
pendingTransactionSpeculatedKeyValueTable(NULL),
committedKeyValueTable(NULL),
speculatedKeyValueTable(NULL),
pendingTransactionSpeculatedKeyValueTable(NULL),
- liveNewKeyTable(NULL),
+
liveNewKeyTable(NULL),
lastMessageTable(NULL),
rejectedMessageWatchVectorTable(NULL),
arbitratorTable(NULL),
lastMessageTable(NULL),
rejectedMessageWatchVectorTable(NULL),
arbitratorTable(NULL),
@@
-94,7
+94,7
@@
Table::Table(CloudComm _cloud, int64_t _localMachineId) :
committedKeyValueTable(NULL),
speculatedKeyValueTable(NULL),
pendingTransactionSpeculatedKeyValueTable(NULL),
committedKeyValueTable(NULL),
speculatedKeyValueTable(NULL),
pendingTransactionSpeculatedKeyValueTable(NULL),
- liveNewKeyTable(NULL),
+
liveNewKeyTable(NULL),
lastMessageTable(NULL),
rejectedMessageWatchVectorTable(NULL),
arbitratorTable(NULL),
lastMessageTable(NULL),
rejectedMessageWatchVectorTable(NULL),
arbitratorTable(NULL),
@@
-176,7
+176,7
@@
synchronized void Table::printSlots() {
int64_t o = buffer.getOldestSeqNum();
int64_t n = buffer.getNewestSeqNum();
int64_t o = buffer.getOldestSeqNum();
int64_t n = buffer.getNewestSeqNum();
- Array<int> *
types = new Array<int>(10);
+ Array<int> *types = new Array<int>(10);
int num = 0;
int num = 0;
@@
-262,7
+262,7
@@
synchronized void Table::initTable() {
localSequenceNumber++;
TableStatus status = new TableStatus(s, numberOfSlots);
s.addEntry(status);
localSequenceNumber++;
TableStatus status = new TableStatus(s, numberOfSlots);
s.addEntry(status);
- Array<Slot> *
array = cloud.putSlot(s, numberOfSlots);
+ Array<Slot> *array = cloud.putSlot(s, numberOfSlots);
if (array == NULL) {
array = new Array<Slot>(1);
if (array == NULL) {
array = new Array<Slot>(1);
@@
-282,7
+282,7
@@
synchronized void Table::initTable() {
*/
synchronized void Table::rebuild() {
// Just pull the latest slots from the server
*/
synchronized void Table::rebuild() {
// Just pull the latest slots from the server
- Array<Slot> *
newslots = cloud.getSlots(sequenceNumber + 1);
+ Array<Slot> *newslots = cloud.getSlots(sequenceNumber + 1);
validateAndUpdate(newslots, true);
sendToServer(NULL);
updateLiveTransactionsAndStatus();
validateAndUpdate(newslots, true);
sendToServer(NULL);
updateLiveTransactionsAndStatus();
@@
-397,7
+397,7
@@
synchronized IoTString Table::getSpeculativeAtomic(IoTString key) {
synchronized bool Table::update() {
try {
synchronized bool Table::update() {
try {
- Array<Slot> *
newSlots = cloud.getSlots(sequenceNumber + 1);
+ Array<Slot> *newSlots = cloud.getSlots(sequenceNumber + 1);
validateAndUpdate(newSlots, false);
sendToServer(NULL);
validateAndUpdate(newSlots, false);
sendToServer(NULL);
@@
-553,10
+553,10
@@
bool Table::sendToServer(NewKey newKey) {
try {
if (hadPartialSendToServer) {
try {
if (hadPartialSendToServer) {
- Array<Slot> *
newSlots = cloud.getSlots(sequenceNumber + 1);
+ Array<Slot> *newSlots = cloud.getSlots(sequenceNumber + 1);
if (newSlots.length == 0) {
fromRetry = true;
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 (sendSlotsReturn.getFirst()) {
if (newKey != NULL) {
@@
-793,7
+793,7
@@
bool Table::sendToServer(NewKey newKey) {
lastPendingSendArbitrationEntriesToDelete = new Vector<Entry>(pendingSendArbitrationEntriesToDelete);
lastPendingSendArbitrationEntriesToDelete = new Vector<Entry>(pendingSendArbitrationEntriesToDelete);
- ThreeTuple<bool, bool, Array<Slot>*> sendSlotsReturn = sendSlotsToServer(slot, newSize, newKey != NULL);
+ ThreeTuple<bool, bool, Array<Slot>
*> sendSlotsReturn = sendSlotsToServer(slot, newSize, newKey != NULL);
if (sendSlotsReturn.getFirst()) {
if (sendSlotsReturn.getFirst()) {
@@
-1172,14
+1172,14
@@
synchronized Array<char> *Table::acceptDataFromLocal(Array<char> *data) {
return returnData;
}
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;
bool inserted = false;
bool lastTryInserted = false;
bool attemptedToSendToServerTmp = attemptedToSendToServer;
attemptedToSendToServer = true;
bool inserted = false;
bool lastTryInserted = false;
- Array<Slot>
*
array = cloud.putSlot(slot, newSize);
+ Array<Slot>
*
array = cloud.putSlot(slot, newSize);
if (array == NULL) {
array = new Array<Slot>();
array->set(0, slot);
if (array == NULL) {
array = new Array<Slot>();
array->set(0, slot);
@@
-1232,7
+1232,7
@@
ThreeTuple<bool, bool, Array<Slot>*> Table::sendSlotsToServer(Slot slot, int new
}
}
}
}
- return new ThreeTuple<bool, bool, Array<Slot>*>(inserted, lastTryInserted, array);
+ return new ThreeTuple<bool, bool, Array<Slot>
*>(inserted, lastTryInserted, array);
}
/**
}
/**
@@
-1476,7
+1476,7
@@
search:
/**
* Checks for malicious activity and updates the local copy of the block chain.
*/
/**
* Checks for malicious activity and updates the local copy of the block chain.
*/
-void Table::validateAndUpdate(Array<Slot>
*
newSlots, bool acceptUpdatesToLocal) {
+void Table::validateAndUpdate(Array<Slot>
*
newSlots, bool acceptUpdatesToLocal) {
// The cloud communication layer has checked slot HMACs already before decoding
if (newSlots.length == 0) {
// The cloud communication layer has checked slot HMACs already before decoding
if (newSlots.length == 0) {
@@
-1858,7
+1858,7
@@
Pair<bool, bool> Table::arbitrateOnLocalTransaction(Transaction transaction) {
newCommit.createCommitParts();
// Append all the commit parts to the end of the pending queue waiting for sending to the server
newCommit.createCommitParts();
// 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 *>());
+ ArbitrationRound *arbitrationRound = new ArbitrationRound(newCommit, new Hashset<Abort *>());
pendingSendArbitrationRounds.add(arbitrationRound);
if (compactArbitrationData()) {
pendingSendArbitrationRounds.add(arbitrationRound);
if (compactArbitrationData()) {
@@
-1893,7
+1893,7
@@
Pair<bool, bool> Table::arbitrateOnLocalTransaction(Transaction transaction) {
status.setStatus(TransactionStatus.StatusAborted);
}
} else {
status.setStatus(TransactionStatus.StatusAborted);
}
} else {
- Hashset<Abort
*> addAbortSet = new Hashset<Abort
* >();
+ Hashset<Abort
*> addAbortSet = new Hashset<Abort
* >();
// Create the abort
// Create the abort
@@
-2141,7
+2141,7
@@
bool Table::updateCommittedTable() {
// 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
*>();
+ Hashset<Commit
*> *commitsToEdit = new Hashset<Commit
*>();
for (KeyValue kv : commit.getKeyValueUpdateSet()) {
commitsToEdit.add(liveCommitsByKeyTable.get(kv.getKey()));
}
for (KeyValue kv : commit.getKeyValueUpdateSet()) {
commitsToEdit.add(liveCommitsByKeyTable.get(kv.getKey()));
}
@@
-2226,7
+2226,7
@@
bool Table::updateSpeculativeTable(bool didProcessNewCommits) {
return false; // did not speculate
}
return false; // did not speculate
}
- Hashset<int64_t> *
incompleteTransactionArbitrator = new Hashset<int64_t>();
+ Hashset<int64_t> *incompleteTransactionArbitrator = new Hashset<int64_t>();
bool didSkip = true;
for (int i = startIndex; i < transactionSequenceNumbersSorted.size(); i++) {
bool didSkip = true;
for (int i = startIndex; i < transactionSequenceNumbersSorted.size(); i++) {
@@
-2346,7
+2346,7
@@
void Table::updateLiveTransactionsAndStatus() {
/**
* Process this slot, entry by entry. Also update the latest message sent by slot
*/
/**
* Process this slot, entry by entry. Also update the latest message sent by slot
*/
-void Table::processSlot(SlotIndexer indexer, Slot slot, bool acceptUpdatesToLocal, Hashset<int64_t> *
machineSet) {
+void Table::processSlot(SlotIndexer indexer, Slot slot, bool acceptUpdatesToLocal, Hashset<int64_t> *machineSet) {
// Update the last message seen
updateLastMessage(slot.getMachineID(), slot.getSequenceNumber(), slot, acceptUpdatesToLocal, machineSet);
// Update the last message seen
updateLastMessage(slot.getMachineID(), slot.getSequenceNumber(), slot, acceptUpdatesToLocal, machineSet);
@@
-2392,7
+2392,7
@@
void Table::processSlot(SlotIndexer indexer, Slot slot, bool acceptUpdatesToLoca
/**
* Update the last message that was sent for a machine Id
*/
/**
* Update the last message that was sent for a machine Id
*/
-void Table::processEntry(LastMessage entry, Hashset<int64_t>
*
machineSet) {
+void Table::processEntry(LastMessage entry, Hashset<int64_t>
*
machineSet) {
// Update what the last message received by a machine was
updateLastMessage(entry.getMachineID(), entry.getSequenceNumber(), entry, false, machineSet);
}
// Update what the last message received by a machine was
updateLastMessage(entry.getMachineID(), entry.getSequenceNumber(), entry, false, machineSet);
}
@@
-2462,7
+2462,7
@@
void Table::processEntry(RejectedMessage entry, SlotIndexer indexer) {
// 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>();
+ 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
for (Map.Entry<int64_t Pair<int64_t Liveness> > lastMessageEntry : lastMessageTable.entrySet()) {
// Machine ID for the last message entry
@@
-2636,13
+2636,13
@@
void Table::processEntry(CommitPart entry) {
* 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.
*/
* 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) {
+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
// 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);
+ Hashset<RejectedMessage
*> *
watchset = rejectedMessageWatchVectorTable.get(machineId);
// If there is a rejected message that this machine Id has not seen yet
if (watchset != NULL) {
// If there is a rejected message that this machine Id has not seen yet
if (watchset != NULL) {
@@
-2739,10
+2739,10
@@
void Table::updateLastMessage(int64_t machineId, int64_t seqNum, Liveness livene
* rejected message entry and which have not.
*/
void Table::addWatchVector(int64_t machineId, RejectedMessage entry) {
* rejected message entry and which have not.
*/
void Table::addWatchVector(int64_t machineId, RejectedMessage entry) {
- Hashset<RejectedMessage
*>*
entries = rejectedMessageWatchVectorTable.get(machineId);
+ Hashset<RejectedMessage
*> *
entries = rejectedMessageWatchVectorTable.get(machineId);
if (entries == NULL) {
// There is no set for this machine ID yet so create one
if (entries == NULL) {
// There is no set for this machine ID yet so create one
- entries = new Hashset<RejectedMessage*>();
+ entries = new Hashset<RejectedMessage
*>();
rejectedMessageWatchVectorTable.put(machineId, entries);
}
entries.add(entry);
rejectedMessageWatchVectorTable.put(machineId, entries);
}
entries.add(entry);
@@
-2751,7
+2751,7
@@
void Table::addWatchVector(int64_t machineId, RejectedMessage entry) {
/**
* Check if the HMAC chain is not violated
*/
/**
* Check if the HMAC chain is not violated
*/
-void Table::checkHMACChain(SlotIndexer indexer, Array<Slot> *
newSlots) {
+void Table::checkHMACChain(SlotIndexer indexer, Array<Slot> *newSlots) {
for (int i = 0; i < newSlots.length; i++) {
Slot currSlot = newSlots[i];
Slot prevSlot = indexer.getSlot(currSlot.getSequenceNumber() - 1);
for (int i = 0; i < newSlots.length; i++) {
Slot currSlot = newSlots[i];
Slot prevSlot = indexer.getSlot(currSlot.getSequenceNumber() - 1);
projects / iotcloud.git / blobdiff