--- /dev/null
+*.iml
+.gradle
+/local.properties
+/.idea/workspace.xml
+/.idea/libraries
+.DS_Store
+/build
+/captures
--- /dev/null
+Control
\ No newline at end of file
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+ <component name="CompilerConfiguration">
+ <resourceExtensions />
+ <wildcardResourcePatterns>
+ <entry name="!?*.java" />
+ <entry name="!?*.form" />
+ <entry name="!?*.class" />
+ <entry name="!?*.groovy" />
+ <entry name="!?*.scala" />
+ <entry name="!?*.flex" />
+ <entry name="!?*.kt" />
+ <entry name="!?*.clj" />
+ <entry name="!?*.aj" />
+ </wildcardResourcePatterns>
+ <annotationProcessing>
+ <profile default="true" name="Default" enabled="false">
+ <processorPath useClasspath="true" />
+ </profile>
+ </annotationProcessing>
+ </component>
+</project>
\ No newline at end of file
--- /dev/null
+<component name="CopyrightManager">
+ <settings default="" />
+</component>
\ No newline at end of file
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+ <component name="GradleSettings">
+ <option name="linkedExternalProjectsSettings">
+ <GradleProjectSettings>
+ <option name="distributionType" value="LOCAL" />
+ <option name="externalProjectPath" value="$PROJECT_DIR$" />
+ <option name="gradleHome" value="$APPLICATION_HOME_DIR$/gradle/gradle-2.8" />
+ <option name="gradleJvm" value="1.8" />
+ <option name="modules">
+ <set>
+ <option value="$PROJECT_DIR$" />
+ <option value="$PROJECT_DIR$/app" />
+ </set>
+ </option>
+ <option name="resolveModulePerSourceSet" value="false" />
+ </GradleProjectSettings>
+ </option>
+ </component>
+</project>
\ No newline at end of file
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+ <component name="EntryPointsManager">
+ <entry_points version="2.0" />
+ </component>
+ <component name="NullableNotNullManager">
+ <option name="myDefaultNullable" value="android.support.annotation.Nullable" />
+ <option name="myDefaultNotNull" value="android.support.annotation.NonNull" />
+ <option name="myNullables">
+ <value>
+ <list size="4">
+ <item index="0" class="java.lang.String" itemvalue="org.jetbrains.annotations.Nullable" />
+ <item index="1" class="java.lang.String" itemvalue="javax.annotation.Nullable" />
+ <item index="2" class="java.lang.String" itemvalue="edu.umd.cs.findbugs.annotations.Nullable" />
+ <item index="3" class="java.lang.String" itemvalue="android.support.annotation.Nullable" />
+ </list>
+ </value>
+ </option>
+ <option name="myNotNulls">
+ <value>
+ <list size="4">
+ <item index="0" class="java.lang.String" itemvalue="org.jetbrains.annotations.NotNull" />
+ <item index="1" class="java.lang.String" itemvalue="javax.annotation.Nonnull" />
+ <item index="2" class="java.lang.String" itemvalue="edu.umd.cs.findbugs.annotations.NonNull" />
+ <item index="3" class="java.lang.String" itemvalue="android.support.annotation.NonNull" />
+ </list>
+ </value>
+ </option>
+ </component>
+ <component name="ProjectRootManager" version="2" languageLevel="JDK_1_8" default="true" project-jdk-name="1.8" project-jdk-type="JavaSDK">
+ <output url="file://$PROJECT_DIR$/build/classes" />
+ </component>
+ <component name="ProjectType">
+ <option name="id" value="Android" />
+ </component>
+</project>
\ No newline at end of file
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+ <component name="ProjectModuleManager">
+ <modules>
+ <module fileurl="file://$PROJECT_DIR$/Control.iml" filepath="$PROJECT_DIR$/Control.iml" />
+ <module fileurl="file://$PROJECT_DIR$/app/app.iml" filepath="$PROJECT_DIR$/app/app.iml" />
+ </modules>
+ </component>
+</project>
\ No newline at end of file
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+ <component name="RunConfigurationProducerService">
+ <option name="ignoredProducers">
+ <set>
+ <option value="org.jetbrains.plugins.gradle.execution.test.runner.AllInPackageGradleConfigurationProducer" />
+ <option value="org.jetbrains.plugins.gradle.execution.test.runner.TestClassGradleConfigurationProducer" />
+ <option value="org.jetbrains.plugins.gradle.execution.test.runner.TestMethodGradleConfigurationProducer" />
+ </set>
+ </option>
+ </component>
+</project>
\ No newline at end of file
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+ <component name="VcsDirectoryMappings">
+ <mapping directory="" vcs="" />
+ </component>
+</project>
\ No newline at end of file
--- /dev/null
+apply plugin: 'com.android.application'
+
+android {
+
+ dexOptions {
+ preDexLibraries = false
+ }
+
+
+ compileSdkVersion 23
+ buildToolsVersion "25.0.2"
+
+ defaultConfig {
+ applicationId "com.example.ali.control"
+ minSdkVersion 15
+ targetSdkVersion 23
+ versionCode 1
+ versionName "1.0"
+ }
+ buildTypes {
+ release {
+ minifyEnabled false
+ proguardFiles getDefaultProguardFile('proguard-android.txt'), 'proguard-rules.pro'
+ }
+ }
+ compileOptions {
+ sourceCompatibility JavaVersion.VERSION_1_8
+ targetCompatibility JavaVersion.VERSION_1_8
+ }
+}
+
+dependencies {
+ compile fileTree(include: ['*.jar'], dir: 'libs')
+ testCompile 'junit:junit:4.12'
+ compile 'com.android.support:appcompat-v7:23.3.0'
+ compile 'com.android.support:design:23.3.0'
+// compile files('libs/bcpg-jdk15on-1.56.0.0.jar')
+// compile files('libs/bcpkix-jdk15on-1.56.0.0.jar')
+// compile files('libs/bctls-jdk15on-1.56.0.0.jar')
+// compile files('libs/core-1.56.0.0.jar')
+// compile files('libs/pg-1.54.0.0.jar')
+// compile files('libs/pkix-1.54.0.0.jar')
+// compile files('libs/prov-1.56.0.0.jar')
+}
--- /dev/null
+# Add project specific ProGuard rules here.
+# By default, the flags in this file are appended to flags specified
+# in /Users/Ali/Library/Android/sdk/tools/proguard/proguard-android.txt
+# You can edit the include path and order by changing the proguardFiles
+# directive in build.gradle.
+#
+# For more details, see
+# http://developer.android.com/guide/developing/tools/proguard.html
+
+# Add any project specific keep options here:
+
+# If your project uses WebView with JS, uncomment the following
+# and specify the fully qualified class name to the JavaScript interface
+# class:
+#-keepclassmembers class fqcn.of.javascript.interface.for.webview {
+# public *;
+#}
--- /dev/null
+package com.example.ali.control;
+
+import android.app.Application;
+import android.test.ApplicationTestCase;
+
+/**
+ * <a href="http://d.android.com/tools/testing/testing_android.html">Testing Fundamentals</a>
+ */
+public class ApplicationTest extends ApplicationTestCase<Application> {
+ public ApplicationTest() {
+ super(Application.class);
+ }
+}
\ No newline at end of file
--- /dev/null
+<?xml version="1.0" encoding="utf-8"?>
+<manifest xmlns:android="http://schemas.android.com/apk/res/android"
+ package="com.example.ali.control">
+
+
+ <application
+ android:allowBackup="true"
+ android:icon="@mipmap/ic_launcher"
+ android:label="@string/app_name"
+ android:supportsRtl="true"
+ android:theme="@style/AppTheme">
+ <activity
+ android:name=".MainActivity"
+ android:label="@string/app_name"
+ android:theme="@style/AppTheme.NoActionBar">
+ <intent-filter>
+ <action android:name="android.intent.action.MAIN" />
+
+ <category android:name="android.intent.category.LAUNCHER" />
+ </intent-filter>
+ </activity>
+ </application>
+
+
+ <uses-permission android:name="android.permission.INTERNET"/>
+ <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE"/>
+
+</manifest>
--- /dev/null
+package com.example.ali.control;
+
+import android.os.Bundle;
+import android.support.design.widget.FloatingActionButton;
+import android.support.design.widget.Snackbar;
+import android.support.v7.app.AppCompatActivity;
+import android.support.v7.widget.Toolbar;
+import android.view.View;
+import android.view.Menu;
+import android.view.MenuItem;
+import android.os.StrictMode;
+
+import android.util.Log;
+import android.os.Bundle;
+import android.support.v7.app.AppCompatActivity;
+import android.widget.CompoundButton;
+import android.widget.Switch;
+import android.widget.TextView;
+import iotcloud.*;
+import java.io.*;
+import java.net.*;
+import java.util.concurrent.*;
+import android.os.Handler;
+import android.content.*;
+
+public class MainActivity extends AppCompatActivity {
+
+ Switch switchButton1, switchButton2, switchButton3, switchButton4, switchButton5;
+ TextView textView;
+ TextView textView2;
+
+ String switchOn = "Switch is ON";
+ String switchOff = "Switch is OFF";
+
+
+ Table t1 = null;
+ Thread thread = null;
+ Semaphore mutex = new Semaphore(1);
+
+ boolean didCrash = false;
+
+ private Handler handler = new Handler();
+ private Handler handler2 = new Handler();
+
+
+ private Runnable runnable = new Runnable() {
+ @Override
+ public void run() {
+
+ if(didCrash)
+ {
+ return;
+ }
+
+
+
+ String pingTimerKey = "bulbController";
+ IoTString ipingTimerKey = new IoTString(pingTimerKey);
+
+ String pingTimerKey2 = "wemoController";
+ IoTString ipingTimerKey2 = new IoTString(pingTimerKey2);
+
+ // Insert custom code here
+ try {
+ Log.e("Ali:::::", "loop............");
+ mutex.acquire();
+ t1.update();
+ IoTString testValA1 = t1.getCommitted(ipingTimerKey);
+ IoTString testValA2 = t1.getCommitted(ipingTimerKey2);
+ mutex.release();
+
+ long l1 = 0;
+ if(testValA1 != null)
+ {
+ String s1 = testValA1.toString();
+ l1 = Long.parseLong(s1);
+ }
+
+ long l2 = 0;
+ if(testValA2 != null)
+ {
+ String s2 = testValA2.toString();
+ l2 = Long.parseLong(s2);
+ }
+
+ Log.e("Ali:::::", "S: " + System.currentTimeMillis() + " " + l1);
+
+ if((System.currentTimeMillis() - l1) > 10000)
+ {
+ textView.setText("Partition" + (System.currentTimeMillis() - l1) );
+ }
+ else if((System.currentTimeMillis() - l2) > 10000)
+ {
+ textView.setText("Partition" + (System.currentTimeMillis() - l2) );
+ }
+ else
+ {
+ textView.setText("All Good");
+ }
+ } catch (Exception e) {
+
+ }
+
+
+ // Repeat every 2 seconds
+// handler.postDelayed(runnable, 1000);
+ }
+ };
+
+
+ int count = 0;
+
+ private Runnable runnable2 = new Runnable() {
+ @Override
+ public void run() {
+
+ if(didCrash)
+ {
+ return;
+ }
+
+
+ String pingTimerKey = "bulbController";
+ IoTString ipingTimerKey = new IoTString(pingTimerKey);
+
+ String pingTimerKey2 = "wemoController";
+ IoTString ipingTimerKey2 = new IoTString(pingTimerKey2);
+
+ String pingTimerKey3 = "sensorController";
+ IoTString ipingTimerKey3 = new IoTString(pingTimerKey3);
+
+
+ String sen1 = "sensor";
+ IoTString is1 = new IoTString(sen1);
+
+
+ // Insert custom code here
+ try {
+// Log.e("Ali:::::", "loop............");
+ mutex.acquire();
+ t1.update();
+ IoTString testValA1 = t1.getCommitted(ipingTimerKey);
+ IoTString testValA2 = t1.getCommitted(ipingTimerKey2);
+ IoTString testValA3 = t1.getCommitted(ipingTimerKey3);
+
+ IoTString testValA4 = t1.getCommitted(is1);
+
+ mutex.release();
+
+
+ if(testValA4 == null) {
+ textView2.setText("No Values");
+ }
+ else
+ {
+ textView2.setText(testValA4.toString());
+ }
+
+
+ long l1 = 0;
+ if(testValA1 != null)
+ {
+ String s1 = testValA1.toString();
+ l1 = Long.parseLong(s1);
+ }
+
+ long l2 = 0;
+ if(testValA2 != null)
+ {
+ String s2 = testValA2.toString();
+ l2 = Long.parseLong(s2);
+ }
+
+ long l3 = 0;
+ if(testValA3 != null)
+ {
+ String s3 = testValA3.toString();
+ l3 = Long.parseLong(s3);
+ }
+
+
+
+ if((System.currentTimeMillis() - l1) > 10000)
+ {
+ textView.setText("Partition 1: " + (System.currentTimeMillis() - l1) );
+ }
+ else if((System.currentTimeMillis() - l2) > 10000)
+ {
+ textView.setText("Partition 2: " + (System.currentTimeMillis() - l2) );
+ }
+ else if((System.currentTimeMillis() - l3) > 10000)
+ {
+ textView.setText("Partition 3: " + (System.currentTimeMillis() - l3) );
+ }
+ else
+ {
+ textView.setText("All Good");
+ }
+
+
+
+
+ } catch (Exception e) {
+
+ }
+
+
+
+ Log.e("Ali::::", "Control Loop: " + count);
+
+
+ count++;
+ count%=3;
+
+ String a1 = "bulb1";
+ String a2 = "bulb2";
+ String a3 = "bulb3";
+ IoTString ia1 = new IoTString(a1);
+ IoTString ia2 = new IoTString(a2);
+ IoTString ia3 = new IoTString(a3);
+
+ String b1 = "wemo1";
+ String b2 = "wemo2";
+ IoTString ib1 = new IoTString(b1);
+ IoTString ib2 = new IoTString(b2);
+
+
+ String valueA = "on";
+ String valueB = "off";
+ IoTString iValueA = new IoTString(valueA);
+ IoTString iValueB = new IoTString(valueB);
+
+
+
+ // Insert custom code here
+ try {
+
+ if(count == 0)
+ {
+ mutex.acquire();
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ia1, iValueA);
+ t1.addKV(ia2, iValueB);
+ t1.addKV(ia3, iValueB);
+ t1.commitTransaction();
+ mutex.release();
+ }
+ else if(count == 1)
+ {
+ mutex.acquire();
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ia1, iValueB);
+ t1.addKV(ia2, iValueA);
+ t1.addKV(ia3, iValueB);
+ t1.commitTransaction();
+ mutex.release();
+ }
+ else if(count == 2)
+ {
+ mutex.acquire();
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ia1, iValueB);
+ t1.addKV(ia2, iValueB);
+ t1.addKV(ia3, iValueA);
+ t1.commitTransaction();
+ mutex.release();
+ }
+
+
+ } catch (Exception e) {
+
+ }
+
+
+
+ handler2.postDelayed(runnable2, 1000);
+ }
+ };
+
+
+
+
+
+
+
+ @Override
+ protected void onCreate(Bundle savedInstanceState) {
+
+ StrictMode.ThreadPolicy policy = new StrictMode.ThreadPolicy.Builder().permitAll().build();
+ StrictMode.setThreadPolicy(policy);
+
+
+
+ super.onCreate(savedInstanceState);
+ setContentView(R.layout.activity_main);
+ Toolbar toolbar = (Toolbar) findViewById(R.id.toolbar);
+ setSupportActionBar(toolbar);
+
+ FloatingActionButton fab = (FloatingActionButton) findViewById(R.id.fab);
+ fab.setOnClickListener(new View.OnClickListener() {
+ @Override
+ public void onClick(View view) {
+ Snackbar.make(view, "Replace with your own action", Snackbar.LENGTH_LONG)
+ .setAction("Action", null).show();
+ }
+ });
+
+
+
+
+
+
+
+ try {
+ Log.e("Ali::::", "Here1");
+ t1 = new Table("http://dc-6.calit2.uci.edu/test.iotcloud/", "reallysecret", 1000, -1, MainActivity.this);
+ Log.e("Ali::::", "Here2");
+ t1.rebuild(); // update
+ Log.e("Ali::::", "Here3");
+
+ } catch (Exception e) {
+
+ StringWriter sw = new StringWriter();
+ PrintWriter pw = new PrintWriter(sw);
+ e.printStackTrace(pw);
+ Log.e("ALI::::", sw.toString());
+
+ }
+
+
+ // Text Box
+ textView = (TextView) findViewById(R.id.textView);
+ textView2 = (TextView) findViewById(R.id.textView4);
+
+
+ if(didCrash)
+ {
+ textView.setText("Error Crashed.....");
+ }
+ else {
+
+
+ // Switches
+ switchButton1 = (Switch) findViewById(R.id.switch1);
+ switchButton2 = (Switch) findViewById(R.id.switch2);
+ switchButton3 = (Switch) findViewById(R.id.switch3);
+ switchButton4 = (Switch) findViewById(R.id.switch4);
+ switchButton5 = (Switch) findViewById(R.id.switch5);
+
+
+ String a1 = "bulb1";
+ String a2 = "bulb2";
+ String a3 = "bulb3";
+ IoTString ia1 = new IoTString(a1);
+ IoTString ia2 = new IoTString(a2);
+ IoTString ia3 = new IoTString(a3);
+
+ String b1 = "wemo1";
+ String b2 = "wemo2";
+ IoTString ib1 = new IoTString(b1);
+ IoTString ib2 = new IoTString(b2);
+
+
+ String valueA = "on";
+ String valueB = "off";
+ IoTString iValueA = new IoTString(valueA);
+ IoTString iValueB = new IoTString(valueB);
+
+
+ Log.e("Ali::::", "Here");
+
+
+ switchButton1.setOnCheckedChangeListener(new CompoundButton.OnCheckedChangeListener() {
+ @Override
+ public void onCheckedChanged(CompoundButton compoundButton, boolean bChecked) {
+
+ if(didCrash)
+ {
+ return;
+ }
+
+ try {
+ mutex.acquire();
+ if (bChecked) {
+
+
+ try {
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ia1, iValueA);
+ t1.commitTransaction();
+ } catch (Exception e) {
+
+ }
+
+ } else {
+
+
+ try {
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ia1, iValueB);
+ t1.commitTransaction();
+ } catch (Exception e) {
+
+ }
+ }
+ } catch (Exception e) {
+
+ }
+
+ finally {
+ mutex.release();
+ }
+ }
+ });
+
+ switchButton2.setOnCheckedChangeListener(new CompoundButton.OnCheckedChangeListener() {
+ @Override
+ public void onCheckedChanged(CompoundButton compoundButton, boolean bChecked) {
+
+ if(didCrash)
+ {
+ return;
+ }
+
+
+
+ try {
+ mutex.acquire();
+ if (bChecked) {
+
+ try {
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ia2, iValueA);
+ t1.commitTransaction();
+ } catch (Exception e) {
+
+ }
+
+ } else {
+
+ try {
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ia2, iValueB);
+ t1.commitTransaction();
+ } catch (Exception e) {
+
+ }
+ }
+ } catch (Exception e) {
+
+ }
+
+ finally {
+ mutex.release();
+ }
+ }
+ });
+
+ switchButton3.setOnCheckedChangeListener(new CompoundButton.OnCheckedChangeListener() {
+ @Override
+ public void onCheckedChanged(CompoundButton compoundButton, boolean bChecked) {
+
+ if(didCrash)
+ {
+ return;
+ }
+
+ try {
+ mutex.acquire();
+ if (bChecked) {
+
+ try {
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ia3, iValueA);
+ t1.commitTransaction();
+ } catch (Exception e) {
+
+ }
+
+ } else {
+
+ try {
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ia3, iValueB);
+ t1.commitTransaction();
+ } catch (Exception e) {
+
+ }
+ }
+ } catch (Exception e) {
+
+ }
+
+ finally {
+ mutex.release();
+ }
+ }
+ });
+
+ switchButton4.setOnCheckedChangeListener(new CompoundButton.OnCheckedChangeListener() {
+ @Override
+ public void onCheckedChanged(CompoundButton compoundButton, boolean bChecked) {
+
+ if(didCrash)
+ {
+ return;
+ }
+
+ try {
+ mutex.acquire();
+ if (bChecked) {
+
+ try {
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ib1, iValueA);
+ t1.commitTransaction();
+ } catch (Exception e) {
+
+ }
+
+ } else {
+
+ try {
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ib1, iValueB);
+ t1.commitTransaction();
+ } catch (Exception e) {
+
+ }
+ }
+ } catch (Exception e) {
+
+ }
+
+ finally {
+ mutex.release();
+ }
+ }
+ });
+
+ switchButton5.setOnCheckedChangeListener(new CompoundButton.OnCheckedChangeListener() {
+ @Override
+ public void onCheckedChanged(CompoundButton compoundButton, boolean bChecked) {
+
+ if(didCrash)
+ {
+ return;
+ }
+
+ try {
+ mutex.acquire();
+
+ if (bChecked) {
+
+ try {
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ib2, iValueA);
+ t1.commitTransaction();
+ } catch (Exception e) {
+
+ }
+
+ } else {
+
+ try {
+ t1.update();
+ t1.startTransaction();
+ t1.addKV(ib2, iValueB);
+ t1.commitTransaction();
+ } catch (Exception e) {
+
+ }
+ }
+ } catch (Exception e) {
+
+ }
+
+ finally {
+ mutex.release();
+ }
+ }
+ });
+
+// handler.post(runnable);
+ handler2.post(runnable2);
+ }
+
+ }
+
+ @Override
+ public boolean onCreateOptionsMenu(Menu menu) {
+ // Inflate the menu; this adds items to the action bar if it is present.
+ getMenuInflater().inflate(R.menu.menu_main, menu);
+ return true;
+ }
+
+ @Override
+ public boolean onOptionsItemSelected(MenuItem item) {
+ // Handle action bar item clicks here. The action bar will
+ // automatically handle clicks on the Home/Up button, so long
+ // as you specify a parent activity in AndroidManifest.xml.
+ int id = item.getItemId();
+
+ //noinspection SimplifiableIfStatement
+ if (id == R.id.action_settings) {
+ return true;
+ }
+
+ return super.onOptionsItemSelected(item);
+ }
+
+
+}
--- /dev/null
+package iotcloud;
+
+import java.nio.ByteBuffer;
+
+import java.lang.Long;
+
+/**
+ * This Entry records the abort sent by a given machine.
+ * @author Ali Younis <ayounis@uci.edu>
+ * @version 1.0
+ */
+
+
+class Abort extends Entry {
+ private long transactionClientLocalSequenceNumber = -1;
+ private long transactionSequenceNumber = -1;
+ private long sequenceNumber = -1;
+ private long transactionMachineId = -1;
+ private long transactionArbitrator = -1;
+ private long arbitratorLocalSequenceNumber = -1;
+
+ private Pair<Long, Long> abortId = null;
+
+
+ public Abort(Slot slot, long _transactionClientLocalSequenceNumber, long _transactionSequenceNumber , long _transactionMachineId, long _transactionArbitrator, long _arbitratorLocalSequenceNumber) {
+ super(slot);
+ transactionClientLocalSequenceNumber = _transactionClientLocalSequenceNumber;
+ transactionSequenceNumber = _transactionSequenceNumber;
+ transactionMachineId = _transactionMachineId;
+ transactionArbitrator = _transactionArbitrator;
+ arbitratorLocalSequenceNumber = _arbitratorLocalSequenceNumber;
+ abortId = new Pair<Long, Long>(transactionMachineId, transactionClientLocalSequenceNumber);
+ }
+
+ public Abort(Slot slot, long _transactionClientLocalSequenceNumber, long _transactionSequenceNumber, long _sequenceNumber , long _transactionMachineId, long _transactionArbitrator, long _arbitratorLocalSequenceNumber) {
+ super(slot);
+ transactionClientLocalSequenceNumber = _transactionClientLocalSequenceNumber;
+ transactionSequenceNumber = _transactionSequenceNumber;
+ sequenceNumber = _sequenceNumber;
+ transactionMachineId = _transactionMachineId;
+ transactionArbitrator = _transactionArbitrator;
+ arbitratorLocalSequenceNumber = _arbitratorLocalSequenceNumber;
+
+ abortId = new Pair<Long, Long>(transactionMachineId, transactionClientLocalSequenceNumber);
+ }
+
+ public Pair<Long, Long> getAbortId() {
+ return abortId;
+ }
+
+ public long getTransactionMachineId() {
+ return transactionMachineId;
+ }
+
+ public long getTransactionSequenceNumber() {
+ return transactionSequenceNumber;
+ }
+
+ public long getTransactionClientLocalSequenceNumber() {
+ return transactionClientLocalSequenceNumber;
+ }
+
+ public long getArbitratorLocalSequenceNumber() {
+ return arbitratorLocalSequenceNumber;
+ }
+
+
+ public void setSlot(Slot s) {
+ parentslot = s;
+ }
+
+ public long getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ public void setSequenceNumber(long _sequenceNumber) {
+ sequenceNumber = _sequenceNumber;
+ }
+
+
+ public long getTransactionArbitrator() {
+ return transactionArbitrator;
+ }
+
+ static Entry decode(Slot slot, ByteBuffer bb) {
+ long transactionClientLocalSequenceNumber = bb.getLong();
+ long transactionSequenceNumber = bb.getLong();
+ long sequenceNumber = bb.getLong();
+ long transactionMachineId = bb.getLong();
+ long transactionArbitrator = bb.getLong();
+ long arbitratorLocalSequenceNumber = bb.getLong();
+
+ return new Abort(slot, transactionClientLocalSequenceNumber, transactionSequenceNumber, sequenceNumber, transactionMachineId, transactionArbitrator, arbitratorLocalSequenceNumber);
+ }
+
+ public void encode(ByteBuffer bb) {
+ bb.put(Entry.TypeAbort);
+ bb.putLong(transactionClientLocalSequenceNumber);
+ bb.putLong(transactionSequenceNumber);
+ bb.putLong(sequenceNumber);
+ bb.putLong(transactionMachineId);
+ bb.putLong(transactionArbitrator);
+ bb.putLong(arbitratorLocalSequenceNumber);
+ }
+
+ public int getSize() {
+ return (6 * Long.BYTES) + Byte.BYTES;
+ }
+
+ public byte getType() {
+ return Entry.TypeAbort;
+ }
+
+ public Entry getCopy(Slot s) {
+ return new Abort(s, transactionClientLocalSequenceNumber, transactionSequenceNumber, sequenceNumber, transactionMachineId, transactionArbitrator, arbitratorLocalSequenceNumber);
+ }
+}
\ No newline at end of file
--- /dev/null
+package iotcloud;
+
+import java.util.Set;
+import java.util.HashSet;
+
+import java.util.List;
+import java.util.ArrayList;
+
+
+class ArbitrationRound {
+
+ public static final int MAX_PARTS = 10;
+
+ Set<Abort> abortsBefore = null;
+ List<Entry> parts = null;
+ Commit commit = null;
+ int currentSize = 0;
+ boolean didSendPart = false;
+ boolean didGenerateParts = false;
+
+ public ArbitrationRound(Commit _commit, Set<Abort> _abortsBefore) {
+
+ parts = new ArrayList<Entry>();
+
+ commit = _commit;
+ abortsBefore = _abortsBefore;
+
+
+ if (commit != null) {
+ commit.createCommitParts();
+ currentSize += commit.getNumberOfParts();
+ }
+
+ currentSize += abortsBefore.size();
+ }
+
+ public void generateParts() {
+ if (didGenerateParts) {
+ return;
+ }
+ parts = new ArrayList<Entry>(abortsBefore);
+ if (commit != null) {
+ parts.addAll(commit.getParts().values());
+ }
+ }
+
+
+ public List<Entry> getParts() {
+ return parts;
+ }
+
+ public void removeParts(List<Entry> removeParts) {
+ parts.removeAll(removeParts);
+ didSendPart = true;
+ }
+
+ public boolean isDoneSending() {
+ if ((commit == null) && abortsBefore.isEmpty()) {
+ return true;
+ }
+
+ return parts.isEmpty();
+ }
+
+ public Commit getCommit() {
+ return commit;
+ }
+
+ public void setCommit(Commit _commit) {
+ if (commit != null) {
+ currentSize -= commit.getNumberOfParts();
+ }
+ commit = _commit;
+
+ if (commit != null) {
+ currentSize += commit.getNumberOfParts();
+ }
+ }
+
+ public void addAbort(Abort abort) {
+ abortsBefore.add(abort);
+ currentSize++;
+ }
+
+ public void addAborts(Set<Abort> aborts) {
+ abortsBefore.addAll(aborts);
+ currentSize += aborts.size();
+ }
+
+
+ public Set<Abort> getAborts() {
+ return abortsBefore;
+ }
+
+ public int getAbortsCount() {
+ return abortsBefore.size();
+ }
+
+ public int getCurrentSize() {
+ return currentSize;
+ }
+
+ public boolean isFull() {
+ return currentSize >= MAX_PARTS;
+ }
+
+ public boolean didSendPart() {
+ return didSendPart;
+ }
+}
\ No newline at end of file
--- /dev/null
+package iotcloud;
+
+import java.io.*;
+import java.net.*;
+import java.util.Arrays;
+import javax.crypto.*;
+import javax.crypto.spec.*;
+import java.security.SecureRandom;
+import android.util.*;
+import java.nio.charset.StandardCharsets;
+import org.spongycastle.crypto.generators.PKCS5S2ParametersGenerator;
+import org.spongycastle.crypto.digests.SHA256Digest;
+import org.spongycastle.crypto.params.KeyParameter;
+import org.spongycastle.crypto.PBEParametersGenerator;
+import android.content.*;
+
+/**
+ * This class provides a communication API to the webserver. It also
+ * validates the HMACs on the slots and handles encryption.
+ * @author Brian Demsky <bdemsky@uci.edu>
+ * @version 1.0
+ */
+
+
+class CloudComm {
+ private static final int SALT_SIZE = 8;
+ private static final int TIMEOUT_MILLIS = 2000; // 100
+
+ /** Sets the size for the HMAC. */
+ static final int HMAC_SIZE = 32;
+
+ private String baseurl;
+ private Cipher encryptCipher;
+ private Cipher decryptCipher;
+ private Mac mac;
+ private String password;
+ private SecureRandom random;
+ private byte salt[];
+ private Table table;
+ private int listeningPort = -1;
+ private Thread localServerThread = null;
+ private boolean doEnd = false;
+
+ private TimingSingleton timer = null;
+
+ private Context context;
+
+
+
+
+ /**
+ * Empty Constructor needed for child class.
+ */
+ CloudComm() {
+ timer = TimingSingleton.getInstance();
+ }
+
+ private void deleteFile(Context context) {
+ File fd = context.getFileStreamPath("config.txt");
+ fd.delete();
+ }
+
+
+ private void writeToFile(byte[] data,Context context) {
+ try {
+// OutputStreamWriter outputStreamWriter = new OutputStreamWriter(context.openFileOutput("config.txt", Context.MODE_PRIVATE));
+// outputStreamWriter.write(data);
+// outputStreamWriter.close();
+
+ FileOutputStream outputStreamWriter = context.openFileOutput("config.txt", Context.MODE_PRIVATE);
+ outputStreamWriter.write(data);
+ outputStreamWriter.close();
+
+ }
+ catch (IOException e) {
+ Log.e("Exception", "File write failed: " + e.toString());
+ }
+ }
+
+ private byte[] readFromFile(Context context) throws FileNotFoundException {
+
+ byte[] ret1 = null;
+
+ try {
+ InputStream inputStream = context.openFileInput("config.txt");
+
+ if ( inputStream != null ) {
+
+
+ ret1 = new byte[inputStream.available()];
+ for(int i = 0; i < ret1.length;i++)
+ {
+ ret1[i] = (byte)inputStream.read();
+ }
+
+
+
+
+// InputStreamReader inputStreamReader = new InputStreamReader(inputStream);
+// BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
+// String receiveString = "";
+// StringBuilder stringBuilder = new StringBuilder();
+
+// while ( (receiveString = bufferedReader.readLine()) != null ) {
+// stringBuilder.append(receiveString);
+// }
+
+ inputStream.close();
+// ret = stringBuilder.toString();
+ }
+ }
+ catch (FileNotFoundException e) {
+ Log.e("login activity", "File not found: " + e.toString());
+
+ throw e;
+ } catch (IOException e) {
+ Log.e("login activity", "Can not read file: " + e.toString());
+ }
+
+ return ret1;
+ }
+
+
+
+ /**
+ * Constructor for actual use. Takes in the url and password.
+ */
+ CloudComm(Table _table, String _baseurl, String _password, int _listeningPort, Context _context) {
+ timer = TimingSingleton.getInstance();
+ this.table = _table;
+ this.baseurl = _baseurl;
+ this.password = _password;
+ this.random = new SecureRandom();
+ this.listeningPort = _listeningPort;
+ this.context = _context;
+
+
+ if (this.listeningPort > 0) {
+ localServerThread = new Thread(new Runnable() {
+ public void run() {
+ localServerWorkerFunction();
+ }
+ });
+ localServerThread.start();
+ }
+ }
+
+ /**
+ * Generates Key from password.
+ */
+ private SecretKeySpec initKey() {
+ try {
+
+ Log.e("Ali:::::", "KEY KEY KEY......");
+
+
+
+ boolean doCrypt = false;
+
+ byte[] keySaved = null;
+
+ try {
+// String file = readFromFile(context);
+ byte[] dat = readFromFile(context);//file.getBytes();
+
+ boolean saltMatch = true;
+ for(int i = 0; i < salt.length; i++)
+ {
+
+ Log.e("ALIasdasdaS:", " " + ((int) salt[i] & 255) + " " + ((int) dat[i] & 255));
+
+ if(dat[i] != salt[i])
+ {
+ saltMatch = false;
+// break;
+ }
+ }
+
+ if(saltMatch )
+ {
+ keySaved = new byte[dat.length - salt.length];
+ for(int i = salt.length; i < dat.length;i++)
+ {
+ keySaved[i-salt.length] = dat[i];
+ }
+ }
+ else
+ {
+ doCrypt = true;
+ Log.e("Ali:::::", "Salt No Match......");
+
+ }
+
+
+
+
+
+ }
+ catch (Exception e)
+ {
+ doCrypt = true;
+ }
+
+
+
+ if(doCrypt) {
+ Log.e("Ali:::::", "Doing Crypt......");
+ PKCS5S2ParametersGenerator generator = new PKCS5S2ParametersGenerator(new SHA256Digest());
+ generator.init(PBEParametersGenerator.PKCS5PasswordToUTF8Bytes(password.toCharArray()), salt, 65536);
+ KeyParameter key = (KeyParameter) generator.generateDerivedMacParameters(128);
+
+
+// PBEKeySpec keyspec = new PBEKeySpec(password.toCharArray(),
+// salt,
+// 65536,
+// 128);
+// SecretKey tmpkey = SecretKeyFactory.getInstance("PBKDF2WithHmacSHA1").generateSecret(keyspec);
+// SecretKey tmpkey = SecretKeyFactory.getInstance("PBKDF2WithHmacSHA256").generateSecret(keyspec);
+
+
+// return new SecretKeySpec(tmpkey.getEncoded(), "AES");
+
+
+ byte[] keyDat = key.getKey();
+ byte[] saveDat = new byte[salt.length + keyDat.length];
+
+ for (int i = 0 ; i < salt.length;i++)
+ {
+ saveDat[i] = salt[i];
+ }
+
+ for (int i = 0 ; i < keyDat.length;i++)
+ {
+ saveDat[i + salt.length] = keyDat[i];
+ }
+
+
+ deleteFile(context);
+ writeToFile(saveDat, context);
+
+ return new SecretKeySpec(key.getKey(), "AES");
+ }
+ else{
+
+ Log.e("Ali:::::", "Using Saved......");
+
+ return new SecretKeySpec(keySaved, "AES");
+ }
+
+
+ } catch (Exception e) {
+ StringWriter sw = new StringWriter();
+ PrintWriter pw = new PrintWriter(sw);
+ e.printStackTrace(pw);
+ // stack trace as a string
+
+
+ throw new Error("Failed generating key. " + sw.toString());
+ }
+ }
+
+ /**
+ * Inits all the security stuff
+ */
+ public void initSecurity() throws ServerException {
+ // try to get the salt and if one does not exist set one
+ if (!getSalt()) {
+ //Set the salt
+ setSalt();
+ }
+
+ initCrypt();
+ }
+
+ /**
+ * Inits the HMAC generator.
+ */
+ private void initCrypt() {
+
+ if (password == null) {
+ return;
+ }
+
+ try {
+ SecretKeySpec key = initKey();
+ password = null; // drop password
+ mac = Mac.getInstance("HmacSHA256");
+ mac.init(key);
+ encryptCipher = Cipher.getInstance("AES/ECB/PKCS5Padding");
+ encryptCipher.init(Cipher.ENCRYPT_MODE, key);
+ decryptCipher = Cipher.getInstance("AES/ECB/PKCS5Padding");
+ decryptCipher.init(Cipher.DECRYPT_MODE, key);
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw new Error("Failed To Initialize Ciphers");
+ }
+ }
+
+ /*
+ * Builds the URL for the given request.
+ */
+ private URL buildRequest(boolean isput, long sequencenumber, long maxentries) throws IOException {
+ String reqstring = isput ? "req=putslot" : "req=getslot";
+ String urlstr = baseurl + "?" + reqstring + "&seq=" + sequencenumber;
+ if (maxentries != 0)
+ urlstr += "&max=" + maxentries;
+ return new URL(urlstr);
+ }
+
+ private void setSalt() throws ServerException {
+
+ if (salt != null) {
+ // Salt already sent to server so dont set it again
+ return;
+ }
+
+ try {
+ byte[] saltTmp = new byte[SALT_SIZE];
+ random.nextBytes(saltTmp);
+
+ URL url = new URL(baseurl + "?req=setsalt");
+
+ timer.startTime();
+ URLConnection con = url.openConnection();
+ HttpURLConnection http = (HttpURLConnection) con;
+
+ http.setRequestMethod("POST");
+ http.setFixedLengthStreamingMode(saltTmp.length);
+ http.setDoOutput(true);
+ http.setConnectTimeout(TIMEOUT_MILLIS);
+
+
+ http.connect();
+
+ OutputStream os = http.getOutputStream();
+ os.write(saltTmp);
+ os.flush();
+
+ int responsecode = http.getResponseCode();
+ if (responsecode != HttpURLConnection.HTTP_OK) {
+ // TODO: Remove this print
+ System.out.println(responsecode);
+ throw new Error("Invalid response");
+ }
+
+ timer.endTime();
+
+ salt = saltTmp;
+ } catch (Exception e) {
+ // e.printStackTrace();
+ timer.endTime();
+ throw new ServerException("Failed setting salt", ServerException.TypeConnectTimeout);
+ }
+ }
+
+ private boolean getSalt() throws ServerException {
+ URL url = null;
+ URLConnection con = null;
+ HttpURLConnection http = null;
+
+ try {
+ url = new URL(baseurl + "?req=getsalt");
+ } catch (Exception e) {
+ // e.printStackTrace();
+ throw new Error("getSlot failed");
+ }
+ try {
+
+ timer.startTime();
+ con = url.openConnection();
+ http = (HttpURLConnection) con;
+ http.setRequestMethod("POST");
+ http.setConnectTimeout(TIMEOUT_MILLIS);
+ http.setReadTimeout(TIMEOUT_MILLIS);
+
+
+ http.connect();
+ timer.endTime();
+
+
+ } catch (SocketTimeoutException e) {
+ timer.endTime();
+ throw new ServerException("getSalt failed", ServerException.TypeConnectTimeout);
+ } catch (Exception e) {
+ // e.printStackTrace();
+ throw new Error("getSlot failed " + e.toString());
+ }
+
+ try {
+
+ timer.startTime();
+
+ int responsecode = http.getResponseCode();
+ if (responsecode != HttpURLConnection.HTTP_OK) {
+ // TODO: Remove this print
+ // System.out.println(responsecode);
+ throw new Error("Invalid response");
+ }
+
+// Log.e("Aaaaa", "Code " + responsecode);
+
+
+ InputStream is = http.getInputStream();
+//
+//
+// BufferedReader rd= new BufferedReader(new InputStreamReader(is));
+// int line;
+// StringBuilder sb= new StringBuilder();
+// while ((line = rd.read())!= -1)
+// {
+// sb.append((char)line);
+// Log.e("Aaaaa", "line " + line);
+//
+// }
+//
+//
+// int sdfsdfds = (int)sb.toString().charAt(0);
+// Log.e("Aaaaa", "length " + (int)sb.toString().charAt(0));
+// Log.e("Aaaaa", "Res " + sb.toString().length());
+
+
+// is = new ByteArrayInputStream(sb.toString().getBytes(StandardCharsets.UTF_8));
+
+
+// if (is.available() > 0) {
+// if (sb.toString().length() > 0) {
+ if(true)
+ {
+ try {
+ DataInputStream dis = new DataInputStream(is);
+ int salt_length = dis.readInt();
+ byte[] tmp = new byte[salt_length];
+// byte [] tmp = new byte[8];
+ dis.readFully(tmp);
+ salt = tmp;
+
+ for (int i = 0; i < 8; i++) {
+ Log.e("ALIasdasdaS:", "asd " + ((int) salt[i] & 255));
+ }
+
+
+ timer.endTime();
+
+ return true;
+ }
+ catch (Exception e)
+ {
+ timer.endTime();
+
+ Log.e("Aaaaa", "Salt No Data");
+
+ return false;
+ }
+ }
+ else {
+
+
+ return false;
+ }
+ } catch (SocketTimeoutException e) {
+ timer.endTime();
+
+ throw new ServerException("getSalt failed", ServerException.TypeInputTimeout);
+ } catch (Exception e) {
+
+ throw new Error("getSlot failed + " + e);
+ }
+ }
+
+ /*
+ * API for putting a slot into the queue. Returns null on success.
+ * On failure, the server will send slots with newer sequence
+ * numbers.
+ */
+ public Slot[] putSlot(Slot slot, int max) throws ServerException {
+ URL url = null;
+ URLConnection con = null;
+ HttpURLConnection http = null;
+
+ try {
+ if (salt == null) {
+ if (!getSalt()) {
+ throw new ServerException("putSlot failed", ServerException.TypeSalt);
+ }
+ initCrypt();
+ }
+
+ long sequencenumber = slot.getSequenceNumber();
+ byte[] bytes = slot.encode(mac);
+ bytes = encryptCipher.doFinal(bytes);
+
+
+
+
+ url = buildRequest(true, sequencenumber, max);
+
+ timer.startTime();
+ con = url.openConnection();
+ http = (HttpURLConnection) con;
+
+ http.setRequestMethod("POST");
+ http.setFixedLengthStreamingMode(bytes.length);
+ http.setDoOutput(true);
+ http.setConnectTimeout(TIMEOUT_MILLIS);
+ http.setReadTimeout(TIMEOUT_MILLIS);
+ http.connect();
+
+ OutputStream os = http.getOutputStream();
+ os.write(bytes);
+ os.flush();
+
+ timer.endTime();
+
+
+ // System.out.println("Bytes Sent: " + bytes.length);
+ } catch (ServerException e) {
+ timer.endTime();
+
+ throw e;
+ } catch (SocketTimeoutException e) {
+ timer.endTime();
+
+ throw new ServerException("putSlot failed", ServerException.TypeConnectTimeout);
+ } catch (Exception e) {
+ // e.printStackTrace();
+ throw new Error("putSlot failed");
+ }
+
+
+
+ try {
+ timer.startTime();
+ InputStream is = http.getInputStream();
+ DataInputStream dis = new DataInputStream(is);
+ byte[] resptype = new byte[7];
+ dis.readFully(resptype);
+ timer.endTime();
+
+ if (Arrays.equals(resptype, "getslot".getBytes()))
+ {
+ return processSlots(dis);
+ }
+ else if (Arrays.equals(resptype, "putslot".getBytes()))
+ {
+ return null;
+ }
+ else
+ throw new Error("Bad response to putslot");
+
+ } catch (SocketTimeoutException e) {
+ timer.endTime();
+ throw new ServerException("putSlot failed", ServerException.TypeInputTimeout);
+ } catch (Exception e) {
+ // e.printStackTrace();
+ throw new Error("putSlot failed");
+ }
+ }
+
+ /**
+ * Request the server to send all slots with the given
+ * sequencenumber or newer.
+ */
+ public Slot[] getSlots(long sequencenumber) throws ServerException {
+ URL url = null;
+ URLConnection con = null;
+ HttpURLConnection http = null;
+
+ try {
+ if (salt == null) {
+ if (!getSalt()) {
+ throw new ServerException("getSlots failed", ServerException.TypeSalt);
+ }
+ initCrypt();
+ }
+
+ url = buildRequest(false, sequencenumber, 0);
+ timer.startTime();
+ con = url.openConnection();
+ http = (HttpURLConnection) con;
+ http.setRequestMethod("POST");
+ http.setConnectTimeout(TIMEOUT_MILLIS);
+ http.setReadTimeout(TIMEOUT_MILLIS);
+
+
+
+ http.connect();
+ timer.endTime();
+
+ } catch (SocketTimeoutException e) {
+ timer.endTime();
+
+ throw new ServerException("getSlots failed", ServerException.TypeConnectTimeout);
+ } catch (ServerException e) {
+ timer.endTime();
+
+ throw e;
+ } catch (Exception e) {
+ // e.printStackTrace();
+ throw new Error("getSlots failed " + e.toString());
+ }
+
+ try {
+
+ timer.startTime();
+ InputStream is = http.getInputStream();
+ DataInputStream dis = new DataInputStream(is);
+ byte[] resptype = new byte[7];
+
+ dis.readFully(resptype);
+ timer.endTime();
+
+ if (!Arrays.equals(resptype, "getslot".getBytes()))
+ throw new Error("Bad Response: " + new String(resptype));
+
+ return processSlots(dis);
+ } catch (SocketTimeoutException e) {
+ timer.endTime();
+
+ throw new ServerException("getSlots failed", ServerException.TypeInputTimeout);
+ } catch (Exception e) {
+ // e.printStackTrace();
+ StringWriter sw = new StringWriter();
+ PrintWriter pw = new PrintWriter(sw);
+ e.printStackTrace(pw);
+ throw new Error("getSlots failed " + sw.toString());
+ }
+ }
+
+ /**
+ * Method that actually handles building Slot objects from the
+ * server response. Shared by both putSlot and getSlots.
+ */
+ private Slot[] processSlots(DataInputStream dis) throws Exception {
+ int numberofslots = dis.readInt();
+ int[] sizesofslots = new int[numberofslots];
+
+ Slot[] slots = new Slot[numberofslots];
+ for (int i = 0; i < numberofslots; i++)
+ sizesofslots[i] = dis.readInt();
+
+ for (int i = 0; i < numberofslots; i++) {
+
+ byte[] data = new byte[sizesofslots[i]];
+ dis.readFully(data);
+
+ data = decryptCipher.doFinal(data);
+
+ slots[i] = Slot.decode(table, data, mac);
+
+ Log.e("Ali::::", "Slot Process");
+ }
+ dis.close();
+ return slots;
+ }
+
+ public byte[] sendLocalData(byte[] sendData, String host, int port) {
+
+ if (salt == null) {
+ return null;
+ }
+ try {
+
+ System.out.println("Passing Locally");
+
+ mac.update(sendData);
+ byte[] genmac = mac.doFinal();
+ byte[] totalData = new byte[sendData.length + genmac.length];
+ System.arraycopy(sendData, 0, totalData, 0, sendData.length);
+ System.arraycopy(genmac, 0, totalData, sendData.length, genmac.length);
+
+ // Encrypt the data for sending
+ // byte[] encryptedData = encryptCipher.doFinal(totalData);
+ byte[] encryptedData = encryptCipher.doFinal(totalData);
+
+ // Open a TCP socket connection to a local device
+ Socket socket = new Socket(host, port);
+ socket.setReuseAddress(true);
+ DataOutputStream output = new DataOutputStream(socket.getOutputStream());
+ DataInputStream input = new DataInputStream(socket.getInputStream());
+
+
+ timer.startTime();
+ // Send data to output (length of data, the data)
+ output.writeInt(encryptedData.length);
+ output.write(encryptedData, 0, encryptedData.length);
+ output.flush();
+
+ int lengthOfReturnData = input.readInt();
+ byte[] returnData = new byte[lengthOfReturnData];
+ input.readFully(returnData);
+
+ timer.endTime();
+
+ returnData = decryptCipher.doFinal(returnData);
+
+ // We are done with this socket
+ socket.close();
+
+ mac.update(returnData, 0, returnData.length - HMAC_SIZE);
+ byte[] realmac = mac.doFinal();
+ byte[] recmac = new byte[HMAC_SIZE];
+ System.arraycopy(returnData, returnData.length - realmac.length, recmac, 0, realmac.length);
+
+ if (!Arrays.equals(recmac, realmac))
+ throw new Error("Local Error: Invalid HMAC! Potential Attack!");
+
+ byte[] returnData2 = new byte[lengthOfReturnData - recmac.length];
+ System.arraycopy(returnData, 0, returnData2, 0, returnData2.length);
+
+ return returnData2;
+ } catch (SocketTimeoutException e) {
+
+ } catch (BadPaddingException e) {
+
+ } catch (IllegalBlockSizeException e) {
+
+ } catch (UnknownHostException e) {
+
+ } catch (IOException e) {
+
+ }
+
+ return null;
+ }
+
+ private void localServerWorkerFunction() {
+
+ ServerSocket inputSocket = null;
+
+ try {
+ // Local server socket
+ inputSocket = new ServerSocket(listeningPort);
+ inputSocket.setReuseAddress(true);
+ inputSocket.setSoTimeout(TIMEOUT_MILLIS);
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw new Error("Local server setup failure...");
+ }
+
+ while (!doEnd) {
+
+ try {
+ // Accept incoming socket
+ Socket socket = inputSocket.accept();
+
+ DataInputStream input = new DataInputStream(socket.getInputStream());
+ DataOutputStream output = new DataOutputStream(socket.getOutputStream());
+
+ // Get the encrypted data from the server
+ int dataSize = input.readInt();
+ byte[] readData = new byte[dataSize];
+ input.readFully(readData);
+
+ timer.endTime();
+
+ // Decrypt the data
+ readData = decryptCipher.doFinal(readData);
+
+ mac.update(readData, 0, readData.length - HMAC_SIZE);
+ byte[] genmac = mac.doFinal();
+ byte[] recmac = new byte[HMAC_SIZE];
+ System.arraycopy(readData, readData.length - recmac.length, recmac, 0, recmac.length);
+
+ if (!Arrays.equals(recmac, genmac))
+ throw new Error("Local Error: Invalid HMAC! Potential Attack!");
+
+ byte[] returnData = new byte[readData.length - recmac.length];
+ System.arraycopy(readData, 0, returnData, 0, returnData.length);
+
+ // Process the data
+ // byte[] sendData = table.acceptDataFromLocal(readData);
+ byte[] sendData = table.acceptDataFromLocal(returnData);
+
+ mac.update(sendData);
+ byte[] realmac = mac.doFinal();
+ byte[] totalData = new byte[sendData.length + realmac.length];
+ System.arraycopy(sendData, 0, totalData, 0, sendData.length);
+ System.arraycopy(realmac, 0, totalData, sendData.length, realmac.length);
+
+ // Encrypt the data for sending
+ byte[] encryptedData = encryptCipher.doFinal(totalData);
+
+
+ timer.startTime();
+ // Send data to output (length of data, the data)
+ output.writeInt(encryptedData.length);
+ output.write(encryptedData, 0, encryptedData.length);
+ output.flush();
+
+ // close the socket
+ socket.close();
+ } catch (SocketTimeoutException e) {
+
+ } catch (BadPaddingException e) {
+
+ } catch (IllegalBlockSizeException e) {
+
+ } catch (UnknownHostException e) {
+
+ } catch (IOException e) {
+
+ }
+ }
+
+ if (inputSocket != null) {
+ try {
+ inputSocket.close();
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw new Error("Local server close failure...");
+ }
+ }
+ }
+
+ public void close() {
+ doEnd = true;
+
+ if (localServerThread != null) {
+ try {
+ localServerThread.join();
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw new Error("Local Server thread join issue...");
+ }
+ }
+
+ // System.out.println("Done Closing Cloud Comm");
+ }
+
+ protected void finalize() throws Throwable {
+ try {
+ close(); // close open files
+ } finally {
+ super.finalize();
+ }
+ }
+
+}
--- /dev/null
+package iotcloud;
+
+import java.util.Map;
+import java.util.HashMap;
+import java.util.Set;
+import java.util.HashSet;
+import java.nio.ByteBuffer;
+
+class Commit {
+
+ private Map<Integer, CommitPart> parts = null;
+ private Set<Integer> missingParts = null;
+ private boolean isComplete = false;
+ private boolean hasLastPart = false;
+ private Set<KeyValue> keyValueUpdateSet = null;
+ private boolean isDead = false;
+ private long sequenceNumber = -1;
+ private long machineId = -1;
+ private long transactionSequenceNumber = -1;
+
+ private Set<IoTString> liveKeys = null;
+
+ public Commit() {
+ parts = new HashMap<Integer, CommitPart>();
+ keyValueUpdateSet = new HashSet<KeyValue>();
+
+ liveKeys = new HashSet<IoTString>();
+ }
+
+ public Commit(long _sequenceNumber, long _machineId, long _transactionSequenceNumber) {
+ parts = new HashMap<Integer, CommitPart>();
+ keyValueUpdateSet = new HashSet<KeyValue>();
+
+ liveKeys = new HashSet<IoTString>();
+
+ sequenceNumber = _sequenceNumber;
+ machineId = _machineId;
+ transactionSequenceNumber = _transactionSequenceNumber;
+ isComplete = true;
+ }
+
+
+ public void addPartDecode(CommitPart newPart) {
+
+ if (isDead) {
+ // If dead then just kill this part and move on
+ newPart.setDead();
+ return;
+ }
+
+ CommitPart previoslySeenPart = parts.put(newPart.getPartNumber(), newPart);
+
+ if (previoslySeenPart != null) {
+ // Set dead the old one since the new one is a rescued version of this part
+ previoslySeenPart.setDead();
+ } else if (newPart.isLastPart()) {
+ missingParts = new HashSet<Integer>();
+ hasLastPart = true;
+
+ for (int i = 0; i < newPart.getPartNumber(); i++) {
+ if (parts.get(i) == null) {
+ missingParts.add(i);
+ }
+ }
+ }
+
+ if (!isComplete && hasLastPart) {
+
+ // We have seen this part so remove it from the set of missing parts
+ missingParts.remove(newPart.getPartNumber());
+
+ // Check if all the parts have been seen
+ if (missingParts.size() == 0) {
+
+ // We have all the parts
+ isComplete = true;
+
+ // Decode all the parts and create the key value guard and update sets
+ decodeCommitData();
+
+ // Get the sequence number and arbitrator of this transaction
+ sequenceNumber = parts.get(0).getSequenceNumber();
+ machineId = parts.get(0).getMachineId();
+ transactionSequenceNumber = parts.get(0).getTransactionSequenceNumber();
+ }
+ }
+ }
+
+ public long getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ public long getTransactionSequenceNumber() {
+ return transactionSequenceNumber;
+ }
+
+ public Map<Integer, CommitPart> getParts() {
+ return parts;
+ }
+
+ public void addKV(KeyValue kv) {
+ keyValueUpdateSet.add(kv);
+ liveKeys.add(kv.getKey());
+ }
+
+ public void invalidateKey(IoTString key) {
+ liveKeys.remove(key);
+
+ if (liveKeys.size() == 0) {
+ setDead();
+ }
+ }
+
+ public Set<KeyValue> getKeyValueUpdateSet() {
+ return keyValueUpdateSet;
+ }
+
+ public int getNumberOfParts() {
+ return parts.size();
+ }
+
+ public long getMachineId() {
+ return machineId;
+ }
+
+ public boolean isComplete() {
+ return isComplete;
+ }
+
+ public boolean isLive() {
+ return !isDead;
+ }
+
+ public void setDead() {
+ if (isDead) {
+ // Already dead
+ return;
+ }
+
+ // Set dead
+ isDead = true;
+
+ // Make all the parts of this transaction dead
+ for (Integer partNumber : parts.keySet()) {
+ CommitPart part = parts.get(partNumber);
+ part.setDead();
+ }
+ }
+
+ public CommitPart getPart(int index) {
+ return parts.get(index);
+ }
+
+ public void createCommitParts() {
+
+ parts.clear();
+
+ // Convert to bytes
+ byte[] byteData = convertDataToBytes();
+
+
+ int commitPartCount = 0;
+ int currentPosition = 0;
+ int remaining = byteData.length;
+
+ while (remaining > 0) {
+
+ Boolean isLastPart = false;
+ // determine how much to copy
+ int copySize = CommitPart.MAX_NON_HEADER_SIZE;
+ if (remaining <= CommitPart.MAX_NON_HEADER_SIZE) {
+ copySize = remaining;
+ isLastPart = true; // last bit of data so last part
+ }
+
+ // Copy to a smaller version
+ byte[] partData = new byte[copySize];
+ System.arraycopy(byteData, currentPosition, partData, 0, copySize);
+
+ CommitPart part = new CommitPart(null, machineId, sequenceNumber, transactionSequenceNumber, commitPartCount, partData, isLastPart);
+ parts.put(part.getPartNumber(), part);
+
+ // Update position, count and remaining
+ currentPosition += copySize;
+ commitPartCount++;
+ remaining -= copySize;
+ }
+ }
+
+ private void decodeCommitData() {
+
+ // Calculate the size of the data section
+ int dataSize = 0;
+ for (int i = 0; i < parts.keySet().size(); i++) {
+ CommitPart tp = parts.get(i);
+ dataSize += tp.getDataSize();
+ }
+
+ byte[] combinedData = new byte[dataSize];
+ int currentPosition = 0;
+
+ // Stitch all the data sections together
+ for (int i = 0; i < parts.keySet().size(); i++) {
+ CommitPart tp = parts.get(i);
+ System.arraycopy(tp.getData(), 0, combinedData, currentPosition, tp.getDataSize());
+ currentPosition += tp.getDataSize();
+ }
+
+ // Decoder Object
+ ByteBuffer bbDecode = ByteBuffer.wrap(combinedData);
+
+ // Decode how many key value pairs need to be decoded
+ int numberOfKVUpdates = bbDecode.getInt();
+
+ // Decode all the updates key values
+ for (int i = 0; i < numberOfKVUpdates; i++) {
+ KeyValue kv = (KeyValue)KeyValue.decode(bbDecode);
+ keyValueUpdateSet.add(kv);
+ liveKeys.add(kv.getKey());
+ }
+ }
+
+ private byte[] convertDataToBytes() {
+
+ // Calculate the size of the data
+ int sizeOfData = Integer.BYTES; // Number of Update KV's
+ for (KeyValue kv : keyValueUpdateSet) {
+ sizeOfData += kv.getSize();
+ }
+
+ // Data handlers and storage
+ byte[] dataArray = new byte[sizeOfData];
+ ByteBuffer bbEncode = ByteBuffer.wrap(dataArray);
+
+ // Encode the size of the updates and guard sets
+ bbEncode.putInt(keyValueUpdateSet.size());
+
+ // Encode all the updates
+ for (KeyValue kv : keyValueUpdateSet) {
+ kv.encode(bbEncode);
+ }
+
+ return bbEncode.array();
+ }
+
+ private void setKVsMap(Map<IoTString, KeyValue> newKVs) {
+ keyValueUpdateSet.clear();
+ liveKeys.clear();
+
+ keyValueUpdateSet.addAll(newKVs.values());
+ liveKeys.addAll(newKVs.keySet());
+
+ }
+
+
+ public static Commit merge(Commit newer, Commit older, long newSequenceNumber) {
+
+ if (older == null) {
+ return newer;
+ } else if (newer == null) {
+ return older;
+ }
+
+ Map<IoTString, KeyValue> kvSet = new HashMap<IoTString, KeyValue>();
+ for (KeyValue kv : older.getKeyValueUpdateSet()) {
+ kvSet.put(kv.getKey(), kv);
+ }
+
+ for (KeyValue kv : newer.getKeyValueUpdateSet()) {
+ kvSet.put(kv.getKey(), kv);
+ }
+
+ long transactionSequenceNumber = newer.getTransactionSequenceNumber();
+
+ if (transactionSequenceNumber == -1) {
+ transactionSequenceNumber = older.getTransactionSequenceNumber();
+ }
+
+ Commit newCommit = new Commit(newSequenceNumber, newer.getMachineId(), transactionSequenceNumber);
+
+ newCommit.setKVsMap(kvSet);
+
+ return newCommit;
+ }
+}
\ No newline at end of file
--- /dev/null
+
+
+package iotcloud;
+
+import java.nio.ByteBuffer;
+
+class CommitPart extends Entry{
+
+ // Max size of the part excluding the fixed size header
+ public static final int MAX_NON_HEADER_SIZE = 512;
+
+
+ // Sequence number of the transaction this commit is for, -1 if not a cloud transaction
+ private long machineId = -1; // Machine Id of the device that made the commit
+ private long sequenceNumber = -1; // commit sequence number for this arbitrator
+ private long transactionSequenceNumber = -1;
+ private int partNumber = -1; // Parts position in the
+ private Boolean isLastPart = false;
+ private byte[] data = null;
+
+ private Pair<Long, Integer> partId = null;
+ private Pair<Long, Long> commitId = null;
+
+
+ public CommitPart(Slot s, long _machineId, long _sequenceNumber, long _transactionSequenceNumber, int _partNumber, byte[] _data, Boolean _isLastPart) {
+ super(s);
+ machineId = _machineId;
+ sequenceNumber = _sequenceNumber;
+ transactionSequenceNumber = _transactionSequenceNumber;
+ partNumber = _partNumber;
+ isLastPart = _isLastPart;
+ data = _data;
+
+ partId = new Pair<Long, Integer>(sequenceNumber, partNumber);
+ commitId = new Pair<Long, Long>(machineId, sequenceNumber);
+ }
+
+ public int getSize() {
+ if (data == null) {
+ return (3 * Long.BYTES) + (2 * Integer.BYTES) + (2 * Byte.BYTES);
+ }
+ return (3 * Long.BYTES) + (2 * Integer.BYTES) + (2 * Byte.BYTES) + data.length;
+ }
+
+ public void setSlot(Slot s) {
+ parentslot = s;
+ }
+
+ public int getPartNumber() {
+ return partNumber;
+ }
+
+ public int getDataSize() {
+ return data.length;
+ }
+
+ public byte[] getData() {
+ return data;
+ }
+
+ public Pair<Long, Integer> getPartId() {
+ return partId;
+ }
+
+ public Pair<Long, Long> getCommitId() {
+ return commitId;
+ }
+
+ public Boolean isLastPart() {
+ return isLastPart;
+ }
+
+ public long getMachineId() {
+ return machineId;
+ }
+
+ public long getTransactionSequenceNumber() {
+ return transactionSequenceNumber;
+ }
+
+ public long getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ static Entry decode(Slot s, ByteBuffer bb) {
+ long machineId = bb.getLong();
+ long sequenceNumber = bb.getLong();
+ long transactionSequenceNumber = bb.getLong();
+ int partNumber = bb.getInt();
+ int dataSize = bb.getInt();
+ Boolean isLastPart = bb.get() == 1;
+
+ // Get the data
+ byte[] data = new byte[dataSize];
+ bb.get(data);
+
+ return new CommitPart(s, machineId, sequenceNumber, transactionSequenceNumber, partNumber, data, isLastPart);
+ }
+
+ public void encode(ByteBuffer bb) {
+ bb.put(Entry.TypeCommitPart);
+ bb.putLong(machineId);
+ bb.putLong(sequenceNumber);
+ bb.putLong(transactionSequenceNumber);
+ bb.putInt(partNumber);
+ bb.putInt(data.length);
+
+ if (isLastPart) {
+ bb.put((byte)1);
+ } else {
+ bb.put((byte)0);
+ }
+
+ bb.put(data);
+ }
+
+ public byte getType() {
+ return Entry.TypeCommitPart;
+ }
+
+ public Entry getCopy(Slot s) {
+ return new CommitPart(s, machineId, sequenceNumber, transactionSequenceNumber, partNumber, data, isLastPart);
+ }
+}
\ No newline at end of file
--- /dev/null
+package iotcloud;
+import java.nio.ByteBuffer;
+
+/**
+ * Generic class that wraps all the different types of information
+ * that can be stored in a Slot.
+ * @author Brian Demsky <bdemsky@uci.edu>
+ * @version 1.0
+ */
+
+abstract class Entry implements Liveness {
+
+ static final byte TypeCommitPart = 1;
+ static final byte TypeAbort = 2;
+ static final byte TypeTransactionPart = 3;
+ static final byte TypeNewKey = 4;
+ static final byte TypeLastMessage = 5;
+ static final byte TypeRejectedMessage = 6;
+ static final byte TypeTableStatus = 7;
+
+
+
+ /* Records whether the information is still live or has been
+ superceded by a newer update. */
+
+ private boolean islive = true;
+ protected Slot parentslot;
+
+ public Entry(Slot _parentslot) {
+ parentslot = _parentslot;
+ }
+
+ /**
+ * Static method for decoding byte array into Entry objects. First
+ * byte tells the type of entry.
+ */
+ static Entry decode(Slot slot, ByteBuffer bb) {
+ byte type = bb.get();
+ switch (type) {
+
+ case TypeCommitPart:
+ return CommitPart.decode(slot, bb);
+
+ case TypeAbort:
+ return Abort.decode(slot, bb);
+
+ case TypeTransactionPart:
+ return TransactionPart.decode(slot, bb);
+
+ case TypeNewKey:
+ return NewKey.decode(slot, bb);
+
+ case TypeLastMessage:
+ return LastMessage.decode(slot, bb);
+
+ case TypeRejectedMessage:
+ return RejectedMessage.decode(slot, bb);
+
+ case TypeTableStatus:
+ return TableStatus.decode(slot, bb);
+
+ default:
+ throw new Error("Unrecognized Entry Type: " + type);
+ }
+ }
+
+ /**
+ * Returns true if the Entry object is still live.
+ */
+ public boolean isLive() {
+ return islive;
+ }
+
+
+ /**
+ * Flags the entry object as dead. Also decrements the live count
+ * of the parent slot.
+ */
+ public void setDead() {
+
+ if (!islive ) {
+ return; // already dead
+ }
+
+ islive = false;
+
+ if (parentslot != null) {
+ parentslot.decrementLiveCount();
+ }
+ }
+
+
+ /**
+ * Serializes the Entry object into the byte buffer.
+ */
+ abstract void encode(ByteBuffer bb);
+
+
+ /**
+ * Returns the size in bytes the entry object will take in the byte
+ * array.
+ */
+ abstract int getSize();
+
+
+ /**
+ * Returns a byte encoding the type of the entry object.
+ */
+ abstract byte getType();
+
+
+ /**
+ * Returns a copy of the Entry that can be added to a different slot.
+ */
+ abstract Entry getCopy(Slot s);
+}
--- /dev/null
+package iotcloud;
+
+import java.util.Arrays;
+
+/**
+ * IoTString is wraps the underlying byte string. We don't use the
+ * standard String class as we have bytes and not chars.
+ * @author Brian Demsky <bdemsky@uci.edu>
+ * @version 1.0
+ */
+
+
+final public class IoTString {
+ byte[] array;
+ int hashcode;
+
+ private IoTString() {
+ }
+
+ /**
+ * Builds an IoTString object around the byte array. This
+ * constructor makes a copy, so the caller is free to modify the byte array.
+ */
+
+ public IoTString(byte[] _array) {
+ array=(byte[]) _array.clone();
+ hashcode=Arrays.hashCode(array);
+ }
+
+ /**
+ * Converts the String object to a byte representation and stores it
+ * into the IoTString object.
+ */
+
+ public IoTString(String str) {
+ array=str.getBytes();
+ hashcode=Arrays.hashCode(array);
+ }
+
+ /**
+ * Internal methods to build an IoTString using the byte[] passed
+ * in. Caller is responsible for ensuring the byte[] is never
+ * modified.
+ */
+
+ static IoTString shallow(byte[] _array) {
+ IoTString i=new IoTString();
+ i.array = _array;
+ i.hashcode = Arrays.hashCode(_array);
+ return i;
+ }
+
+ /**
+ * Internal method to grab a reference to our byte array. Caller
+ * must not modify it.
+ */
+
+ byte[] internalBytes() {
+ return array;
+ }
+
+ /**
+ * Returns the hashCode as computed by Arrays.hashcode(byte[]).
+ */
+
+ public int hashCode() {
+ return hashcode;
+ }
+
+ /**
+ * Returns a String representation of the IoTString.
+ */
+
+ public String toString() {
+ return new String(array);
+ }
+
+ /**
+ * Returns a copy of the underlying byte string.
+ */
+
+ public byte[] getBytes() {
+ return (byte[]) array.clone();
+ }
+
+ /**
+ * Returns true if two byte strings have the same content.
+ */
+
+ public boolean equals(Object o) {
+ if (o instanceof IoTString) {
+ IoTString i=(IoTString)o;
+ return Arrays.equals(array, i.array);
+ }
+ return false;
+ }
+
+ /**
+ * Returns the length in bytes of the IoTString.
+ */
+
+ public int length() {
+ return array.length;
+ }
+}
--- /dev/null
+package iotcloud;
+import java.nio.ByteBuffer;
+
+/**
+ * KeyValue entry for Slot.
+ * @author Brian Demsky <bdemsky@uci.edu>
+ * @version 1.0
+ */
+
+class KeyValue { /*extends Entry */
+ private IoTString key;
+ private IoTString value;
+
+ public KeyValue(IoTString _key, IoTString _value) {
+ key = _key;
+ value = _value;
+ }
+
+ public IoTString getKey() {
+ return key;
+ }
+
+ public IoTString getValue() {
+ return value;
+ }
+
+ static KeyValue decode(ByteBuffer bb) {
+ int keylength = bb.getInt();
+ int valuelength = bb.getInt();
+ byte[] key = new byte[keylength];
+ bb.get(key);
+
+ if (valuelength != 0) {
+ byte[] value = new byte[valuelength];
+ bb.get(value);
+ return new KeyValue(IoTString.shallow(key), IoTString.shallow(value));
+ }
+
+ return new KeyValue(IoTString.shallow(key), null);
+ }
+
+ public void encode(ByteBuffer bb) {
+ bb.putInt(key.length());
+
+ if (value != null) {
+ bb.putInt(value.length());
+ } else {
+ bb.putInt(0);
+ }
+
+ bb.put(key.internalBytes());
+
+ if (value != null) {
+ bb.put(value.internalBytes());
+ }
+ }
+
+ public int getSize() {
+ if (value != null) {
+ return 2 * Integer.BYTES + key.length() + value.length();
+ }
+
+ return 2 * Integer.BYTES + key.length();
+ }
+
+ public String toString() {
+ if (value == null) {
+ return "null";
+ }
+ return value.toString();
+ }
+
+ public KeyValue getCopy() {
+ return new KeyValue(key, value);
+ }
+}
--- /dev/null
+package iotcloud;
+
+import java.nio.ByteBuffer;
+
+/**
+ * This Entry records the last message sent by a given machine.
+ * @author Brian Demsky <bdemsky@uci.edu>
+ * @version 1.0
+ */
+
+
+class LastMessage extends Entry {
+ private long machineid;
+ private long seqnum;
+
+ public LastMessage(Slot slot, long _machineid, long _seqnum) {
+ super(slot);
+ machineid=_machineid;
+ seqnum=_seqnum;
+ }
+
+ public long getMachineID() {
+ return machineid;
+ }
+
+ public long getSequenceNumber() {
+ return seqnum;
+ }
+
+ static Entry decode(Slot slot, ByteBuffer bb) {
+ long machineid=bb.getLong();
+ long seqnum=bb.getLong();
+ return new LastMessage(slot, machineid, seqnum);
+ }
+
+ public void encode(ByteBuffer bb) {
+ bb.put(Entry.TypeLastMessage);
+ bb.putLong(machineid);
+ bb.putLong(seqnum);
+ }
+
+ public int getSize() {
+ return 2*Long.BYTES+Byte.BYTES;
+ }
+
+ public byte getType() {
+ return Entry.TypeLastMessage;
+ }
+
+ public Entry getCopy(Slot s) {
+ return new LastMessage(s, machineid, seqnum);
+ }
+}
+
+
--- /dev/null
+package iotcloud;
+
+/**
+ * Interface common to both classes that record information about the
+ * last message sent by a machine. (Either a Slot or a LastMessage.
+ * @author Brian Demsky <bdemsky@uci.edu>
+ * @version 1.0
+ */
+
+interface Liveness {
+}
--- /dev/null
+package iotcloud;
+
+class LocalComm {
+ private Table t1;
+ private Table t2;
+
+ public LocalComm(Table _t1, Table _t2) {
+ t1 = _t1;
+ t2 = _t2;
+ }
+
+ public byte[] sendDataToLocalDevice(Long deviceId, byte[] data) throws InterruptedException{
+ System.out.println("Passing Locally");
+
+ if (deviceId == t1.getMachineId()) {
+ // return t1.localCommInput(data);
+ } else if (deviceId == t2.getMachineId()) {
+ // return t2.localCommInput(data);
+ } else {
+ throw new Error("Cannot send to " + deviceId + " using this local comm");
+ }
+
+ return new byte[0];
+ }
+}
\ No newline at end of file
--- /dev/null
+package iotcloud;
+
+import java.nio.ByteBuffer;
+
+/**
+ * This Entry records the abort sent by a given machine.
+ * @author Ali Younis <ayounis@uci.edu>
+ * @version 1.0
+ */
+
+
+class NewKey extends Entry {
+ private IoTString key;
+ private long machineid;
+
+ public NewKey(Slot slot, IoTString _key, long _machineid) {
+ super(slot);
+ key = _key;
+ machineid = _machineid;
+ }
+
+ public long getMachineID() {
+ return machineid;
+ }
+
+ public IoTString getKey() {
+ return key;
+ }
+
+ public void setSlot(Slot s) {
+ parentslot = s;
+ }
+
+ static Entry decode(Slot slot, ByteBuffer bb) {
+ int keylength = bb.getInt();
+ byte[] key = new byte[keylength];
+ bb.get(key);
+ long machineid = bb.getLong();
+
+ return new NewKey(slot, IoTString.shallow(key), machineid);
+ }
+
+ public void encode(ByteBuffer bb) {
+ bb.put(Entry.TypeNewKey);
+ bb.putInt(key.length());
+ bb.put(key.internalBytes());
+ bb.putLong(machineid);
+ }
+
+ public int getSize() {
+ return Long.BYTES + Byte.BYTES + Integer.BYTES + key.length();
+ }
+
+ public byte getType() {
+ return Entry.TypeNewKey;
+ }
+
+ public Entry getCopy(Slot s) {
+ return new NewKey(s, key, machineid);
+ }
+}
\ No newline at end of file
--- /dev/null
+package iotcloud;
+
+class Pair<A, B> {
+ private A a;
+ private B b;
+ int hashCode = -1;
+
+ Pair(A a, B b) {
+ this.a = a;
+ this.b = b;
+
+ hashCode = 23;
+ hashCode = hashCode * 31 + a.hashCode();
+ hashCode = hashCode * 31 + b.hashCode();
+ }
+
+ A getFirst() {
+ return a;
+ }
+
+ B getSecond() {
+ return b;
+ }
+
+
+ public int hashCode() {
+ return hashCode;
+ }
+
+ public boolean equals(Object o) {
+ if (o instanceof Pair) {
+ Pair i = (Pair)o;
+ if (a.equals(i.getFirst()) && b.equals(i.getSecond())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ public String toString() {
+ return "<" + a + "," + b + ">";
+ }
+}
--- /dev/null
+package iotcloud;
+
+import java.util.Set;
+import java.util.Map;
+import java.util.HashSet;
+
+import javax.script.ScriptException;
+import java.lang.NullPointerException;
+import java.nio.ByteBuffer;
+
+
+class PendingTransaction {
+
+ private Set<KeyValue> keyValueUpdateSet = null;
+ private Set<KeyValue> keyValueGuardSet = null;
+ private long arbitrator = -1;
+ private long clientLocalSequenceNumber = -1;
+ private long machineId = -1;
+
+ private int currentDataSize = 0;
+
+ public PendingTransaction(long _machineId) {
+ machineId = _machineId;
+ keyValueUpdateSet = new HashSet<KeyValue>();
+ keyValueGuardSet = new HashSet<KeyValue>();
+ }
+
+ /**
+ * Add a new key value to the updates
+ *
+ */
+ public void addKV(KeyValue newKV) {
+
+ KeyValue rmKV = null;
+
+ // Make sure there are no duplicates
+ for (KeyValue kv : keyValueUpdateSet) {
+ if (kv.getKey().equals(newKV.getKey())) {
+
+ // Remove key if we are adding a newer version of the same key
+ rmKV = kv;
+ break;
+ }
+ }
+
+ // Remove key if we are adding a newer version of the same key
+ if (rmKV != null) {
+ keyValueUpdateSet.remove(rmKV);
+ currentDataSize -= rmKV.getSize();
+ }
+
+ // Add the key to the hash set
+ keyValueUpdateSet.add(newKV);
+ currentDataSize += newKV.getSize();
+ }
+
+ /**
+ * Add a new key value to the guard set
+ *
+ */
+ public void addKVGuard(KeyValue newKV) {
+ // Add the key to the hash set
+ keyValueGuardSet.add(newKV);
+ currentDataSize += newKV.getSize();
+ }
+
+ /**
+ * Checks if the arbitrator is the same
+ */
+ public boolean checkArbitrator(long arb) {
+ if (arbitrator == -1) {
+ arbitrator = arb;
+ return true;
+ }
+
+ return arb == arbitrator;
+ }
+
+ /**
+ * Get the transaction arbitrator
+ */
+ public long getArbitrator() {
+ return arbitrator;
+ }
+
+ /**
+ * Get the key value update set
+ */
+ public Set<KeyValue> getKVUpdates() {
+ return keyValueUpdateSet;
+ }
+
+ /**
+ * Get the key value update set
+ */
+ public Set<KeyValue> getKVGuard() {
+ return keyValueGuardSet;
+ }
+
+ public void setClientLocalSequenceNumber(long _clientLocalSequenceNumber) {
+ clientLocalSequenceNumber = _clientLocalSequenceNumber;
+ }
+
+ public long getClientLocalSequenceNumber() {
+ return clientLocalSequenceNumber;
+ }
+
+ public long getMachineId() {
+ return machineId;
+ }
+
+ public boolean evaluateGuard(Map<IoTString, KeyValue> keyValTableCommitted, Map<IoTString, KeyValue> keyValTableSpeculative, Map<IoTString, KeyValue> keyValTablePendingTransSpeculative) {
+ for (KeyValue kvGuard : keyValueGuardSet) {
+
+ // First check if the key is in the speculative table, this is the value of the latest assumption
+ KeyValue kv = keyValTablePendingTransSpeculative.get(kvGuard.getKey());
+
+
+ if (kv == null) {
+ // if it is not in the pending trans table then check the speculative table and use that
+ // value as our latest assumption
+ kv = keyValTableSpeculative.get(kvGuard.getKey());
+ }
+
+
+ if (kv == null) {
+ // if it is not in the speculative table then check the committed table and use that
+ // value as our latest assumption
+ kv = keyValTableCommitted.get(kvGuard.getKey());
+ }
+
+ if (kvGuard.getValue() != null) {
+ if ((kv == null) || (!kvGuard.getValue().equals(kv.getValue()))) {
+ return false;
+ }
+ } else {
+ if (kv != null) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ public Transaction createTransaction() {
+
+ Transaction newTransaction = new Transaction();
+ int transactionPartCount = 0;
+
+ // Convert all the data into a byte array so we can start partitioning
+ byte[] byteData = convertDataToBytes();
+
+ int currentPosition = 0;
+ int remaining = byteData.length;
+
+ while (remaining > 0) {
+
+ Boolean isLastPart = false;
+ // determine how much to copy
+ int copySize = TransactionPart.MAX_NON_HEADER_SIZE;
+ if (remaining <= TransactionPart.MAX_NON_HEADER_SIZE) {
+ copySize = remaining;
+ isLastPart = true; // last bit of data so last part
+ }
+
+ // Copy to a smaller version
+ byte[] partData = new byte[copySize];
+ System.arraycopy(byteData, currentPosition, partData, 0, copySize);
+
+ TransactionPart part = new TransactionPart(null, machineId, arbitrator, clientLocalSequenceNumber, transactionPartCount, partData, isLastPart);
+ newTransaction.addPartEncode(part);
+
+ // Update position, count and remaining
+ currentPosition += copySize;
+ transactionPartCount++;
+ remaining -= copySize;
+ }
+
+ // Add the Guard Conditions
+ for (KeyValue kv : keyValueGuardSet) {
+ newTransaction.addGuardKV(kv);
+ }
+
+ // Add the updates
+ for (KeyValue kv : keyValueUpdateSet) {
+ newTransaction.addUpdateKV(kv);
+ }
+
+ return newTransaction;
+ }
+
+ private byte[] convertDataToBytes() {
+
+ // Calculate the size of the data
+ int sizeOfData = 2 * Integer.BYTES; // Number of Update KV's and Guard KV's
+ sizeOfData += currentDataSize;
+
+ // Data handlers and storage
+ byte[] dataArray = new byte[sizeOfData];
+ ByteBuffer bbEncode = ByteBuffer.wrap(dataArray);
+
+ // Encode the size of the updates and guard sets
+ bbEncode.putInt(keyValueGuardSet.size());
+ bbEncode.putInt(keyValueUpdateSet.size());
+
+ // Encode all the guard conditions
+ for (KeyValue kv : keyValueGuardSet) {
+ kv.encode(bbEncode);
+ }
+
+ // Encode all the updates
+ for (KeyValue kv : keyValueUpdateSet) {
+ kv.encode(bbEncode);
+ }
+
+ return bbEncode.array();
+ }
+}
\ No newline at end of file
--- /dev/null
+package iotcloud;
+import java.nio.ByteBuffer;
+import java.util.HashSet;
+
+/**
+ * Entry for tracking messages that the server rejected. We have to
+ * make sure that all clients know that this message was rejected to
+ * prevent the server from reusing these messages in an attack.
+ * @author Brian Demsky
+ * @version 1.0
+ */
+
+
+class RejectedMessage extends Entry {
+ /* Sequence number */
+ private long sequencenum;
+
+
+ /* Machine identifier */
+ private long machineid;
+ /* Oldest sequence number in range */
+ private long oldseqnum;
+ /* Newest sequence number in range */
+ private long newseqnum;
+ /* Is the machine identifier of the relevant slots equal to (or not
+ * equal to) the specified machine identifier. */
+ private boolean equalto;
+ /* Set of machines that have not received notification. */
+ private HashSet<Long> watchset;
+
+ RejectedMessage(Slot slot, long _sequencenum, long _machineid, long _oldseqnum, long _newseqnum, boolean _equalto) {
+ super(slot);
+ sequencenum = _sequencenum;
+ machineid=_machineid;
+ oldseqnum=_oldseqnum;
+ newseqnum=_newseqnum;
+ equalto=_equalto;
+ }
+
+ long getOldSeqNum() {
+ return oldseqnum;
+ }
+
+ long getNewSeqNum() {
+ return newseqnum;
+ }
+
+ boolean getEqual() {
+ return equalto;
+ }
+
+ long getMachineID() {
+ return machineid;
+ }
+
+
+ long getSequenceNumber() {
+ return sequencenum;
+ }
+
+ static Entry decode(Slot slot, ByteBuffer bb) {
+ long sequencenum=bb.getLong();
+ long machineid=bb.getLong();
+ long oldseqnum=bb.getLong();
+ long newseqnum=bb.getLong();
+ byte equalto=bb.get();
+ return new RejectedMessage(slot,sequencenum, machineid, oldseqnum, newseqnum, equalto==1);
+ }
+
+ void setWatchSet(HashSet<Long> _watchset) {
+ watchset=_watchset;
+ }
+
+ void removeWatcher(long machineid) {
+ if (watchset.remove(machineid))
+ if (watchset.isEmpty())
+ setDead();
+ }
+
+ void encode(ByteBuffer bb) {
+ bb.put(Entry.TypeRejectedMessage);
+ bb.putLong(sequencenum);
+ bb.putLong(machineid);
+ bb.putLong(oldseqnum);
+ bb.putLong(newseqnum);
+ bb.put(equalto?(byte)1:(byte)0);
+ }
+
+ int getSize() {
+ return 4*Long.BYTES + 2*Byte.BYTES;
+ }
+
+ byte getType() {
+ return Entry.TypeRejectedMessage;
+ }
+
+ Entry getCopy(Slot s) {
+ return new RejectedMessage(s,sequencenum, machineid, oldseqnum, newseqnum, equalto);
+ }
+}
--- /dev/null
+package iotcloud;
+
+public class ServerException extends Exception {
+
+ public static final byte TypeConnectTimeout = 1;
+ public static final byte TypeInputTimeout = 2;
+ public static final byte TypeIncorrectResponseCode = 3;
+ public static final byte TypeSalt = 4;
+ private byte type = -1;
+
+ public ServerException(String message, byte _type) {
+ super(message);
+ type = _type;
+ }
+
+ public byte getType() {
+ return type;
+ }
+}
--- /dev/null
+package iotcloud;
+import java.util.Vector;
+import java.nio.ByteBuffer;
+import javax.crypto.Mac;
+import java.util.Arrays;
+
+/**
+ * Data structuring for holding Slot information.
+ * @author Brian Demsky
+ * @version 1.0
+ */
+
+class Slot implements Liveness {
+ /** Sets the slot size. */
+ static final int SLOT_SIZE = 2048;
+ /** Sets the size for the HMAC. */
+ static final int HMAC_SIZE = 32;
+
+ /** Sequence number of the slot. */
+ private long seqnum;
+ /** HMAC of previous slot. */
+ private byte[] prevhmac;
+ /** HMAC of this slot. */
+ private byte[] hmac;
+ /** Machine that sent this slot. */
+ private long machineid;
+ /** Vector of entries in this slot. */
+ private Vector<Entry> entries;
+ /** Pieces of information that are live. */
+ private int livecount;
+ /** Flag that indicates whether this slot is still live for
+ * recording the machine that sent it. */
+ private boolean seqnumlive;
+ /** Number of bytes of free space. */
+ private int freespace;
+ /** Reference to Table */
+ private Table table;
+
+ Slot(Table _table, long _seqnum, long _machineid, byte[] _prevhmac, byte[] _hmac) {
+ seqnum = _seqnum;
+ machineid = _machineid;
+ prevhmac = _prevhmac;
+ hmac = _hmac;
+ entries = new Vector<Entry>();
+ livecount = 1;
+ seqnumlive = true;
+ freespace = SLOT_SIZE - getBaseSize();
+ table = _table;
+ }
+
+ Slot(Table _table, long _seqnum, long _machineid, byte[] _prevhmac) {
+ this(_table, _seqnum, _machineid, _prevhmac, null);
+ }
+
+ Slot(Table _table, long _seqnum, long _machineid) {
+ this(_table, _seqnum, _machineid, new byte[HMAC_SIZE], null);
+ }
+
+ byte[] getHMAC() {
+ return hmac;
+ }
+
+ byte[] getPrevHMAC() {
+ return prevhmac;
+ }
+
+ Entry addEntry(Entry e) {
+ e = e.getCopy(this);
+ entries.add(e);
+ livecount++;
+ freespace -= e.getSize();
+ return e;
+ }
+
+ void removeEntry(Entry e) {
+ entries.remove(e);
+ livecount--;
+ freespace += e.getSize();
+ }
+
+ private void addShallowEntry(Entry e) {
+ entries.add(e);
+ livecount++;
+ freespace -= e.getSize();
+ }
+
+ /**
+ * Returns true if the slot has free space to hold the entry without
+ * using its reserved space. */
+
+ boolean hasSpace(Entry e) {
+ int newfreespace = freespace - e.getSize();
+ return newfreespace >= 0;
+ }
+
+ Vector<Entry> getEntries() {
+ return entries;
+ }
+
+ static Slot decode(Table table, byte[] array, Mac mac) {
+ mac.update(array, HMAC_SIZE, array.length - HMAC_SIZE);
+ byte[] realmac = mac.doFinal();
+
+ ByteBuffer bb = ByteBuffer.wrap(array);
+ byte[] hmac = new byte[HMAC_SIZE];
+ byte[] prevhmac = new byte[HMAC_SIZE];
+ bb.get(hmac);
+ bb.get(prevhmac);
+ if (!Arrays.equals(realmac, hmac))
+ throw new Error("Server Error: Invalid HMAC! Potential Attack!");
+
+ long seqnum = bb.getLong();
+ long machineid = bb.getLong();
+ int numentries = bb.getInt();
+ Slot slot = new Slot(table, seqnum, machineid, prevhmac, hmac);
+
+ for (int i = 0; i < numentries; i++) {
+ slot.addShallowEntry(Entry.decode(slot, bb));
+ }
+
+ return slot;
+ }
+
+ byte[] encode(Mac mac) {
+ byte[] array = new byte[SLOT_SIZE];
+ ByteBuffer bb = ByteBuffer.wrap(array);
+ /* Leave space for the slot HMAC. */
+ bb.position(HMAC_SIZE);
+ bb.put(prevhmac);
+ bb.putLong(seqnum);
+ bb.putLong(machineid);
+ bb.putInt(entries.size());
+ for (Entry entry : entries) {
+ entry.encode(bb);
+ }
+ /* Compute our HMAC */
+ mac.update(array, HMAC_SIZE, array.length - HMAC_SIZE);
+ byte[] realmac = mac.doFinal();
+ hmac = realmac;
+ bb.position(0);
+ bb.put(realmac);
+ return array;
+ }
+
+ /**
+ * Returns the empty size of a Slot. Includes 2 HMACs, the machine
+ * identifier, the sequence number, and the number of entries.
+ */
+ int getBaseSize() {
+ return 2 * HMAC_SIZE + 2 * Long.BYTES + Integer.BYTES;
+ }
+
+ /**
+ * Returns the live set of entries for this Slot. Generates a fake
+ * LastMessage entry to represent the information stored by the slot
+ * itself.
+ */
+
+ Vector<Entry> getLiveEntries(boolean resize) {
+ Vector<Entry> liveEntries = new Vector<Entry>();
+ for (Entry entry : entries) {
+ if (entry.isLive()) {
+ if (!resize || entry.getType() != Entry.TypeTableStatus)
+ liveEntries.add(entry);
+ }
+ }
+
+ if (seqnumlive && !resize)
+ liveEntries.add(new LastMessage(this, machineid, seqnum));
+
+ return liveEntries;
+ }
+
+ /**
+ * Returns the sequence number of the slot.
+ */
+
+ long getSequenceNumber() {
+ return seqnum;
+ }
+
+ /**
+ * Returns the machine that sent this slot.
+ */
+
+ long getMachineID() {
+ return machineid;
+ }
+
+ /**
+ * Records that a newer slot records the fact that this slot was
+ * sent by the relevant machine.
+ */
+
+ void setDead() {
+ seqnumlive = false;
+ decrementLiveCount();
+ }
+
+ /**
+ * Update the count of live entries.
+ */
+
+ void decrementLiveCount() {
+ livecount--;
+ if (livecount == 0) {
+ table.decrementLiveCount();
+ }
+ }
+
+ /**
+ * Returns whether the slot stores any live information.
+ */
+
+ boolean isLive() {
+ return livecount > 0;
+ }
+
+ public String toString() {
+ return "<" + getSequenceNumber() + ">";
+ }
+}
--- /dev/null
+package iotcloud;
+
+/**
+ * Circular buffer that holds the live set of slots.
+ * @author Brian Demsky
+ * @version 1.0
+ */
+
+class SlotBuffer {
+ static final int DEFAULT_SIZE = 16;
+
+ private Slot[] array;
+ private int head;
+ private int tail;
+ public long oldestseqn;
+
+ SlotBuffer() {
+ array = new Slot[DEFAULT_SIZE + 1];
+ head = tail = 0;
+ oldestseqn = 0;
+ }
+
+ int size() {
+ if (head >= tail)
+ return head - tail;
+ return (array.length + head) - tail;
+ }
+
+ int capacity() {
+ return array.length - 1;
+ }
+
+ void resize(int newsize) {
+ if (newsize == (array.length - 1))
+ return;
+
+ Slot[] newarray = new Slot[newsize + 1];
+ int currsize = size();
+ int index = tail;
+ for (int i = 0; i < currsize; i++) {
+ newarray[i] = array[index];
+ if ((++index) == array.length)
+ index = 0;
+ }
+ array = newarray;
+ tail = 0;
+ head = currsize;
+ }
+
+ private void incrementHead() {
+ head++;
+ if (head >= array.length)
+ head = 0;
+ }
+
+ private void incrementTail() {
+ tail++;
+ if (tail >= array.length)
+ tail = 0;
+ }
+
+ void putSlot(Slot s) {
+
+ long checkNum = (getNewestSeqNum() + 1);
+
+ if (checkNum != s.getSequenceNumber()) {
+ // We have a gap so expunge all our slots
+ oldestseqn = s.getSequenceNumber();
+ tail = 0;
+ head = 1;
+ array[0] = s;
+ return;
+ }
+
+ array[head] = s;
+ incrementHead();
+
+ if (oldestseqn == 0) {
+ oldestseqn = s.getSequenceNumber();
+ }
+
+ if (head == tail) {
+ incrementTail();
+ oldestseqn++;
+ }
+ }
+
+ Slot getSlot(long seqnum) {
+ int diff = (int) (seqnum - oldestseqn);
+ int index = diff + tail;
+
+ if (index < 0) {
+ // Really old message so we dont have it anymore
+ return null;
+ }
+
+ if (index >= array.length) {
+ if (head >= tail) {
+ return null;
+ }
+ index -= array.length;
+ }
+
+ if (index >= array.length) {
+
+ return null;
+ }
+ if (head >= tail && index >= head) {
+ return null;
+ }
+
+ return array[index];
+ }
+
+ long getOldestSeqNum() {
+ return oldestseqn;
+ }
+
+ long getNewestSeqNum() {
+ return oldestseqn + size() - 1;
+ }
+}
--- /dev/null
+package iotcloud;
+
+/**
+ * Slot indexer allows slots in both the slot buffer and the new
+ * server response to looked up in a consistent fashion.
+ * @author Brian Demsky
+ * @version 1.0
+ */
+
+class SlotIndexer {
+ private Slot[] updates;
+ private SlotBuffer buffer;
+ private long firstslotseqnum;
+
+ SlotIndexer(Slot[] _updates, SlotBuffer _buffer) {
+ buffer = _buffer;
+ updates = _updates;
+ firstslotseqnum = updates[0].getSequenceNumber();
+ }
+
+ Slot getSlot(long seqnum) {
+ if (seqnum >= firstslotseqnum) {
+ int offset = (int) (seqnum - firstslotseqnum);
+ if (offset >= updates.length)
+ throw new Error("Invalid Slot Sequence Number Reference");
+ else
+ return updates[offset];
+ } else
+ return buffer.getSlot(seqnum);
+ }
+}
--- /dev/null
+package iotcloud;
+
+import java.util.Iterator;
+import java.util.Random;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.List;
+import java.util.Vector;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.nio.ByteBuffer;
+import android.content.*;
+
+/**
+ * IoTTable data structure. Provides client interface.
+ * @author Brian Demsky
+ * @version 1.0
+ */
+
+final public class Table {
+
+ /* Constants */
+ static final int FREE_SLOTS = 10; // Number of slots that should be kept free
+ static final int SKIP_THRESHOLD = 10;
+ static final double RESIZE_MULTIPLE = 1.2;
+ static final double RESIZE_THRESHOLD = 0.75;
+ static final int REJECTED_THRESHOLD = 5;
+
+ /* Helper Objects */
+ private SlotBuffer buffer = null;
+ private CloudComm cloud = null;
+ private Random random = null;
+ private TableStatus liveTableStatus = null;
+ private PendingTransaction pendingTransactionBuilder = null; // Pending Transaction used in building a Pending Transaction
+ private Transaction lastPendingTransactionSpeculatedOn = null; // Last transaction that was speculated on from the pending transaction
+ private Transaction firstPendingTransaction = null; // first transaction in the pending transaction list
+
+ /* Variables */
+ private int numberOfSlots = 0; // Number of slots stored in buffer
+ private int bufferResizeThreshold = 0; // Threshold on the number of live slots before a resize is needed
+ private long liveSlotCount = 0; // Number of currently live slots
+ 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 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
+ private long lastTransactionSequenceNumberSpeculatedOn = -1; // the last transaction that was speculated on
+ private long oldestTransactionSequenceNumberSpeculatedOn = -1; // the oldest transaction that was speculated on
+ private long localArbitrationSequenceNumber = 0;
+ private boolean hadPartialSendToServer = false;
+ private boolean attemptedToSendToServer = false;
+ private long expectedsize;
+ private boolean didFindTableStatus = false;
+ private long currMaxSize = 0;
+
+ private Slot lastSlotAttemptedToSend = null;
+ private boolean lastIsNewKey = false;
+ private int lastNewSize = 0;
+ private Map<Transaction, List<Integer>> lastTransactionPartsSent = null;
+ private List<Entry> lastPendingSendArbitrationEntriesToDelete = null;
+ private NewKey lastNewKey = null;
+
+
+ /* Data Structures */
+ private Map<IoTString, KeyValue> committedKeyValueTable = null; // Table of committed key value pairs
+ private Map<IoTString, KeyValue> speculatedKeyValueTable = null; // Table of speculated key value pairs, if there is a speculative value
+ private Map<IoTString, KeyValue> pendingTransactionSpeculatedKeyValueTable = null; // Table of speculated key value pairs, if there is a speculative value from the pending transactions
+ private Map<IoTString, NewKey> liveNewKeyTable = null; // Table of live new keys
+ private HashMap<Long, Pair<Long, Liveness>> lastMessageTable = null; // Last message sent by a client machine id -> (Seq Num, Slot or LastMessage);
+ private HashMap<Long, HashSet<RejectedMessage>> rejectedMessageWatchListTable = null; // Table of machine Ids and the set of rejected messages they have not seen yet
+ private Map<IoTString, Long> arbitratorTable = null; // Table of keys and their arbitrators
+ private Map<Pair<Long, Long>, Abort> liveAbortTable = null; // Table live abort messages
+ private Map<Long, Map<Pair<Long, Integer>, TransactionPart>> newTransactionParts = null; // transaction parts that are seen in this latest round of slots from the server
+ private Map<Long, Map<Pair<Long, Integer>, CommitPart>> newCommitParts = null; // commit parts that are seen in this latest round of slots from the server
+ private Map<Long, Long> lastArbitratedTransactionNumberByArbitratorTable = null; // Last transaction sequence number that an arbitrator arbitrated on
+ private Map<Long, Transaction> liveTransactionBySequenceNumberTable = null; // live transaction grouped by the sequence number
+ private Map<Pair<Long, Long>, Transaction> liveTransactionByTransactionIdTable = null; // live transaction grouped by the transaction ID
+ private Map<Long, Map<Long, Commit>> liveCommitsTable = null;
+ private Map<IoTString, Commit> liveCommitsByKeyTable = null;
+ private Map<Long, Long> lastCommitSeenSequenceNumberByArbitratorTable = null;
+ private Vector<Long> rejectedSlotList = null; // List of rejected slots that have yet to be sent to the server
+ private List<Transaction> pendingTransactionQueue = null;
+ private List<ArbitrationRound> pendingSendArbitrationRounds = null;
+ private List<Entry> pendingSendArbitrationEntriesToDelete = null;
+ private Map<Transaction, List<Integer>> transactionPartsSent = null;
+ private Map<Long, TransactionStatus> outstandingTransactionStatus = null;
+ private Map<Long, Abort> liveAbortsGeneratedByLocal = null;
+ private Set<Pair<Long, Long>> offlineTransactionsCommittedAndAtServer = null;
+ private Map<Long, Pair<String, Integer>> localCommunicationTable = null;
+ private Map<Long, Long> lastTransactionSeenFromMachineFromServer = null;
+ private Map<Long, Long> lastArbitrationDataLocalSequenceNumberSeenFromArbitrator = null;
+
+
+ public Table(String baseurl, String password, long _localMachineId, int listeningPort, Context context) {
+ localMachineId = _localMachineId;
+ cloud = new CloudComm(this, baseurl, password, listeningPort, context);
+
+ init();
+ }
+
+ public Table(CloudComm _cloud, long _localMachineId) {
+ localMachineId = _localMachineId;
+ cloud = _cloud;
+
+ init();
+ }
+
+ /**
+ * Init all the stuff needed for for table usage
+ */
+ private void init() {
+
+ // Init helper objects
+ random = new Random();
+ buffer = new SlotBuffer();
+
+ // Set Variables
+ oldestLiveSlotSequenceNumver = 1;
+
+ // init data structs
+ committedKeyValueTable = new HashMap<IoTString, KeyValue>();
+ speculatedKeyValueTable = new HashMap<IoTString, KeyValue>();
+ pendingTransactionSpeculatedKeyValueTable = new HashMap<IoTString, KeyValue>();
+ liveNewKeyTable = new HashMap<IoTString, NewKey>();
+ lastMessageTable = new HashMap<Long, Pair<Long, Liveness>>();
+ rejectedMessageWatchListTable = new HashMap<Long, HashSet<RejectedMessage>>();
+ arbitratorTable = new HashMap<IoTString, Long>();
+ liveAbortTable = new HashMap<Pair<Long, Long>, Abort>();
+ newTransactionParts = new HashMap<Long, Map<Pair<Long, Integer>, TransactionPart>>();
+ newCommitParts = new HashMap<Long, Map<Pair<Long, Integer>, CommitPart>>();
+ lastArbitratedTransactionNumberByArbitratorTable = new HashMap<Long, Long>();
+ liveTransactionBySequenceNumberTable = new HashMap<Long, Transaction>();
+ liveTransactionByTransactionIdTable = new HashMap<Pair<Long, Long>, Transaction>();
+ liveCommitsTable = new HashMap<Long, Map<Long, Commit>>();
+ liveCommitsByKeyTable = new HashMap<IoTString, Commit>();
+ lastCommitSeenSequenceNumberByArbitratorTable = new HashMap<Long, Long>();
+ rejectedSlotList = new Vector<Long>();
+ pendingTransactionQueue = new ArrayList<Transaction>();
+ pendingSendArbitrationEntriesToDelete = new ArrayList<Entry>();
+ transactionPartsSent = new HashMap<Transaction, List<Integer>>();
+ outstandingTransactionStatus = new HashMap<Long, TransactionStatus>();
+ liveAbortsGeneratedByLocal = new HashMap<Long, Abort>();
+ offlineTransactionsCommittedAndAtServer = new HashSet<Pair<Long, Long>>();
+ localCommunicationTable = new HashMap<Long, Pair<String, Integer>>();
+ lastTransactionSeenFromMachineFromServer = new HashMap<Long, Long>();
+ pendingSendArbitrationRounds = new ArrayList<ArbitrationRound>();
+ lastArbitrationDataLocalSequenceNumberSeenFromArbitrator = new HashMap<Long, Long>();
+
+
+ // Other init stuff
+ numberOfSlots = buffer.capacity();
+ setResizeThreshold();
+ }
+
+ // TODO: delete method
+ public synchronized void printSlots() {
+ long o = buffer.getOldestSeqNum();
+ long n = buffer.getNewestSeqNum();
+
+ int[] types = new int[10];
+
+ int num = 0;
+
+ int livec = 0;
+ int deadc = 0;
+ for (long i = o; i < (n + 1); i++) {
+ Slot s = buffer.getSlot(i);
+
+ Vector<Entry> entries = s.getEntries();
+
+ for (Entry e : entries) {
+ if (e.isLive()) {
+ int type = e.getType();
+ types[type] = types[type] + 1;
+ num++;
+ livec++;
+ } else {
+ deadc++;
+ }
+ }
+ }
+
+ for (int i = 0; i < 10; i++) {
+ System.out.println(i + " " + types[i]);
+ }
+ System.out.println("Live count: " + livec);
+ System.out.println("Dead count: " + deadc);
+ System.out.println("Old: " + o);
+ System.out.println("New: " + n);
+ System.out.println("Size: " + buffer.size());
+ // System.out.println("Commits: " + liveCommitsTable.size());
+ System.out.println("pendingTrans: " + pendingTransactionQueue.size());
+ System.out.println("Trans Status Out: " + outstandingTransactionStatus.size());
+
+ for (Long k : lastArbitratedTransactionNumberByArbitratorTable.keySet()) {
+ System.out.println(k + ": " + lastArbitratedTransactionNumberByArbitratorTable.get(k));
+ }
+
+
+ for (Long a : liveCommitsTable.keySet()) {
+ for (Long b : liveCommitsTable.get(a).keySet()) {
+ for (KeyValue kv : liveCommitsTable.get(a).get(b).getKeyValueUpdateSet()) {
+ System.out.print(kv + " ");
+ }
+ System.out.print("|| ");
+ }
+ System.out.println();
+ }
+
+ }
+
+ /**
+ * Initialize the table by inserting a table status as the first entry into the table status
+ * also initialize the crypto stuff.
+ */
+ public synchronized void initTable() throws ServerException {
+ cloud.initSecurity();
+
+ // Create the first insertion into the block chain which is the table status
+ Slot s = new Slot(this, 1, localMachineId);
+ TableStatus status = new TableStatus(s, numberOfSlots);
+ s.addEntry(status);
+ Slot[] array = cloud.putSlot(s, numberOfSlots);
+
+ if (array == null) {
+ array = new Slot[] {s};
+ // update local block chain
+ validateAndUpdate(array, true);
+ } else if (array.length == 1) {
+ // in case we did push the slot BUT we failed to init it
+ validateAndUpdate(array, true);
+ } else {
+ throw new Error("Error on initialization");
+ }
+ }
+
+ /**
+ * Rebuild the table from scratch by pulling the latest block chain from the server.
+ */
+ public synchronized void rebuild() throws ServerException {
+ // Just pull the latest slots from the server
+ Slot[] newslots = cloud.getSlots(sequenceNumber + 1);
+ validateAndUpdate(newslots, true);
+ }
+
+ // public String toString() {
+ // String retString = " Committed Table: \n";
+ // retString += "---------------------------\n";
+ // retString += commitedTable.toString();
+
+ // retString += "\n\n";
+
+ // retString += " Speculative Table: \n";
+ // retString += "---------------------------\n";
+ // retString += speculativeTable.toString();
+
+ // return retString;
+ // }
+
+ public synchronized void addLocalCommunication(long arbitrator, String hostName, int portNumber) {
+ localCommunicationTable.put(arbitrator, new Pair<String, Integer>(hostName, portNumber));
+ }
+
+ public synchronized Long getArbitrator(IoTString key) {
+ return arbitratorTable.get(key);
+ }
+
+ public synchronized void close() {
+ cloud.close();
+ }
+
+ public synchronized IoTString getCommitted(IoTString key) {
+ KeyValue kv = committedKeyValueTable.get(key);
+
+ if (kv != null) {
+ return kv.getValue();
+ } else {
+ return null;
+ }
+ }
+
+ public synchronized IoTString getSpeculative(IoTString key) {
+ KeyValue kv = pendingTransactionSpeculatedKeyValueTable.get(key);
+
+ if (kv == null) {
+ kv = speculatedKeyValueTable.get(key);
+ }
+
+ if (kv == null) {
+ kv = committedKeyValueTable.get(key);
+ }
+
+ if (kv != null) {
+ return kv.getValue();
+ } else {
+ return null;
+ }
+ }
+
+ public synchronized IoTString getCommittedAtomic(IoTString key) {
+ KeyValue kv = committedKeyValueTable.get(key);
+
+ if (arbitratorTable.get(key) == null) {
+ throw new Error("Key not Found.");
+ }
+
+ // Make sure new key value pair matches the current arbitrator
+ if (!pendingTransactionBuilder.checkArbitrator(arbitratorTable.get(key))) {
+ // TODO: Maybe not throw en error
+ throw new Error("Not all Key Values Match Arbitrator.");
+ }
+
+ if (kv != null) {
+ pendingTransactionBuilder.addKVGuard(new KeyValue(key, kv.getValue()));
+ return kv.getValue();
+ } else {
+ pendingTransactionBuilder.addKVGuard(new KeyValue(key, null));
+ return null;
+ }
+ }
+
+ public synchronized IoTString getSpeculativeAtomic(IoTString key) {
+ if (arbitratorTable.get(key) == null) {
+ throw new Error("Key not Found.");
+ }
+
+ // Make sure new key value pair matches the current arbitrator
+ if (!pendingTransactionBuilder.checkArbitrator(arbitratorTable.get(key))) {
+ // TODO: Maybe not throw en error
+ throw new Error("Not all Key Values Match Arbitrator.");
+ }
+
+ KeyValue kv = pendingTransactionSpeculatedKeyValueTable.get(key);
+
+ if (kv == null) {
+ kv = speculatedKeyValueTable.get(key);
+ }
+
+ if (kv == null) {
+ kv = committedKeyValueTable.get(key);
+ }
+
+ if (kv != null) {
+ pendingTransactionBuilder.addKVGuard(new KeyValue(key, kv.getValue()));
+ return kv.getValue();
+ } else {
+ pendingTransactionBuilder.addKVGuard(new KeyValue(key, null));
+ return null;
+ }
+ }
+
+ public synchronized boolean update() {
+ try {
+ Slot[] newSlots = cloud.getSlots(sequenceNumber + 1);
+ validateAndUpdate(newSlots, false);
+ sendToServer(null);
+
+
+ updateLiveTransactionsAndStatus();
+
+ return true;
+ } catch (Exception e) {
+ // e.printStackTrace();
+
+ for (Long m : localCommunicationTable.keySet()) {
+ updateFromLocal(m);
+ }
+ }
+
+ return false;
+ }
+
+ public synchronized boolean createNewKey(IoTString keyName, long machineId) throws ServerException {
+ while (true) {
+ if (arbitratorTable.get(keyName) != null) {
+ // There is already an arbitrator
+ return false;
+ }
+
+ NewKey newKey = new NewKey(null, keyName, machineId);
+ if (sendToServer(newKey)) {
+ // If successfully inserted
+ return true;
+ }
+ }
+ }
+
+ public synchronized void startTransaction() {
+ // Create a new transaction, invalidates any old pending transactions.
+ pendingTransactionBuilder = new PendingTransaction(localMachineId);
+ }
+
+ public synchronized void addKV(IoTString key, IoTString value) {
+
+ // Make sure it is a valid key
+ if (arbitratorTable.get(key) == null) {
+ throw new Error("Key not Found.");
+ }
+
+ // Make sure new key value pair matches the current arbitrator
+ if (!pendingTransactionBuilder.checkArbitrator(arbitratorTable.get(key))) {
+ // TODO: Maybe not throw en error
+ throw new Error("Not all Key Values Match Arbitrator.");
+ }
+
+ // Add the key value to this transaction
+ KeyValue kv = new KeyValue(key, value);
+ pendingTransactionBuilder.addKV(kv);
+ }
+
+ public synchronized TransactionStatus commitTransaction() {
+
+ if (pendingTransactionBuilder.getKVUpdates().size() == 0) {
+ // transaction with no updates will have no effect on the system
+ return new TransactionStatus(TransactionStatus.StatusNoEffect, -1);
+ }
+
+ // Set the local transaction sequence number and increment
+ pendingTransactionBuilder.setClientLocalSequenceNumber(localTransactionSequenceNumber);
+ localTransactionSequenceNumber++;
+
+ // Create the transaction status
+ TransactionStatus transactionStatus = new TransactionStatus(TransactionStatus.StatusPending, pendingTransactionBuilder.getArbitrator());
+
+ // Create the new transaction
+ Transaction newTransaction = pendingTransactionBuilder.createTransaction();
+ newTransaction.setTransactionStatus(transactionStatus);
+
+ if (pendingTransactionBuilder.getArbitrator() != localMachineId) {
+ // Add it to the queue and invalidate the builder for safety
+ pendingTransactionQueue.add(newTransaction);
+ } else {
+ arbitrateOnLocalTransaction(newTransaction);
+ updateLiveStateFromLocal();
+ }
+
+ pendingTransactionBuilder = new PendingTransaction(localMachineId);
+
+ try {
+ sendToServer(null);
+ } catch (ServerException e) {
+
+ Set<Long> arbitratorTriedAndFailed = new HashSet<Long>();
+ for (Iterator<Transaction> iter = pendingTransactionQueue.iterator(); iter.hasNext(); ) {
+ Transaction transaction = iter.next();
+
+ if (arbitratorTriedAndFailed.contains(transaction.getArbitrator())) {
+ // Already contacted this client so ignore all attempts to contact this client
+ // to preserve ordering for arbitrator
+ continue;
+ }
+
+ Pair<Boolean, Boolean> sendReturn = sendTransactionToLocal(transaction);
+
+ if (sendReturn.getFirst()) {
+ // Failed to contact over local
+ arbitratorTriedAndFailed.add(transaction.getArbitrator());
+ } else {
+ // Successful contact or should not contact
+
+ if (sendReturn.getSecond()) {
+ // did arbitrate
+ iter.remove();
+ }
+ }
+ }
+ }
+
+ updateLiveStateFromLocal();
+
+ return transactionStatus;
+ }
+
+ /**
+ * Get the machine ID for this client
+ */
+ public long getMachineId() {
+ return localMachineId;
+ }
+
+ /**
+ * Decrement the number of live slots that we currently have
+ */
+ public void decrementLiveCount() {
+ liveSlotCount--;
+ }
+
+ /**
+ * Recalculate the new resize threshold
+ */
+ private void setResizeThreshold() {
+ int resizeLower = (int) (RESIZE_THRESHOLD * numberOfSlots);
+ bufferResizeThreshold = resizeLower - 1 + random.nextInt(numberOfSlots - resizeLower);
+ }
+
+
+ boolean lastInsertedNewKey = false;
+
+ private boolean sendToServer(NewKey newKey) throws ServerException {
+
+ boolean fromRetry = false;
+
+ try {
+ if (hadPartialSendToServer) {
+ Slot[] newSlots = cloud.getSlots(sequenceNumber + 1);
+ if (newSlots.length == 0) {
+ fromRetry = true;
+ ThreeTuple<Boolean, Boolean, Slot[]> sendSlotsReturn = sendSlotsToServer(lastSlotAttemptedToSend, lastNewSize, lastIsNewKey);
+
+ if (sendSlotsReturn.getFirst()) {
+ if (newKey != null) {
+ if (lastInsertedNewKey && (lastNewKey.getKey() == newKey.getKey()) && (lastNewKey.getMachineID() == newKey.getMachineID())) {
+ newKey = null;
+ }
+ }
+
+ for (Transaction transaction : lastTransactionPartsSent.keySet()) {
+ transaction.resetServerFailure();
+
+ // Update which transactions parts still need to be sent
+ transaction.removeSentParts(lastTransactionPartsSent.get(transaction));
+
+ // Add the transaction status to the outstanding list
+ outstandingTransactionStatus.put(transaction.getSequenceNumber(), transaction.getTransactionStatus());
+
+ // Update the transaction status
+ transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentPartial);
+
+ // Check if all the transaction parts were successfully sent and if so then remove it from pending
+ if (transaction.didSendAllParts()) {
+ transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentFully);
+ pendingTransactionQueue.remove(transaction);
+ }
+ }
+ } else {
+
+ newSlots = sendSlotsReturn.getThird();
+
+ boolean isInserted = false;
+ for (Slot s : newSlots) {
+ if ((s.getSequenceNumber() == lastSlotAttemptedToSend.getSequenceNumber()) && (s.getMachineID() == localMachineId)) {
+ isInserted = true;
+ break;
+ }
+ }
+
+ for (Slot s : newSlots) {
+ if (isInserted) {
+ break;
+ }
+
+ // Process each entry in the slot
+ for (Entry entry : s.getEntries()) {
+
+ if (entry.getType() == Entry.TypeLastMessage) {
+ LastMessage lastMessage = (LastMessage)entry;
+ if ((lastMessage.getMachineID() == localMachineId) && (lastMessage.getSequenceNumber() == lastSlotAttemptedToSend.getSequenceNumber())) {
+ isInserted = true;
+ break;
+ }
+ }
+ }
+ }
+
+ if (isInserted) {
+ if (newKey != null) {
+ if (lastInsertedNewKey && (lastNewKey.getKey() == newKey.getKey()) && (lastNewKey.getMachineID() == newKey.getMachineID())) {
+ newKey = null;
+ }
+ }
+
+ for (Transaction transaction : lastTransactionPartsSent.keySet()) {
+ transaction.resetServerFailure();
+
+ // Update which transactions parts still need to be sent
+ transaction.removeSentParts(lastTransactionPartsSent.get(transaction));
+
+ // Add the transaction status to the outstanding list
+ outstandingTransactionStatus.put(transaction.getSequenceNumber(), transaction.getTransactionStatus());
+
+ // Update the transaction status
+ transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentPartial);
+
+ // Check if all the transaction parts were successfully sent and if so then remove it from pending
+ if (transaction.didSendAllParts()) {
+ transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentFully);
+ pendingTransactionQueue.remove(transaction);
+ } else {
+ transaction.resetServerFailure();
+ // Set the transaction sequence number back to nothing
+ if (!transaction.didSendAPartToServer()) {
+ transaction.setSequenceNumber(-1);
+ }
+ }
+ }
+ }
+ }
+
+ for (Transaction transaction : lastTransactionPartsSent.keySet()) {
+ transaction.resetServerFailure();
+ // Set the transaction sequence number back to nothing
+ if (!transaction.didSendAPartToServer()) {
+ transaction.setSequenceNumber(-1);
+ }
+ }
+
+ if (sendSlotsReturn.getThird().length != 0) {
+ // insert into the local block chain
+ validateAndUpdate(sendSlotsReturn.getThird(), true);
+ }
+ // continue;
+ } else {
+ boolean isInserted = false;
+ for (Slot s : newSlots) {
+ if ((s.getSequenceNumber() == lastSlotAttemptedToSend.getSequenceNumber()) && (s.getMachineID() == localMachineId)) {
+ isInserted = true;
+ break;
+ }
+ }
+
+ for (Slot s : newSlots) {
+ if (isInserted) {
+ break;
+ }
+
+ // Process each entry in the slot
+ for (Entry entry : s.getEntries()) {
+
+ if (entry.getType() == Entry.TypeLastMessage) {
+ LastMessage lastMessage = (LastMessage)entry;
+ if ((lastMessage.getMachineID() == localMachineId) && (lastMessage.getSequenceNumber() == lastSlotAttemptedToSend.getSequenceNumber())) {
+ isInserted = true;
+ break;
+ }
+ }
+ }
+ }
+
+ if (isInserted) {
+ if (newKey != null) {
+ if (lastInsertedNewKey && (lastNewKey.getKey() == newKey.getKey()) && (lastNewKey.getMachineID() == newKey.getMachineID())) {
+ newKey = null;
+ }
+ }
+
+ for (Transaction transaction : lastTransactionPartsSent.keySet()) {
+ transaction.resetServerFailure();
+
+ // Update which transactions parts still need to be sent
+ transaction.removeSentParts(lastTransactionPartsSent.get(transaction));
+
+ // Add the transaction status to the outstanding list
+ outstandingTransactionStatus.put(transaction.getSequenceNumber(), transaction.getTransactionStatus());
+
+ // Update the transaction status
+ transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentPartial);
+
+ // Check if all the transaction parts were successfully sent and if so then remove it from pending
+ if (transaction.didSendAllParts()) {
+ transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentFully);
+ pendingTransactionQueue.remove(transaction);
+ } else {
+ transaction.resetServerFailure();
+ // Set the transaction sequence number back to nothing
+ if (!transaction.didSendAPartToServer()) {
+ transaction.setSequenceNumber(-1);
+ }
+ }
+ }
+ } else {
+ for (Transaction transaction : lastTransactionPartsSent.keySet()) {
+ transaction.resetServerFailure();
+ // Set the transaction sequence number back to nothing
+ if (!transaction.didSendAPartToServer()) {
+ transaction.setSequenceNumber(-1);
+ }
+ }
+ }
+
+ // insert into the local block chain
+ validateAndUpdate(newSlots, true);
+ }
+ }
+ } catch (ServerException e) {
+ throw e;
+ }
+
+
+ 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) {
+ throw new Error("Should Be error free");
+ }
+
+
+
+ // If there is a new key with same name then end
+ if ((newKey != null) && (arbitratorTable.get(newKey.getKey()) != null)) {
+ return false;
+ }
+
+ // Create the slot
+ Slot slot = new Slot(this, sequenceNumber + 1, localMachineId, buffer.getSlot(sequenceNumber).getHMAC());
+
+ // Try to fill the slot with data
+ ThreeTuple<Boolean, Integer, Boolean> fillSlotsReturn = fillSlot(slot, false, newKey);
+ boolean needsResize = fillSlotsReturn.getFirst();
+ int newSize = fillSlotsReturn.getSecond();
+ Boolean insertedNewKey = fillSlotsReturn.getThird();
+
+ if (needsResize) {
+ // Reset which transaction to send
+ for (Transaction transaction : transactionPartsSent.keySet()) {
+ transaction.resetNextPartToSend();
+
+ // Set the transaction sequence number back to nothing
+ if (!transaction.didSendAPartToServer() && !transaction.getServerFailure()) {
+ transaction.setSequenceNumber(-1);
+ }
+ }
+
+ // Clear the sent data since we are trying again
+ pendingSendArbitrationEntriesToDelete.clear();
+ transactionPartsSent.clear();
+
+ // We needed a resize so try again
+ fillSlot(slot, true, newKey);
+ }
+
+ lastSlotAttemptedToSend = slot;
+ lastIsNewKey = (newKey != null);
+ lastInsertedNewKey = insertedNewKey;
+ lastNewSize = newSize;
+ lastNewKey = newKey;
+ lastTransactionPartsSent = new HashMap<Transaction, List<Integer>>(transactionPartsSent);
+ lastPendingSendArbitrationEntriesToDelete = new ArrayList<Entry>(pendingSendArbitrationEntriesToDelete);
+
+
+ ThreeTuple<Boolean, Boolean, Slot[]> sendSlotsReturn = sendSlotsToServer(slot, newSize, newKey != null);
+
+ if (sendSlotsReturn.getFirst()) {
+
+ // Did insert into the block chain
+
+ if (insertedNewKey) {
+ // This slot was what was inserted not a previous slot
+
+ // New Key was successfully inserted into the block chain so dont want to insert it again
+ newKey = null;
+ }
+
+ // Remove the aborts and commit parts that were sent from the pending to send queue
+ for (Iterator<ArbitrationRound> iter = pendingSendArbitrationRounds.iterator(); iter.hasNext(); ) {
+ ArbitrationRound round = iter.next();
+ round.removeParts(pendingSendArbitrationEntriesToDelete);
+
+ if (round.isDoneSending()) {
+ // Sent all the parts
+ iter.remove();
+ }
+ }
+
+ for (Transaction transaction : transactionPartsSent.keySet()) {
+ transaction.resetServerFailure();
+
+ // Update which transactions parts still need to be sent
+ transaction.removeSentParts(transactionPartsSent.get(transaction));
+
+ // Add the transaction status to the outstanding list
+ outstandingTransactionStatus.put(transaction.getSequenceNumber(), transaction.getTransactionStatus());
+
+ // Update the transaction status
+ transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentPartial);
+
+ // Check if all the transaction parts were successfully sent and if so then remove it from pending
+ if (transaction.didSendAllParts()) {
+ transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentFully);
+ pendingTransactionQueue.remove(transaction);
+ }
+ }
+ } else {
+
+ // if (!sendSlotsReturn.getSecond()) {
+ // for (Transaction transaction : lastTransactionPartsSent.keySet()) {
+ // transaction.resetServerFailure();
+ // }
+ // } else {
+ // for (Transaction transaction : lastTransactionPartsSent.keySet()) {
+ // transaction.resetServerFailure();
+
+ // // Update which transactions parts still need to be sent
+ // transaction.removeSentParts(transactionPartsSent.get(transaction));
+
+ // // Add the transaction status to the outstanding list
+ // outstandingTransactionStatus.put(transaction.getSequenceNumber(), transaction.getTransactionStatus());
+
+ // // Update the transaction status
+ // transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentPartial);
+
+ // // Check if all the transaction parts were successfully sent and if so then remove it from pending
+ // if (transaction.didSendAllParts()) {
+ // transaction.getTransactionStatus().setStatus(TransactionStatus.StatusSentFully);
+ // pendingTransactionQueue.remove(transaction);
+
+ // for (KeyValue kv : transaction.getKeyValueUpdateSet()) {
+ // System.out.println("Sent: " + kv + " from: " + localMachineId + " Slot:" + lastSlotAttemptedToSend.getSequenceNumber() + " Claimed:" + transaction.getSequenceNumber());
+ // }
+ // }
+ // }
+ // }
+
+ // Reset which transaction to send
+ for (Transaction transaction : transactionPartsSent.keySet()) {
+ transaction.resetNextPartToSend();
+ // transaction.resetNextPartToSend();
+
+ // Set the transaction sequence number back to nothing
+ if (!transaction.didSendAPartToServer() && !transaction.getServerFailure()) {
+ transaction.setSequenceNumber(-1);
+ }
+ }
+ }
+
+ // Clear the sent data in preparation for next send
+ pendingSendArbitrationEntriesToDelete.clear();
+ transactionPartsSent.clear();
+
+ if (sendSlotsReturn.getThird().length != 0) {
+ // insert into the local block chain
+ validateAndUpdate(sendSlotsReturn.getThird(), true);
+ }
+ }
+
+ } catch (ServerException e) {
+
+ if (e.getType() != ServerException.TypeInputTimeout) {
+ // e.printStackTrace();
+
+ // Nothing was able to be sent to the server so just clear these data structures
+ for (Transaction transaction : transactionPartsSent.keySet()) {
+ transaction.resetNextPartToSend();
+
+ // Set the transaction sequence number back to nothing
+ if (!transaction.didSendAPartToServer() && !transaction.getServerFailure()) {
+ transaction.setSequenceNumber(-1);
+ }
+ }
+ } else {
+ // There was a partial send to the server
+ hadPartialSendToServer = true;
+
+
+ // if (!fromRetry) {
+ // lastTransactionPartsSent = new HashMap<Transaction, List<Integer>>(transactionPartsSent);
+ // lastPendingSendArbitrationEntriesToDelete = new ArrayList<Entry>(pendingSendArbitrationEntriesToDelete);
+ // }
+
+ // Nothing was able to be sent to the server so just clear these data structures
+ for (Transaction transaction : transactionPartsSent.keySet()) {
+ transaction.resetNextPartToSend();
+ transaction.setServerFailure();
+ }
+ }
+
+ pendingSendArbitrationEntriesToDelete.clear();
+ transactionPartsSent.clear();
+
+ throw e;
+ }
+
+ return newKey == null;
+ }
+
+ 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
+ return false;
+ }
+
+ // Get the size of the send data
+ int sendDataSize = Integer.BYTES + Long.BYTES;
+
+ Long lastArbitrationDataLocalSequenceNumber = (long) - 1;
+ if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.get(machineId) != null) {
+ lastArbitrationDataLocalSequenceNumber = lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.get(machineId);
+ }
+
+ byte[] sendData = new byte[sendDataSize];
+ ByteBuffer bbEncode = ByteBuffer.wrap(sendData);
+
+ // Encode the data
+ bbEncode.putLong(lastArbitrationDataLocalSequenceNumber);
+ bbEncode.putInt(0);
+
+ // Send by local
+ byte[] returnData = cloud.sendLocalData(sendData, localCommunicationInformation.getFirst(), localCommunicationInformation.getSecond());
+
+ if (returnData == null) {
+ // Could not contact server
+ return false;
+ }
+
+ // Decode the data
+ ByteBuffer bbDecode = ByteBuffer.wrap(returnData);
+ int numberOfEntries = bbDecode.getInt();
+
+ for (int i = 0; i < numberOfEntries; i++) {
+ byte type = bbDecode.get();
+ if (type == Entry.TypeAbort) {
+ Abort abort = (Abort)Abort.decode(null, bbDecode);
+ processEntry(abort);
+ } else if (type == Entry.TypeCommitPart) {
+ CommitPart commitPart = (CommitPart)CommitPart.decode(null, bbDecode);
+ processEntry(commitPart);
+ }
+ }
+
+ updateLiveStateFromLocal();
+
+ return true;
+ }
+
+ private Pair<Boolean, Boolean> sendTransactionToLocal(Transaction transaction) {
+
+ // Get the devices local communications
+ Pair<String, Integer> localCommunicationInformation = localCommunicationTable.get(transaction.getArbitrator());
+
+ if (localCommunicationInformation == null) {
+ // Cant talk to that device locally so do nothing
+ return new Pair<Boolean, Boolean>(true, false);
+ }
+
+ // Get the size of the send data
+ int sendDataSize = Integer.BYTES + Long.BYTES;
+ for (TransactionPart part : transaction.getParts().values()) {
+ sendDataSize += part.getSize();
+ }
+
+ Long lastArbitrationDataLocalSequenceNumber = (long) - 1;
+ if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.get(transaction.getArbitrator()) != null) {
+ lastArbitrationDataLocalSequenceNumber = lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.get(transaction.getArbitrator());
+ }
+
+ // Make the send data size
+ byte[] sendData = new byte[sendDataSize];
+ ByteBuffer bbEncode = ByteBuffer.wrap(sendData);
+
+ // Encode the data
+ bbEncode.putLong(lastArbitrationDataLocalSequenceNumber);
+ bbEncode.putInt(transaction.getParts().size());
+ for (TransactionPart part : transaction.getParts().values()) {
+ part.encode(bbEncode);
+ }
+
+
+ // Send by local
+ byte[] returnData = cloud.sendLocalData(sendData, localCommunicationInformation.getFirst(), localCommunicationInformation.getSecond());
+
+ if (returnData == null) {
+ // Could not contact server
+ return new Pair<Boolean, Boolean>(true, false);
+ }
+
+ // Decode the data
+ ByteBuffer bbDecode = ByteBuffer.wrap(returnData);
+ boolean didCommit = bbDecode.get() == 1;
+ boolean couldArbitrate = bbDecode.get() == 1;
+ int numberOfEntries = bbDecode.getInt();
+ boolean foundAbort = false;
+
+ for (int i = 0; i < numberOfEntries; i++) {
+ byte type = bbDecode.get();
+ if (type == Entry.TypeAbort) {
+ Abort abort = (Abort)Abort.decode(null, bbDecode);
+
+ if ((abort.getTransactionMachineId() == localMachineId) && (abort.getTransactionClientLocalSequenceNumber() == transaction.getClientLocalSequenceNumber())) {
+ foundAbort = true;
+ }
+
+ processEntry(abort);
+ } else if (type == Entry.TypeCommitPart) {
+ CommitPart commitPart = (CommitPart)CommitPart.decode(null, bbDecode);
+ processEntry(commitPart);
+ }
+ }
+
+ updateLiveStateFromLocal();
+
+ if (couldArbitrate) {
+ TransactionStatus status = transaction.getTransactionStatus();
+ if (didCommit) {
+ status.setStatus(TransactionStatus.StatusCommitted);
+ } else {
+ status.setStatus(TransactionStatus.StatusAborted);
+ }
+ } else {
+ TransactionStatus status = transaction.getTransactionStatus();
+ if (foundAbort) {
+ status.setStatus(TransactionStatus.StatusAborted);
+ } else {
+ status.setStatus(TransactionStatus.StatusCommitted);
+ }
+ }
+
+ return new Pair<Boolean, Boolean>(false, true);
+ }
+
+ public synchronized byte[] acceptDataFromLocal(byte[] data) {
+
+ // Decode the data
+ ByteBuffer bbDecode = ByteBuffer.wrap(data);
+ long lastArbitratedSequenceNumberSeen = bbDecode.getLong();
+ int numberOfParts = bbDecode.getInt();
+
+ // If we did commit a transaction or not
+ boolean didCommit = false;
+ boolean couldArbitrate = false;
+
+ if (numberOfParts != 0) {
+
+ // decode the transaction
+ Transaction transaction = new Transaction();
+ for (int i = 0; i < numberOfParts; i++) {
+ bbDecode.get();
+ TransactionPart newPart = (TransactionPart)TransactionPart.decode(null, bbDecode);
+ transaction.addPartDecode(newPart);
+ }
+
+ // Arbitrate on transaction and pull relevant return data
+ Pair<Boolean, Boolean> localArbitrateReturn = arbitrateOnLocalTransaction(transaction);
+ couldArbitrate = localArbitrateReturn.getFirst();
+ didCommit = localArbitrateReturn.getSecond();
+
+ updateLiveStateFromLocal();
+
+ // Transaction was sent to the server so keep track of it to prevent double commit
+ if (transaction.getSequenceNumber() != -1) {
+ offlineTransactionsCommittedAndAtServer.add(transaction.getId());
+ }
+ }
+
+ // The data to send back
+ int returnDataSize = 0;
+ List<Entry> unseenArbitrations = new ArrayList<Entry>();
+
+ // Get the aborts to send back
+ List<Long> abortLocalSequenceNumbers = new ArrayList<Long >(liveAbortsGeneratedByLocal.keySet());
+ Collections.sort(abortLocalSequenceNumbers);
+ for (Long localSequenceNumber : abortLocalSequenceNumbers) {
+ if (localSequenceNumber <= lastArbitratedSequenceNumberSeen) {
+ continue;
+ }
+
+ Abort abort = liveAbortsGeneratedByLocal.get(localSequenceNumber);
+ unseenArbitrations.add(abort);
+ returnDataSize += abort.getSize();
+ }
+
+ // Get the commits to send back
+ Map<Long, Commit> commitForClientTable = liveCommitsTable.get(localMachineId);
+ if (commitForClientTable != null) {
+ List<Long> commitLocalSequenceNumbers = new ArrayList<Long>(commitForClientTable.keySet());
+ Collections.sort(commitLocalSequenceNumbers);
+
+ for (Long localSequenceNumber : commitLocalSequenceNumbers) {
+ Commit commit = commitForClientTable.get(localSequenceNumber);
+
+ if (localSequenceNumber <= lastArbitratedSequenceNumberSeen) {
+ continue;
+ }
+
+ unseenArbitrations.addAll(commit.getParts().values());
+
+ for (CommitPart commitPart : commit.getParts().values()) {
+ returnDataSize += commitPart.getSize();
+ }
+ }
+ }
+
+ // Number of arbitration entries to decode
+ returnDataSize += 2 * Integer.BYTES;
+
+ // Boolean of did commit or not
+ if (numberOfParts != 0) {
+ returnDataSize += Byte.BYTES;
+ }
+
+ // Data to send Back
+ byte[] returnData = new byte[returnDataSize];
+ ByteBuffer bbEncode = ByteBuffer.wrap(returnData);
+
+ if (numberOfParts != 0) {
+ if (didCommit) {
+ bbEncode.put((byte)1);
+ } else {
+ bbEncode.put((byte)0);
+ }
+ if (couldArbitrate) {
+ bbEncode.put((byte)1);
+ } else {
+ bbEncode.put((byte)0);
+ }
+ }
+
+ bbEncode.putInt(unseenArbitrations.size());
+ for (Entry entry : unseenArbitrations) {
+ entry.encode(bbEncode);
+ }
+
+ return returnData;
+ }
+
+ private ThreeTuple<Boolean, Boolean, Slot[]> sendSlotsToServer(Slot slot, int newSize, boolean isNewKey) throws ServerException {
+
+ boolean attemptedToSendToServerTmp = attemptedToSendToServer;
+ attemptedToSendToServer = true;
+
+ boolean inserted = false;
+ boolean lastTryInserted = false;
+
+ Slot[] array = cloud.putSlot(slot, newSize);
+ if (array == null) {
+ array = new Slot[] {slot};
+ rejectedSlotList.clear();
+ inserted = true;
+ } else {
+ if (array.length == 0) {
+ throw new Error("Server Error: Did not send any slots");
+ }
+
+ // if (attemptedToSendToServerTmp) {
+ if (hadPartialSendToServer) {
+
+ boolean isInserted = false;
+ for (Slot s : array) {
+ if ((s.getSequenceNumber() == slot.getSequenceNumber()) && (s.getMachineID() == localMachineId)) {
+ isInserted = true;
+ break;
+ }
+ }
+
+ for (Slot s : array) {
+ if (isInserted) {
+ break;
+ }
+
+ // Process each entry in the slot
+ for (Entry entry : s.getEntries()) {
+
+ if (entry.getType() == Entry.TypeLastMessage) {
+ LastMessage lastMessage = (LastMessage)entry;
+
+ if ((lastMessage.getMachineID() == localMachineId) && (lastMessage.getSequenceNumber() == slot.getSequenceNumber())) {
+ isInserted = true;
+ break;
+ }
+ }
+ }
+ }
+
+ if (!isInserted) {
+ rejectedSlotList.add(slot.getSequenceNumber());
+ lastTryInserted = false;
+ } else {
+ lastTryInserted = true;
+ }
+ } else {
+ rejectedSlotList.add(slot.getSequenceNumber());
+ lastTryInserted = false;
+ }
+ }
+
+ return new ThreeTuple<Boolean, Boolean, Slot[]>(inserted, lastTryInserted, array);
+ }
+
+ /**
+ * 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) {
+ newSize = (int) (numberOfSlots * RESIZE_MULTIPLE);
+ TableStatus status = new TableStatus(slot, newSize);
+ slot.addEntry(status);
+ }
+
+ // Fill with rejected slots first before doing anything else
+ doRejectedMessages(slot);
+
+ // Do mandatory rescue of entries
+ ThreeTuple<Boolean, Boolean, Long> mandatoryRescueReturn = doMandatoryResuce(slot, resize);
+
+ // Extract working variables
+ boolean needsResize = mandatoryRescueReturn.getFirst();
+ boolean seenLiveSlot = mandatoryRescueReturn.getSecond();
+ long currentRescueSequenceNumber = mandatoryRescueReturn.getThird();
+
+ if (needsResize && !resize) {
+ // We need to resize but we are not resizing so return false
+ return new ThreeTuple<Boolean, Integer, Boolean>(true, null, null);
+ }
+
+ boolean inserted = false;
+ if (newKeyEntry != null) {
+ newKeyEntry.setSlot(slot);
+ if (slot.hasSpace(newKeyEntry)) {
+ slot.addEntry(newKeyEntry);
+ inserted = true;
+ }
+ }
+
+ // Clear the transactions, aborts and commits that were sent previously
+ transactionPartsSent.clear();
+ pendingSendArbitrationEntriesToDelete.clear();
+
+ for (ArbitrationRound round : pendingSendArbitrationRounds) {
+ boolean isFull = false;
+ round.generateParts();
+ List<Entry> parts = round.getParts();
+
+ // Insert pending arbitration data
+ for (Entry arbitrationData : parts) {
+
+ // If it is an abort then we need to set some information
+ if (arbitrationData instanceof Abort) {
+ ((Abort)arbitrationData).setSequenceNumber(slot.getSequenceNumber());
+ }
+
+ if (!slot.hasSpace(arbitrationData)) {
+ // No space so cant do anything else with these data entries
+ isFull = true;
+ break;
+ }
+
+ // Add to this current slot and add it to entries to delete
+ slot.addEntry(arbitrationData);
+ pendingSendArbitrationEntriesToDelete.add(arbitrationData);
+ }
+
+ if (isFull) {
+ break;
+ }
+ }
+
+ if (pendingTransactionQueue.size() > 0) {
+
+ Transaction transaction = pendingTransactionQueue.get(0);
+
+ // Set the transaction sequence number if it has yet to be inserted into the block chain
+ // if ((!transaction.didSendAPartToServer() && !transaction.getServerFailure()) || (transaction.getSequenceNumber() == -1)) {
+ // transaction.setSequenceNumber(slot.getSequenceNumber());
+ // }
+
+ if ((!transaction.didSendAPartToServer()) || (transaction.getSequenceNumber() == -1)) {
+ transaction.setSequenceNumber(slot.getSequenceNumber());
+ }
+
+
+ while (true) {
+ TransactionPart part = transaction.getNextPartToSend();
+
+ if (part == null) {
+ // Ran out of parts to send for this transaction so move on
+ break;
+ }
+
+ if (slot.hasSpace(part)) {
+ slot.addEntry(part);
+ List<Integer> partsSent = transactionPartsSent.get(transaction);
+ if (partsSent == null) {
+ partsSent = new ArrayList<Integer>();
+ transactionPartsSent.put(transaction, partsSent);
+ }
+ partsSent.add(part.getPartNumber());
+ transactionPartsSent.put(transaction, partsSent);
+ } else {
+ break;
+ }
+ }
+ }
+
+ // Fill the remainder of the slot with rescue data
+ doOptionalRescue(slot, seenLiveSlot, currentRescueSequenceNumber, resize);
+
+ return new ThreeTuple<Boolean, Integer, Boolean>(false, newSize, inserted);
+ }
+
+ private void doRejectedMessages(Slot s) {
+ if (! rejectedSlotList.isEmpty()) {
+ /* TODO: We should avoid generating a rejected message entry if
+ * there is already a sufficient entry in the queue (e.g.,
+ * equalsto value of true and same sequence number). */
+
+ long old_seqn = rejectedSlotList.firstElement();
+ if (rejectedSlotList.size() > REJECTED_THRESHOLD) {
+ long new_seqn = rejectedSlotList.lastElement();
+ RejectedMessage rm = new RejectedMessage(s, s.getSequenceNumber(), localMachineId, old_seqn, new_seqn, false);
+ s.addEntry(rm);
+ } else {
+ long prev_seqn = -1;
+ int i = 0;
+ /* Go through list of missing messages */
+ for (; i < rejectedSlotList.size(); i++) {
+ long curr_seqn = rejectedSlotList.get(i);
+ Slot s_msg = buffer.getSlot(curr_seqn);
+ if (s_msg != null)
+ break;
+ prev_seqn = curr_seqn;
+ }
+ /* Generate rejected message entry for missing messages */
+ if (prev_seqn != -1) {
+ RejectedMessage rm = new RejectedMessage(s, s.getSequenceNumber(), localMachineId, old_seqn, prev_seqn, false);
+ s.addEntry(rm);
+ }
+ /* Generate rejected message entries for present messages */
+ for (; i < rejectedSlotList.size(); i++) {
+ long curr_seqn = rejectedSlotList.get(i);
+ Slot s_msg = buffer.getSlot(curr_seqn);
+ long machineid = s_msg.getMachineID();
+ RejectedMessage rm = new RejectedMessage(s, s.getSequenceNumber(), machineid, curr_seqn, curr_seqn, true);
+ s.addEntry(rm);
+ }
+ }
+ }
+ }
+
+ private ThreeTuple<Boolean, Boolean, Long> doMandatoryResuce(Slot slot, boolean resize) {
+ long newestSequenceNumber = buffer.getNewestSeqNum();
+ long oldestSequenceNumber = buffer.getOldestSeqNum();
+ if (oldestLiveSlotSequenceNumver < oldestSequenceNumber) {
+ oldestLiveSlotSequenceNumver = oldestSequenceNumber;
+ }
+
+ long currentSequenceNumber = oldestLiveSlotSequenceNumver;
+ boolean seenLiveSlot = false;
+ long firstIfFull = newestSequenceNumber + 1 - numberOfSlots; // smallest seq number in the buffer if it is full
+ long threshold = firstIfFull + FREE_SLOTS; // we want the buffer to be clear of live entries up to this point
+
+
+ // Mandatory Rescue
+ for (; currentSequenceNumber < threshold; currentSequenceNumber++) {
+ Slot previousSlot = buffer.getSlot(currentSequenceNumber);
+ // Push slot number forward
+ if (! seenLiveSlot) {
+ oldestLiveSlotSequenceNumver = currentSequenceNumber;
+ }
+
+ if (!previousSlot.isLive()) {
+ continue;
+ }
+
+ // We have seen a live slot
+ seenLiveSlot = true;
+
+ // Get all the live entries for a slot
+ Vector<Entry> liveEntries = previousSlot.getLiveEntries(resize);
+
+ // Iterate over all the live entries and try to rescue them
+ for (Entry liveEntry : liveEntries) {
+ if (slot.hasSpace(liveEntry)) {
+
+ // Enough space to rescue the entry
+ slot.addEntry(liveEntry);
+ } else if (currentSequenceNumber == firstIfFull) {
+ //if there's no space but the entry is about to fall off the queue
+ System.out.println("B"); //?
+ return new ThreeTuple<Boolean, Boolean, Long>(true, seenLiveSlot, currentSequenceNumber);
+
+ }
+ }
+ }
+
+ // Did not resize
+ return new ThreeTuple<Boolean, Boolean, Long>(false, seenLiveSlot, currentSequenceNumber);
+ }
+
+ private void doOptionalRescue(Slot s, boolean seenliveslot, long seqn, boolean resize) {
+ /* now go through live entries from least to greatest sequence number until
+ * either all live slots added, or the slot doesn't have enough room
+ * for SKIP_THRESHOLD consecutive entries*/
+ int skipcount = 0;
+ long newestseqnum = buffer.getNewestSeqNum();
+ search:
+ for (; seqn <= newestseqnum; seqn++) {
+ Slot prevslot = buffer.getSlot(seqn);
+ //Push slot number forward
+ if (!seenliveslot)
+ oldestLiveSlotSequenceNumver = seqn;
+
+ if (!prevslot.isLive())
+ continue;
+ seenliveslot = true;
+ Vector<Entry> liveentries = prevslot.getLiveEntries(resize);
+ for (Entry liveentry : liveentries) {
+ if (s.hasSpace(liveentry))
+ s.addEntry(liveentry);
+ else {
+ skipcount++;
+ if (skipcount > SKIP_THRESHOLD)
+ break search;
+ }
+ }
+ }
+ }
+
+ /**
+ * Checks for malicious activity and updates the local copy of the block chain.
+ */
+ private void validateAndUpdate(Slot[] newSlots, boolean acceptUpdatesToLocal) {
+
+ // The cloud communication layer has checked slot HMACs already before decoding
+ if (newSlots.length == 0) {
+ return;
+ }
+
+ // Make sure all slots are newer than the last largest slot this client has seen
+ long firstSeqNum = newSlots[0].getSequenceNumber();
+ if (firstSeqNum <= sequenceNumber) {
+ throw new Error("Server Error: Sent older slots!");
+ }
+
+ // Create an object that can access both new slots and slots in our local chain
+ // without committing slots to our local chain
+ SlotIndexer indexer = new SlotIndexer(newSlots, buffer);
+
+ // Check that the HMAC chain is not broken
+ checkHMACChain(indexer, newSlots);
+
+ // Set to keep track of messages from clients
+ HashSet<Long> machineSet = new HashSet<Long>(lastMessageTable.keySet());
+
+ // Process each slots data
+ for (Slot slot : newSlots) {
+ processSlot(indexer, slot, acceptUpdatesToLocal, machineSet);
+ updateExpectedSize();
+ }
+
+ // If there is a gap, check to see if the server sent us everything.
+ if (firstSeqNum != (sequenceNumber + 1)) {
+
+ // Check the size of the slots that were sent down by the server.
+ // Can only check the size if there was a gap
+ checkNumSlots(newSlots.length);
+
+ // Since there was a gap every machine must have pushed a slot or must have
+ // a last message message. If not then the server is hiding slots
+ if (!machineSet.isEmpty()) {
+ throw new Error("Missing record for machines: " + machineSet);
+ }
+ }
+
+ // Update the size of our local block chain.
+ commitNewMaxSize();
+
+ // Commit new to slots to the local block chain.
+ for (Slot slot : newSlots) {
+
+ // Insert this slot into our local block chain copy.
+ buffer.putSlot(slot);
+
+ // Keep track of how many slots are currently live (have live data in them).
+ liveSlotCount++;
+ }
+
+ // Get the sequence number of the latest slot in the system
+ sequenceNumber = newSlots[newSlots.length - 1].getSequenceNumber();
+
+ updateLiveStateFromServer();
+
+ // No Need to remember after we pulled from the server
+ offlineTransactionsCommittedAndAtServer.clear();
+
+ // This is invalidated now
+ hadPartialSendToServer = false;
+ }
+
+ private void updateLiveStateFromServer() {
+ // Process the new transaction parts
+ processNewTransactionParts();
+
+ // Do arbitration on new transactions that were received
+ arbitrateFromServer();
+
+ // Update all the committed keys
+ boolean didCommitOrSpeculate = updateCommittedTable();
+
+ // Delete the transactions that are now dead
+ updateLiveTransactionsAndStatus();
+
+ // Do speculations
+ didCommitOrSpeculate |= updateSpeculativeTable(didCommitOrSpeculate);
+ updatePendingTransactionSpeculativeTable(didCommitOrSpeculate);
+ }
+
+ private void updateLiveStateFromLocal() {
+ // Update all the committed keys
+ boolean didCommitOrSpeculate = updateCommittedTable();
+
+ // Delete the transactions that are now dead
+ updateLiveTransactionsAndStatus();
+
+ // Do speculations
+ didCommitOrSpeculate |= updateSpeculativeTable(didCommitOrSpeculate);
+ updatePendingTransactionSpeculativeTable(didCommitOrSpeculate);
+ }
+
+ private void initExpectedSize(long firstSequenceNumber, long numberOfSlots) {
+
+ long prevslots = firstSequenceNumber;
+ if (didFindTableStatus) {
+// expectedsize = (prevslots < ((long) numberOfSlots)) ? (int) prevslots : expectedsize;
+ } else {
+ expectedsize = (prevslots < ((long) numberOfSlots)) ? (int) prevslots : numberOfSlots;
+ }
+ didFindTableStatus = true;
+
+ currMaxSize = numberOfSlots;
+ }
+
+ private void updateExpectedSize() {
+ expectedsize++;
+ if (expectedsize > currMaxSize) {
+ expectedsize = currMaxSize;
+ }
+ }
+
+
+ /**
+ * Check the size of the block chain to make sure there are enough slots sent back by the server.
+ * This is only called when we have a gap between the slots that we have locally and the slots
+ * sent by the server therefore in the slots sent by the server there will be at least 1 Table
+ * status message
+ */
+ private void checkNumSlots(int numberOfSlots) {
+ if (numberOfSlots != expectedsize) {
+ throw new Error("Server Error: Server did not send all slots. Expected: " + expectedsize + " Received:" + numberOfSlots);
+ }
+ }
+
+ private void updateCurrMaxSize(int newmaxsize) {
+ currMaxSize = newmaxsize;
+ }
+
+
+ /**
+ * Update the size of of the local buffer if it is needed.
+ */
+ private void commitNewMaxSize() {
+ didFindTableStatus = false;
+
+ // Resize the local slot buffer
+ if (numberOfSlots != currMaxSize) {
+ buffer.resize((int)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();
+ }
+
+ /**
+ * Process the new transaction parts from this latest round of slots received from the server
+ */
+ private void processNewTransactionParts() {
+
+ if (newTransactionParts.size() == 0) {
+ // Nothing new to process
+ return;
+ }
+
+ // Iterate through all the machine Ids that we received new parts for
+ for (Long machineId : newTransactionParts.keySet()) {
+ Map<Pair<Long, Integer>, TransactionPart> parts = newTransactionParts.get(machineId);
+
+ // Iterate through all the parts for that machine Id
+ for (Pair<Long, Integer> partId : parts.keySet()) {
+ TransactionPart part = parts.get(partId);
+
+ Long lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable.get(part.getArbitratorId());
+ if ((lastTransactionNumber != null) && (lastTransactionNumber >= part.getSequenceNumber())) {
+ // Set dead the transaction part
+ part.setDead();
+ continue;
+ }
+
+ // Get the transaction object for that sequence number
+ Transaction transaction = liveTransactionBySequenceNumberTable.get(part.getSequenceNumber());
+
+ if (transaction == null) {
+ // This is a new transaction that we dont have so make a new one
+ transaction = new Transaction();
+
+ // Insert this new transaction into the live tables
+ liveTransactionBySequenceNumberTable.put(part.getSequenceNumber(), transaction);
+ liveTransactionByTransactionIdTable.put(part.getTransactionId(), transaction);
+ }
+
+ // Add that part to the transaction
+ transaction.addPartDecode(part);
+ }
+ }
+
+ // Clear all the new transaction parts in preparation for the next time the server sends slots
+ newTransactionParts.clear();
+ }
+
+
+ private long lastSeqNumArbOn = 0;
+
+ private void arbitrateFromServer() {
+
+ if (liveTransactionBySequenceNumberTable.size() == 0) {
+ // Nothing to arbitrate on so move on
+ return;
+ }
+
+ // Get the transaction sequence numbers and sort from oldest to newest
+ List<Long> transactionSequenceNumbers = new ArrayList<Long>(liveTransactionBySequenceNumberTable.keySet());
+ Collections.sort(transactionSequenceNumbers);
+
+ // Collection of key value pairs that are
+ Map<IoTString, KeyValue> speculativeTableTmp = new HashMap<IoTString, KeyValue>();
+
+ // The last transaction arbitrated on
+ long lastTransactionCommitted = -1;
+ Set<Abort> generatedAborts = new HashSet<Abort>();
+
+ for (Long transactionSequenceNumber : transactionSequenceNumbers) {
+ Transaction transaction = liveTransactionBySequenceNumberTable.get(transactionSequenceNumber);
+
+
+
+ // Check if this machine arbitrates for this transaction if not then we cant arbitrate this transaction
+ if (transaction.getArbitrator() != localMachineId) {
+ continue;
+ }
+
+ if (transactionSequenceNumber < lastSeqNumArbOn) {
+ continue;
+ }
+
+ if (offlineTransactionsCommittedAndAtServer.contains(transaction.getId())) {
+ // We have seen this already locally so dont commit again
+ continue;
+ }
+
+
+ if (!transaction.isComplete()) {
+ // Will arbitrate in incorrect order if we continue so just break
+ // Most likely this
+ break;
+ }
+
+
+ // update the largest transaction seen by arbitrator from server
+ if (lastTransactionSeenFromMachineFromServer.get(transaction.getMachineId()) == null) {
+ lastTransactionSeenFromMachineFromServer.put(transaction.getMachineId(), transaction.getClientLocalSequenceNumber());
+ } else {
+ Long lastTransactionSeenFromMachine = lastTransactionSeenFromMachineFromServer.get(transaction.getMachineId());
+ if (transaction.getClientLocalSequenceNumber() > lastTransactionSeenFromMachine) {
+ lastTransactionSeenFromMachineFromServer.put(transaction.getMachineId(), transaction.getClientLocalSequenceNumber());
+ }
+ }
+
+ if (transaction.evaluateGuard(committedKeyValueTable, speculativeTableTmp, null)) {
+ // Guard evaluated as true
+
+ // Update the local changes so we can make the commit
+ for (KeyValue kv : transaction.getKeyValueUpdateSet()) {
+ speculativeTableTmp.put(kv.getKey(), kv);
+ }
+
+ // Update what the last transaction committed was for use in batch commit
+ lastTransactionCommitted = transactionSequenceNumber;
+ } else {
+ // Guard evaluated was false so create abort
+
+ // Create the abort
+ Abort newAbort = new Abort(null,
+ transaction.getClientLocalSequenceNumber(),
+ transaction.getSequenceNumber(),
+ transaction.getMachineId(),
+ transaction.getArbitrator(),
+ localArbitrationSequenceNumber);
+ localArbitrationSequenceNumber++;
+
+ generatedAborts.add(newAbort);
+
+ // Insert the abort so we can process
+ processEntry(newAbort);
+ }
+
+ lastSeqNumArbOn = transactionSequenceNumber;
+
+ // liveTransactionBySequenceNumberTable.remove(transactionSequenceNumber);
+ }
+
+ Commit newCommit = null;
+
+ // If there is something to commit
+ if (speculativeTableTmp.size() != 0) {
+
+ // Create the commit and increment the commit sequence number
+ newCommit = new Commit(localArbitrationSequenceNumber, localMachineId, lastTransactionCommitted);
+ localArbitrationSequenceNumber++;
+
+ // Add all the new keys to the commit
+ for (KeyValue kv : speculativeTableTmp.values()) {
+ newCommit.addKV(kv);
+ }
+
+ // create the commit parts
+ newCommit.createCommitParts();
+
+ // Append all the commit parts to the end of the pending queue waiting for sending to the server
+
+ // Insert the commit so we can process it
+ for (CommitPart commitPart : newCommit.getParts().values()) {
+ processEntry(commitPart);
+ }
+ }
+
+ if ((newCommit != null) || (generatedAborts.size() > 0)) {
+ ArbitrationRound arbitrationRound = new ArbitrationRound(newCommit, generatedAborts);
+ pendingSendArbitrationRounds.add(arbitrationRound);
+
+ if (compactArbitrationData()) {
+ ArbitrationRound newArbitrationRound = pendingSendArbitrationRounds.get(pendingSendArbitrationRounds.size() - 1);
+ if (newArbitrationRound.getCommit() != null) {
+ for (CommitPart commitPart : newArbitrationRound.getCommit().getParts().values()) {
+ processEntry(commitPart);
+ }
+ }
+ }
+ }
+ }
+
+ private Pair<Boolean, Boolean> arbitrateOnLocalTransaction(Transaction transaction) {
+
+ // Check if this machine arbitrates for this transaction if not then we cant arbitrate this transaction
+ if (transaction.getArbitrator() != localMachineId) {
+ return new Pair<Boolean, Boolean>(false, false);
+ }
+
+ if (!transaction.isComplete()) {
+ // Will arbitrate in incorrect order if we continue so just break
+ // Most likely this
+ return new Pair<Boolean, Boolean>(false, false);
+ }
+
+ if (transaction.getMachineId() != localMachineId) {
+ // dont do this check for local transactions
+ if (lastTransactionSeenFromMachineFromServer.get(transaction.getMachineId()) != null) {
+ if (lastTransactionSeenFromMachineFromServer.get(transaction.getMachineId()) > transaction.getClientLocalSequenceNumber()) {
+ // We've have already seen this from the server
+ return new Pair<Boolean, Boolean>(false, false);
+ }
+ }
+ }
+
+ if (transaction.evaluateGuard(committedKeyValueTable, null, null)) {
+ // Guard evaluated as true
+
+ // Create the commit and increment the commit sequence number
+ Commit newCommit = new Commit(localArbitrationSequenceNumber, localMachineId, -1);
+ localArbitrationSequenceNumber++;
+
+ // Update the local changes so we can make the commit
+ for (KeyValue kv : transaction.getKeyValueUpdateSet()) {
+ newCommit.addKV(kv);
+ }
+
+ // create the commit parts
+ newCommit.createCommitParts();
+
+ // Append all the commit parts to the end of the pending queue waiting for sending to the server
+ ArbitrationRound arbitrationRound = new ArbitrationRound(newCommit, new HashSet<Abort>());
+ pendingSendArbitrationRounds.add(arbitrationRound);
+
+ if (compactArbitrationData()) {
+ ArbitrationRound newArbitrationRound = pendingSendArbitrationRounds.get(pendingSendArbitrationRounds.size() - 1);
+ for (CommitPart commitPart : newArbitrationRound.getCommit().getParts().values()) {
+ processEntry(commitPart);
+ }
+ } else {
+ // Insert the commit so we can process it
+ for (CommitPart commitPart : newCommit.getParts().values()) {
+ processEntry(commitPart);
+ }
+ }
+
+ if (transaction.getMachineId() == localMachineId) {
+ TransactionStatus status = transaction.getTransactionStatus();
+ if (status != null) {
+ status.setStatus(TransactionStatus.StatusCommitted);
+ }
+ }
+
+ updateLiveStateFromLocal();
+ return new Pair<Boolean, Boolean>(true, true);
+ } else {
+
+ if (transaction.getMachineId() == localMachineId) {
+ // For locally created messages update the status
+
+ // Guard evaluated was false so create abort
+ TransactionStatus status = transaction.getTransactionStatus();
+ if (status != null) {
+ status.setStatus(TransactionStatus.StatusAborted);
+ }
+ } else {
+ Set addAbortSet = new HashSet<Abort>();
+
+
+ // Create the abort
+ Abort newAbort = new Abort(null,
+ transaction.getClientLocalSequenceNumber(),
+ -1,
+ transaction.getMachineId(),
+ transaction.getArbitrator(),
+ localArbitrationSequenceNumber);
+ localArbitrationSequenceNumber++;
+
+ addAbortSet.add(newAbort);
+
+
+ // Append all the commit parts to the end of the pending queue waiting for sending to the server
+ ArbitrationRound arbitrationRound = new ArbitrationRound(null, addAbortSet);
+ pendingSendArbitrationRounds.add(arbitrationRound);
+
+ if (compactArbitrationData()) {
+ ArbitrationRound newArbitrationRound = pendingSendArbitrationRounds.get(pendingSendArbitrationRounds.size() - 1);
+ for (CommitPart commitPart : newArbitrationRound.getCommit().getParts().values()) {
+ processEntry(commitPart);
+ }
+ }
+ }
+
+ updateLiveStateFromLocal();
+ return new Pair<Boolean, Boolean>(true, false);
+ }
+ }
+
+ /**
+ * 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
+ */
+ private boolean compactArbitrationData() {
+
+ if (pendingSendArbitrationRounds.size() < 2) {
+ // Nothing to compact so do nothing
+ return false;
+ }
+
+ ArbitrationRound lastRound = pendingSendArbitrationRounds.get(pendingSendArbitrationRounds.size() - 1);
+ if (lastRound.didSendPart()) {
+ return false;
+ }
+
+ boolean hadCommit = (lastRound.getCommit() == null);
+ boolean gotNewCommit = false;
+
+ int numberToDelete = 1;
+ while (numberToDelete < pendingSendArbitrationRounds.size()) {
+ ArbitrationRound round = pendingSendArbitrationRounds.get(pendingSendArbitrationRounds.size() - numberToDelete - 1);
+
+ if (round.isFull() || round.didSendPart()) {
+ // Stop since there is a part that cannot be compacted and we need to compact in order
+ break;
+ }
+
+ if (round.getCommit() == null) {
+
+ // Try compacting aborts only
+ int newSize = round.getCurrentSize() + lastRound.getAbortsCount();
+ if (newSize > ArbitrationRound.MAX_PARTS) {
+ // Cant compact since it would be too large
+ break;
+ }
+ lastRound.addAborts(round.getAborts());
+ } else {
+
+ // Create a new larger commit
+ Commit newCommit = Commit.merge(lastRound.getCommit(), round.getCommit(), localArbitrationSequenceNumber);
+ localArbitrationSequenceNumber++;
+
+ // Create the commit parts so that we can count them
+ newCommit.createCommitParts();
+
+ // Calculate the new size of the parts
+ int newSize = newCommit.getNumberOfParts();
+ newSize += lastRound.getAbortsCount();
+ newSize += round.getAbortsCount();
+
+ if (newSize > ArbitrationRound.MAX_PARTS) {
+ // Cant compact since it would be too large
+ break;
+ }
+
+ // Set the new compacted part
+ lastRound.setCommit(newCommit);
+ lastRound.addAborts(round.getAborts());
+ gotNewCommit = true;
+ }
+
+ numberToDelete++;
+ }
+
+ if (numberToDelete != 1) {
+ // If there is a compaction
+
+ // Delete the previous pieces that are now in the new compacted piece
+ if (numberToDelete == pendingSendArbitrationRounds.size()) {
+ pendingSendArbitrationRounds.clear();
+ } else {
+ for (int i = 0; i < numberToDelete; i++) {
+ pendingSendArbitrationRounds.remove(pendingSendArbitrationRounds.size() - 1);
+ }
+ }
+
+ // Add the new compacted into the pending to send list
+ pendingSendArbitrationRounds.add(lastRound);
+
+ // Should reinsert into the commit processor
+ if (hadCommit && gotNewCommit) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+ // private boolean compactArbitrationData() {
+ // return false;
+ // }
+
+ /**
+ * Update all the commits and the committed tables, sets dead the dead transactions
+ */
+ private boolean updateCommittedTable() {
+
+ if (newCommitParts.size() == 0) {
+ // Nothing new to process
+ return false;
+ }
+
+ // Iterate through all the machine Ids that we received new parts for
+ for (Long machineId : newCommitParts.keySet()) {
+ Map<Pair<Long, Integer>, CommitPart> parts = newCommitParts.get(machineId);
+
+ // Iterate through all the parts for that machine Id
+ for (Pair<Long, Integer> partId : parts.keySet()) {
+ CommitPart part = parts.get(partId);
+
+ // Get the transaction object for that sequence number
+ Map<Long, Commit> commitForClientTable = liveCommitsTable.get(part.getMachineId());
+
+ if (commitForClientTable == null) {
+ // This is the first commit from this device
+ commitForClientTable = new HashMap<Long, Commit>();
+ liveCommitsTable.put(part.getMachineId(), commitForClientTable);
+ }
+
+ Commit commit = commitForClientTable.get(part.getSequenceNumber());
+
+ if (commit == null) {
+ // This is a new commit that we dont have so make a new one
+ commit = new Commit();
+
+ // Insert this new commit into the live tables
+ commitForClientTable.put(part.getSequenceNumber(), commit);
+ }
+
+ // Add that part to the commit
+ commit.addPartDecode(part);
+ }
+ }
+
+ // Clear all the new commits parts in preparation for the next time the server sends slots
+ newCommitParts.clear();
+
+ // If we process a new commit keep track of it for future use
+ boolean didProcessANewCommit = false;
+
+ // Process the commits one by one
+ for (Long arbitratorId : liveCommitsTable.keySet()) {
+
+ // Get all the commits for a specific arbitrator
+ Map<Long, Commit> commitForClientTable = liveCommitsTable.get(arbitratorId);
+
+ // Sort the commits in order
+ List<Long> commitSequenceNumbers = new ArrayList<Long>(commitForClientTable.keySet());
+ Collections.sort(commitSequenceNumbers);
+
+ // Get the last commit seen from this arbitrator
+ long lastCommitSeenSequenceNumber = -1;
+ if (lastCommitSeenSequenceNumberByArbitratorTable.get(arbitratorId) != null) {
+ lastCommitSeenSequenceNumber = lastCommitSeenSequenceNumberByArbitratorTable.get(arbitratorId);
+ }
+
+ // Go through each new commit one by one
+ for (int i = 0; i < commitSequenceNumbers.size(); i++) {
+ Long commitSequenceNumber = commitSequenceNumbers.get(i);
+ Commit commit = commitForClientTable.get(commitSequenceNumber);
+
+ // Special processing if a commit is not complete
+ if (!commit.isComplete()) {
+ if (i == (commitSequenceNumbers.size() - 1)) {
+ // If there is an incomplete commit and this commit is the latest one seen then this commit cannot be processed and there are no other commits
+ break;
+ } else {
+ // This is a commit that was already dead but parts of it are still in the block chain (not flushed out yet).
+ // Delete it and move on
+ commit.setDead();
+ commitForClientTable.remove(commit.getSequenceNumber());
+ continue;
+ }
+ }
+
+ // Update the last transaction that was updated if we can
+ if (commit.getTransactionSequenceNumber() != -1) {
+ Long lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable.get(commit.getMachineId());
+
+ // Update the last transaction sequence number that the arbitrator arbitrated on
+ if ((lastTransactionNumber == null) || (lastTransactionNumber < commit.getTransactionSequenceNumber())) {
+ lastArbitratedTransactionNumberByArbitratorTable.put(commit.getMachineId(), commit.getTransactionSequenceNumber());
+ }
+ }
+
+ // Update the last arbitration data that we have seen so far
+ if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.get(commit.getMachineId()) != null) {
+
+ long lastArbitrationSequenceNumber = lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.get(commit.getMachineId());
+ if (commit.getSequenceNumber() > lastArbitrationSequenceNumber) {
+ // Is larger
+ lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.put(commit.getMachineId(), commit.getSequenceNumber());
+ }
+ } else {
+ // Never seen any data from this arbitrator so record the first one
+ lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.put(commit.getMachineId(), commit.getSequenceNumber());
+ }
+
+ // We have already seen this commit before so need to do the full processing on this commit
+ if (commit.getSequenceNumber() <= lastCommitSeenSequenceNumber) {
+
+ // Update the last transaction that was updated if we can
+ if (commit.getTransactionSequenceNumber() != -1) {
+ Long lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable.get(commit.getMachineId());
+
+ // Update the last transaction sequence number that the arbitrator arbitrated on
+ if ((lastTransactionNumber == null) || (lastTransactionNumber < commit.getTransactionSequenceNumber())) {
+ lastArbitratedTransactionNumberByArbitratorTable.put(commit.getMachineId(), commit.getTransactionSequenceNumber());
+ }
+ }
+
+ continue;
+ }
+
+ // If we got here then this is a brand new commit and needs full processing
+
+ // Get what commits should be edited, these are the commits that have live values for their keys
+ Set<Commit> commitsToEdit = new HashSet<Commit>();
+ for (KeyValue kv : commit.getKeyValueUpdateSet()) {
+ commitsToEdit.add(liveCommitsByKeyTable.get(kv.getKey()));
+ }
+ commitsToEdit.remove(null); // remove null since it could be in this set
+
+ // Update each previous commit that needs to be updated
+ for (Commit previousCommit : commitsToEdit) {
+
+ // Only bother with live commits (TODO: Maybe remove this check)
+ if (previousCommit.isLive()) {
+
+ // Update which keys in the old commits are still live
+ for (KeyValue kv : commit.getKeyValueUpdateSet()) {
+ previousCommit.invalidateKey(kv.getKey());
+ }
+
+ // if the commit is now dead then remove it
+ if (!previousCommit.isLive()) {
+ commitForClientTable.remove(previousCommit);
+ }
+ }
+ }
+
+ // Update the last seen sequence number from this arbitrator
+ if (lastCommitSeenSequenceNumberByArbitratorTable.get(commit.getMachineId()) != null) {
+ if (commit.getSequenceNumber() > lastCommitSeenSequenceNumberByArbitratorTable.get(commit.getMachineId())) {
+ lastCommitSeenSequenceNumberByArbitratorTable.put(commit.getMachineId(), commit.getSequenceNumber());
+ }
+ } else {
+ lastCommitSeenSequenceNumberByArbitratorTable.put(commit.getMachineId(), commit.getSequenceNumber());
+ }
+
+ // We processed a new commit that we havent seen before
+ didProcessANewCommit = true;
+
+ // Update the committed table of keys and which commit is using which key
+ for (KeyValue kv : commit.getKeyValueUpdateSet()) {
+ committedKeyValueTable.put(kv.getKey(), kv);
+ liveCommitsByKeyTable.put(kv.getKey(), commit);
+ }
+ }
+ }
+
+ return didProcessANewCommit;
+ }
+
+ /**
+ * Create the speculative table from transactions that are still live and have come from the cloud
+ */
+ private boolean updateSpeculativeTable(boolean didProcessNewCommits) {
+ if (liveTransactionBySequenceNumberTable.keySet().size() == 0) {
+ // There is nothing to speculate on
+ return false;
+ }
+
+ // Create a list of the transaction sequence numbers and sort them from oldest to newest
+ List<Long> transactionSequenceNumbersSorted = new ArrayList<Long>(liveTransactionBySequenceNumberTable.keySet());
+ Collections.sort(transactionSequenceNumbersSorted);
+
+ boolean hasGapInTransactionSequenceNumbers = transactionSequenceNumbersSorted.get(0) != oldestTransactionSequenceNumberSpeculatedOn;
+
+
+ if (hasGapInTransactionSequenceNumbers || didProcessNewCommits) {
+ // If there is a gap in the transaction sequence numbers then there was a commit or an abort of a transaction
+ // OR there was a new commit (Could be from offline commit) so a redo the speculation from scratch
+
+ // Start from scratch
+ speculatedKeyValueTable.clear();
+ lastTransactionSequenceNumberSpeculatedOn = -1;
+ oldestTransactionSequenceNumberSpeculatedOn = -1;
+
+ }
+
+ // Remember the front of the transaction list
+ oldestTransactionSequenceNumberSpeculatedOn = transactionSequenceNumbersSorted.get(0);
+
+ // Find where to start arbitration from
+ int startIndex = transactionSequenceNumbersSorted.indexOf(lastTransactionSequenceNumberSpeculatedOn) + 1;
+
+ if (startIndex >= transactionSequenceNumbersSorted.size()) {
+ // Make sure we are not out of bounds
+ return false; // did not speculate
+ }
+
+ Set<Long> incompleteTransactionArbitrator = new HashSet<Long>();
+ boolean didSkip = true;
+
+ for (int i = startIndex; i < transactionSequenceNumbersSorted.size(); i++) {
+ long transactionSequenceNumber = transactionSequenceNumbersSorted.get(i);
+ Transaction transaction = liveTransactionBySequenceNumberTable.get(transactionSequenceNumber);
+
+ if (!transaction.isComplete()) {
+ // If there is an incomplete transaction then there is nothing we can do
+ // add this transactions arbitrator to the list of arbitrators we should ignore
+ incompleteTransactionArbitrator.add(transaction.getArbitrator());
+ didSkip = true;
+ continue;
+ }
+
+ if (incompleteTransactionArbitrator.contains(transaction.getArbitrator())) {
+ continue;
+ }
+
+ lastTransactionSequenceNumberSpeculatedOn = transactionSequenceNumber;
+
+ if (transaction.evaluateGuard(committedKeyValueTable, speculatedKeyValueTable, null)) {
+ // Guard evaluated to true so update the speculative table
+ for (KeyValue kv : transaction.getKeyValueUpdateSet()) {
+ speculatedKeyValueTable.put(kv.getKey(), kv);
+ }
+ }
+ }
+
+ if (didSkip) {
+ // Since there was a skip we need to redo the speculation next time around
+ lastTransactionSequenceNumberSpeculatedOn = -1;
+ oldestTransactionSequenceNumberSpeculatedOn = -1;
+ }
+
+ // We did some speculation
+ return true;
+ }
+
+ /**
+ * Create the pending transaction speculative table from transactions that are still in the pending transaction buffer
+ */
+ private void updatePendingTransactionSpeculativeTable(boolean didProcessNewCommitsOrSpeculate) {
+ if (pendingTransactionQueue.size() == 0) {
+ // There is nothing to speculate on
+ return;
+ }
+
+
+ if (didProcessNewCommitsOrSpeculate || (firstPendingTransaction != pendingTransactionQueue.get(0))) {
+ // need to reset on the pending speculation
+ lastPendingTransactionSpeculatedOn = null;
+ firstPendingTransaction = pendingTransactionQueue.get(0);
+ pendingTransactionSpeculatedKeyValueTable.clear();
+ }
+
+ // Find where to start arbitration from
+ int startIndex = pendingTransactionQueue.indexOf(firstPendingTransaction) + 1;
+
+ if (startIndex >= pendingTransactionQueue.size()) {
+ // Make sure we are not out of bounds
+ return;
+ }
+
+ for (int i = startIndex; i < pendingTransactionQueue.size(); i++) {
+ Transaction transaction = pendingTransactionQueue.get(i);
+
+ lastPendingTransactionSpeculatedOn = transaction;
+
+ if (transaction.evaluateGuard(committedKeyValueTable, speculatedKeyValueTable, pendingTransactionSpeculatedKeyValueTable)) {
+ // Guard evaluated to true so update the speculative table
+ for (KeyValue kv : transaction.getKeyValueUpdateSet()) {
+ pendingTransactionSpeculatedKeyValueTable.put(kv.getKey(), kv);
+ }
+ }
+ }
+ }
+
+ /**
+ * Set dead and remove from the live transaction tables the transactions that are dead
+ */
+ private void updateLiveTransactionsAndStatus() {
+
+ // Go through each of the transactions
+ for (Iterator<Map.Entry<Long, Transaction>> iter = liveTransactionBySequenceNumberTable.entrySet().iterator(); iter.hasNext();) {
+ Transaction transaction = iter.next().getValue();
+
+ // Check if the transaction is dead
+ Long lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable.get(transaction.getArbitrator());
+ if ((lastTransactionNumber != null) && (lastTransactionNumber >= transaction.getSequenceNumber())) {
+
+ // Set dead the transaction
+ transaction.setDead();
+
+ // Remove the transaction from the live table
+ iter.remove();
+ liveTransactionByTransactionIdTable.remove(transaction.getId());
+ }
+ }
+
+ // Go through each of the transactions
+ for (Iterator<Map.Entry<Long, TransactionStatus>> iter = outstandingTransactionStatus.entrySet().iterator(); iter.hasNext();) {
+ TransactionStatus status = iter.next().getValue();
+
+ // Check if the transaction is dead
+ Long lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable.get(status.getTransactionArbitrator());
+ if ((lastTransactionNumber != null) && (lastTransactionNumber >= status.getTransactionSequenceNumber())) {
+
+ // Set committed
+ status.setStatus(TransactionStatus.StatusCommitted);
+
+ // Remove
+ iter.remove();
+ }
+ }
+ }
+
+ /**
+ * Process this slot, entry by entry. Also update the latest message sent by slot
+ */
+ private void processSlot(SlotIndexer indexer, Slot slot, boolean acceptUpdatesToLocal, HashSet<Long> machineSet) {
+
+ // Update the last message seen
+ updateLastMessage(slot.getMachineID(), slot.getSequenceNumber(), slot, acceptUpdatesToLocal, machineSet);
+
+ // Process each entry in the slot
+ for (Entry entry : slot.getEntries()) {
+ switch (entry.getType()) {
+
+ case Entry.TypeCommitPart:
+ processEntry((CommitPart)entry);
+ break;
+
+ case Entry.TypeAbort:
+ processEntry((Abort)entry);
+ break;
+
+ case Entry.TypeTransactionPart:
+ processEntry((TransactionPart)entry);
+ break;
+
+ case Entry.TypeNewKey:
+ processEntry((NewKey)entry);
+ break;
+
+ case Entry.TypeLastMessage:
+ processEntry((LastMessage)entry, machineSet);
+ break;
+
+ case Entry.TypeRejectedMessage:
+ processEntry((RejectedMessage)entry, indexer);
+ break;
+
+ case Entry.TypeTableStatus:
+ processEntry((TableStatus)entry, slot.getSequenceNumber());
+ break;
+
+ default:
+ throw new Error("Unrecognized type: " + entry.getType());
+ }
+ }
+ }
+
+ /**
+ * Update the last message that was sent for a machine Id
+ */
+ private void processEntry(LastMessage entry, HashSet<Long> machineSet) {
+ // Update what the last message received by a machine was
+ updateLastMessage(entry.getMachineID(), entry.getSequenceNumber(), entry, false, machineSet);
+ }
+
+ /**
+ * Add the new key to the arbitrators table and update the set of live new keys (in case of a rescued new key message)
+ */
+ private void processEntry(NewKey entry) {
+
+ // Update the arbitrator table with the new key information
+ arbitratorTable.put(entry.getKey(), entry.getMachineID());
+
+ // Update what the latest live new key is
+ NewKey oldNewKey = liveNewKeyTable.put(entry.getKey(), entry);
+ if (oldNewKey != null) {
+ // Delete the old new key messages
+ oldNewKey.setDead();
+ }
+ }
+
+ /**
+ * Process new table status entries and set dead the old ones as new ones come in.
+ * keeps track of the largest and smallest table status seen in this current round
+ * of updating the local copy of the block chain
+ */
+ private void processEntry(TableStatus entry, long seq) {
+ int newNumSlots = entry.getMaxSlots();
+ updateCurrMaxSize(newNumSlots);
+
+ initExpectedSize(seq, newNumSlots);
+
+ if (liveTableStatus != null) {
+ // We have a larger table status so the old table status is no longer alive
+ liveTableStatus.setDead();
+ }
+
+ // Make this new table status the latest alive table status
+ liveTableStatus = entry;
+ }
+
+ /**
+ * Check old messages to see if there is a block chain violation. Also
+ */
+ private void processEntry(RejectedMessage entry, SlotIndexer indexer) {
+ long oldSeqNum = entry.getOldSeqNum();
+ long newSeqNum = entry.getNewSeqNum();
+ boolean isequal = entry.getEqual();
+ long machineId = entry.getMachineID();
+ long seq = entry.getSequenceNumber();
+
+
+ // Check if we have messages that were supposed to be rejected in our local block chain
+ for (long seqNum = oldSeqNum; seqNum <= newSeqNum; seqNum++) {
+
+ // Get the slot
+ Slot slot = indexer.getSlot(seqNum);
+
+ if (slot != null) {
+ // If we have this slot make sure that it was not supposed to be a rejected slot
+
+ long slotMachineId = slot.getMachineID();
+ if (isequal != (slotMachineId == machineId)) {
+ throw new Error("Server Error: Trying to insert rejected message for slot " + seqNum);
+ }
+ }
+ }
+
+
+ // Create a list of clients to watch until they see this rejected message entry.
+ HashSet<Long> deviceWatchSet = new HashSet<Long>();
+ for (Map.Entry<Long, Pair<Long, Liveness>> lastMessageEntry : lastMessageTable.entrySet()) {
+
+ // Machine ID for the last message entry
+ long lastMessageEntryMachineId = lastMessageEntry.getKey();
+
+ // We've seen it, don't need to continue to watch. Our next
+ // message will implicitly acknowledge it.
+ if (lastMessageEntryMachineId == localMachineId) {
+ continue;
+ }
+
+ Pair<Long, Liveness> lastMessageValue = lastMessageEntry.getValue();
+ long entrySequenceNumber = lastMessageValue.getFirst();
+
+ if (entrySequenceNumber < seq) {
+
+ // Add this rejected message to the set of messages that this machine ID did not see yet
+ addWatchList(lastMessageEntryMachineId, entry);
+
+ // This client did not see this rejected message yet so add it to the watch set to monitor
+ deviceWatchSet.add(lastMessageEntryMachineId);
+ }
+ }
+
+ if (deviceWatchSet.isEmpty()) {
+ // This rejected message has been seen by all the clients so
+ entry.setDead();
+ } else {
+ // We need to watch this rejected message
+ entry.setWatchSet(deviceWatchSet);
+ }
+ }
+
+ /**
+ * Check if this abort is live, if not then save it so we can kill it later.
+ * update the last transaction number that was arbitrated on.
+ */
+ private void processEntry(Abort entry) {
+
+
+ if (entry.getTransactionSequenceNumber() != -1) {
+ // update the transaction status if it was sent to the server
+ TransactionStatus status = outstandingTransactionStatus.remove(entry.getTransactionSequenceNumber());
+ if (status != null) {
+ status.setStatus(TransactionStatus.StatusAborted);
+ }
+ }
+
+ // Abort has not been seen by the client it is for yet so we need to keep track of it
+ Abort previouslySeenAbort = liveAbortTable.put(entry.getAbortId(), entry);
+ if (previouslySeenAbort != null) {
+ previouslySeenAbort.setDead(); // Delete old version of the abort since we got a rescued newer version
+ }
+
+ if (entry.getTransactionArbitrator() == localMachineId) {
+ liveAbortsGeneratedByLocal.put(entry.getArbitratorLocalSequenceNumber(), entry);
+ }
+
+ if ((entry.getSequenceNumber() != -1) && (lastMessageTable.get(entry.getTransactionMachineId()).getFirst() >= entry.getSequenceNumber())) {
+
+ // The machine already saw this so it is dead
+ entry.setDead();
+ liveAbortTable.remove(entry.getAbortId());
+
+ if (entry.getTransactionArbitrator() == localMachineId) {
+ liveAbortsGeneratedByLocal.remove(entry.getArbitratorLocalSequenceNumber());
+ }
+
+ return;
+ }
+
+
+
+
+ // Update the last arbitration data that we have seen so far
+ if (lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.get(entry.getTransactionArbitrator()) != null) {
+
+ long lastArbitrationSequenceNumber = lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.get(entry.getTransactionArbitrator());
+ if (entry.getSequenceNumber() > lastArbitrationSequenceNumber) {
+ // Is larger
+ lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.put(entry.getTransactionArbitrator(), entry.getSequenceNumber());
+ }
+ } else {
+ // Never seen any data from this arbitrator so record the first one
+ lastArbitrationDataLocalSequenceNumberSeenFromArbitrator.put(entry.getTransactionArbitrator(), entry.getSequenceNumber());
+ }
+
+
+ // Set dead a transaction if we can
+ Transaction transactionToSetDead = liveTransactionByTransactionIdTable.remove(new Pair<Long, Long>(entry.getTransactionMachineId(), entry.getTransactionClientLocalSequenceNumber()));
+ if (transactionToSetDead != null) {
+ liveTransactionBySequenceNumberTable.remove(transactionToSetDead.getSequenceNumber());
+ }
+
+ // Update the last transaction sequence number that the arbitrator arbitrated on
+ Long lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable.get(entry.getTransactionArbitrator());
+ if ((lastTransactionNumber == null) || (lastTransactionNumber < entry.getTransactionSequenceNumber())) {
+
+ // Is a valid one
+ if (entry.getTransactionSequenceNumber() != -1) {
+ lastArbitratedTransactionNumberByArbitratorTable.put(entry.getTransactionArbitrator(), entry.getTransactionSequenceNumber());
+ }
+ }
+ }
+
+ /**
+ * Set dead the transaction part if that transaction is dead and keep track of all new parts
+ */
+ private void processEntry(TransactionPart entry) {
+ // Check if we have already seen this transaction and set it dead OR if it is not alive
+ Long lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable.get(entry.getArbitratorId());
+ if ((lastTransactionNumber != null) && (lastTransactionNumber >= entry.getSequenceNumber())) {
+ // This transaction is dead, it was already committed or aborted
+ entry.setDead();
+ return;
+ }
+
+ // This part is still alive
+ Map<Pair<Long, Integer>, TransactionPart> transactionPart = newTransactionParts.get(entry.getMachineId());
+
+ if (transactionPart == null) {
+ // Dont have a table for this machine Id yet so make one
+ transactionPart = new HashMap<Pair<Long, Integer>, TransactionPart>();
+ newTransactionParts.put(entry.getMachineId(), transactionPart);
+ }
+
+ // Update the part and set dead ones we have already seen (got a rescued version)
+ TransactionPart previouslySeenPart = transactionPart.put(entry.getPartId(), entry);
+ if (previouslySeenPart != null) {
+ previouslySeenPart.setDead();
+ }
+ }
+
+ /**
+ * Process new commit entries and save them for future use. Delete duplicates
+ */
+ private void processEntry(CommitPart entry) {
+
+
+ // Update the last transaction that was updated if we can
+ if (entry.getTransactionSequenceNumber() != -1) {
+ Long lastTransactionNumber = lastArbitratedTransactionNumberByArbitratorTable.get(entry.getMachineId());
+
+ // Update the last transaction sequence number that the arbitrator arbitrated on
+ if ((lastTransactionNumber == null) || (lastTransactionNumber < entry.getTransactionSequenceNumber())) {
+ lastArbitratedTransactionNumberByArbitratorTable.put(entry.getMachineId(), entry.getTransactionSequenceNumber());
+ }
+ }
+
+
+
+
+ Map<Pair<Long, Integer>, CommitPart> commitPart = newCommitParts.get(entry.getMachineId());
+
+ if (commitPart == null) {
+ // Don't have a table for this machine Id yet so make one
+ commitPart = new HashMap<Pair<Long, Integer>, CommitPart>();
+ newCommitParts.put(entry.getMachineId(), commitPart);
+ }
+
+ // Update the part and set dead ones we have already seen (got a rescued version)
+ CommitPart previouslySeenPart = commitPart.put(entry.getPartId(), entry);
+ if (previouslySeenPart != null) {
+ previouslySeenPart.setDead();
+ }
+ }
+
+ /**
+ * Update the last message seen table. Update and set dead the appropriate RejectedMessages as clients see them.
+ * Updates the live aborts, removes those that are dead and sets them dead.
+ * Check that the last message seen is correct and that there is no mismatch of our own last message or that
+ * other clients have not had a rollback on the last message.
+ */
+ private void updateLastMessage(long machineId, long seqNum, Liveness liveness, boolean acceptUpdatesToLocal, HashSet<Long> machineSet) {
+
+ // We have seen this machine ID
+ machineSet.remove(machineId);
+
+ // Get the set of rejected messages that this machine Id is has not seen yet
+ HashSet<RejectedMessage> watchset = rejectedMessageWatchListTable.get(machineId);
+
+ // If there is a rejected message that this machine Id has not seen yet
+ if (watchset != null) {
+
+ // Go through each rejected message that this machine Id has not seen yet
+ for (Iterator<RejectedMessage> rmit = watchset.iterator(); rmit.hasNext(); ) {
+
+ RejectedMessage rm = rmit.next();
+
+ // If this machine Id has seen this rejected message...
+ if (rm.getSequenceNumber() <= seqNum) {
+
+ // Remove it from our watchlist
+ rmit.remove();
+
+ // Decrement machines that need to see this notification
+ rm.removeWatcher(machineId);
+ }
+ }
+ }
+
+ // Set dead the abort
+ for (Iterator<Map.Entry<Pair<Long, Long>, Abort>> i = liveAbortTable.entrySet().iterator(); i.hasNext();) {
+ Abort abort = i.next().getValue();
+
+ if ((abort.getTransactionMachineId() == machineId) && (abort.getSequenceNumber() <= seqNum)) {
+ abort.setDead();
+ i.remove();
+
+ if (abort.getTransactionArbitrator() == localMachineId) {
+ liveAbortsGeneratedByLocal.remove(abort.getArbitratorLocalSequenceNumber());
+ }
+ }
+ }
+
+
+
+ if (machineId == localMachineId) {
+ // Our own messages are immediately dead.
+ if (liveness instanceof LastMessage) {
+ ((LastMessage)liveness).setDead();
+ } else if (liveness instanceof Slot) {
+ ((Slot)liveness).setDead();
+ } else {
+ throw new Error("Unrecognized type");
+ }
+ }
+
+ // Get the old last message for this device
+ Pair<Long, Liveness> lastMessageEntry = lastMessageTable.put(machineId, new Pair<Long, Liveness>(seqNum, liveness));
+ if (lastMessageEntry == null) {
+ // If no last message then there is nothing else to process
+ return;
+ }
+
+ long lastMessageSeqNum = lastMessageEntry.getFirst();
+ Liveness lastEntry = lastMessageEntry.getSecond();
+
+ // If it is not our machine Id since we already set ours to dead
+ if (machineId != localMachineId) {
+ if (lastEntry instanceof LastMessage) {
+ ((LastMessage)lastEntry).setDead();
+ } else if (lastEntry instanceof Slot) {
+ ((Slot)lastEntry).setDead();
+ } else {
+ throw new Error("Unrecognized type");
+ }
+ }
+
+ // Make sure the server is not playing any games
+ if (machineId == localMachineId) {
+
+ if (hadPartialSendToServer) {
+ // We were not making any updates and we had a machine mismatch
+ if (lastMessageSeqNum > seqNum && !acceptUpdatesToLocal) {
+ throw new Error("Server Error: Mismatch on local machine sequence number, needed at least: " + lastMessageSeqNum + " got: " + seqNum);
+ }
+
+ } else {
+ // We were not making any updates and we had a machine mismatch
+ if (lastMessageSeqNum != seqNum && !acceptUpdatesToLocal) {
+ throw new Error("Server Error: Mismatch on local machine sequence number, needed: " + lastMessageSeqNum + " got: " + seqNum);
+ }
+ }
+ } else {
+ if (lastMessageSeqNum > seqNum) {
+ throw new Error("Server Error: Rollback on remote machine sequence number");
+ }
+ }
+ }
+
+ /**
+ * Add a rejected message entry to the watch set to keep track of which clients have seen that
+ * rejected message entry and which have not.
+ */
+ private void addWatchList(long machineId, RejectedMessage entry) {
+ HashSet<RejectedMessage> entries = rejectedMessageWatchListTable.get(machineId);
+ if (entries == null) {
+ // There is no set for this machine ID yet so create one
+ entries = new HashSet<RejectedMessage>();
+ rejectedMessageWatchListTable.put(machineId, entries);
+ }
+ entries.add(entry);
+ }
+
+ /**
+ * Check if the HMAC chain is not violated
+ */
+ private void checkHMACChain(SlotIndexer indexer, Slot[] newSlots) {
+ for (int i = 0; i < newSlots.length; i++) {
+ Slot currSlot = newSlots[i];
+ Slot prevSlot = indexer.getSlot(currSlot.getSequenceNumber() - 1);
+ if (prevSlot != null &&
+ !Arrays.equals(prevSlot.getHMAC(), currSlot.getPrevHMAC()))
+ throw new Error("Server Error: Invalid HMAC Chain" + currSlot + " " + prevSlot);
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+package iotcloud;
+import java.nio.ByteBuffer;
+
+/**
+ * TableStatus entries record the current size of the data structure
+ * in slots. Used to remember the size and to perform resizes.
+ * @author Brian Demsky
+ * @version 1.0
+ */
+
+
+class TableStatus extends Entry {
+ private int maxslots;
+
+ TableStatus(Slot slot, int _maxslots) {
+ super(slot);
+ maxslots=_maxslots;
+ }
+
+ int getMaxSlots() {
+ return maxslots;
+ }
+
+ static Entry decode(Slot slot, ByteBuffer bb) {
+ int maxslots=bb.getInt();
+ return new TableStatus(slot, maxslots);
+ }
+
+ void encode(ByteBuffer bb) {
+ bb.put(Entry.TypeTableStatus);
+ bb.putInt(maxslots);
+ }
+
+ int getSize() {
+ return Integer.BYTES+Byte.BYTES;
+ }
+
+ byte getType() {
+ return Entry.TypeTableStatus;
+ }
+
+ Entry getCopy(Slot s) {
+ return new TableStatus(s, maxslots);
+ }
+}
--- /dev/null
+package iotcloud;
+
+class ThreeTuple<A, B, C> {
+ private A a;
+ private B b;
+ private C c;
+
+ ThreeTuple(A a, B b, C c) {
+ this.a = a;
+ this.b = b;
+ this.c = c;
+ }
+
+ A getFirst() {
+ return a;
+ }
+
+ B getSecond() {
+ return b;
+ }
+
+ C getThird() {
+ return c;
+ }
+
+ public String toString() {
+ return "<" + a + "," + b + "," + c + ">";
+ }
+}
--- /dev/null
+package iotcloud;
+
+
+class TimingSingleton {
+ private static TimingSingleton singleton = new TimingSingleton( );
+ private static long startTime = 0;
+
+ private static long totalTime = 0;
+
+ private TimingSingleton() {
+
+ }
+
+ public static TimingSingleton getInstance( ) {
+ return singleton;
+ }
+
+
+ public static void startTime( ) {
+ startTime = System.nanoTime();
+ }
+
+ public static void endTime( ) {
+ totalTime += System.nanoTime() - startTime;
+
+ }
+
+ public static long getTime( ) {
+ return totalTime;
+ }
+
+
+}
\ No newline at end of file
--- /dev/null
+package iotcloud;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.nio.ByteBuffer;
+
+class Transaction {
+
+ private Map<Integer, TransactionPart> parts = null;
+ private Set<Integer> missingParts = null;
+ private List<Integer> partsPendingSend = null;
+ private boolean isComplete = false;
+ private boolean hasLastPart = false;
+ private Set<KeyValue> keyValueGuardSet = null;
+ private Set<KeyValue> keyValueUpdateSet = null;
+ private boolean isDead = false;
+ private long sequenceNumber = -1;
+ private long clientLocalSequenceNumber = -1;
+ private long arbitratorId = -1;
+ private long machineId = -1;
+ private Pair<Long, Long> transactionId = null;
+
+ private int nextPartToSend = 0;
+ private boolean didSendAPartToServer = false;
+
+ private TransactionStatus transactionStatus = null;
+
+ private boolean hadServerFailure = false;
+
+ public Transaction() {
+ parts = new HashMap<Integer, TransactionPart>();
+ keyValueGuardSet = new HashSet<KeyValue>();
+ keyValueUpdateSet = new HashSet<KeyValue>();
+ partsPendingSend = new ArrayList<Integer>();
+ }
+
+ public void addPartEncode(TransactionPart newPart) {
+ parts.put(newPart.getPartNumber(), newPart);
+ partsPendingSend.add(newPart.getPartNumber());
+
+ // Get the sequence number and other important information
+ sequenceNumber = newPart.getSequenceNumber();
+ arbitratorId = newPart.getArbitratorId();
+ transactionId = newPart.getTransactionId();
+ clientLocalSequenceNumber = newPart.getClientLocalSequenceNumber();
+ machineId = newPart.getMachineId();
+
+ isComplete = true;
+ }
+
+ public void addPartDecode(TransactionPart newPart) {
+
+ if (isDead) {
+ // If dead then just kill this part and move on
+ newPart.setDead();
+ return;
+ }
+
+ // Get the sequence number and other important information
+ sequenceNumber = newPart.getSequenceNumber();
+ arbitratorId = newPart.getArbitratorId();
+ transactionId = newPart.getTransactionId();
+ clientLocalSequenceNumber = newPart.getClientLocalSequenceNumber();
+ machineId = newPart.getMachineId();
+
+ TransactionPart previoslySeenPart = parts.put(newPart.getPartNumber(), newPart);
+
+ if (previoslySeenPart != null) {
+ // Set dead the old one since the new one is a rescued version of this part
+ previoslySeenPart.setDead();
+ } else if (newPart.isLastPart()) {
+ missingParts = new HashSet<Integer>();
+ hasLastPart = true;
+
+ for (int i = 0; i < newPart.getPartNumber(); i++) {
+ if (parts.get(i) == null) {
+ missingParts.add(i);
+ }
+ }
+ }
+
+ if (!isComplete && hasLastPart) {
+
+ // We have seen this part so remove it from the set of missing parts
+ missingParts.remove(newPart.getPartNumber());
+
+ // Check if all the parts have been seen
+ if (missingParts.size() == 0) {
+
+ // We have all the parts
+ isComplete = true;
+
+ // Decode all the parts and create the key value guard and update sets
+ decodeTransactionData();
+ }
+ }
+ }
+
+ public void addUpdateKV(KeyValue kv) {
+ keyValueUpdateSet.add(kv);
+ }
+
+ public void addGuardKV(KeyValue kv) {
+ keyValueGuardSet.add(kv);
+ }
+
+
+ public long getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ public void setSequenceNumber(long _sequenceNumber) {
+ sequenceNumber = _sequenceNumber;
+
+ for (Integer i : parts.keySet()) {
+ parts.get(i).setSequenceNumber(sequenceNumber);
+ }
+ }
+
+ public long getClientLocalSequenceNumber() {
+ return clientLocalSequenceNumber;
+ }
+
+ public Map<Integer, TransactionPart> getParts() {
+ return parts;
+ }
+
+ public boolean didSendAPartToServer() {
+ return didSendAPartToServer;
+ }
+
+ public void resetNextPartToSend() {
+ nextPartToSend = 0;
+ }
+
+ public TransactionPart getNextPartToSend() {
+ if ((partsPendingSend.size() == 0) || (partsPendingSend.size() == nextPartToSend)) {
+ return null;
+ }
+ TransactionPart part = parts.get(partsPendingSend.get(nextPartToSend));
+ nextPartToSend++;
+ return part;
+ }
+
+
+ public void setServerFailure() {
+ hadServerFailure = true;
+ }
+
+ public boolean getServerFailure() {
+ return hadServerFailure;
+ }
+
+
+ public void resetServerFailure() {
+ hadServerFailure = false;
+ }
+
+
+ public void setTransactionStatus(TransactionStatus _transactionStatus) {
+ transactionStatus = _transactionStatus;
+ }
+
+ public TransactionStatus getTransactionStatus() {
+ return transactionStatus;
+ }
+
+ public void removeSentParts(List<Integer> sentParts) {
+ nextPartToSend = 0;
+ if(partsPendingSend.removeAll(sentParts))
+ {
+ didSendAPartToServer = true;
+ transactionStatus.setTransactionSequenceNumber(sequenceNumber);
+ }
+ }
+
+ public boolean didSendAllParts() {
+ return partsPendingSend.isEmpty();
+ }
+
+ public Set<KeyValue> getKeyValueUpdateSet() {
+ return keyValueUpdateSet;
+ }
+
+ public int getNumberOfParts() {
+ return parts.size();
+ }
+
+ public long getMachineId() {
+ return machineId;
+ }
+
+ public long getArbitrator() {
+ return arbitratorId;
+ }
+
+ public boolean isComplete() {
+ return isComplete;
+ }
+
+ public Pair<Long, Long> getId() {
+ return transactionId;
+ }
+
+ public void setDead() {
+ if (isDead) {
+ // Already dead
+ return;
+ }
+
+ // Set dead
+ isDead = true;
+
+ // Make all the parts of this transaction dead
+ for (Integer partNumber : parts.keySet()) {
+ TransactionPart part = parts.get(partNumber);
+ part.setDead();
+ }
+ }
+
+ public TransactionPart getPart(int index) {
+ return parts.get(index);
+ }
+
+ private void decodeTransactionData() {
+
+ // Calculate the size of the data section
+ int dataSize = 0;
+ for (int i = 0; i < parts.keySet().size(); i++) {
+ TransactionPart tp = parts.get(i);
+ dataSize += tp.getDataSize();
+ }
+
+ byte[] combinedData = new byte[dataSize];
+ int currentPosition = 0;
+
+ // Stitch all the data sections together
+ for (int i = 0; i < parts.keySet().size(); i++) {
+ TransactionPart tp = parts.get(i);
+ System.arraycopy(tp.getData(), 0, combinedData, currentPosition, tp.getDataSize());
+ currentPosition += tp.getDataSize();
+ }
+
+ // Decoder Object
+ ByteBuffer bbDecode = ByteBuffer.wrap(combinedData);
+
+ // Decode how many key value pairs need to be decoded
+ int numberOfKVGuards = bbDecode.getInt();
+ int numberOfKVUpdates = bbDecode.getInt();
+
+ // Decode all the guard key values
+ for (int i = 0; i < numberOfKVGuards; i++) {
+ KeyValue kv = (KeyValue)KeyValue.decode(bbDecode);
+ keyValueGuardSet.add(kv);
+ }
+
+ // Decode all the updates key values
+ for (int i = 0; i < numberOfKVUpdates; i++) {
+ KeyValue kv = (KeyValue)KeyValue.decode(bbDecode);
+ keyValueUpdateSet.add(kv);
+ }
+ }
+
+ public boolean evaluateGuard(Map<IoTString, KeyValue> committedKeyValueTable, Map<IoTString, KeyValue> speculatedKeyValueTable, Map<IoTString, KeyValue> pendingTransactionSpeculatedKeyValueTable) {
+ for (KeyValue kvGuard : keyValueGuardSet) {
+
+ // First check if the key is in the speculative table, this is the value of the latest assumption
+ KeyValue kv = null;
+
+ // If we have a speculation table then use it first
+ if (pendingTransactionSpeculatedKeyValueTable != null) {
+ kv = pendingTransactionSpeculatedKeyValueTable.get(kvGuard.getKey());
+ }
+
+ // If we have a speculation table then use it first
+ if ((kv == null) && (speculatedKeyValueTable != null)) {
+ kv = speculatedKeyValueTable.get(kvGuard.getKey());
+ }
+
+ if (kv == null) {
+ // if it is not in the speculative table then check the committed table and use that
+ // value as our latest assumption
+ kv = committedKeyValueTable.get(kvGuard.getKey());
+ }
+
+ if (kvGuard.getValue() != null) {
+ if ((kv == null) || (!kvGuard.getValue().equals(kv.getValue()))) {
+
+
+ if (kv != null) {
+ System.out.println(kvGuard.getValue() + " " + kv.getValue());
+ } else {
+ System.out.println(kvGuard.getValue() + " " + kv);
+ }
+
+ return false;
+ }
+ } else {
+ if (kv != null) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+}
\ No newline at end of file
--- /dev/null
+package iotcloud;
+
+import java.nio.ByteBuffer;
+
+class TransactionPart extends Entry {
+
+ // Max size of the part excluding the fixed size header
+ public static final int MAX_NON_HEADER_SIZE = 512;
+
+ private long sequenceNumber = -1;
+ private long machineId = -1;
+ private long arbitratorId = -1;
+ private long clientLocalSequenceNumber = -1; // Sequence number of the transaction that this is a part of
+ private int partNumber = -1; // Parts position in the
+ private Boolean isLastPart = false;
+
+ private Pair<Long, Long> transactionId = null;
+ private Pair<Long, Integer> partId = null;
+
+ private byte[] data = null;
+
+ public TransactionPart(Slot s, long _machineId, long _arbitratorId, long _clientLocalSequenceNumber, int _partNumber, byte[] _data, Boolean _isLastPart) {
+ super(s);
+ machineId = _machineId;
+ arbitratorId = _arbitratorId;
+ clientLocalSequenceNumber = _clientLocalSequenceNumber;
+ partNumber = _partNumber;
+ data = _data;
+ isLastPart = _isLastPart;
+
+ transactionId = new Pair<Long, Long>(machineId, clientLocalSequenceNumber);
+ partId = new Pair<Long, Integer>(clientLocalSequenceNumber, partNumber);
+
+ }
+
+ public int getSize() {
+ if (data == null) {
+ return (4 * Long.BYTES) + (2 * Integer.BYTES) + (2 * Byte.BYTES);
+ }
+ return (4 * Long.BYTES) + (2 * Integer.BYTES) + (2 * Byte.BYTES) + data.length;
+ }
+
+ public void setSlot(Slot s) {
+ parentslot = s;
+ }
+
+ public Pair<Long, Long> getTransactionId() {
+ return transactionId;
+ }
+
+ public long getArbitratorId() {
+ return arbitratorId;
+ }
+
+ public Pair<Long, Integer> getPartId() {
+ return partId;
+ }
+
+ public int getPartNumber() {
+ return partNumber;
+ }
+
+ public int getDataSize() {
+ return data.length;
+ }
+
+ public byte[] getData() {
+ return data;
+ }
+
+ public Boolean isLastPart() {
+ return isLastPart;
+ }
+
+ public long getMachineId() {
+ return machineId;
+ }
+
+ public long getClientLocalSequenceNumber() {
+ return clientLocalSequenceNumber;
+ }
+
+
+ public long getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ public void setSequenceNumber(long _sequenceNumber) {
+ sequenceNumber = _sequenceNumber;
+ }
+
+ static Entry decode(Slot s, ByteBuffer bb) {
+ long sequenceNumber = bb.getLong();
+ long machineId = bb.getLong();
+ long arbitratorId = bb.getLong();
+ long clientLocalSequenceNumber = bb.getLong();
+ int partNumber = bb.getInt();
+ int dataSize = bb.getInt();
+ Boolean isLastPart = (bb.get() == 1);
+ // Get the data
+ byte[] data = new byte[dataSize];
+ bb.get(data);
+
+ TransactionPart returnTransactionPart = new TransactionPart(s, machineId, arbitratorId, clientLocalSequenceNumber, partNumber, data, isLastPart);
+ returnTransactionPart.setSequenceNumber(sequenceNumber);
+
+ return returnTransactionPart;
+ }
+
+ public void encode(ByteBuffer bb) {
+ bb.put(Entry.TypeTransactionPart);
+ bb.putLong(sequenceNumber);
+ bb.putLong(machineId);
+ bb.putLong(arbitratorId);
+ bb.putLong(clientLocalSequenceNumber);
+ bb.putInt(partNumber);
+ bb.putInt(data.length);
+
+ if (isLastPart) {
+ bb.put((byte)1);
+ } else {
+ bb.put((byte)0);
+ }
+
+ bb.put(data);
+ }
+
+ public byte getType() {
+ return Entry.TypeTransactionPart;
+ }
+
+ public Entry getCopy(Slot s) {
+
+ TransactionPart copyTransaction = new TransactionPart(s, machineId, arbitratorId, clientLocalSequenceNumber, partNumber, data, isLastPart);
+ copyTransaction.setSequenceNumber(sequenceNumber);
+
+ return copyTransaction;
+ }
+}
\ No newline at end of file
--- /dev/null
+package iotcloud;
+
+class TransactionStatus {
+ static final byte StatusAborted = 1;
+ static final byte StatusPending = 2;
+ static final byte StatusCommitted = 3;
+ // static final byte StatusRetrying = 4;
+ static final byte StatusSentPartial = 5;
+ static final byte StatusSentFully = 6;
+ static final byte StatusNoEffect = 10;
+
+ private byte status = 0;
+ private boolean applicationReleased = false;
+ private boolean wasSentInChain = false;
+ private long transactionSequenceNumber = 0;
+ private long arbitrator = -1;
+
+
+ public TransactionStatus(byte _status, long _arbitrator) {
+ status = _status;
+ arbitrator = _arbitrator;
+ }
+
+ public byte getStatus() {
+ return status;
+ }
+
+ public void setStatus(byte _status) {
+ status = _status;
+ }
+
+ public long getTransactionSequenceNumber() {
+ return transactionSequenceNumber;
+ }
+
+ public void setTransactionSequenceNumber(long _transactionSequenceNumber) {
+ transactionSequenceNumber = _transactionSequenceNumber;
+ }
+
+ public long getTransactionArbitrator() {
+ return arbitrator;
+ }
+
+ public void release() {
+ applicationReleased = true;
+ }
+
+ public boolean getReleased() {
+ return applicationReleased;
+ }
+}
\ No newline at end of file
--- /dev/null
+<?xml version="1.0" encoding="utf-8"?>
+<android.support.design.widget.CoordinatorLayout xmlns:android="http://schemas.android.com/apk/res/android"
+ xmlns:app="http://schemas.android.com/apk/res-auto"
+ xmlns:tools="http://schemas.android.com/tools"
+ android:layout_width="match_parent"
+ android:layout_height="match_parent"
+ android:fitsSystemWindows="true"
+ tools:context="com.example.ali.control.MainActivity">
+
+ <android.support.design.widget.AppBarLayout
+ android:layout_width="match_parent"
+ android:layout_height="wrap_content"
+ android:theme="@style/AppTheme.AppBarOverlay">
+
+ <android.support.v7.widget.Toolbar
+ android:id="@+id/toolbar"
+ android:layout_width="match_parent"
+ android:layout_height="?attr/actionBarSize"
+ android:background="?attr/colorPrimary"
+ app:popupTheme="@style/AppTheme.PopupOverlay" />
+
+ </android.support.design.widget.AppBarLayout>
+
+ <include layout="@layout/content_main" />
+
+ <android.support.design.widget.FloatingActionButton
+ android:id="@+id/fab"
+ android:layout_width="wrap_content"
+ android:layout_height="wrap_content"
+ android:layout_gravity="bottom|end"
+ android:layout_margin="@dimen/fab_margin"
+ android:src="@android:drawable/ic_dialog_email" />
+
+</android.support.design.widget.CoordinatorLayout>
--- /dev/null
+<?xml version="1.0" encoding="utf-8"?>
+<RelativeLayout xmlns:android="http://schemas.android.com/apk/res/android"
+ xmlns:app="http://schemas.android.com/apk/res-auto"
+ xmlns:tools="http://schemas.android.com/tools"
+ android:layout_width="match_parent"
+ android:layout_height="match_parent"
+ android:paddingBottom="@dimen/activity_vertical_margin"
+ android:paddingLeft="@dimen/activity_horizontal_margin"
+ android:paddingRight="@dimen/activity_horizontal_margin"
+ android:paddingTop="@dimen/activity_vertical_margin"
+ app:layout_behavior="@string/appbar_scrolling_view_behavior"
+ tools:context="com.example.ali.control.MainActivity"
+ tools:showIn="@layout/activity_main">
+
+ <TextView
+ android:id="@+id/textView4"
+ android:layout_width="wrap_content"
+ android:layout_height="wrap_content"
+ android:layout_marginTop="60dp"
+ android:text="Hello World! This is a test"
+ android:layout_below="@+id/textView"
+ android:layout_alignParentLeft="true"
+ android:layout_alignParentStart="true" />
+
+ <TextView
+ android:layout_width="wrap_content"
+ android:layout_height="wrap_content"
+ android:text="Hello World! This is a test"
+ android:id="@+id/textView"
+ android:layout_marginTop="27dp"
+ android:layout_below="@+id/switch5"
+ android:layout_alignParentLeft="true"
+ android:layout_alignParentStart="true"
+ android:layout_alignParentRight="true"
+ android:layout_alignParentEnd="true" />
+
+ <Switch
+ android:layout_width="wrap_content"
+ android:layout_height="wrap_content"
+ android:text="Light 1 "
+ android:id="@+id/switch1"
+ android:checked="false"
+ android:layout_alignParentTop="true"
+ android:layout_alignParentLeft="true"
+ android:layout_alignParentStart="true"
+ android:layout_marginLeft="104dp"
+ android:layout_marginStart="104dp" />
+
+ <Switch
+ android:layout_width="wrap_content"
+ android:layout_height="wrap_content"
+ android:text="Light 2 "
+ android:id="@+id/switch2"
+ android:checked="false"
+ android:layout_below="@+id/switch1"
+ android:layout_alignLeft="@+id/switch1"
+ android:layout_alignStart="@+id/switch1" />
+
+ <Switch
+ android:layout_width="wrap_content"
+ android:layout_height="wrap_content"
+ android:text="Light 3 "
+ android:id="@+id/switch3"
+ android:checked="false"
+ android:layout_below="@+id/switch2"
+ android:layout_alignLeft="@+id/switch2"
+ android:layout_alignStart="@+id/switch2" />
+
+ <Switch
+ android:layout_width="wrap_content"
+ android:layout_height="wrap_content"
+ android:text="Fan 1 "
+ android:id="@+id/switch4"
+ android:checked="false"
+ android:layout_below="@+id/switch3"
+ android:layout_alignLeft="@+id/switch3"
+ android:layout_alignStart="@+id/switch3" />
+
+ <Switch
+ android:layout_width="wrap_content"
+ android:layout_height="wrap_content"
+ android:text="Fan 2 "
+ android:id="@+id/switch5"
+ android:checked="false"
+ android:layout_below="@+id/switch4"
+ android:layout_alignLeft="@+id/switch3" />
+
+</RelativeLayout>
--- /dev/null
+<menu xmlns:android="http://schemas.android.com/apk/res/android"
+ xmlns:app="http://schemas.android.com/apk/res-auto"
+ xmlns:tools="http://schemas.android.com/tools"
+ tools:context="com.example.ali.control.MainActivity">
+ <item
+ android:id="@+id/action_settings"
+ android:orderInCategory="100"
+ android:title="@string/action_settings"
+ app:showAsAction="never" />
+</menu>
--- /dev/null
+<resources>>
+
+ <style name="AppTheme.NoActionBar">
+ <item name="windowActionBar">false</item>
+ <item name="windowNoTitle">true</item>
+ <item name="android:windowDrawsSystemBarBackgrounds">true</item>
+ <item name="android:statusBarColor">@android:color/transparent</item>
+ </style>
+</resources>
--- /dev/null
+<resources>
+ <!-- Example customization of dimensions originally defined in res/values/dimens.xml
+ (such as screen margins) for screens with more than 820dp of available width. This
+ would include 7" and 10" devices in landscape (~960dp and ~1280dp respectively). -->
+ <dimen name="activity_horizontal_margin">64dp</dimen>
+</resources>
--- /dev/null
+<?xml version="1.0" encoding="utf-8"?>
+<resources>
+ <color name="colorPrimary">#3F51B5</color>
+ <color name="colorPrimaryDark">#303F9F</color>
+ <color name="colorAccent">#FF4081</color>
+</resources>
--- /dev/null
+<resources>
+ <!-- Default screen margins, per the Android Design guidelines. -->
+ <dimen name="activity_horizontal_margin">16dp</dimen>
+ <dimen name="activity_vertical_margin">16dp</dimen>
+ <dimen name="fab_margin">16dp</dimen>
+</resources>
--- /dev/null
+<resources>
+ <string name="app_name">Control</string>
+ <string name="action_settings">Settings</string>
+</resources>
--- /dev/null
+<resources>
+
+ <!-- Base application theme. -->
+ <style name="AppTheme" parent="Theme.AppCompat.Light.DarkActionBar">
+ <!-- Customize your theme here. -->
+ <item name="colorPrimary">@color/colorPrimary</item>
+ <item name="colorPrimaryDark">@color/colorPrimaryDark</item>
+ <item name="colorAccent">@color/colorAccent</item>
+ </style>
+
+ <style name="AppTheme.NoActionBar">
+ <item name="windowActionBar">false</item>
+ <item name="windowNoTitle">true</item>
+ </style>
+
+ <style name="AppTheme.AppBarOverlay" parent="ThemeOverlay.AppCompat.Dark.ActionBar" />
+
+ <style name="AppTheme.PopupOverlay" parent="ThemeOverlay.AppCompat.Light" />
+
+</resources>
--- /dev/null
+package com.example.ali.control;
+
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+/**
+ * To work on unit tests, switch the Test Artifact in the Build Variants view.
+ */
+public class ExampleUnitTest {
+ @Test
+ public void addition_isCorrect() throws Exception {
+ assertEquals(4, 2 + 2);
+ }
+}
\ No newline at end of file
--- /dev/null
+// Top-level build file where you can add configuration options common to all sub-projects/modules.
+
+buildscript {
+ repositories {
+ jcenter()
+ }
+ dependencies {
+ classpath 'com.android.tools.build:gradle:2.4.0-alpha7'
+
+ // NOTE: Do not place your application dependencies here; they belong
+ // in the individual module build.gradle files
+ }
+}
+
+
+
+allprojects {
+ repositories {
+ jcenter()
+ }
+}
+
+task clean(type: Delete) {
+ delete rootProject.buildDir
+}
--- /dev/null
+# Project-wide Gradle settings.
+
+# IDE (e.g. Android Studio) users:
+# Gradle settings configured through the IDE *will override*
+# any settings specified in this file.
+
+# For more details on how to configure your build environment visit
+# http://www.gradle.org/docs/current/userguide/build_environment.html
+
+# Specifies the JVM arguments used for the daemon process.
+# The setting is particularly useful for tweaking memory settings.
+# Default value: -Xmx10248m -XX:MaxPermSize=256m
+# org.gradle.jvmargs=-Xmx2048m -XX:MaxPermSize=512m -XX:+HeapDumpOnOutOfMemoryError -Dfile.encoding=UTF-8
+
+# When configured, Gradle will run in incubating parallel mode.
+# This option should only be used with decoupled projects. More details, visit
+# http://www.gradle.org/docs/current/userguide/multi_project_builds.html#sec:decoupled_projects
+# org.gradle.parallel=true
\ No newline at end of file
--- /dev/null
+#Fri May 12 20:24:22 PDT 2017
+distributionBase=GRADLE_USER_HOME
+distributionPath=wrapper/dists
+zipStoreBase=GRADLE_USER_HOME
+zipStorePath=wrapper/dists
+distributionUrl=https\://services.gradle.org/distributions/gradle-3.4.1-all.zip
--- /dev/null
+#!/usr/bin/env bash
+
+##############################################################################
+##
+## Gradle start up script for UN*X
+##
+##############################################################################
+
+# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+DEFAULT_JVM_OPTS=""
+
+APP_NAME="Gradle"
+APP_BASE_NAME=`basename "$0"`
+
+# Use the maximum available, or set MAX_FD != -1 to use that value.
+MAX_FD="maximum"
+
+warn ( ) {
+ echo "$*"
+}
+
+die ( ) {
+ echo
+ echo "$*"
+ echo
+ exit 1
+}
+
+# OS specific support (must be 'true' or 'false').
+cygwin=false
+msys=false
+darwin=false
+case "`uname`" in
+ CYGWIN* )
+ cygwin=true
+ ;;
+ Darwin* )
+ darwin=true
+ ;;
+ MINGW* )
+ msys=true
+ ;;
+esac
+
+# Attempt to set APP_HOME
+# Resolve links: $0 may be a link
+PRG="$0"
+# Need this for relative symlinks.
+while [ -h "$PRG" ] ; do
+ ls=`ls -ld "$PRG"`
+ link=`expr "$ls" : '.*-> \(.*\)$'`
+ if expr "$link" : '/.*' > /dev/null; then
+ PRG="$link"
+ else
+ PRG=`dirname "$PRG"`"/$link"
+ fi
+done
+SAVED="`pwd`"
+cd "`dirname \"$PRG\"`/" >/dev/null
+APP_HOME="`pwd -P`"
+cd "$SAVED" >/dev/null
+
+CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
+
+# Determine the Java command to use to start the JVM.
+if [ -n "$JAVA_HOME" ] ; then
+ if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+ # IBM's JDK on AIX uses strange locations for the executables
+ JAVACMD="$JAVA_HOME/jre/sh/java"
+ else
+ JAVACMD="$JAVA_HOME/bin/java"
+ fi
+ if [ ! -x "$JAVACMD" ] ; then
+ die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+ fi
+else
+ JAVACMD="java"
+ which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+fi
+
+# Increase the maximum file descriptors if we can.
+if [ "$cygwin" = "false" -a "$darwin" = "false" ] ; then
+ MAX_FD_LIMIT=`ulimit -H -n`
+ if [ $? -eq 0 ] ; then
+ if [ "$MAX_FD" = "maximum" -o "$MAX_FD" = "max" ] ; then
+ MAX_FD="$MAX_FD_LIMIT"
+ fi
+ ulimit -n $MAX_FD
+ if [ $? -ne 0 ] ; then
+ warn "Could not set maximum file descriptor limit: $MAX_FD"
+ fi
+ else
+ warn "Could not query maximum file descriptor limit: $MAX_FD_LIMIT"
+ fi
+fi
+
+# For Darwin, add options to specify how the application appears in the dock
+if $darwin; then
+ GRADLE_OPTS="$GRADLE_OPTS \"-Xdock:name=$APP_NAME\" \"-Xdock:icon=$APP_HOME/media/gradle.icns\""
+fi
+
+# For Cygwin, switch paths to Windows format before running java
+if $cygwin ; then
+ APP_HOME=`cygpath --path --mixed "$APP_HOME"`
+ CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
+ JAVACMD=`cygpath --unix "$JAVACMD"`
+
+ # We build the pattern for arguments to be converted via cygpath
+ ROOTDIRSRAW=`find -L / -maxdepth 1 -mindepth 1 -type d 2>/dev/null`
+ SEP=""
+ for dir in $ROOTDIRSRAW ; do
+ ROOTDIRS="$ROOTDIRS$SEP$dir"
+ SEP="|"
+ done
+ OURCYGPATTERN="(^($ROOTDIRS))"
+ # Add a user-defined pattern to the cygpath arguments
+ if [ "$GRADLE_CYGPATTERN" != "" ] ; then
+ OURCYGPATTERN="$OURCYGPATTERN|($GRADLE_CYGPATTERN)"
+ fi
+ # Now convert the arguments - kludge to limit ourselves to /bin/sh
+ i=0
+ for arg in "$@" ; do
+ CHECK=`echo "$arg"|egrep -c "$OURCYGPATTERN" -`
+ CHECK2=`echo "$arg"|egrep -c "^-"` ### Determine if an option
+
+ if [ $CHECK -ne 0 ] && [ $CHECK2 -eq 0 ] ; then ### Added a condition
+ eval `echo args$i`=`cygpath --path --ignore --mixed "$arg"`
+ else
+ eval `echo args$i`="\"$arg\""
+ fi
+ i=$((i+1))
+ done
+ case $i in
+ (0) set -- ;;
+ (1) set -- "$args0" ;;
+ (2) set -- "$args0" "$args1" ;;
+ (3) set -- "$args0" "$args1" "$args2" ;;
+ (4) set -- "$args0" "$args1" "$args2" "$args3" ;;
+ (5) set -- "$args0" "$args1" "$args2" "$args3" "$args4" ;;
+ (6) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" ;;
+ (7) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" ;;
+ (8) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" ;;
+ (9) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" "$args8" ;;
+ esac
+fi
+
+# Split up the JVM_OPTS And GRADLE_OPTS values into an array, following the shell quoting and substitution rules
+function splitJvmOpts() {
+ JVM_OPTS=("$@")
+}
+eval splitJvmOpts $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS
+JVM_OPTS[${#JVM_OPTS[*]}]="-Dorg.gradle.appname=$APP_BASE_NAME"
+
+exec "$JAVACMD" "${JVM_OPTS[@]}" -classpath "$CLASSPATH" org.gradle.wrapper.GradleWrapperMain "$@"
--- /dev/null
+@if "%DEBUG%" == "" @echo off\r
+@rem ##########################################################################\r
+@rem\r
+@rem Gradle startup script for Windows\r
+@rem\r
+@rem ##########################################################################\r
+\r
+@rem Set local scope for the variables with windows NT shell\r
+if "%OS%"=="Windows_NT" setlocal\r
+\r
+@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.\r
+set DEFAULT_JVM_OPTS=\r
+\r
+set DIRNAME=%~dp0\r
+if "%DIRNAME%" == "" set DIRNAME=.\r
+set APP_BASE_NAME=%~n0\r
+set APP_HOME=%DIRNAME%\r
+\r
+@rem Find java.exe\r
+if defined JAVA_HOME goto findJavaFromJavaHome\r
+\r
+set JAVA_EXE=java.exe\r
+%JAVA_EXE% -version >NUL 2>&1\r
+if "%ERRORLEVEL%" == "0" goto init\r
+\r
+echo.\r
+echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.\r
+echo.\r
+echo Please set the JAVA_HOME variable in your environment to match the\r
+echo location of your Java installation.\r
+\r
+goto fail\r
+\r
+:findJavaFromJavaHome\r
+set JAVA_HOME=%JAVA_HOME:"=%\r
+set JAVA_EXE=%JAVA_HOME%/bin/java.exe\r
+\r
+if exist "%JAVA_EXE%" goto init\r
+\r
+echo.\r
+echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%\r
+echo.\r
+echo Please set the JAVA_HOME variable in your environment to match the\r
+echo location of your Java installation.\r
+\r
+goto fail\r
+\r
+:init\r
+@rem Get command-line arguments, handling Windowz variants\r
+\r
+if not "%OS%" == "Windows_NT" goto win9xME_args\r
+if "%@eval[2+2]" == "4" goto 4NT_args\r
+\r
+:win9xME_args\r
+@rem Slurp the command line arguments.\r
+set CMD_LINE_ARGS=\r
+set _SKIP=2\r
+\r
+:win9xME_args_slurp\r
+if "x%~1" == "x" goto execute\r
+\r
+set CMD_LINE_ARGS=%*\r
+goto execute\r
+\r
+:4NT_args\r
+@rem Get arguments from the 4NT Shell from JP Software\r
+set CMD_LINE_ARGS=%$\r
+\r
+:execute\r
+@rem Setup the command line\r
+\r
+set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar\r
+\r
+@rem Execute Gradle\r
+"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %CMD_LINE_ARGS%\r
+\r
+:end\r
+@rem End local scope for the variables with windows NT shell\r
+if "%ERRORLEVEL%"=="0" goto mainEnd\r
+\r
+:fail\r
+rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of\r
+rem the _cmd.exe /c_ return code!\r
+if not "" == "%GRADLE_EXIT_CONSOLE%" exit 1\r
+exit /b 1\r
+\r
+:mainEnd\r
+if "%OS%"=="Windows_NT" endlocal\r
+\r
+:omega\r
--- /dev/null
+include ':app'
projects / iotcloud.git / commitdiff
