version2/src/C/Table.h

   1 #ifndef Table_H
   2 #define Table_H
   3 #include "common.h"
   4 #include "Pair.h"
   5 #include "ThreeTuple.h"
   6 #include "IoTString.h"
   7
   8 /**
   9  * IoTTable data structure.  Provides client interface.
  10  * @author Brian Demsky
  11  * @version 1.0
  12  */
  13
  14 /* Constants */
  15 #define Table_FREE_SLOTS 2
  16 // Number of slots that should be kept free // 10
  17 #define Table_SKIP_THRESHOLD 10
  18 #define Table_RESIZE_MULTIPLE ((double)1.2)
  19 #define Table_RESIZE_THRESHOLD ((double)0.75)
  20 #define Table_REJECTED_THRESHOLD 5
  21
  22 class Table {
  23 private:
  24         /* Helper Objects */
  25         SlotBuffer *buffer;
  26         CloudComm *cloud;
  27         SecureRandom *random;
  28         TableStatus *liveTableStatus;
  29         PendingTransaction *pendingTransactionBuilder;  // Pending Transaction used in building a Pending Transaction
  30         Transaction *lastPendingTransactionSpeculatedOn;        // Last transaction that was speculated on from the pending transaction
  31         Transaction *firstPendingTransaction;   // first transaction in the pending transaction list
  32
  33         /* Variables */
  34         int numberOfSlots;      // Number of slots stored in buffer
  35         int bufferResizeThreshold;// Threshold on the number of live slots before a resize is needed
  36         int64_t liveSlotCount;// Number of currently live slots
  37         int64_t oldestLiveSlotSequenceNumver;   // Smallest sequence number of the slot with a live entry
  38         int64_t localMachineId; // Machine ID of this client device
  39         int64_t sequenceNumber; // Largest sequence number a client has received
  40         int64_t localSequenceNumber;
  41
  42         //  int smallestTableStatusSeen = -1; // Smallest Table Status that was seen in the latest slots sent from the server
  43         //  int largestTableStatusSeen = -1; // Largest Table Status that was seen in the latest slots sent from the server
  44         int64_t localTransactionSequenceNumber; // Local sequence number counter for transactions
  45         int64_t lastTransactionSequenceNumberSpeculatedOn;      // the last transaction that was speculated on
  46         int64_t oldestTransactionSequenceNumberSpeculatedOn;    // the oldest transaction that was speculated on
  47         int64_t localArbitrationSequenceNumber;
  48         bool hadPartialSendToServer;
  49         bool attemptedToSendToServer;
  50         int64_t expectedsize;
  51         bool didFindTableStatus;
  52         int64_t currMaxSize;
  53
  54         Slot *lastSlotAttemptedToSend;
  55         bool lastIsNewKey;
  56         int lastNewSize;
  57         Hashtable<Transaction *, Vector<int32_t> *> *lastTransactionPartsSent;
  58         Vector<Entry *> *lastPendingSendArbitrationEntriesToDelete;
  59         NewKey *lastNewKey;
  60
  61
  62         /* Data Structures  */
  63         Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals> *committedKeyValueTable;// Table of committed key value pairs
  64         Hashtable<IoTString *, KeyValue *, uintptr_t, 0, hashString, StringEquals> *speculatedKeyValueTable;    // Table of speculated key value pairs, if there is a speculative value
  65         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
  66         Hashtable<IoTString *, NewKey *, uintptr_t, 0, hashString, StringEquals> *liveNewKeyTable;      // Table of live new keys
  67         Hashtable<int64_t, Pair<int64_t, Liveness *> *> *lastMessageTable;      // Last message sent by a client machine id -> (Seq Num, Slot or LastMessage);
  68         Hashtable<int64_t, Hashset<RejectedMessage *> *> *rejectedMessageWatchVectorTable;      // Table of machine Ids and the set of rejected messages they have not seen yet
  69         Hashtable<IoTString *, int64_t, uintptr_t, 0, hashString, StringEquals> *arbitratorTable;// Table of keys and their arbitrators
  70         Hashtable<Pair<int64_t, int64_t> *, Abort *, uintptr_t, 0, pairHashFunction, pairEquals> *liveAbortTable;// Table live abort messages
  71         Hashtable<int64_t, Hashtable<Pair<int64_t, int32_t> *, TransactionPart *, uintptr_t, 0, pairHashFunction, pairEquals> *> *newTransactionParts;  // transaction parts that are seen in this latest round of slots from the server
  72         Hashtable<int64_t, Hashtable<Pair<int64_t, int32_t> *, CommitPart *, uintptr_t, 0, pairHashFunction, pairEquals> *> *newCommitParts;    // commit parts that are seen in this latest round of slots from the server
  73         Hashtable<int64_t, int64_t> *lastArbitratedTransactionNumberByArbitratorTable;  // Last transaction sequence number that an arbitrator arbitrated on
  74         Hashtable<int64_t, Transaction *> *liveTransactionBySequenceNumberTable;        // live transaction grouped by the sequence number
  75         Hashtable<Pair<int64_t, int64_t> *, Transaction *, uintptr_t, 0, pairHashFunction, pairEquals> *liveTransactionByTransactionIdTable;    // live transaction grouped by the transaction ID
  76         Hashtable<int64_t, Hashtable<int64_t, Commit *> *> *liveCommitsTable;
  77         Hashtable<IoTString *, Commit *, uintptr_t, 0, hashString, StringEquals> *liveCommitsByKeyTable;
  78         Hashtable<int64_t, int64_t> *lastCommitSeenSequenceNumberByArbitratorTable;
  79         Vector<int64_t> *rejectedSlotVector;    // Vector of rejected slots that have yet to be sent to the server
  80         Vector<Transaction *> *pendingTransactionQueue;
  81         Vector<ArbitrationRound *> *pendingSendArbitrationRounds;
  82         Vector<Entry *> *pendingSendArbitrationEntriesToDelete;
  83         Hashtable<Transaction *, Vector<int32_t> *> *transactionPartsSent;
  84         Hashtable<int64_t, TransactionStatus *> *outstandingTransactionStatus;
  85         Hashtable<int64_t, Abort *> *liveAbortsGeneratedByLocal;
  86         Hashset<Pair<int64_t, int64_t> *, uintptr_t, 0, pairHashFunction, pairEquals> *offlineTransactionsCommittedAndAtServer;
  87         Hashtable<int64_t, Pair<IoTString *, int32_t> *> *localCommunicationTable;
  88         Hashtable<int64_t, int64_t> *lastTransactionSeenFromMachineFromServer;
  89         Hashtable<int64_t, int64_t> *lastArbitrationDataLocalSequenceNumberSeenFromArbitrator;
  90         bool lastInsertedNewKey;
  91         int64_t lastSeqNumArbOn;
  92
  93
  94         void init();
  95         /**
  96          * Recalculate the new resize threshold
  97          */
  98         void setResizeThreshold();
  99         bool sendToServer(NewKey *newKey);
 100         NewKey * handlePartialSend(NewKey * newKey);
 101         bool checkSend(Array<Slot *> * array, Slot *checkSlot);
 102
 103         bool updateFromLocal(int64_t machineId);
 104         Pair<bool, bool> sendTransactionToLocal(Transaction *transaction);
 105         bool sendSlotsToServer(Slot *slot, int newSize, bool isNewKey, bool * wasInserted, Array<Slot *> **array);
 106         /**
 107          * Returns false if a resize was needed
 108          */
 109         bool fillSlot(Slot *slot, bool resize, NewKey *newKeyEntry, int & newSize, bool & insertedKey);
 110         void doRejectedMessages(Slot *s);
 111
 112         ThreeTuple<bool, bool, int64_t> doMandatoryResuce(Slot *slot, bool resize);
 113
 114         void  doOptionalRescue(Slot *s, bool seenliveslot, int64_t seqn, bool resize);
 115         /**
 116          * Checks for malicious activity and updates the local copy of the block chain.
 117          */
 118         void validateAndUpdate(Array<Slot *> *newSlots, bool acceptUpdatesToLocal);
 119
 120         void updateLiveStateFromServer();
 121
 122         void updateLiveStateFromLocal();
 123
 124         void initExpectedSize(int64_t firstSequenceNumber, int64_t numberOfSlots);
 125
 126         void updateExpectedSize();
 127
 128
 129         /**
 130          * Check the size of the block chain to make sure there are enough slots sent back by the server.
 131          * This is only called when we have a gap between the slots that we have locally and the slots
 132          * sent by the server therefore in the slots sent by the server there will be at least 1 Table
 133          * status message
 134          */
 135         void checkNumSlots(int numberOfSlots);
 136
 137         void updateCurrMaxSize(int newmaxsize) { currMaxSize = newmaxsize; }
 138
 139
 140         /**
 141          * Update the size of of the local buffer if it is needed.
 142          */
 143         void commitNewMaxSize();
 144
 145         /**
 146          * Process the new transaction parts from this latest round of slots received from the server
 147          */
 148         void processNewTransactionParts();
 149
 150
 151
 152         void arbitrateFromServer();
 153
 154         Pair<bool, bool> arbitrateOnLocalTransaction(Transaction *transaction);
 155
 156         /**
 157          * 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
 158          */
 159         bool compactArbitrationData();
 160
 161         /**
 162          * Update all the commits and the committed tables, sets dead the dead transactions
 163          */
 164         bool updateCommittedTable();
 165
 166         /**
 167          * Create the speculative table from transactions that are still live and have come from the cloud
 168          */
 169         bool updateSpeculativeTable(bool didProcessNewCommits);
 170
 171         /**
 172          * Create the pending transaction speculative table from transactions that are still in the pending transaction buffer
 173          */
 174         void updatePendingTransactionSpeculativeTable(bool didProcessNewCommitsOrSpeculate);
 175
 176         /**
 177          * Set dead and remove from the live transaction tables the transactions that are dead
 178          */
 179         void updateLiveTransactionsAndStatus();
 180
 181         /**
 182          * Process this slot, entry by entry.  Also update the latest message sent by slot
 183          */
 184         void processSlot(SlotIndexer *indexer, Slot *slot, bool acceptUpdatesToLocal, Hashset<int64_t> *machineSet);
 185
 186         /**
 187          * Update the last message that was sent for a machine Id
 188          */
 189         void processEntry(LastMessage *entry, Hashset<int64_t> *machineSet);
 190
 191         /**
 192          * Add the new key to the arbitrators table and update the set of live new keys (in case of a rescued new key message)
 193          */
 194         void processEntry(NewKey *entry);
 195
 196         /**
 197          * Process new table status entries and set dead the old ones as new ones come in.
 198          * keeps track of the largest and smallest table status seen in this current round
 199          * of updating the local copy of the block chain
 200          */
 201         void processEntry(TableStatus *entry, int64_t seq);
 202
 203         /**
 204          * Check old messages to see if there is a block chain violation. Also
 205          */
 206         void processEntry(RejectedMessage *entry, SlotIndexer *indexer);
 207
 208         /**
 209          * Check if this abort is live, if not then save it so we can kill it later.
 210          * update the last transaction number that was arbitrated on.
 211          */
 212         void processEntry(Abort *entry);
 213
 214         /**
 215          * Set dead the transaction part if that transaction is dead and keep track of all new parts
 216          */
 217         void processEntry(TransactionPart *entry);
 218
 219         /**
 220          * Process new commit entries and save them for future use.  Delete duplicates
 221          */
 222         void processEntry(CommitPart *entry);
 223
 224         /**
 225          * Update the last message seen table.  Update and set dead the appropriate RejectedMessages as clients see them.
 226          * Updates the live aborts, removes those that are dead and sets them dead.
 227          * Check that the last message seen is correct and that there is no mismatch of our own last message or that
 228          * other clients have not had a rollback on the last message.
 229          */
 230         void updateLastMessage(int64_t machineId, int64_t seqNum, Liveness *liveness, bool acceptUpdatesToLocal, Hashset<int64_t> *machineSet);
 231
 232         /**
 233          * Add a rejected message entry to the watch set to keep track of which clients have seen that
 234          * rejected message entry and which have not.
 235          */
 236         void addWatchVector(int64_t machineId, RejectedMessage *entry);
 237
 238         /**
 239          * Check if the HMAC chain is not violated
 240          */
 241         void checkHMACChain(SlotIndexer *indexer, Array<Slot *> *newSlots);
 242
 243
 244 public:
 245         Table(IoTString *baseurl, IoTString *password, int64_t _localMachineId, int listeningPort);
 246         Table(CloudComm *_cloud, int64_t _localMachineId);
 247         ~Table();
 248
 249         /**
 250          * Initialize the table by inserting a table status as the first entry into the table status
 251          * also initialize the crypto stuff.
 252          */
 253         void initTable();
 254
 255         /**
 256          * Rebuild the table from scratch by pulling the latest block chain from the server.
 257          */
 258
 259         void rebuild();
 260         void addLocalCommunication(int64_t arbitrator, IoTString *hostName, int portNumber);
 261         int64_t getArbitrator(IoTString *key);
 262         void close();
 263         IoTString *getCommitted(IoTString *key);
 264         IoTString *getSpeculative(IoTString *key);
 265         IoTString *getCommittedAtomic(IoTString *key);
 266         IoTString *getSpeculativeAtomic(IoTString *key);
 267         bool update();
 268         bool createNewKey(IoTString *keyName, int64_t machineId);
 269         void startTransaction();
 270         void put(IoTString *key, IoTString *value);
 271         TransactionStatus *commitTransaction();
 272
 273         /**
 274          * Get the machine ID for this client
 275          */
 276         int64_t getMachineId() { return localMachineId; }
 277
 278         /**
 279          * Decrement the number of live slots that we currently have
 280          */
 281         void decrementLiveCount() { liveSlotCount--; }
 282         int64_t getLocalSequenceNumber();
 283         Array<char> *acceptDataFromLocal(Array<char> *data);
 284 };
 285
 286 #endif