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 updateFromLocal(int64_t machineId);
 102         Pair<bool, bool> sendTransactionToLocal(Transaction *transaction);
 103         bool sendSlotsToServer(Slot *slot, int newSize, bool isNewKey, bool * wasInserted, Array<Slot *> **array);
 104         /**
 105          * Returns false if a resize was needed
 106          */
 107         ThreeTuple<bool, int32_t, bool> fillSlot(Slot *slot, bool resize, NewKey *newKeyEntry);
 108         void doRejectedMessages(Slot *s);
 109
 110         ThreeTuple<bool, bool, int64_t> doMandatoryResuce(Slot *slot, bool resize);
 111
 112         void  doOptionalRescue(Slot *s, bool seenliveslot, int64_t seqn, bool resize);
 113         /**
 114          * Checks for malicious activity and updates the local copy of the block chain.
 115          */
 116         void validateAndUpdate(Array<Slot *> *newSlots, bool acceptUpdatesToLocal);
 117
 118         void updateLiveStateFromServer();
 119
 120         void updateLiveStateFromLocal();
 121
 122         void initExpectedSize(int64_t firstSequenceNumber, int64_t numberOfSlots);
 123
 124         void updateExpectedSize();
 125
 126
 127         /**
 128          * Check the size of the block chain to make sure there are enough slots sent back by the server.
 129          * This is only called when we have a gap between the slots that we have locally and the slots
 130          * sent by the server therefore in the slots sent by the server there will be at least 1 Table
 131          * status message
 132          */
 133         void checkNumSlots(int numberOfSlots);
 134
 135         void updateCurrMaxSize(int newmaxsize) { currMaxSize = newmaxsize; }
 136
 137
 138         /**
 139          * Update the size of of the local buffer if it is needed.
 140          */
 141         void commitNewMaxSize();
 142
 143         /**
 144          * Process the new transaction parts from this latest round of slots received from the server
 145          */
 146         void processNewTransactionParts();
 147
 148
 149
 150         void arbitrateFromServer();
 151
 152         Pair<bool, bool> arbitrateOnLocalTransaction(Transaction *transaction);
 153
 154         /**
 155          * 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
 156          */
 157         bool compactArbitrationData();
 158
 159         /**
 160          * Update all the commits and the committed tables, sets dead the dead transactions
 161          */
 162         bool updateCommittedTable();
 163
 164         /**
 165          * Create the speculative table from transactions that are still live and have come from the cloud
 166          */
 167         bool updateSpeculativeTable(bool didProcessNewCommits);
 168
 169         /**
 170          * Create the pending transaction speculative table from transactions that are still in the pending transaction buffer
 171          */
 172         void updatePendingTransactionSpeculativeTable(bool didProcessNewCommitsOrSpeculate);
 173
 174         /**
 175          * Set dead and remove from the live transaction tables the transactions that are dead
 176          */
 177         void updateLiveTransactionsAndStatus();
 178
 179         /**
 180          * Process this slot, entry by entry.  Also update the latest message sent by slot
 181          */
 182         void processSlot(SlotIndexer *indexer, Slot *slot, bool acceptUpdatesToLocal, Hashset<int64_t> *machineSet);
 183
 184         /**
 185          * Update the last message that was sent for a machine Id
 186          */
 187         void processEntry(LastMessage *entry, Hashset<int64_t> *machineSet);
 188
 189         /**
 190          * Add the new key to the arbitrators table and update the set of live new keys (in case of a rescued new key message)
 191          */
 192         void processEntry(NewKey *entry);
 193
 194         /**
 195          * Process new table status entries and set dead the old ones as new ones come in.
 196          * keeps track of the largest and smallest table status seen in this current round
 197          * of updating the local copy of the block chain
 198          */
 199         void processEntry(TableStatus *entry, int64_t seq);
 200
 201         /**
 202          * Check old messages to see if there is a block chain violation. Also
 203          */
 204         void processEntry(RejectedMessage *entry, SlotIndexer *indexer);
 205
 206         /**
 207          * Check if this abort is live, if not then save it so we can kill it later.
 208          * update the last transaction number that was arbitrated on.
 209          */
 210         void processEntry(Abort *entry);
 211
 212         /**
 213          * Set dead the transaction part if that transaction is dead and keep track of all new parts
 214          */
 215         void processEntry(TransactionPart *entry);
 216
 217         /**
 218          * Process new commit entries and save them for future use.  Delete duplicates
 219          */
 220         void processEntry(CommitPart *entry);
 221
 222         /**
 223          * Update the last message seen table.  Update and set dead the appropriate RejectedMessages as clients see them.
 224          * Updates the live aborts, removes those that are dead and sets them dead.
 225          * Check that the last message seen is correct and that there is no mismatch of our own last message or that
 226          * other clients have not had a rollback on the last message.
 227          */
 228         void updateLastMessage(int64_t machineId, int64_t seqNum, Liveness *liveness, bool acceptUpdatesToLocal, Hashset<int64_t> *machineSet);
 229
 230         /**
 231          * Add a rejected message entry to the watch set to keep track of which clients have seen that
 232          * rejected message entry and which have not.
 233          */
 234         void addWatchVector(int64_t machineId, RejectedMessage *entry);
 235
 236         /**
 237          * Check if the HMAC chain is not violated
 238          */
 239         void checkHMACChain(SlotIndexer *indexer, Array<Slot *> *newSlots);
 240
 241
 242 public:
 243         Table(IoTString *baseurl, IoTString *password, int64_t _localMachineId, int listeningPort);
 244         Table(CloudComm *_cloud, int64_t _localMachineId);
 245         ~Table();
 246
 247         /**
 248          * Initialize the table by inserting a table status as the first entry into the table status
 249          * also initialize the crypto stuff.
 250          */
 251         void initTable();
 252
 253         /**
 254          * Rebuild the table from scratch by pulling the latest block chain from the server.
 255          */
 256
 257         void rebuild();
 258         void addLocalCommunication(int64_t arbitrator, IoTString *hostName, int portNumber);
 259         int64_t getArbitrator(IoTString *key);
 260         void close();
 261         IoTString *getCommitted(IoTString *key);
 262         IoTString *getSpeculative(IoTString *key);
 263         IoTString *getCommittedAtomic(IoTString *key);
 264         IoTString *getSpeculativeAtomic(IoTString *key);
 265         bool update();
 266         bool createNewKey(IoTString *keyName, int64_t machineId);
 267         void startTransaction();
 268         void addKV(IoTString *key, IoTString *value);
 269         TransactionStatus *commitTransaction();
 270
 271         /**
 272          * Get the machine ID for this client
 273          */
 274         int64_t getMachineId() { return localMachineId; }
 275
 276         /**
 277          * Decrement the number of live slots that we currently have
 278          */
 279         void decrementLiveCount() { liveSlotCount--; }
 280         int64_t getLocalSequenceNumber();
 281         Array<char> *acceptDataFromLocal(Array<char> *data);
 282 };
 283
 284 #endif