version2/src/java/iotcloud/Transaction.java

   1 package iotcloud;
   2
   3 import java.util.Map;
   4 import java.util.Set;
   5 import java.util.List;
   6 import java.util.ArrayList;
   7 import java.util.HashMap;
   8 import java.util.HashSet;
   9 import java.nio.ByteBuffer;
  10
  11 class Transaction {
  12
  13     private Map<Integer, TransactionPart> parts = null;
  14     private Set<Integer> missingParts = null;
  15     private List<Integer> partsPendingSend = null;
  16     private boolean isComplete = false;
  17     private boolean hasLastPart = false;
  18     private Set<KeyValue> keyValueGuardSet = null;
  19     private Set<KeyValue> keyValueUpdateSet = null;
  20     private boolean isDead = false;
  21     private long sequenceNumber = -1;
  22     private long clientLocalSequenceNumber = -1;
  23     private long arbitratorId = -1;
  24     private long machineId = -1;
  25     private Pair<Long, Long> transactionId = null;
  26
  27     private int nextPartToSend = 0;
  28     private boolean didSendAPartToServer = false;
  29
  30     private TransactionStatus transactionStatus = null;
  31
  32     private boolean hadServerFailure = false;
  33
  34     public Transaction() {
  35         parts = new HashMap<Integer, TransactionPart>();
  36         keyValueGuardSet = new HashSet<KeyValue>();
  37         keyValueUpdateSet = new HashSet<KeyValue>();
  38         partsPendingSend = new ArrayList<Integer>();
  39     }
  40
  41     public void addPartEncode(TransactionPart newPart) {
  42         parts.put(newPart.getPartNumber(), newPart);
  43         partsPendingSend.add(newPart.getPartNumber());
  44
  45         // Get the sequence number and other important information
  46         sequenceNumber = newPart.getSequenceNumber();
  47         arbitratorId = newPart.getArbitratorId();
  48         transactionId = newPart.getTransactionId();
  49         clientLocalSequenceNumber = newPart.getClientLocalSequenceNumber();
  50         machineId = newPart.getMachineId();
  51
  52         isComplete = true;
  53     }
  54
  55     public void addPartDecode(TransactionPart newPart) {
  56
  57         if (isDead) {
  58             // If dead then just kill this part and move on
  59             newPart.setDead();
  60             return;
  61         }
  62
  63         // Get the sequence number and other important information
  64         sequenceNumber = newPart.getSequenceNumber();
  65         arbitratorId = newPart.getArbitratorId();
  66         transactionId = newPart.getTransactionId();
  67         clientLocalSequenceNumber = newPart.getClientLocalSequenceNumber();
  68         machineId = newPart.getMachineId();
  69
  70         TransactionPart previoslySeenPart = parts.put(newPart.getPartNumber(), newPart);
  71
  72         if (previoslySeenPart != null) {
  73             // Set dead the old one since the new one is a rescued version of this part
  74             previoslySeenPart.setDead();
  75         } else if (newPart.isLastPart()) {
  76             missingParts = new HashSet<Integer>();
  77             hasLastPart = true;
  78
  79             for (int i = 0; i < newPart.getPartNumber(); i++) {
  80                 if (parts.get(i) == null) {
  81                     missingParts.add(i);
  82                 }
  83             }
  84         }
  85
  86         if (!isComplete && hasLastPart) {
  87
  88             // We have seen this part so remove it from the set of missing parts
  89             missingParts.remove(newPart.getPartNumber());
  90
  91             // Check if all the parts have been seen
  92             if (missingParts.size() == 0) {
  93
  94                 // We have all the parts
  95                 isComplete = true;
  96
  97                 // Decode all the parts and create the key value guard and update sets
  98                 decodeTransactionData();
  99             }
 100         }
 101     }
 102
 103     public void addUpdateKV(KeyValue kv) {
 104         keyValueUpdateSet.add(kv);
 105     }
 106
 107     public void addGuardKV(KeyValue kv) {
 108         keyValueGuardSet.add(kv);
 109     }
 110
 111
 112     public long getSequenceNumber() {
 113         return sequenceNumber;
 114     }
 115
 116     public void setSequenceNumber(long _sequenceNumber) {
 117         sequenceNumber = _sequenceNumber;
 118
 119         for (Integer i : parts.keySet()) {
 120             parts.get(i).setSequenceNumber(sequenceNumber);
 121         }
 122     }
 123
 124     public long getClientLocalSequenceNumber() {
 125         return clientLocalSequenceNumber;
 126     }
 127
 128     public Map<Integer, TransactionPart> getParts() {
 129         return parts;
 130     }
 131
 132     public boolean didSendAPartToServer() {
 133         return didSendAPartToServer;
 134     }
 135
 136     public void resetNextPartToSend() {
 137         nextPartToSend = 0;
 138     }
 139
 140     public TransactionPart getNextPartToSend() {
 141         if ((partsPendingSend.size() == 0) || (partsPendingSend.size() == nextPartToSend)) {
 142             return null;
 143         }
 144         TransactionPart part = parts.get(partsPendingSend.get(nextPartToSend));
 145         nextPartToSend++;
 146         return part;
 147     }
 148
 149
 150     public void setServerFailure() {
 151         hadServerFailure = true;
 152     }
 153
 154     public boolean getServerFailure() {
 155         return hadServerFailure;
 156     }
 157
 158
 159     public void resetServerFailure() {
 160         hadServerFailure = false;
 161     }
 162
 163
 164     public void setTransactionStatus(TransactionStatus _transactionStatus) {
 165         transactionStatus = _transactionStatus;
 166     }
 167
 168     public TransactionStatus getTransactionStatus() {
 169         return transactionStatus;
 170     }
 171
 172     public void removeSentParts(List<Integer> sentParts) {
 173         nextPartToSend = 0;
 174         partsPendingSend.removeAll(sentParts);
 175         didSendAPartToServer = true;
 176         transactionStatus.setTransactionSequenceNumber(sequenceNumber);
 177     }
 178
 179     public boolean didSendAllParts() {
 180         return partsPendingSend.isEmpty();
 181     }
 182
 183     public Set<KeyValue> getKeyValueUpdateSet() {
 184         return keyValueUpdateSet;
 185     }
 186
 187     public int getNumberOfParts() {
 188         return parts.size();
 189     }
 190
 191     public long getMachineId() {
 192         return machineId;
 193     }
 194
 195     public long getArbitrator() {
 196         return arbitratorId;
 197     }
 198
 199     public boolean isComplete() {
 200         return isComplete;
 201     }
 202
 203     public Pair<Long, Long> getId() {
 204         return transactionId;
 205     }
 206
 207     public void setDead() {
 208         if (isDead) {
 209             // Already dead
 210             return;
 211         }
 212
 213         // Set dead
 214         isDead = true;
 215
 216         // Make all the parts of this transaction dead
 217         for (Integer partNumber : parts.keySet()) {
 218             TransactionPart part = parts.get(partNumber);
 219             part.setDead();
 220         }
 221     }
 222
 223     public TransactionPart getPart(int index) {
 224         return parts.get(index);
 225     }
 226
 227     private void decodeTransactionData() {
 228
 229         // Calculate the size of the data section
 230         int dataSize = 0;
 231         for (int i = 0; i < parts.keySet().size(); i++) {
 232             TransactionPart tp = parts.get(i);
 233             dataSize += tp.getDataSize();
 234         }
 235
 236         byte[] combinedData = new byte[dataSize];
 237         int currentPosition = 0;
 238
 239         // Stitch all the data sections together
 240         for (int i = 0; i < parts.keySet().size(); i++) {
 241             TransactionPart tp = parts.get(i);
 242             System.arraycopy(tp.getData(), 0, combinedData, currentPosition, tp.getDataSize());
 243             currentPosition += tp.getDataSize();
 244         }
 245
 246         // Decoder Object
 247         ByteBuffer bbDecode = ByteBuffer.wrap(combinedData);
 248
 249         // Decode how many key value pairs need to be decoded
 250         int numberOfKVGuards = bbDecode.getInt();
 251         int numberOfKVUpdates = bbDecode.getInt();
 252
 253         // Decode all the guard key values
 254         for (int i = 0; i < numberOfKVGuards; i++) {
 255             KeyValue kv = (KeyValue)KeyValue.decode(bbDecode);
 256             keyValueGuardSet.add(kv);
 257         }
 258
 259         // Decode all the updates key values
 260         for (int i = 0; i < numberOfKVUpdates; i++) {
 261             KeyValue kv = (KeyValue)KeyValue.decode(bbDecode);
 262             keyValueUpdateSet.add(kv);
 263         }
 264     }
 265
 266     public boolean evaluateGuard(Map<IoTString, KeyValue> committedKeyValueTable, Map<IoTString, KeyValue> speculatedKeyValueTable, Map<IoTString, KeyValue> pendingTransactionSpeculatedKeyValueTable) {
 267         for (KeyValue kvGuard : keyValueGuardSet) {
 268
 269             // First check if the key is in the speculative table, this is the value of the latest assumption
 270             KeyValue kv = null;
 271
 272             // If we have a speculation table then use it first
 273             if (pendingTransactionSpeculatedKeyValueTable != null) {
 274                 kv = pendingTransactionSpeculatedKeyValueTable.get(kvGuard.getKey());
 275             }
 276
 277             // If we have a speculation table then use it first
 278             if ((kv == null) && (speculatedKeyValueTable != null)) {
 279                 kv = speculatedKeyValueTable.get(kvGuard.getKey());
 280             }
 281
 282             if (kv == null) {
 283                 // if it is not in the speculative table then check the committed table and use that
 284                 // value as our latest assumption
 285                 kv = committedKeyValueTable.get(kvGuard.getKey());
 286             }
 287
 288             if (kvGuard.getValue() != null) {
 289                 if ((kv == null) || (!kvGuard.getValue().equals(kv.getValue()))) {
 290
 291
 292                     if (kv != null) {
 293                         System.out.println(kvGuard.getValue() + "       " + kv.getValue());
 294                     } else {
 295                         System.out.println(kvGuard.getValue() + "       " + kv);
 296                     }
 297
 298                     return false;
 299                 }
 300             } else {
 301                 if (kv != null) {
 302                     return false;
 303                 }
 304             }
 305         }
 306         return true;
 307     }
 308 }