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