final public class Table {
/* Constants */
- static final int FREE_SLOTS = 10; // Number of slots that should be kept free
+ static final int FREE_SLOTS = 2; // Number of slots that should be kept free // 10
static final int SKIP_THRESHOLD = 10;
static final double RESIZE_MULTIPLE = 1.2;
static final double RESIZE_THRESHOLD = 0.75;
private long oldestLiveSlotSequenceNumver = 0; // Smallest sequence number of the slot with a live entry
private long localMachineId = 0; // Machine ID of this client device
private long sequenceNumber = 0; // Largest sequence number a client has received
+ private long localSequenceNumber = 0;
+
// private int smallestTableStatusSeen = -1; // Smallest Table Status that was seen in the latest slots sent from the server
// private int largestTableStatusSeen = -1; // Largest Table Status that was seen in the latest slots sent from the server
private long localTransactionSequenceNumber = 0; // Local sequence number counter for transactions
int livec = 0;
int deadc = 0;
+
+ int casdasd = 0;
+
+ int liveslo = 0;
+
for (long i = o; i < (n + 1); i++) {
Slot s = buffer.getSlot(i);
+
+ if (s.isLive()) {
+ liveslo++;
+ }
+
Vector<Entry> entries = s.getEntries();
for (Entry e : entries) {
if (e.isLive()) {
int type = e.getType();
+
+
+ if (type == 6) {
+ RejectedMessage rej = (RejectedMessage)e;
+ casdasd++;
+
+ System.out.println(rej.getMachineID());
+ }
+
+
types[type] = types[type] + 1;
num++;
livec++;
System.out.println(i + " " + types[i]);
}
System.out.println("Live count: " + livec);
+ System.out.println("Live Slot count: " + liveslo);
+
System.out.println("Dead count: " + deadc);
System.out.println("Old: " + o);
System.out.println("New: " + n);
cloud.initSecurity();
// Create the first insertion into the block chain which is the table status
- Slot s = new Slot(this, 1, localMachineId);
+ Slot s = new Slot(this, 1, localMachineId, localSequenceNumber);
+ localSequenceNumber++;
TableStatus status = new TableStatus(s, numberOfSlots);
s.addEntry(status);
Slot[] array = cloud.putSlot(s, numberOfSlots);
// Just pull the latest slots from the server
Slot[] newslots = cloud.getSlots(sequenceNumber + 1);
validateAndUpdate(newslots, true);
+ sendToServer(null);
+ updateLiveTransactionsAndStatus();
+
}
// public String toString() {
return true;
} catch (Exception e) {
// e.printStackTrace();
+
+ for (Long m : localCommunicationTable.keySet()) {
+ updateFromLocal(m);
+ }
}
return false;
}
NewKey newKey = new NewKey(null, keyName, machineId);
+
if (sendToServer(newKey)) {
// If successfully inserted
return true;
bufferResizeThreshold = resizeLower - 1 + random.nextInt(numberOfSlots - resizeLower);
}
+ public long getLocalSequenceNumber() {
+ return localSequenceNumber;
+ }
+
boolean lastInsertedNewKey = false;
}
+
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) {
// If there is a new key with same name then end
if ((newKey != null) && (arbitratorTable.get(newKey.getKey()) != null)) {
- System.out.println("New Key Fail");
return false;
}
// Create the slot
- Slot slot = new Slot(this, sequenceNumber + 1, localMachineId, buffer.getSlot(sequenceNumber).getHMAC());
+ Slot slot = new Slot(this, sequenceNumber + 1, localMachineId, buffer.getSlot(sequenceNumber).getHMAC(), localSequenceNumber);
+ localSequenceNumber++;
// Try to fill the slot with data
ThreeTuple<Boolean, Integer, Boolean> fillSlotsReturn = fillSlot(slot, false, newKey);
if (transaction.didSendAllParts()) {
transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentFully);
pendingTransactionQueue.remove(transaction);
-
- for (KeyValue kv : transaction.getKeyValueUpdateSet()) {
- System.out.println("Sent: " + kv + " from: " + localMachineId + " Slot:" + slot.getSequenceNumber() + " Claimed:" + transaction.getSequenceNumber());
- }
}
}
} else {
} catch (ServerException e) {
- System.out.println("Server Failure: " + e.getType());
- for (Transaction transaction : transactionPartsSent.keySet()) {
- for (KeyValue kv : transaction.getKeyValueUpdateSet()) {
- System.out.println("Sent Error: " + kv + " " + e.getType());
- }
- }
-
if (e.getType() != ServerException.TypeInputTimeout) {
// e.printStackTrace();
return newKey == null;
}
- public synchronized boolean updateFromLocal(long machineId) {
+ private synchronized boolean updateFromLocal(long machineId) {
Pair<String, Integer> localCommunicationInformation = localCommunicationTable.get(machineId);
if (localCommunicationInformation == null) {
// Cant talk to that device locally so do nothing
bbEncode.putInt(0);
// Send by local
- byte[] returnData = cloud.sendLocalData(sendData, localCommunicationInformation.getFirst(), localCommunicationInformation.getSecond());
+ byte[] returnData = cloud.sendLocalData(sendData, localSequenceNumber, localCommunicationInformation.getFirst(), localCommunicationInformation.getSecond());
+ localSequenceNumber++;
if (returnData == null) {
// Could not contact server
// Send by local
- byte[] returnData = cloud.sendLocalData(sendData, localCommunicationInformation.getFirst(), localCommunicationInformation.getSecond());
+ byte[] returnData = cloud.sendLocalData(sendData, localSequenceNumber, localCommunicationInformation.getFirst(), localCommunicationInformation.getSecond());
+ localSequenceNumber++;
if (returnData == null) {
// Could not contact server
entry.encode(bbEncode);
}
+
+ localSequenceNumber++;
return returnData;
}
* Returns false if a resize was needed
*/
private ThreeTuple<Boolean, Integer, Boolean> fillSlot(Slot slot, boolean resize, NewKey newKeyEntry) {
+
+
int newSize = 0;
if (liveSlotCount > bufferResizeThreshold) {
resize = true; //Resize is forced
+
}
if (resize) {
if (newKeyEntry != null) {
newKeyEntry.setSlot(slot);
if (slot.hasSpace(newKeyEntry)) {
+
slot.addEntry(newKeyEntry);
inserted = true;
}
// Process each slots data
for (Slot slot : newSlots) {
processSlot(indexer, slot, acceptUpdatesToLocal, machineSet);
+
updateExpectedSize();
}
}
private void initExpectedSize(long firstSequenceNumber, long numberOfSlots) {
+ // if (didFindTableStatus) {
+ // return;
+ // }
+ long prevslots = firstSequenceNumber;
+
+
if (didFindTableStatus) {
- return;
+ // expectedsize = (prevslots < ((long) numberOfSlots)) ? (int) prevslots : expectedsize;
+ // System.out.println("Here2: " + expectedsize + " " + numberOfSlots + " " + prevslots);
+
+ } else {
+ expectedsize = (prevslots < ((long) numberOfSlots)) ? (int) prevslots : numberOfSlots;
+ // System.out.println("Here: " + expectedsize);
}
- long prevslots = firstSequenceNumber;
- expectedsize = (prevslots < ((long) numberOfSlots)) ? (int) prevslots : numberOfSlots;
+
+ // System.out.println(numberOfSlots);
+
+ didFindTableStatus = true;
currMaxSize = numberOfSlots;
}
private void updateExpectedSize() {
expectedsize++;
+
if (expectedsize > currMaxSize) {
expectedsize = currMaxSize;
}
// Change the number of local slots to the new size
numberOfSlots = (int)currMaxSize;
+
// Recalculate the resize threshold since the size of the local buffer has changed
setResizeThreshold();
}
for (Long transactionSequenceNumber : transactionSequenceNumbers) {
Transaction transaction = liveTransactionBySequenceNumberTable.get(transactionSequenceNumber);
- for (KeyValue kv : transaction.getKeyValueUpdateSet()) {
- System.out.println("Arb Seen: " + kv + " " + lastSeqNumArbOn + " " + transactionSequenceNumber + " " + localMachineId + " " + transaction.getArbitrator());
- }
// Check if this machine arbitrates for this transaction if not then we cant arbitrate this transaction
continue;
}
- for (KeyValue kv : transaction.getKeyValueUpdateSet()) {
- System.out.println("Arb Seen: " + kv + " " + lastSeqNumArbOn + " " + transactionSequenceNumber + " " + localMachineId);
- }
-
-
if (offlineTransactionsCommittedAndAtServer.contains(transaction.getId())) {
// We have seen this already locally so dont commit again
continue;
break;
}
- for (KeyValue kv : transaction.getKeyValueUpdateSet()) {
- System.out.println("Arb on: " + kv + " " + lastSeqNumArbOn + " " + transactionSequenceNumber + " " + localMachineId);
- }
-
// update the largest transaction seen by arbitrator from server
if (lastTransactionSeenFromMachineFromServer.get(transaction.getMachineId()) == null) {
// Update what the last transaction committed was for use in batch commit
lastTransactionCommitted = transactionSequenceNumber;
-
- System.out.println("Commit Generated: " + lastTransactionCommitted + " " + localMachineId);
} else {
// Guard evaluated was false so create abort
pendingSendArbitrationRounds.clear();
} else {
for (int i = 0; i < numberToDelete; i++) {
- pendingSendArbitrationRounds.remove(pendingSendArbitrationRounds.size() - 1);
+ pendingSendArbitrationRounds.removeIndex(pendingSendArbitrationRounds.size() - 1);
}
}
}
}
-
- for (KeyValue kv : commit.getKeyValueUpdateSet()) {
- System.out.println("Commit Seen: " + kv + " " + commit.getTransactionSequenceNumber() + " " + localMachineId);
- }
-
-
-
-
-
-
-
-
-
-
// Update the last arbitration data that we have seen so far
if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.get(commit.getMachineId()) != null) {
throw new Error("Server Error: Invalid HMAC Chain" + currSlot + " " + prevSlot);
}
}
-}
\ No newline at end of file
+}