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