edits
authorbdemsky <bdemsky@uci.edu>
Mon, 12 Mar 2018 10:21:12 +0000 (03:21 -0700)
committerbdemsky <bdemsky@uci.edu>
Mon, 12 Mar 2018 10:21:12 +0000 (03:21 -0700)
version2/src/C/Table.cc
version2/src/C/Table.h
version2/src/C/Transaction.cc
version2/src/C/Transaction.h

index 152b377..7c3fde0 100644 (file)
@@ -201,10 +201,10 @@ void Table::init() {
        buffer = new SlotBuffer();
 
        // init data structs
-       committedKeyValueTable = new Hashtable<IoTString *, KeyValue *>();
-       speculatedKeyValueTable = new Hashtable<IoTString *, KeyValue *>();
-       pendingTransactionSpeculatedKeyValueTable = new Hashtable<IoTString *, KeyValue *>();
-       liveNewKeyTable = new Hashtable<IoTString *, NewKey *>();
+       committedKeyValueTable = new Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals>();
+       speculatedKeyValueTable = new Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals>();
+       pendingTransactionSpeculatedKeyValueTable = new Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals>();
+       liveNewKeyTable = new Hashtable<IoTString *, NewKey *, uintptr_t, 0, hashString, StringEquals >();
        lastMessageTable = new Hashtable<int64_t, Pair<int64_t, Liveness *> * >();
        rejectedMessageWatchVectorTable = new Hashtable<int64_t, Hashset<RejectedMessage *> * >();
        arbitratorTable = new Hashtable<IoTString *, int64_t, uintptr_t, 0, hashString, StringEquals>();
@@ -215,7 +215,7 @@ void Table::init() {
        liveTransactionBySequenceNumberTable = new Hashtable<int64_t, Transaction *>();
        liveTransactionByTransactionIdTable = new Hashtable<Pair<int64_t, int64_t> *, Transaction *, uintptr_t, 0, pairHashFunction, pairEquals>();
        liveCommitsTable = new Hashtable<int64_t, Hashtable<int64_t, Commit *> * >();
-       liveCommitsByKeyTable = new Hashtable<IoTString *, Commit *>();
+       liveCommitsByKeyTable = new Hashtable<IoTString *, Commit *, uintptr_t, 0, hashString, StringEquals>();
        lastCommitSeenSequenceNumberByArbitratorTable = new Hashtable<int64_t, int64_t>();
        rejectedSlotVector = new Vector<int64_t>();
        pendingTransactionQueue = new Vector<Transaction *>();
@@ -500,228 +500,224 @@ int64_t Table::getLocalSequenceNumber() {
        return localSequenceNumber;
 }
 
-bool Table::sendToServer(NewKey *newKey) {
-       bool fromRetry = false;
-       try {
-               if (hadPartialSendToServer) {
-                       Array<Slot *> *newSlots = cloud->getSlots(sequenceNumber + 1);
-                       if (newSlots->length() == 0) {
-                               fromRetry = true;
-                               ThreeTuple<bool, bool, Array<Slot *> *> sendSlotsReturn = sendSlotsToServer(lastSlotAttemptedToSend, lastNewSize, lastIsNewKey);
-
-                               if (sendSlotsReturn.getFirst()) {
-                                       if (newKey != NULL) {
-                                               if (lastInsertedNewKey && (lastNewKey->getKey() == newKey->getKey()) && (lastNewKey->getMachineID() == newKey->getMachineID())) {
-                                                       newKey = NULL;
-                                               }
-                                       }
-
-                                       SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
-                                       while (trit->hasNext()) {
-                                               Transaction *transaction = trit->next();
-                                               transaction->resetServerFailure();
-                                               // Update which transactions parts still need to be sent
-                                               transaction->removeSentParts(lastTransactionPartsSent->get(transaction));
-                                               // Add the transaction status to the outstanding list
-                                               outstandingTransactionStatus->put(transaction->getSequenceNumber(), transaction->getTransactionStatus());
-
-                                               // Update the transaction status
-                                               transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentPartial);
-
-                                               // Check if all the transaction parts were successfully
-                                               // sent and if so then remove it from pending
-                                               if (transaction->didSendAllParts()) {
-                                                       transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentFully);
-                                                       pendingTransactionQueue->remove(transaction);
-                                               }
-                                       }
-                                       delete trit;
-                               } else {
-                                       newSlots = sendSlotsReturn.getThird();
-                                       bool isInserted = false;
-                                       for (uint si = 0; si < newSlots->length(); si++) {
-                                               Slot *s = newSlots->get(si);
-                                               if ((s->getSequenceNumber() == lastSlotAttemptedToSend->getSequenceNumber()) && (s->getMachineID() == localMachineId)) {
+NewKey * Table::handlePartialSend(NewKey * newKey) {
+       //Didn't receive acknowledgement for last send
+       //See if the server has received a newer slot
+       
+       Array<Slot *> *newSlots = cloud->getSlots(sequenceNumber + 1);
+       if (newSlots->length() == 0) {
+               //Retry sending old slot
+               bool wasInserted = false;
+               bool sendSlotsReturn = sendSlotsToServer(lastSlotAttemptedToSend, lastNewSize, lastIsNewKey, &wasInserted, &newSlots);
+               
+               if (sendSlotsReturn) {
+                       if (newKey != NULL) {
+                               if (lastInsertedNewKey && (lastNewKey->getKey() == newKey->getKey()) && (lastNewKey->getMachineID() == newKey->getMachineID())) {
+                                       newKey = NULL;
+                               }
+                       }
+                       
+                       SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
+                       while (trit->hasNext()) {
+                               Transaction *transaction = trit->next();
+                               transaction->resetServerFailure();
+                               // Update which transactions parts still need to be sent
+                               transaction->removeSentParts(lastTransactionPartsSent->get(transaction));
+                               // Add the transaction status to the outstanding list
+                               outstandingTransactionStatus->put(transaction->getSequenceNumber(), transaction->getTransactionStatus());
+                               
+                               // Update the transaction status
+                               transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentPartial);
+                               
+                               // Check if all the transaction parts were successfully
+                               // sent and if so then remove it from pending
+                               if (transaction->didSendAllParts()) {
+                                       transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentFully);
+                                       pendingTransactionQueue->remove(transaction);
+                               }
+                       }
+                       delete trit;
+               } else {
+                       bool isInserted = false;
+                       for (uint si = 0; si < newSlots->length(); si++) {
+                               Slot *s = newSlots->get(si);
+                               if ((s->getSequenceNumber() == lastSlotAttemptedToSend->getSequenceNumber()) && (s->getMachineID() == localMachineId)) {
+                                       isInserted = true;
+                                       break;
+                               }
+                       }
+                       
+                       for (uint si = 0; si < newSlots->length(); si++) {
+                               Slot *s = newSlots->get(si);
+                               if (isInserted) {
+                                       break;
+                               }
+                               
+                               // Process each entry in the slot
+                               Vector<Entry *> *ventries = s->getEntries();
+                               uint vesize = ventries->size();
+                               for (uint vei = 0; vei < vesize; vei++) {
+                                       Entry *entry = ventries->get(vei);
+                                       if (entry->getType() == TypeLastMessage) {
+                                               LastMessage *lastMessage = (LastMessage *)entry;
+                                               if ((lastMessage->getMachineID() == localMachineId) && (lastMessage->getSequenceNumber() == lastSlotAttemptedToSend->getSequenceNumber())) {
                                                        isInserted = true;
                                                        break;
                                                }
                                        }
-
-                                       for (uint si = 0; si < newSlots->length(); si++) {
-                                               Slot *s = newSlots->get(si);
-                                               if (isInserted) {
-                                                       break;
-                                               }
-
-                                               // Process each entry in the slot
-                                               Vector<Entry *> *ventries = s->getEntries();
-                                               uint vesize = ventries->size();
-                                               for (uint vei = 0; vei < vesize; vei++) {
-                                                       Entry *entry = ventries->get(vei);
-                                                       if (entry->getType() == TypeLastMessage) {
-                                                               LastMessage *lastMessage = (LastMessage *)entry;
-                                                               if ((lastMessage->getMachineID() == localMachineId) && (lastMessage->getSequenceNumber() == lastSlotAttemptedToSend->getSequenceNumber())) {
-                                                                       isInserted = true;
-                                                                       break;
-                                                               }
-                                                       }
-                                               }
-                                       }
-
-                                       if (isInserted) {
-                                               if (newKey != NULL) {
-                                                       if (lastInsertedNewKey && (lastNewKey->getKey() == newKey->getKey()) && (lastNewKey->getMachineID() == newKey->getMachineID())) {
-                                                               newKey = NULL;
-                                                       }
-                                               }
-
-                                               SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
-                                               while (trit->hasNext()) {
-                                                       Transaction *transaction = trit->next();
-                                                       transaction->resetServerFailure();
-
-                                                       // Update which transactions parts still need to be sent
-                                                       transaction->removeSentParts(lastTransactionPartsSent->get(transaction));
-
-                                                       // Add the transaction status to the outstanding list
-                                                       outstandingTransactionStatus->put(transaction->getSequenceNumber(), transaction->getTransactionStatus());
-
-                                                       // Update the transaction status
-                                                       transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentPartial);
-
-                                                       // Check if all the transaction parts were successfully sent and if so then remove it from pending
-                                                       if (transaction->didSendAllParts()) {
-                                                               transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentFully);
-                                                               pendingTransactionQueue->remove(transaction);
-                                                       } else {
-                                                               transaction->resetServerFailure();
-                                                               // Set the transaction sequence number back to nothing
-                                                               if (!transaction->didSendAPartToServer()) {
-                                                                       transaction->setSequenceNumber(-1);
-                                                               }
-                                                       }
-                                               }
-                                               delete trit;
+                               }
+                       }
+                       
+                       if (isInserted) {
+                               if (newKey != NULL) {
+                                       if (lastInsertedNewKey && (lastNewKey->getKey() == newKey->getKey()) && (lastNewKey->getMachineID() == newKey->getMachineID())) {
+                                               newKey = NULL;
                                        }
                                }
-
+                               
                                SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
                                while (trit->hasNext()) {
                                        Transaction *transaction = trit->next();
                                        transaction->resetServerFailure();
-                                       // Set the transaction sequence number back to nothing
-                                       if (!transaction->didSendAPartToServer()) {
-                                               transaction->setSequenceNumber(-1);
+                                       
+                                       // Update which transactions parts still need to be sent
+                                       transaction->removeSentParts(lastTransactionPartsSent->get(transaction));
+                                       
+                                       // Add the transaction status to the outstanding list
+                                       outstandingTransactionStatus->put(transaction->getSequenceNumber(), transaction->getTransactionStatus());
+                                       
+                                       // Update the transaction status
+                                       transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentPartial);
+                                       
+                                       // Check if all the transaction parts were successfully sent and if so then remove it from pending
+                                       if (transaction->didSendAllParts()) {
+                                               transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentFully);
+                                               pendingTransactionQueue->remove(transaction);
+                                       } else {
+                                               transaction->resetServerFailure();
+                                               // Set the transaction sequence number back to nothing
+                                               if (!transaction->didSendAPartToServer()) {
+                                                       transaction->setSequenceNumber(-1);
+                                               }
                                        }
                                }
                                delete trit;
-
-                               if (sendSlotsReturn.getThird()->length() != 0) {
-                                       // insert into the local block chain
-                                       validateAndUpdate(sendSlotsReturn.getThird(), true);
-                               }
-                               // continue;
-                       } else {
-                               bool isInserted = false;
-                               for (uint si = 0; si < newSlots->length(); si++) {
-                                       Slot *s = newSlots->get(si);
-                                       if ((s->getSequenceNumber() == lastSlotAttemptedToSend->getSequenceNumber()) && (s->getMachineID() == localMachineId)) {
+                       }
+               }
+               
+               SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
+               while (trit->hasNext()) {
+                       Transaction *transaction = trit->next();
+                       transaction->resetServerFailure();
+                       // Set the transaction sequence number back to nothing
+                       if (!transaction->didSendAPartToServer()) {
+                               transaction->setSequenceNumber(-1);
+                       }
+               }
+               delete trit;
+               
+               if (newSlots->length() != 0) {
+                       // insert into the local block chain
+                       validateAndUpdate(newSlots, true);
+               }
+               // continue;
+       } else {
+               bool isInserted = false;
+               for (uint si = 0; si < newSlots->length(); si++) {
+                       Slot *s = newSlots->get(si);
+                       if ((s->getSequenceNumber() == lastSlotAttemptedToSend->getSequenceNumber()) && (s->getMachineID() == localMachineId)) {
+                               isInserted = true;
+                               break;
+                       }
+               }
+               
+               for (uint si = 0; si < newSlots->length(); si++) {
+                       Slot *s = newSlots->get(si);
+                       if (isInserted) {
+                               break;
+                       }
+                       
+                       // Process each entry in the slot
+                       Vector<Entry *> *entries = s->getEntries();
+                       uint eSize = entries->size();
+                       for (uint ei = 0; ei < eSize; ei++) {
+                               Entry *entry = entries->get(ei);
+                               
+                               if (entry->getType() == TypeLastMessage) {
+                                       LastMessage *lastMessage = (LastMessage *)entry;
+                                       if ((lastMessage->getMachineID() == localMachineId) && (lastMessage->getSequenceNumber() == lastSlotAttemptedToSend->getSequenceNumber())) {
                                                isInserted = true;
                                                break;
                                        }
                                }
-
-                               for (uint si = 0; si < newSlots->length(); si++) {
-                                       Slot *s = newSlots->get(si);
-                                       if (isInserted) {
-                                               break;
-                                       }
-
-                                       // Process each entry in the slot
-                                       Vector<Entry *> *entries = s->getEntries();
-                                       uint eSize = entries->size();
-                                       for (uint ei = 0; ei < eSize; ei++) {
-                                               Entry *entry = entries->get(ei);
-
-                                               if (entry->getType() == TypeLastMessage) {
-                                                       LastMessage *lastMessage = (LastMessage *)entry;
-                                                       if ((lastMessage->getMachineID() == localMachineId) && (lastMessage->getSequenceNumber() == lastSlotAttemptedToSend->getSequenceNumber())) {
-                                                               isInserted = true;
-                                                               break;
-                                                       }
-                                               }
-                                       }
+                       }
+               }
+               
+               if (isInserted) {
+                       if (newKey != NULL) {
+                               if (lastInsertedNewKey && (lastNewKey->getKey() == newKey->getKey()) && (lastNewKey->getMachineID() == newKey->getMachineID())) {
+                                       newKey = NULL;
                                }
-
-                               if (isInserted) {
-                                       if (newKey != NULL) {
-                                               if (lastInsertedNewKey && (lastNewKey->getKey() == newKey->getKey()) && (lastNewKey->getMachineID() == newKey->getMachineID())) {
-                                                       newKey = NULL;
-                                               }
-                                       }
-
-                                       SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
-                                       while (trit->hasNext()) {
-                                               Transaction *transaction = trit->next();
-                                               transaction->resetServerFailure();
-
-                                               // Update which transactions parts still need to be sent
-                                               transaction->removeSentParts(lastTransactionPartsSent->get(transaction));
-
-                                               // Add the transaction status to the outstanding list
-                                               outstandingTransactionStatus->put(transaction->getSequenceNumber(), transaction->getTransactionStatus());
-
-                                               // Update the transaction status
-                                               transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentPartial);
-
-                                               // Check if all the transaction parts were successfully sent and if so then remove it from pending
-                                               if (transaction->didSendAllParts()) {
-                                                       transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentFully);
-                                                       pendingTransactionQueue->remove(transaction);
-                                               } else {
-                                                       transaction->resetServerFailure();
-                                                       // Set the transaction sequence number back to nothing
-                                                       if (!transaction->didSendAPartToServer()) {
-                                                               transaction->setSequenceNumber(-1);
-                                                       }
-                                               }
-                                       }
-                                       delete trit;
+                       }
+                       
+                       SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
+                       while (trit->hasNext()) {
+                               Transaction *transaction = trit->next();
+                               transaction->resetServerFailure();
+                               
+                               // Update which transactions parts still need to be sent
+                               transaction->removeSentParts(lastTransactionPartsSent->get(transaction));
+                               
+                               // Add the transaction status to the outstanding list
+                               outstandingTransactionStatus->put(transaction->getSequenceNumber(), transaction->getTransactionStatus());
+                               
+                               // Update the transaction status
+                               transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentPartial);
+                               
+                               // Check if all the transaction parts were successfully sent and if so then remove it from pending
+                               if (transaction->didSendAllParts()) {
+                                       transaction->getTransactionStatus()->setStatus(TransactionStatus_StatusSentFully);
+                                       pendingTransactionQueue->remove(transaction);
                                } else {
-                                       SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
-                                       while (trit->hasNext()) {
-                                               Transaction *transaction = trit->next();
-                                               transaction->resetServerFailure();
-                                               // Set the transaction sequence number back to nothing
-                                               if (!transaction->didSendAPartToServer()) {
-                                                       transaction->setSequenceNumber(-1);
-                                               }
+                                       transaction->resetServerFailure();
+                                       // Set the transaction sequence number back to nothing
+                                       if (!transaction->didSendAPartToServer()) {
+                                               transaction->setSequenceNumber(-1);
                                        }
-                                       delete trit;
                                }
-
-                               // insert into the local block chain
-                               validateAndUpdate(newSlots, true);
                        }
+                       delete trit;
+               } else {
+                       SetIterator<Transaction *, Vector<int> *> *trit = getKeyIterator(lastTransactionPartsSent);
+                       while (trit->hasNext()) {
+                               Transaction *transaction = trit->next();
+                               transaction->resetServerFailure();
+                               // Set the transaction sequence number back to nothing
+                               if (!transaction->didSendAPartToServer()) {
+                                       transaction->setSequenceNumber(-1);
+                               }
+                       }
+                       delete trit;
                }
-       } catch (ServerException *e) {
-               throw e;
+               
+               // insert into the local block chain
+               validateAndUpdate(newSlots, true);
        }
+       return newKey;
+}
 
-
+bool Table::sendToServer(NewKey *newKey) {
+       if (hadPartialSendToServer) {
+               newKey = handlePartialSend(newKey);
+       }
 
        try {
                // While we have stuff that needs inserting into the block chain
                while ((pendingTransactionQueue->size() > 0) || (pendingSendArbitrationRounds->size() > 0) || (newKey != NULL)) {
-
-                       fromRetry = false;
-
                        if (hadPartialSendToServer) {
                                throw new Error("Should Be error free");
                        }
-
-
-
+                       
                        // If there is a new key with same name then end
                        if ((newKey != NULL) && arbitratorTable->contains(newKey->getKey())) {
                                return false;
@@ -767,10 +763,11 @@ bool Table::sendToServer(NewKey *newKey) {
                        lastTransactionPartsSent = transactionPartsSent->clone();
                        lastPendingSendArbitrationEntriesToDelete = new Vector<Entry *>(pendingSendArbitrationEntriesToDelete);
 
-                       ThreeTuple<bool, bool, Array<Slot *> *> sendSlotsReturn = sendSlotsToServer(slot, newSize, newKey != NULL);
-
-                       if (sendSlotsReturn.getFirst()) {
+                       Array<Slot *> * newSlots = NULL;
+                       bool wasInserted = false;
+                       bool sendSlotsReturn = sendSlotsToServer(slot, newSize, newKey != NULL, &wasInserted, &newSlots);
 
+                       if (sendSlotsReturn) {
                                // Did insert into the block chain
 
                                if (insertedNewKey) {
@@ -835,9 +832,9 @@ bool Table::sendToServer(NewKey *newKey) {
                        pendingSendArbitrationEntriesToDelete->clear();
                        transactionPartsSent->clear();
 
-                       if (sendSlotsReturn.getThird()->length() != 0) {
+                       if (newSlots->length() != 0) {
                                // insert into the local block chain
-                               validateAndUpdate(sendSlotsReturn.getThird(), true);
+                               validateAndUpdate(newSlots, true);
                        }
                }
 
@@ -1150,73 +1147,72 @@ Array<char> *Table::acceptDataFromLocal(Array<char> *data) {
        return returnData;
 }
 
-ThreeTuple<bool, bool, Array<Slot *> *> Table::sendSlotsToServer(Slot *slot, int newSize, bool isNewKey) {
-       bool attemptedToSendToServerTmp = attemptedToSendToServer;
-       attemptedToSendToServer = true;
 
-       bool inserted = false;
-       bool lastTryInserted = false;
+/** Method tries to send slot to server.  Returns status in tuple.
+               isInserted returns whether last un-acked send (if any) was
+               successful.  Returns whether send was confirmed.x
+ */
 
-       Array<Slot *> *array = cloud->putSlot(slot, newSize);
-       if (array == NULL) {
-               array = new Array<Slot *>(1);
-               array->set(0, slot);
+bool Table::sendSlotsToServer(Slot *slot, int newSize, bool isNewKey, bool *isInserted, Array<Slot *> **array) {
+       attemptedToSendToServer = true;
+
+       *array = cloud->putSlot(slot, newSize);
+       if (*array == NULL) {
+               *array = new Array<Slot *>(1);
+               (*array)->set(0, slot);
                rejectedSlotVector->clear();
-               inserted = true;
+               *isInserted = false;
+               return true;
        } else {
-               if (array->length() == 0) {
+               if ((*array)->length() == 0) {
                        throw new Error("Server Error: Did not send any slots");
                }
 
-               // if (attemptedToSendToServerTmp) {
                if (hadPartialSendToServer) {
-
-                       bool isInserted = false;
-                       uint size = array->length();
+                       *isInserted = false;
+                       uint size = (*array)->length();
                        for (uint i = 0; i < size; i++) {
-                               Slot *s = array->get(i);
+                               Slot *s = (*array)->get(i);
                                if ((s->getSequenceNumber() == slot->getSequenceNumber()) && (s->getMachineID() == localMachineId)) {
-                                       isInserted = true;
+                                       *isInserted = true;
                                        break;
                                }
                        }
 
-                       for (uint i = 0; i < size; i++) {
-                               Slot *s = array->get(i);
-                               if (isInserted) {
-                                       break;
-                               }
-
-                               // Process each entry in the slot
-                               Vector<Entry *> *entries = s->getEntries();
-                               uint eSize = entries->size();
-                               for (uint ei = 0; ei < eSize; ei++) {
-                                       Entry *entry = entries->get(ei);
-
-                                       if (entry->getType() == TypeLastMessage) {
-                                               LastMessage *lastMessage = (LastMessage *)entry;
-
-                                               if ((lastMessage->getMachineID() == localMachineId) && (lastMessage->getSequenceNumber() == slot->getSequenceNumber())) {
-                                                       isInserted = true;
-                                                       break;
+                       //Also need to see if other machines acknowledged our message
+                       if (!(*isInserted)) {
+                               for (uint i = 0; i < size; i++) {
+                                       Slot *s = (*array)->get(i);
+                                       
+                                       // Process each entry in the slot
+                                       Vector<Entry *> *entries = s->getEntries();
+                                       uint eSize = entries->size();
+                                       for (uint ei = 0; ei < eSize; ei++) {
+                                               Entry *entry = entries->get(ei);
+                                               
+                                               if (entry->getType() == TypeLastMessage) {
+                                                       LastMessage *lastMessage = (LastMessage *)entry;
+                                                       
+                                                       if ((lastMessage->getMachineID() == localMachineId) && (lastMessage->getSequenceNumber() == slot->getSequenceNumber())) {
+                                                               *isInserted = true;
+                                                               goto done;
+                                                       }
                                                }
                                        }
                                }
                        }
-
-                       if (!isInserted) {
+               done:
+                       if (!(*isInserted)) {
                                rejectedSlotVector->add(slot->getSequenceNumber());
-                               lastTryInserted = false;
-                       } else {
-                               lastTryInserted = true;
                        }
+                       
+                       return false;
                } else {
                        rejectedSlotVector->add(slot->getSequenceNumber());
-                       lastTryInserted = false;
+                       *isInserted = false;
+                       return false;
                }
        }
-
-       return ThreeTuple<bool, bool, Array<Slot *> *>(inserted, lastTryInserted, array);
 }
 
 /**
@@ -1694,7 +1690,7 @@ void Table::arbitrateFromServer() {
        qsort(transactionSequenceNumbers->expose(), transactionSequenceNumbers->size(), sizeof(int64_t), compareInt64);
 
        // Collection of key value pairs that are
-       Hashtable<IoTString *, KeyValue *> *speculativeTableTmp = new Hashtable<IoTString *, KeyValue *>();
+       Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals> *speculativeTableTmp = new Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals>();
 
        // The last transaction arbitrated on
        int64_t lastTransactionCommitted = -1;
@@ -1778,7 +1774,7 @@ void Table::arbitrateFromServer() {
                localArbitrationSequenceNumber++;
 
                // Add all the new keys to the commit
-               SetIterator<IoTString *, KeyValue *> *spit = getKeyIterator(speculativeTableTmp);
+               SetIterator<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals> *spit = getKeyIterator(speculativeTableTmp);
                while (spit->hasNext()) {
                        IoTString *string = spit->next();
                        KeyValue *kv = speculativeTableTmp->get(string);
index a97d124..99abb55 100644 (file)
@@ -60,10 +60,10 @@ private:
 
 
        /* Data Structures  */
-       Hashtable<IoTString *, KeyValue *> *committedKeyValueTable;// Table of committed key value pairs
-       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<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals> *committedKeyValueTable;// Table of committed key value pairs
+       Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals> *speculatedKeyValueTable;    // Table of speculated key value pairs, if there is a speculative value
+       Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals> *pendingTransactionSpeculatedKeyValueTable;  // Table of speculated key value pairs, if there is a speculative value from the pending transactions
+       Hashtable<IoTString *, NewKey *, uintptr_t, 0, hashString, StringEquals> *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, Hashset<RejectedMessage *> *> *rejectedMessageWatchVectorTable;      // Table of machine Ids and the set of rejected messages they have not seen yet
        Hashtable<IoTString *, int64_t, uintptr_t, 0, hashString, StringEquals> *arbitratorTable;// Table of keys and their arbitrators
@@ -74,7 +74,7 @@ private:
        Hashtable<int64_t, Transaction *> *liveTransactionBySequenceNumberTable;        // live transaction grouped by the sequence number
        Hashtable<Pair<int64_t, int64_t> *, Transaction *, uintptr_t, 0, pairHashFunction, pairEquals> *liveTransactionByTransactionIdTable;    // live transaction grouped by the transaction ID
        Hashtable<int64_t, Hashtable<int64_t, Commit *> *> *liveCommitsTable;
-       Hashtable<IoTString *, Commit *> *liveCommitsByKeyTable;
+       Hashtable<IoTString *, Commit *, uintptr_t, 0, hashString, StringEquals> *liveCommitsByKeyTable;
        Hashtable<int64_t, int64_t> *lastCommitSeenSequenceNumberByArbitratorTable;
        Vector<int64_t> *rejectedSlotVector;    // Vector of rejected slots that have yet to be sent to the server
        Vector<Transaction *> *pendingTransactionQueue;
@@ -97,9 +97,10 @@ private:
         */
        void setResizeThreshold();
        bool sendToServer(NewKey *newKey);
+       NewKey * handlePartialSend(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);
+       bool sendSlotsToServer(Slot *slot, int newSize, bool isNewKey, bool * wasInserted, Array<Slot *> **array);
        /**
         * Returns false if a resize was needed
         */
index c7785c5..1f862df 100644 (file)
@@ -264,7 +264,7 @@ void Transaction::decodeTransactionData() {
        }
 }
 
-bool Transaction::evaluateGuard(Hashtable<IoTString *, KeyValue *> *committedKeyValueTable, Hashtable<IoTString *, KeyValue *> *speculatedKeyValueTable, Hashtable<IoTString *, KeyValue *> *pendingTransactionSpeculatedKeyValueTable) {
+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();
index d6453d6..08eebea 100644 (file)
@@ -2,6 +2,7 @@
 #define TRANSACTION_H
 #include "common.h"
 #include "Pair.h"
+#include "IoTString.h"
 
 class Transaction {
 private:
@@ -53,6 +54,6 @@ public:
        Pair<int64_t, int64_t> *getId();
        void setDead();
        TransactionPart *getPart(int32_t index);
-       bool evaluateGuard(Hashtable<IoTString *, KeyValue *> *committedKeyValueTable, Hashtable<IoTString *, KeyValue *> *speculatedKeyValueTable, Hashtable<IoTString *, KeyValue *> *pendingTransactionSpeculatedKeyValueTable);
+       bool 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);
 };
 #endif